Add mute intervals feature to suppress notifications during date ranges

docs/notification-routing/routing-by-time.rst

@@ -35,6 +35,33 @@ As seen above, a sink can be active during multiple ``intervals``. Each interval
 
 Sinks that don't define an ``activity`` field are always active.
 
+Mute Intervals
+--------------
+
+You can also mute a sink during specific date ranges using ``mute_intervals``. Unlike ``activity`` (which defines recurring weekly schedules), ``mute_intervals`` defines exact date ranges with a year — useful for holidays, maintenance windows, or one-off silencing.
+
+.. code-block:: yaml
+
+    sinksConfig:
+    - slack_sink:
+        name: main_slack_sink
+        slack_channel: robusta-notifications
+        api_key: xoxb-your-slack-key
+        mute_intervals:
+        - start_date: "2025-12-24 00:00"
+          end_date: "2025-12-26 23:59"
+          timezone: CET
+        - start_date: "2026-01-01 00:00"
+          end_date: "2026-01-01 23:59"
+          timezone: US/Eastern
+
+Each interval has:
+
+- ``start_date`` and ``end_date`` in ``YYYY-MM-DD HH:MM`` format.
+- An optional ``timezone`` (defaults to ``UTC``). Each interval can use a different timezone.
+
+When any mute interval is active, the sink will not emit notifications. You can combine ``mute_intervals`` with ``activity`` — the sink must be active according to ``activity`` *and* not muted by any interval.
+
 .. details:: Supported Timezones
 
     .. code-block::

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: YYYY-MM-DD HH:MM.
+# Example:
+#   sinksConfig:
+#   - slack_sink:
+#       name: my_slack_sink
+#       slack_channel: my-channel
+#       api_key: xoxb-your-key
+#       mute_intervals:
+#       - start_date: "2025-12-24 00:00"
+#         end_date: "2025-12-26 23:59"
+#         timezone: UTC
+#       - start_date: "2026-01-01 00:00"
+#         end_date: "2026-01-01 23:59"
+#         timezone: US/Eastern
 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, MuteInterval, 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,14 @@ 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: Optional[List[MuteInterval]]):
+        if not params:
+            return []
+        return [
+            MuteDateInterval(interval.start_date, interval.end_date, interval.timezone)
+            for interval in params
+        ]
+
     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 +172,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,42 @@ def check_intervals(cls, intervals: List[ActivityInterval]):
         return intervals
 
 
+DATE_TIME_RE = re.compile(r"^\d{4}-\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: YYYY-MM-DD HH:MM")
+    date_part, time_part = value.split(" ", 1)
+    year, month, day = date_part.split("-")
+    hour, minute = time_part.split(":")
+    year, month, day, hour, minute = int(year), 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  # YYYY-MM-DD HH:MM
+    end_date: str    # YYYY-MM-DD HH:MM
+    timezone: str = "UTC"
+
+    _validator_start = validator("start_date", allow_reuse=True)(check_date_time_format)
+    _validator_end = validator("end_date", allow_reuse=True)(check_date_time_format)
+
+    @validator("timezone")
+    def check_timezone(cls, timezone: str):
+        if timezone not in pytz.all_timezones:
+            raise ValueError(f"unknown timezone {timezone}")
+        return timezone
+
+
 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 +144,7 @@ class SinkBaseParams(ABC, BaseModel):
     match: dict = {}
     scope: Optional[ScopeParams]
     activity: Optional[ActivityParams]
+    mute_intervals: Optional[List[MuteInterval]]
     grouping: Optional[GroupingParams]
     stop: bool = False  # Stop processing if this sink has been matched
 

src/robusta/core/sinks/timing.py

@@ -65,3 +65,29 @@ 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 YYYY-MM-DD HH:MM format.
+    """
+
+    def __init__(self, start_date: str, end_date: str, timezone: str = "UTC"):
+        self.start = self._parse(start_date)
+        self.end = 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, int]:
+        date_part, time_part = date_str.strip().split(" ")
+        year, month, day = date_part.split("-")
+        hour, minute = time_part.split(":")
+        return int(year), int(month), int(day), int(hour), int(minute)
+
+    def is_muted_now(self) -> bool:
+        now = datetime.now(self.timezone)
+        current = (now.year, now.month, now.day, now.hour, now.minute)
+        return self.start <= current <= self.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,38 @@ def test_invalid_time(self, time):
             TimeSlice([], [time], "UTC")
 
 
+class TestMuteDateInterval:
+    def test_unknown_timezone(self):
+        with pytest.raises(ValueError):
+            MuteDateInterval("2012-01-01 00:00", "2012-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)
+            ("2012-01-01 00:00", "2012-01-01 23:59", "UTC", True),
+            # Currently muted (multi-day range spanning year boundary)
+            ("2011-12-31 00:00", "2012-01-02 10:00", "UTC", True),
+            # Not muted (range in February)
+            ("2012-02-01 00:00", "2012-02-28 23:59", "UTC", False),
+            # Not muted (same day but end is before current time)
+            ("2012-01-01 00:00", "2012-01-01 13:00", "UTC", False),
+            # Muted (same day, hours match)
+            ("2012-01-01 13:00", "2012-01-01 14:00", "UTC", True),
+            # Not muted (range is entirely in the past)
+            ("2011-06-01 00:00", "2011-06-30 23:59", "UTC", False),
+            # Not muted (range is entirely in the future)
+            ("2013-01-01 00:00", "2013-12-31 23:59", "UTC", False),
+            # Timezone test: 2012-01-01 13:45 UTC = 2012-01-01 14:45 CET
+            ("2012-01-01 14:00", "2012-01-01 15:00", "CET", True),
+            ("2012-01-01 15:00", "2012-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 +82,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 +99,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("2012-01-01 00:00", "2012-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("2012-02-01 00:00", "2012-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