Mutltiple Fixes

src/gallia/cli/gallia.py

@@ -21,15 +21,12 @@
 from gallia.command.base import BaseCommandConfig
 from gallia.command.config import GalliaBaseModel
 from gallia.config import Config, load_config_file
-from gallia.log import Loglevel, setup_logging
+from gallia.log import setup_logging
 from gallia.plugins.plugin import CommandTree, load_commands, load_plugins
 from gallia.pydantic_argparse import ArgumentParser
 from gallia.pydantic_argparse import BaseCommand as PydanticBaseCommand
 from gallia.utils import get_log_level
 
-setup_logging(Loglevel.DEBUG)
-
-
 defaults = dict[type, dict[str, Any]]
 _CLASS_ATTR = "_dynamic_gallia_command_class_reference"
 
@@ -183,6 +180,7 @@ def __call__(
         setup_logging(
             level=get_log_level(config.verbose),
             no_volatile_info=not config.volatile_info,
+            logger_name="",  # Take over the root logger
         )
 
     sys.exit(get_command(config).entry_point())

src/gallia/command/base.py

@@ -14,7 +14,6 @@
 from collections.abc import MutableMapping
 from datetime import UTC, datetime
 from enum import Enum, unique
-from logging import Handler
 from pathlib import Path
 from subprocess import CalledProcessError, run
 from tempfile import gettempdir
@@ -26,7 +25,7 @@
 from gallia.command.config import Field, GalliaBaseModel, Idempotent
 from gallia.db.handler import DBHandler
 from gallia.dumpcap import Dumpcap
-from gallia.log import add_zst_log_handler, get_logger, tz
+from gallia.log import _ZstdFileHandler, add_zst_log_handler, get_logger, remove_zst_log_handler, tz
 from gallia.power_supply import PowerSupply
 from gallia.power_supply.uri import PowerSupplyURI
 from gallia.services.uds.core.exception import UDSException
@@ -114,7 +113,11 @@ def _release_flock(self) -> None:
 class BaseCommandConfig(GalliaBaseModel, cli_group="generic", config_section="gallia"):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    verbose: int = Field(0, description="increase verbosity on the console", short="v")
+    verbose: int = Field(
+        0,
+        description="Increase verbosity of the console log (0: INFO, 1: DEBUG, 2: TRACE)",
+        short="v",
+    )
     volatile_info: bool = Field(
         True, description="Overwrite log lines with level info or lower in terminal output"
     )
@@ -180,7 +183,7 @@ class BaseCommand(FlockMixin, ABC):
     #: a log message with level critical is logged.
     CATCHED_EXCEPTIONS: list[type[Exception]] = []
 
-    log_file_handlers: list[Handler]
+    log_file_handlers: list[_ZstdFileHandler]
 
     def __init__(self, config: BaseCommandConfig) -> None:
         self.id = camel_to_snake(self.__class__.__name__)
@@ -374,6 +377,15 @@ def entry_point(self) -> int:
                 )
                 logger.notice(f"Stored artifacts at {self.artifacts_dir}")
 
+            # Close open log file handlers to ensure logs are properly written
+            # to avoid memory leaks and cross-talking log files
+            logger.info("Syncing log files…")
+            while len(self.log_file_handlers) > 0:
+                remove_zst_log_handler(
+                    logger_name="gallia",
+                    handler=self.log_file_handlers.pop(),
+                )
+
         if self.config.hooks:
             self.run_hook(HookVariant.POST, exit_code)
 

src/gallia/commands/scan/uds/identifiers.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import reprlib
+import sys
 from itertools import product
 
 from gallia.command import UDSScanner
@@ -63,14 +64,16 @@ def __init__(self, config: ScanIdentifiersConfig):
     async def main(self) -> None:
         if self.config.sessions is None:
             logger.notice("Performing scan in current session")
-            await self.perform_scan()
+            if not await self.perform_scan():
+                sys.exit(1)
         else:
             sessions: list[int] = [
                 s
                 for s in self.config.sessions
                 if s not in self.config.skip or self.config.skip[s] is not None
             ]
             logger.info(f"testing sessions {g_repr(sessions)}")
+            clean_returns = True
 
             # TODO: Unified shortened output necessary here
             logger.info(f"skipping identifiers {reprlib.repr(self.config.skip)}")
@@ -84,13 +87,16 @@ async def main(self) -> None:
 
                 logger.result(f"Starting scan in session: {g_repr(session)}")
 
-                await self.perform_scan(session)
+                clean_returns = clean_returns and await self.perform_scan(session)
 
                 logger.result(f"Scan in session {g_repr(session)} is complete!")
                 logger.info(f"Leaving session {g_repr(session)} via hook")
                 await self.ecu.leave_session(session, sleep=self.config.power_cycle_sleep)
 
-    async def perform_scan(self, session: None | int = None) -> None:
+            if not clean_returns:
+                sys.exit(1)
+
+    async def perform_scan(self, session: None | int = None) -> bool:
         positive_DIDs = 0
         abnormal_DIDs = 0
         timeout_DIDs = 0
@@ -132,7 +138,7 @@ async def perform_scan(self, session: None | int = None) -> None:
                     logger.error(
                         f"Aborting scan on session {g_repr(session)}; current DID was {g_repr(DID)}"
                     )
-                    break
+                    return False
 
             if self.config.service == UDSIsoServices.SecurityAccess:
                 if DID & 128:
@@ -191,3 +197,5 @@ async def perform_scan(self, session: None | int = None) -> None:
         logger.result(f"Positive replies: {positive_DIDs}")
         logger.result(f"Abnormal replies: {abnormal_DIDs}")
         logger.result(f"Timeouts: {timeout_DIDs}")
+
+        return True

src/gallia/commands/scan/uds/reset.py

@@ -44,10 +44,12 @@ def __init__(self, config: ResetScannerConfig):
 
     async def main(self) -> None:
         if self.config.sessions is None:
-            await self.perform_scan()
+            if not await self.perform_scan():
+                sys.exit(1)
         else:
             sessions = self.config.sessions
             logger.info(f"testing sessions {g_repr(sessions)}")
+            clean_returns = True
 
             # TODO: Unified shortened output necessary here
             logger.info(f"skipping identifiers {reprlib.repr(self.config.skip)}")
@@ -60,11 +62,14 @@ async def main(self) -> None:
                     continue
 
                 logger.result(f"Scanning in session: {g_repr(session)}")
-                await self.perform_scan(session)
+                clean_returns = clean_returns and await self.perform_scan(session)
 
                 await self.ecu.leave_session(session, sleep=self.config.power_cycle_sleep)
 
-    async def perform_scan(self, session: None | int = None) -> None:
+            if not clean_returns:
+                sys.exit(1)
+
+    async def perform_scan(self, session: None | int = None) -> bool:
         l_ok: list[int] = []
         l_timeout: list[int] = []
         l_error: list[Any] = []
@@ -82,7 +87,7 @@ async def perform_scan(self, session: None | int = None) -> None:
                     logger.error(
                         f"Aborting scan on session {g_repr(session)}; current sub-func was {g_repr(sub_func)}"
                     )
-                    break
+                    return False
 
             try:
                 try:
@@ -153,3 +158,5 @@ async def perform_scan(self, session: None | int = None) -> None:
         logger.result(f"ok: {l_ok}")
         logger.result(f"timeout: {l_timeout}")
         logger.result(f"with error: {l_error}")
+
+        return True

src/gallia/commands/scan/uds/services.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import reprlib
+import sys
 from typing import Any
 
 from gallia.command import UDSScanner
@@ -56,9 +57,10 @@ def __init__(self, config: ServicesScannerConfig):
     async def main(self) -> None:
         self.ecu.max_retry = 0
         found: dict[int, dict[int, Any]] = {}
+        clean_returns = True
 
         if self.config.sessions is None:
-            found[0] = await self.perform_scan()
+            found[0], clean_returns = await self.perform_scan()
         else:
             sessions = [
                 s
@@ -89,7 +91,8 @@ async def main(self) -> None:
 
                 logger.result(f"scanning in session {g_repr(session)}")
 
-                found[session] = await self.perform_scan(session)
+                found[session], ret = await self.perform_scan(session)
+                clean_returns = clean_returns and ret
 
                 await self.ecu.leave_session(session, sleep=self.config.power_cycle_sleep)
 
@@ -102,7 +105,10 @@ async def main(self) -> None:
                 except Exception:
                     logger.result(f"  [{g_repr(sid)}] vendor specific sid: {data}")
 
-    async def perform_scan(self, session: None | int = None) -> dict[int, Any]:
+        if not clean_returns:
+            sys.exit(1)
+
+    async def perform_scan(self, session: None | int = None) -> tuple[dict[int, Any], bool]:
         result: dict[int, Any] = {}
 
         # Starts at 0x00, see first loop iteration.
@@ -123,7 +129,7 @@ async def perform_scan(self, session: None | int = None) -> dict[int, Any]:
                     logger.error(
                         f"Aborting scan on session {g_repr(session)}; current SID was {g_repr(sid)}"
                     )
-                    break
+                    return result, False
 
             for length_payload in [1, 2, 3, 5]:
                 pdu = bytes([sid]) + bytes(length_payload)
@@ -152,4 +158,4 @@ async def perform_scan(self, session: None | int = None) -> dict[int, Any]:
                 result[sid] = resp
                 break
 
-        return result
+        return result, True

src/gallia/log.py

@@ -2,6 +2,8 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+from __future__ import annotations
+
 import atexit
 import dataclasses
 import datetime
@@ -147,7 +149,7 @@ class PenlogPriority(IntEnum):
     TRACE = 8
 
     @classmethod
-    def from_str(cls, string: str) -> "PenlogPriority":
+    def from_str(cls, string: str) -> PenlogPriority:
         """Converts a string to an instance of PenlogPriority.
         ``string`` can be a numeric value (0 to 8 inclusive)
         or a string with a case insensitive name of the level
@@ -179,7 +181,7 @@ def from_str(cls, string: str) -> "PenlogPriority":
                 raise ValueError(f"{string} not a valid priority")
 
     @classmethod
-    def from_level(cls, value: int) -> "PenlogPriority":
+    def from_level(cls, value: int) -> PenlogPriority:
         """Converts an int value (e.g. from python's logging module)
         to an instance of this class.
         """
@@ -299,13 +301,15 @@ def add_stderr_log_handler(
 
 def add_zst_log_handler(
     logger_name: str, filepath: Path, file_log_level: Loglevel
-) -> logging.Handler:
+) -> _ZstdFileHandler:
     queue: Queue[Any] = Queue()
     logger = get_logger(logger_name)
-    logger.addHandler(QueueHandler(queue))
+    qh = QueueHandler(queue)
+    logger.addHandler(qh)
 
     zstd_handler = _ZstdFileHandler(
         filepath,
+        queue_handler=qh,
         level=file_log_level,
     )
     zstd_handler.setLevel(file_log_level)
@@ -317,10 +321,18 @@ def add_zst_log_handler(
         respect_handler_level=True,
     )
     queue_listener.start()
-    atexit.register(queue_listener.stop)
+    zstd_handler.queue_listener = queue_listener
     return zstd_handler
 
 
+def remove_zst_log_handler(logger_name: str, handler: _ZstdFileHandler) -> None:
+    """This function removes the handler from the specified logger and closes it"""
+    logger = get_logger(logger_name)
+    # It is important to remove the Handler, otherwise it would still receive log messages
+    logger.removeHandler(handler.queue_handler)
+    handler.close()
+
+
 @dataclasses.dataclass
 class _PenlogRecordV2:
     module: str
@@ -730,7 +742,9 @@ def format(
 
 
 class _ZstdFileHandler(logging.Handler):
-    def __init__(self, path: Path, level: int | str = logging.NOTSET) -> None:
+    def __init__(
+        self, path: Path, queue_handler: QueueHandler, level: int | str = logging.NOTSET
+    ) -> None:
         super().__init__(level)
         self.file = zstandard.open(
             filename=path,
@@ -741,8 +755,15 @@ def __init__(self, path: Path, level: int | str = logging.NOTSET) -> None:
                 threads=-1,
             ),
         )
+        self.queue_handler = queue_handler
+        self.queue_listener: QueueListener | None = None
 
     def close(self) -> None:
+        """This function closes the queue handler, the queue listener, and the log file."""
+        self.queue_handler.close()
+        # There might be no queue_listener or it might already be closed (_thread is None)
+        if self.queue_listener is not None and self.queue_listener._thread is not None:
+            self.queue_listener.stop()
         self.file.flush()
         self.file.close()
 

src/gallia/utils.py

@@ -219,13 +219,12 @@ def dump_args(args: Any) -> dict[str, str | int | float]:
     return settings
 
 
-def get_log_level(args: Any) -> Loglevel:
+def get_log_level(cli_level: int) -> Loglevel:
     level = Loglevel.INFO
-    if hasattr(args, "verbose"):
-        if args.verbose == 1:
-            level = Loglevel.DEBUG
-        elif args.verbose >= 2:
-            level = Loglevel.TRACE
+    if cli_level == 1:
+        level = Loglevel.DEBUG
+    elif cli_level == 2:
+        level = Loglevel.TRACE
     return level
 
 
@@ -267,8 +266,8 @@ def handle_task_error(fut: asyncio.Future[Any]) -> None:
     except BaseException as e:
         task_name = task_name if task_name is not None else "Task"
 
-        # Info level is enough, since our aim is only to consume the stack trace
-        logger.info(f"{task_name} ended with error: {e!r}")
+        # Debug level is enough, since our aim is only to consume the stack trace
+        logger.debug(f"{task_name} ended with error: {e!r}")
 
 
 P = ParamSpec("P")