fix: add parallel notification delivery

gblanc-1a · gblanc-1a · commit ba683ce66a03 · 2025-10-02T15:50:19.000+02:00
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 <gblanc-1a@users.noreply.github.com>
diff --git a/README.md b/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)
diff --git a/pkg/api/config.go b/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 {
diff --git a/pkg/api/config_test.go b/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"),
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"reflect"
 	"runtime/debug"
+	"sync"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -21,6 +22,12 @@ import (
 	"github.com/argoproj/notifications-engine/pkg/api"
 	"github.com/argoproj/notifications-engine/pkg/services"
 	"github.com/argoproj/notifications-engine/pkg/subscriptions"
+	"github.com/argoproj/notifications-engine/pkg/triggers"
+)
+
+const (
+	// DefaultMaxConcurrentNotifications is the default maximum number of concurrent notification deliveries
+	DefaultMaxConcurrentNotifications = 10
 )
 
 // NotificationDelivery represents a notification that was delivered
@@ -98,6 +105,17 @@ func WithEventCallback(f func(eventSequence NotificationEventSequence)) Opts {
 	}
 }
 
+// WithMaxConcurrentNotifications sets the maximum number of concurrent notification
+// deliveries per resource. This helps prevent resource exhaustion when sending
+// notifications to many destinations. Default is 50 if not specified.
+func WithMaxConcurrentNotifications(maxConcurrent int) Opts {
+	return func(ctrl *notificationController) {
+		if maxConcurrent > 0 {
+			ctrl.maxConcurrentNotifications = maxConcurrent
+		}
+	}
+}
+
 func NewController(
 	client dynamic.NamespaceableResourceInterface,
 	informer cache.SharedIndexInformer,
@@ -123,11 +141,12 @@ func NewController(
 	)
 
 	ctrl := &notificationController{
-		client:          client,
-		informer:        informer,
-		queue:           queue,
-		metricsRegistry: NewMetricsRegistry(""),
-		apiFactory:      apiFactory,
+		client:                     client,
+		informer:                   informer,
+		queue:                      queue,
+		metricsRegistry:            NewMetricsRegistry(""),
+		apiFactory:                 apiFactory,
+		maxConcurrentNotifications: DefaultMaxConcurrentNotifications,
 		toUnstructured: func(obj metav1.Object) (*unstructured.Unstructured, error) {
 			res, ok := obj.(*unstructured.Unstructured)
 			if !ok {
@@ -155,16 +174,17 @@ func NewControllerWithNamespaceSupport(
 }
 
 type notificationController struct {
-	client            dynamic.NamespaceableResourceInterface
-	informer          cache.SharedIndexInformer
-	queue             workqueue.TypedRateLimitingInterface[string]
-	apiFactory        api.Factory
-	metricsRegistry   *MetricsRegistry
-	skipProcessing    func(obj metav1.Object) (bool, string)
-	alterDestinations func(obj metav1.Object, destinations services.Destinations, cfg api.Config) services.Destinations
-	toUnstructured    func(obj metav1.Object) (*unstructured.Unstructured, error)
-	eventCallback     func(eventSequence NotificationEventSequence)
-	namespaceSupport  bool
+	client                     dynamic.NamespaceableResourceInterface
+	informer                   cache.SharedIndexInformer
+	queue                      workqueue.TypedRateLimitingInterface[string]
+	apiFactory                 api.Factory
+	metricsRegistry            *MetricsRegistry
+	skipProcessing             func(obj metav1.Object) (bool, string)
+	alterDestinations          func(obj metav1.Object, destinations services.Destinations, cfg api.Config) services.Destinations
+	toUnstructured             func(obj metav1.Object) (*unstructured.Unstructured, error)
+	eventCallback              func(eventSequence NotificationEventSequence)
+	namespaceSupport           bool
+	maxConcurrentNotifications int
 }
 
 func (c *notificationController) Run(threadiness int, stopCh <-chan struct{}) {
@@ -187,15 +207,74 @@ func (c *notificationController) isSelfServiceConfigureApi(api api.API) bool {
 	return c.namespaceSupport && api.GetConfig().IsSelfServiceConfig
 }
 
+// notificationResult encapsulates the result of sending a single notification.
+// It is used to communicate results from parallel notification goroutines back to the main processing loop.
+type notificationResult struct {
+	success  bool                 // success indicates whether the notification was sent successfully
+	err      error                // err contains the error if the send failed
+	delivery NotificationDelivery // delivery contains the notification delivery information
+}
+
+// sendSingleNotification sends a notification to a single destination and returns the result.
+// Thread-safe: designed for concurrent execution. All parameters are either read-only or thread-safe.
+// The api.Send call is performed without locks to enable parallel execution.
+func (c *notificationController) sendSingleNotification(
+	api api.API,
+	un *unstructured.Unstructured,
+	resource metav1.Object,
+	trigger string,
+	cr triggers.ConditionResult,
+	destination services.Destination,
+	templates []string,
+	apiNamespace string,
+	logEntry *log.Entry,
+) notificationResult {
+	logEntry.Infof("Sending notification about condition '%s.%s' to '%v' using the configuration in namespace %s",
+		trigger, cr.Key, destination, apiNamespace)
+
+	err := api.Send(un.Object, templates, destination)
+
+	result := notificationResult{
+		success: err == nil,
+		err:     err,
+		delivery: NotificationDelivery{
+			Trigger:         trigger,
+			Destination:     destination,
+			AlreadyNotified: false,
+		},
+	}
+
+	if err != nil {
+		logEntry.Errorf("Failed to notify recipient %s defined in resource %s/%s: %v using the configuration in namespace %s",
+			destination, resource.GetNamespace(), resource.GetName(), err, apiNamespace)
+		result.err = fmt.Errorf("failed to deliver notification %s to %s: %w using the configuration in namespace %s",
+			trigger, destination, err, apiNamespace)
+	} else {
+		logEntry.Debugf("Notification %s was sent using the configuration in namespace %s",
+			destination.Recipient, apiNamespace)
+	}
+
+	return result
+}
+
 func (c *notificationController) processResourceWithAPI(api api.API, resource metav1.Object, logEntry *log.Entry, eventSequence *NotificationEventSequence) (map[string]string, error) {
-	apiNamespace := api.GetConfig().Namespace
+	cfg := api.GetConfig()
+	apiNamespace := cfg.Namespace
 	notificationsState := NewStateFromRes(resource)
 
-	destinations := c.getDestinations(resource, api.GetConfig())
+	destinations := c.getDestinations(resource, cfg)
 	if len(destinations) == 0 {
 		return resource.GetAnnotations(), nil
 	}
 
+	// Determine max concurrent notifications
+	// Priority: 1) Programmatic config 2) ConfigMap config 3) Default
+	maxConcurrent := c.maxConcurrentNotifications
+	if maxConcurrent == DefaultMaxConcurrentNotifications && cfg.MaxConcurrentNotifications != DefaultMaxConcurrentNotifications {
+		maxConcurrent = cfg.MaxConcurrentNotifications
+		logEntry.Debugf("Using maxConcurrentNotifications from ConfigMap: %d", maxConcurrent)
+	}
+
 	un, err := c.toUnstructured(resource)
 	if err != nil {
 		return nil, err
@@ -219,6 +298,11 @@ func (c *notificationController) processResourceWithAPI(api api.API, resource me
 				continue
 			}
 
+			// send notifications in parallel using goroutines with a worker pool
+			var wg sync.WaitGroup
+			var notificationsMutex sync.Mutex
+			semaphore := make(chan struct{}, maxConcurrent)
+
 			for _, to := range destinations {
 				if changed := notificationsState.SetAlreadyNotified(c.isSelfServiceConfigureApi(api), apiNamespace, trigger, cr, to, true); !changed {
 					logEntry.Infof("Notification about condition '%s.%s' already sent to '%v' using the configuration in namespace %s", trigger, cr.Key, to, apiNamespace)
@@ -228,24 +312,32 @@ func (c *notificationController) processResourceWithAPI(api api.API, resource me
 						AlreadyNotified: true,
 					})
 				} else {
-					logEntry.Infof("Sending notification about condition '%s.%s' to '%v' using the configuration in namespace %s", trigger, cr.Key, to, apiNamespace)
-					if err := api.Send(un.Object, cr.Templates, to); err != nil {
-						logEntry.Errorf("Failed to notify recipient %s defined in resource %s/%s: %v using the configuration in namespace %s",
-							to, resource.GetNamespace(), resource.GetName(), err, apiNamespace)
-						notificationsState.SetAlreadyNotified(c.isSelfServiceConfigureApi(api), apiNamespace, trigger, cr, to, false)
-						c.metricsRegistry.IncDeliveriesCounter(trigger, to.Service, false)
-						eventSequence.addError(fmt.Errorf("failed to deliver notification %s to %s: %w using the configuration in namespace %s", trigger, to, err, apiNamespace))
-					} else {
-						logEntry.Debugf("Notification %s was sent using the configuration in namespace %s", to.Recipient, apiNamespace)
-						c.metricsRegistry.IncDeliveriesCounter(trigger, to.Service, true)
-						eventSequence.addDelivered(NotificationDelivery{
-							Trigger:         trigger,
-							Destination:     to,
-							AlreadyNotified: false,
-						})
-					}
+					wg.Add(1)
+					semaphore <- struct{}{}
+					go func(destination services.Destination, templates []string) {
+						defer func() {
+							<-semaphore
+							wg.Done()
+						}()
+
+						result := c.sendSingleNotification(api, un, resource, trigger, cr, destination, templates, apiNamespace, logEntry)
+
+						notificationsMutex.Lock()
+						defer notificationsMutex.Unlock()
+
+						if !result.success {
+							notificationsState.SetAlreadyNotified(c.isSelfServiceConfigureApi(api), apiNamespace, trigger, cr, destination, false)
+							c.metricsRegistry.IncDeliveriesCounter(trigger, destination.Service, false)
+							eventSequence.addError(result.err)
+						} else {
+							c.metricsRegistry.IncDeliveriesCounter(trigger, destination.Service, true)
+							eventSequence.addDelivered(result.delivery)
+						}
+					}(to, cr.Templates)
 				}
 			}
+
+			wg.Wait()
 		}
 	}
 
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
