feat: expose artifact cache and origin metrics

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

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

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

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