Merge branch 'release/v11.2.6' of https://github.com/utmstack/UTMStack into release/v11.2.6

Author: Kbayero

plugins/modules-config/config/config.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/threatwinds/go-sdk/catcher"
 	"github.com/threatwinds/go-sdk/utils"
+	"golang.org/x/sync/errgroup"
 )
 
 var (
@@ -45,7 +46,11 @@ func (s *ConfigServer) GetModuleGroup(moduleName PluginType) *ConfigurationSecti
 
 	section, exists := s.cache[moduleName]
 	if !exists {
-		catcher.Error("module group not found", fmt.Errorf("module: %s", moduleName), map[string]any{"process": "plugin_com.utmstack.modules-config"})
+		catcher.Error("module group not found", nil, map[string]any{
+			"process": "plugin_com.utmstack.modules-config",
+			"module":  moduleName,
+		},
+		)
 		return nil
 	}
 
@@ -89,7 +94,11 @@ func (s *ConfigServer) StreamConfig(stream ConfigService_StreamConfigServer) err
 			go s.monitorDisconnect(pluginType, conn)
 
 		default:
-			catcher.Error("unexpected message type", fmt.Errorf("received: %T", payload), map[string]any{"process": "plugin_com.utmstack.modules-config"})
+			catcher.Error("unexpected message type", nil, map[string]any{
+				"process":      "plugin_com.utmstack.modules-config",
+				"message_type": fmt.Sprintf("%T", payload),
+			},
+			)
 		}
 	}
 
@@ -131,59 +140,117 @@ func (s *ConfigServer) NotifyUpdate(moduleName string, section *ConfigurationSec
 	case "CROWDSTRIKE":
 		pluginType = PluginType_CROWDSTRIKE
 	default:
-		_ = catcher.Error("unknown module name", fmt.Errorf("module: %s", moduleName), map[string]any{"process": "plugin_com.utmstack.modules-config"})
+		catcher.Error("unknown module name", nil, map[string]any{"process": "plugin_com.utmstack.modules-config", "module": moduleName})
 		return
 	}
 
 	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	s.cache[pluginType] = section
+	connectedPlugins := append([]*pluginConnection{}, s.plugins[pluginType]...)
+	s.mu.Unlock()
 
-	if len(s.plugins[pluginType]) == 0 {
+	if len(connectedPlugins) == 0 {
 		catcher.Info(fmt.Sprintf("No active connections for plugin type: %s", pluginType), map[string]any{"process": "plugin_com.utmstack.modules-config"})
 		return
 	}
 
-	for _, conn := range s.plugins[pluginType] {
+	for _, conn := range connectedPlugins {
 		err := conn.stream.Send(&BiDirectionalMessage{
 			Payload: &BiDirectionalMessage_Config{
 				Config: section,
 			},
 		})
 		if err != nil {
-			_ = catcher.Error("error sending configuration update", err, map[string]any{"process": "plugin_com.utmstack.modules-config"})
+			catcher.Error("error sending configuration update", err, map[string]any{"process": "plugin_com.utmstack.modules-config"})
 			continue
 		}
 	}
 }
 
-func (s *ConfigServer) SyncConfigs(backend string, internalKey string) {
-	var AllModules = map[string]PluginType{
-		"AWS_IAM_USER": PluginType_AWS_IAM_USER,
-		"AZURE":        PluginType_AZURE,
-		"BITDEFENDER":  PluginType_BITDEFENDER,
-		"GCP":          PluginType_GCP,
-		"O365":         PluginType_O365,
-		"SOC_AI":       PluginType_SOC_AI,
-		"SOPHOS":       PluginType_SOPHOS,
-		"CROWDSTRIKE":  PluginType_CROWDSTRIKE,
-	}
+func (s *ConfigServer) syncModuleWithRetry(
+	moduleName string,
+	pluginType PluginType,
+	backend string,
+	internalKey string,
+) error {
+	const maxRetries = 5
+	baseDelay := 2 * time.Second
 
-	for name, t := range AllModules {
-		url := fmt.Sprintf("%s/api/utm-modules/module-details-decrypted?nameShort=%s&serverId=1", backend, name)
-
-		for {
-			response, status, err := utils.DoReq[ConfigurationSection](url, nil, "GET", map[string]string{"Utm-Internal-Key": internalKey}, true)
-			if err == nil && status == http.StatusOK {
-				s.mu.Lock()
-				s.cache[t] = &response
-				s.mu.Unlock()
-				break
+	url := fmt.Sprintf("%s/api/utm-modules/module-details-decrypted?nameShort=%s&serverId=1", backend, moduleName)
+
+	for attempt := 0; attempt <= maxRetries; attempt++ {
+		response, status, err := utils.DoReq[ConfigurationSection](
+			url,
+			nil,
+			"GET",
+			map[string]string{"Utm-Internal-Key": internalKey},
+			true,
+		)
+
+		if err == nil && status == http.StatusOK {
+			s.mu.Lock()
+			s.cache[pluginType] = &response
+			connectedPlugins := append([]*pluginConnection{}, s.plugins[pluginType]...)
+			s.mu.Unlock()
+
+			if len(connectedPlugins) > 0 {
+				for _, conn := range connectedPlugins {
+					if err := conn.stream.Send(&BiDirectionalMessage{
+						Payload: &BiDirectionalMessage_Config{Config: &response},
+					}); err != nil {
+						catcher.Error(
+							"failed to send late-arrival config",
+							err,
+							map[string]any{
+								"process": "plugin_com.utmstack.modules-config",
+								"module":  moduleName,
+							},
+						)
+					}
+				}
 			}
 
-			fmt.Printf("Error fetching configuration for %s: %v, status code: %d. Retrying...\n", name, err, status)
-			time.Sleep(5 * time.Second)
+			return nil
+		}
+
+		if attempt < maxRetries {
+			delay := time.Duration(1<<attempt) * baseDelay
+			time.Sleep(delay)
 		}
 	}
+
+	return catcher.Error("failed to sync module after max retries", nil, map[string]any{
+		"process": "plugin_com.utmstack.modules-config",
+		"module":  moduleName,
+		"retries": maxRetries + 1,
+	},
+	)
+}
+
+var AllModules = map[string]PluginType{
+	"AWS_IAM_USER": PluginType_AWS_IAM_USER,
+	"AZURE":        PluginType_AZURE,
+	"BITDEFENDER":  PluginType_BITDEFENDER,
+	"GCP":          PluginType_GCP,
+	"O365":         PluginType_O365,
+	"SOC_AI":       PluginType_SOC_AI,
+	"SOPHOS":       PluginType_SOPHOS,
+	"CROWDSTRIKE":  PluginType_CROWDSTRIKE,
+}
+
+func (s *ConfigServer) SyncConfigs(backend string, internalKey string) {
+	g := errgroup.Group{}
+	g.SetLimit(4)
+
+	for name, t := range AllModules {
+		g.Go(func() error {
+			return s.syncModuleWithRetry(name, t, backend, internalKey)
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		catcher.Error("module config sync failed", err, map[string]any{
+			"process": "plugin_com.utmstack.modules-config",
+		})
+	}
 }