Add support to configure log level via config file.

Add marshmallow LoggerConfig and LoggingConfig classes
to load logging configuration.
Add LoggingConfigUpdater actor to listen for config updates
and configure the loggers.

Signed-off-by: Elzbieta Kotulska <[email protected]>

Author: ela-kotulska-frequenz

pyproject.toml

@@ -35,6 +35,8 @@ dependencies = [
   "networkx >= 2.8, < 4",
   "numpy >= 1.26.4, < 2",
   "typing_extensions >= 4.6.1, < 5",
+  "marshmallow >= 3.19.0, < 4",
+  "marshmallow_dataclass >= 8.7.1, < 9",
 ]
 dynamic = ["version"]
 

src/frequenz/sdk/config/__init__.py

@@ -4,5 +4,11 @@
 """Read and update config variables."""
 
 from ._config_managing import ConfigManagingActor
+from ._logging_config_updater import LoggerConfig, LoggingConfig, LoggingConfigUpdater
 
-__all__ = ["ConfigManagingActor"]
+__all__ = [
+    "ConfigManagingActor",
+    "LoggingConfig",
+    "LoggerConfig",
+    "LoggingConfigUpdater",
+]

src/frequenz/sdk/config/_logging_config_updater.py

@@ -0,0 +1,188 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Read and update logging severity from config."""
+
+import logging
+from collections.abc import Mapping
+from dataclasses import field
+from typing import Annotated, Any, Self, cast
+
+import marshmallow
+import marshmallow.validate
+from frequenz.channels import Receiver
+from marshmallow import RAISE
+from marshmallow_dataclass import class_schema, dataclass
+
+from frequenz.sdk.actor import Actor
+
+_logger = logging.getLogger(__name__)
+
+LogLevel = Annotated[
+    str,
+    marshmallow.fields.String(
+        validate=marshmallow.validate.OneOf(choices=logging.getLevelNamesMapping())
+    ),
+]
+
+
+@dataclass
+class LoggerConfig:
+    """A configuration for a logger."""
+
+    level: LogLevel = field(
+        default="INFO",
+        metadata={
+            "metadata": {
+                "description": "Log level for the logger. Uses standard logging levels."
+            },
+            "required": False,
+        },
+    )
+    """The log level for the logger."""
+
+
+@dataclass
+class LoggingConfig:
+    """A configuration for the logging system."""
+
+    default: LoggerConfig = field(
+        default_factory=LoggerConfig,
+        metadata={
+            "metadata": {
+                "description": "Default default configuration for all loggers.",
+            },
+            "required": False,
+        },
+    )
+    """The default log level."""
+
+    loggers: dict[str, LoggerConfig] = field(
+        default_factory=dict,
+        metadata={
+            "metadata": {
+                "description": "Configuration for a logger (the key is the logger name)."
+            },
+            "required": False,
+        },
+    )
+    """The list of loggers configurations."""
+
+    @classmethod
+    def load(cls, configs: Mapping[str, Any]) -> Self:  # noqa: DOC502
+        """Load and validate configs from a dictionary.
+
+        Args:
+            configs: The configuration to validate.
+
+        Returns:
+            The configuration if they are valid.
+
+        Raises:
+            ValidationError: if the configuration are invalid.
+        """
+        schema = class_schema(cls)()
+        return cast(Self, schema.load(configs, unknown=RAISE))
+
+
+class LoggingConfigUpdater(Actor):
+    """Actor that listens for logging configuration changes and sets them.
+
+    Example:
+        `config.toml` file:
+        ```toml
+        [logging.default]
+        level = "INFO"
+
+        [logging.loggers."frequenz.actor.pv_self_consumption"]
+        level = "DEBUG"
+
+        [logging.loggers."frequenz.actor.forecast"]
+        level = "DEBUG"
+        ```
+
+        ```python
+        import asyncio
+        from collections.abc import Mapping
+        from typing import Any
+
+        from frequenz.channels import Broadcast
+        from frequenz.sdk.config import LoggingConfigUpdater, ConfigManager
+        from frequenz.sdk.actor import run as run_actors
+
+        async def run() -> None:
+            config_channel = Broadcast[Mapping[str, Any]](name="config", resend_latest=True)
+            actors = [
+                ConfigManager(config_paths=["config.toml"], output=config_channel.new_sender()),
+                LoggingConfigUpdater(
+                    config_recv=config_channel.new_receiver(limit=1)).map(
+                        lambda app_config: app_config.get("logging", {}
+                    )
+                ),
+            ]
+            await run_actors(*actors)
+
+        asyncio.run(run())
+        ```
+
+        Now whenever the `config.toml` file is updated, the logging configuration
+        will be updated as well.
+    """
+
+    def __init__(
+        self,
+        config_recv: Receiver[Mapping[str, Any]],
+        log_format: str = "%(asctime)s %(levelname)-8s %(name)s:%(lineno)s: %(message)s",
+        log_datefmt: str = "%Y-%m-%dT%H:%M:%S%z",
+    ):
+        """Initialize this instance.
+
+        Args:
+            config_recv: The receiver to listen for configuration changes.
+            log_format: Use the specified format string in logs.
+            log_datefmt: Use the specified date/time format in logs.
+        """
+        super().__init__()
+        self._config_recv = config_recv
+        self._logging_config: LoggingConfig
+        self._format = log_format
+        self._datefmt = log_datefmt
+
+        # Setup default configuration.
+        # This ensures logging is configured even if actor fails to start or
+        # if the configuration cannot be loaded.
+        self._update_logging(LoggingConfig())
+
+    async def _run(self) -> None:
+        """Listen for configuration changes and update logging."""
+        async for message in self._config_recv:
+            try:
+                new_config = LoggingConfig.load(message)
+            except marshmallow.ValidationError:
+                _logger.exception(
+                    "Invalid logging configuration received. Skipping config update"
+                )
+                continue
+
+            if new_config != self._logging_config:
+                self._update_logging(new_config)
+
+    def _update_logging(self, config: LoggingConfig) -> None:
+        """Configure the logging level."""
+        self._logging_config = config
+
+        logging.basicConfig(
+            format=self._format,
+            level=self._logging_config.default.level,
+            datefmt=self._datefmt,
+        )
+
+        for logger_id, logger_config in self._logging_config.loggers.items():
+            _logger.debug(
+                "Setting log level for logger '%s' to '%s'",
+                logger_id,
+                logger_config.level,
+            )
+            logging.getLogger(logger_id).setLevel(logger_config.level)
+
+        _logger.info("Logging config changed to: %s", self._logging_config)

tests/config/test_logging_config_updater.py

@@ -0,0 +1,111 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for logging config updater."""
+
+import asyncio
+from collections.abc import Mapping
+from typing import Any
+
+import pytest
+from frequenz.channels import Broadcast
+from marshmallow import ValidationError
+from pytest_mock import MockerFixture
+
+from frequenz.sdk.config import LoggerConfig, LoggingConfig, LoggingConfigUpdater
+
+
+def test_logging_config() -> None:
+    """Test if logging config is correctly loaded."""
+    config_raw = {
+        "default": {"level": "DEBUG"},
+        "loggers": {
+            "frequenz.sdk.actor": {"level": "INFO"},
+            "frequenz.sdk.timeseries": {"level": "ERROR"},
+        },
+    }
+    config = LoggingConfig(
+        default=LoggerConfig(level="DEBUG"),
+        loggers={
+            "frequenz.sdk.actor": LoggerConfig(level="INFO"),
+            "frequenz.sdk.timeseries": LoggerConfig(level="ERROR"),
+        },
+    )
+
+    assert LoggingConfig.load(config_raw) == config
+
+    config_raw = {}
+    config = LoggingConfig()
+    assert LoggingConfig.load(config_raw) == config
+
+    config_raw = {"default": {"level": "UNKNOWN"}}
+    with pytest.raises(ValidationError):
+        LoggingConfig.load(config_raw)
+
+    config_raw = {"unknown": {"frequenz.sdk.actor": {"level": "DEBUG"}}}
+    with pytest.raises(ValidationError):
+        LoggingConfig.load(config_raw)
+
+
+async def test_logging_config_updater_actor(
+    mocker: MockerFixture,
+) -> None:
+    """Test if logging is configured and updated correctly."""
+    # Mock logging module in this file. Other logging won't be affected.
+    logging_mock = mocker.patch("frequenz.sdk.config._logging_config_updater.logging")
+
+    config_channel = Broadcast[Mapping[str, Any]](name="config")
+    config_sender = config_channel.new_sender()
+    async with LoggingConfigUpdater(
+        config_recv=config_channel.new_receiver().map(
+            lambda app_config: app_config.get("logging", {})
+        )
+    ) as actor:
+        # Check if default config has been set
+        logging_mock.basicConfig.assert_called_once()
+
+        update_logging_spy = mocker.spy(actor, "_update_logging")
+
+        # Send first config
+        await config_sender.send({"logging": {"default": {"level": "ERROR"}}})
+        await asyncio.sleep(0.01)
+        update_logging_spy.assert_called_once_with(
+            LoggingConfig(default=LoggerConfig(level="ERROR"))
+        )
+        update_logging_spy.reset_mock()
+
+        # Update config
+        await config_sender.send(
+            {
+                "logging": {
+                    "default": {"level": "WARNING"},
+                    "loggers": {
+                        "frequenz.sdk.actor": {"level": "INFO"},
+                        "frequenz.sdk.timeseries": {"level": "ERROR"},
+                    },
+                }
+            }
+        )
+        await asyncio.sleep(0.01)
+        expected_config = LoggingConfig(
+            default=LoggerConfig(level="WARNING"),
+            loggers={
+                "frequenz.sdk.actor": LoggerConfig(level="INFO"),
+                "frequenz.sdk.timeseries": LoggerConfig(level="ERROR"),
+            },
+        )
+        update_logging_spy.assert_called_once_with(expected_config)
+        update_logging_spy.reset_mock()
+
+        # Send invalid config to make sure actor doesn't crash and doesn't setup invalid config.
+        await config_sender.send({"logging": {"default": {"level": "UNKNOWN"}}})
+        await asyncio.sleep(0.01)
+        update_logging_spy.assert_not_called()
+        assert actor._logging_config == expected_config
+        update_logging_spy.reset_mock()
+
+        # Send empty config
+        await config_sender.send({"other": {"var1": 1}})
+        await asyncio.sleep(0.01)
+        update_logging_spy.assert_called_once_with(LoggingConfig())
+        update_logging_spy.reset_mock()