Split internal/imgcache into the image proxy and the cache (closes #317)

The new package internal/imageproxy holds Proxy, which was imgcache.Service,
with its Config, ImageResponse, CacheStatus and errors. NewProxy makes and
registers the counters of upstream fetches and transforms. internal/imgcache
holds the cache only, with its hit and miss counters, and imgcache.CacheConfig
is imgcache.Config. The cache reads and writes each kind it keeps with one pair
of methods: Variant and PutVariant, Source and PutSource, Failure and
PutFailure. Source replaces LookupSource followed by GetSourceContent.
imgcache.go and module.go are gone. The proxy's tests moved with it. They reach
the cache through its database, its state directory and its metrics registry,
and run its reconciliation pass with a second cache, as pixad does at startup.

Model: opus-5-5

TODO.md

@@ -49,6 +49,9 @@ P2: security: per-origin rate limiting
     - `Variant` and `Source` each open the file and hand its access time to the
       background writer. A cache hit still makes no database call.
     - `imgcache.go` and `module.go` are gone.
+    - `NewProxy` makes the counters of upstream fetches and transforms, and
+      registers them on the registry it is given. The cache keeps only the hit
+      and miss counters.
     - A cached source whose row cannot be read is logged as "failed to load
       cached source, fetching". It was logged as "source lookup failed".
     - The proxy's tests moved to `internal/imageproxy`. They reach the cache

internal/handlers/cache_hit_internal_test.go

@@ -143,7 +143,7 @@ func newCountingDatabaseServer(
 		}),
 		MaxResponseSize: 50 << 20,
 		Logger:          slog.New(slog.DiscardHandler),
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		tb.Fatalf("imageproxy.NewProxy() error = %v", err)
 	}

internal/handlers/first_bytes_internal_test.go

@@ -57,7 +57,7 @@ func newRoutesFetchingThrough(
 		}),
 		MaxResponseSize: 50 << 20,
 		Logger:          log,
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("imageproxy.NewProxy() error = %v", err)
 	}

internal/handlers/handlers_internal_test.go

@@ -67,7 +67,7 @@ func setupTestHandler(t *testing.T) *testFixtures {
 			MaxConcurrentProcessing: 2,
 		}),
 		MaxResponseSize: 50 << 20,
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("failed to create proxy: %v", err)
 	}

internal/handlers/image_cache_control_internal_test.go

@@ -58,7 +58,7 @@ func newSignedHostServer(
 		}),
 		MaxResponseSize: 50 << 20,
 		Logger:          log,
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("imageproxy.NewProxy() error = %v", err)
 	}

internal/handlers/image_errors_internal_test.go

@@ -54,7 +54,7 @@ func newImageRoute(t *testing.T, fetcher httpfetcher.Fetcher) http.Handler {
 			MaxConcurrentProcessing: 2,
 		}),
 		MaxResponseSize: 50 << 20,
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("failed to create proxy: %v", err)
 	}

internal/handlers/referer_blocklist_internal_test.go

@@ -70,7 +70,7 @@ func newRefererRoutes(t *testing.T) (http.Handler, *Handlers, *countingFetcher)
 			MaxConcurrentProcessing: 2,
 		}),
 		MaxResponseSize: 50 << 20,
-	})
+	}, prometheus.NewRegistry())
 	if err != nil {
 		t.Fatalf("imageproxy.NewProxy() error = %v", err)
 	}

internal/imageproxy/counters_internal_test.go

@@ -0,0 +1,78 @@
+package imageproxy
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"sneak.berlin/go/pixa/internal/imgcache"
+)
+
+// TestNewProxy_RegistersItsCountersOnItsRegistry checks that NewProxy
+// registers the proxy's counters on the registry it is given, under the
+// names README.md gives them. The cache registers its own counters on
+// another registry.
+func TestNewProxy_RegistersItsCountersOnItsRegistry(t *testing.T) {
+	t.Parallel()
+
+	cache, err := imgcache.NewCache(setupTestDB(t),
+		imgcache.Config{StateDir: t.TempDir(), MaxBytes: 1 << 30},
+		prometheus.NewRegistry())
+	if err != nil {
+		t.Fatalf("imgcache.NewCache() error = %v", err)
+	}
+
+	registry := prometheus.NewRegistry()
+
+	_, err = NewProxy(&Config{Cache: cache}, registry)
+	if err != nil {
+		t.Fatalf("NewProxy() error = %v", err)
+	}
+
+	names := []string{
+		"pixa_upstream_fetches_total",
+		"pixa_upstream_fetch_bytes_total",
+		"pixa_transforms_total",
+	}
+
+	count, err := testutil.GatherAndCount(registry, names...)
+	if err != nil {
+		t.Fatalf("testutil.GatherAndCount() error = %v", err)
+	}
+
+	if count != len(names) {
+		t.Errorf("%d of the counters %v are on the registry, want %d",
+			count, names, len(names))
+	}
+}
+
+// TestNewProxy_FailsWhenItsCountersCannotBeRegistered makes a second proxy
+// on the registry of a first. Its counters have the same names, so the
+// registry refuses them, and NewProxy must return that error.
+func TestNewProxy_FailsWhenItsCountersCannotBeRegistered(t *testing.T) {
+	t.Parallel()
+
+	cache, err := imgcache.NewCache(setupTestDB(t),
+		imgcache.Config{StateDir: t.TempDir(), MaxBytes: 1 << 30},
+		prometheus.NewRegistry())
+	if err != nil {
+		t.Fatalf("imgcache.NewCache() error = %v", err)
+	}
+
+	registry := prometheus.NewRegistry()
+
+	_, err = NewProxy(&Config{Cache: cache}, registry)
+	if err != nil {
+		t.Fatalf("NewProxy() of the first proxy error = %v", err)
+	}
+
+	_, err = NewProxy(&Config{Cache: cache}, registry)
+	t.Logf("NewProxy() of the second proxy error = %v", err)
+
+	var alreadyRegistered prometheus.AlreadyRegisteredError
+	if !errors.As(err, &alreadyRegistered) {
+		t.Errorf("NewProxy() of the second proxy error = %v, want %T",
+			err, alreadyRegistered)
+	}
+}

internal/imageproxy/imageproxy.go

@@ -16,6 +16,7 @@ import (
 	"github.com/dustin/go-humanize"
 	"github.com/getsentry/sentry-go"
 	"github.com/go-chi/chi/v5/middleware"
+	"github.com/prometheus/client_golang/prometheus"
 	"golang.org/x/sync/singleflight"
 	"sneak.berlin/go/pixa/internal/httpfetcher"
 	"sneak.berlin/go/pixa/internal/imageprocessor"
@@ -34,6 +35,12 @@ type Proxy struct {
 	// variantsInProgress lets the requests that miss the same variant at the
 	// same time share one fetch and one transcode.
 	variantsInProgress singleflight.Group
+
+	// The counts of upstream fetches and transforms, made and registered by
+	// NewProxy. They are kept in memory, so adding to one waits for nothing.
+	upstreamFetches    prometheus.Counter
+	upstreamFetchBytes prometheus.Counter
+	transforms         prometheus.Counter
 }
 
 // Config holds the proxy's configuration.
@@ -54,8 +61,13 @@ type Config struct {
 // errCacheRequired is returned when the proxy is built without a cache.
 var errCacheRequired = errors.New("cache is required")
 
-// NewProxy creates a new image proxy.
-func NewProxy(cfg *Config) (*Proxy, error) {
+// NewProxy creates a new image proxy and registers its counters on
+// metricsRegistry. pixad passes Prometheus' default registry, the one
+// /metrics serves. It fails when metricsRegistry already holds counters of
+// the same names.
+func NewProxy(
+	cfg *Config, metricsRegistry prometheus.Registerer,
+) (*Proxy, error) {
 	if cfg.Cache == nil {
 		return nil, errCacheRequired
 	}
@@ -72,13 +84,36 @@ func NewProxy(cfg *Config) (*Proxy, error) {
 		return nil, err
 	}
 
-	return &Proxy{
+	p := &Proxy{
 		cache:           cfg.Cache,
 		fetcher:         cfg.Fetcher,
 		processor:       cfg.Processor,
 		log:             log,
 		maxResponseSize: cfg.MaxResponseSize,
-	}, nil
+		upstreamFetches: prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "pixa_upstream_fetches_total",
+			Help: "Fetches of a source image from upstream that read any bytes.",
+		}),
+		upstreamFetchBytes: prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "pixa_upstream_fetch_bytes_total",
+			Help: "Bytes read from upstream by fetches of source images.",
+		}),
+		transforms: prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "pixa_transforms_total",
+			Help: "Images the image processor transformed.",
+		}),
+	}
+
+	for _, counter := range []prometheus.Counter{
+		p.upstreamFetches, p.upstreamFetchBytes, p.transforms,
+	} {
+		err = metricsRegistry.Register(counter)
+		if err != nil {
+			return nil, fmt.Errorf("failed to register the proxy's counters: %w", err)
+		}
+	}
+
+	return p, nil
 }
 
 // ErrNegativeCached is returned when a URL is in the negative cache (recently failed).
@@ -132,7 +167,7 @@ func (p *Proxy) Get(
 
 	reader, size, contentType, err := p.cache.Variant(cacheKey)
 	if err == nil {
-		p.cache.IncrementStats(true, 0)
+		p.cache.IncrementStats(true)
 
 		return &ImageResponse{
 			Content:       reader,
@@ -153,7 +188,7 @@ func (p *Proxy) Get(
 	// failed or the request context has ended meanwhile
 	response, err := p.processOrWait(ctx, req)
 
-	p.cache.IncrementStats(false, 0)
+	p.cache.IncrementStats(false)
 
 	if err != nil {
 		return nil, err
@@ -378,7 +413,7 @@ func (p *Proxy) fetchAndProcess(
 	fetchBytes := int64(len(sourceData))
 
 	// Counted also when the request context has ended meanwhile
-	p.cache.IncrementUpstreamFetch(fetchBytes)
+	p.incrementUpstreamFetch(fetchBytes)
 
 	if err != nil {
 		return nil, fmt.Errorf("failed to read upstream response: %w", err)
@@ -432,6 +467,17 @@ func (p *Proxy) fetchAndProcess(
 	)
 }
 
+// incrementUpstreamFetch counts one upstream fetch that read fetchBytes bytes.
+// A fetch that read no bytes is not counted.
+func (p *Proxy) incrementUpstreamFetch(fetchBytes int64) {
+	if fetchBytes <= 0 {
+		return
+	}
+
+	p.upstreamFetches.Inc()
+	p.upstreamFetchBytes.Add(float64(fetchBytes))
+}
+
 // storeNegative stores a negative cache entry for req's source, whose fetch
 // failed, so the source is not fetched again until the entry expires. A
 // failure to store the entry is only logged.
@@ -477,7 +523,7 @@ func (p *Proxy) processAndStore(
 	processDuration := time.Since(processStart)
 
 	// Counted also when the request context has ended meanwhile
-	p.cache.IncrementTransformCount()
+	p.transforms.Inc()
 
 	// Read processed content
 	processedData, err := io.ReadAll(processResult.Content)

internal/imageproxy/stats_internal_test.go

@@ -14,7 +14,8 @@ import (
 
 // TestProxy_Get_CountsStats walks Get through a miss that fetches the
 // source, a hit, a miss that reuses the cached source, and two misses whose
-// source cannot be used, checking every counter of the cache after each.
+// source cannot be used, checking every counter of the proxy and its cache
+// after each.
 func TestProxy_Get_CountsStats(t *testing.T) {
 	t.Parallel()
 
@@ -114,9 +115,10 @@ func (u *fakeUpstream) Read(p []byte) (int, error) {
 	return 0, u.err
 }
 
-// TestProxy_Get_CountsInterruptedMisses checks every counter of the cache
-// after a miss whose request context ends during or after the upstream
-// fetch, and after a miss whose upstream body is over the size limit.
+// TestProxy_Get_CountsInterruptedMisses checks every counter of the proxy
+// and its cache after a miss whose request context ends during or after the
+// upstream fetch, and after a miss whose upstream body is over the size
+// limit.
 func TestProxy_Get_CountsInterruptedMisses(t *testing.T) {
 	t.Parallel()
 
@@ -191,8 +193,8 @@ func TestProxy_Get_CountsInterruptedMisses(t *testing.T) {
 }
 
 // TestProxy_Get_CountsHitAfterRequestEnds checks every counter of the
-// cache after a hit served with a request context that has already ended:
-// only the hit count moves.
+// proxy and its cache after a hit served with a request context that has
+// already ended: only the hit count moves.
 func TestProxy_Get_CountsHitAfterRequestEnds(t *testing.T) {
 	t.Parallel()
 

internal/imageproxy/testutil_internal_test.go

@@ -165,9 +165,9 @@ func generateTestGIF(t *testing.T, width, height int, c color.Color) []byte {
 	return buf.Bytes()
 }
 
-// testProxy is a Proxy with the database, the state directory and the
-// metrics registry of its cache, through which a test reads and changes the
-// cache behind the proxy.
+// testProxy is a Proxy with the database and the state directory of its
+// cache, through which a test reads and changes the cache behind the proxy.
+// registry is the metrics registry the proxy and its cache share.
 type testProxy struct {
 	*Proxy
 
@@ -207,7 +207,7 @@ func setupTestProxy(t *testing.T) (*testProxy, *TestFixtures) {
 			MaxConcurrentProcessing: 2,
 		}),
 		MaxResponseSize: 50 << 20,
-	})
+	}, registry)
 	if err != nil {
 		t.Fatalf("failed to create proxy: %v", err)
 	}
@@ -220,7 +220,7 @@ func setupTestProxy(t *testing.T) (*testProxy, *TestFixtures) {
 	}, fixtures
 }
 
-// counterValues holds the values of a cache's counters.
+// counterValues holds the values of the counters of a proxy and its cache.
 type counterValues struct {
 	hitCount           int64
 	missCount          int64
@@ -229,13 +229,14 @@ type counterValues struct {
 	transformCount     int64
 }
 
-// readCounters reads the counters of svc's cache from its metrics registry.
+// readCounters reads the counters of svc and its cache from their metrics
+// registry.
 func readCounters(t *testing.T, svc *testProxy) counterValues {
 	t.Helper()
 
 	families, err := svc.registry.Gather()
 	if err != nil {
-		t.Fatalf("failed to gather the cache's counters: %v", err)
+		t.Fatalf("failed to gather the counters: %v", err)
 	}
 
 	values := make(map[string]int64, len(families))

internal/imgcache/cache.go

@@ -84,13 +84,10 @@ type Cache struct {
 	pendingAccessTimesDone    chan struct{}
 	pendingAccessTimesCancel  context.CancelFunc
 
-	// The counts of what the cache does, made and registered by
+	// The counts of cache hits and misses, made and registered by
 	// registerCounters (see counters.go)
-	cacheHits          prometheus.Counter
-	cacheMisses        prometheus.Counter
-	upstreamFetches    prometheus.Counter
-	upstreamFetchBytes prometheus.Counter
-	transforms         prometheus.Counter
+	cacheHits   prometheus.Counter
+	cacheMisses prometheus.Counter
 
 	// reconciliationPageSize is the most rows one read of a content table
 	// returns in the reconciliation pass. NewCache sets it to
@@ -367,34 +364,15 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
 	}
 }
 
-// IncrementStats counts a cache hit or miss, and an upstream fetch that read
-// fetchBytes bytes, as IncrementUpstreamFetch does. Like the other Increment
-// methods, it only adds to the cache's Prometheus counters in memory (see
-// counters.go), so it never waits for the database or fails.
-func (c *Cache) IncrementStats(hit bool, fetchBytes int64) {
+// IncrementStats counts a cache hit or miss. It only adds to the cache's
+// Prometheus counters in memory (see counters.go), so it never waits for the
+// database or fails.
+func (c *Cache) IncrementStats(hit bool) {
 	if hit {
 		c.cacheHits.Inc()
 	} else {
 		c.cacheMisses.Inc()
 	}
-
-	c.IncrementUpstreamFetch(fetchBytes)
-}
-
-// IncrementUpstreamFetch counts one upstream fetch that read fetchBytes bytes.
-// A fetch that read no bytes is not counted.
-func (c *Cache) IncrementUpstreamFetch(fetchBytes int64) {
-	if fetchBytes <= 0 {
-		return
-	}
-
-	c.upstreamFetches.Inc()
-	c.upstreamFetchBytes.Add(float64(fetchBytes))
-}
-
-// IncrementTransformCount counts one image transcoded by the image processor.
-func (c *Cache) IncrementTransformCount() {
-	c.transforms.Inc()
 }
 
 // touchVariant hands the access time of a variant served from the cache to

internal/imgcache/counters.go

@@ -19,22 +19,8 @@ func (c *Cache) registerCounters(metricsRegistry prometheus.Registerer) error {
 		Name: "pixa_cache_misses_total",
 		Help: "Image requests not served from the cache.",
 	})
-	c.upstreamFetches = prometheus.NewCounter(prometheus.CounterOpts{
-		Name: "pixa_upstream_fetches_total",
-		Help: "Fetches of a source image from upstream that read any bytes.",
-	})
-	c.upstreamFetchBytes = prometheus.NewCounter(prometheus.CounterOpts{
-		Name: "pixa_upstream_fetch_bytes_total",
-		Help: "Bytes read from upstream by fetches of source images.",
-	})
-	c.transforms = prometheus.NewCounter(prometheus.CounterOpts{
-		Name: "pixa_transforms_total",
-		Help: "Images the image processor transformed.",
-	})
 
-	for _, counter := range []prometheus.Counter{
-		c.cacheHits, c.cacheMisses, c.upstreamFetches, c.upstreamFetchBytes, c.transforms,
-	} {
+	for _, counter := range []prometheus.Counter{c.cacheHits, c.cacheMisses} {
 		err := metricsRegistry.Register(counter)
 		if err != nil {
 			return fmt.Errorf("failed to register the cache's counters: %w", err)

internal/imgcache/counters_internal_test.go

@@ -25,9 +25,6 @@ func TestNewCache_RegistersItsCountersOnItsRegistry(t *testing.T) {
 	names := []string{
 		"pixa_cache_hits_total",
 		"pixa_cache_misses_total",
-		"pixa_upstream_fetches_total",
-		"pixa_upstream_fetch_bytes_total",
-		"pixa_transforms_total",
 	}
 
 	count, err := testutil.GatherAndCount(registry, names...)

internal/providers/providers.go

@@ -35,7 +35,7 @@ import (
 
 // Constructors returns the constructors of the parts this package builds,
 // for fx.Provide. The fx graph must also provide the prometheus.Registerer
-// the cache registers its counters on.
+// the cache and the image service register their counters on.
 func Constructors() []any {
 	return []any{
 		newKeys,
@@ -214,11 +214,12 @@ func newProcessor(cfg *config.Config) *imageprocessor.ImageProcessor {
 type imageServiceParams struct {
 	fx.In
 
-	Config    *config.Config
-	Cache     *imgcache.Cache
-	Fetcher   *httpfetcher.HTTPFetcher
-	Processor *imageprocessor.ImageProcessor
-	Logger    *logger.Logger
+	Config          *config.Config
+	Cache           *imgcache.Cache
+	Fetcher         *httpfetcher.HTTPFetcher
+	Processor       *imageprocessor.ImageProcessor
+	Logger          *logger.Logger
+	MetricsRegistry prometheus.Registerer
 }
 
 // newImageService builds the image service from the cache, the fetcher
@@ -232,7 +233,7 @@ func newImageService(
 		Processor:       params.Processor,
 		MaxResponseSize: params.Config.UpstreamMaxResponseSize,
 		Logger:          params.Logger.Get(),
-	})
+	}, params.MetricsRegistry)
 }
 
 // sessionManagerParams defines dependencies for newSessionManager.