feat(strategy): ecr request metrics

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://<host>/<repo>@sha256:<digest>` 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`, `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.
+
 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

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

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())

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)
 	}
 }

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(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.
 		ref, err := parseECRPath(rest)
 		if err != nil {
-			http.Error(w, err.Error()+": only /v2/<name>/manifests/sha256:<digest> and /v2/<name>/blobs/sha256:<digest> are served", http.StatusNotFound)
+			observation.refuse()
+			http.Error(rw, err.Error()+": only /v2/<name>/manifests/sha256:<digest> and /v2/<name>/blobs/sha256:<digest> 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(ecrFailureInternal)
 		logging.FromContext(r.Context()).ErrorContext(r.Context(), "Failed to stat ECR cache entry", "error", err)
 		http.Error(w, "cache unavailable", http.StatusInternalServerError)
 	}
@@ -446,36 +464,39 @@ 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)
 	}
 	return w.ResponseWriter.Write(p) //nolint:wrapcheck // ResponseWriter callers expect its errors unwrapped.
 }
 
 // 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(r.Context(), err))
 	for _, logged := range []error{errECRRejected, errECRRefresh} {
 		if errors.Is(err, logged) {
 			// The transport already logged it with its detail.
@@ -524,6 +546,30 @@ func writeECRError(err error, w http.ResponseWriter, r *http.Request) {
 	httputil.ErrorResponse(w, r, http.StatusInternalServerError, err.Error())
 }
 
+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):
+		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 ecrFailureReason(originTransportStatus(ctx, err))
+	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 +614,13 @@ 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 {
+		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")
@@ -612,6 +664,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 +763,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 +791,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 +849,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 +869,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
 			}