From 9223b980406f51abac583c6e55d74346443a9882 Mon Sep 17 00:00:00 2001 From: Ravi Agarwal Date: Sat, 7 Nov 2026 06:58:00 +0000 Subject: [PATCH 1/2] feat(strategy): ecr request metrics Count ECR requests by kind, method and result (cache hit or miss, blob redirect, blob metadata, not found, rate limited, not served, failure), failures by reason, and authorization token events. Labels come from fixed sets; repository names and digests are never labels. --- README.md | 8 ++ internal/strategy/ecr.go | 102 ++++++++++++++---- internal/strategy/ecr_metrics.go | 175 +++++++++++++++++++++++++++++++ internal/strategy/ecr_test.go | 131 +++++++++++++++++++++-- 4 files changed, 387 insertions(+), 29 deletions(-) create mode 100644 internal/strategy/ecr_metrics.go diff --git a/README.md b/README.md index 5c2bd36a..f4a62104 100644 --- a/README.md +++ b/README.md @@ -79,6 +79,14 @@ Read-only proxy for one Amazon ECR registry, including its pull-through cache re - The registry at the host root (`go-containerregistry`, Helm, Docker) or under `/{registry}/v2/` (containerd `override_path`). Only one `ecr` block can own the root; set `prefix-only` on the others. - Helm needs a release that pulls `oci:///@sha256:` without listing tags. Digest pulls arrived in 3.17, and 3.19 is tested; 3.16 calls `tags/list` first and fails on the proxy's `404`, since tags are not served. +**Metrics** + +- `cachew.ecr.requests_total` counts requests by `kind` (`manifest`, `blob`, or `unknown` for a path that names neither), `method` and `result`: `hit` or `miss` for manifests, `redirect` for blob `GET`s, `metadata` for blob `HEAD`s, `not_found`, `rate_limited`, `not_served` for paths and repositories the proxy refuses, or `failure`. The version check is not counted. +- `cachew.ecr.failures_total` counts each failure by `kind` and `reason`: `digest_mismatch`, `no_length`, `partial`, `rejected` (ECR refused the proxy's token), `authorization` (no token could be minted or refreshed), `bad_redirect`, `transport`, `upstream_status`, `cache`, or `internal`. A manifest whose bytes fail verification is a failure even though its `200` was already sent. +- `cachew.ecr.auth_events_total` counts token `reuse`, `refresh`, `forced_refresh` (after a rejection) and `failure` events. + +Repository names and digests are never labels. + Cache hits are not re-hashed, so keep remote `/api/v1/object` writes denied for the `ecr` namespace (the default policy denies remote `/api`); the trust model matches `host` and `codeartifact`. ```hcl diff --git a/internal/strategy/ecr.go b/internal/strategy/ecr.go index 9d8bfe31..37caf901 100644 --- a/internal/strategy/ecr.go +++ b/internal/strategy/ecr.go @@ -104,6 +104,7 @@ type ECR struct { blobClient *http.Client // blobs: returns ECR's redirect unfollowed cache cache.Cache handler *handler.Handler + metric ecrMetricRecorder now func() time.Time // lastForcedRefresh is the UnixNano time of the last refresh a rejection // triggered; see ecrForcedRefreshInterval. @@ -242,8 +243,9 @@ func newECR(config ECRConfig, tokens originTokenSource, base http.RoundTripper, Transport: transport, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, }, - cache: c, - now: time.Now, + cache: c, + metric: newECRMetrics(), + now: time.Now, } s.handler = handler.New(s.client, c). CacheKey(func(r *http.Request) string { @@ -269,29 +271,42 @@ func newECR(config ECRConfig, tokens originTokenSource, base http.RoundTripper, servePing(w, r) return } + ctx, observation := withECRObservation(r.Context()) + r = r.WithContext(ctx) + rw := &ecrResponse{ResponseWriter: w} + kind := ecrKindUnknown + defer func() { + s.metric.recordRequest(ctx, observation.outcome(kind, r.Method, rw.status, rw.Header().Get(handler.CacheResultHeader))) + }() // Reject anything but a digest path before the cache is consulted, so an // entry another strategy wrote under a similar key can never be served here. ref, err := parseECRPath(rest) if err != nil { - http.Error(w, err.Error()+": only /v2//manifests/sha256: and /v2//blobs/sha256: are served", http.StatusNotFound) + observation.refuse() + http.Error(rw, err.Error()+": only /v2//manifests/sha256: and /v2//blobs/sha256: are served", http.StatusNotFound) return } + kind = ecrKindManifest + if ref.kind == ecrBlobs { + kind = ecrKindBlob + } if !s.allowed(ref.name) { - http.Error(w, "repository is not served by this proxy", http.StatusNotFound) + observation.refuse() + http.Error(rw, "repository is not served by this proxy", http.StatusNotFound) return } if ref.kind == ecrBlobs { - s.serveBlob(w, r, ref) + s.serveBlob(rw, r, ref) return } // The upstream fetch is always identity-encoded, so the handler must // not vary the cache key on what the client would have accepted. r.Header.Del("Accept-Encoding") if r.Method == http.MethodHead { - s.serveManifestHead(w, r, ref) + s.serveManifestHead(rw, r, ref) return } - s.handler.ServeHTTP(&hideExpiry{ResponseWriter: w}, r) + s.handler.ServeHTTP(rw, r) }) // GET patterns also match HEAD. Clients such as go-containerregistry, Helm // and Docker want the registry at the host root; containerd mirrors can use @@ -329,7 +344,7 @@ func (s *ECR) serveBlob(w http.ResponseWriter, r *http.Request, ref ecrRef) { raw := resp.Header.Get("Location") location, err := url.Parse(raw) if err != nil || location.Scheme != ecrSchemeHTTPS || location.Host == "" { - s.blobFailure(w, r, "ECR blob redirect is not an absolute https URL", resp.StatusCode) + s.blobFailure(w, r, ecrFailureBadRedirect, "ECR blob redirect is not an absolute https URL", resp.StatusCode) return } // Copied verbatim: re-serializing could change the signed URL. @@ -339,11 +354,11 @@ func (s *ECR) serveBlob(w http.ResponseWriter, r *http.Request, ref ecrRef) { w.WriteHeader(http.StatusTemporaryRedirect) case r.Method == http.MethodHead && resp.StatusCode == http.StatusOK: if digest := resp.Header.Get("Docker-Content-Digest"); digest != ref.digest { - s.blobFailure(w, r, "ECR reported a different digest for the blob", resp.StatusCode) + s.blobFailure(w, r, ecrFailureDigestMismatch, "ECR reported a different digest for the blob", resp.StatusCode) return } if resp.ContentLength < 0 { - s.blobFailure(w, r, "ECR blob HEAD has no Content-Length", resp.StatusCode) + s.blobFailure(w, r, ecrFailureNoLength, "ECR blob HEAD has no Content-Length", resp.StatusCode) return } w.Header().Set("Content-Length", strconv.FormatInt(resp.ContentLength, 10)) @@ -357,15 +372,16 @@ func (s *ECR) serveBlob(w http.ResponseWriter, r *http.Request, ref ecrRef) { } http.Error(w, http.StatusText(resp.StatusCode), resp.StatusCode) default: - s.blobFailure(w, r, "unexpected ECR blob response", resp.StatusCode) + s.blobFailure(w, r, ecrFailureUpstreamStatus, "unexpected ECR blob response", resp.StatusCode) } } // blobFailure logs the upstream status and answers 502. It never logs or echoes // the Location, which carries a presigned credential. -func (s *ECR) blobFailure(w http.ResponseWriter, r *http.Request, reason string, status int) { - logging.FromContext(r.Context()).ErrorContext(r.Context(), reason, "registry", s.registry, "method", r.Method, "status", status) - http.Error(w, reason, http.StatusBadGateway) +func (s *ECR) blobFailure(w http.ResponseWriter, r *http.Request, reason ecrFailureReason, msg string, status int) { + ecrObservationFrom(r.Context()).fail(reason) + logging.FromContext(r.Context()).ErrorContext(r.Context(), msg, "registry", s.registry, "method", r.Method, "status", status) + http.Error(w, msg, http.StatusBadGateway) } // serveManifestHead answers a manifest HEAD from the cache entry's metadata. On @@ -395,6 +411,7 @@ func (s *ECR) serveManifestHead(w http.ResponseWriter, r *http.Request, ref ecrR if errors.Is(err, os.ErrNotExist) { if !fill.complete() { // The fill was cut short: the bytes did not match the digest. + ecrObservationFrom(r.Context()).fail(ecrFailureDigestMismatch) httputil.ErrorResponse(w, r, http.StatusBadGateway, "ECR returned bytes that do not match "+ref.digest) return } @@ -409,6 +426,7 @@ func (s *ECR) serveManifestHead(w http.ResponseWriter, r *http.Request, ref ecrR w.Header().Set(handler.CacheResultHeader, result) if !httputil.ServeCacheStat(w, headers, err) { w.Header().Del(handler.CacheResultHeader) + ecrObservationFrom(r.Context()).fail(ecrFailureCache) logging.FromContext(r.Context()).ErrorContext(r.Context(), "Failed to stat ECR cache entry", "error", err) http.Error(w, "cache unavailable", http.StatusInternalServerError) } @@ -446,21 +464,24 @@ func (r *statusRecorder) complete() bool { return ok && err == nil && r.written == length } -// hideExpiry keeps the cache's internal expiry header out of client responses. -type hideExpiry struct { +// ecrResponse keeps the cache's internal expiry header out of client responses +// and remembers the status for metrics. +type ecrResponse struct { http.ResponseWriter wroteHeader bool + status int } -func (w *hideExpiry) WriteHeader(status int) { +func (w *ecrResponse) WriteHeader(status int) { if !w.wroteHeader { w.wroteHeader = true + w.status = status w.Header().Del(cache.ExpirationKey) } w.ResponseWriter.WriteHeader(status) } -func (w *hideExpiry) Write(p []byte) (int, error) { +func (w *ecrResponse) Write(p []byte) (int, error) { if !w.wroteHeader { w.WriteHeader(http.StatusOK) } @@ -468,14 +489,14 @@ func (w *hideExpiry) Write(p []byte) (int, error) { } // ReadFrom keeps the server's sendfile path for disk-cache hits. -func (w *hideExpiry) ReadFrom(r io.Reader) (int64, error) { +func (w *ecrResponse) ReadFrom(r io.Reader) (int64, error) { if !w.wroteHeader { w.WriteHeader(http.StatusOK) } return io.Copy(w.ResponseWriter, r) //nolint:wrapcheck // ResponseWriter callers expect its errors unwrapped. } -func (w *hideExpiry) Unwrap() http.ResponseWriter { return w.ResponseWriter } +func (w *ecrResponse) Unwrap() http.ResponseWriter { return w.ResponseWriter } func (s *ECR) String() string { return "ecr:" + s.registry } @@ -502,6 +523,7 @@ func (s *ECR) allowed(name string) bool { // errors can quote a raw Location, which may be a presigned S3 URL. Only this // strategy's errors, which name hosts alone, are echoed. func writeECRError(err error, w http.ResponseWriter, r *http.Request) { + ecrObservationFrom(r.Context()).fail(ecrFailureFor(err)) for _, logged := range []error{errECRRejected, errECRRefresh} { if errors.Is(err, logged) { // The transport already logged it with its detail. @@ -524,6 +546,27 @@ func writeECRError(err error, w http.ResponseWriter, r *http.Request) { httputil.ErrorResponse(w, r, http.StatusInternalServerError, err.Error()) } +func ecrFailureFor(err error) ecrFailureReason { + switch { + case errors.Is(err, errECRRejected): + return ecrFailureRejected + case errors.Is(err, errECRRefresh): + return ecrFailureAuthorization + case errors.Is(err, errDigestFailed): + return ecrFailureDigestMismatch + case errors.Is(err, errNoLength): + return ecrFailureNoLength + case errors.Is(err, errPartial): + return ecrFailurePartial + case errors.Is(err, errTooManyHops), errors.Is(err, errInsecureHop): + return ecrFailureBadRedirect + case errors.Is(err, errECRTransport): + return ecrFailureTransport + default: + return ecrFailureInternal + } +} + func ecrOwnError(err error) bool { for _, own := range []error{errTooManyHops, errInsecureHop, errDigestFailed, errNoLength, errPartial, errECRTransport} { if errors.Is(err, own) { @@ -568,7 +611,9 @@ func parseECRPath(rest string) (ecrRef, error) { // token to the registry only, and verifies manifest bytes. func (s *ECR) request(ctx context.Context, ref ecrRef, method string) (*http.Request, error) { token, err := s.tokens.Token(ctx, 0) + s.recordAuth(ctx, token, err) if err != nil { + ecrObservationFrom(ctx).fail(ecrFailureAuthorization) // AWS errors name the role and account; log them, answer with the status. logging.FromContext(ctx).ErrorContext(ctx, "Failed to obtain ECR authorization", "error", err) return nil, httputil.Errorf(http.StatusBadGateway, "could not obtain ECR authorization") @@ -612,6 +657,15 @@ type ecrFetch struct { expiresAt time.Time // zero for blobs } +func (s *ECR) recordAuth(ctx context.Context, token originToken, err error) { + switch { + case err != nil: + s.metric.recordAuth(ctx, originAuthFailure) + case token.event != "": + s.metric.recordAuth(ctx, token.event) + } +} + // allowForcedRefresh reports whether a rejection may mint a new token now. func (s *ECR) allowForcedRefresh() bool { now := s.now().UnixNano() @@ -702,6 +756,7 @@ func (t ecrTransport) retryRejected(req *http.Request, resp *http.Response, fetc rejectedGeneration = fetch.token.generation } refreshed, err := fetch.strategy.tokens.Token(req.Context(), rejectedGeneration) + fetch.strategy.recordAuth(req.Context(), refreshed, err) if err != nil { _ = resp.Body.Close() logging.FromContext(req.Context()).ErrorContext(req.Context(), "Failed to refresh ECR authorization", "error", err) @@ -729,7 +784,9 @@ func verifyECRResponse(req *http.Request, resp *http.Response, fetch *ecrFetch) _ = resp.Body.Close() return nil, errors.Wrapf(errDigestFailed, "%s returned an empty body for sha256:%s", req.URL.Host, expected) } - resp.Body = &digestVerifyingBody{ReadCloser: resp.Body, hash: sha256.New(), expected: expected} + resp.Body = &digestVerifyingBody{ + ReadCloser: resp.Body, hash: sha256.New(), expected: expected, observation: ecrObservationFrom(req.Context()), + } // ECR sends this on HEAD but not on GET for manifests; clients that verify // or resolve by it expect it on both, and the digest is checked as it streams. resp.Header.Set("Docker-Content-Digest", fetch.digest) @@ -785,6 +842,8 @@ type digestVerifyingBody struct { held []byte verified bool failed error + // observation records a mismatch, which the client sees only as a short body. + observation *ecrObservation } func (b *digestVerifyingBody) Read(p []byte) (int, error) { @@ -803,6 +862,7 @@ func (b *digestVerifyingBody) Read(p []byte) (int, error) { if errors.Is(err, io.EOF) { if actual := hex.EncodeToString(b.hash.Sum(nil)); actual != b.expected { b.held = nil + b.observation.fail(ecrFailureDigestMismatch) b.failed = errors.Wrapf(errDigestFailed, "expected sha256:%s, got sha256:%s", b.expected, actual) return 0, b.failed } diff --git a/internal/strategy/ecr_metrics.go b/internal/strategy/ecr_metrics.go new file mode 100644 index 00000000..dfbd968d --- /dev/null +++ b/internal/strategy/ecr_metrics.go @@ -0,0 +1,175 @@ +package strategy + +import ( + "context" + "net/http" + "sync" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + + cachewmetrics "github.com/block/cachew/internal/metrics" + "github.com/block/cachew/internal/strategy/handler" +) + +type ecrResult string + +const ( + ecrResultHit ecrResult = "hit" + ecrResultMiss ecrResult = "miss" + ecrResultRedirect ecrResult = "redirect" + ecrResultMetadata ecrResult = "metadata" + ecrResultNotFound ecrResult = "not_found" + ecrResultRateLimited ecrResult = "rate_limited" + ecrResultNotServed ecrResult = "not_served" + ecrResultFailure ecrResult = "failure" +) + +type ecrFailureReason string + +const ( + ecrFailureDigestMismatch ecrFailureReason = "digest_mismatch" + ecrFailureNoLength ecrFailureReason = "no_length" + ecrFailurePartial ecrFailureReason = "partial" + ecrFailureRejected ecrFailureReason = "rejected" + ecrFailureAuthorization ecrFailureReason = "authorization" + ecrFailureBadRedirect ecrFailureReason = "bad_redirect" + ecrFailureTransport ecrFailureReason = "transport" + ecrFailureUpstreamStatus ecrFailureReason = "upstream_status" + ecrFailureCache ecrFailureReason = "cache" + ecrFailureInternal ecrFailureReason = "internal" +) + +const ( + ecrKindManifest = "manifest" + ecrKindBlob = "blob" + // ecrKindUnknown labels requests refused before the path names a manifest + // or blob. + ecrKindUnknown = "unknown" +) + +// ecrRequestOutcome is what one client request recorded. Every attribute is +// from a fixed set; repository names and digests are never labels. +type ecrRequestOutcome struct { + kind string // manifest, blob or unknown + method string + result ecrResult + reason ecrFailureReason // set when result is failure +} + +type ecrMetricRecorder interface { + recordRequest(context.Context, ecrRequestOutcome) + recordAuth(context.Context, originAuthEvent) +} + +type ecrMetrics struct { + requests metric.Int64Counter + failures metric.Int64Counter + auth metric.Int64Counter +} + +func newECRMetrics() *ecrMetrics { + meter := otel.Meter("cachew.ecr") + return &ecrMetrics{ + requests: cachewmetrics.NewMetric[metric.Int64Counter](meter, "cachew.ecr.requests_total", "{requests}", "ECR requests by kind, method and result"), + failures: cachewmetrics.NewMetric[metric.Int64Counter](meter, "cachew.ecr.failures_total", "{failures}", "Failed ECR requests by kind and reason"), + auth: cachewmetrics.NewMetric[metric.Int64Counter](meter, "cachew.ecr.auth_events_total", "{events}", "ECR authorization token events"), + } +} + +func (m *ecrMetrics) recordRequest(ctx context.Context, outcome ecrRequestOutcome) { + m.requests.Add(ctx, 1, metric.WithAttributes( + attribute.String("kind", outcome.kind), + attribute.String("method", outcome.method), + attribute.String("result", string(outcome.result)), + )) + if outcome.result == ecrResultFailure { + m.failures.Add(ctx, 1, metric.WithAttributes( + attribute.String("kind", outcome.kind), + attribute.String("reason", string(outcome.reason)), + )) + } +} + +func (m *ecrMetrics) recordAuth(ctx context.Context, event originAuthEvent) { + m.auth.Add(ctx, 1, metric.WithAttributes(attribute.String("event", string(event)))) +} + +// ecrObservation collects one request's failure cause as it is found: in the +// serving goroutine, the transport, or the handler's fill goroutine that reads +// the digest-verified body. The first cause wins. +type ecrObservation struct { + mu sync.Mutex + reason ecrFailureReason + notServed bool +} + +type ecrObservationKey struct{} + +func withECRObservation(ctx context.Context) (context.Context, *ecrObservation) { + o := &ecrObservation{} + return context.WithValue(ctx, ecrObservationKey{}, o), o +} + +// ecrObservationFrom returns nil outside a served request; its methods accept +// a nil receiver. +func ecrObservationFrom(ctx context.Context) *ecrObservation { + o, _ := ctx.Value(ecrObservationKey{}).(*ecrObservation) + return o +} + +func (o *ecrObservation) fail(reason ecrFailureReason) { + if o == nil { + return + } + o.mu.Lock() + defer o.mu.Unlock() + if o.reason == "" { + o.reason = reason + } +} + +func (o *ecrObservation) refuse() { + if o == nil { + return + } + o.mu.Lock() + defer o.mu.Unlock() + o.notServed = true +} + +// outcome classifies a finished request from its status, its cache result +// header and any recorded cause. A manifest whose digest failed mid-stream +// still answered 200, so the recorded cause takes precedence over the status. +func (o *ecrObservation) outcome(kind, method string, status int, cacheResult string) ecrRequestOutcome { + o.mu.Lock() + reason, notServed := o.reason, o.notServed + o.mu.Unlock() + outcome := ecrRequestOutcome{kind: kind, method: method} + if status == 0 { + status = http.StatusOK + } + switch { + case notServed: + outcome.result = ecrResultNotServed + case reason != "": + outcome.result, outcome.reason = ecrResultFailure, reason + case status == http.StatusTemporaryRedirect: + outcome.result = ecrResultRedirect + case status == http.StatusNotFound: + outcome.result = ecrResultNotFound + case status == http.StatusTooManyRequests: + outcome.result = ecrResultRateLimited + case cacheResult == handler.CacheResultHit && status < http.StatusInternalServerError: + // Includes 304s and range answers served from the cache entry. + outcome.result = ecrResultHit + case status >= http.StatusBadRequest: + outcome.result, outcome.reason = ecrResultFailure, ecrFailureUpstreamStatus + case kind == ecrKindBlob: + outcome.result = ecrResultMetadata + default: + outcome.result = ecrResultMiss + } + return outcome +} diff --git a/internal/strategy/ecr_test.go b/internal/strategy/ecr_test.go index b182c44c..2248461a 100644 --- a/internal/strategy/ecr_test.go +++ b/internal/strategy/ecr_test.go @@ -189,19 +189,38 @@ func (f *fakeTokens) Token(_ context.Context, rejected uint64) (originToken, err if f.err != nil { return originToken{}, f.err } + event := originAuthReuse if f.gen == 0 || (rejected != 0 && rejected >= f.gen) { + event = originAuthRefresh + if f.gen != 0 { + event = originAuthForcedRefresh + } f.gen++ } - return originToken{value: "tok-" + string(rune('0'+f.gen)), generation: f.gen}, nil + return originToken{value: "tok-" + string(rune('0'+f.gen)), generation: f.gen, event: event}, nil +} + +type recordingECRMetrics struct { + requests []ecrRequestOutcome + auth []originAuthEvent +} + +func (m *recordingECRMetrics) recordRequest(_ context.Context, outcome ecrRequestOutcome) { + m.requests = append(m.requests, outcome) +} + +func (m *recordingECRMetrics) recordAuth(_ context.Context, event originAuthEvent) { + m.auth = append(m.auth, event) } type ecrTest struct { - fake *fakeECR - mux *http.ServeMux - cache cache.Cache - ctx context.Context - tokens *fakeTokens - ecr *ECR + fake *fakeECR + mux *http.ServeMux + cache cache.Cache + ctx context.Context + tokens *fakeTokens + ecr *ECR + metrics *recordingECRMetrics } func setupECRTest(t *testing.T, configure func(*fakeECR)) *ecrTest { @@ -235,7 +254,9 @@ func setupECRTestWith(t *testing.T, configure func(*fakeECR), override cache.Cac s, err := newECR(config, tokens, pinTransport{backend: fake.server}, c, mux) assert.NoError(t, err) assert.Equal(t, "ecr:"+testECRRegistry, s.String()) - return &ecrTest{fake: fake, mux: mux, cache: c, ctx: ctx, tokens: tokens, ecr: s} + metrics := &recordingECRMetrics{} + s.metric = metrics + return &ecrTest{fake: fake, mux: mux, cache: c, ctx: ctx, tokens: tokens, ecr: s, metrics: metrics} } func (e *ecrTest) do(method, path string, headers ...string) *httptest.ResponseRecorder { @@ -748,3 +769,97 @@ func TestECRTokenMinterFeedsTheTokenManager(t *testing.T) { assert.IsError(t, err, denied) assert.Contains(t, err.Error(), "mint ECR authorization token") } + +func TestECRMetricsRecordEachRequestOutcome(t *testing.T) { + e := setupECRTest(t, func(*fakeECR) {}) + manifest := "/v2/" + testECRName + "/manifests/" + testDigest(e.fake.manifest) + blob := "/v2/" + testECRName + "/blobs/" + testDigest(e.fake.blob) + e.do(http.MethodGet, manifest) + e.do(http.MethodGet, manifest) + e.do(http.MethodHead, manifest) + e.do(http.MethodGet, blob) + e.do(http.MethodHead, blob) + e.do(http.MethodGet, "/v2/"+testECRName+"/manifests/sha256:"+strings.Repeat("0", 64)) + e.do(http.MethodGet, "/v2/"+testECRName+"/tags/list") + e.do(http.MethodGet, "/v2/ecr-public/other/chart/blobs/"+testDigest(e.fake.blob)) + e.do(http.MethodGet, "/v2/") + e.fake.tooMany = true + e.do(http.MethodGet, blob) + + assert.Equal(t, []ecrRequestOutcome{ + {kind: ecrKindManifest, method: http.MethodGet, result: ecrResultMiss}, + {kind: ecrKindManifest, method: http.MethodGet, result: ecrResultHit}, + {kind: ecrKindManifest, method: http.MethodHead, result: ecrResultHit}, + {kind: ecrKindBlob, method: http.MethodGet, result: ecrResultRedirect}, + {kind: ecrKindBlob, method: http.MethodHead, result: ecrResultMetadata}, + {kind: ecrKindManifest, method: http.MethodGet, result: ecrResultNotFound}, + {kind: ecrKindUnknown, method: http.MethodGet, result: ecrResultNotServed}, + {kind: ecrKindBlob, method: http.MethodGet, result: ecrResultNotServed}, + {kind: ecrKindBlob, method: http.MethodGet, result: ecrResultRateLimited}, + }, e.metrics.requests, "the version check is not counted") + assert.Equal(t, []originAuthEvent{originAuthRefresh, originAuthReuse, originAuthReuse, originAuthReuse, originAuthReuse}, e.metrics.auth, + "cache hits and refused requests use no token") +} + +func TestECRMetricsRecordWhyARequestFailed(t *testing.T) { + manifest := func(e *ecrTest) string { return "/v2/" + testECRName + "/manifests/" + testDigest(e.fake.manifest) } + blob := func(e *ecrTest) string { return "/v2/" + testECRName + "/blobs/" + testDigest(e.fake.blob) } + for name, tc := range map[string]struct { + configure func(*ecrTest) + method string + path func(*ecrTest) string + kind string + reason ecrFailureReason + }{ + "manifest bytes that do not match": { + method: http.MethodGet, kind: ecrKindManifest, reason: ecrFailureDigestMismatch, + path: func(*ecrTest) string { return "/v2/" + testECRName + "/manifests/" + testDigest([]byte("other")) }, + }, + "manifest without Content-Length": { + configure: func(e *ecrTest) { e.fake.chunked = true }, + method: http.MethodGet, path: manifest, kind: ecrKindManifest, reason: ecrFailureNoLength, + }, + "partial manifest": { + configure: func(e *ecrTest) { e.fake.partial = true }, + method: http.MethodGet, path: manifest, kind: ecrKindManifest, reason: ecrFailurePartial, + }, + "token rejected after a refresh": { + configure: func(e *ecrTest) { e.fake.acceptedToken = "never" }, + method: http.MethodGet, path: manifest, kind: ecrKindManifest, reason: ecrFailureRejected, + }, + "no token": { + configure: func(e *ecrTest) { e.tokens.err = errors.New("AccessDenied") }, + method: http.MethodGet, path: blob, kind: ecrKindBlob, reason: ecrFailureAuthorization, + }, + "plain http redirect": { + configure: func(e *ecrTest) { e.fake.insecureS3 = true }, + method: http.MethodGet, path: blob, kind: ecrKindBlob, reason: ecrFailureBadRedirect, + }, + "blob HEAD for another digest": { + configure: func(e *ecrTest) { e.fake.headDigest = testDigest([]byte("other")) }, + method: http.MethodHead, path: blob, kind: ecrKindBlob, reason: ecrFailureDigestMismatch, + }, + "blob bytes instead of a redirect": { + configure: func(e *ecrTest) { e.fake.noRedirect = true }, + method: http.MethodGet, path: blob, kind: ecrKindBlob, reason: ecrFailureUpstreamStatus, + }, + "ECR unreachable": { + configure: func(e *ecrTest) { e.fake.server.Close() }, + method: http.MethodGet, path: manifest, kind: ecrKindManifest, reason: ecrFailureTransport, + }, + } { + e := setupECRTest(t, func(*fakeECR) {}) + if tc.configure != nil { + tc.configure(e) + } + e.do(tc.method, tc.path(e)) + assert.Equal(t, []ecrRequestOutcome{{kind: tc.kind, method: tc.method, result: ecrResultFailure, reason: tc.reason}}, e.metrics.requests, name) + } + + e := setupECRTest(t, func(f *fakeECR) { f.acceptedToken = "never" }) + e.do(http.MethodGet, blob(e)) + assert.Equal(t, []originAuthEvent{originAuthRefresh, originAuthForcedRefresh}, e.metrics.auth, "a rejection's refresh is counted as forced") + e.tokens.err = errors.New("AccessDenied") + e.do(http.MethodGet, blob(e)) + assert.Equal(t, originAuthFailure, e.metrics.auth[len(e.metrics.auth)-1]) +} From 633f98fbbcfe21406f52717da025ccf4e061dfc5 Mon Sep 17 00:00:00 2001 From: Ravi Agarwal Date: Sat, 7 Nov 2026 14:40:02 +0000 Subject: [PATCH 2/2] fix(strategy): align ecr failure reasons with codeartifact Label transport failures with the statuses the codeartifact metrics use, through a shared originTransportStatus, and count a client that left first as client_canceled rather than a proxy failure. Cache errors on the manifest GET and HEAD paths both report internal. Record with an uncancelable context, as the codeartifact metrics do. --- README.md | 2 +- internal/strategy/codeartifact.go | 24 ++------------------ internal/strategy/codeartifact_assets.go | 2 +- internal/strategy/codeartifact_stream.go | 2 +- internal/strategy/ecr.go | 19 +++++++++++----- internal/strategy/ecr_metrics.go | 7 ++++-- internal/strategy/ecr_test.go | 10 +++++++- internal/strategy/origin_transport.go | 29 ++++++++++++++++++++++++ 8 files changed, 61 insertions(+), 34 deletions(-) create mode 100644 internal/strategy/origin_transport.go diff --git a/README.md b/README.md index f4a62104..5948335b 100644 --- a/README.md +++ b/README.md @@ -82,7 +82,7 @@ Read-only proxy for one Amazon ECR registry, including its pull-through cache re **Metrics** - `cachew.ecr.requests_total` counts requests by `kind` (`manifest`, `blob`, or `unknown` for a path that names neither), `method` and `result`: `hit` or `miss` for manifests, `redirect` for blob `GET`s, `metadata` for blob `HEAD`s, `not_found`, `rate_limited`, `not_served` for paths and repositories the proxy refuses, or `failure`. The version check is not counted. -- `cachew.ecr.failures_total` counts each failure by `kind` and `reason`: `digest_mismatch`, `no_length`, `partial`, `rejected` (ECR refused the proxy's token), `authorization` (no token could be minted or refreshed), `bad_redirect`, `transport`, `upstream_status`, `cache`, or `internal`. A manifest whose bytes fail verification is a failure even though its `200` was already sent. +- `cachew.ecr.failures_total` counts each failure by `kind` and `reason`: `digest_mismatch`, `no_length`, `partial`, `rejected` (ECR refused the proxy's token), `authorization` (no token could be minted or refreshed), `bad_redirect`, `upstream_status` (any other ECR error status), `internal` (a cache or request error), `client_canceled` (the client left first), or, when the connection to ECR or S3 failed, the same `transport_timeout`, `transport_dns`, `transport_connection` and `transport_error` statuses the CodeArtifact metrics use. A manifest whose bytes fail verification is a failure even though its `200` was already sent. - `cachew.ecr.auth_events_total` counts token `reuse`, `refresh`, `forced_refresh` (after a rejection) and `failure` events. Repository names and digests are never labels. diff --git a/internal/strategy/codeartifact.go b/internal/strategy/codeartifact.go index 32c3b6be..05a90a57 100644 --- a/internal/strategy/codeartifact.go +++ b/internal/strategy/codeartifact.go @@ -3,7 +3,6 @@ package strategy import ( "context" "log/slog" - "net" "net/http" "net/url" "strings" @@ -497,7 +496,7 @@ func (c *CodeArtifact) do(r *http.Request, token string) (*http.Response, error) resp, err := c.client.Do(upstream) if err != nil { cancel(nil) - status := codeArtifactTransportStatus(r.Context(), err) + status := originTransportStatus(r.Context(), err) if c.metric != nil { c.metric.recordOrigin(context.WithoutCancel(r.Context()), codeArtifactOriginObservation{ status: status, format: codeArtifactMetricFormat(target.Path), duration: time.Since(started), @@ -587,7 +586,7 @@ func (c *CodeArtifact) followCrossOriginRedirect( resp, err := c.client.Do(redirected) if err != nil { cancel(nil) - status := codeArtifactTransportStatus(r.Context(), err) + status := originTransportStatus(r.Context(), err) if c.metric != nil { c.metric.recordOrigin(context.WithoutCancel(r.Context()), codeArtifactOriginObservation{ status: status, format: codeArtifactMetricFormat(resolved.Path), duration: time.Since(started), @@ -602,25 +601,6 @@ func (c *CodeArtifact) followCrossOriginRedirect( return resp, nil } -func codeArtifactTransportStatus(ctx context.Context, err error) string { - if errors.Is(ctx.Err(), context.Canceled) { - return "client_canceled" - } - var dnsErr *net.DNSError - if errors.As(err, &dnsErr) { - return "transport_dns" - } - var timeoutErr net.Error - if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &timeoutErr) && timeoutErr.Timeout()) { - return "transport_timeout" - } - var opErr *net.OpError - if errors.As(err, &opErr) { - return "transport_connection" - } - return "transport_error" -} - func (c *CodeArtifact) observeOriginResponse(ctx context.Context, resp *http.Response, path string, started time.Time) { if c.metric == nil { return diff --git a/internal/strategy/codeartifact_assets.go b/internal/strategy/codeartifact_assets.go index 835ce231..87b7d31c 100644 --- a/internal/strategy/codeartifact_assets.go +++ b/internal/strategy/codeartifact_assets.go @@ -40,7 +40,7 @@ func (c *CodeArtifact) serveAsset(w http.ResponseWriter, r *http.Request, mode c started := time.Now() output, err := c.assets.client.GetPackageVersionAsset(ctx, input) if err != nil { - status := codeArtifactTransportStatus(ctx, err) + status := originTransportStatus(ctx, err) var responseError *smithyhttp.ResponseError if errors.As(err, &responseError) { status = strconv.Itoa(responseError.HTTPStatusCode()) diff --git a/internal/strategy/codeartifact_stream.go b/internal/strategy/codeartifact_stream.go index a7afdc8a..761babd6 100644 --- a/internal/strategy/codeartifact_stream.go +++ b/internal/strategy/codeartifact_stream.go @@ -90,6 +90,6 @@ func codeArtifactBodyError(ctx context.Context, err error) string { case errors.Is(err, syscall.EPIPE): return "broken_pipe" default: - return codeArtifactTransportStatus(ctx, err) + return originTransportStatus(ctx, err) } } diff --git a/internal/strategy/ecr.go b/internal/strategy/ecr.go index 37caf901..ca152315 100644 --- a/internal/strategy/ecr.go +++ b/internal/strategy/ecr.go @@ -276,7 +276,7 @@ func newECR(config ECRConfig, tokens originTokenSource, base http.RoundTripper, rw := &ecrResponse{ResponseWriter: w} kind := ecrKindUnknown defer func() { - s.metric.recordRequest(ctx, observation.outcome(kind, r.Method, rw.status, rw.Header().Get(handler.CacheResultHeader))) + s.metric.recordRequest(context.WithoutCancel(ctx), observation.outcome(kind, r.Method, rw.status, rw.Header().Get(handler.CacheResultHeader))) }() // Reject anything but a digest path before the cache is consulted, so an // entry another strategy wrote under a similar key can never be served here. @@ -426,7 +426,7 @@ func (s *ECR) serveManifestHead(w http.ResponseWriter, r *http.Request, ref ecrR w.Header().Set(handler.CacheResultHeader, result) if !httputil.ServeCacheStat(w, headers, err) { w.Header().Del(handler.CacheResultHeader) - ecrObservationFrom(r.Context()).fail(ecrFailureCache) + ecrObservationFrom(r.Context()).fail(ecrFailureInternal) logging.FromContext(r.Context()).ErrorContext(r.Context(), "Failed to stat ECR cache entry", "error", err) http.Error(w, "cache unavailable", http.StatusInternalServerError) } @@ -523,7 +523,7 @@ func (s *ECR) allowed(name string) bool { // errors can quote a raw Location, which may be a presigned S3 URL. Only this // strategy's errors, which name hosts alone, are echoed. func writeECRError(err error, w http.ResponseWriter, r *http.Request) { - ecrObservationFrom(r.Context()).fail(ecrFailureFor(err)) + ecrObservationFrom(r.Context()).fail(ecrFailureFor(r.Context(), err)) for _, logged := range []error{errECRRejected, errECRRefresh} { if errors.Is(err, logged) { // The transport already logged it with its detail. @@ -546,8 +546,11 @@ func writeECRError(err error, w http.ResponseWriter, r *http.Request) { httputil.ErrorResponse(w, r, http.StatusInternalServerError, err.Error()) } -func ecrFailureFor(err error) ecrFailureReason { +func ecrFailureFor(ctx context.Context, err error) ecrFailureReason { switch { + case errors.Is(ctx.Err(), context.Canceled): + // Whatever failed, the client had already gone. + return ecrFailureClientCanceled case errors.Is(err, errECRRejected): return ecrFailureRejected case errors.Is(err, errECRRefresh): @@ -561,7 +564,7 @@ func ecrFailureFor(err error) ecrFailureReason { case errors.Is(err, errTooManyHops), errors.Is(err, errInsecureHop): return ecrFailureBadRedirect case errors.Is(err, errECRTransport): - return ecrFailureTransport + return ecrFailureReason(originTransportStatus(ctx, err)) default: return ecrFailureInternal } @@ -613,7 +616,11 @@ func (s *ECR) request(ctx context.Context, ref ecrRef, method string) (*http.Req token, err := s.tokens.Token(ctx, 0) s.recordAuth(ctx, token, err) if err != nil { - ecrObservationFrom(ctx).fail(ecrFailureAuthorization) + reason := ecrFailureAuthorization + if errors.Is(ctx.Err(), context.Canceled) { + reason = ecrFailureClientCanceled + } + ecrObservationFrom(ctx).fail(reason) // AWS errors name the role and account; log them, answer with the status. logging.FromContext(ctx).ErrorContext(ctx, "Failed to obtain ECR authorization", "error", err) return nil, httputil.Errorf(http.StatusBadGateway, "could not obtain ECR authorization") diff --git a/internal/strategy/ecr_metrics.go b/internal/strategy/ecr_metrics.go index dfbd968d..a5c4648b 100644 --- a/internal/strategy/ecr_metrics.go +++ b/internal/strategy/ecr_metrics.go @@ -35,10 +35,13 @@ const ( ecrFailureRejected ecrFailureReason = "rejected" ecrFailureAuthorization ecrFailureReason = "authorization" ecrFailureBadRedirect ecrFailureReason = "bad_redirect" - ecrFailureTransport ecrFailureReason = "transport" ecrFailureUpstreamStatus ecrFailureReason = "upstream_status" - ecrFailureCache ecrFailureReason = "cache" ecrFailureInternal ecrFailureReason = "internal" + // The client went away first; not a proxy or ECR failure. + ecrFailureClientCanceled ecrFailureReason = "client_canceled" + // Other transport failures are labeled by originTransportStatus, as in the + // codeartifact metrics: transport_timeout, transport_dns, + // transport_connection or transport_error. ) const ( diff --git a/internal/strategy/ecr_test.go b/internal/strategy/ecr_test.go index 2248461a..20708857 100644 --- a/internal/strategy/ecr_test.go +++ b/internal/strategy/ecr_test.go @@ -845,7 +845,15 @@ func TestECRMetricsRecordWhyARequestFailed(t *testing.T) { }, "ECR unreachable": { configure: func(e *ecrTest) { e.fake.server.Close() }, - method: http.MethodGet, path: manifest, kind: ecrKindManifest, reason: ecrFailureTransport, + method: http.MethodGet, path: manifest, kind: ecrKindManifest, reason: "transport_connection", + }, + "client gone before ECR answered": { + configure: func(e *ecrTest) { + ctx, cancel := context.WithCancel(e.ctx) + cancel() + e.ctx = ctx + }, + method: http.MethodGet, path: blob, kind: ecrKindBlob, reason: ecrFailureClientCanceled, }, } { e := setupECRTest(t, func(*fakeECR) {}) diff --git a/internal/strategy/origin_transport.go b/internal/strategy/origin_transport.go new file mode 100644 index 00000000..76d36d9c --- /dev/null +++ b/internal/strategy/origin_transport.go @@ -0,0 +1,29 @@ +package strategy + +import ( + "context" + "net" + + "github.com/alecthomas/errors" +) + +// originTransportStatus names why an origin request failed before response +// headers, from a fixed set suitable for a metric label. +func originTransportStatus(ctx context.Context, err error) string { + if errors.Is(ctx.Err(), context.Canceled) { + return "client_canceled" + } + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) { + return "transport_dns" + } + var timeoutErr net.Error + if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &timeoutErr) && timeoutErr.Timeout()) { + return "transport_timeout" + } + var opErr *net.OpError + if errors.As(err, &opErr) { + return "transport_connection" + } + return "transport_error" +}