fix: prevent concurrent registry credential refreshing (#1175)

Signed-off-by: Jongwon Youn <[email protected]>

Author: korECM

registry-scanner/pkg/registry/endpoints.go

@@ -13,6 +13,7 @@ import (
 	"github.com/argoproj-labs/argocd-image-updater/registry-scanner/pkg/log"
 
 	"go.uber.org/ratelimit"
+	"golang.org/x/sync/singleflight"
 )
 
 // TagListSort defines how the registry returns the list of tags
@@ -117,6 +118,9 @@ var defaultRegistry *RegistryEndpoint
 // Simple RW mutex for concurrent access to registries map
 var registryLock sync.RWMutex
 
+// credentialGroup ensures only one credential refresh happens per registry
+var credentialGroup singleflight.Group
+
 func AddRegistryEndpointFromConfig(epc RegistryConfiguration) error {
 	ep := NewRegistryEndpoint(epc.Prefix, epc.Name, epc.ApiURL, epc.Credentials, epc.DefaultNS, epc.Insecure, TagListSortFromString(epc.TagSortMode), epc.Limit, epc.CredsExpire)
 	return AddRegistryEndpoint(ep)

registry-scanner/pkg/registry/registry.go

@@ -188,8 +188,17 @@ func (ep *RegistryEndpoint) expireCredentials() bool {
 	return false
 }
 
-// Sets endpoint credentials for this registry from a reference to a K8s secret
+// SetEndpointCredentials Sets endpoint credentials for this registry from a reference to a K8s secret
 func (ep *RegistryEndpoint) SetEndpointCredentials(kubeClient *kube.KubernetesClient) error {
+	// Use singleflight to prevent concurrent credential fetching for the same registry
+	_, err, _ := credentialGroup.Do(ep.RegistryAPI, func() (interface{}, error) {
+		return nil, ep.setEndpointCredentialsInternal(kubeClient)
+	})
+	return err
+}
+
+// setEndpointCredentialsInternal performs the actual credential fetching
+func (ep *RegistryEndpoint) setEndpointCredentialsInternal(kubeClient *kube.KubernetesClient) error {
 	if ep.expireCredentials() {
 		log.Debugf("expired credentials for registry %s (updated:%s, expiry:%0fs)", ep.RegistryAPI, ep.CredsUpdated, ep.CredsExpire.Seconds())
 	}

registry-scanner/pkg/registry/registry_test.go

@@ -2,6 +2,9 @@ package registry
 
 import (
 	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -161,3 +164,121 @@ registries:
 	})
 
 }
+
+func Test_ConcurrentCredentialFetching(t *testing.T) {
+	t.Run("Multiple goroutines fetching credentials should only call once", func(t *testing.T) {
+		// Create a mock script that counts how many times it's called
+		scriptContent := `#!/bin/sh
+echo "counter" >> /tmp/test_ecr_calls.log
+echo "AWS:mock-token-12345"
+`
+		scriptPath := "/tmp/test_ecr_auth.sh"
+		err := os.WriteFile(scriptPath, []byte(scriptContent), 0755)
+		require.NoError(t, err)
+		defer os.Remove(scriptPath)
+		
+		// Clean up any existing log file
+		os.Remove("/tmp/test_ecr_calls.log")
+		defer os.Remove("/tmp/test_ecr_calls.log")
+
+		epYAML := `
+registries:
+- name: ECR Registry
+  api_url: https://123456789.dkr.ecr.us-east-1.amazonaws.com
+  prefix: 123456789.dkr.ecr.us-east-1.amazonaws.com
+  credentials: ext:` + scriptPath + `
+  credsexpire: 1s
+`
+		epl, err := ParseRegistryConfiguration(epYAML)
+		require.NoError(t, err)
+		require.Len(t, epl.Items, 1)
+
+		// Add registry configuration
+		err = AddRegistryEndpointFromConfig(epl.Items[0])
+		require.NoError(t, err)
+		ep, err := GetRegistryEndpoint("123456789.dkr.ecr.us-east-1.amazonaws.com")
+		require.NoError(t, err)
+
+		// Force credentials to be expired
+		ep.CredsUpdated = time.Now().Add(-2 * time.Second)
+
+		// Launch multiple goroutines to fetch credentials concurrently
+		var wg sync.WaitGroup
+		numGoroutines := 10
+		errors := make([]error, numGoroutines)
+
+		for i := 0; i < numGoroutines; i++ {
+			wg.Add(1)
+			go func(idx int) {
+				defer wg.Done()
+				errors[idx] = ep.SetEndpointCredentials(nil)
+			}(i)
+		}
+
+		wg.Wait()
+
+		// Check that no errors occurred
+		for i, err := range errors {
+			assert.NoError(t, err, "goroutine %d returned error", i)
+		}
+
+		// Verify credentials were set
+		assert.Equal(t, "AWS", ep.Username)
+		assert.Equal(t, "mock-token-12345", ep.Password)
+
+		// Check that the script was called only once
+		data, err := os.ReadFile("/tmp/test_ecr_calls.log")
+		if err != nil {
+			// File might not exist if script wasn't called at all
+			assert.Equal(t, 0, 0)
+		} else {
+			lines := strings.Count(string(data), "counter")
+			assert.Equal(t, 1, lines, "Expected script to be called exactly once, but was called %d times", lines)
+		}
+	})
+
+	t.Run("Concurrent calls with unexpired credentials should not refetch", func(t *testing.T) {
+		var callCount int32
+		
+		epYAML := `
+registries:
+- name: Test Registry
+  api_url: https://test.registry.io
+  prefix: test.registry.io
+  credentials: env:TEST_CONCURRENT_CREDS
+  credsexpire: 10m
+`
+		epl, err := ParseRegistryConfiguration(epYAML)
+		require.NoError(t, err)
+		
+		err = AddRegistryEndpointFromConfig(epl.Items[0])
+		require.NoError(t, err)
+		ep, err := GetRegistryEndpoint("test.registry.io")
+		require.NoError(t, err)
+
+		// Set environment variable
+		os.Setenv("TEST_CONCURRENT_CREDS", "user:pass")
+		
+		// First call to set credentials
+		err = ep.SetEndpointCredentials(nil)
+		require.NoError(t, err)
+		atomic.AddInt32(&callCount, 1)
+
+		// Launch concurrent calls - these should not refetch
+		var wg sync.WaitGroup
+		for i := 0; i < 10; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				err := ep.SetEndpointCredentials(nil)
+				assert.NoError(t, err)
+			}()
+		}
+		wg.Wait()
+
+		// Credentials should still be cached, so total calls should be 1
+		assert.Equal(t, int32(1), atomic.LoadInt32(&callCount))
+		assert.Equal(t, "user", ep.Username)
+		assert.Equal(t, "pass", ep.Password)
+	})
+}