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. internal/imgcache holds
the cache only, 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,9 +49,6 @@ 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

@@ -1,78 +0,0 @@
-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,7 +16,6 @@ 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"
@@ -35,12 +34,6 @@ 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.
@@ -61,13 +54,8 @@ 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 and registers its counters on
+// NewProxy creates a new image proxy.
-// metricsRegistry. pixad passes Prometheus' default registry, the one
+func NewProxy(cfg *Config) (*Proxy, error) {
-// /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
 	}
@@ -84,36 +72,13 @@ func NewProxy(
 		return nil, err
 	}
 
-	p := &Proxy{
+	return &Proxy{
 		cache:           cfg.Cache,
 		fetcher:         cfg.Fetcher,
 		processor:       cfg.Processor,
 		log:             log,
 		maxResponseSize: cfg.MaxResponseSize,
-		upstreamFetches: prometheus.NewCounter(prometheus.CounterOpts{
+	}, nil
-			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).
@@ -167,7 +132,7 @@ func (p *Proxy) Get(
 
 	reader, size, contentType, err := p.cache.Variant(cacheKey)
 	if err == nil {
-		p.cache.IncrementStats(true)
+		p.cache.IncrementStats(true, 0)
 
 		return &ImageResponse{
 			Content:       reader,
@@ -188,7 +153,7 @@ func (p *Proxy) Get(
 	// failed or the request context has ended meanwhile
 	response, err := p.processOrWait(ctx, req)
 
-	p.cache.IncrementStats(false)
+	p.cache.IncrementStats(false, 0)
 
 	if err != nil {
 		return nil, err
@@ -413,7 +378,7 @@ func (p *Proxy) fetchAndProcess(
 	fetchBytes := int64(len(sourceData))
 
 	// Counted also when the request context has ended meanwhile
-	p.incrementUpstreamFetch(fetchBytes)
+	p.cache.IncrementUpstreamFetch(fetchBytes)
 
 	if err != nil {
 		return nil, fmt.Errorf("failed to read upstream response: %w", err)
@@ -467,17 +432,6 @@ 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.
@@ -523,7 +477,7 @@ func (p *Proxy) processAndStore(
 	processDuration := time.Since(processStart)
 
 	// Counted also when the request context has ended meanwhile
-	p.transforms.Inc()
+	p.cache.IncrementTransformCount()
 
 	// Read processed content
 	processedData, err := io.ReadAll(processResult.Content)

internal/imageproxy/stats_internal_test.go

@@ -14,8 +14,7 @@ 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 proxy and its cache
+// source cannot be used, checking every counter of the cache after each.
-// after each.
 func TestProxy_Get_CountsStats(t *testing.T) {
 	t.Parallel()
 
@@ -115,10 +114,9 @@ func (u *fakeUpstream) Read(p []byte) (int, error) {
 	return 0, u.err
 }
 
-// TestProxy_Get_CountsInterruptedMisses checks every counter of the proxy
+// TestProxy_Get_CountsInterruptedMisses checks every counter of the cache
-// and its cache after a miss whose request context ends during or after the
+// after a miss whose request context ends during or after the upstream
-// upstream fetch, and after a miss whose upstream body is over the size
+// fetch, and after a miss whose upstream body is over the size limit.
-// limit.
 func TestProxy_Get_CountsInterruptedMisses(t *testing.T) {
 	t.Parallel()
 
@@ -193,8 +191,8 @@ func TestProxy_Get_CountsInterruptedMisses(t *testing.T) {
 }
 
 // TestProxy_Get_CountsHitAfterRequestEnds checks every counter of the
-// proxy and its cache after a hit served with a request context that has
+// cache after a hit served with a request context that has already ended:
-// already ended: only the hit count moves.
+// 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 and the state directory of its
+// testProxy is a Proxy with the database, the state directory and the
-// cache, through which a test reads and changes the cache behind the proxy.
+// metrics registry of its cache, through which a test reads and changes the
-// registry is the metrics registry the proxy and its cache share.
+// cache behind the proxy.
 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 the counters of a proxy and its cache.
+// counterValues holds the values of a cache's counters.
 type counterValues struct {
 	hitCount           int64
 	missCount          int64
@@ -229,14 +229,13 @@ type counterValues struct {
 	transformCount     int64
 }
 
-// readCounters reads the counters of svc and its cache from their metrics
+// readCounters reads the counters of svc's cache from its metrics registry.
-// 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 counters: %v", err)
+		t.Fatalf("failed to gather the cache's counters: %v", err)
 	}
 
 	values := make(map[string]int64, len(families))

internal/imgcache/cache.go

@@ -84,10 +84,13 @@ type Cache struct {
 	pendingAccessTimesDone    chan struct{}
 	pendingAccessTimesCancel  context.CancelFunc
 
-	// The counts of cache hits and misses, made and registered by
+	// The counts of what the cache does, made and registered by
 	// registerCounters (see counters.go)
-	cacheHits   prometheus.Counter
+	cacheHits          prometheus.Counter
-	cacheMisses prometheus.Counter
+	cacheMisses        prometheus.Counter
+	upstreamFetches    prometheus.Counter
+	upstreamFetchBytes prometheus.Counter
+	transforms         prometheus.Counter
 
 	// reconciliationPageSize is the most rows one read of a content table
 	// returns in the reconciliation pass. NewCache sets it to
@@ -364,15 +367,34 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
 	}
 }
 
-// IncrementStats counts a cache hit or miss. It only adds to the cache's
+// IncrementStats counts a cache hit or miss, and an upstream fetch that read
-// Prometheus counters in memory (see counters.go), so it never waits for the
+// fetchBytes bytes, as IncrementUpstreamFetch does. Like the other Increment
-// database or fails.
+// methods, it only adds to the cache's Prometheus counters in memory (see
-func (c *Cache) IncrementStats(hit bool) {
+// counters.go), so it never waits for the database or fails.
+func (c *Cache) IncrementStats(hit bool, fetchBytes int64) {
 	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,8 +19,22 @@ 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} {
+	for _, counter := range []prometheus.Counter{
+		c.cacheHits, c.cacheMisses, c.upstreamFetches, c.upstreamFetchBytes, c.transforms,
+	} {
 		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,6 +25,9 @@ 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 and the image service register their counters on.
+// the cache registers its counters on.
 func Constructors() []any {
 	return []any{
 		newKeys,
@@ -214,12 +214,11 @@ func newProcessor(cfg *config.Config) *imageprocessor.ImageProcessor {
 type imageServiceParams struct {
 	fx.In
 
-	Config          *config.Config
+	Config    *config.Config
-	Cache           *imgcache.Cache
+	Cache     *imgcache.Cache
-	Fetcher         *httpfetcher.HTTPFetcher
+	Fetcher   *httpfetcher.HTTPFetcher
-	Processor       *imageprocessor.ImageProcessor
+	Processor *imageprocessor.ImageProcessor
-	Logger          *logger.Logger
+	Logger    *logger.Logger
-	MetricsRegistry prometheus.Registerer
 }
 
 // newImageService builds the image service from the cache, the fetcher
@@ -233,7 +232,7 @@ func newImageService(
 		Processor:       params.Processor,
 		MaxResponseSize: params.Config.UpstreamMaxResponseSize,
 		Logger:          params.Logger.Get(),
-	}, params.MetricsRegistry)
+	})
 }
 
 // sessionManagerParams defines dependencies for newSessionManager.