fix: Process webhook refresh in background to not block the request (a…

…rgoproj#14269) (argoproj#18173)

Signed-off-by: dhruvang1 <[email protected]>

applicationset/webhook/webhook.go

@@ -10,6 +10,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
@@ -26,16 +27,20 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+const payloadQueueSize = 50000
+
 var errBasicAuthVerificationFailed = errors.New("basic auth verification failed")
 
 type WebhookHandler struct {
+	sync.WaitGroup         // for testing
 	namespace              string
 	github                 *github.Webhook
 	gitlab                 *gitlab.Webhook
 	azuredevops            *azuredevops.Webhook
 	azuredevopsAuthHandler func(r *http.Request) error
 	client                 client.Client
 	generators             map[string]generators.Generator
+	queue                  chan interface{}
 }
 
 type gitGeneratorInfo struct {
@@ -66,7 +71,7 @@ type prGeneratorGitlabInfo struct {
 	APIHostname string
 }
 
-func NewWebhookHandler(namespace string, argocdSettingsMgr *argosettings.SettingsManager, client client.Client, generators map[string]generators.Generator) (*WebhookHandler, error) {
+func NewWebhookHandler(namespace string, webhookParallelism int, argocdSettingsMgr *argosettings.SettingsManager, client client.Client, generators map[string]generators.Generator) (*WebhookHandler, error) {
 	// register the webhook secrets stored under "argocd-secret" for verifying incoming payloads
 	argocdSettings, err := argocdSettingsMgr.GetSettings()
 	if err != nil {
@@ -94,15 +99,36 @@ func NewWebhookHandler(namespace string, argocdSettingsMgr *argosettings.Setting
 		return nil
 	}
 
-	return &WebhookHandler{
+	webhookHandler := &WebhookHandler{
 		namespace:              namespace,
 		github:                 githubHandler,
 		gitlab:                 gitlabHandler,
 		azuredevops:            azuredevopsHandler,
 		azuredevopsAuthHandler: azuredevopsAuthHandler,
 		client:                 client,
 		generators:             generators,
-	}, nil
+		queue:                  make(chan interface{}, payloadQueueSize),
+	}
+
+	webhookHandler.startWorkerPool(webhookParallelism)
+
+	return webhookHandler, nil
+}
+
+func (h *WebhookHandler) startWorkerPool(webhookParallelism int) {
+	for i := 0; i < webhookParallelism; i++ {
+		h.Add(1)
+		go func() {
+			defer h.Done()
+			for {
+				payload, ok := <-h.queue
+				if !ok {
+					return
+				}
+				h.HandleEvent(payload)
+			}
+		}()
+	}
 }
 
 func (h *WebhookHandler) HandleEvent(payload interface{}) {
@@ -176,7 +202,12 @@ func (h *WebhookHandler) Handler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	h.HandleEvent(payload)
+	select {
+	case h.queue <- payload:
+	default:
+		log.Info("Queue is full, discarding webhook payload")
+		http.Error(w, "Queue is full, discarding webhook payload", http.StatusServiceUnavailable)
+	}
 }
 
 func parseRevision(ref string) string {

applicationset/webhook/webhook_test.go

@@ -178,6 +178,7 @@ func TestWebhookHandler(t *testing.T) {
 	}
 
 	namespace := "test"
+	webhookParallelism := 10
 	fakeClient := newFakeClient(namespace)
 	scheme := runtime.NewScheme()
 	err := v1alpha1.AddToScheme(scheme)
@@ -206,7 +207,7 @@ func TestWebhookHandler(t *testing.T) {
 				fakeAppWithMergeAndNestedGitGenerator("merge-nested-git-github", namespace, "https://github.com/org/repo"),
 			).Build()
 			set := argosettings.NewSettingsManager(context.TODO(), fakeClient, namespace)
-			h, err := NewWebhookHandler(namespace, set, fc, mockGenerators())
+			h, err := NewWebhookHandler(namespace, webhookParallelism, set, fc, mockGenerators())
 			require.NoError(t, err)
 
 			req := httptest.NewRequest(http.MethodPost, "/api/webhook", nil)
@@ -217,6 +218,8 @@ func TestWebhookHandler(t *testing.T) {
 			w := httptest.NewRecorder()
 
 			h.Handler(w, req)
+			close(h.queue)
+			h.Wait()
 			assert.Equal(t, test.expectedStatusCode, w.Code)
 
 			list := &v1alpha1.ApplicationSetList{}

cmd/argocd-applicationset-controller/commands/applicationset_controller.go

@@ -69,6 +69,7 @@ func NewCommand() *cobra.Command {
 		globalPreservedAnnotations   []string
 		globalPreservedLabels        []string
 		enableScmProviders           bool
+		webhookParallelism           int
 	)
 	scheme := runtime.NewScheme()
 	_ = clientgoscheme.AddToScheme(scheme)
@@ -182,7 +183,7 @@ func NewCommand() *cobra.Command {
 			topLevelGenerators := generators.GetGenerators(ctx, mgr.GetClient(), k8sClient, namespace, argoCDService, dynamicClient, scmConfig)
 
 			// start a webhook server that listens to incoming webhook payloads
-			webhookHandler, err := webhook.NewWebhookHandler(namespace, argoSettingsMgr, mgr.GetClient(), topLevelGenerators)
+			webhookHandler, err := webhook.NewWebhookHandler(namespace, webhookParallelism, argoSettingsMgr, mgr.GetClient(), topLevelGenerators)
 			if err != nil {
 				log.Error(err, "failed to create webhook handler")
 			}
@@ -248,6 +249,7 @@ func NewCommand() *cobra.Command {
 	command.Flags().StringVar(&scmRootCAPath, "scm-root-ca-path", env.StringFromEnv("ARGOCD_APPLICATIONSET_CONTROLLER_SCM_ROOT_CA_PATH", ""), "Provide Root CA Path for self-signed TLS Certificates")
 	command.Flags().StringSliceVar(&globalPreservedAnnotations, "preserved-annotations", env.StringsFromEnv("ARGOCD_APPLICATIONSET_CONTROLLER_GLOBAL_PRESERVED_ANNOTATIONS", []string{}, ","), "Sets global preserved field values for annotations")
 	command.Flags().StringSliceVar(&globalPreservedLabels, "preserved-labels", env.StringsFromEnv("ARGOCD_APPLICATIONSET_CONTROLLER_GLOBAL_PRESERVED_LABELS", []string{}, ","), "Sets global preserved field values for labels")
+	command.Flags().IntVar(&webhookParallelism, "webhook-parallelism-limit", env.ParseNumFromEnv("ARGOCD_APPLICATIONSET_CONTROLLER_WEBHOOK_PARALLELISM_LIMIT", 50, 1, 1000), "Number of webhook requests processed concurrently")
 	return &command
 }
 

cmd/argocd-server/commands/argocd_server.go

@@ -82,6 +82,7 @@ func NewCommand() *cobra.Command {
 		staticAssetsDir          string
 		applicationNamespaces    []string
 		enableProxyExtension     bool
+		webhookParallelism       int
 
 		// ApplicationSet
 		enableNewGitFileGlobbing bool
@@ -221,6 +222,7 @@ func NewCommand() *cobra.Command {
 				StaticAssetsDir:         staticAssetsDir,
 				ApplicationNamespaces:   applicationNamespaces,
 				EnableProxyExtension:    enableProxyExtension,
+				WebhookParallelism:      webhookParallelism,
 			}
 
 			appsetOpts := server.ApplicationSetOpts{
@@ -294,6 +296,7 @@ func NewCommand() *cobra.Command {
 	command.Flags().BoolVar(&dexServerStrictTLS, "dex-server-strict-tls", env.ParseBoolFromEnv("ARGOCD_SERVER_DEX_SERVER_STRICT_TLS", false), "Perform strict validation of TLS certificates when connecting to dex server")
 	command.Flags().StringSliceVar(&applicationNamespaces, "application-namespaces", env.StringsFromEnv("ARGOCD_APPLICATION_NAMESPACES", []string{}, ","), "List of additional namespaces where application resources can be managed in")
 	command.Flags().BoolVar(&enableProxyExtension, "enable-proxy-extension", env.ParseBoolFromEnv("ARGOCD_SERVER_ENABLE_PROXY_EXTENSION", false), "Enable Proxy Extension feature")
+	command.Flags().IntVar(&webhookParallelism, "webhook-parallelism-limit", env.ParseNumFromEnv("ARGOCD_SERVER_WEBHOOK_PARALLELISM_LIMIT", 50, 1, 1000), "Number of webhook requests processed concurrently")
 
 	// Flags related to the applicationSet component.
 	command.Flags().StringVar(&scmRootCAPath, "appset-scm-root-ca-path", env.StringFromEnv("ARGOCD_APPLICATIONSET_CONTROLLER_SCM_ROOT_CA_PATH", ""), "Provide Root CA Path for self-signed TLS Certificates")

docs/operator-manual/argocd-cmd-params-cm.yaml

@@ -93,6 +93,8 @@ data:
   # Semicolon-separated list of content types allowed on non-GET requests. Set an empty string to allow all. Be aware
   # that allowing content types besides application/json may make your API more vulnerable to CSRF attacks.
   server.api.content.types: "application/json"
+  # Number of webhook requests processed concurrently (default 50)
+  server.webhook.parallelism.limit: "50"
 
   # Set the logging format. One of: text|json (default "text")
   server.log.format: "text"
@@ -213,6 +215,8 @@ data:
   applicationsetcontroller.allowed.scm.providers: "https://git.example.com/,https://gitlab.example.com/"
   # To disable SCM providers entirely (i.e. disable the SCM and PR generators), set this to "false". Default is "true".
   applicationsetcontroller.enable.scm.providers: "false"
+  # Number of webhook requests processed concurrently (default 50)
+  applicationsetcontroller.webhook.parallelism.limit: "50"
 
   ## Argo CD Notifications Controller Properties
   # Set the logging level. One of: debug|info|warn|error (default "info")

docs/operator-manual/server-commands/argocd-server.md

@@ -110,6 +110,7 @@ argocd-server [flags]
       --token string                                    Bearer token for authentication to the API server
       --user string                                     The name of the kubeconfig user to use
       --username string                                 Username for basic authentication to the API server
+      --webhook-parallelism-limit int                   Number of webhook requests processed concurrently (default 50)
       --x-frame-options value                           Set X-Frame-Options header in HTTP responses to value. To disable, set to "". (default "sameorigin")
 ```
 

manifests/base/applicationset-controller/argocd-applicationset-controller-deployment.yaml

@@ -157,6 +157,12 @@ spec:
                   name: argocd-cmd-params-cm
                   key: applicationsetcontroller.enable.scm.providers
                   optional: true
+            - name: ARGOCD_APPLICATIONSET_CONTROLLER_WEBHOOK_PARALLELISM_LIMIT
+              valueFrom:
+                configMapKeyRef:
+                  name: argocd-cmd-params-cm
+                  key: applicationsetcontroller.webhook.parallelism.limit
+                  optional: true
           volumeMounts:
             - mountPath: /app/config/ssh
               name: ssh-known-hosts

manifests/base/server/argocd-server-deployment.yaml

@@ -262,6 +262,12 @@ spec:
                   name: argocd-cmd-params-cm
                   key: server.api.content.types
                   optional: true
+            - name: ARGOCD_SERVER_WEBHOOK_PARALLELISM_LIMIT
+              valueFrom:
+                configMapKeyRef:
+                  name: argocd-cmd-params-cm
+                  key: server.webhook.parallelism.limit
+                  optional: true
             - name: ARGOCD_APPLICATIONSET_CONTROLLER_ENABLE_NEW_GIT_FILE_GLOBBING
               valueFrom:
                 configMapKeyRef:

manifests/core-install.yaml

@@ -21268,6 +21268,12 @@ spec:
               key: applicationsetcontroller.enable.scm.providers
               name: argocd-cmd-params-cm
               optional: true
+        - name: ARGOCD_APPLICATIONSET_CONTROLLER_WEBHOOK_PARALLELISM_LIMIT
+          valueFrom:
+            configMapKeyRef:
+              key: applicationsetcontroller.webhook.parallelism.limit
+              name: argocd-cmd-params-cm
+              optional: true
         image: quay.io/argoproj/argocd:latest
         imagePullPolicy: Always
         name: argocd-applicationset-controller

manifests/ha/install.yaml

@@ -22609,6 +22609,12 @@ spec:
               key: applicationsetcontroller.enable.scm.providers
               name: argocd-cmd-params-cm
               optional: true
+        - name: ARGOCD_APPLICATIONSET_CONTROLLER_WEBHOOK_PARALLELISM_LIMIT
+          valueFrom:
+            configMapKeyRef:
+              key: applicationsetcontroller.webhook.parallelism.limit
+              name: argocd-cmd-params-cm
+              optional: true
         image: quay.io/argoproj/argocd:latest
         imagePullPolicy: Always
         name: argocd-applicationset-controller
@@ -23590,6 +23596,12 @@ spec:
               key: server.api.content.types
               name: argocd-cmd-params-cm
               optional: true
+        - name: ARGOCD_SERVER_WEBHOOK_PARALLELISM_LIMIT
+          valueFrom:
+            configMapKeyRef:
+              key: server.webhook.parallelism.limit
+              name: argocd-cmd-params-cm
+              optional: true
         - name: ARGOCD_APPLICATIONSET_CONTROLLER_ENABLE_NEW_GIT_FILE_GLOBBING
           valueFrom:
             configMapKeyRef:

manifests/ha/namespace-install.yaml

@@ -1686,6 +1686,12 @@ spec:
               key: applicationsetcontroller.enable.scm.providers
               name: argocd-cmd-params-cm
               optional: true
+        - name: ARGOCD_APPLICATIONSET_CONTROLLER_WEBHOOK_PARALLELISM_LIMIT
+          valueFrom:
+            configMapKeyRef:
+              key: applicationsetcontroller.webhook.parallelism.limit
+              name: argocd-cmd-params-cm
+              optional: true
         image: quay.io/argoproj/argocd:latest
         imagePullPolicy: Always
         name: argocd-applicationset-controller
@@ -2667,6 +2673,12 @@ spec:
               key: server.api.content.types
               name: argocd-cmd-params-cm
               optional: true
+        - name: ARGOCD_SERVER_WEBHOOK_PARALLELISM_LIMIT
+          valueFrom:
+            configMapKeyRef:
+              key: server.webhook.parallelism.limit
+              name: argocd-cmd-params-cm
+              optional: true
         - name: ARGOCD_APPLICATIONSET_CONTROLLER_ENABLE_NEW_GIT_FILE_GLOBBING
           valueFrom:
             configMapKeyRef:

manifests/install.yaml

@@ -21726,6 +21726,12 @@ spec:
               key: applicationsetcontroller.enable.scm.providers
               name: argocd-cmd-params-cm
               optional: true
+        - name: ARGOCD_APPLICATIONSET_CONTROLLER_WEBHOOK_PARALLELISM_LIMIT
+          valueFrom:
+            configMapKeyRef:
+              key: applicationsetcontroller.webhook.parallelism.limit
+              name: argocd-cmd-params-cm
+              optional: true
         image: quay.io/argoproj/argocd:latest
         imagePullPolicy: Always
         name: argocd-applicationset-controller
@@ -22658,6 +22664,12 @@ spec:
               key: server.api.content.types
               name: argocd-cmd-params-cm
               optional: true
+        - name: ARGOCD_SERVER_WEBHOOK_PARALLELISM_LIMIT
+          valueFrom:
+            configMapKeyRef:
+              key: server.webhook.parallelism.limit
+              name: argocd-cmd-params-cm
+              optional: true
         - name: ARGOCD_APPLICATIONSET_CONTROLLER_ENABLE_NEW_GIT_FILE_GLOBBING
           valueFrom:
             configMapKeyRef:

manifests/namespace-install.yaml

@@ -803,6 +803,12 @@ spec:
               key: applicationsetcontroller.enable.scm.providers
               name: argocd-cmd-params-cm
               optional: true
+        - name: ARGOCD_APPLICATIONSET_CONTROLLER_WEBHOOK_PARALLELISM_LIMIT
+          valueFrom:
+            configMapKeyRef:
+              key: applicationsetcontroller.webhook.parallelism.limit
+              name: argocd-cmd-params-cm
+              optional: true
         image: quay.io/argoproj/argocd:latest
         imagePullPolicy: Always
         name: argocd-applicationset-controller
@@ -1735,6 +1741,12 @@ spec:
               key: server.api.content.types
               name: argocd-cmd-params-cm
               optional: true
+        - name: ARGOCD_SERVER_WEBHOOK_PARALLELISM_LIMIT
+          valueFrom:
+            configMapKeyRef:
+              key: server.webhook.parallelism.limit
+              name: argocd-cmd-params-cm
+              optional: true
         - name: ARGOCD_APPLICATIONSET_CONTROLLER_ENABLE_NEW_GIT_FILE_GLOBBING
           valueFrom:
             configMapKeyRef:

server/server.go

@@ -228,6 +228,7 @@ type ArgoCDServerOpts struct {
 	ContentSecurityPolicy   string
 	ApplicationNamespaces   []string
 	EnableProxyExtension    bool
+	WebhookParallelism      int
 }
 
 type ApplicationSetOpts struct {
@@ -1072,7 +1073,7 @@ func (a *ArgoCDServer) newHTTPServer(ctx context.Context, port int, grpcWebHandl
 
 	// Webhook handler for git events (Note: cache timeouts are hardcoded because API server does not write to cache and not really using them)
 	argoDB := db.NewDB(a.Namespace, a.settingsMgr, a.KubeClientset)
-	acdWebhookHandler := webhook.NewHandler(a.Namespace, a.ArgoCDServerOpts.ApplicationNamespaces, a.AppClientset, a.settings, a.settingsMgr, a.RepoServerCache, a.Cache, argoDB)
+	acdWebhookHandler := webhook.NewHandler(a.Namespace, a.ArgoCDServerOpts.ApplicationNamespaces, a.ArgoCDServerOpts.WebhookParallelism, a.AppClientset, a.settings, a.settingsMgr, a.RepoServerCache, a.Cache, argoDB)
 
 	mux.HandleFunc("/api/webhook", acdWebhookHandler.Handler)