feat: implement config_file_monitor

README.md

@@ -16,6 +16,11 @@ Currently only ECU id will be checked, IP checking is not performed as sub ECU o
 
 NOTE that if `ecu_info.yaml` file is not presented, the filtering will be DISABLED.
 
+## Auto restart on config files changed
+
+By default, the `EXIT_ON_CONFIG_FILE_CHANGED` is enabled.
+Together with systemd.service `Restart` policy configured, automatically restart iot-logger server on config files changed can be achieved.
+
 ## Usage
 
 ### Environmental variables
@@ -37,3 +42,4 @@ The behaviors of the iot_logging_server can be configured with the following env
 | MAX_LOGS_BACKLOG | `4096` | Max pending log entries. |
 | MAX_LOGS_PER_MERGE | `512` | Max log entries in a merge group. |
 | UPLOAD_INTERVAL | `3` | Interval of uploading log batches to cloud. **Note that if the logger is restarted before next upload occurs, the pending loggings will be dropped.** |
+| EXIT_ON_CONFIG_FILE_CHANGED | `true` | Whether to kill the server on config files changed. **Note that this feature is expected to be used together with systemd.service Restart.** |

src/otaclient_iot_logging_server/__main__.py

@@ -21,14 +21,14 @@
 from otaclient_iot_logging_server._common import LogsQueue
 from otaclient_iot_logging_server._log_setting import config_logging
 from otaclient_iot_logging_server.aws_iot_logger import start_aws_iot_logger_thread
+from otaclient_iot_logging_server.config_file_monitor import config_file_monitor_thread
 from otaclient_iot_logging_server.configs import server_cfg
 from otaclient_iot_logging_server.log_proxy_server import launch_server
 
 
 def main() -> None:
     # server scope log entries pipe
     queue: LogsQueue = Queue(maxsize=server_cfg.MAX_LOGS_BACKLOG)
-
     # ------ configure local logging ------ #
     root_logger = config_logging(
         queue,
@@ -38,13 +38,16 @@ def main() -> None:
         server_logstream_suffix=server_cfg.SERVER_LOGSTREAM_SUFFIX,
     )
 
-    # ------ start server ------ #
     root_logger.info(
         f"launching iot_logging_server({__version__}) at http://{server_cfg.LISTEN_ADDRESS}:{server_cfg.LISTEN_PORT}"
     )
     root_logger.info(f"iot_logging_server config: \n{server_cfg}")
-
+    # ------ launch aws cloudwatch client ------ #
     start_aws_iot_logger_thread(queue)
+    # ------ launch config file monitor ------ #
+    if server_cfg.EXIT_ON_CONFIG_FILE_CHANGED:
+        config_file_monitor_thread()
+    # ------ start server ------ #
     launch_server(queue=queue)  # NoReturn
 
 

src/otaclient_iot_logging_server/config_file_monitor.py

@@ -0,0 +1,79 @@
+# 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.
+"""Monitor the used config files.
+
+Monitor the files listed in <monitored_config_files>, kill the server
+if any of the files are changed.
+
+This is expected to be used together with systemd.unit Restart policy
+to achieve automatically restart on configuration files changed.
+"""
+
+
+from __future__ import annotations
+
+import logging
+import os
+import signal
+import threading
+import time
+from os import stat_result
+from pathlib import Path
+from typing import NamedTuple, NoReturn
+
+logger = logging.getLogger(__name__)
+
+_CHECK_INTERVAL = 3  # second
+
+monitored_config_files: set[str] = set()
+_monitored_files_stat: dict[str, _MCTime] = {}
+
+
+class _MCTime(NamedTuple):
+    mtime: int
+    ctime: int
+
+    def file_changed(self, new_mctime: _MCTime) -> bool:
+        # if create time is newer in <new_mctime>, it means the file is recreated.
+        # if modified time is newer in <new_mctime>, it means the file is modified.
+        return self.ctime < new_mctime.ctime or self.mtime < new_mctime.mtime
+
+    @classmethod
+    def from_stat(cls, stat: stat_result) -> _MCTime:
+        return cls(int(stat.st_mtime), int(stat.st_ctime))
+
+
+def _config_file_monitor() -> NoReturn:
+    # initialize, record the original status
+    logger.info(f"start to monitor the changes of {monitored_config_files}")
+    while True:
+        for entry in monitored_config_files:
+            new_f_mctime = _MCTime.from_stat(Path(entry).stat())
+            if entry not in _monitored_files_stat:
+                _monitored_files_stat[entry] = new_f_mctime
+                continue
+
+            f_mctime = _monitored_files_stat[entry]
+            if f_mctime.file_changed(new_f_mctime):
+                logger.warning(f"detect change on config file {entry}, exit")
+                # NOTE: sys.exit is not working in thread
+                os.kill(os.getpid(), signal.SIGINT)
+
+        time.sleep(_CHECK_INTERVAL)
+
+
+def config_file_monitor_thread() -> threading.Thread:
+    t = threading.Thread(target=_config_file_monitor, daemon=True)
+    t.start()
+    return t

src/otaclient_iot_logging_server/configs.py

@@ -24,6 +24,8 @@
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from typing_extensions import Annotated
 
+from otaclient_iot_logging_server.config_file_monitor import monitored_config_files
+
 _LoggingLevelName = Literal["INFO", "DEBUG", "CRITICAL", "ERROR", "WARNING"]
 
 
@@ -52,6 +54,9 @@ class ConfigurableLoggingServerConfig(BaseSettings):
 
     ECU_INFO_YAML: str = "/boot/ota/ecu_info.yaml"
 
+    EXIT_ON_CONFIG_FILE_CHANGED: bool = True
+    """Kill the server when any config files changed."""
+
 
 class _AWSProfile(BaseModel):
     model_config = SettingsConfigDict(frozen=True)
@@ -76,3 +81,4 @@ def load_profile_info(_cfg_fpath: str) -> AWSProfileInfo:
 
 server_cfg = ConfigurableLoggingServerConfig()
 profile_info = load_profile_info(server_cfg.AWS_PROFILE_INFO)
+monitored_config_files.add(server_cfg.AWS_PROFILE_INFO)

src/otaclient_iot_logging_server/ecu_info.py

@@ -27,6 +27,9 @@
 import yaml
 from pydantic import BaseModel, ConfigDict, Field, IPvAnyAddress
 
+from otaclient_iot_logging_server.config_file_monitor import monitored_config_files
+from otaclient_iot_logging_server.configs import server_cfg
+
 logger = logging.getLogger(__name__)
 
 
@@ -65,3 +68,8 @@ def parse_ecu_info(ecu_info_file: Path | str) -> Optional[ECUInfo]:
         return ECUInfo.model_validate(loaded_ecu_info, strict=True)
     except Exception as e:
         logger.info(f"{ecu_info_file=} is invalid or missing: {e!r}")
+
+
+ecu_info = parse_ecu_info(server_cfg.ECU_INFO_YAML)
+if ecu_info:
+    monitored_config_files.add(server_cfg.ECU_INFO_YAML)

src/otaclient_iot_logging_server/greengrass_config.py

@@ -28,6 +28,7 @@
 from pydantic import computed_field
 
 from otaclient_iot_logging_server._utils import FixedConfig, chain_query, remove_prefix
+from otaclient_iot_logging_server.config_file_monitor import monitored_config_files
 from otaclient_iot_logging_server.configs import profile_info, server_cfg
 
 logger = logging.getLogger(__name__)
@@ -76,8 +77,6 @@ def profile(self) -> str:
     @property
     def thing_name(self) -> str:
         return remove_prefix(self.resource_id, "thing/")
-
-
 #
 # ------ v1 configuration parse ------ #
 #
@@ -117,8 +116,6 @@ def parse_v1_config(_raw_cfg: str) -> IoTSessionConfig:
         region=thing_arn.region,
         aws_credential_provider_endpoint=str(this_profile_info.credential_endpoint),
     )
-
-
 #
 # ------ v2 configuration parse ------ #
 #
@@ -137,7 +134,6 @@ def parse_v2_config(_raw_cfg: str) -> IoTSessionConfig:
     this_profile_info = profile_info.get_profile_info(
         get_profile_from_thing_name(thing_name)
     )
-
     # NOTE(20240207): use credential endpoint defined in the config.yml in prior,
     #                 only when this information is not available, we use the
     #                 <_AWS_CREDENTIAL_PROVIDER_ENDPOINT_MAPPING> to get endpoint.
@@ -153,7 +149,6 @@ def parse_v2_config(_raw_cfg: str) -> IoTSessionConfig:
         cred_endpoint = _cred_endpoint
     else:
         cred_endpoint = this_profile_info.credential_endpoint
-
     # ------ parse pkcs11 config if any ------ #
     _raw_pkcs11_cfg: dict[str, str]
     pkcs11_cfg = None
@@ -188,13 +183,9 @@ def parse_v2_config(_raw_cfg: str) -> IoTSessionConfig:
         aws_credential_provider_endpoint=cred_endpoint,
         pkcs11_config=pkcs11_cfg,
     )
-
-
 #
 # ------ main config parser ------ #
 #
-
-
 class PKCS11Config(FixedConfig):
     """
     See services.aws.greengrass.crypto.Pkcs11Provider section for more details.
@@ -259,10 +250,12 @@ def parse_config() -> IoTSessionConfig:
         if (_v2_cfg_f := Path(server_cfg.GREENGRASS_V2_CONFIG)).is_file():
             _v2_cfg = parse_v2_config(_v2_cfg_f.read_text())
             logger.debug(f"gg config v2 is in used: {_v2_cfg}")
+            monitored_config_files.add(server_cfg.GREENGRASS_V2_CONFIG)
             return _v2_cfg
 
         _v1_cfg = parse_v1_config(Path(server_cfg.GREENGRASS_V1_CONFIG).read_text())
         logger.debug(f"gg config v1 is in used: {_v1_cfg}")
+        monitored_config_files.add(server_cfg.GREENGRASS_V1_CONFIG)
         return _v1_cfg
     except Exception as e:
         _msg = f"failed to parse config: {e!r}"

src/otaclient_iot_logging_server/log_proxy_server.py

@@ -25,7 +25,7 @@
 
 from otaclient_iot_logging_server._common import LogMessage, LogsQueue
 from otaclient_iot_logging_server.configs import server_cfg
-from otaclient_iot_logging_server.ecu_info import parse_ecu_info
+from otaclient_iot_logging_server.ecu_info import ecu_info
 
 logger = logging.getLogger(__name__)
 
@@ -37,11 +37,10 @@ def __init__(self, queue: LogsQueue) -> None:
         self._queue = queue
         self._allowed_ecus = None
 
-        stripped_ecu_info = parse_ecu_info(server_cfg.ECU_INFO_YAML)
-        if stripped_ecu_info:
-            self._allowed_ecus = stripped_ecu_info.ecu_id_set
+        if ecu_info:
+            self._allowed_ecus = ecu_info.ecu_id_set
             logger.info(
-                f"setup allowed_ecu_id from ecu_info.yaml: {stripped_ecu_info.ecu_id_set}"
+                f"setup allowed_ecu_id from ecu_info.yaml: {ecu_info.ecu_id_set}"
             )
         else:
             logger.warning(