feat: Allow application updates to run simultaneously (#11)

1 files changed, 128 insertions, 89 deletions

diff --git a/cmd/main.go b/cmd/main.go
index 8171736..37893ba 100644
--- a/cmd/main.go
+++ b/cmd/main.go
@@ -1,10 +1,12 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/argoproj-labs/argocd-image-updater/pkg/argocd"
@@ -15,6 +17,7 @@ import (
 	"github.com/argoproj-labs/argocd-image-updater/pkg/version"
 
 	"github.com/spf13/cobra"
+	"golang.org/x/sync/semaphore"
 )
 
 var lastRun time.Time
@@ -45,6 +48,105 @@ type ImageUpdaterResult struct {
 	NumErrors                int
 }
 
+// Update all images of a single application. Will run in a goroutine.
+func updateApplication(argoClient *argocd.ArgoCD, kubeClient *client.KubernetesClient, curApplication *argocd.ApplicationImages) ImageUpdaterResult {
+	result := ImageUpdaterResult{}
+	app := curApplication.Application.GetName()
+
+	// Get all images that are deployed with the current application
+	applicationImages := argocd.GetImagesFromApplication(&curApplication.Application)
+
+	result.NumApplicationsProcessed += 1
+
+	// Loop through all images of current application, and check whether one of
+	// its images is egilible for updating.
+	//
+	// Whether an image qualifies for update is dependent on semantic version
+	// constraints which are part of the application's annotation values.
+	//
+	for _, applicationImage := range applicationImages {
+		updateableImage := curApplication.Images.ContainsImage(applicationImage, false)
+		if updateableImage == nil {
+			log.WithContext().AddField("application", app).Debugf("Image %s not in list of allowed images, skipping", applicationImage.ImageName)
+			result.NumSkipped += 1
+			continue
+		}
+
+		result.NumImagesConsidered += 1
+
+		imgCtx := log.WithContext().
+			AddField("application", app).
+			AddField("registry", applicationImage.RegistryURL).
+			AddField("image_name", applicationImage.ImageName).
+			AddField("image_tag", applicationImage.ImageTag)
+
+		imgCtx.Debugf("Considering this image for update")
+
+		rep, err := registry.GetRegistryEndpoint(applicationImage.RegistryURL)
+		if err != nil {
+			imgCtx.Errorf("Could not get registry endpoint from configuration: %v", err)
+			result.NumErrors += 1
+			continue
+		}
+
+		// Get list of available image tags from the repository
+		tags, err := rep.GetTags(applicationImage, kubeClient)
+		if err != nil {
+			imgCtx.Errorf("Could not get tags from registry: %v", err)
+			result.NumErrors += 1
+			continue
+		}
+
+		imgCtx.Tracef("List of available tags found: %v", tags)
+
+		// Get the latest available tag matching any constraint that might be set
+		// for allowed updates.
+		latest, err := applicationImage.GetNewestVersionFromTags(updateableImage.ImageTag, tags)
+		if err != nil {
+			imgCtx.Errorf("Unable to find newest version from available tags: %v", err)
+			result.NumErrors += 1
+			continue
+		}
+
+		// If we have no latest tag information, it means there was no tag which
+		// has met our version constraint (or there was no semantic versioned tag
+		// at all in the repository)
+		if latest == "" {
+			imgCtx.Debugf("No suitable image tag for upgrade found in list of available tags.")
+			result.NumSkipped += 1
+			continue
+		}
+
+		// If the latest tag does not match image's current tag, it means we have
+		// an update candidate.
+		if applicationImage.ImageTag != latest {
+			imgCtx.Infof("Upgrading image to %s", applicationImage.WithTag(latest).String())
+
+			if appType := argocd.GetApplicationType(&curApplication.Application); appType == argocd.ApplicationTypeKustomize {
+				err = argoClient.SetKustomizeImage(&curApplication.Application, applicationImage.WithTag(latest))
+			} else if appType == argocd.ApplicationTypeHelm {
+				err = argoClient.SetHelmImage(&curApplication.Application, applicationImage.WithTag(latest))
+			} else {
+				result.NumErrors += 1
+				err = fmt.Errorf("Could not update application %s - neither Helm nor Kustomize application", app)
+			}
+
+			if err != nil {
+				imgCtx.Errorf("Error while trying to update image: %v", err)
+				result.NumErrors += 1
+				continue
+			} else {
+				imgCtx.Infof("Successfully updated image '%s' to '%s'", applicationImage.GetFullNameWithTag(), applicationImage.WithTag(latest).GetFullNameWithTag())
+				result.NumImagesUpdated += 1
+			}
+		} else {
+			imgCtx.Debugf("Image '%s' already on latest allowed version", applicationImage.GetFullNameWithTag())
+		}
+	}
+
+	return result
+}
+
 // Main loop for argocd-image-controller
 func runImageUpdater(cfg *ImageUpdaterConfig) (ImageUpdaterResult, error) {
 	result := ImageUpdaterResult{}
@@ -75,100 +177,37 @@ func runImageUpdater(cfg *ImageUpdaterConfig) (ImageUpdaterResult, error) {
 
 	log.Debugf("Considering %d applications with annotations for update", len(appList))
 
-	for app, curApplication := range appList {
-
-		// Get all images that are deployed with the current application
-		applicationImages := argocd.GetImagesFromApplication(&curApplication.Application)
-
-		result.NumApplicationsProcessed += 1
-
-		// Loop through all images of current application, and check whether one of
-		// its images is egilible for updating.
-		//
-		// Whether an image qualifies for update is dependent on semantic version
-		// constraints which are part of the application's annotation values.
-		//
-		for _, applicationImage := range applicationImages {
-			updateableImage := curApplication.Images.ContainsImage(applicationImage, false)
-			if updateableImage == nil {
-				log.WithContext().AddField("application", app).Debugf("Image %s not in list of allowed images, skipping", applicationImage.ImageName)
-				result.NumSkipped += 1
-				continue
-			}
+	// Allow a maximum of MaxConcurrency number of goroutines to exist at the
+	// same time.
+	sem := semaphore.NewWeighted(int64(cfg.MaxConcurrency))
 
-			result.NumImagesConsidered += 1
+	var wg sync.WaitGroup
 
-			imgCtx := log.WithContext().
-				AddField("application", app).
-				AddField("registry", applicationImage.RegistryURL).
-				AddField("image_name", applicationImage.ImageName).
-				AddField("image_tag", applicationImage.ImageTag)
-
-			imgCtx.Debugf("Considering this image for update")
-
-			rep, err := registry.GetRegistryEndpoint(applicationImage.RegistryURL)
-			if err != nil {
-				imgCtx.Errorf("Could not get registry endpoint from configuration: %v", err)
-				result.NumErrors += 1
-				continue
-			}
-
-			// Get list of available image tags from the repository
-			tags, err := rep.GetTags(applicationImage, cfg.KubeClient)
-			if err != nil {
-				imgCtx.Errorf("Could not get tags from registry: %v", err)
-				result.NumErrors += 1
-				continue
-			}
-
-			imgCtx.Tracef("List of available tags found: %v", tags)
-
-			// Get the latest available tag matching any constraint that might be set
-			// for allowed updates.
-			latest, err := applicationImage.GetNewestVersionFromTags(updateableImage.ImageTag, tags)
-			if err != nil {
-				imgCtx.Errorf("Unable to find newest version from available tags: %v", err)
-				result.NumErrors += 1
-				continue
-			}
-
-			// If we have no latest tag information, it means there was no tag which
-			// has met our version constraint (or there was no semantic versioned tag
-			// at all in the repository)
-			if latest == "" {
-				imgCtx.Debugf("No suitable image tag for upgrade found in list of available tags.")
-				result.NumSkipped += 1
-				continue
-			}
-
-			// If the latest tag does not match image's current tag, it means we have
-			// an update candidate.
-			if applicationImage.ImageTag != latest {
-				imgCtx.Infof("Upgrading image to %s", applicationImage.WithTag(latest).String())
-
-				if appType := argocd.GetApplicationType(&curApplication.Application); appType == argocd.ApplicationTypeKustomize {
-					err = argoClient.SetKustomizeImage(&curApplication.Application, applicationImage.WithTag(latest))
-				} else if appType == argocd.ApplicationTypeHelm {
-					err = argoClient.SetHelmImage(&curApplication.Application, applicationImage.WithTag(latest))
-				} else {
-					result.NumErrors += 1
-					err = fmt.Errorf("Could not update application %s - neither Helm nor Kustomize application", app)
-				}
-
-				if err != nil {
-					imgCtx.Errorf("Error while trying to update image: %v", err)
-					result.NumErrors += 1
-					continue
-				} else {
-					imgCtx.Infof("Successfully updated image '%s' to '%s'", applicationImage.GetFullNameWithTag(), applicationImage.WithTag(latest).GetFullNameWithTag())
-					result.NumImagesUpdated += 1
-				}
-			} else {
-				imgCtx.Debugf("Image '%s' already on latest allowed version", applicationImage.GetFullNameWithTag())
-			}
+	for app, curApplication := range appList {
+		lockErr := sem.Acquire(context.TODO(), 1)
+		if lockErr != nil {
+			log.Errorf("could not acquire semaphore for application %s: %v", app, lockErr)
+			continue
 		}
+
+		wg.Add(1)
+
+		go func(app string, curApplication argocd.ApplicationImages) {
+			defer sem.Release(1)
+			log.Debugf("Processing application %s", app)
+			res := updateApplication(cfg.ArgoClient, cfg.KubeClient, &curApplication)
+			result.NumApplicationsProcessed += 1
+			result.NumErrors += res.NumErrors
+			result.NumImagesConsidered += res.NumImagesConsidered
+			result.NumImagesUpdated += res.NumImagesUpdated
+			result.NumSkipped += res.NumSkipped
+			wg.Done()
+		}(app, curApplication)
 	}
 
+	// Wait for all goroutines to finish
+	wg.Wait()
+
 	return result, nil
 }