Allow ignoring sinks init errors

Fix finding dates timezone issues

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

@@ -4,7 +4,7 @@
 import urllib.parse
 import uuid
 from abc import ABC, abstractmethod
-from datetime import datetime
+from datetime import datetime, timezone
 from enum import Enum
 from typing import Any, Dict, List, Optional, Union
 from urllib.parse import urlencode
@@ -289,7 +289,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 datetime.now(timezone.utc)
         self.ends_at = ends_at
         self.dirty = False
 

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

@@ -2,7 +2,7 @@
 import json
 import logging
 import uuid
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any, Dict, List
 
 from robusta.core.model.env_vars import ENABLE_GRAPH_BLOCK
@@ -52,7 +52,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(datetime.now(timezone.utc)),
         }
 
         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