Resolve #8: Add savant_rs based logging

* can't log configuration errors as the logger isn't configured yet, but if successful, the current config values are logged (api_key is scrapped)
* an option for log spec to be YAML dict in addition to just string, there are tests for this
* log to STDOUT as this is common for k8s
* termination signal handler for asyncio to shut down gracefully
* debug as a mininal logging level seems enough, no need for trace level right now

Author: abukharov268

.gitignore

@@ -186,7 +186,7 @@ cython_debug/
 #  that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
 #  and can be added to the global gitignore or merged into this file. However, if you prefer, 
 #  you could uncomment the following to ignore the entire vscode folder
-# .vscode/
+.vscode/
 
 # Ruff stuff:
 .ruff_cache/

savant_cloudpin/__main__.py

@@ -1,13 +1,25 @@
 import asyncio
 
-from savant_cloudpin.cfg import load_config
+from savant_rs.py.log import get_logger, init_logging
+
+from savant_cloudpin.cfg import SENSITIVE_KEYS, dump_to_yaml, load_config
 from savant_cloudpin.services import create_service
 
 
 async def serve() -> None:
     config = load_config()
+
+    init_logging(config.observability.log_spec)
+    logger = get_logger(__name__)
+
+    logger.info("Configuration loaded")
+    config_yaml = dump_to_yaml(config, scrap_keys=SENSITIVE_KEYS)
+    logger.debug(f"Configuration details:\n{config_yaml}")
+
+    logger.info("Running main loop ...")
     async with create_service(config) as service:
         await service.run()
+    logger.info("Main loop stopped")
 
 
 asyncio.run(serve())

savant_cloudpin/cfg/__init__.py

@@ -1,8 +1,10 @@
-from savant_cloudpin.cfg._bootstrap import load_config
+from savant_cloudpin.cfg._bootstrap import dump_to_yaml, load_config
 from savant_cloudpin.cfg._models import (
+    SENSITIVE_KEYS,
     ClientServiceConfig,
     ClientSSLConfig,
     ClientWSConfig,
+    ObservabilityConfig,
     ReaderConfig,
     ServerServiceConfig,
     ServerSSLConfig,
@@ -14,8 +16,11 @@
     "ClientServiceConfig",
     "ClientSSLConfig",
     "ClientWSConfig",
+    "dump_to_yaml",
     "load_config",
+    "ObservabilityConfig",
     "ReaderConfig",
+    "SENSITIVE_KEYS",
     "ServerServiceConfig",
     "ServerSSLConfig",
     "ServerWSConfig",

savant_cloudpin/cfg/_bootstrap.py

@@ -1,10 +1,10 @@
+import copy
 import os.path
 import re
+from collections.abc import Sequence
 from typing import Any
 
-from omegaconf import OmegaConf
-from omegaconf.base import SCMode
-from omegaconf.dictconfig import DictConfig
+from omegaconf import DictConfig, ListConfig, OmegaConf, SCMode
 
 from savant_cloudpin.cfg import _utils as utils
 from savant_cloudpin.cfg._defaults import (
@@ -62,15 +62,28 @@ def merge_env_config(
     return cfg
 
 
+def join_log_spec(cfg: DictConfig | ListConfig) -> None:
+    if not isinstance(cfg, DictConfig):
+        return
+    if "observability" not in cfg or "log_spec" not in cfg.observability:
+        return
+    if not isinstance(cfg.observability.log_spec, (DictConfig, dict)):
+        return
+    cfg.observability.log_spec = ",".join(
+        f"{k}={v}" for k, v in cfg.observability.log_spec.items()
+    )
+
+
 def load_config(
     args_list: list[str] | None = None,
 ) -> ClientServiceConfig | ServerServiceConfig:
     cli_cfg = OmegaConf.from_cli(args_list)
-    env_cfg = utils.as_value_dict(utils.env_override(DEFAULT_LOAD_CONFIG))
+    env_cfg = utils.env_override(DEFAULT_LOAD_CONFIG)
     cfg = OmegaConf.merge(env_cfg, cli_cfg)
 
     yml_exists = cfg.config and os.path.exists(cfg.config)
     yml_cfg = OmegaConf.load(cfg.config) if yml_exists else OmegaConf.create({})
+    join_log_spec(yml_cfg)
 
     cfg = OmegaConf.merge(yml_cfg, env_cfg, cli_cfg)
     assert isinstance(cfg, DictConfig) and isinstance(yml_cfg, DictConfig)
@@ -88,3 +101,18 @@ def load_config(
             return validated_dataclass(cfg, ClientServiceConfig)
         case _:
             raise ValueError("Invalid service mode")
+
+
+def dump_to_yaml(
+    config: ClientServiceConfig | ServerServiceConfig, scrap_keys: Sequence[str] = ()
+) -> str:
+    if scrap_keys:
+        config = copy.deepcopy(config)
+        utils.scrap_sensitive_keys(config, scrap_keys)
+
+    mode = "server" if isinstance(config, ServerServiceConfig) else "client"
+    summary = OmegaConf.to_container(
+        OmegaConf.structured(config), structured_config_mode=SCMode.DICT
+    )
+    summary = OmegaConf.merge(dict(mode=mode), summary)
+    return OmegaConf.to_yaml(summary)

savant_cloudpin/cfg/_defaults.py

@@ -2,27 +2,28 @@
     ClientServiceConfig,
     ClientSSLConfig,
     ClientWSConfig,
-    LoadConfig,
+    ObservabilityConfig,
     ReaderConfig,
     ServerServiceConfig,
     ServerSSLConfig,
     ServerWSConfig,
     WriterConfig,
 )
 
+DEFAULT_LOAD_CONFIG = dict(
+    config="./cloudpin.yml",
+    mode="client",
+)
+
 DEFAULT_SOURCE_CONFIG = ReaderConfig(
-    results_queue_size=100,
     url="???",
 )
 DEFAULT_SINK_CONFIG = WriterConfig(
-    max_inflight_messages=100,
     url="???",
 )
 
-
-DEFAULT_LOAD_CONFIG = LoadConfig(
-    config="./cloudpin.yml",
-    mode="client",
+DEFAULT_OBSERVABILITY_CONFIG = ObservabilityConfig(
+    log_spec="${oc.env:LOGLEVEL,warning}"
 )
 
 DEFAULT_CLIENT_CONFIG = ClientServiceConfig(
@@ -36,9 +37,9 @@
             check_hostname=False,
         ),
     ),
-    io_timeout=0.1,
     source=DEFAULT_SOURCE_CONFIG,
     sink=DEFAULT_SINK_CONFIG,
+    observability=DEFAULT_OBSERVABILITY_CONFIG,
 )
 
 DEFAULT_SERVER_CONFIG = ServerServiceConfig(
@@ -52,7 +53,7 @@
             client_cert_required=True,
         ),
     ),
-    io_timeout=0.1,
     source=DEFAULT_SOURCE_CONFIG,
     sink=DEFAULT_SINK_CONFIG,
+    observability=DEFAULT_OBSERVABILITY_CONFIG,
 )

savant_cloudpin/cfg/_models.py

@@ -1,15 +1,17 @@
-from dataclasses import dataclass, replace
-from typing import Self
+from dataclasses import dataclass, field, replace
+from typing import Protocol, Self
 
 from savant_rs import zmq
 
 from savant_cloudpin.cfg._utils import to_map_config
 
+SENSITIVE_KEYS = ("api_key",)
+
 
 @dataclass
 class ReaderConfig:
-    results_queue_size: int
     url: str
+    results_queue_size: int = 100
     receive_timeout: int | None = None
     receive_hwm: int | None = None
     topic_prefix_spec: str | None = None
@@ -29,8 +31,8 @@ def to_args(self) -> tuple[zmq.ReaderConfig, int]:
 
 @dataclass
 class WriterConfig:
-    max_inflight_messages: int
     url: str
+    max_inflight_messages: int = 100
     send_timeout: int | None = None
     send_retries: int | None = None
     send_hwm: int | None = None
@@ -81,23 +83,30 @@ class ClientWSConfig:
 
 
 @dataclass
-class BaseServiceConfig:
-    io_timeout: float
+class ObservabilityConfig:
+    log_spec: str | None = "warning"
+
+
+class BaseServiceConfig(Protocol):
     source: ReaderConfig
     sink: WriterConfig
+    io_timeout: float
+    observability: ObservabilityConfig
 
 
 @dataclass
 class ServerServiceConfig(BaseServiceConfig):
     websockets: ServerWSConfig
+    source: ReaderConfig
+    sink: WriterConfig
+    io_timeout: float = 0.1
+    observability: ObservabilityConfig = field(default_factory=ObservabilityConfig)
 
 
 @dataclass
 class ClientServiceConfig(BaseServiceConfig):
     websockets: ClientWSConfig
-
-
-@dataclass
-class LoadConfig:
-    config: str | None
-    mode: str | None
+    source: ReaderConfig
+    sink: WriterConfig
+    io_timeout: float = 0.1
+    observability: ObservabilityConfig = field(default_factory=ObservabilityConfig)

savant_cloudpin/cfg/_utils.py

@@ -1,7 +1,10 @@
 import dataclasses
-from collections.abc import Mapping
+import operator
+from collections.abc import Mapping, Sequence
 from dataclasses import asdict
-from typing import Any
+from typing import Any, cast
+
+from omegaconf import DictConfig
 
 ENV_PREFIX = "CLOUDPIN"
 
@@ -32,16 +35,21 @@ def as_value_dict(obj: Any) -> dict[str, Any]:
     return drop_none_values(dct)
 
 
-def env_override(
-    obj: Any,
+def env_override[T](
+    obj: T,
     default: str | None = None,
     prefix: str = ENV_PREFIX,
-) -> Any:
+) -> T:
     if isinstance(obj, type):
         raise ValueError("Instance is expected")
     updates = dict[str, Any]()
-    for field in dataclasses.fields(obj):
-        name, val = field.name, getattr(obj, field.name)
+    if dataclasses.is_dataclass(obj):
+        items = ((f.name, getattr(obj, f.name)) for f in dataclasses.fields(obj))
+    elif isinstance(obj, dict):
+        items = obj.items()
+    else:
+        raise TypeError("Unsupported type")
+    for name, val in items:
         env_name = f"{prefix}_{name.upper()}"
         match val:
             case str() | int() | float() if default is None:
@@ -57,4 +65,29 @@ def env_override(
             case _:
                 updates[name] = env_override(val, default, env_name)
 
+    if isinstance(obj, dict):
+        obj.update(**updates)
+        return cast(T, obj)
     return dataclasses.replace(obj, **updates)
+
+
+def scrap_sensitive_keys(obj: Any, sensitive_keys: Sequence[str]) -> None:
+    if dataclasses.is_dataclass(obj):
+        items = ((f.name, getattr(obj, f.name)) for f in dataclasses.fields(obj))
+        setkey = setattr
+    elif isinstance(obj, (dict, DictConfig)):
+        items = obj.items()
+        setkey = operator.setitem
+    else:
+        raise TypeError(f"Unsupported type {type(obj)}")
+
+    for key, val in items:
+        match val:
+            case str() if isinstance(key, str) and key in sensitive_keys:
+                setkey(obj, key, "*****")
+            case int() if isinstance(key, str) and key in sensitive_keys:
+                setkey(obj, key, 0)
+            case str() | int() | float() | bool() | None:
+                continue
+            case _:
+                scrap_sensitive_keys(val, sensitive_keys)

savant_cloudpin/services/__init__.py

@@ -1,3 +1,7 @@
+from typing import overload
+
+from savant_rs.py.log import get_logger
+
 from savant_cloudpin.cfg import ClientServiceConfig, ServerServiceConfig
 from savant_cloudpin.services._client import ClientService
 from savant_cloudpin.services._server import ServerService
@@ -8,11 +12,21 @@
     "ServerService",
 ]
 
+logger = get_logger(__package__ or __name__)
+
 
+@overload
+def create_service(config: ClientServiceConfig) -> ClientService: ...
+@overload
+def create_service(config: ServerServiceConfig) -> ServerService: ...
 def create_service(
     config: ClientServiceConfig | ServerServiceConfig,
 ) -> ClientService | ServerService:
-    if isinstance(config, ServerServiceConfig):
-        return ServerService(config)
-    else:
-        return ClientService(config)
+    try:
+        if isinstance(config, ServerServiceConfig):
+            return ServerService(config)
+        else:
+            return ClientService(config)
+    except:
+        logger.exception("Error service configuring")
+        raise