refactor(crowdstrike): migrate from polling to real-time event streaming

- Replaced polling-based event collection with real-time streaming architecture
- Implemented persistent stream management with automatic reconnection
- Added dynamic configuration reloading with live stream updates
- Introduced per-stream offset tracking to prevent duplicate events

Author: JocLRojas

plugins/crowdstrike/check.go

@@ -58,7 +58,7 @@ func infiniteRetryIfXError(f func() error, exception string) error {
 				_ = catcher.Error("An error occurred (%s), will keep retrying indefinitely...", err, map[string]any{"process": "plugin_com.utmstack.crowdstrike"})
 				xErrorWasLogged = true
 			}
-			time.Sleep(wait)
+			time.Sleep(reconnectDelay)
 			continue
 		}
 

plugins/crowdstrike/config/config.go

@@ -23,13 +23,17 @@ const (
 )
 
 var (
-	cnf *ConfigurationSection
-	mu  sync.Mutex
-
+	cnf               *ConfigurationSection
+	mu                sync.Mutex
+	configUpdateChan  chan *ConfigurationSection
 	internalKey       string
 	modulesConfigHost string
 )
 
+func init() {
+	configUpdateChan = make(chan *ConfigurationSection, 1)
+}
+
 func GetConfig() *ConfigurationSection {
 	mu.Lock()
 	defer mu.Unlock()
@@ -39,6 +43,10 @@ func GetConfig() *ConfigurationSection {
 	return cnf
 }
 
+func GetConfigUpdateChannel() <-chan *ConfigurationSection {
+	return configUpdateChan
+}
+
 func StartConfigurationSystem() {
 	for {
 		pluginConfig := plugins.PluginCfg("com.utmstack")
@@ -133,7 +141,18 @@ func StartConfigurationSystem() {
 			switch message := in.Payload.(type) {
 			case *BiDirectionalMessage_Config:
 				catcher.Info("Received configuration update", map[string]any{"process": "plugin_com.utmstack.crowdstrike"})
+
+				mu.Lock()
 				cnf = message.Config
+				mu.Unlock()
+
+				select {
+				case configUpdateChan <- message.Config:
+
+				default:
+
+					catcher.Info("Configuration update channel full, skipping notification", map[string]any{"process": "plugin_com.utmstack.crowdstrike"})
+				}
 			}
 		}
 	}