Allow ignoring sinks init errors (#1915)

* Allow ignoring sinks init errors
Fix finding dates timezone issues

* CR fixes

* CR fixes

Author: arikalon1

docs/notification-routing/configuring-sinks.rst

@@ -90,6 +90,22 @@ Description of each option:
 | *parameters*     | Slack and webhook_url for MSTeams                       |                                                          |                                               |
 +------------------+---------------------------------------------------------+----------------------------------------------------------+-----------------------------------------------+
 
+Ignoring Sinks Initialization Errors
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+By default, when Robusta fails to initialize any of the Sinks, it will not start.
+
+On some scenarios, you may want to ignore Sinks initialization errors.
+
+For example, if Robusta is not allowed to connect to Slack, but you still want to receive notifications on the Robusta UI.
+
+In order to enable that, add the below to ``globalConfig`` in your ``generated_values.yaml`` file:
+
+.. code-block:: yaml
+
+    globalConfig:
+      continue_on_sink_errors: True
+
 Learn More
 ^^^^^^^^^^^^
 

src/robusta/core/model/cluster_status.py

@@ -23,6 +23,7 @@ class ActivityStats(BaseModel):
     holmesModel: Optional[str]
     clusterTimeZone: str
     errors: List[str]
+    sinksInitializationErrors: bool = False
 
 
 class ClusterStatus(BaseModel):

src/robusta/core/reporting/base.py

@@ -17,6 +17,7 @@
 from robusta.core.reporting.consts import FindingSource, FindingSubjectType, FindingType
 from robusta.integrations.kubernetes.api_client_utils import get_namespace_labels
 from robusta.utils.scope import BaseScopeMatcher
+from robusta.utils.time_utils import current_utc_timestamp
 
 
 class BaseBlock(BaseModel):
@@ -289,7 +290,7 @@ def __init__(
         self.fingerprint = (
             fingerprint if fingerprint else self.__calculate_fingerprint(subject, source, aggregation_key)
         )
-        self.starts_at = starts_at if starts_at else datetime.now()
+        self.starts_at = starts_at if starts_at else current_utc_timestamp()
         self.ends_at = ends_at
         self.dirty = False
 

src/robusta/core/sinks/robusta/dal/model_conversion.py

@@ -26,6 +26,7 @@
 from robusta.core.reporting.holmes import HolmesChatResultsBlock, HolmesResultsBlock, ToolCallResult
 from robusta.core.sinks.transformer import Transformer
 from robusta.utils.parsing import datetime_to_db_str
+from robusta.utils.time_utils import current_utc_timestamp
 
 
 class ModelConversion:
@@ -52,7 +53,7 @@ def to_finding_json(account_id: str, cluster_id: str, finding: Finding):
             "account_id": account_id,
             "video_links": [link.dict() for link in finding.links],  # TD: Migrate column in table.
             "starts_at": datetime_to_db_str(finding.starts_at),
-            "updated_at": datetime_to_db_str(datetime.now()),
+            "updated_at": datetime_to_db_str(current_utc_timestamp()),
         }
 
         if finding.creation_date:

src/robusta/core/sinks/robusta/robusta_sink.py

@@ -563,6 +563,7 @@ def __update_cluster_status(self):
             holmesModel=holmes_model,
             clusterTimeZone=str(datetime.now().astimezone().tzinfo),
             errors=self.__errors,
+            sinksInitializationErrors=self.registry.has_sink_initialization_errors(),
         )
 
         # checking the status of relay connection

src/robusta/model/config.py

@@ -1,6 +1,6 @@
 import logging
 from collections import defaultdict
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Tuple
 
 from robusta.core.model.env_vars import PROMETHEUS_ENABLED, RUNNER_VERSION
 from robusta.core.playbooks.actions_registry import ActionsRegistry
@@ -46,7 +46,8 @@ def construct_new_sinks(
         new_sinks_config: List[SinkConfigBase],
         existing_sinks: Dict[str, SinkBase],
         registry,
-    ) -> Dict[str, SinkBase]:
+        continue_on_sink_errors: bool = False,
+    ) -> Tuple[Dict[str, SinkBase], bool]:
         new_sink_names = [sink_config.get_name() for sink_config in new_sinks_config]
         # remove deleted sinks
         deleted_sink_names = [sink_name for sink_name in existing_sinks.keys() if sink_name not in new_sink_names]
@@ -56,6 +57,7 @@ def construct_new_sinks(
             del existing_sinks[deleted_sink]
 
         new_sinks: Dict[str, SinkBase] = dict()
+        has_sink_errors = False
 
         # Reload sinks, order does matter and should be loaded & added to the dict by config order.
         for sink_config in new_sinks_config:
@@ -68,24 +70,33 @@ def construct_new_sinks(
 
             sink_name = sink_config.get_name()
             exists_sink = existing_sinks.get(sink_name, None)
-            if not exists_sink:
-                logging.info(f"Adding {type(sink_config)} sink named {sink_name}")
-                new_sinks[sink_name] = SinkFactory.create_sink(sink_config, registry)
-                continue
-
-            is_global_config_changed = exists_sink.is_global_config_changed()
-            is_sink_changed = sink_config.get_params() != exists_sink.params or is_global_config_changed
-            if is_sink_changed:
-                config_change_msg = "due to global config change" if is_global_config_changed else "due to param change"
-                logging.info(f"Updating {type(sink_config)} sink named {sink_config.get_name()} {config_change_msg}")
-                exists_sink.stop()
-                new_sinks[sink_name] = SinkFactory.create_sink(sink_config, registry)
-                continue
-
-            logging.info("Sink %s not changed", sink_name)
-            new_sinks[sink_name] = exists_sink
-
-        return new_sinks
+            
+            try:
+                if not exists_sink:
+                    logging.info(f"Adding {type(sink_config)} sink named {sink_name}")
+                    new_sinks[sink_name] = SinkFactory.create_sink(sink_config, registry)
+                    continue
+
+                is_global_config_changed = exists_sink.is_global_config_changed()
+                is_sink_changed = sink_config.get_params() != exists_sink.params or is_global_config_changed
+                if is_sink_changed:
+                    config_change_msg = "due to global config change" if is_global_config_changed else "due to param change"
+                    logging.info(f"Updating {type(sink_config)} sink named {sink_config.get_name()} {config_change_msg}")
+                    exists_sink.stop()
+                    new_sinks[sink_name] = SinkFactory.create_sink(sink_config, registry)
+                    continue
+
+                logging.info("Sink %s not changed", sink_name)
+                new_sinks[sink_name] = exists_sink
+                
+            except Exception as e:
+                has_sink_errors = True
+                logging.error(f"Failed to initialize sink {sink_name}: {e}", exc_info=True)
+                if not continue_on_sink_errors:
+                    raise
+                # Skip this sink if continue_on_sink_errors is True
+
+        return new_sinks, has_sink_errors
 
 
 class PlaybooksRegistry:
@@ -171,6 +182,7 @@ class Registry:
         prometheus_enabled=PROMETHEUS_ENABLED,
     )
     _pubsub: EventsPubSub = EventsPubSub()
+    _sink_initialization_errors: bool = False
 
     def set_light_actions(self, light_actions: List[str]):
         self._light_actions = light_actions
@@ -231,3 +243,9 @@ def subscribe(self, event_name: str, handler: EventHandler):
 
     def unsubscribe(self, event_name: str, handler: EventHandler):
         self._pubsub.unsubscribe(event_name, handler)
+    
+    def set_sink_initialization_errors(self, has_errors: bool):
+        self._sink_initialization_errors = has_errors
+    
+    def has_sink_initialization_errors(self) -> bool:
+        return self._sink_initialization_errors

src/robusta/runner/config_loader.py

@@ -287,10 +287,16 @@ def __prepare_runtime_config(
         sinks_registry: SinksRegistry,
         actions_registry: ActionsRegistry,
         registry: Registry,
-    ) -> (SinksRegistry, PlaybooksRegistry):
+    ) -> tuple[SinksRegistry, PlaybooksRegistry]:
         existing_sinks = sinks_registry.get_all() if sinks_registry else {}
-        new_sinks = SinksRegistry.construct_new_sinks(runner_config.sinks_config, existing_sinks, registry)
+        new_sinks, has_sink_errors = SinksRegistry.construct_new_sinks(
+            runner_config.sinks_config, 
+            existing_sinks, 
+            registry,
+            runner_config.global_config.get("continue_on_sink_errors", False)
+        )
         sinks_registry = SinksRegistry(new_sinks)
+        registry.set_sink_initialization_errors(has_sink_errors)
 
         # TODO we will replace it with a more generic mechanism, as part of the triggers separation task
         # First, we load the internal playbooks, then add the user activated playbooks

src/robusta/utils/time_utils.py

@@ -0,0 +1,5 @@
+from datetime import datetime, timezone
+
+
+def current_utc_timestamp() -> datetime:
+    return datetime.now(timezone.utc)

tests/config/test_sink_registry.py

@@ -24,7 +24,7 @@ def setup_method(self, method):
       self.config_data = yaml.safe_load(CONFIG)
       self.runner_config = RunnerConfig(**self.config_data)
       self.registry = Registry()
-      new_sinks = SinksRegistry.construct_new_sinks(self.runner_config.sinks_config, {}, self.registry)
+      new_sinks, _ = SinksRegistry.construct_new_sinks(self.runner_config.sinks_config, {}, self.registry)
       self.sinks_registry = SinksRegistry(new_sinks)
 
 
@@ -34,20 +34,20 @@ def test_create_sinks(self):
 
   def test_delete_sinks(self):
       del self.runner_config.sinks_config[0]
-      new_sinks = SinksRegistry.construct_new_sinks(self.runner_config.sinks_config, self.sinks_registry.sinks, self.registry)
+      new_sinks, _ = SinksRegistry.construct_new_sinks(self.runner_config.sinks_config, self.sinks_registry.sinks, self.registry)
       assert list(new_sinks.keys()) == ["my_sink2"]
 
 
   def test_add_new_sinks(self):
       self.config_data["sinks_config"].insert(1, {'mail_sink': {'name': 'my_sink3', 'mailto': 'mailtos://user:password@example.com?from=a@x&to=b@y'}})
       self.runner_config = RunnerConfig(**self.config_data)
-      new_sinks = SinksRegistry.construct_new_sinks(self.runner_config.sinks_config, self.sinks_registry.sinks, self.registry)
+      new_sinks, _ = SinksRegistry.construct_new_sinks(self.runner_config.sinks_config, self.sinks_registry.sinks, self.registry)
       assert list(new_sinks.keys()) == ["my_sink1", "my_sink3", "my_sink2"]
 
 
   def test_sink_config_changed(self):
       new_email = "dev@robusta.dev"
       self.runner_config.sinks_config[0].mail_sink.mailto = new_email
-      new_sinks = SinksRegistry.construct_new_sinks(self.runner_config.sinks_config, self.sinks_registry.sinks, self.registry)
+      new_sinks, _ = SinksRegistry.construct_new_sinks(self.runner_config.sinks_config, self.sinks_registry.sinks, self.registry)
       assert new_sinks["my_sink1"].params.mailto == new_email
       assert list(new_sinks.keys()) == ["my_sink1", "my_sink2"]