From 4afbd72d3cb27ec28f9ba544f85f67fd65f040d7 Mon Sep 17 00:00:00 2001 From: Joel Robotham Date: Sat, 7 Nov 2026 17:07:31 -0800 Subject: [PATCH 1/3] feat: expose artifact cache and origin metrics Distinguish cache resolution, outbound HTTP exchanges, and cache publication so cache hits cannot hide origin failures. Co-authored-by: Daics (6.1 sol) --- README.md | 22 + internal/cache/instrumentation.go | 95 +++++ .../cache/instrumentation_internal_test.go | 64 +++ internal/cache/tiered.go | 50 ++- internal/metrics/artifact.go | 182 +++++++++ internal/strategy/artifactory.go | 2 +- internal/strategy/github_releases.go | 2 +- internal/strategy/handler/handler.go | 61 ++- internal/strategy/handler/handler_test.go | 2 +- .../strategy/handler/instrumentation_test.go | 382 ++++++++++++++++++ internal/strategy/hermit.go | 8 +- internal/strategy/host.go | 2 +- internal/strategy/proxy.go | 2 +- 13 files changed, 858 insertions(+), 16 deletions(-) create mode 100644 internal/cache/instrumentation.go create mode 100644 internal/cache/instrumentation_internal_test.go create mode 100644 internal/metrics/artifact.go create mode 100644 internal/strategy/handler/instrumentation_test.go diff --git a/README.md b/README.md index bdf2273c..6fefa79b 100644 --- a/README.md +++ b/README.md @@ -353,6 +353,28 @@ Quota gauges describe the installation's shared API budget; do not sum them acro replicas. They do not track Git transport or LFS limits. Polling does not consume the primary API budget, but can count toward GitHub's secondary limits. +### Artifact cache metrics + +Generic artifact handlers (host, Artifactory, Hermit, HTTP proxy and GitHub release downloads) export: + +| OpenTelemetry instrument | Type | Extra labels | +| --- | --- | --- | +| `cachew.cache.resolutions` | Counter | `result`, `source`, `tier` | +| `cachew.origin.requests` | Counter | `status_code`, `error_type` | +| `cachew.origin.request_duration_seconds` | Histogram | `status_code`, `error_type` | +| `cachew.cache.fills` | Counter | `purpose`, `result` | +| `cachew.cache.tier_writes` | Counter | `backend`, `tier`, `purpose`, `result` | + +All carry `strategy` and `registry`. Registry values come from configured host/Artifactory targets or fixed strategy labels; dynamic proxy/Hermit hosts use `other`. URLs, paths, cache keys, credentials and caller namespaces are not labels. Prometheus converts dots to underscores and appends `_total` to counters. + +A resolution is one final lookup decision: `hit`, `miss`, `error`, `conditional`, or `precondition_failed`. Hit sources identify the accepted backend (`disk`, `s3`, `memory`, `remote`, or `other`), after stale and conditional probes. `tier` is zero-based; a miss/error has source `none` and tier `-1`. This measures cache lookup, not complete body delivery. + +Origin metrics count HTTP exchanges in the artifact download client, including redirect hops and transport failures. `status_code=none` means no HTTP response; `error_type` is `none`, `transport`, `timeout`, or `canceled`. Duration ends at response headers or transport failure. Registry identifies the logical artifact provider, including its redirects, rather than each physical destination. These metrics do not count every low-level connection retry, Git transport, Go-module fetch, GitHub metadata lookup, or S3 operation. Internal Hermit forwarding to another Cachew route is excluded from origin/fill counts; the receiving handler measures its own work. + +Fills record a terminal outcome, rather than treating an origin 200 as successful publication. `purpose` separates `origin`, `backfill` and ranged `heal`. Results distinguish successful completion from creation, write, commit, transfer or read errors, cancellation, and superseded heals. Tier writes report backend outcomes separately: a failed S3 write can coexist with a successful disk write. Early close or canceled backfill is not a successful fill. Client responses continue normally when cache writes fail. + +Use resolutions to compare local/shared hits and misses, origin requests by status to measure dependency traffic, and fills/tier writes to find failed publication. An origin error percentage must use origin exchanges as its denominator; client response percentages include cache hits. Idle periods have no percentage. + ## Full Configuration Example ```hcl diff --git a/internal/cache/instrumentation.go b/internal/cache/instrumentation.go new file mode 100644 index 00000000..ebf87902 --- /dev/null +++ b/internal/cache/instrumentation.go @@ -0,0 +1,95 @@ +package cache + +import ( + "context" + "sync" + + "github.com/block/cachew/internal/metrics" +) + +// BackendName is a bounded cache type label, excluding paths, buckets and namespaces. +func BackendName(c Cache) string { + switch c := c.(type) { + case *Disk: + return "disk" + case *S3: + return "s3" + case *Memory: + return "memory" + case *Remote: + return "remote" + case *noOpCache: + return "disabled" + case authoritativeCache: + return BackendName(c.Cache) + default: + return "other" + } +} + +const ( + writeSuccess = "success" + writeCommitError = "commit_error" + writeCanceled = "canceled" +) + +func observeWriter(ctx context.Context, c Cache, tier int, w Writer, err error) Writer { + if !metrics.HasArtifactRequest(ctx) { + return w + } + backend := BackendName(c) + if err != nil { + metrics.RecordCacheWrite(ctx, backend, tier, "create_error") + return w + } + return &observedWriter{Writer: w, ctx: ctx, backend: backend, tier: tier} +} + +type observedWriter struct { + Writer + ctx context.Context + backend string + tier int + result string + once sync.Once +} + +func (w *observedWriter) Write(p []byte) (int, error) { + n, err := w.Writer.Write(p) + if err != nil && w.result == "" { + w.result = "write_error" + } + return n, err //nolint:wrapcheck // Instrumentation preserves backend errors. +} + +func (w *observedWriter) Close() error { + err := w.Writer.Close() + w.record(err) + return err //nolint:wrapcheck // Instrumentation preserves backend errors. +} + +func (w *observedWriter) Abort(cause error) error { + if w.result == "" { + w.result = "aborted" + } + err := w.Writer.Abort(cause) + w.record(err) + return err //nolint:wrapcheck // Instrumentation preserves backend errors. +} + +func (w *observedWriter) record(err error) { + w.once.Do(func() { + result := w.result + if result == "" { + switch { + case w.ctx.Err() != nil: + result = writeCanceled + case err != nil: + result = writeCommitError + default: + result = writeSuccess + } + } + metrics.RecordCacheWrite(w.ctx, w.backend, w.tier, result) + }) +} diff --git a/internal/cache/instrumentation_internal_test.go b/internal/cache/instrumentation_internal_test.go new file mode 100644 index 00000000..7d9c1896 --- /dev/null +++ b/internal/cache/instrumentation_internal_test.go @@ -0,0 +1,64 @@ +package cache + +import ( + "context" + "io" + "os" + "strings" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + "go.opentelemetry.io/otel" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + + "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/metrics" +) + +func TestObservedBackfillEarlyClose(t *testing.T) { + previous := otel.GetMeterProvider() + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + otel.SetMeterProvider(provider) + defer func() { otel.SetMeterProvider(previous); assert.NoError(t, provider.Shutdown(context.Background())) }() + _, ctx := logging.Configure(t.Context(), logging.Config{}) + ctx = metrics.NewArtifactMetrics().WithRequest(ctx, "host", "registry.example", true) + ctx = metrics.WithCachePurpose(ctx, "backfill") + ctx, cancel := context.WithCancel(ctx) + defer cancel() + c, err := NewMemory(ctx, MemoryConfig{MaxTTL: time.Hour}) + assert.NoError(t, err) + defer c.Close() + key := NewKey("artifact") + w, err := c.Create(ctx, key, nil, time.Hour) + assert.NoError(t, err) + w = observeWriter(ctx, c, 0, w, nil) + r := newBackfillReadCloser(ctx, io.NopCloser(strings.NewReader("artifact")), w, cancel) + p := make([]byte, 2) + _, err = r.Read(p) + assert.NoError(t, err) + assert.NoError(t, r.Close()) + _, _, err = c.Open(t.Context(), key) + assert.IsError(t, err, os.ErrNotExist) + var data metricdata.ResourceMetrics + assert.NoError(t, reader.Collect(t.Context(), &data)) + for _, name := range []string{"cachew.cache.fills", "cachew.cache.tier_writes"} { + count := int64(0) + for _, scope := range data.ScopeMetrics { + for _, m := range scope.Metrics { + if m.Name != name { + continue + } + sum := m.Data.(metricdata.Sum[int64]) + for _, point := range sum.DataPoints { + result, _ := point.Attributes.Value("result") + assert.Equal(t, "canceled", result.AsString()) + count += point.Value + } + } + } + assert.Equal(t, int64(1), count) + } +} diff --git a/internal/cache/tiered.go b/internal/cache/tiered.go index e851ceeb..696246b4 100644 --- a/internal/cache/tiered.go +++ b/internal/cache/tiered.go @@ -14,6 +14,7 @@ import ( "github.com/block/cachew/internal/logging" "github.com/block/cachew/internal/metadatadb" + "github.com/block/cachew/internal/metrics" ) // The Tiered cache combines multiple caches. @@ -49,6 +50,17 @@ type authoritativeCache struct { Cache } +func (c authoritativeCache) Open(ctx context.Context, key Key, opts ...Option) (io.ReadCloser, http.Header, error) { + r, headers, err := c.Cache.Open(ctx, key, opts...) + metrics.ObserveCacheSource(ctx, BackendName(c.Cache), 0) + return r, headers, errors.WithStack(err) +} + +func (c authoritativeCache) Create(ctx context.Context, key Key, headers http.Header, ttl time.Duration, opts ...Option) (Writer, error) { + w, err := c.Cache.Create(ctx, key, headers, ttl, opts...) + return observeWriter(ctx, c.Cache, 0, w, err), err +} + func (c authoritativeCache) Invalidate(context.Context, Key) error { return nil } @@ -107,7 +119,7 @@ func (t Tiered) Create(ctx context.Context, key Key, headers http.Header, ttl ti if err != nil { cancel(err) } - tw.writers[i] = w + tw.writers[i] = observeWriter(ctx, cache, i, w, err) }) } done := make(chan struct{}) @@ -285,6 +297,8 @@ func StatAuthoritative(ctx context.Context, c Cache, key Key, opts ...Option) (h // not misreported as missing versions. // // If all caches fail, all errors are returned. +// +//nolint:funlen // Conditional fallback and tier selection share one ordered traversal. func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadCloser, http.Header, error) { ro := NewRequestOptions(opts...) // A Range request yields a partial body, which must never be backfilled @@ -295,6 +309,7 @@ func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadClose // fallback, served only if no deeper tier holds the pinned version. var fallback io.ReadCloser var fallbackHeaders http.Header + fallbackTier := -1 var probeErrs []error rejected := false // a tier failed If-Match; deeper tiers may hold the named version @@ -317,6 +332,7 @@ func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadClose if fallback != nil { discardTieredReader(ctx, key, fallback) } + metrics.ObserveCacheSource(ctx, BackendName(c), i) return nil, headers, errors.WithStack(err) case err != nil: // A hard error is definitive when no earlier tier produced a @@ -338,12 +354,13 @@ func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadClose discardTieredReader(ctx, key, r) continue } - fallback, fallbackHeaders = r, headers + fallback, fallbackHeaders, fallbackTier = r, headers, i continue } if fallback != nil { discardTieredReader(ctx, key, fallback) } + metrics.ObserveCacheSource(ctx, BackendName(c), i) return t.convergeTier0(ctx, key, r, headers, c, i, partial), headers, nil } if len(probeErrs) > 0 { @@ -353,6 +370,7 @@ func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadClose return nil, nil, errors.Join(probeErrs...) } if fallback != nil { + metrics.ObserveCacheSource(ctx, BackendName(t.caches[fallbackTier]), fallbackTier) return fallback, fallbackHeaders, nil } if rejected { @@ -372,18 +390,21 @@ func discardTieredReader(ctx context.Context, key Key, r io.ReadCloser) { // On error or partial read the dst entry is discarded per the Cache contract // (the context is cancelled, causing the writer to discard on Close). func (t Tiered) backfillReader(ctx context.Context, key Key, src io.ReadCloser, headers http.Header, dst Cache) io.ReadCloser { + ctx = metrics.WithCachePurpose(ctx, "backfill") logger := logging.FromContext(ctx) // Use a cancellable context so we can abort the write on failure. // The Cache contract guarantees that cancelled-context writes are discarded. writeCtx, cancel := context.WithCancel(ctx) createOpts := backfillCreateOptions(headers) w, err := dst.Create(writeCtx, key, headers, 0, createOpts...) // 0 → use the cache's max TTL + w = observeWriter(writeCtx, dst, 0, w, err) if err != nil { cancel() + metrics.RecordCacheFill(ctx, "create_error") logger.WarnContext(ctx, "Tier backfill: failed to create writer, skipping", "error", err) return src } - return newBackfillReadCloser(ctx, src, w, cancel) + return newBackfillReadCloser(writeCtx, src, w, cancel) } const tieredETagsMap = "cache-etags" @@ -472,6 +493,7 @@ func (h *tieredHealer) trigger(reqCtx context.Context, id healKey, heal func(con h.mu.Unlock() ctx := logging.ContextWithLogger(h.ctx, logging.FromContext(reqCtx)) + ctx = metrics.ContinueArtifactRequest(ctx, reqCtx) h.wg.Go(func() { defer func() { h.mu.Lock() @@ -513,26 +535,34 @@ func (t Tiered) healTier0(reqCtx context.Context, key Key, source Cache, servedE } func (t Tiered) backfillTier0FromSource(ctx context.Context, key Key, source Cache, wantETag string) { + ctx = metrics.WithCachePurpose(ctx, "heal") logger := logging.FromContext(ctx) + result := writeSuccess + defer func() { metrics.RecordCacheFill(ctx, result) }() ctx, cancel := context.WithTimeout(ctx, tieredHealTimeout) defer cancel() r, headers, err := source.Open(ctx, key, IfMatch(wantETag)) if err != nil { + result = "read_error" logger.WarnContext(ctx, "Tiered: ranged heal source read failed", "key", key, "etag", wantETag, "error", err) return } defer discardTieredReader(ctx, key, r) if headers.Get(ETagKey) != wantETag { + result = "superseded" return } w, err := t.caches[0].Create(ctx, key, headers, 0, backfillCreateOptions(headers)...) // 0 → cache's max TTL + w = observeWriter(ctx, t.caches[0], 0, w, err) if err != nil { + result = "create_error" logger.WarnContext(ctx, "Tiered: ranged heal writer create failed", "key", key, "error", err) return } if _, err := io.Copy(w, r); err != nil { + result = "transfer_error" logger.WarnContext(ctx, "Tiered: ranged heal copy failed", "key", key, "error", errors.Join(err, w.Abort(err))) return } @@ -540,13 +570,17 @@ func (t Tiered) backfillTier0FromSource(ctx context.Context, key Key, source Cac // object that invalidateStale can no longer clean up (it only fires on an // etag mismatch, not a missing entry). Re-check narrows that window. if got, ok := t.etags.Get(key); !ok || got != wantETag { + result = "superseded" if err := w.Abort(errHealSuperseded); err != nil { logger.WarnContext(ctx, "Tiered: ranged heal abort failed", "key", key, "error", err) } return } if err := w.Close(); err != nil { + result = writeCommitError logger.WarnContext(ctx, "Tiered: ranged heal commit failed", "key", key, "error", err) + } else if ctx.Err() != nil { + result = writeCanceled } } @@ -586,6 +620,16 @@ func newBackfillReadCloser(ctx context.Context, src io.ReadCloser, dst io.WriteC } } closeErr := dst.Close() + result := writeSuccess + switch { + case err != nil: + result = "write_error" + case ctx.Err() != nil: + result = writeCanceled + case closeErr != nil: + result = writeCommitError + } + metrics.RecordCacheFill(ctx, result) switch { case err != nil: done <- err diff --git a/internal/metrics/artifact.go b/internal/metrics/artifact.go new file mode 100644 index 00000000..6945f0b5 --- /dev/null +++ b/internal/metrics/artifact.go @@ -0,0 +1,182 @@ +package metrics + +import ( + "context" + "net" + "net/http" + "strconv" + "sync/atomic" + "time" + + "github.com/alecthomas/errors" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// ArtifactMetrics separates cache decisions, origin exchanges, and cache publication. +type ArtifactMetrics struct { + resolutions metric.Int64Counter + origin metric.Int64Counter + duration metric.Float64Histogram + fills metric.Int64Counter + writes metric.Int64Counter +} + +// NewArtifactMetrics uses the same provider as the server's HTTP metrics. +func NewArtifactMetrics() *ArtifactMetrics { + m := otel.Meter("cachew.artifact") + return &ArtifactMetrics{ + resolutions: NewMetric[metric.Int64Counter](m, "cachew.cache.resolutions", "{request}", "Final cache lookup decisions for artifact requests"), + origin: NewMetric[metric.Int64Counter](m, "cachew.origin.requests", "{request}", "Outbound artifact HTTP exchanges, including redirects and transport failures"), + duration: NewHistogram(m, "cachew.origin.request_duration_seconds", "s", "Time to origin response headers or transport failure", FastLatencyBuckets()), + fills: NewMetric[metric.Int64Counter](m, "cachew.cache.fills", "{fill}", "Terminal artifact cache fill and backfill outcomes"), + writes: NewMetric[metric.Int64Counter](m, "cachew.cache.tier_writes", "{write}", "Terminal cache write outcomes by tier"), + } +} + +const noValue = "none" + +type artifactKey struct{} +type purposeKey struct{} + +type cacheSource struct { + backend string + tier int +} + +type artifactRequest struct { + metrics *ArtifactMetrics + attrs []attribute.KeyValue + external bool + source atomic.Pointer[cacheSource] +} + +// WithRequest scopes observations to a configured strategy and registry, never an artifact URL. +func (m *ArtifactMetrics) WithRequest(ctx context.Context, strategy, registry string, external bool) context.Context { + return context.WithValue(ctx, artifactKey{}, &artifactRequest{ + metrics: m, external: external, + attrs: []attribute.KeyValue{attribute.String("strategy", strategy), attribute.String("registry", registry)}, + }) +} + +// ContinueArtifactRequest retains metric attribution without inheriting a request's cancellation. +func ContinueArtifactRequest(ctx, request context.Context) context.Context { + if r, ok := request.Value(artifactKey{}).(*artifactRequest); ok { + return context.WithValue(ctx, artifactKey{}, r) + } + return ctx +} + +// WithCachePurpose distinguishes origin fills from background cache replication. +func WithCachePurpose(ctx context.Context, purpose string) context.Context { + return context.WithValue(ctx, purposeKey{}, purpose) +} + +// ObserveCacheSource reports the accepted tier, after stale and conditional probes are resolved. +func ObserveCacheSource(ctx context.Context, backend string, tier int) { + if r, ok := ctx.Value(artifactKey{}).(*artifactRequest); ok { + r.source.CompareAndSwap(nil, &cacheSource{backend: backend, tier: tier}) + } +} + +// RecordCacheResolution records one decision, independently of subsequent body transfer. +func RecordCacheResolution(ctx context.Context, result, fallback string) { + r, ok := ctx.Value(artifactKey{}).(*artifactRequest) + if !ok { + return + } + source, tier := noValue, -1 + if result == "hit" || result == "conditional" { + source, tier = fallback, 0 + if s := r.source.Load(); s != nil { + source, tier = s.backend, s.tier + } + } + r.metrics.resolutions.Add(ctx, 1, r.options(attribute.String("result", result), attribute.String("source", source), attribute.Int("tier", tier))) +} + +// RecordCacheFill records completion rather than assuming an origin 200 was cached. +func RecordCacheFill(ctx context.Context, result string) { + if r, ok := ctx.Value(artifactKey{}).(*artifactRequest); ok { + r.metrics.fills.Add(ctx, 1, r.options(attribute.String("purpose", cachePurpose(ctx)), attribute.String("result", result))) + } +} + +// RecordCacheWrite records a terminal outcome for one backend writer. +func RecordCacheWrite(ctx context.Context, backend string, tier int, result string) { + if r, ok := ctx.Value(artifactKey{}).(*artifactRequest); ok { + r.metrics.writes.Add(ctx, 1, r.options(attribute.String("backend", backend), attribute.Int("tier", tier), + attribute.String("purpose", cachePurpose(ctx)), attribute.String("result", result))) + } +} + +// HasArtifactRequest avoids allocating observation wrappers for unrelated cache operations. +func HasArtifactRequest(ctx context.Context) bool { + _, ok := ctx.Value(artifactKey{}).(*artifactRequest) + return ok +} + +func cachePurpose(ctx context.Context) string { + if p, ok := ctx.Value(purposeKey{}).(string); ok { + return p + } + return "origin" +} + +func (r *artifactRequest) options(extra ...attribute.KeyValue) metric.MeasurementOption { + attrs := make([]attribute.KeyValue, 0, len(r.attrs)+len(extra)) + attrs = append(attrs, r.attrs...) + return metric.WithAttributes(append(attrs, extra...)...) +} + +// ArtifactClient preserves client settings and measures individual HTTP exchanges without global transport changes. +func ArtifactClient(base *http.Client) *http.Client { + client := *base + transport := client.Transport + if transport == nil { + transport = http.DefaultTransport + } + if _, ok := transport.(*artifactTransport); !ok { + client.Transport = &artifactTransport{base: transport} + } + return &client +} + +type artifactTransport struct { + base http.RoundTripper +} + +func (t *artifactTransport) RoundTrip(req *http.Request) (*http.Response, error) { + r, ok := req.Context().Value(artifactKey{}).(*artifactRequest) + if !ok || !r.external { + return t.base.RoundTrip(req) //nolint:wrapcheck // Preserve the transport contract. + } + start := time.Now() + resp, err := t.base.RoundTrip(req) + status, errorType := noValue, noValue + if resp != nil { + status = strconv.Itoa(resp.StatusCode) + } + switch { + case errors.Is(err, context.Canceled): + errorType = "canceled" + case errors.Is(err, context.DeadlineExceeded): + errorType = "timeout" + case err != nil: + errorType = "transport" + if n, isNetError := errors.AsType[net.Error](err); isNetError && n.Timeout() { + errorType = "timeout" + } + } + attrs := r.options(attribute.String("status_code", status), attribute.String("error_type", errorType)) + r.metrics.origin.Add(req.Context(), 1, attrs) + r.metrics.duration.Record(req.Context(), time.Since(start).Seconds(), attrs) + return resp, err //nolint:wrapcheck // Preserve the transport error for http.Client. +} + +func (t *artifactTransport) CloseIdleConnections() { + if c, ok := t.base.(interface{ CloseIdleConnections() }); ok { + c.CloseIdleConnections() + } +} diff --git a/internal/strategy/artifactory.go b/internal/strategy/artifactory.go index d0ead261..28132463 100644 --- a/internal/strategy/artifactory.go +++ b/internal/strategy/artifactory.go @@ -74,7 +74,7 @@ func NewArtifactory(ctx context.Context, config ArtifactoryConfig, cache cache.C logger: logging.FromContext(ctx), } - hdlr := handler.New(a.client, cache, spools). + hdlr := handler.New(a.client, cache, spools).Metrics("artifactory", u.Hostname()). CacheKey(func(r *http.Request) string { return a.buildTargetURL(r).String() }). diff --git a/internal/strategy/github_releases.go b/internal/strategy/github_releases.go index 4b012771..c5dcc494 100644 --- a/internal/strategy/github_releases.go +++ b/internal/strategy/github_releases.go @@ -65,7 +65,7 @@ func NewGitHubReleases(ctx context.Context, config GitHubReleasesConfig, cache c } // eg. https://github.com/alecthomas/chroma/releases/download/v2.21.1/chroma-2.21.1-darwin-amd64.tar.gz // eg. https://github.com/kubernetes-sigs/kustomize/releases/download/kustomize/v5.8.1/kustomize_v5.8.1_linux_amd64.tar.gz - h := handler.New(s.client, cache, spools). + h := handler.New(s.client, cache, spools).Metrics("github-releases", "github.com"). CacheKey(func(r *http.Request) string { org := r.PathValue("org") repo := r.PathValue("repo") diff --git a/internal/strategy/handler/handler.go b/internal/strategy/handler/handler.go index dd3b53c0..3251619e 100644 --- a/internal/strategy/handler/handler.go +++ b/internal/strategy/handler/handler.go @@ -22,6 +22,7 @@ import ( "github.com/block/cachew/internal/cache" "github.com/block/cachew/internal/httputil" "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/metrics" "github.com/block/cachew/internal/spool" ) @@ -70,6 +71,10 @@ func (p CacheKeyParts) Key() cache.Key { // return modifiedRequest, nil // }) type Handler struct { + metrics *metrics.ArtifactMetrics + strategy string + registry string + external bool client *http.Client scope *httputil.BackoffScope cache cache.Cache @@ -102,14 +107,17 @@ const checkedAtHeader = "X-Cachew-Checked-At" // - No request transformation is performed // - Standard error handling is used. func New(client *http.Client, c cache.Cache, spools *spool.Registry) *Handler { + client = metrics.ArtifactClient(client) scope := httputil.NewBackoffScope(client.Transport) scoped := *client scoped.Transport = scope return &Handler{ - client: &scoped, - scope: scope, - cache: c, - spools: spools.NewGroup(spool.ReleaseCompleted), + client: &scoped, + scope: scope, + cache: c, + spools: spools.NewGroup(spool.ReleaseCompleted), + metrics: metrics.NewArtifactMetrics(), + strategy: "other", registry: "other", external: true, cacheKeyFunc: func(r *http.Request) CacheKeyParts { return NewCacheKeyParts(r.URL.String()) }, @@ -123,6 +131,18 @@ func New(client *http.Client, c cache.Cache, spools *spool.Registry) *Handler { } } +// Metrics labels requests with a strategy and configured registry; dynamic hosts must use a bounded fallback. +func (h *Handler) Metrics(strategy, registry string) *Handler { + h.strategy, h.registry = strategy, registry + return h +} + +// InternalForward excludes delegation to another cache route from external-origin and fill metrics. +func (h *Handler) InternalForward() *Handler { + h.external = false + return h +} + // CacheKey sets the function used to determine the cache key for a request. // The function receives the original incoming request. func (h *Handler) CacheKey(f func(*http.Request) string) *Handler { @@ -214,7 +234,8 @@ func (h *Handler) ParallelFetch(chunkSize int64, concurrency int) *Handler { // HEAD misses are forwarded upstream without caching, so metadata probes // never trigger full downloads. func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - ctx := r.Context() + ctx := h.metrics.WithRequest(r.Context(), h.strategy, h.registry, h.external) + r = r.WithContext(ctx) logger := logging.FromContext(ctx) @@ -322,6 +343,18 @@ func (h *Handler) serveCached(w http.ResponseWriter, r *http.Request, key cache. return h.cacheMiss(w, r, err), nil } cr, headers, err := h.cache.Open(ctx, key, opts...) + result := "error" + switch { + case err == nil: + result = "hit" + case errors.Is(err, os.ErrNotExist): + result = "miss" + case errors.Is(err, cache.ErrPreconditionFailed): + result = "precondition_failed" + case errors.Is(err, cache.ErrNotModified), errors.Is(err, cache.ErrRangeNotSatisfiable): + result = "conditional" + } + metrics.RecordCacheResolution(ctx, result, cache.BackendName(h.cache)) if cacheEntryExists(err) && !h.fresh(r, key, headers) { if cr != nil { return false, errors.Wrap(cr.Close(), "close stale entry") @@ -605,9 +638,14 @@ func (h *Handler) streamNonOKResponse(w http.ResponseWriter, resp *http.Response // discarded and the response is streamed in full. func (h *Handler) streamAndCache(w http.ResponseWriter, r *http.Request, key cache.Key, resp *http.Response) error { maps.Copy(w.Header(), resp.Header) + result := "success" + if h.external { + defer func() { metrics.RecordCacheFill(r.Context(), result) }() + } cw, err := h.cache.Create(r.Context(), key, h.cacheHeaders(resp), h.ttlFunc(r)) if err != nil { + result = "create_error" _, copyErr := io.Copy(w, resp.Body) return errors.Join(errors.Wrap(err, "create cache entry"), errors.Wrap(copyErr, "stream response")) } @@ -640,9 +678,20 @@ func (h *Handler) streamAndCache(w http.ResponseWriter, r *http.Request, key cac _, copyErr := io.Copy(out, pr) closeErr := pr.Close() + cacheErr := <-cacheErrCh + switch { + case tee.err != nil: + result = "write_error" + case r.Context().Err() != nil: + result = "canceled" + case copyErr != nil || closeErr != nil: + result = "transfer_error" + case cacheErr != nil: + result = "commit_error" + } return errors.Join( errors.Wrap(errors.Join(copyErr, closeErr), "stream response"), - errors.Wrap(<-cacheErrCh, "cache response"), + errors.Wrap(cacheErr, "cache response"), ) } diff --git a/internal/strategy/handler/handler_test.go b/internal/strategy/handler/handler_test.go index a9e92d30..e6e4c9f9 100644 --- a/internal/strategy/handler/handler_test.go +++ b/internal/strategy/handler/handler_test.go @@ -413,7 +413,7 @@ func TestHandlerMethodChaining(t *testing.T) { OnError(func(_ error, _ http.ResponseWriter, _ *http.Request) {}). TTL(func(_ *http.Request) time.Duration { return time.Hour }) - assert.Equal(t, h, result, "methods should return the same handler instance") + assert.True(t, h == result, "methods should return the same handler instance") } func TestStreamAndCacheAbortsOnUpstreamError(t *testing.T) { diff --git a/internal/strategy/handler/instrumentation_test.go b/internal/strategy/handler/instrumentation_test.go new file mode 100644 index 00000000..1c4a78e8 --- /dev/null +++ b/internal/strategy/handler/instrumentation_test.go @@ -0,0 +1,382 @@ +package handler_test + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + "github.com/alecthomas/errors" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + + "github.com/block/cachew/internal/cache" + "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/metadatadb" + "github.com/block/cachew/internal/metrics" + "github.com/block/cachew/internal/s3client" + "github.com/block/cachew/internal/s3client/s3clienttest" + "github.com/block/cachew/internal/strategy/handler" +) + +func artifactReader(t *testing.T) *sdkmetric.ManualReader { + t.Helper() + previous := otel.GetMeterProvider() + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + otel.SetMeterProvider(provider) + t.Cleanup(func() { otel.SetMeterProvider(previous); assert.NoError(t, provider.Shutdown(context.Background())) }) + return reader +} + +func artifactCount(t *testing.T, reader *sdkmetric.ManualReader, name string, labels ...string) int64 { + t.Helper() + var data metricdata.ResourceMetrics + assert.NoError(t, reader.Collect(context.Background(), &data)) + var total int64 + for _, scope := range data.ScopeMetrics { + for _, m := range scope.Metrics { + if m.Name != name { + continue + } + sum, ok := m.Data.(metricdata.Sum[int64]) + assert.True(t, ok) + for _, point := range sum.DataPoints { + match := true + for i := 0; i < len(labels); i += 2 { + value, exists := point.Attributes.Value(attribute.Key(labels[i])) + if !exists || value.Emit() != labels[i+1] { + match = false + break + } + } + + if match { + total += point.Value + } + } + } + } + return total +} + +func TestArtifactCacheSourcesAndBackfill(t *testing.T) { + reader := artifactReader(t) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = io.WriteString(w, "artifact") })) + defer origin.Close() + disk, err := cache.NewDisk(artifactContext(t), cache.DiskConfig{Root: t.TempDir()}) + assert.NoError(t, err) + memory := mustNewMemoryCache() + metadata := metadatadb.New(artifactContext(t), metadatadb.NewMemoryBackend()) + c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{disk, memory}, metadata) + t.Cleanup(func() { assert.NoError(t, c.Close()) }) + h := handler.New(origin.Client(), c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { + return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL+r.URL.Path, nil) + }) + serve := func(headers http.Header) *httptest.ResponseRecorder { + r := artifactHTTPRequest(t, "/artifact") + r.Header = headers + w := httptest.NewRecorder() + h.ServeHTTP(w, r) + return w + } + first := serve(http.Header{}) + assert.Equal(t, http.StatusOK, first.Code) + assert.Equal(t, "artifact", first.Body.String()) + second := serve(http.Header{}) + assert.Equal(t, "artifact", second.Body.String()) + assert.NoError(t, disk.Delete(artifactContext(t), cache.NewKey("/artifact"))) + third := serve(http.Header{}) + assert.Equal(t, "artifact", third.Body.String()) + fourth := serve(http.Header{}) + assert.Equal(t, "artifact", fourth.Body.String()) + conditional := serve(http.Header{"If-None-Match": []string{fourth.Header().Get("Etag")}}) + assert.Equal(t, http.StatusNotModified, conditional.Code) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.origin.requests", "status_code", "200")) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "result", "miss")) + assert.Equal(t, int64(2), artifactCount(t, reader, "cachew.cache.resolutions", "source", "disk", "result", "hit")) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "source", "memory", "tier", "1")) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "source", "disk", "result", "conditional")) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "purpose", "origin", "result", "success")) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "purpose", "backfill", "result", "success")) + assert.Equal(t, int64(3), artifactCount(t, reader, "cachew.cache.tier_writes", "result", "success")) +} + +func TestArtifactRedirectsAndThrottling(t *testing.T) { + reader := artifactReader(t) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/redirect" { + http.Redirect(w, r, "/blocked", http.StatusFound) + return + } + w.WriteHeader(http.StatusTooManyRequests) + _, _ = io.WriteString(w, "rate limited") + })) + defer origin.Close() + c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{mustNewMemoryCache()}, nil) + defer c.Close() + h := handler.New(origin.Client(), c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { + return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL+"/redirect", nil) + }) + for range 2 { + w := httptest.NewRecorder() + h.ServeHTTP(w, artifactHTTPRequest(t, "/artifact")) + assert.Equal(t, http.StatusTooManyRequests, w.Code) + assert.Equal(t, "rate limited", w.Body.String()) + } + assert.Equal(t, int64(2), artifactCount(t, reader, "cachew.origin.requests", "status_code", "302")) + assert.Equal(t, int64(2), artifactCount(t, reader, "cachew.origin.requests", "status_code", "429")) + assert.Equal(t, int64(2), artifactCount(t, reader, "cachew.cache.resolutions", "result", "miss")) + assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.fills")) +} + +type artifactTransportFunc func(*http.Request) (*http.Response, error) + +func (f artifactTransportFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +func TestArtifactTransportFailures(t *testing.T) { + for _, tt := range []struct { + name string + err error + label string + }{ + {"connection", errors.New("connection unavailable"), "transport"}, + {"timeout", context.DeadlineExceeded, "timeout"}, + {"canceled", context.Canceled, "canceled"}, + } { + t.Run(tt.name, func(t *testing.T) { + reader := artifactReader(t) + client := &http.Client{Transport: artifactTransportFunc(func(*http.Request) (*http.Response, error) { return nil, tt.err })} + c := mustNewMemoryCache() + defer c.Close() + h := handler.New(client, c, newTestSpools(t)).Metrics("host", "registry.example") + w := httptest.NewRecorder() + h.ServeHTTP(w, artifactHTTPRequest(t, "http://registry.example/a")) + assert.Equal(t, http.StatusBadGateway, w.Code) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.origin.requests", "status_code", "none", "error_type", tt.label)) + assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.fills")) + }) + } +} + +type artifactFailCache struct { + cache.Cache + phase string +} + +func (c artifactFailCache) Create(ctx context.Context, key cache.Key, headers http.Header, ttl time.Duration, opts ...cache.Option) (cache.Writer, error) { + if c.phase == "create" { + return nil, errors.New("create failed") + } + w, err := c.Cache.Create(ctx, key, headers, ttl, opts...) + return &artifactFailWriter{Writer: w, phase: c.phase}, err +} + +type artifactFailWriter struct { + cache.Writer + phase string +} + +func (w *artifactFailWriter) Write(p []byte) (int, error) { + if w.phase == "write" { + return 0, errors.New("write failed") + } + return w.Writer.Write(p) +} +func (w *artifactFailWriter) Close() error { + if w.phase == "commit" { + _ = w.Writer.Abort(errors.New("commit failed")) + return errors.New("commit failed") + } + return w.Writer.Close() +} + +func TestArtifactFillFailuresKeepClientResponse(t *testing.T) { + for _, phase := range []string{"create", "write", "commit"} { + t.Run(phase, func(t *testing.T) { + reader := artifactReader(t) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, strings.Repeat("artifact", 10000)) + })) + defer origin.Close() + c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{artifactFailCache{Cache: mustNewMemoryCache(), phase: phase}}, nil) + defer c.Close() + h := handler.New(origin.Client(), c, newTestSpools(t)).Transform(func(r *http.Request) (*http.Request, error) { + return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL, nil) + }) + w := httptest.NewRecorder() + h.ServeHTTP(w, artifactHTTPRequest(t, "/artifact")) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, strings.Repeat("artifact", 10000), w.Body.String()) + _, _, err := c.Open(artifactContext(t), cache.NewKey("/artifact")) + assert.IsError(t, err, os.ErrNotExist) + result := phase + "_error" + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "result", result)) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.tier_writes", "result", result)) + assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.fills", "result", "success")) + }) + } +} + +func TestArtifactInterruptedBodyIsNotSuccessfulFill(t *testing.T) { + reader := artifactReader(t) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", "10000") + _, _ = io.WriteString(w, "partial") + })) + defer origin.Close() + c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{mustNewMemoryCache()}, nil) + defer c.Close() + h := handler.New(origin.Client(), c, newTestSpools(t)).Transform(func(r *http.Request) (*http.Request, error) { + return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL, nil) + }) + w := httptest.NewRecorder() + h.ServeHTTP(w, artifactHTTPRequest(t, "/artifact")) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.origin.requests", "status_code", "200")) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "result", "transfer_error")) + assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.tier_writes", "result", "success")) +} + +func TestArtifactInternalForwardAndCardinality(t *testing.T) { + reader := artifactReader(t) + client := &http.Client{Transport: artifactTransportFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: io.NopCloser(strings.NewReader("artifact"))}, nil + })} + c := cache.NoOpCache() + h := handler.New(client, c, newTestSpools(t)).Metrics("proxy", "other").InternalForward() + var wg sync.WaitGroup + for i := range 50 { + wg.Go(func() { + w := httptest.NewRecorder() + h.ServeHTTP(w, artifactHTTPRequest(t, "http://"+strings.Repeat("a", i+1)+".example/secret?token=value")) + assert.Equal(t, http.StatusOK, w.Code) + }) + } + wg.Wait() + assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.origin.requests")) + assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.fills")) + assert.Equal(t, int64(50), artifactCount(t, reader, "cachew.cache.resolutions", "registry", "other")) + var data metricdata.ResourceMetrics + assert.NoError(t, reader.Collect(artifactContext(t), &data)) + for _, scope := range data.ScopeMetrics { + for _, m := range scope.Metrics { + if sum, ok := m.Data.(metricdata.Sum[int64]); ok { + assert.Equal(t, 1, len(sum.DataPoints)) + for _, a := range sum.DataPoints[0].Attributes.ToSlice() { + assert.False(t, strings.Contains(a.Value.Emit(), ".example")) + assert.False(t, strings.Contains(a.Value.Emit(), "secret")) + } + } + } + } +} + +func TestArtifactBackfillFailureRemainsVisible(t *testing.T) { + reader := artifactReader(t) + local := artifactFailCache{Cache: mustNewMemoryCache(), phase: "write"} + remote := mustNewMemoryCache() + key := cache.NewKey("/artifact") + assert.NoError(t, cache.WriteFunc(artifactContext(t), remote, key, nil, time.Hour, func(w io.Writer) error { _, err := io.WriteString(w, "artifact"); return err })) + c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{local, remote}, metadatadb.New(artifactContext(t), metadatadb.NewMemoryBackend())) + defer c.Close() + h := handler.New(http.DefaultClient, c, newTestSpools(t)) + w := httptest.NewRecorder() + h.ServeHTTP(w, artifactHTTPRequest(t, "/artifact")) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "artifact", w.Body.String()) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "purpose", "backfill", "result", "write_error")) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.tier_writes", "purpose", "backfill", "result", "write_error")) + assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.origin.requests")) +} + +func TestArtifactStaleTierIsNotCountedAsHit(t *testing.T) { + reader := artifactReader(t) + disk, err := cache.NewDisk(artifactContext(t), cache.DiskConfig{Root: t.TempDir()}) + assert.NoError(t, err) + memory := mustNewMemoryCache() + key := cache.NewKey("/artifact") + for _, seed := range []struct { + c cache.Cache + etag string + body string + }{{disk, "old", "old"}, {memory, "new", "new"}} { + assert.NoError(t, cache.WriteFunc(artifactContext(t), seed.c, key, nil, time.Hour, func(w io.Writer) error { _, err := io.WriteString(w, seed.body); return err }, cache.WithETag(seed.etag))) + } + c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{disk, memory}, metadatadb.New(artifactContext(t), metadatadb.NewMemoryBackend())) + defer c.Close() + h := handler.New(http.DefaultClient, c, newTestSpools(t)) + r := artifactHTTPRequest(t, "/artifact") + r.Header.Set("If-Match", `"new"`) + w := httptest.NewRecorder() + h.ServeHTTP(w, r) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "new", w.Body.String()) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "source", "memory", "tier", "1", "result", "hit")) + assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.resolutions", "source", "disk")) +} + +func TestArtifactS3(t *testing.T) { + reader := artifactReader(t) + bucket := s3clienttest.Start(t) + provider := s3client.NewClientProvider(artifactContext(t), s3client.Config{Endpoint: s3clienttest.Addr, UseSSL: false}) + shared, err := cache.NewS3(artifactContext(t), cache.S3Config{Bucket: bucket, MaxTTL: time.Hour, UploadPartSizeMB: 16}, provider) + assert.NoError(t, err) + disk, err := cache.NewDisk(artifactContext(t), cache.DiskConfig{Root: t.TempDir()}) + assert.NoError(t, err) + key := cache.NewKey("/artifact") + assert.NoError(t, cache.WriteFunc(artifactContext(t), shared, key, nil, time.Hour, func(w io.Writer) error { _, err := io.WriteString(w, "artifact"); return err })) + c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{disk, shared}, metadatadb.New(artifactContext(t), metadatadb.NewMemoryBackend())) + defer c.Close() + h := handler.New(http.DefaultClient, c, newTestSpools(t)).Metrics("host", "registry.example") + w := httptest.NewRecorder() + h.ServeHTTP(w, artifactHTTPRequest(t, "/artifact")) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "artifact", w.Body.String()) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "source", "s3", "tier", "1", "result", "hit")) + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "purpose", "backfill", "result", "success")) + assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.origin.requests")) +} + +func artifactContext(t *testing.T) context.Context { + t.Helper() + _, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelError}) + return ctx +} + +func artifactHTTPRequest(t *testing.T, target string) *http.Request { + t.Helper() + r := httptest.NewRequest(http.MethodGet, target, nil).WithContext(artifactContext(t)) + r.RequestURI = "" + return r +} + +func TestArtifactMetricsPrometheusExport(t *testing.T) { + previous := otel.GetMeterProvider() + exporter, err := metrics.New(artifactContext(t), metrics.Config{ServiceName: "cachew", EnablePrometheus: true}) + assert.NoError(t, err) + defer func() { otel.SetMeterProvider(previous); assert.NoError(t, exporter.Close()) }() + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = io.WriteString(w, "artifact") })) + defer origin.Close() + c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{mustNewMemoryCache()}, nil) + defer c.Close() + h := handler.New(origin.Client(), c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { + return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL, nil) + }) + h.ServeHTTP(httptest.NewRecorder(), artifactHTTPRequest(t, "/artifact")) + w := httptest.NewRecorder() + exporter.Handler().ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + text := w.Body.String() + for _, name := range []string{"cachew_cache_resolutions_total{", "cachew_origin_requests_total{", "cachew_origin_request_duration_seconds_count{", "cachew_cache_fills_total{", "cachew_cache_tier_writes_total{"} { + assert.True(t, strings.Contains(text, name), "missing Prometheus instrument %s", name) + } + assert.True(t, strings.Contains(text, `registry="registry.example"`)) +} diff --git a/internal/strategy/hermit.go b/internal/strategy/hermit.go index 186b08f8..bcc524af 100644 --- a/internal/strategy/hermit.go +++ b/internal/strategy/hermit.go @@ -86,7 +86,7 @@ func NewHermit(ctx context.Context, config HermitConfig, _ jobscheduler.Schedule func (s *Hermit) String() string { return "hermit" } func (s *Hermit) createDirectHandler(c cache.Cache, spools *spool.Registry) http.Handler { - return handler.New(s.client, c, spools). + return handler.New(s.client, c, spools).Metrics("hermit", "other"). CacheKey(func(r *http.Request) string { return s.buildOriginalURL(r) }). @@ -109,7 +109,11 @@ func (s *Hermit) createRedirectHandler(isInternalRedirect bool, c cache.Cache, s cacheBackend = c } - return handler.New(s.client, cacheBackend, spools). + h := handler.New(s.client, cacheBackend, spools).Metrics("hermit", "github.com") + if isInternalRedirect { + h.InternalForward() + } + return h. CacheKey(func(r *http.Request) string { return s.buildGitHubURL(r) }). diff --git a/internal/strategy/host.go b/internal/strategy/host.go index 60010f75..138c2dac 100644 --- a/internal/strategy/host.go +++ b/internal/strategy/host.go @@ -89,7 +89,7 @@ func NewHost(_ context.Context, config HostConfig, cache cache.Cache, mux Mux, s h.client.Transport = transport } - hdlr := handler.New(h.client, cache, spools). + hdlr := handler.New(h.client, cache, spools).Metrics("host", u.Hostname()). CacheKey(func(r *http.Request) string { return h.buildTargetURL(r).String() }). diff --git a/internal/strategy/proxy.go b/internal/strategy/proxy.go index 642f313f..8a16355b 100644 --- a/internal/strategy/proxy.go +++ b/internal/strategy/proxy.go @@ -59,7 +59,7 @@ func NewHTTPProxy(ctx context.Context, c cache.Cache, _ Mux, spools *spool.Regis client := &http.Client{} p := &HTTPProxy{logger: logger} - p.handler = handler.New(client, c, spools). + p.handler = handler.New(client, c, spools).Metrics("proxy", "other"). CacheKey(func(r *http.Request) string { target := p.parseProxyURI(r) if target == nil { From d46a758865ceeff922f4aa28009ff2b3862cbba1 Mon Sep 17 00:00:00 2001 From: Elizabeth Worstell Date: Mon, 9 Nov 2026 14:20:07 -0800 Subject: [PATCH 2/3] fix: align artifact metrics with coalescing, cooldowns and cache result Origin metrics now wrap the transport below the 429 cooldowns, so a request that a cooldown rejects is not counted as origin traffic. Coalesced fills record their outcome in cachew.cache.fills. The fill writer and the tier write wrapper keep pending reads, so coalesced readers still follow the cache write. The cachew.cache.resolutions counter is removed: cachew.cache.result already gives hit and miss counts. Hits now also get the cachew.cache.source label on the same HTTP server metrics. OCI blob and manifest handlers get metric labels. Co-authored-by: Chain <492cad42391b4a5c84d831885f468e6f1699c0a8c39241f1c002235ece37ee69@buzz> Co-Authored-By: Claude Opus 5.5 --- README.md | 9 +- cmd/cachewd/main.go | 5 +- internal/cache/instrumentation.go | 14 ++ .../cache/instrumentation_internal_test.go | 26 ++++ internal/cache/tiered.go | 14 +- internal/metrics/artifact.go | 76 ++++------ internal/strategy/handler/handler.go | 138 ++++++++++++------ .../strategy/handler/instrumentation_test.go | 97 +++++++----- internal/strategy/oci.go | 4 +- 9 files changed, 240 insertions(+), 143 deletions(-) diff --git a/README.md b/README.md index 6fefa79b..e41653aa 100644 --- a/README.md +++ b/README.md @@ -355,11 +355,10 @@ the primary API budget, but can count toward GitHub's secondary limits. ### Artifact cache metrics -Generic artifact handlers (host, Artifactory, Hermit, HTTP proxy and GitHub release downloads) export: +Generic artifact handlers (host, Artifactory, Hermit, HTTP proxy, OCI blobs and manifests, and GitHub release downloads) export: | OpenTelemetry instrument | Type | Extra labels | | --- | --- | --- | -| `cachew.cache.resolutions` | Counter | `result`, `source`, `tier` | | `cachew.origin.requests` | Counter | `status_code`, `error_type` | | `cachew.origin.request_duration_seconds` | Histogram | `status_code`, `error_type` | | `cachew.cache.fills` | Counter | `purpose`, `result` | @@ -367,13 +366,13 @@ Generic artifact handlers (host, Artifactory, Hermit, HTTP proxy and GitHub rele All carry `strategy` and `registry`. Registry values come from configured host/Artifactory targets or fixed strategy labels; dynamic proxy/Hermit hosts use `other`. URLs, paths, cache keys, credentials and caller namespaces are not labels. Prometheus converts dots to underscores and appends `_total` to counters. -A resolution is one final lookup decision: `hit`, `miss`, `error`, `conditional`, or `precondition_failed`. Hit sources identify the accepted backend (`disk`, `s3`, `memory`, `remote`, or `other`), after stale and conditional probes. `tier` is zero-based; a miss/error has source `none` and tier `-1`. This measures cache lookup, not complete body delivery. +The HTTP server metrics of these handlers carry `cachew.cache.result` (`hit`, `miss`, `coalesced` or `error`). Hits also carry `cachew.cache.source`, the backend that served them (`disk`, `s3`, `memory`, `remote`, or `other`), after stale and conditional probes. -Origin metrics count HTTP exchanges in the artifact download client, including redirect hops and transport failures. `status_code=none` means no HTTP response; `error_type` is `none`, `transport`, `timeout`, or `canceled`. Duration ends at response headers or transport failure. Registry identifies the logical artifact provider, including its redirects, rather than each physical destination. These metrics do not count every low-level connection retry, Git transport, Go-module fetch, GitHub metadata lookup, or S3 operation. Internal Hermit forwarding to another Cachew route is excluded from origin/fill counts; the receiving handler measures its own work. +Origin metrics count HTTP exchanges that reach an origin through `http.DefaultTransport`, including redirect hops and transport failures. Requests rejected during an upstream 429 cooldown are not counted. `status_code=none` means no HTTP response; `error_type` is `none`, `transport`, `timeout`, or `canceled`. Duration ends at response headers or transport failure. Registry identifies the logical artifact provider, including its redirects, rather than each physical destination. These metrics do not count every low-level connection retry, Git transport, Go-module fetch, GitHub metadata lookup, OCI tag-list or referrer pass-through, or S3 operation. Internal Hermit forwarding to another Cachew route is excluded from origin/fill counts; the receiving handler measures its own work. Fills record a terminal outcome, rather than treating an origin 200 as successful publication. `purpose` separates `origin`, `backfill` and ranged `heal`. Results distinguish successful completion from creation, write, commit, transfer or read errors, cancellation, and superseded heals. Tier writes report backend outcomes separately: a failed S3 write can coexist with a successful disk write. Early close or canceled backfill is not a successful fill. Client responses continue normally when cache writes fail. -Use resolutions to compare local/shared hits and misses, origin requests by status to measure dependency traffic, and fills/tier writes to find failed publication. An origin error percentage must use origin exchanges as its denominator; client response percentages include cache hits. Idle periods have no percentage. +Use `cachew.cache.result` and `cachew.cache.source` to compare local/shared hits and misses, origin requests by status to measure dependency traffic, and fills/tier writes to find failed publication. An origin error percentage must use origin exchanges as its denominator; client response percentages include cache hits. Idle periods have no percentage. ## Full Configuration Example diff --git a/cmd/cachewd/main.go b/cmd/cachewd/main.go index 2e8b23e9..a2b2597d 100644 --- a/cmd/cachewd/main.go +++ b/cmd/cachewd/main.go @@ -457,8 +457,9 @@ func configureProcess(mutexProfileFraction int, upstreamBackoff time.Duration) { runtime.SetMutexProfileFraction(mutexProfileFraction) // Every strategy that uses http.DefaultTransport shares one set of - // per-origin 429 cooldowns. - http.DefaultTransport = httputil.NewBackoffTransport(http.DefaultTransport, upstreamBackoff) //nolint:reassign + // per-origin 429 cooldowns. Origin metrics sit below the cooldowns so that + // requests rejected during a cooldown are not counted as origin traffic. + http.DefaultTransport = httputil.NewBackoffTransport(metrics.NewOriginTransport(http.DefaultTransport), upstreamBackoff) //nolint:reassign } // loadGlobalConfig unmarshals the global config from HCL, using a two-pass diff --git a/internal/cache/instrumentation.go b/internal/cache/instrumentation.go index ebf87902..4d832f1d 100644 --- a/internal/cache/instrumentation.go +++ b/internal/cache/instrumentation.go @@ -4,6 +4,8 @@ import ( "context" "sync" + "github.com/alecthomas/errors" + "github.com/block/cachew/internal/metrics" ) @@ -54,6 +56,18 @@ type observedWriter struct { once sync.Once } +var _ PendingReader = (*observedWriter)(nil) + +// OpenPending keeps pending reads available through the wrapper, so coalesced +// readers can follow the write instead of spooling a second copy. +func (w *observedWriter) OpenPending() (ReadAtCloser, error) { + pending, ok := w.Writer.(PendingReader) + if !ok { + return nil, errors.WithStack(ErrPendingReadUnsupported) + } + return errors.WithStack2(pending.OpenPending()) +} + func (w *observedWriter) Write(p []byte) (int, error) { n, err := w.Writer.Write(p) if err != nil && w.result == "" { diff --git a/internal/cache/instrumentation_internal_test.go b/internal/cache/instrumentation_internal_test.go index 7d9c1896..a635ea75 100644 --- a/internal/cache/instrumentation_internal_test.go +++ b/internal/cache/instrumentation_internal_test.go @@ -9,11 +9,13 @@ import ( "time" "github.com/alecthomas/assert/v2" + "github.com/alecthomas/errors" "go.opentelemetry.io/otel" sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/metadatadb" "github.com/block/cachew/internal/metrics" ) @@ -62,3 +64,27 @@ func TestObservedBackfillEarlyClose(t *testing.T) { assert.Equal(t, int64(1), count) } } + +func TestObservedWriterKeepsPendingReads(t *testing.T) { + _, ctx := logging.Configure(t.Context(), logging.Config{}) + ctx = metrics.NewArtifactMetrics().WithRequest(ctx, "host", "registry.example", true) + disk, err := NewDisk(ctx, DiskConfig{Root: t.TempDir()}) + assert.NoError(t, err) + memory, err := NewMemory(ctx, MemoryConfig{MaxTTL: time.Hour}) + assert.NoError(t, err) + c := MaybeNewTiered(ctx, []Cache{disk, memory}, metadatadb.New(ctx, metadatadb.NewMemoryBackend())) + defer c.Close() + w, err := c.Create(ctx, NewKey("artifact"), nil, time.Hour) + assert.NoError(t, err) + pending, ok := w.(PendingReader) + assert.True(t, ok) + r, err := pending.OpenPending() + assert.NoError(t, err) + _, err = io.WriteString(w, "artifact") + assert.NoError(t, err) + p := make([]byte, len("artifact")) + _, err = r.ReadAt(p, 0) + assert.NoError(t, err) + assert.Equal(t, "artifact", string(p)) + assert.NoError(t, errors.Join(w.Close(), r.Close())) +} diff --git a/internal/cache/tiered.go b/internal/cache/tiered.go index 696246b4..52e6e37a 100644 --- a/internal/cache/tiered.go +++ b/internal/cache/tiered.go @@ -52,7 +52,9 @@ type authoritativeCache struct { func (c authoritativeCache) Open(ctx context.Context, key Key, opts ...Option) (io.ReadCloser, http.Header, error) { r, headers, err := c.Cache.Open(ctx, key, opts...) - metrics.ObserveCacheSource(ctx, BackendName(c.Cache), 0) + if err == nil || errors.Is(err, ErrNotModified) || errors.Is(err, ErrRangeNotSatisfiable) { + metrics.ObserveCacheSource(ctx, BackendName(c.Cache)) + } return r, headers, errors.WithStack(err) } @@ -309,7 +311,7 @@ func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadClose // fallback, served only if no deeper tier holds the pinned version. var fallback io.ReadCloser var fallbackHeaders http.Header - fallbackTier := -1 + var fallbackSource Cache var probeErrs []error rejected := false // a tier failed If-Match; deeper tiers may hold the named version @@ -332,7 +334,7 @@ func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadClose if fallback != nil { discardTieredReader(ctx, key, fallback) } - metrics.ObserveCacheSource(ctx, BackendName(c), i) + metrics.ObserveCacheSource(ctx, BackendName(c)) return nil, headers, errors.WithStack(err) case err != nil: // A hard error is definitive when no earlier tier produced a @@ -354,13 +356,13 @@ func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadClose discardTieredReader(ctx, key, r) continue } - fallback, fallbackHeaders, fallbackTier = r, headers, i + fallback, fallbackHeaders, fallbackSource = r, headers, c continue } if fallback != nil { discardTieredReader(ctx, key, fallback) } - metrics.ObserveCacheSource(ctx, BackendName(c), i) + metrics.ObserveCacheSource(ctx, BackendName(c)) return t.convergeTier0(ctx, key, r, headers, c, i, partial), headers, nil } if len(probeErrs) > 0 { @@ -370,7 +372,7 @@ func (t Tiered) Open(ctx context.Context, key Key, opts ...Option) (io.ReadClose return nil, nil, errors.Join(probeErrs...) } if fallback != nil { - metrics.ObserveCacheSource(ctx, BackendName(t.caches[fallbackTier]), fallbackTier) + metrics.ObserveCacheSource(ctx, BackendName(fallbackSource)) return fallback, fallbackHeaders, nil } if rejected { diff --git a/internal/metrics/artifact.go b/internal/metrics/artifact.go index 6945f0b5..a816f47f 100644 --- a/internal/metrics/artifact.go +++ b/internal/metrics/artifact.go @@ -14,24 +14,22 @@ import ( "go.opentelemetry.io/otel/metric" ) -// ArtifactMetrics separates cache decisions, origin exchanges, and cache publication. +// ArtifactMetrics separates origin exchanges from cache publication. type ArtifactMetrics struct { - resolutions metric.Int64Counter - origin metric.Int64Counter - duration metric.Float64Histogram - fills metric.Int64Counter - writes metric.Int64Counter + origin metric.Int64Counter + duration metric.Float64Histogram + fills metric.Int64Counter + writes metric.Int64Counter } // NewArtifactMetrics uses the same provider as the server's HTTP metrics. func NewArtifactMetrics() *ArtifactMetrics { m := otel.Meter("cachew.artifact") return &ArtifactMetrics{ - resolutions: NewMetric[metric.Int64Counter](m, "cachew.cache.resolutions", "{request}", "Final cache lookup decisions for artifact requests"), - origin: NewMetric[metric.Int64Counter](m, "cachew.origin.requests", "{request}", "Outbound artifact HTTP exchanges, including redirects and transport failures"), - duration: NewHistogram(m, "cachew.origin.request_duration_seconds", "s", "Time to origin response headers or transport failure", FastLatencyBuckets()), - fills: NewMetric[metric.Int64Counter](m, "cachew.cache.fills", "{fill}", "Terminal artifact cache fill and backfill outcomes"), - writes: NewMetric[metric.Int64Counter](m, "cachew.cache.tier_writes", "{write}", "Terminal cache write outcomes by tier"), + origin: NewMetric[metric.Int64Counter](m, "cachew.origin.requests", "{request}", "Outbound artifact HTTP exchanges, including redirects and transport failures"), + duration: NewHistogram(m, "cachew.origin.request_duration_seconds", "s", "Time to origin response headers or transport failure", FastLatencyBuckets()), + fills: NewMetric[metric.Int64Counter](m, "cachew.cache.fills", "{fill}", "Terminal artifact cache fill and backfill outcomes"), + writes: NewMetric[metric.Int64Counter](m, "cachew.cache.tier_writes", "{write}", "Terminal cache write outcomes by tier"), } } @@ -40,16 +38,11 @@ const noValue = "none" type artifactKey struct{} type purposeKey struct{} -type cacheSource struct { - backend string - tier int -} - type artifactRequest struct { metrics *ArtifactMetrics attrs []attribute.KeyValue external bool - source atomic.Pointer[cacheSource] + source atomic.Pointer[string] } // WithRequest scopes observations to a configured strategy and registry, never an artifact URL. @@ -73,27 +66,21 @@ func WithCachePurpose(ctx context.Context, purpose string) context.Context { return context.WithValue(ctx, purposeKey{}, purpose) } -// ObserveCacheSource reports the accepted tier, after stale and conditional probes are resolved. -func ObserveCacheSource(ctx context.Context, backend string, tier int) { +// ObserveCacheSource reports the backend that answered, after stale and conditional probes are resolved. +func ObserveCacheSource(ctx context.Context, backend string) { if r, ok := ctx.Value(artifactKey{}).(*artifactRequest); ok { - r.source.CompareAndSwap(nil, &cacheSource{backend: backend, tier: tier}) + r.source.CompareAndSwap(nil, &backend) } } -// RecordCacheResolution records one decision, independently of subsequent body transfer. -func RecordCacheResolution(ctx context.Context, result, fallback string) { - r, ok := ctx.Value(artifactKey{}).(*artifactRequest) - if !ok { - return - } - source, tier := noValue, -1 - if result == "hit" || result == "conditional" { - source, tier = fallback, 0 +// CacheSource returns the backend reported by [ObserveCacheSource], if any. +func CacheSource(ctx context.Context) (string, bool) { + if r, ok := ctx.Value(artifactKey{}).(*artifactRequest); ok { if s := r.source.Load(); s != nil { - source, tier = s.backend, s.tier + return *s, true } } - r.metrics.resolutions.Add(ctx, 1, r.options(attribute.String("result", result), attribute.String("source", source), attribute.Int("tier", tier))) + return "", false } // RecordCacheFill records completion rather than assuming an origin 200 was cached. @@ -130,24 +117,20 @@ func (r *artifactRequest) options(extra ...attribute.KeyValue) metric.Measuremen return metric.WithAttributes(append(attrs, extra...)...) } -// ArtifactClient preserves client settings and measures individual HTTP exchanges without global transport changes. -func ArtifactClient(base *http.Client) *http.Client { - client := *base - transport := client.Transport - if transport == nil { - transport = http.DefaultTransport - } - if _, ok := transport.(*artifactTransport); !ok { - client.Transport = &artifactTransport{base: transport} - } - return &client +// OriginTransport measures the exchanges of artifact requests that reach base. +// It must sit below any transport that answers without contacting the origin, +// such as upstream cooldowns, so that those answers are not counted. +type OriginTransport struct { + base http.RoundTripper } -type artifactTransport struct { - base http.RoundTripper +// NewOriginTransport measures exchanges sent to base. +func NewOriginTransport(base http.RoundTripper) *OriginTransport { + return &OriginTransport{base: base} } -func (t *artifactTransport) RoundTrip(req *http.Request) (*http.Response, error) { +// RoundTrip implements http.RoundTripper. +func (t *OriginTransport) RoundTrip(req *http.Request) (*http.Response, error) { r, ok := req.Context().Value(artifactKey{}).(*artifactRequest) if !ok || !r.external { return t.base.RoundTrip(req) //nolint:wrapcheck // Preserve the transport contract. @@ -175,7 +158,8 @@ func (t *artifactTransport) RoundTrip(req *http.Request) (*http.Response, error) return resp, err //nolint:wrapcheck // Preserve the transport error for http.Client. } -func (t *artifactTransport) CloseIdleConnections() { +// CloseIdleConnections forwards to base, so http.Client.CloseIdleConnections still works. +func (t *OriginTransport) CloseIdleConnections() { if c, ok := t.base.(interface{ CloseIdleConnections() }); ok { c.CloseIdleConnections() } diff --git a/internal/strategy/handler/handler.go b/internal/strategy/handler/handler.go index 3251619e..ea18bd52 100644 --- a/internal/strategy/handler/handler.go +++ b/internal/strategy/handler/handler.go @@ -107,7 +107,6 @@ const checkedAtHeader = "X-Cachew-Checked-At" // - No request transformation is performed // - Standard error handling is used. func New(client *http.Client, c cache.Cache, spools *spool.Registry) *Handler { - client = metrics.ArtifactClient(client) scope := httputil.NewBackoffScope(client.Transport) scoped := *client scoped.Transport = scope @@ -312,11 +311,24 @@ const ( CacheError = "error" ) +// CacheSourceAttribute labels a cache hit with the backend that served it, +// such as disk, s3 or remote, so local and shared hits can be told apart. +const CacheSourceAttribute = "cachew.cache.source" + func labelCacheResult(ctx context.Context, result string) { labeler, _ := otelhttp.LabelerFromContext(ctx) labeler.Add(attribute.String(CacheResultAttribute, result)) } +func (h *Handler) labelCacheHit(ctx context.Context) { + source, ok := metrics.CacheSource(ctx) + if !ok { + source = cache.BackendName(h.cache) + } + labeler, _ := otelhttp.LabelerFromContext(ctx) + labeler.Add(attribute.String(CacheResultAttribute, CacheHit), attribute.String(CacheSourceAttribute, source)) +} + // coalescable reports whether a miss can share the plain response for its key. // Range and conditional requests need a different response. func coalescable(r *http.Request) bool { @@ -337,24 +349,12 @@ func (h *Handler) serveCached(w http.ResponseWriter, r *http.Request, key cache. return false, nil } if httputil.ServeCacheStat(w, clientHeaders(headers), err) { - labelCacheResult(ctx, CacheHit) + h.labelCacheHit(ctx) return true, nil } return h.cacheMiss(w, r, err), nil } cr, headers, err := h.cache.Open(ctx, key, opts...) - result := "error" - switch { - case err == nil: - result = "hit" - case errors.Is(err, os.ErrNotExist): - result = "miss" - case errors.Is(err, cache.ErrPreconditionFailed): - result = "precondition_failed" - case errors.Is(err, cache.ErrNotModified), errors.Is(err, cache.ErrRangeNotSatisfiable): - result = "conditional" - } - metrics.RecordCacheResolution(ctx, result, cache.BackendName(h.cache)) if cacheEntryExists(err) && !h.fresh(r, key, headers) { if cr != nil { return false, errors.Wrap(cr.Close(), "close stale entry") @@ -362,7 +362,7 @@ func (h *Handler) serveCached(w http.ResponseWriter, r *http.Request, key cache. return false, nil } if handled, _, serveErr := httputil.ServeCacheHit(w, clientHeaders(headers), cr, err); handled { - labelCacheResult(ctx, CacheHit) + h.labelCacheHit(ctx) logging.FromContext(ctx).DebugContext(ctx, "Cache hit") return true, errors.WithStack(serveErr) } @@ -519,16 +519,30 @@ func (h *Handler) fill(ctx context.Context, w *spool.Writer, r *http.Request, ke maps.Copy(w.Header(), resp.Header) w.WriteHeader(http.StatusOK) + result := fillSuccess + if h.external { + defer func() { metrics.RecordCacheFill(ctx, result) }() + } + var fw *fillWriter cw, err := h.cache.Create(ctx, key, h.cacheHeaders(resp), h.ttlFunc(r)) if err != nil { + result = fillCreateError logging.FromContext(ctx).WarnContext(ctx, "Failed to create cache entry", "error", err) - } else if err := w.Store(h.verified(r, cw)); err != nil { - return errors.Wrap(err, "store response") + } else { + fw = h.fillWriter(r, cw) + if err := w.Store(fw); err != nil { + result = fillCreateError + return errors.Wrap(err, "store response") + } } if _, err := io.Copy(w, resp.Body); err != nil { + if fw != nil { + result = fw.result(ctx, fillTransferError) + } return errors.Wrap(err, "stream response") } if err := w.Commit(); err != nil { + result = fw.result(ctx, fillCommitError) logging.FromContext(ctx).WarnContext(ctx, "Failed to cache response", "error", err) } return nil @@ -542,55 +556,83 @@ func (h *Handler) cacheHeaders(resp *http.Response) http.Header { return headers } -// verified commits cw only if the body verifies. A failed commit does not fail -// the shared response, so clients still receive the bytes, as in streamAndCache. -func (h *Handler) verified(r *http.Request, cw cache.Writer) cache.Writer { - if h.verifyFunc == nil { - return cw - } - verifier := h.verifyFunc(r) - if verifier == nil { - return cw +// Outcomes recorded by cachew.cache.fills. +const ( + fillSuccess = "success" + fillCreateError = "create_error" + fillWriteError = "write_error" + fillTransferError = "transfer_error" + fillCommitError = "commit_error" + fillCanceled = "canceled" +) + +// fillWriter commits cw only if the body verifies, and records whether the +// cache write itself failed. A failed commit does not fail the shared response, +// so clients still receive the bytes, as in streamAndCache. +func (h *Handler) fillWriter(r *http.Request, cw cache.Writer) *fillWriter { + fw := &fillWriter{Writer: cw} + if h.verifyFunc != nil { + fw.verifier = h.verifyFunc(r) } - return &verifyingWriter{Writer: cw, verifier: verifier} + return fw } -type verifyingWriter struct { +type fillWriter struct { cache.Writer - verifier Verifier - err error + verifier Verifier + verifyErr error + writeErr error } -var _ cache.PendingReader = (*verifyingWriter)(nil) +var _ cache.PendingReader = (*fillWriter)(nil) -func (v *verifyingWriter) Write(p []byte) (int, error) { - n, err := v.Writer.Write(p) - if v.err == nil { - _, v.err = v.verifier.Write(p[:n]) +func (f *fillWriter) Write(p []byte) (int, error) { + n, err := f.Writer.Write(p) + if err != nil && f.writeErr == nil { + f.writeErr = err + } + if f.verifier != nil && f.verifyErr == nil { + _, f.verifyErr = f.verifier.Write(p[:n]) } return n, errors.WithStack(err) } -func (v *verifyingWriter) Close() error { - err := v.err +func (f *fillWriter) Close() error { + if f.verifier == nil { + return errors.WithStack(f.Writer.Close()) + } + err := f.verifyErr if err == nil { - err = v.verifier.Verify() + err = f.verifier.Verify() } if err != nil { err = errors.Wrap(err, "verify response") - return errors.Join(err, v.Writer.Abort(err)) + return errors.Join(err, f.Writer.Abort(err)) } - return errors.WithStack(v.Writer.Close()) + return errors.WithStack(f.Writer.Close()) } -func (v *verifyingWriter) OpenPending() (cache.ReadAtCloser, error) { - pending, ok := v.Writer.(cache.PendingReader) +func (f *fillWriter) OpenPending() (cache.ReadAtCloser, error) { + pending, ok := f.Writer.(cache.PendingReader) if !ok { return nil, errors.WithStack(cache.ErrPendingReadUnsupported) } return errors.WithStack2(pending.OpenPending()) } +// result classifies a failed fill, preferring the cache's own write failure +// over the stream or commit failure it caused. +func (f *fillWriter) result(ctx context.Context, otherwise string) string { + switch { + case f.writeErr != nil: + return fillWriteError + case ctx.Err() != nil: + return fillCanceled + default: + return otherwise + } +} + // fetch sends the transformed request upstream. On failure it writes an error // response to w and returns false. func (h *Handler) fetch(w http.ResponseWriter, r *http.Request) (*http.Response, bool) { @@ -638,14 +680,14 @@ func (h *Handler) streamNonOKResponse(w http.ResponseWriter, resp *http.Response // discarded and the response is streamed in full. func (h *Handler) streamAndCache(w http.ResponseWriter, r *http.Request, key cache.Key, resp *http.Response) error { maps.Copy(w.Header(), resp.Header) - result := "success" + result := fillSuccess if h.external { defer func() { metrics.RecordCacheFill(r.Context(), result) }() } cw, err := h.cache.Create(r.Context(), key, h.cacheHeaders(resp), h.ttlFunc(r)) if err != nil { - result = "create_error" + result = fillCreateError _, copyErr := io.Copy(w, resp.Body) return errors.Join(errors.Wrap(err, "create cache entry"), errors.Wrap(copyErr, "stream response")) } @@ -681,13 +723,13 @@ func (h *Handler) streamAndCache(w http.ResponseWriter, r *http.Request, key cac cacheErr := <-cacheErrCh switch { case tee.err != nil: - result = "write_error" + result = fillWriteError case r.Context().Err() != nil: - result = "canceled" + result = fillCanceled case copyErr != nil || closeErr != nil: - result = "transfer_error" + result = fillTransferError case cacheErr != nil: - result = "commit_error" + result = fillCommitError } return errors.Join( errors.Wrap(errors.Join(copyErr, closeErr), "stream response"), diff --git a/internal/strategy/handler/instrumentation_test.go b/internal/strategy/handler/instrumentation_test.go index 1c4a78e8..ca00a21e 100644 --- a/internal/strategy/handler/instrumentation_test.go +++ b/internal/strategy/handler/instrumentation_test.go @@ -14,12 +14,14 @@ import ( "github.com/alecthomas/assert/v2" "github.com/alecthomas/errors" + "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" "github.com/block/cachew/internal/cache" + "github.com/block/cachew/internal/httputil" "github.com/block/cachew/internal/logging" "github.com/block/cachew/internal/metadatadb" "github.com/block/cachew/internal/metrics" @@ -79,33 +81,32 @@ func TestArtifactCacheSourcesAndBackfill(t *testing.T) { metadata := metadatadb.New(artifactContext(t), metadatadb.NewMemoryBackend()) c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{disk, memory}, metadata) t.Cleanup(func() { assert.NoError(t, c.Close()) }) - h := handler.New(origin.Client(), c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { + h := handler.New(originClient(origin.Client().Transport), c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL+r.URL.Path, nil) }) - serve := func(headers http.Header) *httptest.ResponseRecorder { + serve := func(headers http.Header) (*httptest.ResponseRecorder, map[string]string) { r := artifactHTTPRequest(t, "/artifact") r.Header = headers - w := httptest.NewRecorder() - h.ServeHTTP(w, r) - return w + return serveLabeled(h, r) } - first := serve(http.Header{}) + first, labels := serve(http.Header{}) assert.Equal(t, http.StatusOK, first.Code) assert.Equal(t, "artifact", first.Body.String()) - second := serve(http.Header{}) + assert.Equal(t, map[string]string{handler.CacheResultAttribute: handler.CacheMiss}, labels) + second, labels := serve(http.Header{}) assert.Equal(t, "artifact", second.Body.String()) + assert.Equal(t, "disk", labels[handler.CacheSourceAttribute]) assert.NoError(t, disk.Delete(artifactContext(t), cache.NewKey("/artifact"))) - third := serve(http.Header{}) + third, labels := serve(http.Header{}) assert.Equal(t, "artifact", third.Body.String()) - fourth := serve(http.Header{}) + assert.Equal(t, map[string]string{handler.CacheResultAttribute: handler.CacheHit, handler.CacheSourceAttribute: "memory"}, labels) + fourth, labels := serve(http.Header{}) assert.Equal(t, "artifact", fourth.Body.String()) - conditional := serve(http.Header{"If-None-Match": []string{fourth.Header().Get("Etag")}}) + assert.Equal(t, "disk", labels[handler.CacheSourceAttribute]) + conditional, labels := serve(http.Header{"If-None-Match": []string{fourth.Header().Get("ETag")}}) assert.Equal(t, http.StatusNotModified, conditional.Code) + assert.Equal(t, map[string]string{handler.CacheResultAttribute: handler.CacheHit, handler.CacheSourceAttribute: "disk"}, labels) assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.origin.requests", "status_code", "200")) - assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "result", "miss")) - assert.Equal(t, int64(2), artifactCount(t, reader, "cachew.cache.resolutions", "source", "disk", "result", "hit")) - assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "source", "memory", "tier", "1")) - assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "source", "disk", "result", "conditional")) assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "purpose", "origin", "result", "success")) assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "purpose", "backfill", "result", "success")) assert.Equal(t, int64(3), artifactCount(t, reader, "cachew.cache.tier_writes", "result", "success")) @@ -124,7 +125,7 @@ func TestArtifactRedirectsAndThrottling(t *testing.T) { defer origin.Close() c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{mustNewMemoryCache()}, nil) defer c.Close() - h := handler.New(origin.Client(), c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { + h := handler.New(originClient(origin.Client().Transport), c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL+"/redirect", nil) }) for range 2 { @@ -135,7 +136,6 @@ func TestArtifactRedirectsAndThrottling(t *testing.T) { } assert.Equal(t, int64(2), artifactCount(t, reader, "cachew.origin.requests", "status_code", "302")) assert.Equal(t, int64(2), artifactCount(t, reader, "cachew.origin.requests", "status_code", "429")) - assert.Equal(t, int64(2), artifactCount(t, reader, "cachew.cache.resolutions", "result", "miss")) assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.fills")) } @@ -155,7 +155,7 @@ func TestArtifactTransportFailures(t *testing.T) { } { t.Run(tt.name, func(t *testing.T) { reader := artifactReader(t) - client := &http.Client{Transport: artifactTransportFunc(func(*http.Request) (*http.Response, error) { return nil, tt.err })} + client := originClient(artifactTransportFunc(func(*http.Request) (*http.Response, error) { return nil, tt.err })) c := mustNewMemoryCache() defer c.Close() h := handler.New(client, c, newTestSpools(t)).Metrics("host", "registry.example") @@ -210,7 +210,7 @@ func TestArtifactFillFailuresKeepClientResponse(t *testing.T) { defer origin.Close() c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{artifactFailCache{Cache: mustNewMemoryCache(), phase: phase}}, nil) defer c.Close() - h := handler.New(origin.Client(), c, newTestSpools(t)).Transform(func(r *http.Request) (*http.Request, error) { + h := handler.New(originClient(origin.Client().Transport), c, newTestSpools(t)).Transform(func(r *http.Request) (*http.Request, error) { return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL, nil) }) w := httptest.NewRecorder() @@ -236,11 +236,10 @@ func TestArtifactInterruptedBodyIsNotSuccessfulFill(t *testing.T) { defer origin.Close() c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{mustNewMemoryCache()}, nil) defer c.Close() - h := handler.New(origin.Client(), c, newTestSpools(t)).Transform(func(r *http.Request) (*http.Request, error) { + h := handler.New(originClient(origin.Client().Transport), c, newTestSpools(t)).Transform(func(r *http.Request) (*http.Request, error) { return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL, nil) }) - w := httptest.NewRecorder() - h.ServeHTTP(w, artifactHTTPRequest(t, "/artifact")) + assert.Panics(t, func() { h.ServeHTTP(httptest.NewRecorder(), artifactHTTPRequest(t, "/artifact")) }) assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.origin.requests", "status_code", "200")) assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "result", "transfer_error")) assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.tier_writes", "result", "success")) @@ -248,9 +247,9 @@ func TestArtifactInterruptedBodyIsNotSuccessfulFill(t *testing.T) { func TestArtifactInternalForwardAndCardinality(t *testing.T) { reader := artifactReader(t) - client := &http.Client{Transport: artifactTransportFunc(func(*http.Request) (*http.Response, error) { + client := originClient(artifactTransportFunc(func(*http.Request) (*http.Response, error) { return &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: io.NopCloser(strings.NewReader("artifact"))}, nil - })} + })) c := cache.NoOpCache() h := handler.New(client, c, newTestSpools(t)).Metrics("proxy", "other").InternalForward() var wg sync.WaitGroup @@ -264,7 +263,6 @@ func TestArtifactInternalForwardAndCardinality(t *testing.T) { wg.Wait() assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.origin.requests")) assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.fills")) - assert.Equal(t, int64(50), artifactCount(t, reader, "cachew.cache.resolutions", "registry", "other")) var data metricdata.ResourceMetrics assert.NoError(t, reader.Collect(artifactContext(t), &data)) for _, scope := range data.ScopeMetrics { @@ -299,7 +297,6 @@ func TestArtifactBackfillFailureRemainsVisible(t *testing.T) { } func TestArtifactStaleTierIsNotCountedAsHit(t *testing.T) { - reader := artifactReader(t) disk, err := cache.NewDisk(artifactContext(t), cache.DiskConfig{Root: t.TempDir()}) assert.NoError(t, err) memory := mustNewMemoryCache() @@ -316,12 +313,10 @@ func TestArtifactStaleTierIsNotCountedAsHit(t *testing.T) { h := handler.New(http.DefaultClient, c, newTestSpools(t)) r := artifactHTTPRequest(t, "/artifact") r.Header.Set("If-Match", `"new"`) - w := httptest.NewRecorder() - h.ServeHTTP(w, r) + w, labels := serveLabeled(h, r) assert.Equal(t, http.StatusOK, w.Code) assert.Equal(t, "new", w.Body.String()) - assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "source", "memory", "tier", "1", "result", "hit")) - assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.cache.resolutions", "source", "disk")) + assert.Equal(t, "memory", labels[handler.CacheSourceAttribute]) } func TestArtifactS3(t *testing.T) { @@ -337,15 +332,29 @@ func TestArtifactS3(t *testing.T) { c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{disk, shared}, metadatadb.New(artifactContext(t), metadatadb.NewMemoryBackend())) defer c.Close() h := handler.New(http.DefaultClient, c, newTestSpools(t)).Metrics("host", "registry.example") - w := httptest.NewRecorder() - h.ServeHTTP(w, artifactHTTPRequest(t, "/artifact")) + w, labels := serveLabeled(h, artifactHTTPRequest(t, "/artifact")) assert.Equal(t, http.StatusOK, w.Code) assert.Equal(t, "artifact", w.Body.String()) - assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.resolutions", "source", "s3", "tier", "1", "result", "hit")) + assert.Equal(t, "s3", labels[handler.CacheSourceAttribute]) assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.cache.fills", "purpose", "backfill", "result", "success")) assert.Equal(t, int64(0), artifactCount(t, reader, "cachew.origin.requests")) } +func originClient(rt http.RoundTripper) *http.Client { + return &http.Client{Transport: metrics.NewOriginTransport(rt)} +} + +func serveLabeled(h http.Handler, r *http.Request) (*httptest.ResponseRecorder, map[string]string) { + labeler := &otelhttp.Labeler{} + w := httptest.NewRecorder() + h.ServeHTTP(w, r.WithContext(otelhttp.ContextWithLabeler(r.Context(), labeler))) + labels := map[string]string{} + for _, kv := range labeler.Get() { + labels[string(kv.Key)] = kv.Value.Emit() + } + return w, labels +} + func artifactContext(t *testing.T) context.Context { t.Helper() _, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelError}) @@ -368,15 +377,35 @@ func TestArtifactMetricsPrometheusExport(t *testing.T) { defer origin.Close() c := cache.MaybeNewTiered(artifactContext(t), []cache.Cache{mustNewMemoryCache()}, nil) defer c.Close() - h := handler.New(origin.Client(), c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { + h := handler.New(originClient(origin.Client().Transport), c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL, nil) }) h.ServeHTTP(httptest.NewRecorder(), artifactHTTPRequest(t, "/artifact")) w := httptest.NewRecorder() exporter.Handler().ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/metrics", nil)) text := w.Body.String() - for _, name := range []string{"cachew_cache_resolutions_total{", "cachew_origin_requests_total{", "cachew_origin_request_duration_seconds_count{", "cachew_cache_fills_total{", "cachew_cache_tier_writes_total{"} { + for _, name := range []string{"cachew_origin_requests_total{", "cachew_origin_request_duration_seconds_count{", "cachew_cache_fills_total{", "cachew_cache_tier_writes_total{"} { assert.True(t, strings.Contains(text, name), "missing Prometheus instrument %s", name) } assert.True(t, strings.Contains(text, `registry="registry.example"`)) } + +func TestArtifactCooldownIsNotOriginTraffic(t *testing.T) { + reader := artifactReader(t) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusTooManyRequests) + })) + defer origin.Close() + client := &http.Client{Transport: httputil.NewBackoffTransport(metrics.NewOriginTransport(origin.Client().Transport), time.Minute)} + c := mustNewMemoryCache() + defer c.Close() + h := handler.New(client, c, newTestSpools(t)).Metrics("host", "registry.example").Transform(func(r *http.Request) (*http.Request, error) { + return http.NewRequestWithContext(r.Context(), http.MethodGet, origin.URL, nil) + }) + for range 3 { + w := httptest.NewRecorder() + h.ServeHTTP(w, artifactHTTPRequest(t, "/artifact")) + assert.Equal(t, http.StatusTooManyRequests, w.Code) + } + assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.origin.requests", "status_code", "429")) +} diff --git a/internal/strategy/oci.go b/internal/strategy/oci.go index 6a74562d..0e8ca08e 100644 --- a/internal/strategy/oci.go +++ b/internal/strategy/oci.go @@ -116,12 +116,12 @@ func NewOCI(ctx context.Context, config OCIConfig, c cache.Cache, mux Mux, spool client: &http.Client{Transport: transport}, logger: logging.FromContext(ctx), } - o.blobs = handler.New(o.client, c, spools). + o.blobs = handler.New(o.client, c, spools).Metrics("oci", upstream.Hostname()). CacheKey(o.blobKey). Transform(o.transform). Verify(o.verifier). ParallelFetch(int64(config.BlobChunkSizeMB)<<20, int(config.BlobConcurrency)) - o.manifests = handler.New(o.client, c, spools). + o.manifests = handler.New(o.client, c, spools).Metrics("oci", upstream.Hostname()). CacheKey(o.manifestKey). Transform(o.transform). Vary(o.manifestVary). From 76da2ca02e6a7d76962269162d74b7c3957fa19a Mon Sep 17 00:00:00 2001 From: Elizabeth Worstell Date: Mon, 9 Nov 2026 14:36:30 -0800 Subject: [PATCH 3/3] fix: label HEAD cache hits with the tier that answered Tiered and single-tier Stat now report the backend that found the object, so HEAD hits carry the correct cachew.cache.source instead of the fallback value. Co-authored-by: Chain <492cad42391b4a5c84d831885f468e6f1699c0a8c39241f1c002235ece37ee69@buzz> Co-Authored-By: Claude Opus 5.5 --- internal/cache/instrumentation.go | 7 ++++++ internal/cache/tiered.go | 11 +++++--- .../strategy/handler/instrumentation_test.go | 25 +++++++++++++++++++ 3 files changed, 40 insertions(+), 3 deletions(-) diff --git a/internal/cache/instrumentation.go b/internal/cache/instrumentation.go index 4d832f1d..a4744e91 100644 --- a/internal/cache/instrumentation.go +++ b/internal/cache/instrumentation.go @@ -29,6 +29,13 @@ func BackendName(c Cache) string { } } +// observeSource reports c as the backend that answered a lookup that found the object. +func observeSource(ctx context.Context, c Cache, err error) { + if err == nil || errors.Is(err, ErrNotModified) || errors.Is(err, ErrRangeNotSatisfiable) { + metrics.ObserveCacheSource(ctx, BackendName(c)) + } +} + const ( writeSuccess = "success" writeCommitError = "commit_error" diff --git a/internal/cache/tiered.go b/internal/cache/tiered.go index 52e6e37a..d1dcb241 100644 --- a/internal/cache/tiered.go +++ b/internal/cache/tiered.go @@ -52,12 +52,16 @@ type authoritativeCache struct { func (c authoritativeCache) Open(ctx context.Context, key Key, opts ...Option) (io.ReadCloser, http.Header, error) { r, headers, err := c.Cache.Open(ctx, key, opts...) - if err == nil || errors.Is(err, ErrNotModified) || errors.Is(err, ErrRangeNotSatisfiable) { - metrics.ObserveCacheSource(ctx, BackendName(c.Cache)) - } + observeSource(ctx, c.Cache, err) return r, headers, errors.WithStack(err) } +func (c authoritativeCache) Stat(ctx context.Context, key Key, opts ...Option) (http.Header, error) { + headers, err := c.Cache.Stat(ctx, key, opts...) + observeSource(ctx, c.Cache, err) + return headers, errors.WithStack(err) +} + func (c authoritativeCache) Create(ctx context.Context, key Key, headers http.Header, ttl time.Duration, opts ...Option) (Writer, error) { w, err := c.Cache.Create(ctx, key, headers, ttl, opts...) return observeWriter(ctx, c.Cache, 0, w, err), err @@ -246,6 +250,7 @@ func (t Tiered) Stat(ctx context.Context, key Key, opts ...Option) (http.Header, } // Any other outcome (success, ErrNotModified, or a hard error) is // definitive for this tier; surface it with its headers. + observeSource(ctx, c, err) return headers, errors.WithStack(err) } if len(probeErrs) > 0 { diff --git a/internal/strategy/handler/instrumentation_test.go b/internal/strategy/handler/instrumentation_test.go index ca00a21e..f7a590ea 100644 --- a/internal/strategy/handler/instrumentation_test.go +++ b/internal/strategy/handler/instrumentation_test.go @@ -409,3 +409,28 @@ func TestArtifactCooldownIsNotOriginTraffic(t *testing.T) { } assert.Equal(t, int64(1), artifactCount(t, reader, "cachew.origin.requests", "status_code", "429")) } + +func TestArtifactHeadHitSource(t *testing.T) { + disk, err := cache.NewDisk(artifactContext(t), cache.DiskConfig{Root: t.TempDir()}) + assert.NoError(t, err) + memory := mustNewMemoryCache() + key := cache.NewKey("/artifact") + assert.NoError(t, cache.WriteFunc(artifactContext(t), memory, key, nil, time.Hour, func(w io.Writer) error { _, err := io.WriteString(w, "artifact"); return err })) + for _, tt := range []struct { + name string + c cache.Cache + source string + }{ + {"tiered", cache.MaybeNewTiered(artifactContext(t), []cache.Cache{disk, memory}, metadatadb.New(artifactContext(t), metadatadb.NewMemoryBackend())), "memory"}, + {"single", cache.MaybeNewTiered(artifactContext(t), []cache.Cache{memory}, nil), "memory"}, + } { + t.Run(tt.name, func(t *testing.T) { + h := handler.New(http.DefaultClient, tt.c, newTestSpools(t)) + r := artifactHTTPRequest(t, "/artifact") + r.Method = http.MethodHead + w, labels := serveLabeled(h, r) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, map[string]string{handler.CacheResultAttribute: handler.CacheHit, handler.CacheSourceAttribute: tt.source}, labels) + }) + } +}