ROB-1323 cluster status init crash take 2 (#1835)

* emit event when config load is done

* call status and discover cluster after config is done

* change event on config reload

* change event on config reload

* just check if thread attribute exists

---------

Co-authored-by: arik <alon.arik@gmail.com>

Author: RoiGlinik

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

@@ -86,9 +86,9 @@ def __init__(self, sink_config: RobustaSinkConfigWrapper, registry):
         )
         self.__rrm_checker = RRM(dal=self.dal, cluster=self.cluster_name, account_id=self.account_id)
         self.__pods_running_count: int = 0
-        self.__update_cluster_status()  # send runner version initially, then force prometheus alert time periodically.
 
         self.registry.subscribe("scan_updated", self)
+        self.registry.subscribe("config_reload", self)
 
         # start cluster discovery
         self.__active = True
@@ -101,14 +101,14 @@ def __init__(self, sink_config: RobustaSinkConfigWrapper, registry):
         self.__jobs_cache_initialized: bool = False
         self.__helm_releases_cache: Optional[Dict[str, HelmRelease]] = None
         self.__init_service_resolver()
-        self.__thread = threading.Thread(target=self.__discover_cluster)
         self.__watchdog_thread = threading.Thread(target=self.__discovery_watchdog)
-        self.__thread.start()
         self.__watchdog_thread.start()
 
     def handle_event(self, event_name: str, **kwargs):
         if event_name == "scan_updated":
             self._on_scan_updated(**kwargs)
+        elif event_name == "config_reload":
+            self._on_config_reload(**kwargs)
         else:
             logging.warning("RobustaSink subscriber called with unknown event")
 
@@ -120,6 +120,14 @@ def _on_scan_updated(
 
         self.dal.set_scan_state(scan_id, state, metadata)
 
+    def _on_config_reload(self) -> None:
+        # make sure cluster status and periodic check start after all config has been reloaded successfuly.
+        self.__update_cluster_status()  # send runner version initially, then force prometheus alert time periodically.
+
+        if not hasattr(self, '_thread'):
+            self._thread = threading.Thread(target=self.__discover_cluster, daemon=True)
+            self._thread.start()
+
     def set_cluster_active(self, active: bool):
         self.dal.set_cluster_active(active)
 

src/robusta/runner/config_loader.py

@@ -64,16 +64,17 @@ def __init__(
         self.event_handler = event_handler
         self.root_playbook_path = PLAYBOOKS_ROOT
         self.reload_lock = threading.RLock()
-        self.watcher = FileSystemWatcher(self.root_playbook_path, self.__reload_playbook_packages)
-        self.conf_watcher = FileSystemWatcher(self.config_file_path, self.__reload_playbook_packages)
-        self.__reload_playbook_packages("initialization")
+        self.watcher = FileSystemWatcher(self.root_playbook_path, self.reload)
+        self.conf_watcher = FileSystemWatcher(self.config_file_path, self.reload)
+        self.reload("initialization")
 
     def close(self):
         self.watcher.stop_watcher()
         self.conf_watcher.stop_watcher()
 
     def reload(self, description: str):
         self.__reload_playbook_packages(description)
+        self.registry.get_event_emitter().emit_event("config_reload")
 
     def __reload_scheduler(self, playbooks_registry: PlaybooksRegistry):
         scheduler = self.registry.get_scheduler()