Rewrite merging strategy design

Implement merge strategy based on TYPE+TARGET

Signed-off-by: Mathias L. Baumann <[email protected]>

Author: Marenz

RELEASE_NOTES.md

@@ -13,7 +13,7 @@
 
 ## New Features
 
-* A new feature "unify running intervals" has been added to the `Dispatcher.new_running_state_event_receiver` method. Using it, you can automatically merge & unify consecutive and overlapping dispatch start/stop events of the same type. E.g. dispatch `A` starting at 10:10 and ending at 10:30 and dispatch `B` starts at 10:30 until 11:00, with the feature enabled this would in total trigger one start event, one reconfigure event at 10:30 and one stop event at 11:00.
+* A new feature "merger strategy" (`MergeByType`, `MergeByTypeTarget`) has been added to the `Dispatcher.new_running_state_event_receiver` method. Using it, you can automatically merge & unify consecutive and overlapping dispatch start/stop events of the same type. E.g. dispatch `A` starting at 10:10 and ending at 10:30 and dispatch `B` starts at 10:30 until 11:00, with the feature enabled this would in total trigger one start event, one reconfigure event at 10:30 and one stop event at 11:00.
 
 * The SDK dependency was widened to allow versions up to (excluding) v1.0.0-rc1600.
 

src/frequenz/dispatch/__init__.py

@@ -15,6 +15,7 @@
 
 """
 
+from ._bg_service import MergeByType, MergeByTypeTarget, MergeStrategy
 from ._dispatch import Dispatch
 from ._dispatcher import Dispatcher
 from ._event import Created, Deleted, DispatchEvent, Updated
@@ -29,4 +30,7 @@
     "Dispatch",
     "DispatchManagingActor",
     "DispatchUpdate",
+    "MergeStrategy",  # To allow for user strategies
+    "MergeByType",
+    "MergeByTypeTarget",
 ]

src/frequenz/dispatch/_bg_service.py

@@ -3,18 +3,24 @@
 
 """The dispatch background service."""
 
+from __future__ import annotations
+
 import asyncio
 import logging
+from abc import ABC, abstractmethod
+from collections.abc import ValuesView
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from heapq import heappop, heappush
+from typing import Callable
 
 import grpc.aio
 from frequenz.channels import Broadcast, Receiver, select, selected_from
 from frequenz.channels.timer import SkipMissedAndResync, Timer
 from frequenz.client.dispatch import Client
 from frequenz.client.dispatch.types import Event
 from frequenz.sdk.actor import BackgroundService
+from typing_extensions import override
 
 from ._dispatch import Dispatch
 from ._event import Created, Deleted, DispatchEvent, Updated
@@ -23,6 +29,76 @@
 """The logger for this module."""
 
 
+class MergeStrategy(ABC):
+    """Base class for strategies to merge running intervals.
+
+    All strategies will be initialized with the dispatch scheduler.
+    """
+
+    def __init__(self, scheduler: DispatchScheduler) -> None:
+        """Initialize the strategy.
+
+        Args:
+            scheduler: The dispatch scheduler.
+        """
+        self._scheduler = scheduler
+
+    @property
+    @abstractmethod
+    def filter(self) -> Callable[[Dispatch], bool]:
+        """Get a filter function for dispatches.
+
+        Returns:
+            A filter function.
+        """
+
+
+class MergeByType(MergeStrategy):
+    """Merge running intervals based on the dispatch type."""
+
+    @property
+    @override
+    def filter(self) -> Callable[[Dispatch], bool]:
+        """Get a filter function for dispatches."""
+        return self._filter_func
+
+    def criteria(self, previous_dispatch: Dispatch, new_dispatch: Dispatch) -> bool:
+        """Define the criteria for checking other running dispatches."""
+        return previous_dispatch.type == new_dispatch.type
+
+    def _filter_func(self, new_dispatch: Dispatch) -> bool:
+        """Filter dispatches based on the merge strategy.
+
+        Keeps start events.
+        Keeps stop events only if no other dispatches matching the
+        strategy's criteria are running.
+        """
+        if new_dispatch.started:
+            return True
+
+        # pylint: disable=protected-access
+        other_dispatches_running = any(
+            dispatch.started
+            for dispatch in self._scheduler.dispatches
+            if self.criteria(dispatch, new_dispatch)
+        )
+        # pylint: enable=protected-access
+
+        return not other_dispatches_running
+
+
+class MergeByTypeTarget(MergeByType):
+    """Merge running intervals based on the dispatch type and target."""
+
+    @override
+    def criteria(self, previous_dispatch: Dispatch, new_dispatch: Dispatch) -> bool:
+        """Define the criteria for checking other running dispatches."""
+        return (
+            previous_dispatch.type == new_dispatch.type
+            and previous_dispatch.target == new_dispatch.target
+        )
+
+
 # pylint: disable=too-many-instance-attributes
 class DispatchScheduler(BackgroundService):
     """Dispatch background service.
@@ -104,6 +180,11 @@ def __init__(
         always at index 0.
         """
 
+    @property
+    def dispatches(self) -> ValuesView[Dispatch]:
+        """All currently cached dispatches."""
+        return self._dispatches.values()
+
     # pylint: disable=redefined-builtin
     def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]:
         """Create a new receiver for lifecycle events.
@@ -119,54 +200,34 @@ def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]:
         )
 
     async def new_running_state_event_receiver(
-        self, type: str, *, unify_running_intervals: bool = True
+        self, type: str, *, merge_strategy: type[MergeStrategy] | None = None
     ) -> Receiver[Dispatch]:
         """Create a new receiver for running state events of the specified type.
 
-        If `unify_running_intervals` is True, running intervals from multiple
-        dispatches of the same type are considered as one continuous running
-        period. In this mode, any stop events are ignored as long as at least
-        one dispatch remains active.
+        `merge_strategy` can be one of `MergeByType` or `MergeByTypeTarget`.
+        If set, running intervals from multiple dispatches will be merged,
+        depending on the chosen strategy.
+        When merging, stop events are ignored as long as at least one
+        merge-criteria-matching dispatch remains active.
 
         Args:
             type: The type of events to receive.
-            unify_running_intervals: Whether to unify running intervals.
-
+            merge_strategy: The merge strategy to use.
         Returns:
             A new receiver for running state status.
         """
-        # Find all matching dispatches based on the type and collect them
         dispatches = [
             dispatch for dispatch in self._dispatches.values() if dispatch.type == type
         ]
 
-        # Create receiver with enough capacity to hold all matching dispatches
         receiver = self._running_state_status_channel.new_receiver(
             limit=max(1, len(dispatches))
         ).filter(lambda dispatch: dispatch.type == type)
 
-        if unify_running_intervals:
-
-            def _is_type_still_running(new_dispatch: Dispatch) -> bool:
-                """Merge time windows of running dispatches.
-
-                Any event that would cause a stop is filtered if at least one
-                dispatch of the same type is running.
-                """
-                if new_dispatch.started:
-                    return True
-
-                other_dispatches_running = any(
-                    dispatch.started
-                    for dispatch in self._dispatches.values()
-                    if dispatch.type == type
-                )
-                # If no other dispatches are running, we can allow the stop event
-                return not other_dispatches_running
-
-            receiver = receiver.filter(_is_type_still_running)
+        if merge_strategy:
+            instance = merge_strategy(self)
+            receiver = receiver.filter(instance.filter)
 
-        # Send all matching dispatches to the receiver
         for dispatch in dispatches:
             await self._send_running_state_change(dispatch)
 

src/frequenz/dispatch/_dispatcher.py

@@ -7,7 +7,7 @@
 from frequenz.channels import Receiver
 from frequenz.client.dispatch import Client
 
-from ._bg_service import DispatchScheduler
+from ._bg_service import DispatchScheduler, MergeStrategy
 from ._dispatch import Dispatch
 from ._event import DispatchEvent
 
@@ -200,7 +200,10 @@ def new_lifecycle_events_receiver(
         return self._bg_service.new_lifecycle_events_receiver(dispatch_type)
 
     async def new_running_state_event_receiver(
-        self, dispatch_type: str, *, unify_running_intervals: bool = True
+        self,
+        dispatch_type: str,
+        *,
+        merge_strategy: type[MergeStrategy] | None = None,
     ) -> Receiver[Dispatch]:
         """Return running state event receiver.
 
@@ -228,18 +231,21 @@ async def new_running_state_event_receiver(
          - The payload changed
          - The dispatch was deleted
 
-        If `unify_running_intervals` is True, running intervals from multiple
-        dispatches of the same type are considered as one continuous running
-        period. In this mode, any stop events are ignored as long as at least
-        one dispatch remains active.
+        If `unify_running_intervals` is set, running intervals from multiple
+        dispatches of the same type/type&target (depending on the chosen
+        strategy) are considered as one continuous running
+        period.
+        In this mode, stop events are ignored as long as at least one (criteria
+        matching) dispatch remains active.
 
         Args:
             dispatch_type: The type of the dispatch to listen for.
-            unify_running_intervals: Whether to unify running intervals.
+            merge_strategy: The type of the strategy to merge running intervals.
+                            One of `MergeByType` or `MergeByTypeTarget`.
 
         Returns:
             A new receiver for dispatches whose running status changed.
         """
         return await self._bg_service.new_running_state_event_receiver(
-            dispatch_type, unify_running_intervals=unify_running_intervals
+            dispatch_type, merge_strategy=merge_strategy
         )