feat(scheduler): add LRU/fail-first scheduling, cooldown, and per-repo cap; improve logs for queued git writes

Author: Eidmantas Ivanauskas

cmd/main.go

@@ -52,6 +52,10 @@ type ImageUpdaterConfig struct {
 	DisableKubeEvents      bool
 	GitCreds               git.CredsStore
 	EnableWebhook          bool
+    // Scheduler options
+    Schedule              string
+    Cooldown              time.Duration
+    PerRepoCap            int
 }
 
 // newRootCommand implements the root command of argocd-image-updater

cmd/run.go

@@ -32,6 +32,36 @@ import (
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+// orderApplications reorders apps by schedule policy: lru (least-recent success first),
+// fail-first (recent failures first), with optional cooldown to deprioritize recently
+// successful apps.
+func orderApplications(names []string, appList map[string]argocd.ApplicationImages, state *argocd.SyncIterationState, cfg *ImageUpdaterConfig) []string {
+    stats := state.GetStats()
+    type item struct{ name string; score int64 }
+    items := make([]item, 0, len(names))
+    now := time.Now()
+    for _, n := range names {
+        s := stats[n]
+        score := int64(0)
+        switch cfg.Schedule {
+        case "lru":
+            // Older success => higher priority (lower score)
+            if !s.LastSuccess.IsZero() { score -= int64(now.Sub(s.LastSuccess).Milliseconds()) }
+        case "fail-first":
+            score += int64(s.FailCount) * 1_000_000 // dominate by failures
+            if !s.LastAttempt.IsZero() { score -= int64(now.Sub(s.LastAttempt).Milliseconds()) }
+        }
+        if cfg.Cooldown > 0 && !s.LastSuccess.IsZero() && now.Sub(s.LastSuccess) < cfg.Cooldown {
+            score -= 1 // slight deprioritization
+        }
+        items = append(items, item{name: n, score: score})
+    }
+    sort.Slice(items, func(i, j int) bool { return items[i].score > items[j].score })
+    out := make([]string, len(items))
+    for i := range items { out[i] = items[i].name }
+    return out
+}
+
 // newRunCommand implements "run" command
 func newRunCommand() *cobra.Command {
 	var cfg *ImageUpdaterConfig = &ImageUpdaterConfig{}
@@ -60,7 +90,7 @@ func newRunCommand() *cobra.Command {
 				return fmt.Errorf("--max-concurrency must be greater than 1")
 			}
 
-			log.Infof("%s %s starting [loglevel:%s, interval:%s, healthport:%s]",
+            log.Infof("%s %s starting [loglevel:%s, interval:%s, healthport:%s]",
 				version.BinaryName(),
 				version.Version(),
 				strings.ToUpper(cfg.LogLevel),
@@ -144,7 +174,7 @@ func newRunCommand() *cobra.Command {
 			)
 
 			// Initialize metrics before starting the metrics server or using any counters
-			metrics.InitMetrics()
+            metrics.InitMetrics()
 
 			// Health server will start in a go routine and run asynchronously
 			var hsErrCh chan error
@@ -255,7 +285,7 @@ func newRunCommand() *cobra.Command {
 
 			// This is our main loop. We leave it only when our health probe server
 			// returns an error.
-			for {
+            for {
 				select {
 				case err := <-hsErrCh:
 					if err != nil {
@@ -288,7 +318,7 @@ func newRunCommand() *cobra.Command {
 					return nil
 				default:
 					if lastRun.IsZero() || time.Since(lastRun) > cfg.CheckInterval {
-						result, err := runImageUpdater(cfg, false)
+                        result, err := runImageUpdater(cfg, false)
 						if err != nil {
 							log.Errorf("Error: %v", err)
 						} else {
@@ -340,6 +370,9 @@ func newRunCommand() *cobra.Command {
 	runCmd.Flags().StringVar(&cfg.AppLabel, "match-application-label", "", "label selector to match application labels against. DEPRECATED: this flag will be removed in a future version.")
 
 	runCmd.Flags().BoolVar(&warmUpCache, "warmup-cache", true, "whether to perform a cache warm-up on startup")
+	runCmd.Flags().StringVar(&cfg.Schedule, "schedule", env.GetStringVal("IMAGE_UPDATER_SCHEDULE", "default"), "scheduling policy: default|lru|fail-first")
+	runCmd.Flags().DurationVar(&cfg.Cooldown, "cooldown", env.GetDurationVal("IMAGE_UPDATER_COOLDOWN", 0), "deprioritize apps updated within this duration")
+	runCmd.Flags().IntVar(&cfg.PerRepoCap, "per-repo-cap", env.ParseNumFromEnv("IMAGE_UPDATER_PER_REPO_CAP", 0, 0, 100000), "max updates per repo per cycle (0 = unlimited)")
 	runCmd.Flags().StringVar(&cfg.GitCommitUser, "git-commit-user", env.GetStringVal("GIT_COMMIT_USER", "argocd-image-updater"), "Username to use for Git commits")
 	runCmd.Flags().StringVar(&cfg.GitCommitMail, "git-commit-email", env.GetStringVal("GIT_COMMIT_EMAIL", "[email protected]"), "E-Mail address to use for Git commits")
 	runCmd.Flags().StringVar(&cfg.GitCommitSigningKey, "git-commit-signing-key", env.GetStringVal("GIT_COMMIT_SIGNING_KEY", ""), "GnuPG key ID or path to Private SSH Key used to sign the commits")
@@ -403,7 +436,7 @@ func runImageUpdater(cfg *ImageUpdaterConfig, warmUp bool) (argocd.ImageUpdaterR
 		log.Infof("Starting image update cycle, considering %d annotated application(s) for update", len(appList))
 	}
 
-	syncState := argocd.NewSyncIterationState()
+            syncState := argocd.NewSyncIterationState()
 
 	// Allow a maximum of MaxConcurrency number of goroutines to exist at the
 	// same time. If in warm-up mode, set to 1 explicitly.
@@ -420,7 +453,25 @@ func runImageUpdater(cfg *ImageUpdaterConfig, warmUp bool) (argocd.ImageUpdaterR
 	var wg sync.WaitGroup
 	wg.Add(len(appList))
 
-	for app, curApplication := range appList {
+            // Optionally reorder apps by scheduling policy
+            ordered := make([]string, 0, len(appList))
+            for app := range appList { ordered = append(ordered, app) }
+            if cfg.Schedule != "default" || cfg.Cooldown > 0 || cfg.PerRepoCap > 0 {
+                ordered = orderApplications(ordered, appList, syncState, cfg)
+            }
+
+            perRepoCounter := map[string]int{}
+
+            for _, app := range ordered {
+                curApplication := appList[app]
+                // Per-repo cap if configured
+                if cfg.PerRepoCap > 0 {
+                    repo := argocd.GetApplicationSource(&curApplication.Application).RepoURL
+                    if perRepoCounter[repo] >= cfg.PerRepoCap {
+                        continue
+                    }
+                }
+                syncState.RecordAttempt(app)
 		lockErr := sem.Acquire(context.Background(), 1)
 		if lockErr != nil {
 			log.Errorf("Could not acquire semaphore for application %s: %v", app, lockErr)
@@ -447,7 +498,12 @@ func runImageUpdater(cfg *ImageUpdaterConfig, warmUp bool) (argocd.ImageUpdaterR
 				DisableKubeEvents:      cfg.DisableKubeEvents,
 				GitCreds:               cfg.GitCreds,
 			}
-			res := argocd.UpdateApplication(upconf, syncState)
+            res := argocd.UpdateApplication(upconf, syncState)
+            syncState.RecordResult(app, res.NumErrors > 0)
+            if cfg.PerRepoCap > 0 {
+                repo := argocd.GetApplicationSource(&curApplication.Application).RepoURL
+                perRepoCounter[repo] = perRepoCounter[repo] + 1
+            }
 			result.NumApplicationsProcessed += 1
 			result.NumErrors += res.NumErrors
 			result.NumImagesConsidered += res.NumImagesConsidered

pkg/argocd/update.go

@@ -123,15 +123,65 @@ type ChangeEntry struct {
 type SyncIterationState struct {
 	lock            sync.Mutex
 	repositoryLocks map[string]*sync.Mutex
+    lastSuccess     map[string]time.Time
+    lastAttempt     map[string]time.Time
+    failCount       map[string]int
 }
 
 // NewSyncIterationState returns a new instance of SyncIterationState
 func NewSyncIterationState() *SyncIterationState {
 	return &SyncIterationState{
-		repositoryLocks: make(map[string]*sync.Mutex),
+        repositoryLocks: make(map[string]*sync.Mutex),
+        lastSuccess:     make(map[string]time.Time),
+        lastAttempt:     make(map[string]time.Time),
+        failCount:       make(map[string]int),
 	}
 }
 
+// AppStats is a snapshot of per-application scheduling stats
+type AppStats struct {
+    LastSuccess time.Time
+    LastAttempt time.Time
+    FailCount   int
+}
+
+// RecordAttempt notes that an attempt was made to process the application
+func (state *SyncIterationState) RecordAttempt(app string) {
+    state.lock.Lock()
+    state.lastAttempt[app] = time.Now()
+    state.lock.Unlock()
+}
+
+// RecordResult records the outcome for an application
+func (state *SyncIterationState) RecordResult(app string, hadErrors bool) {
+    state.lock.Lock()
+    defer state.lock.Unlock()
+    if hadErrors {
+        state.failCount[app] = state.failCount[app] + 1
+    } else {
+        state.failCount[app] = 0
+        state.lastSuccess[app] = time.Now()
+    }
+}
+
+// GetStats returns a copy of current stats for scheduling decisions
+func (state *SyncIterationState) GetStats() map[string]AppStats {
+    state.lock.Lock()
+    defer state.lock.Unlock()
+    out := make(map[string]AppStats, len(state.lastAttempt))
+    // Union of keys
+    for k := range state.lastAttempt { out[k] = AppStats{} }
+    for k := range state.lastSuccess { out[k] = AppStats{} }
+    for k := range state.failCount { out[k] = AppStats{} }
+    for k, v := range out {
+        v.LastAttempt = state.lastAttempt[k]
+        v.LastSuccess = state.lastSuccess[k]
+        v.FailCount = state.failCount[k]
+        out[k] = v
+    }
+    return out
+}
+
 // GetRepositoryLock returns the lock for a specified repository
 func (state *SyncIterationState) GetRepositoryLock(repository string) *sync.Mutex {
 	state.lock.Lock()
@@ -169,7 +219,7 @@ func UpdateApplication(updateConf *UpdateConfiguration, state *SyncIterationStat
 
 	result.NumApplicationsProcessed += 1
 
-	// Loop through all images of current application, and check whether one of
+    // Loop through all images of current application, and check whether one of
 	// its images is eligible for updating.
 	//
 	// Whether an image qualifies for update is dependent on semantic version
@@ -250,7 +300,7 @@ func UpdateApplication(updateConf *UpdateConfiguration, state *SyncIterationStat
 				imgCtx.Warnf("Could not fetch credentials: %v", err)
 				result.NumErrors += 1
 				continue
-			}
+        }
 		}
 
 		regClient, err := updateConf.NewRegFN(rep, creds.Username, creds.Password)
@@ -357,11 +407,11 @@ func UpdateApplication(updateConf *UpdateConfiguration, state *SyncIterationStat
 		wbc.GitCommitSignOff = updateConf.GitCommitSignOff
 	}
 
-	if needUpdate {
+    if needUpdate {
 		logCtx := log.WithContext().AddField("application", app)
-		log.Debugf("Using commit message: %s", wbc.GitCommitMessage)
+        log.Debugf("Using commit message: %s", wbc.GitCommitMessage)
 		if !updateConf.DryRun {
-			logCtx.Infof("Committing %d parameter update(s) for application %s", result.NumImagesUpdated, app)
+            logCtx.Infof("Queuing %d parameter update(s) for application %s (git write pending)", result.NumImagesUpdated, app)
             // Enqueue batched write intent unless tests explicitly disable batching
             var err error
             if env.GetBoolVal("GIT_BATCH_DISABLE", false) {
@@ -375,8 +425,8 @@ func UpdateApplication(updateConf *UpdateConfiguration, state *SyncIterationStat
 				logCtx.Errorf("Could not update application spec: %v", err)
 				result.NumErrors += 1
 				result.NumImagesUpdated = 0
-			} else {
-				logCtx.Infof("Successfully updated the live application spec")
+            } else {
+                logCtx.Infof("Application spec updated in-memory; write-back scheduled")
 				if !updateConf.DisableKubeEvents && updateConf.KubeClient != nil {
 					annotations := map[string]string{}
 					for i, c := range changeList {