Make the LoggingConfigUpdatingActor use the ConfigManager

The `LoggingConfigUpdatingActor` can now also receive `None` as
configuration (for example if the configuration is removed), in which
case it will go back to the default configuration.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

RELEASE_NOTES.md

@@ -11,7 +11,9 @@
     * `LoggingConfigUpdater`
 
         + Renamed to `LoggingConfigUpdatingActor` to follow the actor naming convention.
-        + Make all arguments to the constructor keyword-only.
+        + The actor must now be constructed using a `ConfigManager` instead of a receiver.
+        + Make all arguments to the constructor keyword-only, except for the `config_manager` argument.
+        + If the configuration is removed, the actor will now load back the default configuration.
 
     * `LoggingConfig`
 

src/frequenz/sdk/config/_logging_actor.py

@@ -4,16 +4,14 @@
 """Read and update logging severity from config."""
 
 import logging
-from collections.abc import Mapping
 from dataclasses import dataclass, field
-from typing import Annotated, Any
+from typing import Annotated, Sequence, assert_never
 
 import marshmallow
 import marshmallow.validate
-from frequenz.channels import Receiver
 
 from ..actor import Actor
-from ._util import load_config
+from ._manager import ConfigManager
 
 _logger = logging.getLogger(__name__)
 
@@ -84,26 +82,13 @@ class LoggingConfigUpdatingActor(Actor):
 
         ```python
         import asyncio
-        from collections.abc import Mapping
-        from typing import Any
 
-        from frequenz.channels import Broadcast
-        from frequenz.sdk.config import LoggingConfigUpdatingActor, ConfigManagingActor
+        from frequenz.sdk.config import LoggingConfigUpdatingActor
         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 = [
-                ConfigManagingActor(
-                    config_paths=["config.toml"], output=config_channel.new_sender()
-                ),
-                LoggingConfigUpdatingActor(
-                    config_recv=config_channel.new_receiver(limit=1)).map(
-                        lambda app_config: app_config.get("logging", {}
-                    )
-                ),
-            ]
-            await run_actors(*actors)
+            config_manager: ConfigManager = ...
+            await run_actors(LoggingConfigUpdatingActor(config_manager))
 
         asyncio.run(run())
         ```
@@ -112,18 +97,25 @@ async def run() -> None:
         will be updated as well.
     """
 
+    # pylint: disable-next=too-many-arguments
     def __init__(
         self,
+        config_manager: ConfigManager,
+        /,
         *,
-        config_recv: Receiver[Mapping[str, Any]],
+        config_key: str | Sequence[str] = "logging",
         log_datefmt: str = "%Y-%m-%dT%H:%M:%S%z",
         log_format: str = "%(asctime)s %(levelname)-8s %(name)s:%(lineno)s: %(message)s",
         name: str | None = None,
     ):
         """Initialize this instance.
 
         Args:
-            config_recv: The receiver to listen for configuration changes.
+            config_manager: The configuration manager to use.  If `None`, the [global
+                configuration manager][frequenz.sdk.config.get_config_manager] will be
+                used.
+            config_key: The key to use to retrieve the configuration from the
+                configuration manager.  If `None`, the whole configuration will be used.
             log_datefmt: Use the specified date/time format in logs.
             log_format: Use the specified format string in logs.
             name: The name of this actor. If `None`, `str(id(self))` will be used. This
@@ -135,7 +127,9 @@ def __init__(
             in the application (through a previous `basicConfig()` call), then the format
             settings specified here will be ignored.
         """
-        self._config_recv = config_recv
+        self._config_receiver = config_manager.new_receiver(
+            config_key, LoggingConfig, base_schema=None
+        )
 
         # Setup default configuration.
         # This ensures logging is configured even if actor fails to start or
@@ -153,17 +147,19 @@ def __init__(
 
     async def _run(self) -> None:
         """Listen for configuration changes and update logging."""
-        async for message in self._config_recv:
-            try:
-                new_config = load_config(LoggingConfig, 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)
+        async for new_config in self._config_receiver:
+            match new_config:
+                case None:
+                    # When we receive None, we want to reset the logging configuration
+                    # to the default
+                    self._update_logging(LoggingConfig())
+                case LoggingConfig():
+                    self._update_logging(new_config)
+                case Exception():
+                    # We ignore errors and just keep the old configuration
+                    pass
+                case unexpected:
+                    assert_never(unexpected)
 
     def _update_logging(self, config: LoggingConfig) -> None:
         """Configure the logging level."""

tests/config/test_logging_actor.py

@@ -5,7 +5,6 @@
 
 import asyncio
 import logging
-from collections.abc import Mapping
 from typing import Any
 
 import pytest
@@ -77,78 +76,69 @@ async def test_logging_config_updating_actor(
     # is not working anyway - python ignores it.
     mocker.patch("frequenz.sdk.config._logging_actor.logging.basicConfig")
 
-    config_channel = Broadcast[Mapping[str, Any]](name="config")
-    config_sender = config_channel.new_sender()
-    async with LoggingConfigUpdatingActor(
-        config_recv=config_channel.new_receiver().map(
-            lambda app_config: app_config.get("logging", {})
-        )
-    ) as actor:
+    # Mock ConfigManager
+    mock_config_manager = mocker.Mock()
+    mock_config_manager.config_channel = Broadcast[LoggingConfig | Exception | None](
+        name="config"
+    )
+    mock_config_manager.new_receiver = mocker.Mock(
+        return_value=mock_config_manager.config_channel.new_receiver()
+    )
+
+    async with LoggingConfigUpdatingActor(mock_config_manager) 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"},
-                    },
-                }
-            }
+        expected_config = LoggingConfig(
+            root_logger=LoggerConfig(level="ERROR"),
+            loggers={
+                "frequenz.sdk.actor": LoggerConfig(level="DEBUG"),
+                "frequenz.sdk.timeseries": LoggerConfig(level="ERROR"),
+            },
         )
+        await mock_config_manager.config_channel.new_sender().send(expected_config)
         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"),
-                },
-            )
-        )
+        update_logging_spy.assert_called_once_with(expected_config)
         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"},
-                    },
-                }
-            }
+        # Send an exception and verify the previous config is maintained
+        await mock_config_manager.config_channel.new_sender().send(
+            ValueError("Test error")
         )
         await asyncio.sleep(0.01)
+        update_logging_spy.assert_not_called()  # Should not try to update logging
+        # Previous config should be maintained
+        assert logging.getLogger("frequenz.sdk.actor").level == logging.DEBUG
+        assert logging.getLogger("frequenz.sdk.timeseries").level == logging.ERROR
+        assert (
+            actor._current_config == expected_config  # pylint: disable=protected-access
+        )  # pylint: disable=protected-access
+        update_logging_spy.reset_mock()
+
+        # Update config
         expected_config = LoggingConfig(
             root_logger=LoggerConfig(level="WARNING"),
             loggers={
                 "frequenz.sdk.actor": LoggerConfig(level="INFO"),
             },
         )
+        await mock_config_manager.config_channel.new_sender().send(expected_config)
+        await asyncio.sleep(0.01)
         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}})
+        # Send a None config to make sure actor doesn't crash and configures a default logging
+        await mock_config_manager.config_channel.new_sender().send(None)
         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
+        assert (
+            actor._current_config == LoggingConfig()  # pylint: disable=protected-access
+        )
         update_logging_spy.reset_mock()