Add mute_intervals config for sinks to mute notifications by date range

Adds a new `mute_intervals` field parallel to `activity` in sink config.
Users can specify date/time ranges (MM-DD HH:MM format, no year) during
which all notifications to a sink are muted. Supports timezone config
and year-boundary wrapping (e.g. Dec 24 to Jan 2).

https://claude.ai/code/session_01FGiv1N3QcFMWAPT4e1pu93

Author: claude

helm/robusta/values.yaml

@@ -10,6 +10,21 @@ fullnameOverride: ""
 playbookRepos: {}
 
 # sinks configurations
+# Each sink supports an optional mute_intervals field (parallel to activity) that mutes
+# all notifications during specified date/time ranges. Format: MM-DD HH:MM (no year).
+# Example:
+#   sinksConfig:
+#   - slack_sink:
+#       name: my_slack_sink
+#       slack_channel: my-channel
+#       api_key: xoxb-your-key
+#       mute_intervals:
+#         timezone: UTC
+#         intervals:
+#         - start_date: "12-24 00:00"
+#           end_date: "12-26 23:59"
+#         - start_date: "01-01 00:00"
+#           end_date: "01-01 23:59"
 sinksConfig: []
 
 # global parameters

src/robusta/core/sinks/sink_base.py

@@ -8,8 +8,8 @@
 
 from robusta.core.model.k8s_operation_type import K8sOperationType
 from robusta.core.reporting.base import Finding
-from robusta.core.sinks.sink_base_params import ActivityInterval, ActivityParams, SinkBaseParams
-from robusta.core.sinks.timing import TimeSlice, TimeSliceAlways
+from robusta.core.sinks.sink_base_params import ActivityInterval, ActivityParams, MuteParams, SinkBaseParams
+from robusta.core.sinks.timing import MuteDateInterval, TimeSlice, TimeSliceAlways
 
 
 KeyT = Tuple[str, ...]
@@ -71,6 +71,7 @@ def __init__(self, sink_params: SinkBaseParams, registry):
         self.signing_key = global_config.get("signing_key", "")
 
         self.time_slices = self._build_time_slices_from_params(self.params.activity)
+        self.mute_date_intervals = self._build_mute_intervals_from_params(self.params.mute_intervals)
 
         self.grouping_summary_mode = False
         self.grouping_enabled = False
@@ -151,6 +152,15 @@ def _build_time_slices_from_params(self, params: ActivityParams):
     def _interval_to_time_slice(self, timezone: str, interval: ActivityInterval):
         return TimeSlice(interval.days, [(time.start, time.end) for time in interval.hours], timezone)
 
+    def _build_mute_intervals_from_params(self, params: MuteParams):
+        if params is None:
+            return []
+        timezone = params.timezone
+        return [
+            MuteDateInterval(interval.start_date, interval.end_date, timezone)
+            for interval in params.intervals
+        ]
+
     def is_global_config_changed(self) -> bool:
         # registry global config can be updated without these stored values being changed
         global_config = self.registry.get_global_config()
@@ -163,6 +173,8 @@ def stop(self):
         pass
 
     def accepts(self, finding: Finding) -> bool:
+        if any(mute.is_muted_now() for mute in self.mute_date_intervals):
+            return False
         return (
             finding.matches(self.params.match, self.params.scope)
             and any(time_slice.is_active_now() for time_slice in self.time_slices)

src/robusta/core/sinks/sink_base_params.py

@@ -57,6 +57,52 @@ def check_intervals(cls, intervals: List[ActivityInterval]):
         return intervals
 
 
+DATE_TIME_RE = re.compile(r"^\d{2}-\d{2} \d{2}:\d{2}$")
+
+
+def check_date_time_format(value: str) -> str:
+    if not DATE_TIME_RE.match(value):
+        raise ValueError(f"invalid date-time: {value}. Expected format: MM-DD HH:MM")
+    month, rest = value.split("-", 1)
+    day, time_part = rest.split(" ", 1)
+    hour, minute = time_part.split(":")
+    month, day, hour, minute = int(month), int(day), int(hour), int(minute)
+    if not (1 <= month <= 12):
+        raise ValueError(f"invalid month: {month}")
+    if not (1 <= day <= 31):
+        raise ValueError(f"invalid day: {day}")
+    if not (0 <= hour <= 23):
+        raise ValueError(f"invalid hour: {hour}")
+    if not (0 <= minute <= 59):
+        raise ValueError(f"invalid minute: {minute}")
+    return value
+
+
+class MuteInterval(BaseModel):
+    start_date: str  # MM-DD HH:MM
+    end_date: str    # MM-DD HH:MM
+
+    _validator_start = validator("start_date", allow_reuse=True)(check_date_time_format)
+    _validator_end = validator("end_date", allow_reuse=True)(check_date_time_format)
+
+
+class MuteParams(BaseModel):
+    timezone: str = "UTC"
+    intervals: List[MuteInterval]
+
+    @validator("timezone")
+    def check_timezone(cls, timezone: str):
+        if timezone not in pytz.all_timezones:
+            raise ValueError(f"unknown timezone {timezone}")
+        return timezone
+
+    @validator("intervals")
+    def check_intervals(cls, intervals: List[MuteInterval]):
+        if not intervals:
+            raise ValueError("at least one interval has to be specified for mute_intervals")
+        return intervals
+
+
 class RegularNotificationModeParams(BaseModel):
     # This is mandatory because using the regular mode without setting it
     # would make no sense - all the notifications would just pass through
@@ -108,6 +154,7 @@ class SinkBaseParams(ABC, BaseModel):
     match: dict = {}
     scope: Optional[ScopeParams]
     activity: Optional[ActivityParams]
+    mute_intervals: Optional[MuteParams]
     grouping: Optional[GroupingParams]
     stop: bool = False  # Stop processing if this sink has been matched
 

src/robusta/core/sinks/timing.py

@@ -65,3 +65,41 @@ def is_active_now(self) -> bool:
 class TimeSliceAlways(TimeSliceBase):
     def is_active_now(self) -> bool:
         return True
+
+
+class MuteDateInterval:
+    """Checks if the current date/time falls within a mute interval.
+
+    start_date and end_date are in MM-DD HH:MM format (no year).
+    The interval applies to the current year. If start_date > end_date
+    (e.g. 12-20 to 01-05), it wraps across the year boundary.
+    """
+
+    def __init__(self, start_date: str, end_date: str, timezone: str = "UTC"):
+        self.start_month, self.start_day, self.start_hour, self.start_minute = self._parse(start_date)
+        self.end_month, self.end_day, self.end_hour, self.end_minute = self._parse(end_date)
+        try:
+            self.timezone = pytz.timezone(timezone)
+        except pytz.exceptions.UnknownTimeZoneError:
+            raise ValueError(f"Unknown time zone {timezone}")
+
+    def _parse(self, date_str: str) -> Tuple[int, int, int, int]:
+        date_part, time_part = date_str.strip().split(" ")
+        month, day = date_part.split("-")
+        hour, minute = time_part.split(":")
+        return int(month), int(day), int(hour), int(minute)
+
+    def _to_tuple(self, month: int, day: int, hour: int, minute: int) -> Tuple[int, int, int, int]:
+        return (month, day, hour, minute)
+
+    def is_muted_now(self) -> bool:
+        now = datetime.now(self.timezone)
+        current = self._to_tuple(now.month, now.day, now.hour, now.minute)
+        start = self._to_tuple(self.start_month, self.start_day, self.start_hour, self.start_minute)
+        end = self._to_tuple(self.end_month, self.end_day, self.end_hour, self.end_minute)
+
+        if start <= end:
+            return start <= current <= end
+        else:
+            # Wraps across year boundary (e.g. 12-20 00:00 to 01-05 00:00)
+            return current >= start or current <= end

tests/test_sink_timing.py

@@ -6,7 +6,7 @@
 from robusta.core.reporting import Finding
 from robusta.core.sinks.sink_base import SinkBase
 from robusta.core.sinks.sink_base_params import ActivityParams
-from robusta.core.sinks.timing import TimeSlice
+from robusta.core.sinks.timing import MuteDateInterval, TimeSlice
 
 
 class TestTimeSlice:
@@ -39,6 +39,40 @@ def test_invalid_time(self, time):
             TimeSlice([], [time], "UTC")
 
 
+class TestMuteDateInterval:
+    def test_unknown_timezone(self):
+        with pytest.raises(ValueError):
+            MuteDateInterval("01-01 00:00", "01-02 00:00", "Mars/Cydonia")
+
+    @pytest.mark.parametrize(
+        "start_date,end_date,timezone,expected_muted",
+        [
+            # 2012-01-01 13:45 UTC - currently muted (within range)
+            ("01-01 00:00", "01-01 23:59", "UTC", True),
+            # Currently muted (multi-day range)
+            ("12-31 00:00", "01-02 10:00", "UTC", True),
+            # Not muted (range in February)
+            ("02-01 00:00", "02-28 23:59", "UTC", False),
+            # Not muted (same day but hours don't match - before current time)
+            ("01-01 00:00", "01-01 13:00", "UTC", False),
+            # Muted (same day, hours match)
+            ("01-01 13:00", "01-01 14:00", "UTC", True),
+            # Year-boundary wrap: Dec 20 to Jan 5 should mute on Jan 1
+            ("12-20 00:00", "01-05 23:59", "UTC", True),
+            # Year-boundary wrap: March to Feb wraps around, Jan 1 IS inside that range
+            ("03-01 00:00", "02-15 23:59", "UTC", True),
+            # Not muted: range is Feb 1 to Feb 28, Jan 1 is outside
+            ("02-01 00:00", "02-10 23:59", "UTC", False),
+            # Timezone test: 2012-01-01 13:45 UTC = 2012-01-01 14:45 CET
+            ("01-01 14:00", "01-01 15:00", "CET", True),
+            ("01-01 15:00", "01-01 16:00", "CET", False),
+        ],
+    )
+    def test_is_muted_now(self, start_date, end_date, timezone, expected_muted):
+        with freeze_time("2012-01-01 13:45"):  # UTC time
+            assert MuteDateInterval(start_date, end_date, timezone).is_muted_now() is expected_muted
+
+
 class _TestSinkBase(SinkBase):
     def write_finding(self, finding: Finding, platform_enabled: bool):
         pass
@@ -50,6 +84,10 @@ def _build_time_slices_from_params(self, params: ActivityParams):
         # We'll construct time_slices explicitly below in TestSinkBase.test_accepts
         pass
 
+    def _build_mute_intervals_from_params(self, params):
+        # We'll construct mute_date_intervals explicitly below
+        pass
+
 
 class TestSinkBase:
     @pytest.mark.parametrize(
@@ -63,6 +101,29 @@ def test_accepts(self, days, time_intervals, expected_result):
         mock_registry = Mock(get_global_config=lambda: Mock())
         sink = _TestSinkBase(registry=mock_registry, sink_params=Mock())
         sink.time_slices = [TimeSlice(days, time_intervals, "UTC")]
+        sink.mute_date_intervals = []
         mock_finding = Mock(matches=Mock(return_value=True))
         with freeze_time("2012-01-01 13:45"):  # this is UTC time
             assert sink.accepts(mock_finding) is expected_result
+
+    def test_accepts_muted(self):
+        """When a mute interval is active, accepts() should return False."""
+        mock_registry = Mock(get_global_config=lambda: Mock())
+        sink = _TestSinkBase(registry=mock_registry, sink_params=Mock())
+        sink.time_slices = [TimeSlice(["sun"], [("13:30", "14:00")], "UTC")]
+        sink.mute_date_intervals = [MuteDateInterval("01-01 00:00", "01-01 23:59", "UTC")]
+        mock_finding = Mock(matches=Mock(return_value=True))
+        with freeze_time("2012-01-01 13:45"):  # this is UTC time, Sunday
+            # Would normally be accepted (Sunday 13:45 in 13:30-14:00), but muted
+            assert sink.accepts(mock_finding) is False
+
+    def test_accepts_not_muted(self):
+        """When no mute interval is active, accepts() works normally."""
+        mock_registry = Mock(get_global_config=lambda: Mock())
+        sink = _TestSinkBase(registry=mock_registry, sink_params=Mock())
+        sink.time_slices = [TimeSlice(["sun"], [("13:30", "14:00")], "UTC")]
+        sink.mute_date_intervals = [MuteDateInterval("02-01 00:00", "02-28 23:59", "UTC")]
+        mock_finding = Mock(matches=Mock(return_value=True))
+        with freeze_time("2012-01-01 13:45"):  # this is UTC time, Sunday
+            # Mute is for February, so should still accept
+            assert sink.accepts(mock_finding) is True