Rewrite merging strategy design

Implement merge strategy based on TYPE+TARGET

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

Author: Marenz

Diff for: RELEASE_NOTES.md

@@ -12,7 +12,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.
 
 ## Bug Fixes
 

Diff for: 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",
 ]

Diff for: src/frequenz/dispatch/_bg_service.py

@@ -3,11 +3,15 @@
 
 """The dispatch background service."""
 
+from __future__ import annotations
+
 import asyncio
 import logging
+from abc import ABC, abstractmethod
 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
@@ -23,6 +27,77 @@
 """The logger for this module."""
 
 
+class _MergeStrategy(ABC):
+    """Base class for strategies to merge running intervals."""
+
+    @abstractmethod
+    def _get_filter_function(
+        self,
+        scheduler: DispatchScheduler,
+    ) -> Callable[[Dispatch], bool]:
+        """Get a filter function for dispatches.
+
+        Args:
+            scheduler: The dispatch scheduler.
+
+        Returns:
+            A filter function.
+        """
+
+
+class MergeByType(_MergeStrategy):
+    """Merge running intervals based on the dispatch type."""
+
+    def __init__(self) -> None:
+        """Initialize the strategy."""
+        self._scheduler: DispatchScheduler
+        self._new_dispatch: Dispatch
+
+    def _get_filter_function(
+        self, scheduler: DispatchScheduler
+    ) -> Callable[[Dispatch], bool]:
+        """Get a filter function for dispatches."""
+        self._scheduler = scheduler
+        return self._filter_func
+
+    def _criteria(self, dispatch: Dispatch) -> bool:
+        """Define the criteria for checking other running dispatches."""
+        return dispatch.type == self._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
+
+        self._new_dispatch = new_dispatch
+
+        # pylint: disable=protected-access
+        other_dispatches_running = any(
+            dispatch.started
+            for dispatch in self._scheduler._dispatches.values()
+            if self._criteria(dispatch)
+        )
+        # pylint: enable=protected-access
+
+        return not other_dispatches_running
+
+
+class MergeByTypeTarget(MergeByType):
+    """Merge running intervals based on the dispatch type and target."""
+
+    def _criteria(self, dispatch: Dispatch) -> bool:
+        """Define the criteria for checking other running dispatches."""
+        return (
+            dispatch.type == self._new_dispatch.type
+            and dispatch.target == self._new_dispatch.target
+        )
+
+
 # pylint: disable=too-many-instance-attributes
 class DispatchScheduler(BackgroundService):
     """Dispatch background service.
@@ -119,54 +194,35 @@ 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: _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:
+            # pylint: disable=protected-access
+            receiver = receiver.filter(merge_strategy._get_filter_function(self))
+            # pylint: enable=protected-access
 
-        # Send all matching dispatches to the receiver
         for dispatch in dispatches:
             await self._send_running_state_change(dispatch)
 

Diff for: 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: _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 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
         )