melosh/mal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'integrate-codex-work'

Browse Source
This commit is contained in:
mkelvers
2026-06-21 17:22:24 +02:00
parent 07f6080451 a5875de32e
commit 7010f72a26
12 changed files with 442 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
integrations/jikan/cache/store.go vendored
View File

@@ -22,6 +22,7 @@ func NewStore(queries db.Querier, metrics *observability.Metrics) *Store {
func (s *Store) Get(parentCtx context.Context, key string, out any) bool {
ctx, cancel := context.WithTimeout(parentCtx, 2*time.Second)
defer cancel()
defer s.observeStats(parentCtx)
data, err := s.db.GetJikanCache(ctx, key)
if err != nil {
@@ -42,6 +43,7 @@ func (s *Store) Get(parentCtx context.Context, key string, out any) bool {
func (s *Store) GetStale(parentCtx context.Context, key string, out any) bool {
ctx, cancel := context.WithTimeout(parentCtx, 2*time.Second)
defer cancel()
defer s.observeStats(parentCtx)
data, err := s.db.GetJikanCacheStale(ctx, key)
if err != nil {
@@ -62,6 +64,7 @@ func (s *Store) GetStale(parentCtx context.Context, key string, out any) bool {
func (s *Store) Set(parentCtx context.Context, key string, data any, ttl time.Duration) {
ctx, cancel := context.WithTimeout(parentCtx, 2*time.Second)
defer cancel()
defer s.observeStats(parentCtx)
bytes, err := json.Marshal(data)
if err != nil {
@@ -84,3 +87,22 @@ func (s *Store) Set(parentCtx context.Context, key string, data any, ttl time.Du
)
}
}
func (s *Store) observeStats(parentCtx context.Context) {
ctx, cancel := context.WithTimeout(parentCtx, 2*time.Second)
defer cancel()
stats, err := s.db.GetJikanCacheStats(ctx)
if err != nil {
observability.LogJSON(
observability.LogLevelWarn,
"jikan_cache_stats",
"jikan",
"",
nil,
err,
)
return
}
s.metrics.ObserveJikanCacheStats(stats.TotalRows, stats.ExpiredRows, stats.OldestExpiresAtSeconds)
}

74
integrations/jikan/client_test.go
View File

@@ -86,6 +86,59 @@ func TestGetWithCacheAllowsEmptySearchResults(t *testing.T) {
}
}
func TestLoadCachedRandomPoolIgnoresExpiredAnimeCache(t *testing.T) {
sqlDB := newTestCacheDB(t)
defer func() {
if err := sqlDB.Close(); err != nil {
t.Errorf("close sqlite: %v", err)
}
}()
queries := db.New(sqlDB)
client := NewClient(config.Config{}, queries, observability.NewMetrics())
insertCachedAnime(t, sqlDB, "anime:1", Anime{MalID: 1, Title: "fresh"}, time.Now().Add(time.Hour))
insertCachedAnime(t, sqlDB, "anime:2", Anime{MalID: 2, Title: "expired"}, time.Now().Add(-time.Hour))
client.loadCachedRandomPool(context.Background())
client.poolMu.RLock()
defer client.poolMu.RUnlock()
if len(client.randomPool) != 1 {
t.Fatalf("randomPool length = %d, want 1", len(client.randomPool))
}
if client.randomPool[0].MalID != 1 || client.randomPool[0].Title != "fresh" {
t.Fatalf("randomPool[0] = %+v, want fresh anime", client.randomPool[0])
}
}
func TestGetJikanCacheStatsCountsRowsAndExpiry(t *testing.T) {
sqlDB := newTestCacheDB(t)
defer func() {
if err := sqlDB.Close(); err != nil {
t.Errorf("close sqlite: %v", err)
}
}()
oldest := time.Date(2026, 6, 20, 12, 0, 0, 0, time.UTC)
insertCachedResponse(t, sqlDB, "expired", TopAnimeResponse{Data: []Anime{{MalID: 1}}}, oldest)
insertCachedResponse(t, sqlDB, "fresh", TopAnimeResponse{Data: []Anime{{MalID: 2}}}, time.Now().Add(time.Hour))
stats, err := db.New(sqlDB).GetJikanCacheStats(context.Background())
if err != nil {
t.Fatalf("GetJikanCacheStats: %v", err)
}
if stats.TotalRows != 2 {
t.Fatalf("TotalRows = %d, want 2", stats.TotalRows)
}
if stats.ExpiredRows != 1 {
t.Fatalf("ExpiredRows = %d, want 1", stats.ExpiredRows)
}
if stats.OldestExpiresAtSeconds != oldest.Unix() {
t.Fatalf("OldestExpiresAtSeconds = %d, want %d", stats.OldestExpiresAtSeconds, oldest.Unix())
}
}
func newTestCacheDB(t *testing.T) *sql.DB {
t.Helper()
ctx := context.Background()
@@ -135,6 +188,27 @@ func insertCachedResponse(t *testing.T, sqlDB *sql.DB, key string, value TopAnim
}
}
func insertCachedAnime(t *testing.T, sqlDB *sql.DB, key string, value Anime, expiresAt time.Time) {
t.Helper()
ctx := context.Background()
encoded, err := json.Marshal(value)
if err != nil {
t.Fatalf("marshal cached anime: %v", err)
}
_, err = sqlDB.ExecContext(
ctx,
`INSERT INTO jikan_cache (key, data, expires_at) VALUES (?, ?, ?)`,
key,
string(encoded),
expiresAt,
)
if err != nil {
t.Fatalf("insert cached anime: %v", err)
}
}
func waitForFreshCache(t *testing.T, sqlDB *sql.DB, client *Client, key string) {
t.Helper()

1
internal/database/database.go
View File

@@ -22,6 +22,7 @@ var Module = fx.Options(
ProvideSQLDB,
ProvideQueries,
),
fx.Invoke(RegisterJikanCacheCleanupWorker),
)
func ProvideSQLDB(cfg config.Config) (*sql.DB, error) {

80
internal/database/database_test.go
View File

@@ -3,6 +3,8 @@ package database
import (
"context"
"database/sql"
"mal/internal/db"
"mal/internal/observability"
"testing"
_ "github.com/mattn/go-sqlite3"
@@ -43,3 +45,81 @@ func TestRunMigrationsCreatesHotPathIndexes(t *testing.T) {
})
}
}
func TestCleanupExpiredJikanCache(t *testing.T) {
sqlDB := newMigratedTestDB(t)
defer closeTestDB(t, sqlDB)
ctx := context.Background()
for _, row := range []struct {
key string
expiresAt string
}{
{key: "expired", expiresAt: "2000-01-01T00:00:00Z"},
{key: "fresh", expiresAt: "2999-01-01T00:00:00Z"},
} {
_, err := sqlDB.ExecContext(ctx, `INSERT INTO jikan_cache (key, data, expires_at) VALUES (?, ?, ?)`, row.key, "{}", row.expiresAt)
if err != nil {
t.Fatalf("insert %s cache row: %v", row.key, err)
}
}
cleanupExpiredJikanCache(ctx, db.New(sqlDB), observability.NewMetrics())
keys := jikanCacheKeys(ctx, t, sqlDB)
if len(keys) != 1 || keys[0] != "fresh" {
t.Fatalf("remaining cache keys = %v, want [fresh]", keys)
}
}
func newMigratedTestDB(t *testing.T) *sql.DB {
t.Helper()
sqlDB, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
sqlDB.SetMaxOpenConns(1)
if err := RunMigrations(sqlDB); err != nil {
closeTestDB(t, sqlDB)
t.Fatalf("RunMigrations: %v", err)
}
return sqlDB
}
func closeTestDB(t *testing.T, sqlDB *sql.DB) {
t.Helper()
if err := sqlDB.Close(); err != nil {
t.Errorf("close sqlite: %v", err)
}
}
func jikanCacheKeys(ctx context.Context, t *testing.T, sqlDB *sql.DB) []string {
t.Helper()
var keys []string
rows, err := sqlDB.QueryContext(ctx, `SELECT key FROM jikan_cache ORDER BY key`)
if err != nil {
t.Fatalf("query cache keys: %v", err)
}
defer func() {
if err := rows.Close(); err != nil {
t.Errorf("close rows: %v", err)
}
}()
for rows.Next() {
var key string
if err := rows.Scan(&key); err != nil {
t.Fatalf("scan key: %v", err)
}
keys = append(keys, key)
}
if err := rows.Err(); err != nil {
t.Fatalf("iterate keys: %v", err)
}
return keys
}

72
internal/database/jikan_cache_cleanup.go Normal file
View File

@@ -0,0 +1,72 @@
package database
import (
"context"
"mal/internal/db"
"mal/internal/observability"
"time"
"go.uber.org/fx"
)
const (
jikanCacheCleanupInterval = time.Hour
jikanCacheCleanupTimeout = 30 * time.Second
jikanCacheCleanupWorker = "jikan_cache_cleanup"
)
func RegisterJikanCacheCleanupWorker(lc fx.Lifecycle, queries *db.Queries, metrics *observability.Metrics) {
ctx, cancel := context.WithCancel(context.Background())
lc.Append(fx.Hook{
OnStart: func(startCtx context.Context) error {
go func() {
<-startCtx.Done()
cancel()
}()
go runJikanCacheCleanupWorker(ctx, queries, metrics)
return nil
},
OnStop: func(context.Context) error {
cancel()
return nil
},
})
}
func runJikanCacheCleanupWorker(ctx context.Context, queries *db.Queries, metrics *observability.Metrics) {
observability.Info("jikan_cache_cleanup_worker_start", "database", "", nil)
ticker := time.NewTicker(jikanCacheCleanupInterval)
defer ticker.Stop()
for {
cleanupExpiredJikanCache(ctx, queries, metrics)
select {
case <-ticker.C:
case <-ctx.Done():
observability.Info("jikan_cache_cleanup_worker_stop", "database", "", nil)
return
}
}
}
func cleanupExpiredJikanCache(ctx context.Context, queries *db.Queries, metrics *observability.Metrics) {
cleanupCtx, cancel := context.WithTimeout(ctx, jikanCacheCleanupTimeout)
defer cancel()
err := queries.DeleteExpiredJikanCache(cleanupCtx)
metrics.ObserveWorkerTick(jikanCacheCleanupWorker, err)
if err != nil {
observability.Warn(
"jikan_cache_cleanup_failed",
"database",
"",
map[string]any{
"worker": jikanCacheCleanupWorker,
},
err,
)
}
}

2
internal/db/querier.go
View File

@@ -15,6 +15,7 @@ type Querier interface {
CreateSession(ctx context.Context, arg CreateSessionParams) (Session, error)
DeleteAnimeFetchRetry(ctx context.Context, animeID int64) error
DeleteContinueWatchingEntry(ctx context.Context, arg DeleteContinueWatchingEntryParams) error
DeleteExpiredFailedEpisodeProviderMappings(ctx context.Context) error
DeleteExpiredJikanCache(ctx context.Context) error
DeleteSession(ctx context.Context, id string) error
DeleteWatchListEntry(ctx context.Context, arg DeleteWatchListEntryParams) error
@@ -30,6 +31,7 @@ type Querier interface {
GetEpisodeAvailabilityCache(ctx context.Context, animeID int64) (EpisodeAvailabilityCache, error)
GetEpisodeProviderMapping(ctx context.Context, arg GetEpisodeProviderMappingParams) (EpisodeProviderMapping, error)
GetJikanCache(ctx context.Context, key string) (string, error)
GetJikanCacheStats(ctx context.Context) (GetJikanCacheStatsRow, error)
GetJikanCacheStale(ctx context.Context, key string) (string, error)
GetSession(ctx context.Context, id string) (Session, error)
GetTrackedAiringAnimeIDsDueForEpisodeRefresh(ctx context.Context, limit int64) ([]int64, error)

16
internal/db/queries.sql
View File

@@ -233,7 +233,14 @@ WHERE key = ? AND datetime(expires_at) > CURRENT_TIMESTAMP LIMIT 1;
-- name: GetJikanCacheStale :one
SELECT data FROM jikan_cache
WHERE key = ? LIMIT 1;
WHERE key = ? AND datetime(expires_at) > datetime(CURRENT_TIMESTAMP, '-14 days') LIMIT 1;
-- name: GetJikanCacheStats :one
SELECT
COUNT(*) AS total_rows,
COUNT(*) FILTER (WHERE datetime(expires_at) <= CURRENT_TIMESTAMP) AS expired_rows,
COALESCE(unixepoch(MIN(expires_at)), 0) AS oldest_expires_at_seconds
FROM jikan_cache;
-- name: SetJikanCache :exec
INSERT INTO jikan_cache (key, data, expires_at)
@@ -333,6 +340,11 @@ SELECT anime_id, provider, provider_show_id, failed_until, last_error, updated_a
FROM episode_provider_mapping
WHERE anime_id = ? AND provider = ? LIMIT 1;
-- name: DeleteExpiredFailedEpisodeProviderMappings :exec
DELETE FROM episode_provider_mapping
WHERE provider_show_id = ''
AND failed_until <= CURRENT_TIMESTAMP;
-- name: GetTrackedAiringAnimeIDsDueForEpisodeRefresh :many
WITH tracked AS (
SELECT DISTINCT w.anime_id
@@ -357,4 +369,4 @@ LIMIT ?;
-- name: GetAllCachedAnime :many
SELECT data FROM jikan_cache
WHERE key LIKE 'anime:%' LIMIT 1000;
WHERE key LIKE 'anime:%' AND datetime(expires_at) > CURRENT_TIMESTAMP LIMIT 1000;

36
internal/db/queries.sql.go
View File

@@ -149,6 +149,17 @@ func (q *Queries) DeleteContinueWatchingEntry(ctx context.Context, arg DeleteCon
return err
}
const deleteExpiredFailedEpisodeProviderMappings = `-- name: DeleteExpiredFailedEpisodeProviderMappings :exec
DELETE FROM episode_provider_mapping
WHERE provider_show_id = ''
AND failed_until <= CURRENT_TIMESTAMP
`
func (q *Queries) DeleteExpiredFailedEpisodeProviderMappings(ctx context.Context) error {
_, err := q.db.ExecContext(ctx, deleteExpiredFailedEpisodeProviderMappings)
return err
}
const deleteExpiredJikanCache = `-- name: DeleteExpiredJikanCache :exec
DELETE FROM jikan_cache WHERE datetime(expires_at) <= CURRENT_TIMESTAMP
`
@@ -227,7 +238,7 @@ func (q *Queries) GetAPITokenByHash(ctx context.Context, tokenHash string) (ApiT
const getAllCachedAnime = `-- name: GetAllCachedAnime :many
SELECT data FROM jikan_cache
WHERE key LIKE 'anime:%' LIMIT 1000
WHERE key LIKE 'anime:%' AND datetime(expires_at) > CURRENT_TIMESTAMP LIMIT 1000
`
func (q *Queries) GetAllCachedAnime(ctx context.Context) ([]string, error) {
@@ -565,9 +576,30 @@ func (q *Queries) GetJikanCache(ctx context.Context, key string) (string, error)
return data, err
}
const getJikanCacheStats = `-- name: GetJikanCacheStats :one
SELECT
COUNT(*) AS total_rows,
COUNT(*) FILTER (WHERE datetime(expires_at) <= CURRENT_TIMESTAMP) AS expired_rows,
COALESCE(unixepoch(MIN(expires_at)), 0) AS oldest_expires_at_seconds
FROM jikan_cache
`
type GetJikanCacheStatsRow struct {
TotalRows int64 `json:"total_rows"`
ExpiredRows int64 `json:"expired_rows"`
OldestExpiresAtSeconds int64 `json:"oldest_expires_at_seconds"`
}
func (q *Queries) GetJikanCacheStats(ctx context.Context) (GetJikanCacheStatsRow, error) {
row := q.db.QueryRowContext(ctx, getJikanCacheStats)
var i GetJikanCacheStatsRow
err := row.Scan(&i.TotalRows, &i.ExpiredRows, &i.OldestExpiresAtSeconds)
return i, err
}
const getJikanCacheStale = `-- name: GetJikanCacheStale :one
SELECT data FROM jikan_cache
WHERE key = ? LIMIT 1
WHERE key = ? AND datetime(expires_at) > datetime(CURRENT_TIMESTAMP, '-14 days') LIMIT 1
`
func (q *Queries) GetJikanCacheStale(ctx context.Context, key string) (string, error) {

3
internal/db/sqlite.go
View File

@@ -19,6 +19,9 @@ func Open(dbFile string) (*sql.DB, error) {
if err != nil {
return nil, fmt.Errorf("failed to open db: %w", err)
}
db.SetMaxOpenConns(1)
db.SetMaxIdleConns(1)
// WAL improves concurrency between readers and writers.
if _, err := db.ExecContext(context.Background(), "PRAGMA journal_mode=WAL;"); err != nil {
return nil, fmt.Errorf("failed to enable WAL mode: %w", err)

94
internal/episodes/service/cache_store.go
View File

@@ -132,41 +132,7 @@ func (s *EpisodeService) markFailure(ctx context.Context, anime domain.Anime, ca
}
func (s *EpisodeService) getCached(ctx context.Context, anime domain.Anime) (domain.CanonicalEpisodeList, bool) {
row, err := s.queries.GetEpisodeAvailabilityCache(ctx, int64(anime.MalID))
if err != nil {
s.metrics.ObserveCache("episode_availability", "miss")
return domain.CanonicalEpisodeList{}, false
}
var payload domain.CanonicalEpisodeList
if err := json.Unmarshal([]byte(row.Data), &payload); err != nil {
s.metrics.ObserveCache("episode_availability", "miss")
observability.Warn(
"episodes_cached_payload_invalid",
"episodes",
"",
map[string]any{
"anime_id": anime.MalID,
},
err,
)
return domain.CanonicalEpisodeList{}, false
}
if !isCanonicalEpisodePayloadValid(payload, anime.Episodes) {
s.metrics.ObserveCache("episode_availability", "miss")
observability.Info(
"episodes_cached_payload_rejected",
"episodes",
"",
map[string]any{
"anime_id": anime.MalID,
"expected_count": anime.Episodes,
"cached_episodes": len(payload.Episodes),
},
)
return domain.CanonicalEpisodeList{}, false
}
s.metrics.ObserveCache("episode_availability", "hit")
return payload, true
return s.getDecodedCached(ctx, anime, "episode_availability")
}
func (s *EpisodeService) getFreshCached(ctx context.Context, anime domain.Anime) (domain.CanonicalEpisodeList, bool) {
@@ -181,24 +147,10 @@ func (s *EpisodeService) getFreshCached(ctx context.Context, anime domain.Anime)
return domain.CanonicalEpisodeList{}, false
}
payload, ok := s.decodeFreshCachedPayload(anime, row.Data)
payload, ok := s.decodeCachedPayload(anime, row.Data, "episode_availability_fresh")
if !ok {
return domain.CanonicalEpisodeList{}, false
}
if !isCanonicalEpisodePayloadValid(payload, anime.Episodes) {
s.metrics.ObserveCache("episode_availability_fresh", "miss")
observability.Info(
"episodes_cached_payload_rejected",
"episodes",
"",
map[string]any{
"anime_id": anime.MalID,
"expected_count": anime.Episodes,
"cached_episodes": len(payload.Episodes),
},
)
return domain.CanonicalEpisodeList{}, false
}
s.metrics.ObserveCache("episode_availability_fresh", "hit")
observability.Info(
"episodes_cache_served",
@@ -213,6 +165,20 @@ func (s *EpisodeService) getFreshCached(ctx context.Context, anime domain.Anime)
return payload, true
}
func (s *EpisodeService) getDecodedCached(ctx context.Context, anime domain.Anime, metric string) (domain.CanonicalEpisodeList, bool) {
row, err := s.queries.GetEpisodeAvailabilityCache(ctx, int64(anime.MalID))
if err != nil {
s.metrics.ObserveCache(metric, "miss")
return domain.CanonicalEpisodeList{}, false
}
payload, ok := s.decodeCachedPayload(anime, row.Data, metric)
if !ok {
return domain.CanonicalEpisodeList{}, false
}
s.metrics.ObserveCache(metric, "hit")
return payload, true
}
func (s *EpisodeService) isFreshEpisodeCache(anime domain.Anime, row db.EpisodeAvailabilityCache, now time.Time) bool {
if row.NextRefreshAt.Valid && !row.NextRefreshAt.Time.After(now) {
s.metrics.ObserveCache("episode_availability_fresh", "miss")
@@ -243,14 +209,10 @@ func (s *EpisodeService) isFreshEpisodeCache(anime domain.Anime, row db.EpisodeA
return true
}
func (s *EpisodeService) decodeFreshCachedPayload(anime domain.Anime, raw string) (domain.CanonicalEpisodeList, bool) {
func (s *EpisodeService) decodeCachedPayload(anime domain.Anime, raw string, metric string) (domain.CanonicalEpisodeList, bool) {
var payload domain.CanonicalEpisodeList
err := json.Unmarshal([]byte(raw), &payload)
if err == nil {
return payload, true
}
s.metrics.ObserveCache("episode_availability_fresh", "miss")
if err := json.Unmarshal([]byte(raw), &payload); err != nil {
s.metrics.ObserveCache(metric, "miss")
observability.Warn(
"episodes_cached_payload_invalid",
"episodes",
@@ -261,4 +223,22 @@ func (s *EpisodeService) decodeFreshCachedPayload(anime domain.Anime, raw string
err,
)
return domain.CanonicalEpisodeList{}, false
}
if !isCanonicalEpisodePayloadValid(payload, anime.Episodes) {
s.metrics.ObserveCache(metric, "miss")
observability.Info(
"episodes_cached_payload_rejected",
"episodes",
"",
map[string]any{
"anime_id": anime.MalID,
"expected_count": anime.Episodes,
"cached_episodes": len(payload.Episodes),
},
)
return domain.CanonicalEpisodeList{}, false
}
return payload, true
}

92
internal/observability/metrics.go
View File

@@ -37,6 +37,11 @@ type histogramSample struct {
sum float64
}
type gaugeSample struct {
labels map[string]string
value float64
}
type counterVec struct {
mu sync.Mutex
labelNames []string
@@ -50,6 +55,12 @@ type histogramVec struct {
samples map[string]*histogramSample
}
type gaugeVec struct {
mu sync.Mutex
labelNames []string
samples map[string]*gaugeSample
}
type Metrics struct {
httpRequests *counterVec
httpRequestLatency *histogramVec
@@ -59,6 +70,8 @@ type Metrics struct {
dbQueryLatency *histogramVec
workerTicks *counterVec
cacheOperations *counterVec
jikanCacheRows *gaugeVec
jikanCacheOldest *gaugeVec
}
func NewMetrics() *Metrics {
@@ -71,6 +84,8 @@ func NewMetrics() *Metrics {
dbQueryLatency: newHistogramVec(defaultDurationBuckets, "operation", "result"),
workerTicks: newCounterVec("worker", "result"),
cacheOperations: newCounterVec("cache", "result"),
jikanCacheRows: newGaugeVec("state"),
jikanCacheOldest: newGaugeVec(),
}
}
@@ -119,6 +134,12 @@ func (m *Metrics) ObserveCache(cache string, result string) {
m.cacheOperations.Inc(cache, result)
}
func (m *Metrics) ObserveJikanCacheStats(totalRows int64, expiredRows int64, oldestExpiresAtSeconds int64) {
m.jikanCacheRows.Set(float64(totalRows), "total")
m.jikanCacheRows.Set(float64(expiredRows), "expired")
m.jikanCacheOldest.Set(float64(oldestExpiresAtSeconds))
}
func (m *Metrics) writePrometheus(w http.ResponseWriter) error {
if err := writeCounterMetric(w, "mal_http_requests_total", "Total HTTP requests by method, route, and status.", m.httpRequests.snapshot()); err != nil {
return err
@@ -141,7 +162,13 @@ func (m *Metrics) writePrometheus(w http.ResponseWriter) error {
if err := writeCounterMetric(w, "mal_worker_ticks_total", "Total background worker ticks by worker and result.", m.workerTicks.snapshot()); err != nil {
return err
}
return writeCounterMetric(w, "mal_cache_operations_total", "Total cache hits and misses by cache name.", m.cacheOperations.snapshot())
if err := writeCounterMetric(w, "mal_cache_operations_total", "Total cache hits and misses by cache name.", m.cacheOperations.snapshot()); err != nil {
return err
}
if err := writeGaugeMetric(w, "mal_jikan_cache_rows", "Current jikan_cache row count by state.", m.jikanCacheRows.snapshot()); err != nil {
return err
}
return writeGaugeMetric(w, "mal_jikan_cache_oldest_expires_at_seconds", "Unix timestamp for the oldest jikan_cache expires_at value, or 0 when empty.", m.jikanCacheOldest.snapshot())
}
func newCounterVec(labelNames ...string) *counterVec {
@@ -237,6 +264,45 @@ func (h *histogramVec) snapshot() []histogramSample {
return out
}
func newGaugeVec(labelNames ...string) *gaugeVec {
return &gaugeVec{
labelNames: append([]string(nil), labelNames...),
samples: make(map[string]*gaugeSample),
}
}
func (g *gaugeVec) Set(value float64, labelValues ...string) {
g.mu.Lock()
defer g.mu.Unlock()
key, labels := buildLabelKey(g.labelNames, labelValues)
if labels == nil {
return
}
sample, ok := g.samples[key]
if !ok {
sample = &gaugeSample{labels: labels}
g.samples[key] = sample
}
sample.value = value
}
func (g *gaugeVec) snapshot() []gaugeSample {
g.mu.Lock()
defer g.mu.Unlock()
keys := sortedGaugeSampleKeys(g.samples)
out := make([]gaugeSample, 0, len(keys))
for _, key := range keys {
sample := g.samples[key]
out = append(out, gaugeSample{
labels: copyLabels(sample.labels),
value: sample.value,
})
}
return out
}
func buildLabelKey(labelNames []string, labelValues []string) (string, map[string]string) {
if len(labelNames) != len(labelValues) {
return "", nil
@@ -276,6 +342,15 @@ func sortedHistogramSampleKeys(samples map[string]*histogramSample) []string {
return keys
}
func sortedGaugeSampleKeys(samples map[string]*gaugeSample) []string {
keys := make([]string, 0, len(samples))
for key := range samples {
keys = append(keys, key)
}
sort.Strings(keys)
return keys
}
func writeCounterMetric(w http.ResponseWriter, name string, help string, samples []counterSample) error {
if _, err := fmt.Fprintf(w, "# HELP %s %s\n", name, help); err != nil {
return err
@@ -291,6 +366,21 @@ func writeCounterMetric(w http.ResponseWriter, name string, help string, samples
return nil
}
func writeGaugeMetric(w http.ResponseWriter, name string, help string, samples []gaugeSample) error {
if _, err := fmt.Fprintf(w, "# HELP %s %s\n", name, help); err != nil {
return err
}
if _, err := fmt.Fprintf(w, "# TYPE %s gauge\n", name); err != nil {
return err
}
for _, sample := range samples {
if _, err := fmt.Fprintf(w, "%s%s %s\n", name, formatLabels(sample.labels), formatFloat(sample.value)); err != nil {
return err
}
}
return nil
}
func writeHistogramMetric(w http.ResponseWriter, name string, help string, samples []histogramSample, bounds []float64) error {
if _, err := fmt.Fprintf(w, "# HELP %s %s\n", name, help); err != nil {
return err

4
internal/observability/metrics_test.go
View File

@@ -21,6 +21,7 @@ func TestMetricsHandlerRendersPrometheusFamilies(t *testing.T) {
metrics.ObserveWorkerTick("episodes_availability", nil)
metrics.ObserveCache("jikan", "hit")
metrics.ObserveCache("episode_availability", "miss")
metrics.ObserveJikanCacheStats(12, 3, 1770000000)
req := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/metrics", nil)
rec := httptest.NewRecorder()
@@ -39,6 +40,9 @@ func TestMetricsHandlerRendersPrometheusFamilies(t *testing.T) {
assertContains(t, text, `mal_db_query_duration_seconds_count{operation="query",result="success"} 1`)
assertContains(t, text, `mal_worker_ticks_total{result="success",worker="episodes_availability"} 1`)
assertContains(t, text, `mal_cache_operations_total{cache="episode_availability",result="miss"} 1`)
assertContains(t, text, `mal_jikan_cache_rows{state="total"} 12`)
assertContains(t, text, `mal_jikan_cache_rows{state="expired"} 3`)
assertContains(t, text, `mal_jikan_cache_oldest_expires_at_seconds 1770000000`)
}
func TestInstrumentDBRecordsQueryLatency(t *testing.T) {