[APMLP-876] Add bucketTagResolver implementation for gradual rollout (#45951)

### What does this PR do?
Add a new `imageresolver.Resolver` implementation called `bucketTagResolver`. There are a few changes in the other files in order to mock the HTTP calls in the resolver testing (namely the addition of the `http.RoundTripper` to the `newHTTPDigestFetcher(...)` constructor).

NOTE: This does **NOT** change existing gradual rollout behavior. The new resolver implementation is additive, and enabling this to be the default will be its own separate PR.

### Motivation
This is specifically for the tag-based gradual rollout initiative, and will allow the gradual rollout of APM tracer libraries and APM Injector releases to subsets of customers by generating a bucket ID between 0-9 based on the API key configured on the cluster agent. This is to ensure as much of an even distribution across the gradual rollout buckets as possible.

This change also uses the new `imageresolver.cache` that was implemented (#45451) to help proactively promote the use of newer releases based on a configurable cache TTL (to be implemented in a subsequent PR, it is currently hard coded to default to 1 hour). If a tag -> digest mapping is expired according to the TTL, or a tag entry is not available in the cache, we will fetch the digest using the new `imageresolver.httpDigestFetcher` (#45653). This is essentially our own implementation of `crane.Digest` (see source code [here](https://github.com/google/go-containerregistry/blob/main/pkg/crane/digest.go)).

### Describe how you validated your changes
- Addition of new unit tests in `resolver_test.go`
- Manual E2E testing with `injector-dev`

### Additional Notes
- (WIP) E2E testing integrated into CI (we will have a separate PR to add these when we actually enable the tag-based gradual rollout in favor of the current remote configuration-based gradual rollout).


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

Author: erikayasuda

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

@@ -8,6 +8,7 @@
 package imageresolver
 
 import (
+	"net/http"
 	"sync"
 	"time"
 )
@@ -27,17 +28,17 @@ type httpDigestCache struct {
 	fetcher *httpDigestFetcher
 }
 
-func (c *httpDigestCache) get(registry string, repository string, tag string) (string, bool) {
+func (c *httpDigestCache) get(registry string, repository string, tag string) (string, error) {
 	if digest := c.checkCache(registry, repository, tag); digest != "" {
-		return digest, true
+		return digest, nil
 	}
 
 	digest, err := c.fetcher.digest(registry + "/" + repository + ":" + tag)
 	if err != nil {
-		return "", false
+		return "", err
 	}
 
-	return c.store(registry, repository, tag, digest), true
+	return c.store(registry, repository, tag, digest), nil
 }
 
 func (c *httpDigestCache) checkCache(registry, repository, tag string) string {
@@ -75,3 +76,16 @@ func (c *httpDigestCache) store(registry, repository, tag, digest string) string
 	}
 	return digest
 }
+
+func newHTTPDigestCache(ttl time.Duration, ddRegistries map[string]struct{}, rt http.RoundTripper) *httpDigestCache {
+	cache := make(registryCache)
+	for registry := range ddRegistries {
+		cache[registry] = make(repositoryCache)
+	}
+
+	return &httpDigestCache{
+		cache:   cache,
+		ttl:     ttl,
+		fetcher: newHTTPDigestFetcher(rt),
+	}
+}

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

@@ -145,9 +145,9 @@ func TestHttpDigestCache_Get_Success(t *testing.T) {
 			tt.setupCache(cc)
 			tt.setupMock(transport)
 
-			resolved, ok := cc.get("test-registry", tt.repository, tt.tag)
+			resolved, err := cc.get("test-registry", tt.repository, tt.tag)
 
-			require.True(t, ok, "Expected successful get")
+			require.NoError(t, err, "Expected successful get")
 			require.NotNil(t, resolved, "Expected non-nil resolved image")
 			require.Equal(t, tt.expectedDigest, resolved)
 			require.Equal(t, tt.expectedCallCount, transport.CallCount())
@@ -157,9 +157,9 @@ func TestHttpDigestCache_Get_Success(t *testing.T) {
 
 func TestHttpDigestCache_Get_Failure(t *testing.T) {
 	cc, transport := mockHTTPDigestCache(1 * time.Minute)
-	resolved, ok := cc.get("test-registry", "dd-lib-python-init", "v1")
+	resolved, err := cc.get("test-registry", "dd-lib-python-init", "v1")
 
-	require.False(t, ok, "Expected failed get")
+	require.Error(t, err, "Expected failed get")
 	require.Empty(t, resolved, "Expected empty digest")
 	require.Equal(t, 1, transport.CallCount())
 }
@@ -169,11 +169,11 @@ func TestHttpDigestCache_Get_MultipleRepositories(t *testing.T) {
 	transport.addImage("registry1", "dd-lib-python-init", "v1", "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
 	transport.addImage("registry2", "dd-lib-java-init", "v2", "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
 
-	resolved1, ok1 := cc.get("registry1", "dd-lib-python-init", "v1")
-	resolved2, ok2 := cc.get("registry2", "dd-lib-java-init", "v2")
+	resolved1, err1 := cc.get("registry1", "dd-lib-python-init", "v1")
+	resolved2, err2 := cc.get("registry2", "dd-lib-java-init", "v2")
 
-	require.True(t, ok1, "Should fetch python lib")
-	require.True(t, ok2, "Should fetch java lib")
+	require.NoError(t, err1, "Should fetch python lib")
+	require.NoError(t, err2, "Should fetch java lib")
 	require.Equal(t, "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", resolved1)
 	require.Equal(t, "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", resolved2)
 	require.Equal(t, 2, transport.CallCount(), "Should have fetched digest twice")
@@ -185,13 +185,13 @@ func TestHttpDigestCache_Get_SameRepoMultipleTags(t *testing.T) {
 	transport.addImage("registry", "dd-lib-python-init", "v2", "sha256:2222222222222222222222222222222222222222222222222222222222222222")
 	transport.addImage("registry", "dd-lib-python-init", "latest", "sha256:ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff")
 
-	resolved1, ok1 := cc.get("registry", "dd-lib-python-init", "v1")
-	resolved2, ok2 := cc.get("registry", "dd-lib-python-init", "v2")
-	resolved3, ok3 := cc.get("registry", "dd-lib-python-init", "latest")
+	resolved1, err1 := cc.get("registry", "dd-lib-python-init", "v1")
+	resolved2, err2 := cc.get("registry", "dd-lib-python-init", "v2")
+	resolved3, err3 := cc.get("registry", "dd-lib-python-init", "latest")
 
-	require.True(t, ok1)
-	require.True(t, ok2)
-	require.True(t, ok3)
+	require.NoError(t, err1)
+	require.NoError(t, err2)
+	require.NoError(t, err3)
 	require.Equal(t, "sha256:1111111111111111111111111111111111111111111111111111111111111111", resolved1)
 	require.Equal(t, "sha256:2222222222222222222222222222222222222222222222222222222222222222", resolved2)
 	require.Equal(t, "sha256:ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff", resolved3)
@@ -202,8 +202,8 @@ func TestHttpDigestCache_Get_ConcurrentCacheHit(t *testing.T) {
 	cc, transport := mockHTTPDigestCache(5 * time.Minute)
 	transport.addImage("registry", "repo", "v1", "sha256:abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890")
 
-	resolved, ok := cc.get("registry", "repo", "v1")
-	require.True(t, ok)
+	resolved, err := cc.get("registry", "repo", "v1")
+	require.NoError(t, err)
 	require.NotNil(t, resolved)
 	require.Equal(t, 1, transport.CallCount(), "Should have made exactly one fetch to warm cache")
 
@@ -213,8 +213,8 @@ func TestHttpDigestCache_Get_ConcurrentCacheHit(t *testing.T) {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			resolved, ok := cc.get("registry", "repo", "v1")
-			require.True(t, ok)
+			resolved, err := cc.get("registry", "repo", "v1")
+			require.NoError(t, err)
 			require.NotNil(t, resolved)
 			require.Equal(t, "sha256:abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890", resolved)
 		}()

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

@@ -37,6 +37,7 @@ type Config struct {
 	MaxInitRetries int
 	InitRetryDelay time.Duration
 	BucketID       string
+	DigestCacheTTL time.Duration
 }
 
 func calculateRolloutBucket(apiKey string) string {
@@ -46,7 +47,6 @@ func calculateRolloutBucket(apiKey string) string {
 	return strconv.Itoa(int(hashInt % rolloutBucketCount))
 }
 
-// NewConfig creates a new Config
 func NewConfig(cfg config.Component, rcClient RemoteConfigClient) Config {
 	return Config{
 		Site:           cfg.GetString("site"),
@@ -55,5 +55,6 @@ func NewConfig(cfg config.Component, rcClient RemoteConfigClient) Config {
 		MaxInitRetries: 5,
 		InitRetryDelay: 1 * time.Second,
 		BucketID:       calculateRolloutBucket(cfg.GetString("api_key")),
+		DigestCacheTTL: 1 * time.Hour, // DEV: Make this configurable
 	}
 }

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

@@ -38,6 +38,7 @@ func TestNewConfig(t *testing.T) {
 				MaxInitRetries: 5,
 				InitRetryDelay: 1 * time.Second,
 				BucketID:       "2",
+				DigestCacheTTL: 1 * time.Hour,
 			},
 		},
 		{
@@ -55,6 +56,7 @@ func TestNewConfig(t *testing.T) {
 				MaxInitRetries: 5,
 				InitRetryDelay: 1 * time.Second,
 				BucketID:       "2",
+				DigestCacheTTL: 1 * time.Hour,
 			},
 		},
 		{
@@ -71,6 +73,7 @@ func TestNewConfig(t *testing.T) {
 				MaxInitRetries: 5,
 				InitRetryDelay: 1 * time.Second,
 				BucketID:       "2",
+				DigestCacheTTL: 1 * time.Hour,
 			},
 		},
 		{
@@ -88,6 +91,7 @@ func TestNewConfig(t *testing.T) {
 				MaxInitRetries: 5,
 				InitRetryDelay: 1 * time.Second,
 				BucketID:       "0",
+				DigestCacheTTL: 1 * time.Hour,
 			},
 		},
 	}

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

@@ -101,12 +101,14 @@ func (h *httpDigestFetcher) digest(ref string) (string, error) {
 	return digest, nil
 }
 
-func newHTTPDigestFetcher() *httpDigestFetcher {
-	transport := http.DefaultTransport.(*http.Transport).Clone()
+func newHTTPDigestFetcher(rt http.RoundTripper) *httpDigestFetcher {
+	if rt == nil {
+		rt = http.DefaultTransport.(*http.Transport).Clone()
+	}
 	return &httpDigestFetcher{
 		client: &http.Client{
 			Timeout:   requestTimeout,
-			Transport: transport,
+			Transport: rt,
 		},
 	}
 }

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

@@ -30,7 +30,10 @@ func makeTestImageRef(server *httptest.Server) string {
 }
 
 func TestHttpDigestFetcher_buildManifestRequest_Success(t *testing.T) {
-	f := newHTTPDigestFetcher()
+	transport := &mockRoundTripper{
+		responses: make(map[string]*http.Response),
+	}
+	f := newHTTPDigestFetcher(transport)
 	tests := []struct {
 		name               string
 		imageRef           string
@@ -94,7 +97,10 @@ func TestHttpDigestFetcher_buildManifestRequest_Success(t *testing.T) {
 }
 
 func TestHttpDigestFetcher_buildManifestRequest_Error(t *testing.T) {
-	f := newHTTPDigestFetcher()
+	transport := &mockRoundTripper{
+		responses: make(map[string]*http.Response),
+	}
+	f := newHTTPDigestFetcher(transport)
 	tests := []struct {
 		name     string
 		imageRef string

…entation/imageresolver/image_resolver.go …nstrumentation/imageresolver/resolver.gopkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver/image_resolver.go renamed to pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver/resolver.go pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver/image_resolver.go renamed to pkg/clusteragent/admission/mutate/autoinstrumentation/imageresolver/resolver.go

@@ -9,6 +9,7 @@ package imageresolver
 
 import (
 	"fmt"
+	"net/http"
 	"reflect"
 	"strings"
 	"sync"
@@ -31,6 +32,8 @@ type Resolver interface {
 // This is used when no remote config client is available.
 type noOpResolver struct{}
 
+var _ Resolver = (*noOpResolver)(nil)
+
 // NewNoOpResolver creates a new noOpImageResolver.
 // This is useful for testing or when image resolution is not needed.
 func NewNoOpResolver() Resolver {
@@ -58,6 +61,8 @@ type rcResolver struct {
 	retryDelay time.Duration
 }
 
+var _ Resolver = (*rcResolver)(nil)
+
 func newRcResolver(cfg Config) Resolver {
 	resolver := &rcResolver{
 		rcClient:            cfg.RCClient,
@@ -211,13 +216,68 @@ func (r *rcResolver) processUpdate(update map[string]state.RawConfig, applyState
 	}
 }
 
+// bucketTagResolver resolves image references using bucket tags.
+// It maintains a cache of image digests and resolves bucket tags to digests if possible.
+// Defaults to returning the original tag if resolution fails for any reason.
+type bucketTagResolver struct {
+	cache               *httpDigestCache
+	bucketID            string
+	datadoghqRegistries map[string]struct{}
+}
+
+var _ Resolver = (*bucketTagResolver)(nil)
+
+func (r *bucketTagResolver) createBucketTag(tag string) string {
+	normalizedTag := strings.TrimPrefix(tag, "v")
+
+	// DEV: Only create bucket tag for major versions (single number like "1" or "v1")
+	if !strings.Contains(normalizedTag, ".") {
+		return fmt.Sprintf("%s-gr%s", normalizedTag, r.bucketID)
+	}
+	return normalizedTag
+}
+
+func (r *bucketTagResolver) Resolve(registry string, repository string, tag string) (*ResolvedImage, bool) {
+	var result = tag
+	var resolvedTag = tag
+	defer func() {
+		metrics.ImageResolutionAttempts.Inc(repository, resolvedTag, result)
+	}()
+	if !isDatadoghqRegistry(registry, r.datadoghqRegistries) {
+		log.Debugf("%s is not a Datadoghq registry, not opting into gradual rollout", registry)
+		return nil, false
+	}
+
+	bucketTag := r.createBucketTag(tag)
+
+	digest, err := r.cache.get(registry, repository, bucketTag)
+	resolvedTag = bucketTag
+	if err != nil {
+		log.Debugf("Failed to resolve %s/%s:%s for gradual rollout - %v", registry, repository, bucketTag, err)
+		return nil, false
+	}
+	result = digest
+	return &ResolvedImage{
+		FullImageRef:     registry + "/" + repository + "@" + digest,
+		CanonicalVersion: tag, // DEV: This is the customer provided tag
+	}, true
+}
+
+func newBucketTagResolver(cfg Config) *bucketTagResolver {
+	rt := http.DefaultTransport.(*http.Transport).Clone()
+	return &bucketTagResolver{
+		cache:               newHTTPDigestCache(cfg.DigestCacheTTL, cfg.DDRegistries, rt),
+		bucketID:            cfg.BucketID,
+		datadoghqRegistries: cfg.DDRegistries,
+	}
+}
+
 // New creates the appropriate Resolver based on whether
 // a remote config client is available.
 func New(cfg Config) Resolver {
 	if cfg.RCClient == nil || reflect.ValueOf(cfg.RCClient).IsNil() {
 		log.Debugf("No remote config client available")
 		return NewNoOpResolver()
 	}
-
 	return newRcResolver(cfg)
 }