diff --git a/README.md b/README.md index bdf2273c..e41653aa 100644 --- a/README.md +++ b/README.md @@ -353,6 +353,27 @@ 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, OCI blobs and manifests, and GitHub release downloads) export: + +| OpenTelemetry instrument | Type | Extra labels | +| --- | --- | --- | +| `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. + +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 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 `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 ```hcl 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 new file mode 100644 index 00000000..a4744e91 --- /dev/null +++ b/internal/cache/instrumentation.go @@ -0,0 +1,116 @@ +package cache + +import ( + "context" + "sync" + + "github.com/alecthomas/errors" + + "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" + } +} + +// 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" + 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 +} + +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 == "" { + 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..a635ea75 --- /dev/null +++ b/internal/cache/instrumentation_internal_test.go @@ -0,0 +1,90 @@ +package cache + +import ( + "context" + "io" + "os" + "strings" + "testing" + "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" +) + +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) + } +} + +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 e851ceeb..d1dcb241 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,23 @@ 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...) + 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 +} + func (c authoritativeCache) Invalidate(context.Context, Key) error { return nil } @@ -107,7 +125,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{}) @@ -232,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 { @@ -285,6 +304,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 +316,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 + var fallbackSource Cache var probeErrs []error rejected := false // a tier failed If-Match; deeper tiers may hold the named version @@ -317,6 +339,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)) return nil, headers, errors.WithStack(err) case err != nil: // A hard error is definitive when no earlier tier produced a @@ -338,12 +361,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, fallbackSource = r, headers, c continue } if fallback != nil { discardTieredReader(ctx, key, fallback) } + metrics.ObserveCacheSource(ctx, BackendName(c)) return t.convergeTier0(ctx, key, r, headers, c, i, partial), headers, nil } if len(probeErrs) > 0 { @@ -353,6 +377,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(fallbackSource)) return fallback, fallbackHeaders, nil } if rejected { @@ -372,18 +397,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 +500,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 +542,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 +577,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 +627,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..a816f47f --- /dev/null +++ b/internal/metrics/artifact.go @@ -0,0 +1,166 @@ +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 origin exchanges from cache publication. +type ArtifactMetrics struct { + 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{ + 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 artifactRequest struct { + metrics *ArtifactMetrics + attrs []attribute.KeyValue + external bool + source atomic.Pointer[string] +} + +// 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 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, &backend) + } +} + +// 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 { + return *s, true + } + } + return "", false +} + +// 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...)...) +} + +// 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 +} + +// NewOriginTransport measures exchanges sent to base. +func NewOriginTransport(base http.RoundTripper) *OriginTransport { + return &OriginTransport{base: base} +} + +// 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. + } + 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. +} + +// 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/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..ea18bd52 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 @@ -106,10 +111,12 @@ func New(client *http.Client, c cache.Cache, spools *spool.Registry) *Handler { 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 +130,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 +233,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) @@ -291,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 { @@ -316,7 +349,7 @@ 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 @@ -329,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) } @@ -486,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 @@ -509,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) { @@ -605,9 +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 := 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 = fillCreateError _, copyErr := io.Copy(w, resp.Body) return errors.Join(errors.Wrap(err, "create cache entry"), errors.Wrap(copyErr, "stream response")) } @@ -640,9 +720,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 = fillWriteError + case r.Context().Err() != nil: + result = fillCanceled + case copyErr != nil || closeErr != nil: + result = fillTransferError + case cacheErr != nil: + result = fillCommitError + } 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..f7a590ea --- /dev/null +++ b/internal/strategy/handler/instrumentation_test.go @@ -0,0 +1,436 @@ +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/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" + "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(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, map[string]string) { + r := artifactHTTPRequest(t, "/artifact") + r.Header = headers + return serveLabeled(h, r) + } + first, labels := serve(http.Header{}) + assert.Equal(t, http.StatusOK, first.Code) + assert.Equal(t, "artifact", first.Body.String()) + 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, labels := serve(http.Header{}) + assert.Equal(t, "artifact", third.Body.String()) + 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()) + 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.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(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 { + 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(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 := 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") + 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(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.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(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) + }) + 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")) +} + +func TestArtifactInternalForwardAndCardinality(t *testing.T) { + reader := artifactReader(t) + 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 + 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")) + 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) { + 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, labels := serveLabeled(h, r) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "new", w.Body.String()) + assert.Equal(t, "memory", labels[handler.CacheSourceAttribute]) +} + +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, labels := serveLabeled(h, artifactHTTPRequest(t, "/artifact")) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "artifact", w.Body.String()) + 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}) + 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(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_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")) +} + +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) + }) + } +} 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/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). 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 {