fix: add parallel notification delivery

Implements parallel notification delivery with configurable concurrency

Solve performance issues described in #146 and slow or failing destinations

Adds maxConcurrentNotifications config option (default: 10)

Signed-off-by: gblanc-1a <[email protected]>

Author: gblanc-1a

README.md

@@ -42,6 +42,13 @@ data:
 
   service.slack: |
     token: $slack-token
+  
+  # Optional: Maximum concurrent notification deliveries (default: 10)
+  # Controls parallel processing of notifications to multiple destinations.
+  # Higher values speed up delivery when destinations are slow/timing out but increase resource usage.
+  # Recommended: 10-25 (small/medium clusters), 50-100 (large clusters)
+  # Note: Some services have rate limits; start low and increase if needed.
+  maxConcurrentNotifications: "25"
 ---
 apiVersion: v1
 kind: Secret
@@ -76,6 +83,7 @@ notifications.argoproj.io/subscriptions: |
       - service: slack
         recipients: [my-channel-21, my-channel-22]
 ```
+
 ## Getting Started
 
 Ready to add notifications to your project? Check out sample notifications for [cert-manager](./examples/certmanager/README.md)

pkg/api/config.go

@@ -29,8 +29,10 @@ type Config struct {
 	DefaultTriggers []string
 	// ServiceDefaultTriggers holds list of default triggers per service
 	ServiceDefaultTriggers map[string][]string
-	Namespace              string
-	IsSelfServiceConfig    bool
+	// MaxConcurrentNotifications is the maximum number of notifications to send concurrently (default: 10)
+	MaxConcurrentNotifications int
+	Namespace                  string
+	IsSelfServiceConfig        bool
 }
 
 // Returns list of destinations for the specified trigger
@@ -71,14 +73,20 @@ func replaceStringSecret(val string, secretValues map[string][]byte) string {
 	})
 }
 
+const (
+	// DefaultMaxConcurrentNotifications is the default maximum number of concurrent notification deliveries
+	DefaultMaxConcurrentNotifications = 10
+)
+
 // ParseConfig retrieves Config from given ConfigMap and Secret
 func ParseConfig(configMap *corev1.ConfigMap, secret *corev1.Secret) (*Config, error) {
 	cfg := Config{
-		Services:               map[string]ServiceFactory{},
-		Triggers:               map[string][]triggers.Condition{},
-		ServiceDefaultTriggers: map[string][]string{},
-		Templates:              map[string]services.Notification{},
-		Namespace:              configMap.Namespace,
+		Services:                   map[string]ServiceFactory{},
+		Triggers:                   map[string][]triggers.Condition{},
+		ServiceDefaultTriggers:     map[string][]string{},
+		Templates:                  map[string]services.Notification{},
+		Namespace:                  configMap.Namespace,
+		MaxConcurrentNotifications: DefaultMaxConcurrentNotifications,
 	}
 	if subscriptionYaml, ok := configMap.Data["subscriptions"]; ok {
 		if err := yaml.Unmarshal([]byte(subscriptionYaml), &cfg.Subscriptions); err != nil {
@@ -92,6 +100,23 @@ func ParseConfig(configMap *corev1.ConfigMap, secret *corev1.Secret) (*Config, e
 		}
 	}
 
+	if maxConcurrentYaml, ok := configMap.Data["maxConcurrentNotifications"]; ok {
+		var maxConcurrent int
+		if err := yaml.Unmarshal([]byte(maxConcurrentYaml), &maxConcurrent); err != nil {
+			log.Warnf("Invalid maxConcurrentNotifications value '%s' (must be a positive integer), using default: %d", maxConcurrentYaml, DefaultMaxConcurrentNotifications)
+		} else {
+			switch {
+			case maxConcurrent <= 0:
+				log.Warnf("maxConcurrentNotifications must be positive, got %d, using default: %d", maxConcurrent, DefaultMaxConcurrentNotifications)
+			case maxConcurrent > 1000:
+				log.Warnf("maxConcurrentNotifications value %d is very high (>1000), consider using a lower value", maxConcurrent)
+				cfg.MaxConcurrentNotifications = maxConcurrent
+			default:
+				cfg.MaxConcurrentNotifications = maxConcurrent
+			}
+		}
+	}
+
 	for k, v := range configMap.Data {
 		parts := strings.Split(k, ".")
 		switch {

pkg/api/config_test.go

@@ -62,6 +62,80 @@ func TestParseConfig_DefaultServiceTriggers(t *testing.T) {
 	}, cfg.ServiceDefaultTriggers)
 }
 
+func TestParseConfig_MaxConcurrentNotifications(t *testing.T) {
+	tests := []struct {
+		name     string
+		data     map[string]string
+		expected int
+	}{
+		{
+			name: "valid positive value",
+			data: map[string]string{
+				"maxConcurrentNotifications": "25",
+			},
+			expected: 25,
+		},
+		{
+			name: "valid value 100",
+			data: map[string]string{
+				"maxConcurrentNotifications": "100",
+			},
+			expected: 100,
+		},
+		{
+			name:     "not set uses default",
+			data:     map[string]string{},
+			expected: DefaultMaxConcurrentNotifications,
+		},
+		{
+			name: "zero uses default",
+			data: map[string]string{
+				"maxConcurrentNotifications": "0",
+			},
+			expected: DefaultMaxConcurrentNotifications,
+		},
+		{
+			name: "negative uses default",
+			data: map[string]string{
+				"maxConcurrentNotifications": "-10",
+			},
+			expected: DefaultMaxConcurrentNotifications,
+		},
+		{
+			name: "invalid non-numeric uses default",
+			data: map[string]string{
+				"maxConcurrentNotifications": "invalid",
+			},
+			expected: DefaultMaxConcurrentNotifications,
+		},
+		{
+			name: "empty string uses default",
+			data: map[string]string{
+				"maxConcurrentNotifications": "",
+			},
+			expected: DefaultMaxConcurrentNotifications,
+		},
+		{
+			name: "very high value is accepted with warning",
+			data: map[string]string{
+				"maxConcurrentNotifications": "1500",
+			},
+			expected: 1500,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg, err := ParseConfig(&corev1.ConfigMap{Data: tt.data}, emptySecret)
+			if !assert.NoError(t, err) {
+				return
+			}
+			assert.Equal(t, tt.expected, cfg.MaxConcurrentNotifications,
+				"MaxConcurrentNotifications should match expected value")
+		})
+	}
+}
+
 func TestReplaceStringSecret_KeyPresent(t *testing.T) {
 	val := replaceStringSecret("hello $secret-value", map[string][]byte{
 		"secret-value": []byte("world"),