Export cache, eviction, upstream and transform metrics (closes #85)
check / check (push) Canceled after 0s

New pixa_ metrics, none with labels:
- source cache hits and misses, counted once per variant made or checked;
  a source whose cached file cannot be used is a miss
- cache usage and limit gauges; usage is set each time eviction reads it
- eviction count and bytes, from the size_bytes the deleted row returns
- upstream fetch duration histogram and fetch error counter
- transform duration histogram and an in-progress gauge, both including
  the wait for a processing slot

Removing a damaged file during reconciliation is not counted as an
eviction. README lists every metric; the metrics config comment now says
the pixa_ metrics are measured whether or not /metrics is served.

Model: opus-5-5
This commit was merged in pull request #353.
This commit is contained in:
sneak committed 2026-11-11 14:32:38 +00:00
1 parent a041e73550
commit fa2ec3da6a
9 files changed
+319 -113

No files matched your search

+105 -33
View File
@@ -37,11 +37,15 @@ type Proxy struct {
// same time share one fetch and one transcode.
variantsInProgress singleflight.Group
// The counts of upstream fetches and transforms, made and registered by
// New. They are kept in memory, so adding to one waits for nothing.
upstreamFetches prometheus.Counter
upstreamFetchBytes prometheus.Counter
transforms prometheus.Counter
// The metrics of upstream fetches and transforms, made and registered by
// New. They are kept in memory, so updating one waits for nothing.
upstreamFetches prometheus.Counter
upstreamFetchBytes prometheus.Counter
upstreamFetchDurations prometheus.Histogram
upstreamFetchErrors prometheus.Counter
transforms prometheus.Counter
transformDurations prometheus.Histogram
transformsInProgress prometheus.Gauge
}
// Config holds the proxy's configuration.
@@ -62,9 +66,9 @@ type Config struct {
// errCacheRequired is returned when the proxy is built without a cache.
var errCacheRequired = errors.New("cache is required")
// New creates a new image proxy and registers its counters on
// New creates a new image proxy and registers its metrics on
// metricsRegistry. pixad passes Prometheus' default registry, the one
// /metrics serves. It fails when metricsRegistry already holds counters of
// /metrics serves. It fails when metricsRegistry already holds metrics of
// the same names.
func New(
cfg *Config, metricsRegistry prometheus.Registerer,
@@ -99,18 +103,39 @@ func New(
Name: "pixa_upstream_fetch_bytes_total",
Help: "Bytes read from upstream by fetches of source images.",
}),
upstreamFetchDurations: prometheus.NewHistogram(prometheus.HistogramOpts{
Name: "pixa_upstream_fetch_duration_seconds",
Help: "Duration of the fetches pixa_upstream_fetches_total counts, " +
"until the response's last byte.",
}),
upstreamFetchErrors: prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_upstream_fetch_errors_total",
Help: "Fetches of a source image from upstream that failed, " +
"before or while reading the response.",
}),
transforms: prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_transforms_total",
Help: "Images the image processor transformed.",
}),
transformDurations: prometheus.NewHistogram(prometheus.HistogramOpts{
Name: "pixa_transform_duration_seconds",
Help: "Duration of the transforms pixa_transforms_total counts, " +
"waiting for a processing slot included.",
}),
transformsInProgress: prometheus.NewGauge(prometheus.GaugeOpts{
Name: "pixa_transforms_in_progress",
Help: "Images being transformed or waiting for a processing slot.",
}),
}
for _, counter := range []prometheus.Counter{
p.upstreamFetches, p.upstreamFetchBytes, p.transforms,
for _, metric := range []prometheus.Collector{
p.upstreamFetches, p.upstreamFetchBytes, p.upstreamFetchDurations,
p.upstreamFetchErrors, p.transforms, p.transformDurations,
p.transformsInProgress,
} {
err = metricsRegistry.Register(counter)
err = metricsRegistry.Register(metric)
if err != nil {
return nil, fmt.Errorf("failed to register the proxy's counters: %w", err)
return nil, fmt.Errorf("failed to register the proxy's metrics: %w", err)
}
}
@@ -341,7 +366,10 @@ func (p *Proxy) loadCachedSource(
}
// processFromSourceOrFetch makes the variant req asks for, from the cached
// source when there is one and from a fetched source otherwise.
// source when there is one and from a fetched source otherwise. It counts a
// source cache miss when it fetches the source because the source has no row
// in the cache or its cached file cannot be used, and a source cache hit
// otherwise.
//
// A cached source last checked with upstream CacheTTL or more ago is checked
// first: upstream is asked to send it only if it changed since. When it
@@ -362,6 +390,8 @@ func (p *Proxy) processFromSourceOrFetch(
p.log.Warn("failed to look up cached source, fetching", "error", err)
}
p.cache.IncrementSourceStats(false)
return p.fetchAndProcess(ctx, req, cacheKey)
}
@@ -375,16 +405,21 @@ func (p *Proxy) processFromSourceOrFetch(
case err != nil:
p.recordFailedCheck(ctx, req, err)
default:
p.cache.IncrementSourceStats(true)
return p.processFetched(ctx, req, cacheKey, fetched)
}
}
variant := p.renewVariant(ctx, cacheKey, cached)
if variant != nil {
p.cache.IncrementSourceStats(true)
return variant, nil
}
source, sourceSize, sourceFormat := p.loadCachedSource(ctx, req.Source)
p.cache.IncrementSourceStats(source != nil)
// Fetch from upstream if we don't have source data or it's empty
if source == nil {
@@ -581,36 +616,32 @@ type fetchedSource struct {
// fetchSource fetches req's source from upstream and stores it in the cache.
// A non-empty ifNoneMatch or ifModifiedSince asks upstream to send the source
// only if it changed since; errSourceNotModified says it did not. It counts
// the fetch with the bytes read from upstream, including when reading the
// response or a later step fails. It fetches nothing while the source URL is
// in the negative cache. The caller closes the returned response.
// the fetch with the bytes read from upstream and its duration, including
// when reading the response or a later step fails, and counts a fetch that
// fails or whose response cannot be read as an upstream fetch error. It
// fetches nothing while the source URL is in the negative cache. The caller
// closes the returned response.
func (p *Proxy) fetchSource(
ctx context.Context,
req *imagerequest.Request,
ifNoneMatch, ifModifiedSince string,
) (*fetchedSource, error) {
// Check negative cache - skip fetching for recently-failed URLs
negHit, err := p.cache.Failure(ctx, req.Source)
err := p.refuseNegativeCached(ctx, req)
if err != nil {
p.log.Warn("negative cache check failed", "error", err)
}
if negHit {
p.log.Debug("negative cache hit",
"host", req.Source.Host,
"path", req.Source.Path,
)
return nil, fmt.Errorf("%w: %w", httpfetcher.ErrUpstreamError, ErrNegativeCached)
return nil, err
}
// Fetch from upstream
p.log.Debug("fetching from upstream", "host", req.Source.Host,
"path", req.Source.Path, "query", req.Source.Query)
fetchStart := time.Now()
fetchResult, err := p.fetcher.Fetch(ctx, req.Source,
ifNoneMatch, ifModifiedSince)
if err != nil {
p.upstreamFetchErrors.Inc()
// Store negative cache for certain errors
if isNegativeCacheable(err) {
p.storeNegative(ctx, req)
@@ -634,9 +665,11 @@ func (p *Proxy) fetchSource(
fetchBytes := int64(len(sourceData))
// Counted also when the request context has ended meanwhile
p.incrementUpstreamFetch(fetchBytes)
p.incrementUpstreamFetch(fetchBytes, time.Since(fetchStart))
if err != nil {
p.upstreamFetchErrors.Inc()
_ = fetchResult.Content.Close()
return nil, fmt.Errorf("failed to read upstream response: %w", err)
@@ -677,6 +710,29 @@ func (p *Proxy) fetchSource(
}, nil
}
// refuseNegativeCached returns an error wrapping ErrNegativeCached while
// req's source URL is in the negative cache, and nil otherwise. A failure to
// read the negative cache is only logged, and returns nil.
func (p *Proxy) refuseNegativeCached(
ctx context.Context, req *imagerequest.Request,
) error {
negHit, err := p.cache.Failure(ctx, req.Source)
if err != nil {
p.log.Warn("negative cache check failed", "error", err)
}
if !negHit {
return nil
}
p.log.Debug("negative cache hit",
"host", req.Source.Host,
"path", req.Source.Path,
)
return fmt.Errorf("%w: %w", httpfetcher.ErrUpstreamError, ErrNegativeCached)
}
// fetchedSourceInfo returns what the cache keeps about a source upstream
// just sent: fetched and checked now, with the ETag and Last-Modified
// headers of fetchResult.
@@ -722,15 +778,16 @@ func (p *Proxy) logUpstreamFetch(
)
}
// incrementUpstreamFetch counts one upstream fetch that read fetchBytes bytes.
// A fetch that read no bytes is not counted.
func (p *Proxy) incrementUpstreamFetch(fetchBytes int64) {
// incrementUpstreamFetch counts one upstream fetch that read fetchBytes bytes
// and took duration. A fetch that read no bytes is not counted.
func (p *Proxy) incrementUpstreamFetch(fetchBytes int64, duration time.Duration) {
if fetchBytes <= 0 {
return
}
p.upstreamFetches.Inc()
p.upstreamFetchBytes.Add(float64(fetchBytes))
p.upstreamFetchDurations.Observe(duration.Seconds())
}
// storeNegative stores a negative cache entry for req's source, whose fetch
@@ -772,8 +829,7 @@ func (p *Proxy) processAndStore(
// ctx has the deadline of the request doing the processing, which can
// end while a request waiting for its image has time left
processResult, err := p.processor.Process(ctx, source, sourceFormat,
&req.Transform)
processResult, err := p.transform(ctx, source, sourceFormat, &req.Transform)
if errors.Is(err, context.DeadlineExceeded) {
return nil, fmt.Errorf("%w: %w", ErrProcessingTimedOut, err)
}
@@ -786,6 +842,7 @@ func (p *Proxy) processAndStore(
// Counted also when the request context has ended meanwhile
p.transforms.Inc()
p.transformDurations.Observe(processDuration.Seconds())
// Read processed content
processedData, err := io.ReadAll(processResult.Content)
@@ -838,6 +895,21 @@ func (p *Proxy) processAndStore(
}, nil
}
// transform has the image processor transform the image read from source
// as requested says, and counts it in pixa_transforms_in_progress until the
// processor returns or panics.
func (p *Proxy) transform(
ctx context.Context,
source io.Reader,
sourceFormat imagerequest.Format,
requested *imagerequest.Transform,
) (*imageprocessor.Result, error) {
p.transformsInProgress.Inc()
defer p.transformsInProgress.Dec()
return p.processor.Process(ctx, source, sourceFormat, requested)
}
// isNegativeCacheable returns true if the error should be cached.
func isNegativeCacheable(err error) bool {
return errors.Is(err, httpfetcher.ErrUpstreamError)
+12 -17
View File
@@ -111,10 +111,16 @@ type Cache struct {
pendingAccessTimesDone chan struct{}
pendingAccessTimesCancel context.CancelFunc
// The counts of cache hits and misses, made and registered by
// registerCounters (see counters.go)
cacheHits prometheus.Counter
cacheMisses prometheus.Counter
// The cache's metrics, made and registered by registerMetrics (see
// metrics.go)
cacheHits prometheus.Counter
cacheMisses prometheus.Counter
sourceCacheHits prometheus.Counter
sourceCacheMisses prometheus.Counter
usageBytes prometheus.Gauge
maxBytes prometheus.Gauge
evictions prometheus.Counter
evictedBytes prometheus.Counter
// reconciliationPageSize is the most rows one read of a content table
// returns in the reconciliation pass. NewCache sets it to
@@ -128,7 +134,7 @@ type Cache struct {
reconciliationReadTestHook func(rows int)
}
// NewCache creates a new cache instance and registers its counters on
// NewCache creates a new cache instance and registers its metrics on
// metricsRegistry. pixad passes Prometheus' default registry, the one
// /metrics serves. It makes no database call.
func NewCache(
@@ -157,7 +163,7 @@ func NewCache(
reconciliationPageSize: defaultReconciliationPageSize,
}
err := c.registerCounters(metricsRegistry)
err := c.registerMetrics(metricsRegistry)
if err != nil {
return nil, err
}
@@ -656,17 +662,6 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
}
}
// 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()
}
}
// touchVariant hands the access time of a variant served from the cache to
// the goroutine StartPendingAccessTimeWrites starts, which writes it in its
// next batch. It never waits: while maxPendingAccessTimes access times of
-31
View File
@@ -1,31 +0,0 @@
package imgcache
import (
"fmt"
"github.com/prometheus/client_golang/prometheus"
)
// registerCounters makes the cache's counters and registers them on
// metricsRegistry. The counters are kept in memory, so adding to one waits
// for nothing and makes no database call. It fails when metricsRegistry
// already holds counters of the same names.
func (c *Cache) registerCounters(metricsRegistry prometheus.Registerer) error {
c.cacheHits = prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_cache_hits_total",
Help: "Image requests served from the cache.",
})
c.cacheMisses = prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_cache_misses_total",
Help: "Image requests not served from the cache.",
})
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)
}
}
return nil
}
+87 -22
View File
@@ -92,8 +92,9 @@ type evictionCandidate struct {
}
// EvictToLimit evicts least-recently-used cache entries until total
// tracked usage is at or below the configured MaxBytes limit. It is a
// no-op when the cache is disabled.
// tracked usage is at or below the configured MaxBytes limit. It sets the
// pixa_cache_usage_bytes gauge to each usage it reads. It is a no-op when
// the cache is disabled.
func (c *Cache) EvictToLimit(ctx context.Context) error {
if c.config.MaxBytes == 0 {
return nil
@@ -105,6 +106,8 @@ func (c *Cache) EvictToLimit(ctx context.Context) error {
return err
}
c.usageBytes.Set(float64(usage))
if usage <= c.config.MaxBytes {
return nil
}
@@ -345,37 +348,95 @@ func (c *Cache) sourceCandidates(ctx context.Context) ([]evictionCandidate, erro
return candidates, nil
}
// evictVariant removes one variant: accounting row first, then the
// file, so the database never references a deleted file.
// evictVariant removes one variant (see removeVariant) and counts the
// eviction with the size its row recorded.
func (c *Cache) evictVariant(
ctx context.Context, cacheKey imagerequest.VariantKey,
) error {
_, err := c.db.ExecContext(ctx,
`DELETE FROM variant_content WHERE cache_key = ?`, string(cacheKey))
if err != nil {
return fmt.Errorf("failed to delete variant accounting row: %w", err)
}
err = c.variants.Delete(string(cacheKey))
sizeBytes, err := c.removeVariant(ctx, cacheKey)
if err != nil {
return err
}
c.countEviction(sizeBytes)
return nil
}
// evictSourceBlob removes one source: its row first, then its file, as
// evictVariant does for a variant. It holds no lock: a store of the same
// source in between can leave the source's row with no file, which
// Source treats as a miss.
func (c *Cache) evictSourceBlob(ctx context.Context, key imagerequest.SourceKey) error {
_, err := c.db.ExecContext(ctx,
`DELETE FROM source_content WHERE source_key = ?`, string(key))
// removeVariant removes one variant: accounting row first, then the
// file, so the database never references a deleted file. It returns the
// size the row recorded, or 0 when the variant had no row.
func (c *Cache) removeVariant(
ctx context.Context, cacheKey imagerequest.VariantKey,
) (int64, error) {
sizeBytes, err := c.deleteRowReturningSize(ctx,
`DELETE FROM variant_content WHERE cache_key = ? RETURNING size_bytes`,
string(cacheKey))
if err != nil {
return fmt.Errorf("failed to delete source content row: %w", err)
return 0, fmt.Errorf("failed to delete variant accounting row: %w", err)
}
return c.srcContent.Delete(string(key))
err = c.variants.Delete(string(cacheKey))
if err != nil {
return 0, err
}
return sizeBytes, nil
}
// evictSourceBlob removes one source (see removeSourceBlob) and counts the
// eviction with the size its row recorded.
func (c *Cache) evictSourceBlob(ctx context.Context, key imagerequest.SourceKey) error {
sizeBytes, err := c.removeSourceBlob(ctx, key)
if err != nil {
return err
}
c.countEviction(sizeBytes)
return nil
}
// removeSourceBlob removes one source: its row first, then its file, as
// removeVariant does for a variant, and returns the size the row recorded,
// or 0 when the source had no row. It holds no lock: a store of the same
// source in between can leave the source's row with no file, which Source
// treats as a miss.
func (c *Cache) removeSourceBlob(
ctx context.Context, key imagerequest.SourceKey,
) (int64, error) {
sizeBytes, err := c.deleteRowReturningSize(ctx,
`DELETE FROM source_content WHERE source_key = ? RETURNING size_bytes`,
string(key))
if err != nil {
return 0, fmt.Errorf("failed to delete source content row: %w", err)
}
err = c.srcContent.Delete(string(key))
if err != nil {
return 0, err
}
return sizeBytes, nil
}
// deleteRowReturningSize runs deleteQuery, which deletes the row of key and
// returns its size_bytes, and returns that size, or 0 when there was no row.
func (c *Cache) deleteRowReturningSize(
ctx context.Context, deleteQuery string, key string,
) (int64, error) {
var sizeBytes int64
err := c.db.QueryRowContext(ctx, deleteQuery, key).Scan(&sizeBytes)
if errors.Is(err, sql.ErrNoRows) {
return 0, nil
}
if err != nil {
return 0, err
}
return sizeBytes, nil
}
// notifyWritePressure wakes the background evictor after a store, so
@@ -673,7 +734,9 @@ func (c *Cache) reconcileVariantFile(
c.log.Warn("removing damaged cached variant file", "path", path,
"file_bytes", info.Size(), "recorded_bytes", recordedSize)
return c.evictVariant(ctx, cacheKey)
_, err = c.removeVariant(ctx, cacheKey)
return err
}
// adoptVariantFile inserts an accounting row for a variant file that
@@ -835,7 +898,9 @@ func (c *Cache) reconcileSourceFile(
c.log.Warn("removing damaged cached source file", "path", path,
"file_bytes", info.Size(), "recorded_bytes", recordedSize)
return c.evictSourceBlob(ctx, key)
_, err = c.removeSourceBlob(ctx, key)
return err
}
// removeUntrackedSourceFile deletes a source blob file that has no
+88
View File
@@ -0,0 +1,88 @@
package imgcache
import (
"fmt"
"github.com/prometheus/client_golang/prometheus"
)
// registerMetrics makes the cache's metrics and registers them on
// metricsRegistry. The metrics are kept in memory, so updating one waits
// for nothing and makes no database call. It fails when metricsRegistry
// already holds metrics of the same names.
func (c *Cache) registerMetrics(metricsRegistry prometheus.Registerer) error {
c.cacheHits = prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_cache_hits_total",
Help: "Image requests served from the cache.",
})
c.cacheMisses = prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_cache_misses_total",
Help: "Image requests not served from the cache.",
})
c.sourceCacheHits = prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_source_cache_hits_total",
Help: "Variants made or checked whose source was in the cache.",
})
c.sourceCacheMisses = prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_source_cache_misses_total",
Help: "Variants made or checked whose source was not in the cache.",
})
c.usageBytes = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "pixa_cache_usage_bytes",
Help: "Bytes of sources and variants in the cache, as eviction last read them.",
})
c.maxBytes = prometheus.NewGauge(prometheus.GaugeOpts{
Name: "pixa_cache_max_bytes",
Help: "The disk cache limit in bytes, cache_max_bytes.",
})
c.evictions = prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_cache_evictions_total",
Help: "Sources and variants removed by eviction.",
})
c.evictedBytes = prometheus.NewCounter(prometheus.CounterOpts{
Name: "pixa_cache_evicted_bytes_total",
Help: "Bytes of the sources and variants removed by eviction.",
})
c.maxBytes.Set(float64(c.config.MaxBytes))
for _, metric := range []prometheus.Collector{
c.cacheHits, c.cacheMisses, c.sourceCacheHits, c.sourceCacheMisses,
c.usageBytes, c.maxBytes, c.evictions, c.evictedBytes,
} {
err := metricsRegistry.Register(metric)
if err != nil {
return fmt.Errorf("failed to register the cache's metrics: %w", err)
}
}
return nil
}
// IncrementStats counts a cache hit or miss of an image request. It only
// adds to the cache's counters in memory, so it never waits for the
// database or fails.
func (c *Cache) IncrementStats(hit bool) {
if hit {
c.cacheHits.Inc()
} else {
c.cacheMisses.Inc()
}
}
// IncrementSourceStats counts whether a variant being made or checked found
// its source in the cache. Like IncrementStats, it never waits or fails.
func (c *Cache) IncrementSourceStats(hit bool) {
if hit {
c.sourceCacheHits.Inc()
} else {
c.sourceCacheMisses.Inc()
}
}
// countEviction counts one source or variant of sizeBytes bytes removed by
// eviction.
func (c *Cache) countEviction(sizeBytes int64) {
c.evictions.Inc()
c.evictedBytes.Add(float64(sizeBytes))
}
+1 -1
View File
@@ -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 and the image service register their metrics on.
func Constructors() []any {
return []any{
newKeys,