feat: split and run grpc_server, ota_core and otaclient main as separated processes

src/otaclient/_otaproxy_ctx.py

@@ -0,0 +1,138 @@
+# 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.
+"""Control of the otaproxy server startup/shutdown.
+
+The API exposed by this module is meant to be controlled by otaproxy managing thread only.
+"""
+
+
+from __future__ import annotations
+
+import asyncio
+import atexit
+import logging
+import multiprocessing as mp
+import multiprocessing.context as mp_ctx
+import shutil
+import time
+from functools import partial
+from pathlib import Path
+
+from ota_proxy import config as local_otaproxy_cfg
+from ota_proxy import run_otaproxy
+from ota_proxy.config import config as otaproxy_cfg
+from otaclient._types import MultipleECUStatusFlags
+from otaclient.configs.cfg import cfg, proxy_info
+from otaclient_common.common import ensure_otaproxy_start
+
+logger = logging.getLogger(__name__)
+
+_otaproxy_p: mp_ctx.SpawnProcess | None = None
+
+
+def shutdown_otaproxy_server() -> None:
+    global _otaproxy_p
+    if _otaproxy_p:
+        _otaproxy_p.terminate()
+        _otaproxy_p.join()
+        _otaproxy_p = None
+
+
+OTAPROXY_CHECK_INTERVAL = 3
+OTAPROXY_MIN_STARTUP_TIME = 120
+"""Keep otaproxy running at least 60 seconds after startup."""
+OTA_CACHE_DIR_CHECK_INTERVAL = 60
+
+
+def otaproxy_process(*, init_cache: bool) -> None:
+    from otaclient._logging import configure_logging
+
+    configure_logging()
+    logger.info("otaproxy process started")
+
+    external_cache_mnt_point = None
+    if cfg.OTAPROXY_ENABLE_EXTERNAL_CACHE:
+        external_cache_mnt_point = cfg.EXTERNAL_CACHE_DEV_MOUNTPOINT
+
+    host, port = (
+        str(proxy_info.local_ota_proxy_listen_addr),
+        proxy_info.local_ota_proxy_listen_port,
+    )
+
+    upper_proxy = str(proxy_info.upper_ota_proxy or "")
+    logger.info(f"will launch otaproxy at http://{host}:{port}, with {upper_proxy=}")
+    if upper_proxy:
+        logger.info(f"wait for {upper_proxy=} online...")
+        ensure_otaproxy_start(str(upper_proxy))
+
+    asyncio.run(
+        run_otaproxy(
+            host=host,
+            port=port,
+            init_cache=init_cache,
+            cache_dir=local_otaproxy_cfg.BASE_DIR,
+            cache_db_f=local_otaproxy_cfg.DB_FILE,
+            upper_proxy=upper_proxy,
+            enable_cache=proxy_info.enable_local_ota_proxy_cache,
+            enable_https=proxy_info.gateway_otaproxy,
+            external_cache_mnt_point=external_cache_mnt_point,
+        )
+    )
+
+
+def otaproxy_control_thread(
+    ecu_status_flags: MultipleECUStatusFlags,
+) -> None:  # pragma: no cover
+    atexit.register(shutdown_otaproxy_server)
+
+    _mp_ctx = mp.get_context("spawn")
+
+    ota_cache_dir = Path(otaproxy_cfg.BASE_DIR)
+    next_ota_cache_dir_checkpoint = 0
+
+    global _otaproxy_p
+    while True:
+        time.sleep(OTAPROXY_CHECK_INTERVAL)
+        _now = time.time()
+
+        _otaproxy_running = _otaproxy_p and _otaproxy_p.is_alive()
+        _otaproxy_should_run = ecu_status_flags.any_requires_network.is_set()
+        _all_success = ecu_status_flags.all_success.is_set()
+
+        if not _otaproxy_should_run and not _otaproxy_running:
+            if (
+                _now > next_ota_cache_dir_checkpoint
+                and _all_success
+                and ota_cache_dir.is_dir()
+            ):
+                logger.info(
+                    "all tracked ECUs are in SUCCESS OTA status, cleanup ota cache dir ..."
+                )
+                next_ota_cache_dir_checkpoint = _now + OTA_CACHE_DIR_CHECK_INTERVAL
+                shutil.rmtree(ota_cache_dir, ignore_errors=True)
+
+        elif _otaproxy_should_run and not _otaproxy_running:
+            # NOTE: always try to re-use cache. If the cache dir is empty, otaproxy
+            #   will still init the cache even init_cache is False.
+            _otaproxy_p = _mp_ctx.Process(
+                target=partial(otaproxy_process, init_cache=False),
+                name="otaproxy",
+            )
+            _otaproxy_p.start()
+            next_ota_cache_dir_checkpoint = _now + OTAPROXY_MIN_STARTUP_TIME
+            time.sleep(OTAPROXY_MIN_STARTUP_TIME)  # prevent pre-mature shutdown
+
+        elif _otaproxy_p and _otaproxy_running and not _otaproxy_should_run:
+            logger.info("shutting down otaproxy as not needed now ...")
+            shutdown_otaproxy_server()

src/otaclient/_status_monitor.py

@@ -23,7 +23,7 @@
 from dataclasses import asdict, dataclass
 from enum import Enum, auto
 from threading import Thread
-from typing import Union, cast
+from typing import Literal, Union, cast
 
 from otaclient._types import (
     FailureType,
@@ -34,17 +34,25 @@
     UpdateProgress,
     UpdateTiming,
 )
+from otaclient._utils import SharedOTAClientStatusWriter
+from otaclient_common.logging import BurstSuppressFilter
 
 logger = logging.getLogger(__name__)
+burst_suppressed_logger = logging.getLogger(f"{__name__}.shm_push")
+# NOTE: for request_error, only allow max 6 lines of logging per 30 seconds
+burst_suppressed_logger.addFilter(
+    BurstSuppressFilter(
+        f"{__name__}.shm_push",
+        upper_logger_name=__name__,
+        burst_round_length=30,
+        burst_max=6,
+    )
+)
 
-_otaclient_shutdown = False
 _status_report_queue: queue.Queue | None = None
 
 
 def _global_shutdown():
-    global _otaclient_shutdown
-    _otaclient_shutdown = True
-
     if _status_report_queue:
         _status_report_queue.put_nowait(TERMINATE_SENTINEL)
 
@@ -120,7 +128,7 @@ class StatusReport:
 #
 def _on_session_finished(
     status_storage: OTAClientStatus, payload: OTAStatusChangeReport
-):
+) -> Literal[True]:
     status_storage.session_id = ""
     status_storage.update_phase = UpdatePhase.INITIALIZING
     status_storage.update_meta = UpdateMeta()
@@ -137,10 +145,12 @@ def _on_session_finished(
         status_storage.failure_reason = ""
         status_storage.failure_traceback = ""
 
+    return True
+
 
 def _on_new_ota_session(
     status_storage: OTAClientStatus, payload: OTAStatusChangeReport
-):
+) -> Literal[True]:
     status_storage.ota_status = payload.new_ota_status
     status_storage.update_phase = UpdatePhase.INITIALIZING
     status_storage.update_meta = UpdateMeta()
@@ -149,6 +159,8 @@ def _on_new_ota_session(
     status_storage.failure_type = FailureType.NO_FAILURE
     status_storage.failure_reason = ""
 
+    return True
+
 
 def _on_update_phase_changed(
     status_storage: OTAClientStatus, payload: OTAUpdatePhaseChangeReport
@@ -157,7 +169,7 @@ def _on_update_phase_changed(
         logger.warning(
             "attempt to update update_timing when no OTA update session on-going"
         )
-        return
+        return False
 
     phase, trigger_timestamp = payload.new_update_phase, payload.trigger_timestamp
     if phase == UpdatePhase.PROCESSING_POSTUPDATE:
@@ -170,14 +182,17 @@ def _on_update_phase_changed(
         update_timing.update_apply_start_timestamp = trigger_timestamp
 
     status_storage.update_phase = phase
+    return True
 
 
-def _on_update_progress(status_storage: OTAClientStatus, payload: UpdateProgressReport):
+def _on_update_progress(
+    status_storage: OTAClientStatus, payload: UpdateProgressReport
+) -> bool:
     if (update_progress := status_storage.update_progress) is None:
         logger.warning(
             "attempt to update update_progress when no OTA update session on-going"
         )
-        return
+        return False
 
     op = payload.operation
     if (
@@ -195,6 +210,7 @@ def _on_update_progress(status_storage: OTAClientStatus, payload: UpdateProgress
         update_progress.downloading_errors += payload.errors
     elif op == UpdateProgressReport.Type.APPLY_REMOVE_DELTA:
         update_progress.removed_files_num += payload.processed_file_num
+    return True
 
 
 def _on_update_meta(status_storage: OTAClientStatus, payload: SetUpdateMetaReport):
@@ -204,7 +220,7 @@ def _on_update_meta(status_storage: OTAClientStatus, payload: SetUpdateMetaRepor
         logger.warning(
             "attempt to update update_meta when no OTA update session on-going"
         )
-        return
+        return False
 
     _input = asdict(payload)
     for k, v in _input.items():
@@ -213,31 +229,44 @@ def _on_update_meta(status_storage: OTAClientStatus, payload: SetUpdateMetaRepor
             continue
         if v:
             setattr(update_meta, k, v)
+    return True
 
 
 #
 # ------ status monitor implementation ------ #
 #
 
 TERMINATE_SENTINEL = cast(StatusReport, object())
+MIN_COLLECT_INTERVAL = 0.5
+SHM_PUSH_INTERVAL = 0.5
 
 
 class OTAClientStatusCollector:
+    """NOTE: status_monitor should only be started once during whole otaclient lifecycle!"""
 
     def __init__(
         self,
         msg_queue: queue.Queue[StatusReport],
+        shm_status: SharedOTAClientStatusWriter,
         *,
-        min_collect_interval: int = 1,
-        min_push_interval: int = 1,
+        min_collect_interval: float = MIN_COLLECT_INTERVAL,
+        shm_push_interval: float = SHM_PUSH_INTERVAL,
+        max_traceback_size: int,
     ) -> None:
+        self.max_traceback_size = max_traceback_size
         self.min_collect_interval = min_collect_interval
-        self.min_push_interval = min_push_interval
+        self.shm_push_interval = shm_push_interval
 
         self._input_queue = msg_queue
+        global _status_report_queue
+        _status_report_queue = msg_queue
+
         self._status = None
+        self._shm_status = shm_status
+
+        atexit.register(shm_status.atexit)
 
-    def load_report(self, report: StatusReport):
+    def load_report(self, report: StatusReport) -> bool:
         if self._status is None:
             self._status = OTAClientStatus()
         status_storage = self._status
@@ -246,37 +275,56 @@ def load_report(self, report: StatusReport):
         # ------ update otaclient meta ------ #
         if isinstance(payload, SetOTAClientMetaReport):
             status_storage.firmware_version = payload.firmware_version
+            return True
 
         # ------ on session start/end ------ #
         if isinstance(payload, OTAStatusChangeReport):
+            if (_traceback := payload.failure_traceback) and len(
+                _traceback
+            ) > self.max_traceback_size:
+                payload.failure_traceback = _traceback[-self.max_traceback_size :]
+
             new_ota_status = payload.new_ota_status
             if new_ota_status in [OTAStatus.UPDATING, OTAStatus.ROLLBACKING]:
                 status_storage.session_id = report.session_id
                 return _on_new_ota_session(status_storage, payload)
-
             status_storage.session_id = ""  # clear session if we are not in an OTA
             return _on_session_finished(status_storage, payload)
 
         # ------ during OTA session ------ #
         report_session_id = report.session_id
         if report_session_id != status_storage.session_id:
-            logger.warning(f"drop reports from mismatched session: {report}")
-            return  # drop invalid report
+            logger.warning(
+                f"drop reports from mismatched session (expect {status_storage.session_id=}): {report}"
+            )
+            return False
         if isinstance(payload, OTAUpdatePhaseChangeReport):
             return _on_update_phase_changed(status_storage, payload)
         if isinstance(payload, UpdateProgressReport):
             return _on_update_progress(status_storage, payload)
         if isinstance(payload, SetUpdateMetaReport):
             return _on_update_meta(status_storage, payload)
+        return False
 
     def _status_collector_thread(self) -> None:
         """Main entry of status monitor working thread."""
-        while not _otaclient_shutdown:
+        _next_shm_push = 0
+        while True:
+            _now = time.time()
             try:
                 report = self._input_queue.get_nowait()
                 if report is TERMINATE_SENTINEL:
                     break
-                self.load_report(report)
+
+                # ------ push status on load_report ------ #
+                if self.load_report(report) and self._status and _now > _next_shm_push:
+                    try:
+                        self._shm_status.write_msg(self._status)
+                        _next_shm_push = _now + self.shm_push_interval
+                    except Exception as e:
+                        burst_suppressed_logger.debug(
+                            f"failed to push status to shm: {e!r}"
+                        )
             except queue.Empty:
                 time.sleep(self.min_collect_interval)