Fix unreachable-power formula leak in the power distributor (#1411)

The power distributor runs a formula for each set of unreachable
components. They were created on the shared logical-meter pool, which
never evicts unused formulas.

On teardown only the cache was stopped, not the formula — so a
formula-evaluating actor leaked per distinct set, and CPU climbed over
time until a restart.

## Changes

* `FormulaPool`: `stop()` now also stops string formulas; new
`stop_string_formula()` stops + evicts a single one.
* Each `ComponentManager` lazily owns a dedicated `FormulaPool` (new
`_DataPipeline.new_formula_pool`) and stops its formulas on teardown,
via a per-type `_unreachable_power_formula` hook. Battery and PV use it.
* Fix `from_reactive_power_formula` checking the wrong cache dict, which
re-created and leaked the grid reactive-power formula on every access.

## Breaking

* PV distribution now subtracts the power of unreachable PV inverters
from the target, so PV results differ when some inverters are
unreachable.

Author: shsms

RELEASE_NOTES.md

@@ -7,11 +7,13 @@
 ## Upgrading
 
 * Update microgrid client to v0.18.3+, which fixes a problem with missing steam boilers on formula generation.
+* The PV inverter manager now subtracts the power measured on unreachable PV inverters from the distribution target, so the power sent to the reachable inverters can change when some requested PV inverters are unreachable.
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* The PV inverter manager now accounts for the power measured on unreachable PV inverters when distributing power, so the reachable inverters compensate for it (matching the battery manager's behavior).
 
 ## Bug Fixes
 
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+* Fixed a resource leak in the power distributor: the formulas created for unreachable batteries were never stopped, so CPU usage slowly climbed over time until the application was restarted.
+* Fixed the grid reactive-power formula being recreated and leaked on every access instead of being reused from the cache.

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -12,6 +12,7 @@
 
 import logging
 import typing
+import uuid
 from collections import abc
 from dataclasses import dataclass
 from datetime import timedelta
@@ -28,6 +29,9 @@
 from ._power_managing._base_classes import DefaultPower, PowerManagerAlgorithm
 from ._power_wrapper import PowerWrapper
 
+if typing.TYPE_CHECKING:
+    from ..timeseries.formulas._formula_pool import FormulaPool
+
 # A number of imports had to be done inside functions where they are used, to break
 # import cycles.
 #
@@ -184,6 +188,31 @@ def logical_meter(self) -> LogicalMeter:
             )
         return self._logical_meter
 
+    def new_formula_pool(self, namespace: str) -> FormulaPool:
+        """Create a new, empty formula pool.
+
+        The caller owns the returned pool and is responsible for stopping it.
+
+        Args:
+            namespace: A label describing what the pool is for; a unique suffix is
+                appended so its resampler subscriptions don't collide with other
+                pools that share the channel registry.
+
+        Returns:
+            A new, empty formula pool.
+        """
+        # Imported here (not at module level) to avoid an import cycle, mirroring
+        # `logical_meter()` above.
+        from ..timeseries.formulas._formula_pool import (  # pylint: disable=import-outside-toplevel
+            FormulaPool,
+        )
+
+        return FormulaPool(
+            f"{namespace}-{uuid.uuid4()}",
+            self._channel_registry,
+            self._resampling_request_sender(),
+        )
+
     def consumer(self) -> Consumer:
         """Return the consumption measuring point of the microgrid."""
         from ..timeseries.consumer import Consumer
@@ -557,6 +586,19 @@ def logical_meter() -> LogicalMeter:
     return _get().logical_meter()
 
 
+def _new_formula_pool(namespace: str) -> FormulaPool:
+    """Create a new, empty formula pool owned by the caller.
+
+    Args:
+        namespace: A label describing what the pool is for (a unique suffix is
+            appended).
+
+    Returns:
+        A new, empty formula pool.
+    """
+    return _get().new_formula_pool(namespace)
+
+
 def consumer() -> Consumer:
     """Return the [`Consumption`][frequenz.sdk.timeseries.consumer.Consumer] measuring point."""
     return _get().consumer()

src/frequenz/sdk/microgrid/_power_distributing/_component_managers/_battery_manager.py

@@ -14,11 +14,9 @@
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid import ApiClientError, OperationOutOfRange
 from frequenz.client.microgrid.component import Battery, Inverter
-from frequenz.client.microgrid.metrics import Metric
 from frequenz.quantities import Power
 from typing_extensions import override
 
-from ....timeseries import Sample
 from ... import connection_manager
 from ..._old_component_data import BatteryData, InverterData
 from .._component_pool_status_tracker import ComponentPoolStatusTracker
@@ -146,6 +144,7 @@ def __init__(
             api_power_request_timeout: Timeout to use when making power requests to
                 the microgrid API.
         """
+        super().__init__()
         self._results_sender = results_sender
         self._api_power_request_timeout = api_power_request_timeout
         self._batteries = connection_manager.get().component_graph.components(
@@ -163,20 +162,6 @@ def __init__(
         self._battery_caches: dict[ComponentId, LatestValueCache[BatteryData]] = {}
         self._inverter_caches: dict[ComponentId, LatestValueCache[InverterData]] = {}
 
-        self._unreachable_battery_powers: dict[
-            frozenset[ComponentId],
-            tuple[collections.abc.Set[ComponentId], LatestValueCache[Sample[Power]]],
-        ] = {}
-        """Map from battery sets to data from inaccessible battery subset.
-
-        When batteries are inaccessible, for example, due to network issues and can't be
-        controlled, they might still be producing or consuming power.  This power needs
-        to be considered when distributing power to the other batteries.
-
-        So for each battery set, we track of the inaccessible subset of this battery set
-        and the result of the power formulas for the inaccessible subset here.
-        """
-
         self._component_pool_status_tracker = ComponentPoolStatusTracker(
             component_ids=set(self._battery_ids),
             component_status_sender=component_pool_status_sender,
@@ -219,6 +204,7 @@ async def stop(self) -> None:
         for inv_cache in self._inverter_caches.values():
             await inv_cache.stop()
         await self._component_pool_status_tracker.stop()
+        await self._stop_all_unreachable_power_subscriptions()
 
     @override
     async def distribute_power(self, request: Request) -> None:
@@ -536,76 +522,12 @@ def nan_metric_in_list(data: list[DataType], metrics: list[str]) -> bool:
 
         return InvBatPair(AggregatedBatteryData(battery_data), inverter_data)
 
-    async def _subscribe_to_unreachable_battery_power(
-        self,
-        requested_battery_ids: collections.abc.Set[ComponentId],
-        working_battery_ids: collections.abc.Set[ComponentId],
-    ) -> None:
-        requested_battery_ids = frozenset(requested_battery_ids)
-        unreachable_battery_ids = requested_battery_ids - working_battery_ids
-
-        if not unreachable_battery_ids:
-            if unreachable_power := self._unreachable_battery_powers.pop(
-                requested_battery_ids, None
-            ):
-                _logger.debug(
-                    "All batteries are reachable, stopping unreachable battery power "
-                    + "subscription for batteries %s",
-                    self._str_ids(requested_battery_ids),
-                )
-                await unreachable_power[1].stop()
-            return
-
-        if unreachable_power := self._unreachable_battery_powers.get(
-            requested_battery_ids
-        ):
-            if unreachable_power[0] == unreachable_battery_ids:
-                return
-            _logger.debug(
-                "Unreachable battery set for batteries %s changed from %s to %s, "
-                + "restarting subscription",
-                self._str_ids(requested_battery_ids),
-                self._str_ids(unreachable_power[0]),
-                self._str_ids(unreachable_battery_ids),
-            )
-            await unreachable_power[1].stop()
-
-        formula = connection_manager.get().component_graph.battery_formula(
-            unreachable_battery_ids
-        )
-        _logger.debug(
-            "Subscribing to unreachable battery power for batteries %s with formula: %s",
-            self._str_ids(unreachable_battery_ids),
-            formula,
-        )
-
-        # This is to avoid a circular import.  Pylint doesn't detect that this
-        # is not a circular import, so we need to disable the warning also.
-        #
-        # pylint: disable-next=import-outside-toplevel,cyclic-import
-        from ... import (
-            logical_meter,
-        )
-
-        formula_cache = LatestValueCache(
-            logical_meter()
-            .start_formula(formula, Metric.AC_ACTIVE_POWER)
-            .new_receiver()
-            .map(
-                lambda sample: Sample[Power](
-                    sample.timestamp,
-                    (
-                        Power.from_watts(sample.value.base_value)
-                        if sample.value is not None
-                        else None
-                    ),
-                )
-            )
-        )
-        self._unreachable_battery_powers[requested_battery_ids] = (
-            unreachable_battery_ids,
-            formula_cache,
-        )
+    @override
+    def _unreachable_power_formula(
+        self, component_ids: collections.abc.Set[ComponentId]
+    ) -> str:
+        """Return the formula for the active power of the given batteries."""
+        return connection_manager.get().component_graph.battery_formula(component_ids)
 
     async def _get_components_data(
         self, batteries: collections.abc.Set[ComponentId]
@@ -629,7 +551,7 @@ async def _get_components_data(
             batteries
         )
 
-        await self._subscribe_to_unreachable_battery_power(
+        await self._subscribe_to_unreachable_power(
             batteries,
             working_batteries,
         )
@@ -676,16 +598,9 @@ async def _get_components_data(
             assert len(data.inverter) > 0
             pairs_data.append(data)
 
-        unreachable_power: Power | None = None
-        if unreachable_power_cache := self._unreachable_battery_powers.get(
-            frozenset(batteries)
-        ):
-            if unreachable_power_cache[1].has_value():
-                unreachable_power = unreachable_power_cache[1].get().value
-
         return BatteryComponentsData(
             inv_bat_pairs=pairs_data,
-            unreachable_power=unreachable_power,
+            unreachable_power=self._unreachable_power(batteries),
         )
 
     def _str_ids(self, ids: collections.abc.Set[ComponentId]) -> str: