Set cluster inactive on SIGINT and active on start (#1100)

* Set cluster inactive on SIGINT and active on start

* Make set_cluster_active available for all sinks

* Add logging on setting cluster active

* Close websocket on SIGINT

* set_cluster_active change updated_at

src/robusta/core/playbooks/playbooks_event_handler_impl.py

@@ -17,6 +17,7 @@
 from robusta.core.reporting import MarkdownBlock
 from robusta.core.reporting.base import Finding
 from robusta.core.reporting.consts import SYNC_RESPONSE_SINK
+from robusta.core.sinks.robusta import RobustaSink
 from robusta.core.sinks.robusta.dal.model_conversion import ModelConversion
 from robusta.model.alert_relabel_config import AlertRelabel
 from robusta.model.config import Registry
@@ -334,8 +335,20 @@ def is_healthy(
             return True
         return all(sink.is_healthy() for sink in sinks_registry.get_all().values())
 
+    def set_cluster_active(self, active: bool):
+        logging.info(f"Setting cluster active to {active}")
+        for sink in self.registry.get_sinks().get_all().values():
+            sink.set_cluster_active(active)
+
     def handle_sigint(self, sig, frame):
         logging.info("SIGINT handler called")
+
         if not self.is_healthy():  # dump stuck trace only when the runner is unhealthy
             StackTracer.dump()
+
+        receiver = self.registry.get_receiver()
+        if receiver is not None:
+            receiver.stop()
+
+        self.set_cluster_active(False)
         sys.exit(0)

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

@@ -527,3 +527,16 @@ def __update_token_patch(self, event, session):
         logging.debug(f"Event {event}, Session {session}")
         if session and event == "TOKEN_REFRESHED":
             self.client.postgrest.auth(session.access_token)
+
+    def set_cluster_active(self, active: bool) -> None:
+        try:
+            (
+                self.client.table(CLUSTERS_STATUS_TABLE)
+                .update({"active": active, "updated_at": "now()"})
+                .eq("cluster_id", self.cluster)
+                .eq("account_id", self.account_id)
+                .execute()
+            )
+        except Exception as e:
+            logging.error(f"Failed to set cluster status active=False error: {e}")
+            self.handle_supabase_error()

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

@@ -91,6 +91,9 @@ def __init__(self, sink_config: RobustaSinkConfigWrapper, registry):
         self.__thread.start()
         self.__watchdog_thread.start()
 
+    def set_cluster_active(self, active: bool):
+        self.dal.set_cluster_active(active)
+
     def __init_service_resolver(self):
         """
         Init service resolver from the service stored in storage.
@@ -342,7 +345,9 @@ def __to_node_info(cls, node: Union[V1Node, Node]) -> Dict:
         return node_info
 
     @classmethod
-    def __from_api_server_node(cls, api_server_node: Union[V1Node, Node], pod_requests_list: List[PodResources]) -> NodeInfo:
+    def __from_api_server_node(
+        cls, api_server_node: Union[V1Node, Node], pod_requests_list: List[PodResources]
+    ) -> NodeInfo:
         addresses = api_server_node.status.addresses or []
         external_addresses = [address for address in addresses if "externalip" in address.type.lower()]
         external_ip = ",".join([addr.address for addr in external_addresses])
@@ -353,8 +358,12 @@ def __from_api_server_node(cls, api_server_node: Union[V1Node, Node], pod_reques
         capacity = api_server_node.status.capacity or {}
         allocatable = api_server_node.status.allocatable or {}
         # V1Node and Node use snake case and camelCase respectively, handle this for more than 1 word attributes.
-        creation_ts = getattr(api_server_node.metadata, "creation_timestamp", None) or getattr(api_server_node.metadata, "creationTimestamp", None)
-        version = getattr(api_server_node.metadata, "resource_version", None) or getattr(api_server_node.metadata, "resourceVersion", None)
+        creation_ts = getattr(api_server_node.metadata, "creation_timestamp", None) or getattr(
+            api_server_node.metadata, "creationTimestamp", None
+        )
+        version = getattr(api_server_node.metadata, "resource_version", None) or getattr(
+            api_server_node.metadata, "resourceVersion", None
+        )
         return NodeInfo(
             name=api_server_node.metadata.name,
             node_creation_time=str(creation_ts),
@@ -371,7 +380,7 @@ def __from_api_server_node(cls, api_server_node: Union[V1Node, Node], pod_reques
             pods_count=len(pod_requests_list),
             pods=",".join([pod_req.pod_name for pod_req in pod_requests_list]),
             node_info=cls.__to_node_info(api_server_node),
-            resource_version=int(version) if version else 0
+            resource_version=int(version) if version else 0,
         )
 
     def __publish_new_nodes(self, current_nodes: V1NodeList, node_requests: Dict[str, List[PodResources]]):
@@ -612,7 +621,6 @@ def __update_node(self, new_node: Node, operation: K8sOperationType):
             self.__discovery_metrics.on_nodes_updated(1)
 
     def __update_job(self, new_job: Job, operation: K8sOperationType):
-
         new_info = JobInfo.from_api_server(new_job, [])
         job_key = new_info.get_service_key()
         with self.services_publish_lock:

src/robusta/core/sinks/sink_base.py

@@ -42,3 +42,6 @@ def is_healthy(self) -> bool:
 
     def handle_service_diff(self, new_obj: Any, operation: K8sOperationType):
         pass
+
+    def set_cluster_active(self, active: bool):
+        pass

src/robusta/model/config.py

@@ -40,7 +40,6 @@ def construct_new_sinks(
         existing_sinks: Dict[str, SinkBase],
         registry,
     ) -> Dict[str, SinkBase]:
-
         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]

src/robusta/runner/main.py

@@ -45,7 +45,9 @@ def main():
         logging.info(f"Running alerts workers pool of {ALERT_BUILDER_WORKERS}")
 
     Web.init(event_handler, loader)
+
     signal.signal(signal.SIGINT, event_handler.handle_sigint)
+    event_handler.set_cluster_active(True)
     Web.run()  # blocking
     loader.close()