[Feature] configs: new config infra with pydantic, with config via en…

…v and dynamic rootfs support (#264)

This PR introduces otaclient package scope configuration, otaclient runtime configurable configs via environmental variables with validation and dynamic rooted path constants support.

* Standalone otaclient.configs package for holding otaclient package scope configs
Configs for otaclient top-level modules(i.e., app) and common used configs (i.e., debug configs, logging configs, etc,) are grouped together and placed under otaclient.configs package.

* pydantic as configs storage back-end
With pydantic as configs storing back-end, validation to otaclient configs are introduced and enforced. Other features from pydantic are also become available for Configs classes instances.

* User configurable configs can be configured via env var
With pydantic and pydantic-settings, some of the configs(configs in app_cfg.ConfigurableConfigs, debug_cfg and logging_cfg) that configure otaclient's runtime behavior are exposed to end-user and configurable via environmental variables.

* Rootfs specifying support and dynamic rooted path constants support
Now for otaclient running under container environment, the actual host rootfs mount point can be specified by OTA_HOST_ROOTFS, allowing otaclient to properly calling commands and executables from the host.

otaclient/_utils/__init__.py

@@ -11,34 +11,29 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from typing import Any, Callable, TypeVar
-from typing_extensions import ParamSpec, Concatenate
-
-P = ParamSpec("P")
-
 
-def copy_callable_typehint(_source: Callable[P, Any]):
-    """This helper function return a decorator that can type hint the target
-    function as the _source function.
 
-    At runtime, this decorator actually does nothing, but just return the input function as it.
-    But the returned function will have the same type hint as the source function in ide.
-    It will not impact the runtime behavior of the decorated function.
-    """
-
-    def _decorator(target) -> Callable[P, Any]:
-        return target
-
-    return _decorator
+from __future__ import annotations
+import os.path
+from functools import cached_property
+from pydantic import computed_field
+from typing import Any, Callable, TypeVar
 
+T = TypeVar("T")
 
-RT = TypeVar("RT")
+_CONTAINER_INDICATOR_FILES = [
+    "/.dockerenv",
+    "/run/.dockerenv",
+    "/run/.containerenv",
+]
 
 
-def copy_callable_typehint_to_method(_source: Callable[P, Any]):
-    """Works the same as copy_callable_typehint, but omit the first arg."""
+def if_run_as_container() -> bool:
+    for indicator in _CONTAINER_INDICATOR_FILES:
+        if os.path.isfile(indicator):
+            return True
+    return False
 
-    def _decorator(target: Callable[..., RT]) -> Callable[Concatenate[Any, P], RT]:
-        return target  # type: ignore
 
-    return _decorator
+def cached_computed_field(_f: Callable[[Any], Any]) -> cached_property[Any]:
+    return computed_field(cached_property(_f))

otaclient/_utils/logging.py

@@ -61,3 +61,9 @@ def filter(self, _: logging.LogRecord) -> bool:
             )
             self._round_warned = True
         return False
+
+
+def check_loglevel(_in: int) -> int:
+    """Pydantic validator for logging level number."""
+    assert _in in logging._levelToName, f"{_in} is not a valid logging level"
+    return _in

otaclient/_utils/path.py

@@ -0,0 +1,35 @@
+# Copyright 2022 TIER IV, INC. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import os.path
+from otaclient._utils.typing import StrOrPath
+
+
+def replace_root(path: StrOrPath, old_root: StrOrPath, new_root: StrOrPath) -> str:
+    """Replace a <path> relative to <old_root> to <new_root>.
+
+    For example, if path="/abc", old_root="/", new_root="/new_root",
+    then we will have "/new_root/abc".
+    """
+    # normalize all the input args
+    path = os.path.normpath(path)
+    old_root = os.path.normpath(old_root)
+    new_root = os.path.normpath(new_root)
+
+    if not (old_root.startswith("/") and new_root.startswith("/")):
+        raise ValueError(f"{old_root=} and/or {new_root=} is not valid root")
+    if os.path.commonpath([path, old_root]) != old_root:
+        raise ValueError(f"{old_root=} is not the root of {path=}")
+    return os.path.join(new_root, os.path.relpath(path, old_root))

otaclient/_utils/typing.py

@@ -0,0 +1,51 @@
+# Copyright 2022 TIER IV, INC. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from pathlib import Path
+from typing import Any, Callable, TypeVar, Union
+from typing_extensions import Concatenate, ParamSpec
+
+P = ParamSpec("P")
+RT = TypeVar("RT")
+
+StrOrPath = Union[str, Path]
+
+
+def copy_callable_typehint(_source: Callable[P, Any]):
+    """This helper function return a decorator that can type hint the target
+    function as the _source function.
+
+    At runtime, this decorator actually does nothing, but just return the input function as it.
+    But the returned function will have the same type hint as the source function in ide.
+    It will not impact the runtime behavior of the decorated function.
+    """
+
+    def _decorator(target) -> Callable[P, Any]:
+        return target
+
+    return _decorator
+
+
+def copy_callable_typehint_to_method(_source: Callable[P, Any]):
+    """Works the same as copy_callable_typehint, but omit the first arg."""
+
+    def _decorator(target: Callable[..., RT]) -> Callable[Concatenate[Any, P], RT]:
+        return target  # type: ignore
+
+    return _decorator
+
+
+def check_port(_in: Any) -> bool:
+    return isinstance(_in, int) and _in >= 0 and _in <= 65535

otaclient/app/boot_control/_cboot.py

@@ -20,7 +20,7 @@
 from subprocess import CalledProcessError
 from typing import Generator, Optional
 
-
+from ..configs import config as cfg
 from .. import log_setting, errors as ota_errors
 from ..common import (
     copytree_identical,
@@ -39,14 +39,12 @@
     SlotInUseMixin,
     VersionControlMixin,
 )
-from .configs import cboot_cfg as cfg
+from .configs import cboot_cfg as boot_cfg
 from .protocol import BootControllerProtocol
 from .firmware import Firmware
 
 
-logger = log_setting.get_logger(
-    __name__, cfg.LOG_LEVEL_TABLE.get(__name__, cfg.DEFAULT_LOG_LEVEL)
-)
+logger = log_setting.get_logger(__name__)
 
 
 class NvbootctrlError(Exception):
@@ -160,14 +158,14 @@ class _CBootControl:
     def __init__(self):
         # NOTE: only support rqx-580, rqx-58g platform right now!
         # detect the chip id
-        self.chip_id = read_str_from_file(cfg.TEGRA_CHIP_ID_PATH)
-        if not self.chip_id or int(self.chip_id) not in cfg.CHIP_ID_MODEL_MAP:
+        self.chip_id = read_str_from_file(boot_cfg.TEGRA_CHIP_ID_FPATH)
+        if not self.chip_id or int(self.chip_id) not in boot_cfg.CHIP_ID_MODEL_MAP:
             raise NotImplementedError(
                 f"unsupported platform found (chip_id: {self.chip_id}), abort"
             )
 
         self.chip_id = int(self.chip_id)
-        self.model = cfg.CHIP_ID_MODEL_MAP[self.chip_id]
+        self.model = boot_cfg.CHIP_ID_MODEL_MAP[self.chip_id]
         logger.info(f"{self.model=}, (chip_id={hex(self.chip_id)})")
 
         # initializing dev info
@@ -300,33 +298,36 @@ def __init__(self) -> None:
         try:
             self._cboot_control: _CBootControl = _CBootControl()
 
+            # ------ prepare mount space ------ #
+            otaclient_ms = Path(cfg.OTACLIENT_MOUNT_SPACE_DPATH)
+            otaclient_ms.mkdir(exist_ok=True, parents=True)
+            otaclient_ms.chmod(0o700)
+
             # load paths
             ## first try to unmount standby dev if possible
             self.standby_slot_dev = self._cboot_control.get_standby_rootfs_dev()
             CMDHelperFuncs.umount(self.standby_slot_dev)
 
-            self.standby_slot_mount_point = Path(cfg.MOUNT_POINT)
-            self.standby_slot_mount_point.mkdir(exist_ok=True)
+            self.standby_slot_mount_point = Path(cfg.STANDBY_SLOT_MP)
+            self.standby_slot_mount_point.mkdir(exist_ok=True, parents=True)
 
             ## refroot mount point
-            _refroot_mount_point = cfg.ACTIVE_ROOT_MOUNT_POINT
-            # first try to umount refroot mount point
-            CMDHelperFuncs.umount(_refroot_mount_point)
-            if not os.path.isdir(_refroot_mount_point):
-                os.mkdir(_refroot_mount_point)
+            _refroot_mount_point = cfg.ACTIVE_SLOT_MP
             self.ref_slot_mount_point = Path(_refroot_mount_point)
 
+            if os.path.isdir(_refroot_mount_point):
+                # first try to umount refroot mount point
+                CMDHelperFuncs.umount(_refroot_mount_point)
+            elif not os.path.exists(_refroot_mount_point):
+                self.ref_slot_mount_point.mkdir(exist_ok=True, parents=True)
+
             ## ota-status dir
             ### current slot
-            self.current_ota_status_dir = Path(cfg.ACTIVE_ROOTFS_PATH) / Path(
-                cfg.OTA_STATUS_DIR
-            ).relative_to("/")
+            self.current_ota_status_dir = Path(boot_cfg.ACTIVE_BOOT_OTA_STATUS_DPATH)
             self.current_ota_status_dir.mkdir(parents=True, exist_ok=True)
             ### standby slot
             # NOTE: might not yet be populated before OTA update applied!
-            self.standby_ota_status_dir = self.standby_slot_mount_point / Path(
-                cfg.OTA_STATUS_DIR
-            ).relative_to("/")
+            self.standby_ota_status_dir = Path(boot_cfg.STANDBY_BOOT_OTA_STATUS_DPATH)
 
             # init ota-status
             self._init_boot_control()
@@ -418,7 +419,7 @@ def _is_switching_boot(self) -> bool:
 
     def _populate_boot_folder_to_separate_bootdev(self):
         # mount the actual standby_boot_dev now
-        _boot_dir_mount_point = Path(cfg.SEPARATE_BOOT_MOUNT_POINT)
+        _boot_dir_mount_point = Path(boot_cfg.SEPARATE_BOOT_MOUNT_POINT)
         _boot_dir_mount_point.mkdir(exist_ok=True, parents=True)
 
         try:
@@ -513,16 +514,11 @@ def pre_update(self, version: str, *, standby_as_ref: bool, erase_standby=False)
     def post_update(self) -> Generator[None, None, None]:
         try:
             # firmware update
-            firmware = Firmware(
-                self.standby_slot_mount_point
-                / Path(cfg.FIRMWARE_CONFIG).relative_to("/")
-            )
+            firmware = Firmware(Path(boot_cfg.FIRMWARE_CFG_STANDBY_FPATH))
             firmware.update(int(self._cboot_control.get_standby_slot()))
 
             # update extlinux_cfg file
-            _extlinux_cfg = self.standby_slot_mount_point / Path(
-                cfg.EXTLINUX_FILE
-            ).relative_to("/")
+            _extlinux_cfg = Path(boot_cfg.STANDBY_EXTLINUX_FPATH)
             self._cboot_control.update_extlinux_cfg(
                 dst=_extlinux_cfg,
                 ref=_extlinux_cfg,

otaclient/app/boot_control/_common.py

@@ -21,6 +21,7 @@
 from subprocess import CalledProcessError
 from typing import List, Optional, Union, Callable
 
+from otaclient._utils.path import replace_root
 from ._errors import (
     BootControlError,
     MountError,
@@ -39,9 +40,7 @@
 from ..proto import wrapper
 
 
-logger = log_setting.get_logger(
-    __name__, cfg.LOG_LEVEL_TABLE.get(__name__, cfg.DEFAULT_LOG_LEVEL)
-)
+logger = log_setting.get_logger(__name__)
 
 
 class CMDHelperFuncs:
@@ -798,9 +797,7 @@ def __init__(
         # NOTE(20230907): this will always be <standby_slot_mp>/boot,
         #                 in the future this attribute will not be used by
         #                 standby slot creater.
-        self.standby_boot_dir = self.standby_slot_mount_point / Path(
-            cfg.BOOT_DIR
-        ).relative_to("/")
+        self.standby_boot_dir = self.standby_slot_mount_point / "boot"
 
     def mount_standby(self, *, raise_exc: bool = True) -> bool:
         """Mount standby slot dev to <standby_slot_mount_point>.
@@ -861,12 +858,15 @@ def preserve_ota_folder_to_standby(self):
         so we should preserve it for each slot, accross each update.
         """
         logger.debug("copy /boot/ota from active to standby.")
+
+        _src = Path(cfg.BOOT_OTA_DPATH)
+        _dst = Path(
+            replace_root(cfg.BOOT_OTA_DPATH, cfg.ACTIVE_ROOTFS, cfg.STANDBY_SLOT_MP)
+        )
         try:
-            _src = self.active_slot_mount_point / Path(cfg.OTA_DIR).relative_to("/")
-            _dst = self.standby_slot_mount_point / Path(cfg.OTA_DIR).relative_to("/")
             shutil.copytree(_src, _dst, dirs_exist_ok=True)
         except Exception as e:
-            raise ValueError(f"failed to copy /boot/ota from active to standby: {e!r}")
+            raise ValueError(f"failed to copy {_src=} to {_dst=}: {e!r}")
 
     def umount_all(self, *, ignore_error: bool = False):
         logger.debug("unmount standby slot and active slot mount point...")