feat: add parallel notification delivery to improve performance

gblanc-1a · gblanc-1a · commit 900d042834d7 · 2025-10-02T14:49:34.000+02:00
Implements parallel notification delivery with bounded concurrency to significantly improve performance and prevent queue overload when sending notifications to multiple destinations. Fixes #146 This change addresses the issue where notification-engine becomes overloaded when processing resources with multiple notification destinations, especially when notification services are slow to respond. The sequential processing was causing: - Queue buildup and missed notifications - Worker threads blocked during network I/O - Processing delays as subscription count increases - Annotation state issues due to slow processing Changes include: Controller enhancements: - Send notifications in parallel using goroutines with worker pool pattern - Add WithMaxConcurrentNotifications() option for programmatic configuration - Implement thread-safe state management with proper mutex protection - Extract sendSingleNotification() helper method for better testability - Comprehensive test coverage with race condition validation Configuration support: - Add maxConcurrentNotifications config option (default: 50) - ConfigMap-based configuration with validation and warnings - Support for values 1-1000 with recommendations for optimal ranges - Tests for configuration parsing and validation Documentation: - Add Performance Configuration section to README - Document configuration options and best practices - Provide guidance on tuning for different scenarios - Include examples for high-volume environments Performance impact: - Before: n destinations × avg_latency (sequential processing) - After: max(latencies) with bounded concurrency (parallel processing) - Example: 10 destinations @ 200ms each: 2000ms → 200ms (~90% improvement) - Prevents queue buildup in high-volume environments (160+ apps) Testing: - Parallel execution with timing validation - Error handling in concurrent scenarios - Configuration validation (default, custom, invalid values) - Thread-safety verified with race detector Breaking changes: None - Backward compatible with existing configurations - Default behavior provides automatic performance improvements - Optional configuration for specific tuning needs Signed-off-by: gblanc-1a <gblanc-1a@users.noreply.github.com>
diff --git a/README.md b/README.md
@@ -42,6 +42,12 @@ data:
 
   service.slack: |
     token: $slack-token
+  
+  # Optional: Configure maximum concurrent notifications (default: 50)
+  # Valid range: 1-1000 (recommended: 10-200)
+  # Higher values = faster notification delivery but more resource usage
+  # Lower values = slower delivery but less resource consumption
+  maxConcurrentNotifications: "25"
 ---
 apiVersion: v1
 kind: Secret
@@ -76,6 +82,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: 50)
+	MaxConcurrentNotifications int
+	Namespace                  string
+	IsSelfServiceConfig        bool
 }
 
 // Returns list of destinations for the specified trigger
@@ -74,11 +76,12 @@ func replaceStringSecret(val string, secretValues map[string][]byte) string {
 // 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: 50, // Default value
 	}
 	if subscriptionYaml, ok := configMap.Data["subscriptions"]; ok {
 		if err := yaml.Unmarshal([]byte(subscriptionYaml), &cfg.Subscriptions); err != nil {
@@ -92,6 +95,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', using default: 50", maxConcurrentYaml)
+		} else {
+			switch {
+			case maxConcurrent <= 0:
+				log.Warnf("maxConcurrentNotifications must be positive, got %d, using default: 50", maxConcurrent)
+			case maxConcurrent > 1000:
+				log.Warnf("maxConcurrentNotifications value %d is very high (>1000), consider using a lower value for better resource management", 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: 50,
+		},
+		{
+			name: "zero uses default",
+			data: map[string]string{
+				"maxConcurrentNotifications": "0",
+			},
+			expected: 50,
+		},
+		{
+			name: "negative uses default",
+			data: map[string]string{
+				"maxConcurrentNotifications": "-10",
+			},
+			expected: 50,
+		},
+		{
+			name: "invalid non-numeric uses default",
+			data: map[string]string{
+				"maxConcurrentNotifications": "invalid",
+			},
+			expected: 50,
+		},
+		{
+			name: "empty string uses default",
+			data: map[string]string{
+				"maxConcurrentNotifications": "",
+			},
+			expected: 50,
+		},
+		{
+			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,7 @@ 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"
 )
 
 // NotificationDelivery represents a notification that was delivered
@@ -98,6 +100,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 +136,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: 50, // Default limit
 		toUnstructured: func(obj metav1.Object) (*unstructured.Unstructured, error) {
 			res, ok := obj.(*unstructured.Unstructured)
 			if !ok {
@@ -155,16 +169,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 +202,84 @@ 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.
+// The actual api.Send call is performed without holding any locks to enable parallel execution.
+// logEntry is safe to use concurrently as logrus.Entry is immutable and logging operations are thread-safe.
+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 (if explicitly set and != default)
+	//           2) ConfigMap config
+	//           3) Default (50)
+	maxConcurrent := c.maxConcurrentNotifications
+	if cfg.MaxConcurrentNotifications > 0 && cfg.MaxConcurrentNotifications != 50 {
+		// ConfigMap has a non-default value, use it unless programmatic config overrides
+		if maxConcurrent == 50 || maxConcurrent == 0 {
+			maxConcurrent = cfg.MaxConcurrentNotifications
+			logEntry.Debugf("Using maxConcurrentNotifications from ConfigMap: %d", maxConcurrent)
+		} else {
+			logEntry.Debugf("Using maxConcurrentNotifications from programmatic config: %d", maxConcurrent)
+		}
+	} else if maxConcurrent == 0 {
+		// Fallback to default if not set anywhere
+		maxConcurrent = 50
+	}
+
 	un, err := c.toUnstructured(resource)
 	if err != nil {
 		return nil, err
@@ -219,6 +303,12 @@ 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
+			// TODO: add metrics for concurrency usage (current concurrent sends, max reached, wait time)
+			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 +318,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
