Count hits, misses, fetches and transforms as Prometheus counters (closes #311) #328
@@ -88,14 +88,17 @@ A load balancer's health check can request `/.well-known/healthcheck.json`,
|
||||
which answers 200 whenever pixa is running, in maintenance mode too (see
|
||||
`maintenance_mode`).
|
||||
|
||||
On SIGTERM or SIGINT pixa stops accepting connections, gives the requests in
|
||||
progress and the images being processed 5 seconds to finish, then writes to the
|
||||
database the counts of cache hits, misses, fetches and conversions and the times
|
||||
cached images were last used that are not written yet (see Storage), and exits:
|
||||
with 0, or with 1 when images were still being processed after those 5 seconds,
|
||||
some of those counts or times could not be written, or another part of pixa
|
||||
failed to stop. A request not finished by then is cut off. `docker stop` waits
|
||||
10 seconds before it kills the container.
|
||||
On SIGTERM or SIGINT pixa stops accepting connections. The requests in progress
|
||||
and the images being processed get 5 seconds to finish. A request not finished
|
||||
by then is cut off. pixa then writes the waiting last-used times of cached
|
||||
images to the database (see Storage) and exits. It exits with 1 when:
|
||||
|
||||
- images were still being processed after those 5 seconds,
|
||||
- some of those times could not be written, or
|
||||
- another part of pixa failed to stop.
|
||||
|
||||
Otherwise it exits with 0. `docker stop` waits 10 seconds before it kills the
|
||||
container.
|
||||
|
||||
Outside Docker, pixa needs libvips (the image has 8.16) and libheif to run, as
|
||||
it uses libvips through CGO. pixad does not start unless libvips has its JPEG XL
|
||||
@@ -209,14 +212,14 @@ Test). Toward it, a cache hit opens the transformed image's file and nothing
|
||||
else, and makes no database call. The content type comes from the file's first
|
||||
bytes, which are those of the format pixa wrote it in; a file whose first bytes
|
||||
are not those of a format pixa writes is not served, and the image is converted
|
||||
again. The time the image was last served, which eviction orders by, and the hit
|
||||
count are handed to a background writer without waiting for it, as are the
|
||||
counts of misses, fetches and conversions, and the time a miss last used a
|
||||
cached source image. Once a second that writer writes the waiting times in one
|
||||
transaction and the counts in one statement. While 10,000 times of transformed
|
||||
images, or 10,000 of source images, wait to be written, a further one is
|
||||
dropped, and that image only looks less recently used to eviction.
|
||||
`script/bench` measures how many hits pixa serves per second (see Load Test).
|
||||
again. A hit hands the time it was served to a background writer and does not
|
||||
wait for the writer. Eviction orders by that time. The hit is counted in memory.
|
||||
A miss that uses a cached source image also hands that writer the time it used
|
||||
the image. Once a second that writer writes the waiting times in one
|
||||
transaction. While 10,000 times of transformed images, or 10,000 of source
|
||||
images, wait to be written, a further one is dropped, and that image only looks
|
||||
less recently used to eviction. `script/bench` measures how many hits pixa
|
||||
serves per second (see Load Test).
|
||||
|
||||
### Routes
|
||||
|
||||
@@ -814,15 +817,27 @@ See `configs/config.example.yml` for all options with defaults.
|
||||
- **Image processing**: govips (CGO wrapper for libvips)
|
||||
- **Database**: SQLite via modernc.org/sqlite
|
||||
- **Static assets**: embedded via `//go:embed`
|
||||
- **Metrics**: Prometheus, at `/metrics`: generic HTTP request metrics
|
||||
(duration, response size, requests in flight) and the Go runtime and process
|
||||
metrics; requests are measured and `/metrics` is served only when
|
||||
`metrics.username` and `metrics.password` are set. A request's `handler` label
|
||||
is the pattern of the route it matches, such as `/v1/image/*`, never its path.
|
||||
Requests that match no route share the label `unmatched`: those the router
|
||||
answers 404 or 405, and CORS preflights. A request's `method` label is its
|
||||
method when that is `GET`, `HEAD`, `POST`, `PUT`, `PATCH`, `DELETE`,
|
||||
`OPTIONS`, `CONNECT` or `TRACE`, and `other` for any other method
|
||||
- **Metrics**: Prometheus, at `/metrics`. pixa serves `/metrics` and measures
|
||||
HTTP requests only when `metrics.username` and `metrics.password` are set.
|
||||
`/metrics` holds:
|
||||
- the duration of HTTP requests
|
||||
- the size of HTTP responses
|
||||
- the HTTP requests in flight
|
||||
- the Go runtime and process metrics
|
||||
- `pixa_cache_hits_total`: image requests served from the cache
|
||||
- `pixa_cache_misses_total`: image requests not served from the cache
|
||||
- `pixa_upstream_fetches_total`: fetches of a source image that read any
|
||||
bytes
|
||||
- `pixa_upstream_fetch_bytes_total`: the bytes those fetches read
|
||||
- `pixa_transforms_total`: images converted
|
||||
|
||||
A request's `handler` label is the pattern of the route it matches, such as
|
||||
`/v1/image/*`, never its path. Requests that match no route share the label
|
||||
`unmatched`: those the router answers 404 or 405, and CORS preflights. A
|
||||
request's `method` label is its method when that is `GET`, `HEAD`, `POST`,
|
||||
`PUT`, `PATCH`, `DELETE`, `OPTIONS`, `CONNECT` or `TRACE`, and `other` for
|
||||
any other method.
|
||||
|
||||
- **Logging**: stdlib slog
|
||||
|
||||
## Entrypoints
|
||||
|
||||
@@ -30,6 +30,18 @@ P2: security: per-origin rate limiting
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-26 cache hits, misses, upstream fetches, the bytes those fetches read
|
||||
and transforms are counted as Prometheus counters in memory (closes #311).
|
||||
They are `pixa_cache_hits_total`, `pixa_cache_misses_total`,
|
||||
`pixa_upstream_fetches_total`, `pixa_upstream_fetch_bytes_total` and
|
||||
`pixa_transforms_total`, fields of the cache. `NewCache` makes them and
|
||||
registers them on the registry it is given (`internal/imgcache/counters.go`).
|
||||
pixad gives it Prometheus' default registry, which `/metrics` serves; each
|
||||
test gives it a new one. The `Increment` methods add to their cache's
|
||||
counters. `cache_stats`, `Cache.Stats`, `Service.Stats`, `CacheStats` and the
|
||||
pending counts are gone. The background goroutine writes only access times
|
||||
now: `StartPendingAccessTimeWrites` starts it and
|
||||
`StopPendingAccessTimeWrites` stops it, in `pendingaccesstimes.go`.
|
||||
- 2026-10-21 the schema file declares only the tables and columns pixa writes
|
||||
and reads (closes #310): `output_content` and `request_cache` are gone, and so
|
||||
are the columns nothing read: the host, path, query, `content_type` and
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/spf13/cobra"
|
||||
"go.uber.org/fx"
|
||||
"sneak.berlin/go/pixa/internal/config"
|
||||
@@ -61,6 +62,8 @@ func run(_ *cobra.Command, _ []string) {
|
||||
server.New,
|
||||
middleware.New,
|
||||
healthcheck.New,
|
||||
// The registry /metrics serves
|
||||
func() prometheus.Registerer { return prometheus.DefaultRegisterer },
|
||||
),
|
||||
fx.Invoke(
|
||||
func(log *logger.Logger) { log.Identify() },
|
||||
|
||||
@@ -104,33 +104,24 @@ func writeConcurrently(t *testing.T, db *sql.DB) {
|
||||
|
||||
wg.Wait()
|
||||
|
||||
var hits, sources int
|
||||
var sources int
|
||||
|
||||
err := db.QueryRowContext(ctx, `
|
||||
SELECT hit_count, (SELECT COUNT(*) FROM source_content)
|
||||
FROM cache_stats WHERE id = 1
|
||||
`).Scan(&hits, &sources)
|
||||
err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM source_content`).
|
||||
Scan(&sources)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to count the writes: %v", err)
|
||||
}
|
||||
|
||||
if hits != totalRequests || sources != totalRequests {
|
||||
t.Errorf("hit_count = %d and %d source_content rows, want %d of each",
|
||||
hits, sources, totalRequests)
|
||||
if sources != totalRequests {
|
||||
t.Errorf("%d source_content rows, want %d", sources, totalRequests)
|
||||
}
|
||||
}
|
||||
|
||||
// writeLikeOneRequest makes the writes one request and the eviction pass
|
||||
// make: it counts a cache hit, stores a source, records a transformed image
|
||||
// and deletes that record again.
|
||||
// make: it stores a source, records a transformed image and deletes that
|
||||
// record again.
|
||||
func writeLikeOneRequest(ctx context.Context, db *sql.DB, key string) error {
|
||||
_, err := db.ExecContext(ctx,
|
||||
`UPDATE cache_stats SET hit_count = hit_count + 1 WHERE id = 1`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("counting a cache hit: %w", err)
|
||||
}
|
||||
|
||||
_, err = db.ExecContext(ctx, `INSERT INTO source_content
|
||||
_, err := db.ExecContext(ctx, `INSERT INTO source_content
|
||||
(source_key, size_bytes) VALUES (?, 1)`, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("storing a source: %w", err)
|
||||
|
||||
@@ -159,7 +159,7 @@ func TestApplyMigrations_CreatesSchemaAndTables(t *testing.T) {
|
||||
}
|
||||
|
||||
// Verify that the application tables created by 001.sql exist.
|
||||
tables := []string{"source_content", "negative_cache", "cache_stats"}
|
||||
tables := []string{"source_content", "negative_cache"}
|
||||
for _, table := range tables {
|
||||
var count int
|
||||
|
||||
|
||||
@@ -108,18 +108,6 @@ CREATE TABLE IF NOT EXISTS negative_cache (
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_negative_cache_expires ON negative_cache(expires_at);
|
||||
|
||||
-- Cache statistics for monitoring
|
||||
CREATE TABLE IF NOT EXISTS cache_stats (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
hit_count INTEGER NOT NULL DEFAULT 0,
|
||||
miss_count INTEGER NOT NULL DEFAULT 0,
|
||||
upstream_fetch_count INTEGER NOT NULL DEFAULT 0,
|
||||
upstream_fetch_bytes INTEGER NOT NULL DEFAULT 0,
|
||||
transform_count INTEGER NOT NULL DEFAULT 0,
|
||||
last_updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
INSERT OR IGNORE INTO cache_stats (id) VALUES (1);
|
||||
|
||||
-- A random value that a login session carries from its login. pixa makes it
|
||||
-- when the table has no row, and each logout replaces it; a login session
|
||||
-- carrying another value is refused, so logging out ends every login session.
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/database"
|
||||
"sneak.berlin/go/pixa/internal/encurl"
|
||||
@@ -101,9 +102,8 @@ func newCountingDatabase(tb testing.TB) (*sql.DB, *atomic.Int64) {
|
||||
// newCountingDatabaseServer returns a router for both image routes whose
|
||||
// image cache uses newCountingDatabase, the count of the statements run on
|
||||
// that database, and a URL on each route for a different image of
|
||||
// photoPath on countingHost. Nothing writes the cache's pending counts and
|
||||
// access times in the background, so every statement counted comes from a
|
||||
// request.
|
||||
// photoPath on countingHost. Nothing writes the cache's access times in the
|
||||
// background, so every statement counted comes from a request.
|
||||
func newCountingDatabaseServer(
|
||||
tb testing.TB,
|
||||
) (http.Handler, *atomic.Int64, []string) {
|
||||
@@ -121,7 +121,7 @@ func newCountingDatabaseServer(
|
||||
cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
|
||||
StateDir: tb.TempDir(),
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
tb.Fatalf("imgcache.NewCache() error = %v", err)
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"go.uber.org/fx/fxtest"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/config"
|
||||
@@ -127,7 +128,10 @@ func TestDiskCacheOffOnlyForExplicitZeroCacheMaxBytes(t *testing.T) {
|
||||
t.Fatalf("database.New() error = %v", err)
|
||||
}
|
||||
|
||||
_, err = New(lc, Params{Logger: log, Database: db, Config: cfg})
|
||||
_, err = New(lc, Params{
|
||||
Logger: log, Database: db, Config: cfg,
|
||||
MetricsRegistry: prometheus.NewRegistry(),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("New() error = %v", err)
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"go.uber.org/fx"
|
||||
"go.uber.org/fx/fxtest"
|
||||
|
||||
@@ -47,7 +48,8 @@ func TestHandlersBuildTheirOwnFetcherWhenNoneIsProvided(t *testing.T) {
|
||||
|
||||
app := fxtest.New(t,
|
||||
fx.Supply(cfg),
|
||||
fx.Provide(globals.New, logger.New, database.New, healthcheck.New, New),
|
||||
fx.Provide(globals.New, logger.New, database.New, healthcheck.New, New,
|
||||
func() prometheus.Registerer { return prometheus.NewRegistry() }),
|
||||
fx.Populate(&h),
|
||||
)
|
||||
app.RequireStart()
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"sneak.berlin/go/pixa/internal/encurl"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
@@ -35,7 +36,7 @@ func newRoutesFetchingThrough(
|
||||
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("imgcache.NewCache() error = %v", err)
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"go.uber.org/fx"
|
||||
"sneak.berlin/go/pixa/internal/allowlist"
|
||||
"sneak.berlin/go/pixa/internal/config"
|
||||
@@ -30,6 +31,11 @@ type Params struct {
|
||||
Database *database.Database
|
||||
Config *config.Config
|
||||
|
||||
// MetricsRegistry is the registry the image cache registers its
|
||||
// counters on. pixad provides Prometheus' default registry, the one
|
||||
// /metrics serves.
|
||||
MetricsRegistry prometheus.Registerer
|
||||
|
||||
// Fetcher, when provided, fetches upstream images in place of the
|
||||
// fetcher the handlers build from the config. Only tests provide one;
|
||||
// pixad does not.
|
||||
@@ -52,6 +58,10 @@ type Handlers struct {
|
||||
// refererBlocklist matches the hosts of referer_blocklist; its IsAllowed
|
||||
// reports whether a URL's host is on that list.
|
||||
refererBlocklist *allowlist.HostAllowList
|
||||
|
||||
// metricsRegistry is the registry the image cache registers its
|
||||
// counters on.
|
||||
metricsRegistry prometheus.Registerer
|
||||
}
|
||||
|
||||
// New creates a new Handlers instance.
|
||||
@@ -69,6 +79,7 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
|
||||
fetcher: params.Fetcher,
|
||||
csrfProtect: csrfProtect,
|
||||
refererBlocklist: allowlist.New(params.Config.RefererBlocklist),
|
||||
metricsRegistry: params.MetricsRegistry,
|
||||
}
|
||||
|
||||
lc.Append(fx.Hook{
|
||||
@@ -76,8 +87,8 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
|
||||
OnStart: func(_ context.Context) error {
|
||||
return s.initImageService()
|
||||
},
|
||||
// The pending counts and access times are written here, before the
|
||||
// database's stop hook closes it.
|
||||
// The pending access times are written here, before the database's
|
||||
// stop hook closes it.
|
||||
OnStop: func(ctx context.Context) error {
|
||||
if s.imgCache == nil {
|
||||
return nil
|
||||
@@ -85,7 +96,7 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) {
|
||||
|
||||
return errors.Join(
|
||||
s.imgCache.StopEviction(ctx),
|
||||
s.imgCache.StopPendingCountWrites(ctx),
|
||||
s.imgCache.StopPendingAccessTimeWrites(ctx),
|
||||
)
|
||||
},
|
||||
})
|
||||
@@ -116,7 +127,8 @@ func newCacheConfig(cfg *config.Config, log *slog.Logger) imgcache.CacheConfig {
|
||||
|
||||
// initImageService initializes the image cache and service.
|
||||
func (s *Handlers) initImageService() error {
|
||||
cache, err := imgcache.NewCache(s.db.DB(), newCacheConfig(s.config, s.log))
|
||||
cache, err := imgcache.NewCache(
|
||||
s.db.DB(), newCacheConfig(s.config, s.log), s.metricsRegistry)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -128,9 +140,9 @@ func (s *Handlers) initImageService() error {
|
||||
// rows, which is all it does when the disk cache is disabled.
|
||||
cache.StartEviction(imgcache.DefaultEvictionInterval)
|
||||
|
||||
// Writes the counts and the access times of cache hits that requests
|
||||
// hand over, so that no request waits for the database to record them
|
||||
cache.StartPendingCountWrites()
|
||||
// Writes the access times that requests hand over, so that no request
|
||||
// waits for the database to record them
|
||||
cache.StartPendingAccessTimeWrites()
|
||||
|
||||
// Create the fetcher config
|
||||
fetcherCfg := httpfetcher.DefaultConfig()
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"sneak.berlin/go/pixa/internal/database"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imgcache"
|
||||
@@ -48,7 +49,7 @@ func setupTestHandler(t *testing.T) *testFixtures {
|
||||
cache, err := imgcache.NewCache(db, imgcache.CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/encurl"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
@@ -33,7 +34,7 @@ func newSignedHostServer(
|
||||
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("imgcache.NewCache() error = %v", err)
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
"sneak.berlin/go/pixa/internal/imgcache"
|
||||
@@ -31,7 +32,7 @@ func newImageRoute(t *testing.T, fetcher httpfetcher.Fetcher) http.Handler {
|
||||
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"sneak.berlin/go/pixa/internal/allowlist"
|
||||
"sneak.berlin/go/pixa/internal/encurl"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
@@ -50,7 +51,7 @@ func newRefererRoutes(t *testing.T) (http.Handler, *Handlers, *countingFetcher)
|
||||
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("imgcache.NewCache() error = %v", err)
|
||||
}
|
||||
|
||||
+49
-96
@@ -8,10 +8,9 @@ import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
@@ -81,33 +80,30 @@ type Cache struct {
|
||||
evictionDone chan struct{}
|
||||
evictionCancel context.CancelFunc
|
||||
|
||||
// The pending counts: hits, misses, upstream fetches and transforms
|
||||
// that requests counted, kept here until they are written by the
|
||||
// goroutine that StartPendingCountWrites starts, so that no request
|
||||
// waits for the database to count it. pendingVariantAccessTimes holds
|
||||
// the variants served from the cache, and pendingSourceAccessTimes the
|
||||
// cached sources a miss used, whose access time is still to be written
|
||||
// by that goroutine. As for eviction, the channels are created in
|
||||
// pendingVariantAccessTimes holds the variants served from the cache,
|
||||
// and pendingSourceAccessTimes the cached sources a miss used, whose
|
||||
// access time is still to be written by the goroutine that
|
||||
// StartPendingAccessTimeWrites starts, so that no request waits for the
|
||||
// database to record it. As for eviction, the channels are created in
|
||||
// NewCache, so that requests can hand over access times before that
|
||||
// goroutine starts; pendingCountsDone is closed when it returns.
|
||||
// pendingCountsCancel, set by StartPendingCountWrites, cancels its
|
||||
// context, which tells it to finish.
|
||||
pendingHits atomic.Int64
|
||||
pendingMisses atomic.Int64
|
||||
pendingUpstreamFetches atomic.Int64
|
||||
pendingUpstreamFetchBytes atomic.Int64
|
||||
pendingTransforms atomic.Int64
|
||||
// goroutine starts; pendingAccessTimesDone is closed when it returns.
|
||||
// pendingAccessTimesCancel, set by StartPendingAccessTimeWrites,
|
||||
// cancels its context, which tells it to finish.
|
||||
pendingVariantAccessTimes chan imagerequest.VariantKey
|
||||
pendingSourceAccessTimes chan imagerequest.SourceKey
|
||||
pendingCountsDone chan struct{}
|
||||
pendingCountsCancel context.CancelFunc
|
||||
pendingAccessTimesDone chan struct{}
|
||||
pendingAccessTimesCancel context.CancelFunc
|
||||
|
||||
// Held through each write of the pending counts, and by
|
||||
// readHitAndMissCounts to read them
|
||||
pendingCountsWriteMutex sync.Mutex
|
||||
// The counts of what the cache does, made and registered by
|
||||
// registerCounters (see counters.go)
|
||||
cacheHits 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 and in Stats. newCache sets it to
|
||||
// returns in the reconciliation pass. newCache sets it to
|
||||
// defaultReconciliationPageSize; tests set it smaller.
|
||||
reconciliationPageSize int
|
||||
|
||||
@@ -118,15 +114,22 @@ type Cache struct {
|
||||
reconciliationReadTestHook func(rows int)
|
||||
}
|
||||
|
||||
// NewCache creates a new cache instance.
|
||||
func NewCache(db *sql.DB, config CacheConfig) (*Cache, error) {
|
||||
return newCache(db, config, defaultFreeSpaceProbe)
|
||||
// NewCache creates a new cache instance and registers its counters on
|
||||
// metricsRegistry. pixad passes Prometheus' default registry, the one
|
||||
// /metrics serves.
|
||||
func NewCache(
|
||||
db *sql.DB, config CacheConfig, metricsRegistry prometheus.Registerer,
|
||||
) (*Cache, error) {
|
||||
return newCache(db, config, metricsRegistry, defaultFreeSpaceProbe)
|
||||
}
|
||||
|
||||
// newCache is NewCache with the free-space probe passed in, so tests
|
||||
// can fake the free space the default limit is worked out from.
|
||||
func newCache(
|
||||
db *sql.DB, config CacheConfig, probe FreeSpaceProbeFunc,
|
||||
db *sql.DB,
|
||||
config CacheConfig,
|
||||
metricsRegistry prometheus.Registerer,
|
||||
probe FreeSpaceProbeFunc,
|
||||
) (*Cache, error) {
|
||||
log := config.Logger
|
||||
if log == nil {
|
||||
@@ -143,11 +146,16 @@ func newCache(
|
||||
|
||||
pendingVariantAccessTimes: make(chan imagerequest.VariantKey, maxPendingAccessTimes),
|
||||
pendingSourceAccessTimes: make(chan imagerequest.SourceKey, maxPendingAccessTimes),
|
||||
pendingCountsDone: make(chan struct{}),
|
||||
pendingAccessTimesDone: make(chan struct{}),
|
||||
|
||||
reconciliationPageSize: defaultReconciliationPageSize,
|
||||
}
|
||||
|
||||
err := c.registerCounters(metricsRegistry)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if c.disabled {
|
||||
return c, nil
|
||||
}
|
||||
@@ -373,51 +381,15 @@ func (c *Cache) CleanExpired(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
|
||||
// Stats returns cache statistics.
|
||||
func (c *Cache) Stats(ctx context.Context) (*CacheStats, error) {
|
||||
var stats CacheStats
|
||||
|
||||
err := c.readHitAndMissCounts(ctx, &stats)
|
||||
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, fmt.Errorf("failed to get cache stats: %w", err)
|
||||
}
|
||||
|
||||
// Count and size the cached source images and processed variants from
|
||||
// their tables. A disabled cache holds none, whatever rows an earlier
|
||||
// run left.
|
||||
if !c.disabled {
|
||||
err = c.db.QueryRowContext(ctx, `
|
||||
SELECT (SELECT COUNT(*) FROM source_content)
|
||||
+ (SELECT COUNT(*) FROM variant_content)
|
||||
`).Scan(&stats.TotalItems)
|
||||
if err != nil {
|
||||
c.log.Warn("failed to count cache items for stats", "error", err)
|
||||
}
|
||||
|
||||
stats.TotalSizeBytes, err = c.sumContentSizeBytes(ctx)
|
||||
if err != nil {
|
||||
c.log.Warn("failed to sum cache size for stats", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Compute hit rate as a ratio
|
||||
if stats.HitCount+stats.MissCount > 0 {
|
||||
stats.HitRate = float64(stats.HitCount) / float64(stats.HitCount+stats.MissCount)
|
||||
}
|
||||
|
||||
return &stats, nil
|
||||
}
|
||||
|
||||
// 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 pending counts, which the goroutine
|
||||
// StartPendingCountWrites starts writes to the database, so it never waits
|
||||
// for the database or fails.
|
||||
// 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) {
|
||||
if hit {
|
||||
c.pendingHits.Add(1)
|
||||
c.cacheHits.Inc()
|
||||
} else {
|
||||
c.pendingMisses.Add(1)
|
||||
c.cacheMisses.Inc()
|
||||
}
|
||||
|
||||
c.IncrementUpstreamFetch(fetchBytes)
|
||||
@@ -430,37 +402,18 @@ func (c *Cache) IncrementUpstreamFetch(fetchBytes int64) {
|
||||
return
|
||||
}
|
||||
|
||||
c.pendingUpstreamFetches.Add(1)
|
||||
c.pendingUpstreamFetchBytes.Add(fetchBytes)
|
||||
c.upstreamFetches.Inc()
|
||||
c.upstreamFetchBytes.Add(float64(fetchBytes))
|
||||
}
|
||||
|
||||
// IncrementTransformCount counts one image transcoded by the image processor.
|
||||
func (c *Cache) IncrementTransformCount() {
|
||||
c.pendingTransforms.Add(1)
|
||||
}
|
||||
|
||||
// readHitAndMissCounts sets the hit and miss counts of stats to those of the
|
||||
// cache_stats row plus the pending hits and misses (see
|
||||
// StartPendingCountWrites). It holds pendingCountsWriteMutex, so that no
|
||||
// write of the pending counts falls between the two reads.
|
||||
func (c *Cache) readHitAndMissCounts(ctx context.Context, stats *CacheStats) error {
|
||||
c.pendingCountsWriteMutex.Lock()
|
||||
defer c.pendingCountsWriteMutex.Unlock()
|
||||
|
||||
err := c.db.QueryRowContext(ctx, `
|
||||
SELECT hit_count, miss_count
|
||||
FROM cache_stats WHERE id = 1
|
||||
`).Scan(&stats.HitCount, &stats.MissCount)
|
||||
|
||||
stats.HitCount += c.pendingHits.Load()
|
||||
stats.MissCount += c.pendingMisses.Load()
|
||||
|
||||
return err
|
||||
c.transforms.Inc()
|
||||
}
|
||||
|
||||
// touchVariant hands the access time of a variant served from the cache to
|
||||
// the goroutine StartPendingCountWrites starts, which writes it in its next
|
||||
// batch. It never waits: while maxPendingAccessTimes access times of
|
||||
// the goroutine StartPendingAccessTimeWrites starts, which writes it in its
|
||||
// next batch. It never waits: while maxPendingAccessTimes access times of
|
||||
// variants wait to be written, this one is dropped, and the variant only
|
||||
// looks colder to eviction.
|
||||
func (c *Cache) touchVariant(cacheKey imagerequest.VariantKey) {
|
||||
@@ -471,10 +424,10 @@ func (c *Cache) touchVariant(cacheKey imagerequest.VariantKey) {
|
||||
}
|
||||
|
||||
// touchSourceContent hands the access time of a cached source a miss used to
|
||||
// the goroutine StartPendingCountWrites starts, which writes it in its next
|
||||
// batch. It never waits: while maxPendingAccessTimes access times of sources
|
||||
// wait to be written, this one is dropped, and the source only looks colder
|
||||
// to eviction.
|
||||
// the goroutine StartPendingAccessTimeWrites starts, which writes it in its
|
||||
// next batch. It never waits: while maxPendingAccessTimes access times of
|
||||
// sources wait to be written, this one is dropped, and the source only looks
|
||||
// colder to eviction.
|
||||
func (c *Cache) touchSourceContent(key imagerequest.SourceKey) {
|
||||
select {
|
||||
case c.pendingSourceAccessTimes <- key:
|
||||
|
||||
@@ -167,8 +167,8 @@ func TestService_Get_MakesAgainVariantOfNoOutputFormat(t *testing.T) {
|
||||
|
||||
// TestService_Get_HitWaitsForNoDatabaseWrite holds the database's one
|
||||
// connection while a cached variant is requested. The hit must be answered
|
||||
// while the connection is still held, and its access time and hit count
|
||||
// must reach the database once it is free.
|
||||
// while the connection is still held, and its access time must reach the
|
||||
// database once it is free.
|
||||
func TestService_Get_HitWaitsForNoDatabaseWrite(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -177,14 +177,18 @@ func TestService_Get_HitWaitsForNoDatabaseWrite(t *testing.T) {
|
||||
key := imagerequest.CacheKey(req)
|
||||
|
||||
getImage(t, svc, req)
|
||||
waitForPendingCountsWritten(t, svc.cache)
|
||||
|
||||
// Each connection to an in-memory SQLite database opens a new, empty one,
|
||||
// so the goroutine that writes the access times must share the
|
||||
// connection that holds the schema.
|
||||
svc.cache.db.SetMaxOpenConns(1)
|
||||
svc.cache.StartPendingAccessTimeWrites()
|
||||
|
||||
defer func() { _ = svc.cache.StopPendingAccessTimeWrites(t.Context()) }()
|
||||
|
||||
anHourAgo := time.Now().Add(-time.Hour)
|
||||
setVariantLastAccessed(t, svc.cache, key, anHourAgo)
|
||||
|
||||
want := readCacheStatsCounters(t, svc.cache)
|
||||
want.hitCount++
|
||||
|
||||
releaseDatabase := holdDatabase(t, svc.cache)
|
||||
defer releaseDatabase()
|
||||
|
||||
@@ -195,15 +199,13 @@ func TestService_Get_HitWaitsForNoDatabaseWrite(t *testing.T) {
|
||||
if got.err != nil {
|
||||
t.Fatalf("Get() of the cached variant error = %v", got.err)
|
||||
}
|
||||
case <-time.After(pendingCountsWait):
|
||||
case <-time.After(pendingAccessTimesWait):
|
||||
t.Fatal("the cache hit waited for the database")
|
||||
}
|
||||
|
||||
releaseDatabase()
|
||||
|
||||
waitForCounters(t, svc.cache, want)
|
||||
|
||||
deadline := time.Now().Add(pendingCountsWait)
|
||||
deadline := time.Now().Add(pendingAccessTimesWait)
|
||||
|
||||
for countRows(t, svc.cache, `
|
||||
SELECT COUNT(*) FROM variant_content
|
||||
@@ -239,7 +241,7 @@ func TestTouchVariantAndTouchSourceContent_DropAccessTimesPastTheLimit(t *testin
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(pendingCountsWait):
|
||||
case <-time.After(pendingAccessTimesWait):
|
||||
t.Fatal("touchVariant or touchSourceContent waited once the access " +
|
||||
"times reached the limit")
|
||||
}
|
||||
@@ -255,11 +257,11 @@ func TestTouchVariantAndTouchSourceContent_DropAccessTimesPastTheLimit(t *testin
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopPendingCountWrites_WritesPendingAccessTimes hands the access time
|
||||
// of one of two variants, three times, then stops the goroutine that writes
|
||||
// them, which never started. The stop must write that access time and
|
||||
// TestStopPendingAccessTimeWrites_WritesPendingAccessTimes hands the access
|
||||
// time of one of two variants, three times, then stops the goroutine that
|
||||
// writes them, which never started. The stop must write that access time and
|
||||
// leave the other variant's.
|
||||
func TestStopPendingCountWrites_WritesPendingAccessTimes(t *testing.T) {
|
||||
func TestStopPendingAccessTimeWrites_WritesPendingAccessTimes(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
@@ -274,9 +276,9 @@ func TestStopPendingCountWrites_WritesPendingAccessTimes(t *testing.T) {
|
||||
cache.touchVariant(testVariantKeyOne)
|
||||
}
|
||||
|
||||
err := cache.StopPendingCountWrites(t.Context())
|
||||
err := cache.StopPendingAccessTimeWrites(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("StopPendingCountWrites() error = %v", err)
|
||||
t.Fatalf("StopPendingAccessTimeWrites() error = %v", err)
|
||||
}
|
||||
|
||||
for key, wantTouched := range map[imagerequest.VariantKey]int{
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
@@ -33,16 +34,6 @@ func setupTestDB(t *testing.T) *sql.DB {
|
||||
expires_at DATETIME NOT NULL,
|
||||
UNIQUE(source_host, source_path, source_query)
|
||||
);
|
||||
CREATE TABLE cache_stats (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
hit_count INTEGER NOT NULL DEFAULT 0,
|
||||
miss_count INTEGER NOT NULL DEFAULT 0,
|
||||
upstream_fetch_count INTEGER NOT NULL DEFAULT 0,
|
||||
upstream_fetch_bytes INTEGER NOT NULL DEFAULT 0,
|
||||
transform_count INTEGER NOT NULL DEFAULT 0,
|
||||
last_updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
INSERT INTO cache_stats (id) VALUES (1);
|
||||
`
|
||||
|
||||
_, err = db.ExecContext(t.Context(), schema)
|
||||
@@ -53,48 +44,6 @@ func setupTestDB(t *testing.T) *sql.DB {
|
||||
return db
|
||||
}
|
||||
|
||||
func setupTestCache(t *testing.T) *Cache {
|
||||
t.Helper()
|
||||
|
||||
tmpDir := t.TempDir()
|
||||
db := setupTestDB(t)
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
return cache
|
||||
}
|
||||
|
||||
func TestCache_Stats(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache := setupTestCache(t)
|
||||
ctx := context.Background()
|
||||
|
||||
// Increment some stats
|
||||
cache.IncrementStats(true, 0)
|
||||
cache.IncrementStats(true, 0)
|
||||
cache.IncrementStats(false, 1024)
|
||||
|
||||
stats, err := cache.Stats(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("Stats() error = %v", err)
|
||||
}
|
||||
|
||||
if stats.HitCount != 2 {
|
||||
t.Errorf("Stats() HitCount = %d, want 2", stats.HitCount)
|
||||
}
|
||||
|
||||
if stats.MissCount != 1 {
|
||||
t.Errorf("Stats() MissCount = %d, want 1", stats.MissCount)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCache_CleanExpired(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -104,7 +53,7 @@ func TestCache_CleanExpired(t *testing.T) {
|
||||
cache, _ := NewCache(db, CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
NegativeTTL: 1 * time.Millisecond,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -155,7 +104,7 @@ func TestCache_StorageDirectoriesCreated(t *testing.T) {
|
||||
|
||||
_, err := NewCache(db, CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("NewCache() error = %v", err)
|
||||
}
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
)
|
||||
|
||||
// Static errors returned by the stub free-space probes below.
|
||||
@@ -144,7 +146,7 @@ func TestNewCacheComputesDefaultMaxBytesWhenAsked(t *testing.T) {
|
||||
cache, err := newCache(evictionTestDB(t), CacheConfig{
|
||||
StateDir: stateDir,
|
||||
UseDefaultMaxBytes: true,
|
||||
}, probe)
|
||||
}, prometheus.NewRegistry(), probe)
|
||||
if err != nil {
|
||||
t.Fatalf("newCache returned error: %v", err)
|
||||
}
|
||||
@@ -175,7 +177,7 @@ func TestNewCacheKeepsExplicitMaxBytes(t *testing.T) {
|
||||
cache, err := newCache(evictionTestDB(t), CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
MaxBytes: 1024,
|
||||
}, probe)
|
||||
}, prometheus.NewRegistry(), probe)
|
||||
if err != nil {
|
||||
t.Fatalf("newCache returned error: %v", err)
|
||||
}
|
||||
|
||||
@@ -191,11 +191,9 @@ func TestService_Get_ConcurrentMissesShareOneFetch(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
want := cacheStatsCounters{0, requests, 1, int64(len(photo)), 1}
|
||||
want := counterValues{0, requests, 1, int64(len(photo)), 1}
|
||||
|
||||
waitForPendingCountsWritten(t, svc.cache)
|
||||
|
||||
if got := readCacheStatsCounters(t, svc.cache); got != want {
|
||||
if got := readCounters(svc.cache); got != want {
|
||||
t.Errorf("counters = %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
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.",
|
||||
})
|
||||
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,
|
||||
} {
|
||||
err := metricsRegistry.Register(counter)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to register the cache's counters: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/testutil"
|
||||
)
|
||||
|
||||
// counterValues holds the values of a cache's counters.
|
||||
type counterValues struct {
|
||||
hitCount int64
|
||||
missCount int64
|
||||
upstreamFetchCount int64
|
||||
upstreamFetchBytes int64
|
||||
transformCount int64
|
||||
}
|
||||
|
||||
// readCounters reads cache's counters.
|
||||
func readCounters(cache *Cache) counterValues {
|
||||
return counterValues{
|
||||
hitCount: int64(testutil.ToFloat64(cache.cacheHits)),
|
||||
missCount: int64(testutil.ToFloat64(cache.cacheMisses)),
|
||||
upstreamFetchCount: int64(testutil.ToFloat64(cache.upstreamFetches)),
|
||||
upstreamFetchBytes: int64(testutil.ToFloat64(cache.upstreamFetchBytes)),
|
||||
transformCount: int64(testutil.ToFloat64(cache.transforms)),
|
||||
}
|
||||
}
|
||||
|
||||
// TestNewCache_RegistersItsCountersOnItsRegistry checks that NewCache
|
||||
// registers the cache's counters on the registry it is given, under the
|
||||
// names README.md gives them.
|
||||
func TestNewCache_RegistersItsCountersOnItsRegistry(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
registry := prometheus.NewRegistry()
|
||||
|
||||
_, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, registry)
|
||||
if err != nil {
|
||||
t.Fatalf("NewCache() error = %v", err)
|
||||
}
|
||||
|
||||
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...)
|
||||
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))
|
||||
}
|
||||
}
|
||||
|
||||
// TestNewCache_FailsWhenItsCountersCannotBeRegistered makes a second cache
|
||||
// on the registry of a first. Its counters have the same names, so the
|
||||
// registry refuses them, and NewCache must return that error.
|
||||
func TestNewCache_FailsWhenItsCountersCannotBeRegistered(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
registry := prometheus.NewRegistry()
|
||||
|
||||
_, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, registry)
|
||||
if err != nil {
|
||||
t.Fatalf("NewCache() of the first cache error = %v", err)
|
||||
}
|
||||
|
||||
_, err = NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()}, registry)
|
||||
t.Logf("NewCache() of the second cache error = %v", err)
|
||||
|
||||
var alreadyRegistered prometheus.AlreadyRegisteredError
|
||||
if !errors.As(err, &alreadyRegistered) {
|
||||
t.Errorf("NewCache() of the second cache error = %v, want %T",
|
||||
err, alreadyRegistered)
|
||||
}
|
||||
}
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"modernc.org/sqlite"
|
||||
"sneak.berlin/go/pixa/internal/database"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
@@ -74,7 +75,7 @@ func newEvictionTestCache(t *testing.T, maxBytes int64) (*Cache, string) {
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
MaxBytes: maxBytes,
|
||||
DisableDiskCache: maxBytes == 0,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
@@ -44,23 +44,4 @@ type ImageCache interface {
|
||||
|
||||
// Warm pre-fetches and caches an image without returning it
|
||||
Warm(ctx context.Context, req *imagerequest.Request) error
|
||||
|
||||
// Stats returns cache statistics
|
||||
Stats(ctx context.Context) (*CacheStats, error)
|
||||
}
|
||||
|
||||
// CacheStats contains cache statistics
|
||||
type CacheStats struct {
|
||||
// TotalItems is the number of cached source images plus processed
|
||||
// variants
|
||||
TotalItems int64
|
||||
// TotalSizeBytes is the total size of cached source images and
|
||||
// processed variants
|
||||
TotalSizeBytes int64
|
||||
// HitCount is the number of cache hits
|
||||
HitCount int64
|
||||
// MissCount is the number of cache misses
|
||||
MissCount int64
|
||||
// HitRate is HitCount / (HitCount + MissCount)
|
||||
HitRate float64
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
@@ -19,7 +20,7 @@ func TestNegativeCache_StoreAndCheck(t *testing.T) {
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: dir,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -68,7 +69,7 @@ func TestNegativeCache_Expired(t *testing.T) {
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: dir,
|
||||
NegativeTTL: 1 * time.Millisecond, // very short TTL
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -191,7 +192,7 @@ func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) {
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: time.Second,
|
||||
DisableDiskCache: tt.disableDiskCache,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
@@ -265,7 +266,8 @@ func negativeCacheRowCount(tb testing.TB, cache *Cache, sourcePath string) int {
|
||||
func TestCleanExpiredDeletesInBatches(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()},
|
||||
prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
@@ -319,7 +321,8 @@ func TestCleanExpiredDeletesInBatches(t *testing.T) {
|
||||
func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()},
|
||||
prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
@@ -349,7 +352,8 @@ func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) {
|
||||
func TestCleanExpiredStopsOnceContextEnds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()},
|
||||
prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
@@ -128,9 +128,7 @@ func TestService_Get_FetchesAgainForEmptyCachedSource(t *testing.T) {
|
||||
getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover))
|
||||
getImage(t, svc, photoVariant(fixtures, 30, imagerequest.FitCover))
|
||||
|
||||
waitForPendingCountsWritten(t, svc.cache)
|
||||
|
||||
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
|
||||
fetches := readCounters(svc.cache).upstreamFetchCount
|
||||
t.Logf("upstream fetches = %d", fetches)
|
||||
|
||||
if fetches != 2 {
|
||||
@@ -166,9 +164,7 @@ func TestService_Get_FetchesAgainForPartialCachedSource(t *testing.T) {
|
||||
getImage(t, svc, photoVariant(fixtures, 40, imagerequest.FitCover))
|
||||
getImage(t, svc, photoVariant(fixtures, 30, imagerequest.FitCover))
|
||||
|
||||
waitForPendingCountsWritten(t, svc.cache)
|
||||
|
||||
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
|
||||
fetches := readCounters(svc.cache).upstreamFetchCount
|
||||
t.Logf("upstream fetches = %d", fetches)
|
||||
|
||||
if fetches != 2 {
|
||||
|
||||
@@ -0,0 +1,160 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
// pendingWriteInterval is how often the goroutine
|
||||
// StartPendingAccessTimeWrites starts writes the pending access times.
|
||||
// SQLite's CURRENT_TIMESTAMP, which an access time is written as, counts
|
||||
// whole seconds.
|
||||
const pendingWriteInterval = time.Second
|
||||
|
||||
// StartPendingAccessTimeWrites starts the goroutine that writes, every
|
||||
// pendingWriteInterval, the pending access times in one transaction.
|
||||
// Requests only hand these to it, so none waits for the database to record
|
||||
// them, and only this goroutine's write ever waits for it. It is a no-op
|
||||
// when already started. The goroutine outlives the caller, so it runs with
|
||||
// its own context, which StopPendingAccessTimeWrites cancels to tell it to
|
||||
// finish.
|
||||
func (c *Cache) StartPendingAccessTimeWrites() {
|
||||
if c.pendingAccessTimesCancel != nil {
|
||||
return
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
c.pendingAccessTimesCancel = cancel
|
||||
|
||||
go func() {
|
||||
c.pendingAccessTimeWriteLoop(ctx)
|
||||
}()
|
||||
}
|
||||
|
||||
// StopPendingAccessTimeWrites tells the goroutine
|
||||
// StartPendingAccessTimeWrites started, if it did, to finish, and waits for
|
||||
// its write under way to end and for it to return. It then writes the
|
||||
// pending access times left, once, so that they are written before the
|
||||
// database closes. It waits for all of this at most until ctx ends. When
|
||||
// access times are not written, it logs that and returns an error.
|
||||
func (c *Cache) StopPendingAccessTimeWrites(ctx context.Context) error {
|
||||
var err error
|
||||
|
||||
if c.pendingAccessTimesCancel != nil {
|
||||
c.pendingAccessTimesCancel()
|
||||
|
||||
select {
|
||||
case <-c.pendingAccessTimesDone:
|
||||
case <-ctx.Done():
|
||||
// The write given up on here may not reach the database:
|
||||
// database/sql closes without waiting for a query, and a write
|
||||
// still waiting for the connection fails when the database
|
||||
// closes. Nothing is written here.
|
||||
err = fmt.Errorf("access times still being written: %w", ctx.Err())
|
||||
}
|
||||
}
|
||||
|
||||
if err == nil {
|
||||
err = c.writePendingAccessTimes(ctx)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
c.log.Error("access times not written at shutdown", "error", err)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// pendingAccessTimeWriteLoop is the body of the goroutine
|
||||
// StartPendingAccessTimeWrites starts. Once ctx is cancelled it starts no
|
||||
// write and returns, but does not cut off a write under way then. Access
|
||||
// times whose write fails are dropped.
|
||||
func (c *Cache) pendingAccessTimeWriteLoop(ctx context.Context) {
|
||||
defer close(c.pendingAccessTimesDone)
|
||||
|
||||
ticker := time.NewTicker(pendingWriteInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
|
||||
// select picks at random when ctx is cancelled and a tick waits
|
||||
// too.
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err := c.writePendingAccessTimes(context.WithoutCancel(ctx))
|
||||
if err != nil {
|
||||
c.log.Warn("failed to write access times", "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// writePendingAccessTimes takes out the variants waiting in
|
||||
// pendingVariantAccessTimes and the sources waiting in
|
||||
// pendingSourceAccessTimes, and sets their last_accessed_at to now in one
|
||||
// transaction, once for each however often it was used. Nothing else takes
|
||||
// from those channels while it runs: only the goroutine
|
||||
// StartPendingAccessTimeWrites starts calls it, or
|
||||
// StopPendingAccessTimeWrites once that goroutine has returned.
|
||||
func (c *Cache) writePendingAccessTimes(ctx context.Context) error {
|
||||
variants := make(map[imagerequest.VariantKey]struct{})
|
||||
|
||||
for range len(c.pendingVariantAccessTimes) {
|
||||
variants[<-c.pendingVariantAccessTimes] = struct{}{}
|
||||
}
|
||||
|
||||
sources := make(map[imagerequest.SourceKey]struct{})
|
||||
|
||||
for range len(c.pendingSourceAccessTimes) {
|
||||
sources[<-c.pendingSourceAccessTimes] = struct{}{}
|
||||
}
|
||||
|
||||
accessTimes := len(variants) + len(sources)
|
||||
if accessTimes == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
tx, err := c.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
|
||||
// Does nothing once the transaction is committed
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
for cacheKey := range variants {
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE cache_key = ?
|
||||
`, string(cacheKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
}
|
||||
|
||||
for sourceKey := range sources {
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE source_key = ?
|
||||
`, string(sourceKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
}
|
||||
|
||||
err = tx.Commit()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,319 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
// pendingAccessTimesWait is how long a test waits for the database's
|
||||
// connection, for an access time to reach the database, for a write to start
|
||||
// waiting for the connection, for a request to return past its deadline, or
|
||||
// for StopPendingAccessTimeWrites to return.
|
||||
const pendingAccessTimesWait = 5 * time.Second
|
||||
|
||||
// otherWritesWait is how long a test gives other writes, once one write waits
|
||||
// for the database's connection, to start waiting for it too.
|
||||
const otherWritesWait = 100 * time.Millisecond
|
||||
|
||||
// holdDatabase takes the one connection of cache's database, so that every
|
||||
// other query waits for it, and returns the func that frees it.
|
||||
func holdDatabase(t *testing.T, cache *Cache) func() {
|
||||
t.Helper()
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), pendingAccessTimesWait)
|
||||
defer cancel()
|
||||
|
||||
conn, err := cache.db.Conn(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to take the database connection: %v", err)
|
||||
}
|
||||
|
||||
return func() { _ = conn.Close() }
|
||||
}
|
||||
|
||||
// waitForConnectionWaits waits until db.Stats().WaitCount, the number of
|
||||
// times a caller has waited for a connection, reaches waits.
|
||||
func waitForConnectionWaits(t *testing.T, db *sql.DB, waits int64) {
|
||||
t.Helper()
|
||||
|
||||
deadline := time.Now().Add(pendingAccessTimesWait)
|
||||
|
||||
for db.Stats().WaitCount < waits {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("callers waited for the database connection %d times, want %d",
|
||||
db.Stats().WaitCount, waits)
|
||||
}
|
||||
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the
|
||||
// database's one connection while a request whose fetch is held reaches its
|
||||
// deadline. The request must return with the deadline's error while the
|
||||
// connection is still held, as it waits for no database write past its
|
||||
// deadline.
|
||||
func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures, fetcher := setupHeldFetchService(t)
|
||||
|
||||
svc.cache.StartPendingAccessTimeWrites()
|
||||
defer func() { _ = svc.cache.StopPendingAccessTimeWrites(t.Context()) }()
|
||||
|
||||
// Room for the request to reach its fetch on a busy host
|
||||
const timeout = 2 * time.Second
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), timeout)
|
||||
defer cancel()
|
||||
|
||||
deadline, _ := ctx.Deadline()
|
||||
|
||||
results := startGet(ctx, svc, photoVariant(fixtures, 85, imagerequest.FitCover))
|
||||
|
||||
// The request has made its database reads by the time it fetches.
|
||||
select {
|
||||
case <-fetcher.started:
|
||||
case <-time.After(timeout):
|
||||
t.Fatal("request did not reach its fetch by its deadline")
|
||||
}
|
||||
|
||||
releaseDatabase := holdDatabase(t, svc.cache)
|
||||
defer releaseDatabase()
|
||||
|
||||
select {
|
||||
case got := <-results:
|
||||
t.Logf("Get() returned %v after its deadline, error = %v",
|
||||
time.Since(deadline), got.err)
|
||||
|
||||
if !errors.Is(got.err, context.DeadlineExceeded) {
|
||||
t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded)
|
||||
}
|
||||
case <-time.After(time.Until(deadline) + pendingAccessTimesWait):
|
||||
t.Fatal("request did not return while the database was busy")
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopPendingAccessTimeWrites_LetsTheWriteUnderWayFinish holds the
|
||||
// database's one connection while the goroutine that writes the pending
|
||||
// access times waits for it, hands over another access time, then stops that
|
||||
// goroutine. The stop must leave the write under way to finish rather than
|
||||
// cut it off, and once the connection is free it must write the access time
|
||||
// that write left.
|
||||
func TestStopPendingAccessTimeWrites_LetsTheWriteUnderWayFinish(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
anHourAgo := time.Now().Add(-time.Hour)
|
||||
|
||||
for _, key := range []imagerequest.VariantKey{testVariantKeyOne, testVariantKeyTwo} {
|
||||
storeEvictionTestVariant(t, cache, key, []byte("variant bytes"))
|
||||
setVariantLastAccessed(t, cache, key, anHourAgo)
|
||||
}
|
||||
|
||||
cache.StartPendingAccessTimeWrites()
|
||||
defer func() { _ = cache.StopPendingAccessTimeWrites(t.Context()) }()
|
||||
|
||||
releaseDatabase := holdDatabase(t, cache)
|
||||
defer releaseDatabase()
|
||||
|
||||
waitsBefore := cache.db.Stats().WaitCount
|
||||
|
||||
cache.touchVariant(testVariantKeyOne)
|
||||
|
||||
// The goroutine's write waits for the connection.
|
||||
waitForConnectionWaits(t, cache.db, waitsBefore+1)
|
||||
|
||||
// Handed over after that write took the pending access times
|
||||
cache.touchVariant(testVariantKeyTwo)
|
||||
|
||||
// The goroutine is told to finish here as well as by the stop, so that it
|
||||
// is told before its write under way ends, however late the stop runs.
|
||||
// The access time above then reaches the database only through the
|
||||
// stop's own write.
|
||||
cache.pendingAccessTimesCancel()
|
||||
|
||||
stopped := make(chan error, 1)
|
||||
|
||||
go func() {
|
||||
stopped <- cache.StopPendingAccessTimeWrites(t.Context())
|
||||
}()
|
||||
|
||||
// Had the stop cut off the write under way, its own write would wait for
|
||||
// the connection too.
|
||||
time.Sleep(otherWritesWait)
|
||||
|
||||
waits := cache.db.Stats().WaitCount - waitsBefore
|
||||
t.Logf("%d writes waited for the database", waits)
|
||||
|
||||
if waits > 1 {
|
||||
t.Errorf("%d writes waited for the database, want 1: "+
|
||||
"the stop cut off the write under way", waits)
|
||||
}
|
||||
|
||||
releaseDatabase()
|
||||
|
||||
select {
|
||||
case err := <-stopped:
|
||||
if err != nil {
|
||||
t.Fatalf("StopPendingAccessTimeWrites() error = %v", err)
|
||||
}
|
||||
case <-time.After(pendingAccessTimesWait):
|
||||
t.Fatal("StopPendingAccessTimeWrites() did not return once the database was free")
|
||||
}
|
||||
|
||||
for _, key := range []imagerequest.VariantKey{testVariantKeyOne, testVariantKeyTwo} {
|
||||
if countRows(t, cache, `
|
||||
SELECT COUNT(*) FROM variant_content
|
||||
WHERE cache_key = ? AND last_accessed_at > ?
|
||||
`, string(key), anHourAgo.UTC().Format(sqliteTimestampFormat)) != 1 {
|
||||
t.Errorf("the access time of variant %s was not written", key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopPendingAccessTimeWrites_ReturnsWhenItsContextEnds holds the
|
||||
// database's one connection while the goroutine that writes the pending
|
||||
// access times waits for it, then stops that goroutine with a context that
|
||||
// ends first. The stop must return the context's error, and the write under
|
||||
// way must still reach the database once the connection is free.
|
||||
func TestStopPendingAccessTimeWrites_ReturnsWhenItsContextEnds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyOne, []byte("variant bytes"))
|
||||
|
||||
anHourAgo := time.Now().Add(-time.Hour)
|
||||
setVariantLastAccessed(t, cache, testVariantKeyOne, anHourAgo)
|
||||
|
||||
cache.StartPendingAccessTimeWrites()
|
||||
defer func() { _ = cache.StopPendingAccessTimeWrites(t.Context()) }()
|
||||
|
||||
releaseDatabase := holdDatabase(t, cache)
|
||||
defer releaseDatabase()
|
||||
|
||||
waitsBefore := cache.db.Stats().WaitCount
|
||||
|
||||
cache.touchVariant(testVariantKeyOne)
|
||||
|
||||
// The goroutine's write waits for the connection.
|
||||
waitForConnectionWaits(t, cache.db, waitsBefore+1)
|
||||
|
||||
stopCtx, cancelStop := context.WithCancel(t.Context())
|
||||
cancelStop()
|
||||
|
||||
stopped := make(chan error, 1)
|
||||
|
||||
go func() {
|
||||
stopped <- cache.StopPendingAccessTimeWrites(stopCtx)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-stopped:
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("StopPendingAccessTimeWrites() error = %v, want %v",
|
||||
err, context.Canceled)
|
||||
}
|
||||
case <-time.After(pendingAccessTimesWait):
|
||||
t.Fatal("StopPendingAccessTimeWrites() did not return once its context ended")
|
||||
}
|
||||
|
||||
releaseDatabase()
|
||||
|
||||
deadline := time.Now().Add(pendingAccessTimesWait)
|
||||
|
||||
for countRows(t, cache, `
|
||||
SELECT COUNT(*) FROM variant_content
|
||||
WHERE cache_key = ? AND last_accessed_at > ?
|
||||
`, string(testVariantKeyOne), anHourAgo.UTC().Format(sqliteTimestampFormat)) == 0 {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("the access time being written was not written")
|
||||
}
|
||||
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLookupSource_HandsAccessTimeToTheBackgroundWriter finds a cached source
|
||||
// whose access time is an hour old. LookupSource must not write the access
|
||||
// time itself but hand it over to be written, and StopPendingAccessTimeWrites
|
||||
// must then write it.
|
||||
func TestLookupSource_HandsAccessTimeToTheBackgroundWriter(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
key := storeEvictionTestSource(t, cache, "example.com", "/photo.jpg",
|
||||
[]byte("source bytes"))
|
||||
anHourAgo := time.Now().Add(-time.Hour)
|
||||
setSourceLastAccessed(t, cache, key, anHourAgo)
|
||||
|
||||
touchedQuery := `
|
||||
SELECT COUNT(*) FROM source_content
|
||||
WHERE source_key = ? AND last_accessed_at > ?
|
||||
`
|
||||
anHourAgoInDatabase := anHourAgo.UTC().Format(sqliteTimestampFormat)
|
||||
|
||||
found, err := cache.LookupSource(t.Context(), &imagerequest.Request{
|
||||
Source: imagerequest.Source{Host: "example.com", Path: "/photo.jpg"},
|
||||
})
|
||||
if err != nil || found != key {
|
||||
t.Fatalf("LookupSource() = %q, %v; want %q", found, err, key)
|
||||
}
|
||||
|
||||
if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 0 {
|
||||
t.Error("LookupSource wrote the source's access time itself")
|
||||
}
|
||||
|
||||
if waiting := len(cache.pendingSourceAccessTimes); waiting != 1 {
|
||||
t.Errorf("%d access times of sources wait to be written, want 1", waiting)
|
||||
}
|
||||
|
||||
err = cache.StopPendingAccessTimeWrites(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("StopPendingAccessTimeWrites() error = %v", err)
|
||||
}
|
||||
|
||||
if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 1 {
|
||||
t.Error("the source's access time was not written")
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopPendingAccessTimeWrites_LogsWhatWasNotWritten makes the write of
|
||||
// the access times fail at shutdown, by dropping the table they are written
|
||||
// to. The stop must return an error and log that access times were not
|
||||
// written.
|
||||
func TestStopPendingAccessTimeWrites_LogsWhatWasNotWritten(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
var logs bytes.Buffer
|
||||
|
||||
cache.log = slog.New(slog.NewJSONHandler(&logs, nil))
|
||||
|
||||
cache.touchVariant(testVariantKeyOne)
|
||||
|
||||
_, err := cache.db.ExecContext(t.Context(), `DROP TABLE variant_content`)
|
||||
if err != nil {
|
||||
t.Fatalf("dropping variant_content: %v", err)
|
||||
}
|
||||
|
||||
err = cache.StopPendingAccessTimeWrites(t.Context())
|
||||
t.Logf("StopPendingAccessTimeWrites() error = %v; logged:\n%s", err, logs.String())
|
||||
|
||||
if err == nil {
|
||||
t.Error("StopPendingAccessTimeWrites() error = nil, want the access times' error")
|
||||
}
|
||||
|
||||
if !strings.Contains(logs.String(), "access times not written at shutdown") {
|
||||
t.Error("the access times not written were not logged")
|
||||
}
|
||||
}
|
||||
@@ -1,221 +0,0 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
// pendingWriteInterval is how often the goroutine StartPendingCountWrites
|
||||
// starts writes the pending access times and counts. SQLite's
|
||||
// CURRENT_TIMESTAMP, which an access time is written as, counts whole
|
||||
// seconds.
|
||||
const pendingWriteInterval = time.Second
|
||||
|
||||
// StartPendingCountWrites starts the goroutine that writes, every
|
||||
// pendingWriteInterval, the pending access times in one transaction and then
|
||||
// the pending counts in one UPDATE. Requests only hand these to it, so none
|
||||
// waits for the database to record what it did, and only this goroutine's
|
||||
// write ever waits for it. It is a no-op when already started. The goroutine
|
||||
// outlives the caller, so it runs with its own context, which
|
||||
// StopPendingCountWrites cancels to tell it to finish.
|
||||
func (c *Cache) StartPendingCountWrites() {
|
||||
if c.pendingCountsCancel != nil {
|
||||
return
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
c.pendingCountsCancel = cancel
|
||||
|
||||
go func() {
|
||||
c.pendingCountWriteLoop(ctx)
|
||||
}()
|
||||
}
|
||||
|
||||
// StopPendingCountWrites tells the goroutine StartPendingCountWrites
|
||||
// started, if it did, to finish, and waits for its write under way to end
|
||||
// and for it to return. It then writes the pending access times and counts
|
||||
// left, once, so that they are written before the database closes. It waits
|
||||
// for all of this at most until ctx ends. It logs what was not written,
|
||||
// access times or counts, and returns an error.
|
||||
func (c *Cache) StopPendingCountWrites(ctx context.Context) error {
|
||||
var accessTimesErr, countsErr error
|
||||
|
||||
if c.pendingCountsCancel != nil {
|
||||
c.pendingCountsCancel()
|
||||
|
||||
select {
|
||||
case <-c.pendingCountsDone:
|
||||
case <-ctx.Done():
|
||||
// The write given up on here, of access times or of counts,
|
||||
// may not reach the database: database/sql closes without
|
||||
// waiting for a query, and a write still waiting for the
|
||||
// connection fails when the database closes. Neither is
|
||||
// written here.
|
||||
accessTimesErr = fmt.Errorf("access times still being written: %w", ctx.Err())
|
||||
countsErr = fmt.Errorf("pending counts still being written: %w", ctx.Err())
|
||||
}
|
||||
}
|
||||
|
||||
if accessTimesErr == nil && countsErr == nil {
|
||||
accessTimesErr = c.writePendingAccessTimes(ctx)
|
||||
countsErr = c.writePendingCounts(ctx)
|
||||
}
|
||||
|
||||
if accessTimesErr != nil {
|
||||
c.log.Error("access times not written at shutdown", "error", accessTimesErr)
|
||||
}
|
||||
|
||||
if countsErr != nil {
|
||||
c.log.Error("counts not written at shutdown",
|
||||
"hits", c.pendingHits.Load(),
|
||||
"misses", c.pendingMisses.Load(),
|
||||
"upstream_fetches", c.pendingUpstreamFetches.Load(),
|
||||
"upstream_fetch_bytes", c.pendingUpstreamFetchBytes.Load(),
|
||||
"transforms", c.pendingTransforms.Load(),
|
||||
"error", countsErr,
|
||||
)
|
||||
}
|
||||
|
||||
return errors.Join(accessTimesErr, countsErr)
|
||||
}
|
||||
|
||||
// pendingCountWriteLoop is the body of the goroutine
|
||||
// StartPendingCountWrites starts. Once ctx is cancelled it starts no write
|
||||
// and returns, but does not cut off a write under way then. Counts whose
|
||||
// write fails stay pending, for the next write; access times whose write
|
||||
// fails are dropped.
|
||||
func (c *Cache) pendingCountWriteLoop(ctx context.Context) {
|
||||
defer close(c.pendingCountsDone)
|
||||
|
||||
ticker := time.NewTicker(pendingWriteInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
|
||||
// select picks at random when ctx is cancelled and a tick waits
|
||||
// too.
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err := c.writePendingAccessTimes(context.WithoutCancel(ctx))
|
||||
if err != nil {
|
||||
c.log.Warn("failed to write access times", "error", err)
|
||||
}
|
||||
|
||||
err = c.writePendingCounts(context.WithoutCancel(ctx))
|
||||
if err != nil {
|
||||
c.log.Warn("failed to write pending counts", "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// writePendingAccessTimes takes out the variants waiting in
|
||||
// pendingVariantAccessTimes and the sources waiting in
|
||||
// pendingSourceAccessTimes, and sets their last_accessed_at to now in one
|
||||
// transaction, once for each however often it was used. Nothing else takes
|
||||
// from those channels while it runs: only the goroutine
|
||||
// StartPendingCountWrites starts calls it, or StopPendingCountWrites once
|
||||
// that goroutine has returned.
|
||||
func (c *Cache) writePendingAccessTimes(ctx context.Context) error {
|
||||
variants := make(map[imagerequest.VariantKey]struct{})
|
||||
|
||||
for range len(c.pendingVariantAccessTimes) {
|
||||
variants[<-c.pendingVariantAccessTimes] = struct{}{}
|
||||
}
|
||||
|
||||
sources := make(map[imagerequest.SourceKey]struct{})
|
||||
|
||||
for range len(c.pendingSourceAccessTimes) {
|
||||
sources[<-c.pendingSourceAccessTimes] = struct{}{}
|
||||
}
|
||||
|
||||
accessTimes := len(variants) + len(sources)
|
||||
if accessTimes == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
tx, err := c.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
|
||||
// Does nothing once the transaction is committed
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
for cacheKey := range variants {
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE variant_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE cache_key = ?
|
||||
`, string(cacheKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
}
|
||||
|
||||
for sourceKey := range sources {
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE source_content SET last_accessed_at = CURRENT_TIMESTAMP
|
||||
WHERE source_key = ?
|
||||
`, string(sourceKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
}
|
||||
|
||||
err = tx.Commit()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write %d access times: %w", accessTimes, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// writePendingCounts adds the pending counts to the cache_stats row in one
|
||||
// UPDATE, then takes what it wrote out of them, leaving any added
|
||||
// meanwhile. When the UPDATE fails, they all stay pending.
|
||||
func (c *Cache) writePendingCounts(ctx context.Context) error {
|
||||
c.pendingCountsWriteMutex.Lock()
|
||||
defer c.pendingCountsWriteMutex.Unlock()
|
||||
|
||||
hits := c.pendingHits.Load()
|
||||
misses := c.pendingMisses.Load()
|
||||
upstreamFetches := c.pendingUpstreamFetches.Load()
|
||||
upstreamFetchBytes := c.pendingUpstreamFetchBytes.Load()
|
||||
transforms := c.pendingTransforms.Load()
|
||||
|
||||
if hits+misses+upstreamFetches+upstreamFetchBytes+transforms == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
UPDATE cache_stats
|
||||
SET hit_count = hit_count + ?,
|
||||
miss_count = miss_count + ?,
|
||||
upstream_fetch_count = upstream_fetch_count + ?,
|
||||
upstream_fetch_bytes = upstream_fetch_bytes + ?,
|
||||
transform_count = transform_count + ?,
|
||||
last_updated_at = CURRENT_TIMESTAMP
|
||||
WHERE id = 1
|
||||
`, hits, misses, upstreamFetches, upstreamFetchBytes, transforms)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to write pending counts: %w", err)
|
||||
}
|
||||
|
||||
c.pendingHits.Add(-hits)
|
||||
c.pendingMisses.Add(-misses)
|
||||
c.pendingUpstreamFetches.Add(-upstreamFetches)
|
||||
c.pendingUpstreamFetchBytes.Add(-upstreamFetchBytes)
|
||||
c.pendingTransforms.Add(-transforms)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -1,431 +0,0 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
// pendingCountsWait is how long a test waits for the database's connection,
|
||||
// for a count to reach the database, for a write to start waiting for the
|
||||
// connection, for a request to return past its deadline, or for
|
||||
// StopPendingCountWrites to return.
|
||||
const pendingCountsWait = 5 * time.Second
|
||||
|
||||
// otherWritesWait is how long a test gives other writes, once one write waits
|
||||
// for the database's connection, to start waiting for it too.
|
||||
const otherWritesWait = 100 * time.Millisecond
|
||||
|
||||
// holdDatabase takes the one connection of cache's database, so that every
|
||||
// other query waits for it, and returns the func that frees it.
|
||||
func holdDatabase(t *testing.T, cache *Cache) func() {
|
||||
t.Helper()
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), pendingCountsWait)
|
||||
defer cancel()
|
||||
|
||||
conn, err := cache.db.Conn(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to take the database connection: %v", err)
|
||||
}
|
||||
|
||||
return func() { _ = conn.Close() }
|
||||
}
|
||||
|
||||
// waitForConnectionWaits waits until db.Stats().WaitCount, the number of
|
||||
// times a caller has waited for a connection, reaches waits.
|
||||
func waitForConnectionWaits(t *testing.T, db *sql.DB, waits int64) {
|
||||
t.Helper()
|
||||
|
||||
deadline := time.Now().Add(pendingCountsWait)
|
||||
|
||||
for db.Stats().WaitCount < waits {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("callers waited for the database connection %d times, want %d",
|
||||
db.Stats().WaitCount, waits)
|
||||
}
|
||||
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// waitForCounters waits until the cache_stats row holds want.
|
||||
func waitForCounters(t *testing.T, cache *Cache, want cacheStatsCounters) {
|
||||
t.Helper()
|
||||
|
||||
deadline := time.Now().Add(pendingCountsWait)
|
||||
|
||||
for {
|
||||
got := readCacheStatsCounters(t, cache)
|
||||
if got == want {
|
||||
return
|
||||
}
|
||||
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("counters = %+v, want %+v", got, want)
|
||||
}
|
||||
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// waitForPendingCountsWritten starts the goroutine that writes cache's
|
||||
// pending counts, unless it is running, and waits until it has written them
|
||||
// all, so that the database holds every count made so far. The goroutine is
|
||||
// stopped when the test ends. The database is limited to one connection
|
||||
// first: each connection to an in-memory SQLite database opens a new, empty
|
||||
// one, so the goroutine must share the connection that holds the schema.
|
||||
func waitForPendingCountsWritten(t *testing.T, cache *Cache) {
|
||||
t.Helper()
|
||||
|
||||
if cache.pendingCountsCancel == nil {
|
||||
cache.db.SetMaxOpenConns(1)
|
||||
cache.StartPendingCountWrites()
|
||||
|
||||
t.Cleanup(func() { _ = cache.StopPendingCountWrites(context.Background()) })
|
||||
}
|
||||
|
||||
deadline := time.Now().Add(pendingCountsWait)
|
||||
|
||||
for cache.pendingHits.Load()+cache.pendingMisses.Load()+
|
||||
cache.pendingUpstreamFetches.Load()+cache.pendingUpstreamFetchBytes.Load()+
|
||||
cache.pendingTransforms.Load() > 0 {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("the pending counts were not written")
|
||||
}
|
||||
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy holds the
|
||||
// database's one connection while a request whose fetch is held reaches its
|
||||
// deadline. The request must return with the deadline's error while the
|
||||
// connection is still held, as it waits for no database write past its
|
||||
// deadline, and its miss must reach the database once the connection is free.
|
||||
func TestService_Get_ReturnsByItsDeadlineWhileTheDatabaseIsBusy(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures, fetcher := setupHeldFetchService(t)
|
||||
|
||||
svc.cache.StartPendingCountWrites()
|
||||
defer func() { _ = svc.cache.StopPendingCountWrites(t.Context()) }()
|
||||
|
||||
// Room for the request to reach its fetch on a busy host
|
||||
const timeout = 2 * time.Second
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), timeout)
|
||||
defer cancel()
|
||||
|
||||
deadline, _ := ctx.Deadline()
|
||||
|
||||
results := startGet(ctx, svc, photoVariant(fixtures, 85, imagerequest.FitCover))
|
||||
|
||||
// The request has made its database reads by the time it fetches.
|
||||
select {
|
||||
case <-fetcher.started:
|
||||
case <-time.After(timeout):
|
||||
t.Fatal("request did not reach its fetch by its deadline")
|
||||
}
|
||||
|
||||
releaseDatabase := holdDatabase(t, svc.cache)
|
||||
defer releaseDatabase()
|
||||
|
||||
select {
|
||||
case got := <-results:
|
||||
t.Logf("Get() returned %v after its deadline, error = %v",
|
||||
time.Since(deadline), got.err)
|
||||
|
||||
if !errors.Is(got.err, context.DeadlineExceeded) {
|
||||
t.Errorf("Get() error = %v, want %v", got.err, context.DeadlineExceeded)
|
||||
}
|
||||
case <-time.After(time.Until(deadline) + pendingCountsWait):
|
||||
t.Fatal("request did not return while the database was busy")
|
||||
}
|
||||
|
||||
releaseDatabase()
|
||||
|
||||
waitForCounters(t, svc.cache, cacheStatsCounters{missCount: 1})
|
||||
}
|
||||
|
||||
// TestIncrementStats_ManyCountsLeaveOneWriteWaiting holds the database's one
|
||||
// connection while many misses are added to the pending counts, which the
|
||||
// goroutine StartPendingCountWrites starts writes to the database. At most one
|
||||
// write may then wait for the connection, and every miss must reach the
|
||||
// database once it is free.
|
||||
func TestIncrementStats_ManyCountsLeaveOneWriteWaiting(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
cache.StartPendingCountWrites()
|
||||
defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
|
||||
|
||||
releaseDatabase := holdDatabase(t, cache)
|
||||
defer releaseDatabase()
|
||||
|
||||
waitsBefore := cache.db.Stats().WaitCount
|
||||
|
||||
const misses = 20
|
||||
|
||||
for range misses {
|
||||
cache.IncrementStats(false, 0)
|
||||
}
|
||||
|
||||
// Once one write waits for the connection, any others start waiting too.
|
||||
waitForConnectionWaits(t, cache.db, waitsBefore+1)
|
||||
time.Sleep(otherWritesWait)
|
||||
|
||||
waits := cache.db.Stats().WaitCount - waitsBefore
|
||||
t.Logf("%d writes waited for the database", waits)
|
||||
|
||||
if waits > 1 {
|
||||
t.Errorf("%d writes waited for the database, want at most 1", waits)
|
||||
}
|
||||
|
||||
releaseDatabase()
|
||||
|
||||
waitForCounters(t, cache, cacheStatsCounters{missCount: misses})
|
||||
}
|
||||
|
||||
// TestStats_IncludesPendingCounts adds hits and misses to the pending counts,
|
||||
// with no goroutine started to write them, and checks that Stats adds them to
|
||||
// the database's counts before they are written.
|
||||
func TestStats_IncludesPendingCounts(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
_, err := cache.db.ExecContext(t.Context(),
|
||||
`UPDATE cache_stats SET hit_count = 75, miss_count = 25 WHERE id = 1`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
cache.IncrementStats(true, 0)
|
||||
cache.IncrementStats(true, 0)
|
||||
cache.IncrementStats(false, 0)
|
||||
|
||||
stats, err := cache.Stats(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("Stats() error = %v", err)
|
||||
}
|
||||
|
||||
if stats.HitCount != 77 || stats.MissCount != 26 {
|
||||
t.Errorf("HitCount = %d, MissCount = %d, want 77 and 26",
|
||||
stats.HitCount, stats.MissCount)
|
||||
}
|
||||
|
||||
want := cacheStatsCounters{hitCount: 75, missCount: 25}
|
||||
if got := readCacheStatsCounters(t, cache); got != want {
|
||||
t.Errorf("counters in the database = %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopPendingCountWrites_WritesPendingCounts holds the database's one
|
||||
// connection while the goroutine that writes the pending counts waits for it,
|
||||
// counts more, then stops that goroutine. The stop must leave the write under
|
||||
// way to finish rather than cut it off, and once the connection is free it
|
||||
// must write the counts that write left.
|
||||
func TestStopPendingCountWrites_WritesPendingCounts(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
cache.StartPendingCountWrites()
|
||||
defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
|
||||
|
||||
releaseDatabase := holdDatabase(t, cache)
|
||||
defer releaseDatabase()
|
||||
|
||||
waitsBefore := cache.db.Stats().WaitCount
|
||||
|
||||
cache.IncrementStats(true, 0)
|
||||
|
||||
// The goroutine's write waits for the connection.
|
||||
waitForConnectionWaits(t, cache.db, waitsBefore+1)
|
||||
|
||||
// Counted after that write read the pending counts
|
||||
cache.IncrementStats(false, 0)
|
||||
cache.IncrementUpstreamFetch(1024)
|
||||
cache.IncrementTransformCount()
|
||||
|
||||
// The goroutine is told to finish here as well as by the stop, so that it
|
||||
// is told before its write under way ends, however late the stop runs.
|
||||
// The counts above then reach the database only through the stop's own
|
||||
// write.
|
||||
cache.pendingCountsCancel()
|
||||
|
||||
stopped := make(chan error, 1)
|
||||
|
||||
go func() {
|
||||
stopped <- cache.StopPendingCountWrites(t.Context())
|
||||
}()
|
||||
|
||||
// Had the stop cut off the write under way, its own write would wait for
|
||||
// the connection too.
|
||||
time.Sleep(otherWritesWait)
|
||||
|
||||
waits := cache.db.Stats().WaitCount - waitsBefore
|
||||
t.Logf("%d writes waited for the database", waits)
|
||||
|
||||
if waits > 1 {
|
||||
t.Errorf("%d writes waited for the database, want 1: "+
|
||||
"the stop cut off the write under way", waits)
|
||||
}
|
||||
|
||||
releaseDatabase()
|
||||
|
||||
select {
|
||||
case err := <-stopped:
|
||||
if err != nil {
|
||||
t.Fatalf("StopPendingCountWrites() error = %v", err)
|
||||
}
|
||||
case <-time.After(pendingCountsWait):
|
||||
t.Fatal("StopPendingCountWrites() did not return once the database was free")
|
||||
}
|
||||
|
||||
want := cacheStatsCounters{1, 1, 1, 1024, 1}
|
||||
if got := readCacheStatsCounters(t, cache); got != want {
|
||||
t.Errorf("counters = %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopPendingCountWrites_ReturnsWhenItsContextEnds holds the database's
|
||||
// one connection while the goroutine that writes the pending counts waits for
|
||||
// it, then stops that goroutine with a context that ends first. The stop must
|
||||
// return the context's error, and the write under way must still reach the
|
||||
// database once the connection is free.
|
||||
func TestStopPendingCountWrites_ReturnsWhenItsContextEnds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
cache.StartPendingCountWrites()
|
||||
defer func() { _ = cache.StopPendingCountWrites(t.Context()) }()
|
||||
|
||||
releaseDatabase := holdDatabase(t, cache)
|
||||
defer releaseDatabase()
|
||||
|
||||
waitsBefore := cache.db.Stats().WaitCount
|
||||
|
||||
cache.IncrementStats(true, 0)
|
||||
|
||||
// The goroutine's write waits for the connection.
|
||||
waitForConnectionWaits(t, cache.db, waitsBefore+1)
|
||||
|
||||
stopCtx, cancelStop := context.WithCancel(t.Context())
|
||||
cancelStop()
|
||||
|
||||
stopped := make(chan error, 1)
|
||||
|
||||
go func() {
|
||||
stopped <- cache.StopPendingCountWrites(stopCtx)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-stopped:
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("StopPendingCountWrites() error = %v, want %v",
|
||||
err, context.Canceled)
|
||||
}
|
||||
case <-time.After(pendingCountsWait):
|
||||
t.Fatal("StopPendingCountWrites() did not return once its context ended")
|
||||
}
|
||||
|
||||
releaseDatabase()
|
||||
|
||||
waitForCounters(t, cache, cacheStatsCounters{hitCount: 1})
|
||||
}
|
||||
|
||||
// TestLookupSource_HandsAccessTimeToTheBackgroundWriter finds a cached source
|
||||
// whose access time is an hour old. LookupSource must not write the access
|
||||
// time itself but hand it over to be written, and StopPendingCountWrites must
|
||||
// then write it.
|
||||
func TestLookupSource_HandsAccessTimeToTheBackgroundWriter(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
key := storeEvictionTestSource(t, cache, "example.com", "/photo.jpg",
|
||||
[]byte("source bytes"))
|
||||
anHourAgo := time.Now().Add(-time.Hour)
|
||||
setSourceLastAccessed(t, cache, key, anHourAgo)
|
||||
|
||||
touchedQuery := `
|
||||
SELECT COUNT(*) FROM source_content
|
||||
WHERE source_key = ? AND last_accessed_at > ?
|
||||
`
|
||||
anHourAgoInDatabase := anHourAgo.UTC().Format(sqliteTimestampFormat)
|
||||
|
||||
found, err := cache.LookupSource(t.Context(), &imagerequest.Request{
|
||||
Source: imagerequest.Source{Host: "example.com", Path: "/photo.jpg"},
|
||||
})
|
||||
if err != nil || found != key {
|
||||
t.Fatalf("LookupSource() = %q, %v; want %q", found, err, key)
|
||||
}
|
||||
|
||||
if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 0 {
|
||||
t.Error("LookupSource wrote the source's access time itself")
|
||||
}
|
||||
|
||||
if waiting := len(cache.pendingSourceAccessTimes); waiting != 1 {
|
||||
t.Errorf("%d access times of sources wait to be written, want 1", waiting)
|
||||
}
|
||||
|
||||
err = cache.StopPendingCountWrites(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("StopPendingCountWrites() error = %v", err)
|
||||
}
|
||||
|
||||
if countRows(t, cache, touchedQuery, string(key), anHourAgoInDatabase) != 1 {
|
||||
t.Error("the source's access time was not written")
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopPendingCountWrites_LogsWhatWasNotWritten makes the write of the
|
||||
// access times fail at shutdown, by dropping the table they are written to,
|
||||
// while a hit count is written. The stop must return an error and log that
|
||||
// access times were not written, and must not log the counts as not written.
|
||||
func TestStopPendingCountWrites_LogsWhatWasNotWritten(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
var logs bytes.Buffer
|
||||
|
||||
cache.log = slog.New(slog.NewJSONHandler(&logs, nil))
|
||||
|
||||
cache.touchVariant(testVariantKeyOne)
|
||||
cache.IncrementStats(true, 0)
|
||||
|
||||
_, err := cache.db.ExecContext(t.Context(), `DROP TABLE variant_content`)
|
||||
if err != nil {
|
||||
t.Fatalf("dropping variant_content: %v", err)
|
||||
}
|
||||
|
||||
err = cache.StopPendingCountWrites(t.Context())
|
||||
t.Logf("StopPendingCountWrites() error = %v; logged:\n%s", err, logs.String())
|
||||
|
||||
if err == nil {
|
||||
t.Error("StopPendingCountWrites() error = nil, want the access times' error")
|
||||
}
|
||||
|
||||
if !strings.Contains(logs.String(), "access times not written at shutdown") {
|
||||
t.Error("the access times not written were not logged")
|
||||
}
|
||||
|
||||
if strings.Contains(logs.String(), "counts not written at shutdown") {
|
||||
t.Error("the counts, which were written, were logged as not written")
|
||||
}
|
||||
|
||||
if got := readCacheStatsCounters(t, cache); got.hitCount != 1 {
|
||||
t.Errorf("hit count in the database = %d, want 1", got.hitCount)
|
||||
}
|
||||
}
|
||||
@@ -142,8 +142,8 @@ var ErrProcessingTimedOut = errors.New(
|
||||
// only a request that needs the source fetched is refused.
|
||||
//
|
||||
// A cache hit opens the variant file and makes no database call: its access
|
||||
// time and its hit count are handed to the goroutine that
|
||||
// StartPendingCountWrites starts, without waiting for it. A variant file
|
||||
// time is handed to the goroutine that StartPendingAccessTimeWrites starts,
|
||||
// without waiting for it, and its hit is counted in memory. A variant file
|
||||
// that cannot be served, such as one whose first bytes are not those of an
|
||||
// output format, is a miss, and the variant is made again.
|
||||
func (s *Service) Get(
|
||||
@@ -196,11 +196,6 @@ func (s *Service) Warm(ctx context.Context, req *imagerequest.Request) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// Stats returns cache statistics.
|
||||
func (s *Service) Stats(ctx context.Context) (*CacheStats, error) {
|
||||
return s.cache.Stats(ctx)
|
||||
}
|
||||
|
||||
// WaitForProcessing waits until no image is being processed, or until ctx
|
||||
// ends, and returns how many images were still being processed then.
|
||||
func (s *Service) WaitForProcessing(ctx context.Context) int {
|
||||
|
||||
@@ -595,9 +595,7 @@ func TestService_Get_ServesSourceAndVariantCachedTwoDaysAgo(t *testing.T) {
|
||||
|
||||
_ = resp.Content.Close()
|
||||
|
||||
waitForPendingCountsWritten(t, svc.cache)
|
||||
|
||||
fetches := readCacheStatsCounters(t, svc.cache).upstreamFetchCount
|
||||
fetches := readCounters(svc.cache).upstreamFetchCount
|
||||
if fetches != 1 {
|
||||
t.Errorf("upstream fetches = %d, want 1", fetches)
|
||||
}
|
||||
|
||||
@@ -3,245 +3,18 @@ package imgcache
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"image/color"
|
||||
"io"
|
||||
"io/fs"
|
||||
"log/slog"
|
||||
"math"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/database"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imagerequest"
|
||||
)
|
||||
|
||||
func setupStatsTestDB(t *testing.T) *sql.DB {
|
||||
t.Helper()
|
||||
|
||||
db, err := sql.Open("sqlite", ":memory:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = database.ApplyMigrations(context.Background(), db, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
func TestStats_HitRateIsRatio(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db := setupStatsTestDB(t)
|
||||
dir := t.TempDir()
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: dir,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Set some hit/miss counts and a transform_count
|
||||
_, err = db.ExecContext(ctx, `
|
||||
UPDATE cache_stats
|
||||
SET hit_count = 75, miss_count = 25, transform_count = 9999
|
||||
WHERE id = 1
|
||||
`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
stats, err := cache.Stats(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if stats.HitCount != 75 {
|
||||
t.Errorf("HitCount = %d, want 75", stats.HitCount)
|
||||
}
|
||||
|
||||
if stats.MissCount != 25 {
|
||||
t.Errorf("MissCount = %d, want 25", stats.MissCount)
|
||||
}
|
||||
|
||||
// HitRate should be 0.75, NOT 9999 (transform_count)
|
||||
expectedRate := 0.75
|
||||
if math.Abs(stats.HitRate-expectedRate) > 0.001 {
|
||||
t.Errorf("HitRate = %f, want %f (was it scanning transform_count?)",
|
||||
stats.HitRate, expectedRate)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStats_ZeroCounts(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db := setupStatsTestDB(t)
|
||||
dir := t.TempDir()
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: dir,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
stats, err := cache.Stats(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// With zero hits and misses, HitRate should be 0, not some garbage value
|
||||
if stats.HitRate != 0.0 {
|
||||
t.Errorf("HitRate = %f, want 0.0 for zero counts", stats.HitRate)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStats_LogsFailedCountQueries verifies that a failed item count query
|
||||
// and a failed size query are each logged at warn and Stats still succeeds.
|
||||
func TestStats_LogsFailedCountQueries(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db := setupStatsTestDB(t)
|
||||
|
||||
var logBuf bytes.Buffer
|
||||
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
Logger: slog.New(slog.NewJSONHandler(&logBuf, nil)),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
_, err = db.ExecContext(t.Context(),
|
||||
`DROP TABLE source_content; DROP TABLE variant_content`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
_, err = cache.Stats(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("Stats() error = %v, want nil", err)
|
||||
}
|
||||
|
||||
for _, msg := range []string{
|
||||
"failed to count cache items for stats",
|
||||
"failed to sum cache size for stats",
|
||||
} {
|
||||
want := `"level":"WARN","msg":"` + msg + `"`
|
||||
if !strings.Contains(logBuf.String(), want) {
|
||||
t.Errorf("log missing %s; got %q", want, logBuf.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestStats_TotalsCountSourcesAndVariants verifies that TotalItems and
|
||||
// TotalSizeBytes cover the stored source images and processed variants.
|
||||
func TestStats_TotalsCountSourcesAndVariants(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
storeEvictionTestSource(t, cache, testHostCDN, testPathCat,
|
||||
bytes.Repeat([]byte{0xAA}, 1000))
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyOne,
|
||||
bytes.Repeat([]byte{0xAB}, 500))
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyTwo,
|
||||
bytes.Repeat([]byte{0xAC}, 250))
|
||||
|
||||
stats, err := cache.Stats(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("Stats() error = %v", err)
|
||||
}
|
||||
|
||||
if stats.TotalItems != 3 {
|
||||
t.Errorf("TotalItems = %d, want 3 (1 source, 2 variants)", stats.TotalItems)
|
||||
}
|
||||
|
||||
if stats.TotalSizeBytes != 1750 {
|
||||
t.Errorf("TotalSizeBytes = %d, want 1750 (1000+500+250)",
|
||||
stats.TotalSizeBytes)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStats_DisabledCacheReportsNoItems verifies that a disabled disk cache
|
||||
// reports no items and no size, even when its database still holds the
|
||||
// rows of an earlier run with the disk cache enabled.
|
||||
func TestStats_DisabledCacheReportsNoItems(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
enabled, _ := newEvictionTestCache(t, 1<<30)
|
||||
|
||||
storeEvictionTestSource(t, enabled, testHostCDN, testPathCat,
|
||||
bytes.Repeat([]byte{0xAA}, 1000))
|
||||
storeEvictionTestVariant(t, enabled, testVariantKeyOne,
|
||||
bytes.Repeat([]byte{0xAB}, 500))
|
||||
|
||||
disabled, err := NewCache(enabled.db, CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
DisableDiskCache: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
stats, err := disabled.Stats(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("Stats() error = %v", err)
|
||||
}
|
||||
|
||||
if stats.TotalItems != 0 || stats.TotalSizeBytes != 0 {
|
||||
t.Errorf("TotalItems = %d, TotalSizeBytes = %d, want 0 and 0",
|
||||
stats.TotalItems, stats.TotalSizeBytes)
|
||||
}
|
||||
}
|
||||
|
||||
// cacheStatsCounters holds the counters of the cache_stats row, in column
|
||||
// order.
|
||||
type cacheStatsCounters struct {
|
||||
hitCount int64
|
||||
missCount int64
|
||||
upstreamFetchCount int64
|
||||
upstreamFetchBytes int64
|
||||
transformCount int64
|
||||
}
|
||||
|
||||
// readCacheStatsCounters reads the counters of the cache_stats row.
|
||||
func readCacheStatsCounters(t *testing.T, cache *Cache) cacheStatsCounters {
|
||||
t.Helper()
|
||||
|
||||
var got cacheStatsCounters
|
||||
|
||||
err := cache.db.QueryRowContext(t.Context(), `
|
||||
SELECT hit_count, miss_count, upstream_fetch_count,
|
||||
upstream_fetch_bytes, transform_count
|
||||
FROM cache_stats WHERE id = 1
|
||||
`).Scan(&got.hitCount, &got.missCount, &got.upstreamFetchCount,
|
||||
&got.upstreamFetchBytes, &got.transformCount)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to read cache_stats: %v", err)
|
||||
}
|
||||
|
||||
return got
|
||||
}
|
||||
|
||||
// TestService_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 cache_stats counter after each.
|
||||
// source cannot be used, checking every counter of the cache after each.
|
||||
func TestService_Get_CountsStats(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -268,18 +41,18 @@ func TestService_Get_CountsStats(t *testing.T) {
|
||||
path string
|
||||
size int
|
||||
wantErr bool
|
||||
want cacheStatsCounters
|
||||
want counterValues
|
||||
}{
|
||||
{"miss that fetches the source", testPathPhoto, 50, false,
|
||||
cacheStatsCounters{0, 1, 1, photoBytes, 1}},
|
||||
counterValues{0, 1, 1, photoBytes, 1}},
|
||||
{"hit", testPathPhoto, 50, false,
|
||||
cacheStatsCounters{1, 1, 1, photoBytes, 1}},
|
||||
counterValues{1, 1, 1, photoBytes, 1}},
|
||||
{"miss that reuses the cached source", testPathPhoto, 25, false,
|
||||
cacheStatsCounters{1, 2, 1, photoBytes, 2}},
|
||||
counterValues{1, 2, 1, photoBytes, 2}},
|
||||
{"miss whose source fails the magic byte check", "/images/fake.jpg", 50, true,
|
||||
cacheStatsCounters{1, 3, 2, photoBytes + fakeBytes, 2}},
|
||||
counterValues{1, 3, 2, photoBytes + fakeBytes, 2}},
|
||||
{"miss whose source is not found", "/images/nonexistent.jpg", 50, true,
|
||||
cacheStatsCounters{1, 4, 2, photoBytes + fakeBytes, 2}},
|
||||
counterValues{1, 4, 2, photoBytes + fakeBytes, 2}},
|
||||
}
|
||||
|
||||
for _, step := range steps {
|
||||
@@ -303,9 +76,7 @@ func TestService_Get_CountsStats(t *testing.T) {
|
||||
_ = resp.Content.Close()
|
||||
}
|
||||
|
||||
waitForPendingCountsWritten(t, svc.cache)
|
||||
|
||||
got := readCacheStatsCounters(t, svc.cache)
|
||||
got := readCounters(svc.cache)
|
||||
if got != step.want {
|
||||
t.Fatalf("after the %s: counters = %+v, want %+v", step.name, got, step.want)
|
||||
}
|
||||
@@ -343,7 +114,7 @@ func (u *fakeUpstream) Read(p []byte) (int, error) {
|
||||
return 0, u.err
|
||||
}
|
||||
|
||||
// TestService_Get_CountsInterruptedMisses checks every cache_stats counter
|
||||
// TestService_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.
|
||||
func TestService_Get_CountsInterruptedMisses(t *testing.T) {
|
||||
@@ -359,15 +130,15 @@ func TestService_Get_CountsInterruptedMisses(t *testing.T) {
|
||||
cancel bool // whether the body then ends the request context
|
||||
readErr error // what the body then returns
|
||||
wantErr bool
|
||||
want cacheStatsCounters
|
||||
want counterValues
|
||||
}{
|
||||
{"request context ends during the fetch", half, true, context.Canceled, true,
|
||||
cacheStatsCounters{0, 1, 1, int64(half), 0}},
|
||||
counterValues{0, 1, 1, int64(half), 0}},
|
||||
{"request context ends after the fetch", len(photo), true, io.EOF, false,
|
||||
cacheStatsCounters{0, 1, 1, int64(len(photo)), 1}},
|
||||
counterValues{0, 1, 1, int64(len(photo)), 1}},
|
||||
{"upstream body over the size limit", half, false,
|
||||
httpfetcher.ErrResponseTooLarge, true,
|
||||
cacheStatsCounters{0, 1, 1, int64(half), 0}},
|
||||
counterValues{0, 1, 1, int64(half), 0}},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
@@ -411,9 +182,7 @@ func TestService_Get_CountsInterruptedMisses(t *testing.T) {
|
||||
_ = resp.Content.Close()
|
||||
}
|
||||
|
||||
waitForPendingCountsWritten(t, svc.cache)
|
||||
|
||||
got := readCacheStatsCounters(t, svc.cache)
|
||||
got := readCounters(svc.cache)
|
||||
if got != tc.want {
|
||||
t.Errorf("counters = %+v, want %+v", got, tc.want)
|
||||
}
|
||||
@@ -421,9 +190,9 @@ func TestService_Get_CountsInterruptedMisses(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Get_CountsHitAfterRequestEnds checks every cache_stats counter
|
||||
// after a hit served with a request context that has already ended: only
|
||||
// the hit count moves.
|
||||
// TestService_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.
|
||||
func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -450,9 +219,7 @@ func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) {
|
||||
|
||||
_ = resp.Content.Close()
|
||||
|
||||
waitForPendingCountsWritten(t, svc.cache)
|
||||
|
||||
want := readCacheStatsCounters(t, svc.cache)
|
||||
want := readCounters(svc.cache)
|
||||
want.hitCount++
|
||||
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
@@ -469,9 +236,7 @@ func TestService_Get_CountsHitAfterRequestEnds(t *testing.T) {
|
||||
t.Fatalf("CacheStatus = %v, want %v", resp.CacheStatus, CacheHit)
|
||||
}
|
||||
|
||||
waitForPendingCountsWritten(t, svc.cache)
|
||||
|
||||
got := readCacheStatsCounters(t, svc.cache)
|
||||
got := readCounters(svc.cache)
|
||||
if got != want {
|
||||
t.Errorf("counters = %+v, want %+v", got, want)
|
||||
}
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"testing/fstest"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"sneak.berlin/go/pixa/internal/database"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
)
|
||||
@@ -184,7 +185,7 @@ func SetupTestService(
|
||||
cache, err := NewCache(db, CacheConfig{
|
||||
StateDir: tmpDir,
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
}, prometheus.NewRegistry())
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"go.uber.org/fx"
|
||||
"go.uber.org/fx/fxtest"
|
||||
|
||||
@@ -170,6 +171,7 @@ func startImageProxy(
|
||||
handlers.New,
|
||||
middleware.New,
|
||||
func() httpfetcher.Fetcher { return fetcher },
|
||||
func() prometheus.Registerer { return prometheus.NewRegistry() },
|
||||
),
|
||||
fx.Populate(&h, &mw, &db),
|
||||
)
|
||||
@@ -223,6 +225,7 @@ func startImageProxyWithBlockedNetworks(
|
||||
healthcheck.New,
|
||||
handlers.New,
|
||||
middleware.New,
|
||||
func() prometheus.Registerer { return prometheus.NewRegistry() },
|
||||
),
|
||||
fx.Populate(&h, &mw),
|
||||
)
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"go.uber.org/fx/fxtest"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/config"
|
||||
@@ -81,6 +82,7 @@ func newTestServer(t *testing.T) *Server {
|
||||
|
||||
h, err := handlers.New(lc, handlers.Params{
|
||||
Logger: log, Healthcheck: hc, Database: db, Config: cfg,
|
||||
MetricsRegistry: prometheus.NewRegistry(),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("handlers.New() error = %v", err)
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"go.uber.org/fx"
|
||||
"go.uber.org/fx/fxtest"
|
||||
|
||||
@@ -46,6 +47,7 @@ func startPixa(t *testing.T, cfg *config.Config) *Server {
|
||||
healthcheck.New,
|
||||
handlers.New,
|
||||
middleware.New,
|
||||
func() prometheus.Registerer { return prometheus.NewRegistry() },
|
||||
),
|
||||
fx.Populate(&h, &mw),
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user