From 552ae3ec0918487c9520390aa603dd72ab63f9fc Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 4 Mar 2026 10:50:39 +0000 Subject: [PATCH 1/4] 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 --- helm/robusta/values.yaml | 15 ++++++ src/robusta/core/sinks/sink_base.py | 16 +++++- src/robusta/core/sinks/sink_base_params.py | 47 ++++++++++++++++ src/robusta/core/sinks/timing.py | 38 +++++++++++++ tests/test_sink_timing.py | 63 +++++++++++++++++++++- 5 files changed, 176 insertions(+), 3 deletions(-) diff --git a/helm/robusta/values.yaml b/helm/robusta/values.yaml index 16a48f1b5..b082d725b 100644 --- a/helm/robusta/values.yaml +++ b/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 diff --git a/src/robusta/core/sinks/sink_base.py b/src/robusta/core/sinks/sink_base.py index eac2123b5..264c77824 100644 --- a/src/robusta/core/sinks/sink_base.py +++ b/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) diff --git a/src/robusta/core/sinks/sink_base_params.py b/src/robusta/core/sinks/sink_base_params.py index f9c055ecc..c8faf25a1 100644 --- a/src/robusta/core/sinks/sink_base_params.py +++ b/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 diff --git a/src/robusta/core/sinks/timing.py b/src/robusta/core/sinks/timing.py index 93d9be8c8..99e4eea3e 100644 --- a/src/robusta/core/sinks/timing.py +++ b/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 diff --git a/tests/test_sink_timing.py b/tests/test_sink_timing.py index 01e9313a0..6845ce224 100644 --- a/tests/test_sink_timing.py +++ b/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 From 5b6af545fd020ee480aad86f7b995eac21d59a7c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 4 Mar 2026 10:53:13 +0000 Subject: [PATCH 2/4] Change mute_intervals to use exact dates with year (YYYY-MM-DD HH:MM) Switches from MM-DD HH:MM (yearless) to YYYY-MM-DD HH:MM format. This simplifies the logic (no year-boundary wrapping needed) and gives users precise control over mute windows. https://claude.ai/code/session_01FGiv1N3QcFMWAPT4e1pu93 --- helm/robusta/values.yaml | 10 +++---- src/robusta/core/sinks/sink_base_params.py | 14 +++++----- src/robusta/core/sinks/timing.py | 28 ++++++------------- tests/test_sink_timing.py | 32 ++++++++++------------ 4 files changed, 35 insertions(+), 49 deletions(-) diff --git a/helm/robusta/values.yaml b/helm/robusta/values.yaml index b082d725b..a06582700 100644 --- a/helm/robusta/values.yaml +++ b/helm/robusta/values.yaml @@ -11,7 +11,7 @@ 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). +# all notifications during specified date/time ranges. Format: YYYY-MM-DD HH:MM. # Example: # sinksConfig: # - slack_sink: @@ -21,10 +21,10 @@ playbookRepos: {} # 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" +# - start_date: "2025-12-24 00:00" +# end_date: "2025-12-26 23:59" +# - start_date: "2026-01-01 00:00" +# end_date: "2026-01-01 23:59" sinksConfig: [] # global parameters diff --git a/src/robusta/core/sinks/sink_base_params.py b/src/robusta/core/sinks/sink_base_params.py index c8faf25a1..2fde84b12 100644 --- a/src/robusta/core/sinks/sink_base_params.py +++ b/src/robusta/core/sinks/sink_base_params.py @@ -57,16 +57,16 @@ def check_intervals(cls, intervals: List[ActivityInterval]): return intervals -DATE_TIME_RE = re.compile(r"^\d{2}-\d{2} \d{2}:\d{2}$") +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: MM-DD HH:MM") - month, rest = value.split("-", 1) - day, time_part = rest.split(" ", 1) + 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(":") - month, day, hour, minute = int(month), int(day), int(hour), int(minute) + 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): @@ -79,8 +79,8 @@ def check_date_time_format(value: str) -> str: class MuteInterval(BaseModel): - start_date: str # MM-DD HH:MM - end_date: str # MM-DD HH:MM + start_date: str # YYYY-MM-DD HH:MM + end_date: str # YYYY-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) diff --git a/src/robusta/core/sinks/timing.py b/src/robusta/core/sinks/timing.py index 99e4eea3e..224f503d1 100644 --- a/src/robusta/core/sinks/timing.py +++ b/src/robusta/core/sinks/timing.py @@ -70,36 +70,24 @@ def is_active_now(self) -> bool: 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. + 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_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) + 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]: + def _parse(self, date_str: str) -> Tuple[int, int, int, int, int]: date_part, time_part = date_str.strip().split(" ") - month, day = date_part.split("-") + year, 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) + return int(year), int(month), int(day), int(hour), int(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 + current = (now.year, now.month, now.day, now.hour, now.minute) + return self.start <= current <= self.end diff --git a/tests/test_sink_timing.py b/tests/test_sink_timing.py index 6845ce224..73d2ae78f 100644 --- a/tests/test_sink_timing.py +++ b/tests/test_sink_timing.py @@ -48,24 +48,22 @@ def test_unknown_timezone(self): "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), + ("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) - ("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), + ("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) - ("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), + ("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 - ("01-01 14:00", "01-01 15:00", "CET", True), - ("01-01 15:00", "01-01 16:00", "CET", False), + ("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): @@ -111,7 +109,7 @@ def test_accepts_muted(self): 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")] + 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 @@ -122,7 +120,7 @@ def test_accepts_not_muted(self): 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")] + 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 From 937861ccb937c3fda0880c3b745e91f84a71a465 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 4 Mar 2026 11:01:14 +0000 Subject: [PATCH 3/4] Flatten mute_intervals: remove MuteParams wrapper, put timezone on each interval mute_intervals is now a direct list of intervals instead of a nested object with timezone + intervals. Each interval carries its own timezone. https://claude.ai/code/session_01FGiv1N3QcFMWAPT4e1pu93 --- helm/robusta/values.yaml | 10 +++++----- src/robusta/core/sinks/sink_base.py | 11 +++++------ src/robusta/core/sinks/sink_base_params.py | 14 ++------------ tests/test_sink_timing.py | 2 +- 4 files changed, 13 insertions(+), 24 deletions(-) diff --git a/helm/robusta/values.yaml b/helm/robusta/values.yaml index a06582700..abda1dd7a 100644 --- a/helm/robusta/values.yaml +++ b/helm/robusta/values.yaml @@ -19,12 +19,12 @@ playbookRepos: {} # 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 -# intervals: -# - start_date: "2025-12-24 00:00" -# end_date: "2025-12-26 23:59" -# - start_date: "2026-01-01 00:00" -# end_date: "2026-01-01 23:59" +# - start_date: "2026-01-01 00:00" +# end_date: "2026-01-01 23:59" +# timezone: US/Eastern sinksConfig: [] # global parameters diff --git a/src/robusta/core/sinks/sink_base.py b/src/robusta/core/sinks/sink_base.py index 264c77824..80fb6de66 100644 --- a/src/robusta/core/sinks/sink_base.py +++ b/src/robusta/core/sinks/sink_base.py @@ -8,7 +8,7 @@ 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, MuteParams, SinkBaseParams +from robusta.core.sinks.sink_base_params import ActivityInterval, ActivityParams, MuteInterval, SinkBaseParams from robusta.core.sinks.timing import MuteDateInterval, TimeSlice, TimeSliceAlways @@ -152,13 +152,12 @@ 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: + def _build_mute_intervals_from_params(self, params: Optional[List[MuteInterval]]): + if not params: return [] - timezone = params.timezone return [ - MuteDateInterval(interval.start_date, interval.end_date, timezone) - for interval in params.intervals + MuteDateInterval(interval.start_date, interval.end_date, interval.timezone) + for interval in params ] def is_global_config_changed(self) -> bool: diff --git a/src/robusta/core/sinks/sink_base_params.py b/src/robusta/core/sinks/sink_base_params.py index 2fde84b12..434b8a4f6 100644 --- a/src/robusta/core/sinks/sink_base_params.py +++ b/src/robusta/core/sinks/sink_base_params.py @@ -81,27 +81,17 @@ def check_date_time_format(value: str) -> str: 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) - -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 @@ -154,7 +144,7 @@ class SinkBaseParams(ABC, BaseModel): match: dict = {} scope: Optional[ScopeParams] activity: Optional[ActivityParams] - mute_intervals: Optional[MuteParams] + mute_intervals: Optional[List[MuteInterval]] grouping: Optional[GroupingParams] stop: bool = False # Stop processing if this sink has been matched diff --git a/tests/test_sink_timing.py b/tests/test_sink_timing.py index 73d2ae78f..aa661799d 100644 --- a/tests/test_sink_timing.py +++ b/tests/test_sink_timing.py @@ -42,7 +42,7 @@ def test_invalid_time(self, time): class TestMuteDateInterval: def test_unknown_timezone(self): with pytest.raises(ValueError): - MuteDateInterval("01-01 00:00", "01-02 00:00", "Mars/Cydonia") + MuteDateInterval("2012-01-01 00:00", "2012-01-02 00:00", "Mars/Cydonia") @pytest.mark.parametrize( "start_date,end_date,timezone,expected_muted", From ed34127ff448e0c88e73441f9025a1e18846eda3 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 4 Mar 2026 11:12:09 +0000 Subject: [PATCH 4/4] Add mute_intervals section to routing-by-time docs https://claude.ai/code/session_01FGiv1N3QcFMWAPT4e1pu93 --- docs/notification-routing/routing-by-time.rst | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/docs/notification-routing/routing-by-time.rst b/docs/notification-routing/routing-by-time.rst index c492eccb0..7b46c290f 100644 --- a/docs/notification-routing/routing-by-time.rst +++ b/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::