[APMLP-819] Add imageresolver.Config to initialize ImageResolvers (#43308)

### What does this PR do?
Adds a new `Config` type in a separate `imageresolver` package. The config should contain all necessary configurations to determine what type of `ImageResolver` implementation should be used.

There is no functionality/behavior change from this PR.

### Motivation
- **Code clean up** -- Removes the different varieties of RC ImageResolver constructors that made it difficult to follow the code.
- **Simpler interface** -- Calling `NewImageResolver()` now only requires the `ImageResolverConfig`, which can be updated later if needed. 
- **`imageresolver` package** -- Starting this so other ImageResolver specific code can be nested in here in a follow up PR

### Describe how you validated your changes
Local E2E validation using `injector-dev` to ensure existing behavior is retained

### Additional Notes


Co-authored-by: erika.yasuda <erika.yasuda@datadoghq.com>

Author: erikayasuda

pkg/clusteragent/admission/mutate/autoinstrumentation/auto_instrumentation.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/DataDog/datadog-agent/comp/core/config"
 	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver"
 	mutatecommon "github.com/DataDog/datadog-agent/pkg/clusteragent/admission/mutate/common"
 	configWebhook "github.com/DataDog/datadog-agent/pkg/clusteragent/admission/mutate/config"
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/mutate/tagsfromlabels"
@@ -28,7 +29,7 @@ func NewAutoInstrumentation(datadogConfig config.Component, wmeta workloadmeta.C
 		return nil, fmt.Errorf("failed to create auto instrumentation config: %v", err)
 	}
 
-	imageResolver := NewImageResolver(rcClient, datadogConfig)
+	imageResolver := NewImageResolver(imageresolver.NewConfig(datadogConfig, rcClient))
 	apm, err := NewTargetMutator(config, wmeta, imageResolver)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create auto instrumentation namespace mutator: %v", err)

pkg/clusteragent/admission/mutate/autoinstrumentation/image_resolver.go

@@ -17,18 +17,12 @@ import (
 
 	"github.com/opencontainers/go-digest"
 
-	"github.com/DataDog/datadog-agent/comp/core/config"
 	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/metrics"
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver"
 	"github.com/DataDog/datadog-agent/pkg/remoteconfig/state"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
-// RemoteConfigClient defines the interface we need for remote config operations
-type RemoteConfigClient interface {
-	GetConfigs(product string) map[string]state.RawConfig
-	Subscribe(product string, callback func(map[string]state.RawConfig, func(string, state.ApplyStatus)))
-}
-
 // ImageResolver resolves container image references from tag-based to digest-based.
 type ImageResolver interface {
 	// Resolve takes a registry, repository, and tag string (e.g., "gcr.io/datadoghq", "dd-lib-python-init", "v3")
@@ -56,7 +50,7 @@ func (r *noOpImageResolver) Resolve(registry string, repository string, tag stri
 // remoteConfigImageResolver resolves image references using remote configuration data.
 // It maintains a cache of image mappings received from the remote config service.
 type remoteConfigImageResolver struct {
-	rcClient RemoteConfigClient
+	rcClient imageresolver.RemoteConfigClient
 
 	mu                  sync.RWMutex
 	imageMappings       map[string]map[string]ImageInfo // repository name -> tag -> resolved image
@@ -67,24 +61,16 @@ type remoteConfigImageResolver struct {
 	retryDelay time.Duration
 }
 
-// newRemoteConfigImageResolver creates a new remoteConfigImageResolver.
-// Assumes rcClient is non-nil.
-func newRemoteConfigImageResolver(rcClient RemoteConfigClient, datadoghqRegistries map[string]struct{}) ImageResolver {
-	return newRemoteConfigImageResolverWithRetryConfig(rcClient, 5, 1*time.Second, datadoghqRegistries)
-}
-
-// newRemoteConfigImageResolverWithRetryConfig creates a resolver with configurable retry behavior.
-// Useful for testing with faster retry settings.
-func newRemoteConfigImageResolverWithRetryConfig(rcClient RemoteConfigClient, maxRetries int, retryDelay time.Duration, datadoghqRegistries map[string]struct{}) ImageResolver {
+func newRcImageResolver(cfg imageresolver.Config) ImageResolver {
 	resolver := &remoteConfigImageResolver{
-		rcClient:            rcClient,
+		rcClient:            cfg.RCClient,
 		imageMappings:       make(map[string]map[string]ImageInfo),
-		maxRetries:          maxRetries,
-		retryDelay:          retryDelay,
-		datadoghqRegistries: datadoghqRegistries,
+		maxRetries:          cfg.MaxInitRetries,
+		retryDelay:          cfg.InitRetryDelay,
+		datadoghqRegistries: cfg.DDRegistries,
 	}
 
-	rcClient.Subscribe(state.ProductGradualRollout, resolver.processUpdate)
+	resolver.rcClient.Subscribe(state.ProductGradualRollout, resolver.processUpdate)
 	log.Debugf("Subscribed to %s", state.ProductGradualRollout)
 
 	go func() {
@@ -286,25 +272,16 @@ func newResolvedImage(registry string, repositoryName string, imageInfo ImageInf
 
 // NewImageResolver creates the appropriate ImageResolver based on whether
 // a remote config client is available.
-func NewImageResolver(rcClient RemoteConfigClient, cfg config.Component) ImageResolver {
-
-	if rcClient == nil || reflect.ValueOf(rcClient).IsNil() {
+func NewImageResolver(cfg imageresolver.Config) ImageResolver {
+	if cfg.RCClient == nil || reflect.ValueOf(cfg.RCClient).IsNil() {
 		log.Debugf("No remote config client available")
 		return newNoOpImageResolver()
 	}
 
-	datadogRegistriesList := cfg.GetStringSlice("admission_controller.auto_instrumentation.default_dd_registries")
-	datadogRegistries := newDatadoghqRegistries(datadogRegistriesList)
-
-	return newRemoteConfigImageResolverWithDefaultDatadoghqRegistries(rcClient, datadogRegistries)
-}
-
-func newRemoteConfigImageResolverWithDefaultDatadoghqRegistries(rcClient RemoteConfigClient, datadoghqRegistries map[string]struct{}) ImageResolver {
-	resolver := newRemoteConfigImageResolver(rcClient, datadoghqRegistries)
-	resolver.(*remoteConfigImageResolver).datadoghqRegistries = datadoghqRegistries
-	return resolver
+	return newRcImageResolver(cfg)
 }
 
+// DEV: Delete this in favor of the imageresolver package one after the refactor is complete
 func newDatadoghqRegistries(datadogRegistriesList []string) map[string]struct{} {
 	datadoghqRegistries := make(map[string]struct{})
 	for _, registry := range datadogRegistriesList {

pkg/clusteragent/admission/mutate/autoinstrumentation/image_resolver_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/DataDog/datadog-agent/comp/core/config"
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver"
 	"github.com/DataDog/datadog-agent/pkg/remoteconfig/state"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
@@ -107,24 +108,24 @@ func (m *mockRCClient) setBlocking(block bool) {
 func TestNewImageResolver(t *testing.T) {
 	t.Run("with_remote_config_client", func(t *testing.T) {
 		mockClient := newMockRCClient("image_resolver_multi_repo.json")
-		mockConfig := config.NewMock(t)
-		resolver := NewImageResolver(mockClient, mockConfig)
+		mockConfig := imageresolver.NewConfig(config.NewMock(t), mockClient)
+		resolver := NewImageResolver(mockConfig)
 
 		_, ok := resolver.(*remoteConfigImageResolver)
 		assert.True(t, ok, "Should return remoteConfigImageResolver when rcClient is not nil")
 	})
 
 	t.Run("without_remote_config_client__typed_nil", func(t *testing.T) {
-		mockConfig := config.NewMock(t)
-		resolver := NewImageResolver((*mockRCClient)(nil), mockConfig)
+		mockConfig := imageresolver.NewConfig(config.NewMock(t), (*mockRCClient)(nil))
+		resolver := NewImageResolver(mockConfig)
 
 		_, ok := resolver.(*noOpImageResolver)
 		assert.True(t, ok, "Should return noOpImageResolver when rcClient is nil")
 	})
 
 	t.Run("without_remote_config_client__untyped_nil", func(t *testing.T) {
-		mockConfig := config.NewMock(t)
-		resolver := NewImageResolver(nil, mockConfig)
+		mockConfig := imageresolver.NewConfig(config.NewMock(t), nil)
+		resolver := NewImageResolver(mockConfig)
 
 		_, ok := resolver.(*noOpImageResolver)
 		assert.True(t, ok, "Should return noOpImageResolver when rcClient is nil")
@@ -228,8 +229,7 @@ func TestImageResolverEmptyConfig(t *testing.T) {
 
 func TestRemoteConfigImageResolver_Resolve(t *testing.T) {
 	mockRCClient := newMockRCClient("image_resolver_multi_repo.json")
-	datadoghqRegistries := newDatadoghqRegistries(config.NewMock(t).GetStringSlice("admission_controller.auto_instrumentation.default_dd_registries"))
-	resolver := newRemoteConfigImageResolver(mockRCClient, datadoghqRegistries)
+	resolver := newRcImageResolver(imageresolver.NewConfig(config.NewMock(t), mockRCClient))
 
 	testCases := []struct {
 		name           string
@@ -414,8 +414,7 @@ func TestRemoteConfigImageResolver_InvalidDigestValidation(t *testing.T) {
 }
 
 func TestRemoteConfigImageResolver_ConcurrentAccess(t *testing.T) {
-	datadoghqRegistries := newDatadoghqRegistries(config.NewMock(t).GetStringSlice("admission_controller.auto_instrumentation.default_dd_registries"))
-	resolver := newRemoteConfigImageResolver(newMockRCClient("image_resolver_multi_repo.json"), datadoghqRegistries).(*remoteConfigImageResolver)
+	resolver := newRcImageResolver(imageresolver.NewConfig(config.NewMock(t), newMockRCClient("image_resolver_multi_repo.json")))
 
 	t.Run("concurrent_read_write", func(_ *testing.T) {
 		var wg sync.WaitGroup
@@ -437,7 +436,7 @@ func TestRemoteConfigImageResolver_ConcurrentAccess(t *testing.T) {
 			go func() {
 				defer wg.Done()
 				for j := 0; j < 10; j++ {
-					resolver.processUpdate(map[string]state.RawConfig{}, func(string, state.ApplyStatus) {})
+					resolver.(*remoteConfigImageResolver).processUpdate(map[string]state.RawConfig{}, func(string, state.ApplyStatus) {})
 					time.Sleep(10 * time.Millisecond)
 				}
 			}()
@@ -495,9 +494,8 @@ func TestAsyncInitialization(t *testing.T) {
 	t.Run("noop_during_initialization", func(t *testing.T) {
 		mockClient := newMockRCClient("image_resolver_multi_repo.json")
 		mockClient.setBlocking(true) // Block initialization
-		datadoghqRegistries := newDatadoghqRegistries(config.NewMock(t).GetStringSlice("admission_controller.auto_instrumentation.default_dd_registries"))
 
-		resolver := newRemoteConfigImageResolverWithRetryConfig(mockClient, 2, 10*time.Millisecond, datadoghqRegistries)
+		resolver := newRcImageResolver(imageresolver.NewConfig(config.NewMock(t), mockClient))
 
 		resolved, ok := resolver.Resolve("gcr.io/datadoghq", "dd-lib-python-init", "latest")
 		assert.False(t, ok, "Should not complete image resolution during initialization")
@@ -508,8 +506,7 @@ func TestAsyncInitialization(t *testing.T) {
 		mockClient := newMockRCClient("image_resolver_multi_repo.json")
 		mockClient.setBlocking(true)
 
-		datadoghqRegistries := newDatadoghqRegistries(config.NewMock(t).GetStringSlice("admission_controller.auto_instrumentation.default_dd_registries"))
-		resolver := newRemoteConfigImageResolverWithRetryConfig(mockClient, 2, 10*time.Millisecond, datadoghqRegistries)
+		resolver := newRcImageResolver(imageresolver.NewConfig(config.NewMock(t), mockClient))
 
 		resolved, ok := resolver.Resolve("gcr.io/datadoghq", "dd-lib-python-init", "latest")
 		assert.False(t, ok, "Should not complete image resolution during initialization")
@@ -533,8 +530,7 @@ func TestAsyncInitialization(t *testing.T) {
 		}
 		close(mockClient.configsReady)
 
-		datadoghqRegistries := newDatadoghqRegistries(config.NewMock(t).GetStringSlice("admission_controller.auto_instrumentation.default_dd_registries"))
-		resolver := newRemoteConfigImageResolverWithRetryConfig(mockClient, 2, 10*time.Millisecond, datadoghqRegistries)
+		resolver := newRcImageResolver(imageresolver.NewConfig(config.NewMock(t), mockClient))
 		time.Sleep(50 * time.Millisecond)
 
 		resolved, ok := resolver.Resolve("gcr.io/datadoghq", "dd-lib-python-init", "latest")

pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver/config.go

@@ -0,0 +1,43 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+//go:build kubeapiserver
+
+// Package imageresolver provides configuration and utilities for resolving
+// container image references from mutable tags to digests.
+package imageresolver
+
+import (
+	"time"
+
+	"github.com/DataDog/datadog-agent/comp/core/config"
+	"github.com/DataDog/datadog-agent/pkg/remoteconfig/state"
+)
+
+// RemoteConfigClient defines the interface we need for remote config operations
+type RemoteConfigClient interface {
+	GetConfigs(product string) map[string]state.RawConfig
+	Subscribe(product string, callback func(map[string]state.RawConfig, func(string, state.ApplyStatus)))
+}
+
+// Config contains information needed to create an ImageResolver
+type Config struct {
+	Site           string
+	DDRegistries   map[string]struct{}
+	RCClient       RemoteConfigClient
+	MaxInitRetries int
+	InitRetryDelay time.Duration
+}
+
+// NewConfig creates a new Config
+func NewConfig(cfg config.Component, rcClient RemoteConfigClient) Config {
+	return Config{
+		Site:           cfg.GetString("site"),
+		DDRegistries:   newDatadoghqRegistries(cfg.GetStringSlice("admission_controller.auto_instrumentation.default_dd_registries")),
+		RCClient:       rcClient,
+		MaxInitRetries: 5,
+		InitRetryDelay: 1 * time.Second,
+	}
+}

pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver/config_test.go

@@ -0,0 +1,81 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+//go:build kubeapiserver
+
+package imageresolver
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/DataDog/datadog-agent/comp/core/config"
+)
+
+func TestNewConfig(t *testing.T) {
+	tests := []struct {
+		name          string
+		configFactory func(*testing.T) config.Component
+		expectedState Config
+	}{
+		{
+			name: "default_config",
+			configFactory: func(t *testing.T) config.Component {
+				mockConfig := config.NewMock(t)
+				mockConfig.SetWithoutSource("site", "datadoghq.com")
+				return mockConfig
+			},
+			expectedState: Config{
+				Site:           "datadoghq.com",
+				DDRegistries:   map[string]struct{}{"gcr.io/datadoghq": {}, "docker.io/datadog": {}, "public.ecr.aws/datadog": {}},
+				RCClient:       nil,
+				MaxInitRetries: 5,
+				InitRetryDelay: 1 * time.Second,
+			},
+		},
+		{
+			name: "custom_dd_registries",
+			configFactory: func(t *testing.T) config.Component {
+				mockConfig := config.NewMock(t)
+				mockConfig.SetWithoutSource("site", "datadoghq.com")
+				mockConfig.SetWithoutSource("admission_controller.auto_instrumentation.default_dd_registries", []string{"helloworld.io/datadog"})
+				return mockConfig
+			},
+			expectedState: Config{
+				Site:           "datadoghq.com",
+				DDRegistries:   map[string]struct{}{"helloworld.io/datadog": {}},
+				RCClient:       nil,
+				MaxInitRetries: 5,
+				InitRetryDelay: 1 * time.Second,
+			},
+		},
+		{
+			name: "configured_site",
+			configFactory: func(t *testing.T) config.Component {
+				mockConfig := config.NewMock(t)
+				mockConfig.SetWithoutSource("site", "datad0g.com")
+				return mockConfig
+			},
+			expectedState: Config{
+				Site:           "datad0g.com",
+				DDRegistries:   map[string]struct{}{"gcr.io/datadoghq": {}, "docker.io/datadog": {}, "public.ecr.aws/datadog": {}},
+				RCClient:       nil,
+				MaxInitRetries: 5,
+				InitRetryDelay: 1 * time.Second,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockConfig := tt.configFactory(t)
+			result := NewConfig(mockConfig, nil)
+
+			require.Equal(t, tt.expectedState, result)
+		})
+	}
+}

pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver/util.go

@@ -0,0 +1,18 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+//go:build kubeapiserver
+
+// Package imageresolver provides configuration and utilities for resolving
+// container image references from mutable tags to digests.
+package imageresolver
+
+func newDatadoghqRegistries(datadogRegistriesList []string) map[string]struct{} {
+	datadoghqRegistries := make(map[string]struct{})
+	for _, registry := range datadogRegistriesList {
+		datadoghqRegistries[registry] = struct{}{}
+	}
+	return datadoghqRegistries
+}

pkg/clusteragent/admission/mutate/autoinstrumentation/injector_test.go

@@ -16,6 +16,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 
 	"github.com/DataDog/datadog-agent/comp/core/config"
+	"github.com/DataDog/datadog-agent/pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver"
 	"github.com/DataDog/datadog-agent/pkg/util/pointer"
 )
 
@@ -97,13 +98,7 @@ func TestInjectorWithRemoteConfigImageResolver(t *testing.T) {
 			var resolver ImageResolver
 			if tc.hasRemoteData {
 				mockClient := newMockRCClient("image_resolver_multi_repo.json")
-				datadoghqRegistries := newDatadoghqRegistries(config.NewMock(t).GetStringSlice("admission_controller.auto_instrumentation.default_dd_registries"))
-				resolver = newRemoteConfigImageResolverWithRetryConfig(
-					mockClient,
-					2,
-					1*time.Millisecond,
-					datadoghqRegistries,
-				)
+				resolver = newRcImageResolver(imageresolver.NewConfig(config.NewMock(t), mockClient))
 			} else {
 				resolver = newNoOpImageResolver()
 			}
@@ -119,13 +114,7 @@ func TestInjectorWithRemoteConfigImageResolver(t *testing.T) {
 
 func TestInjectorWithRemoteConfigImageResolverAfterInit(t *testing.T) {
 	mockClient := newMockRCClient("image_resolver_multi_repo.json")
-	datadoghqRegistries := newDatadoghqRegistries(config.NewMock(t).GetStringSlice("admission_controller.auto_instrumentation.default_dd_registries"))
-	resolver := newRemoteConfigImageResolverWithRetryConfig(
-		mockClient,
-		2,
-		1*time.Millisecond,
-		datadoghqRegistries,
-	)
+	resolver := newRcImageResolver(imageresolver.NewConfig(config.NewMock(t), mockClient))
 
 	assert.Eventually(t, func() bool {
 		_, ok := resolver.Resolve("gcr.io/datadoghq", "apm-inject", "0")