Implement and standardize logging configuration via config file

RELEASE_NOTES.md

@@ -18,6 +18,10 @@
 ## New Features
 
 - The `ConfigManagingActor` can now take multiple configuration files as input, allowing to override default configurations with custom configurations.
+- Implement and standardize logging configuration with the following changes:
+   * Add LoggerConfig and LoggingConfig to standardize logging configuration.
+   * Create LoggingConfigUpdater to handle runtime config updates.
+   * Support individual log level settings for each module.
 
 ## Bug Fixes
 

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,194 @@
+# 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="NOTSET",
+        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."""
+
+    root_logger: 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.root_logger]
+        level = "INFO"
+
+        [logging.loggers."frequenz.sdk.actor.power_distributing"]
+        level = "DEBUG"
+
+        [logging.loggers."frequenz.channels"]
+        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._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._current_config: LoggingConfig = LoggingConfig()
+        self._update_logging(self._current_config)
+
+    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._current_config:
+                self._update_logging(new_config)
+
+    def _update_logging(self, config: LoggingConfig) -> None:
+        """Configure the logging level."""
+        # If the logger is not in the new config, set it to NOTSET
+        loggers_to_unset = self._current_config.loggers.keys() - config.loggers.keys()
+        for logger_id in loggers_to_unset:
+            _logger.debug("Unsetting log level for logger '%s'", logger_id)
+            logging.getLogger(logger_id).setLevel(logging.NOTSET)
+
+        self._current_config = config
+        logging.basicConfig(
+            format=self._format,
+            level=self._current_config.root_logger.level,
+            datefmt=self._datefmt,
+        )
+
+        # For each logger in the new config, set the log level
+        for logger_id, logger_config in self._current_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._current_config)

tests/config/test_logging_config_updater.py

@@ -0,0 +1,149 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for logging config updater."""
+
+import asyncio
+import logging
+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 = {
+        "root_logger": {"level": "DEBUG"},
+        "loggers": {
+            "frequenz.sdk.actor": {"level": "INFO"},
+            "frequenz.sdk.timeseries": {"level": "ERROR"},
+        },
+    }
+    config = LoggingConfig(
+        root_logger=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 = {"root_logger": {"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)
+
+
+@pytest.fixture
+def cleanup_logs() -> Any:
+    """Reset logging to default.
+
+    Python doesn't cleanup logging configuration after tests, so we need to do it manually.
+    """
+    yield
+
+    logging.getLogger("frequenz.sdk.actor").setLevel(logging.NOTSET)
+    logging.getLogger("frequenz.sdk.timeseries").setLevel(logging.NOTSET)
+
+
+async def test_logging_config_updater_actor(
+    mocker: MockerFixture,
+    cleanup_logs: Any,
+) -> None:
+    """Test if logging is configured and updated correctly."""
+    # Mock method that sets root level logging.
+    # Python doesn't cleanup logging configuration after tests.
+    # Overriding logging.basicConfig would mess up other tests, so we mock it.
+    # This is just for extra safety because changing root logging level in unit tests
+    # is not working anyway - python ignores it.
+    mocker.patch("frequenz.sdk.config._logging_config_updater.logging.basicConfig")
+
+    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:
+        assert logging.getLogger("frequenz.sdk.actor").level == logging.NOTSET
+        assert logging.getLogger("frequenz.sdk.timeseries").level == logging.NOTSET
+
+        update_logging_spy = mocker.spy(actor, "_update_logging")
+
+        # Send first config
+        await config_sender.send(
+            {
+                "logging": {
+                    "root_logger": {"level": "ERROR"},
+                    "loggers": {
+                        "frequenz.sdk.actor": {"level": "DEBUG"},
+                        "frequenz.sdk.timeseries": {"level": "ERROR"},
+                    },
+                }
+            }
+        )
+        await asyncio.sleep(0.01)
+        update_logging_spy.assert_called_once_with(
+            LoggingConfig(
+                root_logger=LoggerConfig(level="ERROR"),
+                loggers={
+                    "frequenz.sdk.actor": LoggerConfig(level="DEBUG"),
+                    "frequenz.sdk.timeseries": LoggerConfig(level="ERROR"),
+                },
+            )
+        )
+        assert logging.getLogger("frequenz.sdk.actor").level == logging.DEBUG
+        assert logging.getLogger("frequenz.sdk.timeseries").level == logging.ERROR
+        update_logging_spy.reset_mock()
+
+        # Update config
+        await config_sender.send(
+            {
+                "logging": {
+                    "root_logger": {"level": "WARNING"},
+                    "loggers": {
+                        "frequenz.sdk.actor": {"level": "INFO"},
+                    },
+                }
+            }
+        )
+        await asyncio.sleep(0.01)
+        expected_config = LoggingConfig(
+            root_logger=LoggerConfig(level="WARNING"),
+            loggers={
+                "frequenz.sdk.actor": LoggerConfig(level="INFO"),
+            },
+        )
+        update_logging_spy.assert_called_once_with(expected_config)
+        assert logging.getLogger("frequenz.sdk.actor").level == logging.INFO
+        assert logging.getLogger("frequenz.sdk.timeseries").level == logging.NOTSET
+        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": {"root_logger": {"level": "UNKNOWN"}}})
+        await asyncio.sleep(0.01)
+        update_logging_spy.assert_not_called()
+        assert actor._current_config == expected_config
+        update_logging_spy.reset_mock()
+
+        # Send empty config to reset logging to default
+        await config_sender.send({"other": {"var1": 1}})
+        await asyncio.sleep(0.01)
+        update_logging_spy.assert_called_once_with(LoggingConfig())
+        assert logging.getLogger("frequenz.sdk.actor").level == logging.NOTSET
+        assert logging.getLogger("frequenz.sdk.timeseries").level == logging.NOTSET
+        update_logging_spy.reset_mock()