From fe1a933fd0973addbf4c789690d69dc1d752c736 Mon Sep 17 00:00:00 2001 From: Admin Date: Sat, 28 Mar 2026 14:32:40 +0500 Subject: [PATCH] feat(queue): replace PocketBase polling with Asynq + Redis MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce a Redis-backed Asynq task queue so the runner consumes TTS jobs pushed by the backend instead of polling PocketBase. - backend/internal/asynqqueue: Producer and Consumer wrappers - backend/internal/runner: AsynqRunner mux, per-instance Prometheus registry (fixes duplicate-collector panic in tests), redisConnOpt - backend/internal/config: REDIS_ADDR / REDIS_PASSWORD env vars - backend/cmd/{backend,runner}/main.go: wire Redis when env set; fall back to legacy poll mode when unset - Caddyfile: caddy-l4 TCP proxy for redis.libnovel.cc:6380 → homelab - caddy/Dockerfile: add --with github.com/mholt/caddy-l4 - docker-compose.yml: Caddy exposes 6380, backend/runner get Redis env - homelab/runner/docker-compose.yml: Redis sidecar, runner depends_on - homelab/otel/grafana: Grafana dashboards (backend, catalogue, runner) and alerting rules / contact-points provisioning --- Caddyfile | 24 ++ backend/cmd/backend/main.go | 47 ++- backend/cmd/runner/main.go | 18 +- backend/go.mod | 12 + backend/go.sum | 32 ++ backend/internal/asynqqueue/consumer.go | 56 +++ backend/internal/asynqqueue/producer.go | 90 +++++ backend/internal/asynqqueue/tasks.go | 46 +++ backend/internal/config/config.go | 19 + backend/internal/runner/asynq_runner.go | 149 +++++++ backend/internal/runner/metrics.go | 95 +++-- backend/internal/runner/runner.go | 45 ++- caddy/Dockerfile | 3 +- docker-compose.yml | 13 +- .../provisioning/alerting/contact-points.yaml | 16 + .../alerting/notification-policy.yaml | 15 + .../grafana/provisioning/alerting/rules.yaml | 214 ++++++++++ .../provisioning/dashboards/backend.json | 338 ++++++++++++++++ .../provisioning/dashboards/catalogue.json | 275 +++++++++++++ .../provisioning/dashboards/runner.json | 377 ++++++++++++++++++ homelab/runner/docker-compose.yml | 32 +- 21 files changed, 1869 insertions(+), 47 deletions(-) create mode 100644 backend/internal/asynqqueue/consumer.go create mode 100644 backend/internal/asynqqueue/producer.go create mode 100644 backend/internal/asynqqueue/tasks.go create mode 100644 backend/internal/runner/asynq_runner.go create mode 100644 homelab/otel/grafana/provisioning/alerting/contact-points.yaml create mode 100644 homelab/otel/grafana/provisioning/alerting/notification-policy.yaml create mode 100644 homelab/otel/grafana/provisioning/alerting/rules.yaml create mode 100644 homelab/otel/grafana/provisioning/dashboards/backend.json create mode 100644 homelab/otel/grafana/provisioning/dashboards/catalogue.json create mode 100644 homelab/otel/grafana/provisioning/dashboards/runner.json diff --git a/Caddyfile b/Caddyfile index 6379318..2ab5d3c 100644 --- a/Caddyfile +++ b/Caddyfile @@ -234,3 +234,27 @@ search.libnovel.cc { import security_headers reverse_proxy meilisearch:7700 } + +# ── Redis TCP proxy: exposes homelab Redis over TLS for Asynq ───────────────── +# The backend (prod) connects to rediss://redis.libnovel.cc:6380 to enqueue +# Asynq jobs. Caddy terminates TLS (Let's Encrypt cert for redis.libnovel.cc) +# and proxies the raw TCP stream to the homelab Redis via this reverse proxy. +# +# NOTE: Redis is NOT running on the prod server — it runs on the homelab +# (192.168.0.109:6379) and is exposed to the internet via this Caddy proxy. +# The homelab Redis is protected by REDIS_PASSWORD (requirepass). +# +# Caddy layer4 app handles this; requires the caddy-l4 module in the build. +{ + layer4 { + redis.libnovel.cc:6380 { + route { + tls + proxy { + # Homelab Redis — replace with actual homelab IP or FQDN + upstream {$HOMELAB_REDIS_ADDR:192.168.0.109:6379} + } + } + } + } +} diff --git a/backend/cmd/backend/main.go b/backend/cmd/backend/main.go index eda95a4..553489c 100644 --- a/backend/cmd/backend/main.go +++ b/backend/cmd/backend/main.go @@ -22,12 +22,16 @@ import ( "time" "github.com/getsentry/sentry-go" + "github.com/hibiken/asynq" + "github.com/libnovel/backend/internal/asynqqueue" "github.com/libnovel/backend/internal/backend" "github.com/libnovel/backend/internal/config" "github.com/libnovel/backend/internal/kokoro" "github.com/libnovel/backend/internal/meili" "github.com/libnovel/backend/internal/otelsetup" + "github.com/libnovel/backend/internal/pockettts" "github.com/libnovel/backend/internal/storage" + "github.com/libnovel/backend/internal/taskqueue" ) // version and commit are set at build time via -ldflags. @@ -100,6 +104,15 @@ func run() error { kokoroClient = &noopKokoro{} } + // ── Pocket-TTS (voice list + sample generation; audio generation is the runner's job) ── + var pocketTTSClient pockettts.Client + if cfg.PocketTTS.URL != "" { + pocketTTSClient = pockettts.New(cfg.PocketTTS.URL) + log.Info("pocket-tts voices enabled", "url", cfg.PocketTTS.URL) + } else { + log.Info("POCKET_TTS_URL not set — pocket-tts voices unavailable in backend") + } + // ── Meilisearch (search reads only; indexing is the runner's job) ──────── var searchIndex meili.Client if cfg.Meilisearch.URL != "" { @@ -110,6 +123,24 @@ func run() error { searchIndex = meili.NoopClient{} } + // ── Task Producer ──────────────────────────────────────────────────────── + // When REDIS_ADDR is set the backend dual-writes: PocketBase record (audit) + // + Asynq job (immediate delivery). Otherwise it writes to PocketBase only + // and the runner picks up on the next poll tick. + var producer taskqueue.Producer = store + if cfg.Redis.Addr != "" { + redisOpt, parseErr := parseRedisOpt(cfg.Redis) + if parseErr != nil { + return fmt.Errorf("parse REDIS_ADDR: %w", parseErr) + } + asynqProducer := asynqqueue.NewProducer(store, redisOpt) + defer asynqProducer.Close() //nolint:errcheck + producer = asynqProducer + log.Info("backend: asynq task dispatch enabled", "addr", cfg.Redis.Addr) + } else { + log.Info("backend: poll-mode task dispatch (REDIS_ADDR not set)") + } + // ── Backend server ─────────────────────────────────────────────────────── srv := backend.New( backend.Config{ @@ -125,10 +156,11 @@ func run() error { PresignStore: store, ProgressStore: store, CoverStore: store, - Producer: store, + Producer: producer, TaskReader: store, SearchIndex: searchIndex, Kokoro: kokoroClient, + PocketTTS: pocketTTSClient, Log: log, }, ) @@ -165,3 +197,16 @@ func (n *noopKokoro) GenerateAudio(_ context.Context, _, _ string) ([]byte, erro func (n *noopKokoro) ListVoices(_ context.Context) ([]string, error) { return nil, nil } + +// parseRedisOpt converts a config.Redis into an asynq.RedisConnOpt. +// Handles full "redis://" / "rediss://" URLs and plain "host:port". +func parseRedisOpt(cfg config.Redis) (asynq.RedisConnOpt, error) { + addr := cfg.Addr + if len(addr) > 7 && (addr[:8] == "redis://" || (len(addr) > 8 && addr[:9] == "rediss://")) { + return asynq.ParseRedisURI(addr) + } + return asynq.RedisClientOpt{ + Addr: addr, + Password: cfg.Password, + }, nil +} diff --git a/backend/cmd/runner/main.go b/backend/cmd/runner/main.go index 47dc329..8dbc561 100644 --- a/backend/cmd/runner/main.go +++ b/backend/cmd/runner/main.go @@ -20,6 +20,7 @@ import ( "time" "github.com/getsentry/sentry-go" + "github.com/libnovel/backend/internal/asynqqueue" "github.com/libnovel/backend/internal/browser" "github.com/libnovel/backend/internal/config" "github.com/libnovel/backend/internal/kokoro" @@ -29,6 +30,7 @@ import ( "github.com/libnovel/backend/internal/pockettts" "github.com/libnovel/backend/internal/runner" "github.com/libnovel/backend/internal/storage" + "github.com/libnovel/backend/internal/taskqueue" ) // version and commit are set at build time via -ldflags. @@ -151,9 +153,23 @@ func run() error { MetricsAddr: cfg.Runner.MetricsAddr, CatalogueRefreshInterval: cfg.Runner.CatalogueRefreshInterval, SkipInitialCatalogueRefresh: cfg.Runner.SkipInitialCatalogueRefresh, + RedisAddr: cfg.Redis.Addr, + RedisPassword: cfg.Redis.Password, } + + // In Asynq mode the Consumer is a thin wrapper: claim/heartbeat/reap are + // no-ops, but FinishAudioTask / FinishScrapeTask / FailTask write back to + // PocketBase as before. + var consumer taskqueue.Consumer = store + if cfg.Redis.Addr != "" { + log.Info("runner: asynq mode — using Redis for task dispatch", "addr", cfg.Redis.Addr) + consumer = asynqqueue.NewConsumer(store) + } else { + log.Info("runner: poll mode — using PocketBase for task dispatch") + } + deps := runner.Dependencies{ - Consumer: store, + Consumer: consumer, BookWriter: store, BookReader: store, AudioStore: store, diff --git a/backend/go.mod b/backend/go.mod index bbf3a10..0be5bc0 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -9,6 +9,7 @@ require ( require ( github.com/andybalholm/brotli v1.1.1 // indirect + github.com/beorn7/perks v1.0.1 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect @@ -22,16 +23,25 @@ require ( github.com/golang-jwt/jwt/v5 v5.3.1 // indirect github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect + github.com/hibiken/asynq v0.26.0 // indirect + github.com/hibiken/asynq/x v0.0.0-20260203063626-d704b68a426d // indirect github.com/klauspost/compress v1.18.2 // indirect github.com/klauspost/cpuid/v2 v2.2.11 // indirect github.com/klauspost/crc32 v1.3.0 // indirect github.com/meilisearch/meilisearch-go v0.36.1 // indirect github.com/minio/crc64nvme v1.1.1 // indirect github.com/minio/md5-simd v1.1.2 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/philhofer/fwd v1.2.0 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/prometheus/client_golang v1.23.2 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.66.1 // indirect + github.com/prometheus/procfs v0.16.1 // indirect github.com/redis/go-redis/v9 v9.18.0 // indirect + github.com/robfig/cron/v3 v3.0.1 // indirect github.com/rs/xid v1.6.0 // indirect + github.com/spf13/cast v1.10.0 // indirect github.com/tinylib/msgp v1.6.1 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/bridges/otelslog v0.17.0 // indirect @@ -47,10 +57,12 @@ require ( go.opentelemetry.io/otel/trace v1.42.0 // indirect go.opentelemetry.io/proto/otlp v1.9.0 // indirect go.uber.org/atomic v1.11.0 // indirect + go.yaml.in/yaml/v2 v2.4.2 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/crypto v0.48.0 // indirect golang.org/x/sys v0.41.0 // indirect golang.org/x/text v0.34.0 // indirect + golang.org/x/time v0.14.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect google.golang.org/grpc v1.79.2 // indirect diff --git a/backend/go.sum b/backend/go.sum index bc91505..39bddd0 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -1,5 +1,7 @@ github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA= github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= @@ -27,6 +29,10 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= +github.com/hibiken/asynq v0.26.0 h1:1Zxr92MlDnb1Zt/QR5g2vSCqUS03i95lUfqx5X7/wrw= +github.com/hibiken/asynq v0.26.0/go.mod h1:Qk4e57bTnWDoyJ67VkchuV6VzSM9IQW2nPvAGuDyw58= +github.com/hibiken/asynq/x v0.0.0-20260203063626-d704b68a426d h1:Ld5m8EIK5QVOq/owOexKIbETij3skACg4eU1pArHsrw= +github.com/hibiken/asynq/x v0.0.0-20260203063626-d704b68a426d/go.mod h1:hhpStehaxSGg3ib9wJXzw5AXY1YS6lQ9BNavAgPbIhE= github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk= github.com/klauspost/compress v1.18.2/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= @@ -42,14 +48,36 @@ github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34= github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM= github.com/minio/minio-go/v7 v7.0.98 h1:MeAVKjLVz+XJ28zFcuYyImNSAh8Mq725uNW4beRisi0= github.com/minio/minio-go/v7 v7.0.98/go.mod h1:cY0Y+W7yozf0mdIclrttzo1Iiu7mEf9y7nk2uXqMOvM= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM= github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y= +github.com/prometheus/client_golang v1.20.5/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE= +github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= +github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= +github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E= +github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G1dc= +github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8= +github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= +github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= +github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc= +github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= +github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= +github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs= github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0= +github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= +github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU= github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= +github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= +github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo= github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= @@ -84,6 +112,8 @@ go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjce go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= +go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= @@ -94,6 +124,8 @@ golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= +golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 h1:JLQynH/LBHfCTSbDWl+py8C+Rg/k1OVH3xfcaiANuF0= google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:kSJwQxqmFXeo79zOmbrALdflXQeAYcUbgS7PbpMknCY= google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 h1:mWPCjDEyshlQYzBpMNHaEof6UX1PmHcaUODUywQ0uac= diff --git a/backend/internal/asynqqueue/consumer.go b/backend/internal/asynqqueue/consumer.go new file mode 100644 index 0000000..188cf2c --- /dev/null +++ b/backend/internal/asynqqueue/consumer.go @@ -0,0 +1,56 @@ +package asynqqueue + +import ( + "context" + "time" + + "github.com/libnovel/backend/internal/domain" + "github.com/libnovel/backend/internal/taskqueue" +) + +// Consumer wraps the PocketBase-backed Consumer for result write-back only. +// +// When using Asynq, the runner no longer polls for work — Asynq delivers +// tasks via the ServeMux handlers. The only Consumer operations the handlers +// need are: +// - FinishAudioTask / FinishScrapeTask — write result back to PocketBase +// - FailTask — mark PocketBase record as failed +// +// ClaimNextAudioTask, ClaimNextScrapeTask, HeartbeatTask, and ReapStaleTasks +// are all no-ops here because Asynq owns those responsibilities. +type Consumer struct { + pb taskqueue.Consumer // underlying PocketBase consumer (for write-back) +} + +// NewConsumer wraps an existing PocketBase Consumer. +func NewConsumer(pb taskqueue.Consumer) *Consumer { + return &Consumer{pb: pb} +} + +// ── Write-back (delegated to PocketBase) ────────────────────────────────────── + +func (c *Consumer) FinishScrapeTask(ctx context.Context, id string, result domain.ScrapeResult) error { + return c.pb.FinishScrapeTask(ctx, id, result) +} + +func (c *Consumer) FinishAudioTask(ctx context.Context, id string, result domain.AudioResult) error { + return c.pb.FinishAudioTask(ctx, id, result) +} + +func (c *Consumer) FailTask(ctx context.Context, id, errMsg string) error { + return c.pb.FailTask(ctx, id, errMsg) +} + +// ── No-ops (Asynq owns claiming / heartbeating / reaping) ─────────────────── + +func (c *Consumer) ClaimNextScrapeTask(_ context.Context, _ string) (domain.ScrapeTask, bool, error) { + return domain.ScrapeTask{}, false, nil +} + +func (c *Consumer) ClaimNextAudioTask(_ context.Context, _ string) (domain.AudioTask, bool, error) { + return domain.AudioTask{}, false, nil +} + +func (c *Consumer) HeartbeatTask(_ context.Context, _ string) error { return nil } + +func (c *Consumer) ReapStaleTasks(_ context.Context, _ time.Duration) (int, error) { return 0, nil } diff --git a/backend/internal/asynqqueue/producer.go b/backend/internal/asynqqueue/producer.go new file mode 100644 index 0000000..dec69ef --- /dev/null +++ b/backend/internal/asynqqueue/producer.go @@ -0,0 +1,90 @@ +package asynqqueue + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/hibiken/asynq" + "github.com/libnovel/backend/internal/taskqueue" +) + +// Producer dual-writes every task: first to PocketBase (via pb, for audit / +// UI status), then to Redis via Asynq so the runner picks it up immediately. +type Producer struct { + pb taskqueue.Producer // underlying PocketBase producer + client *asynq.Client +} + +// NewProducer wraps an existing PocketBase Producer with Asynq dispatch. +func NewProducer(pb taskqueue.Producer, redisOpt asynq.RedisConnOpt) *Producer { + return &Producer{ + pb: pb, + client: asynq.NewClient(redisOpt), + } +} + +// Close shuts down the underlying Asynq client connection. +func (p *Producer) Close() error { + return p.client.Close() +} + +// CreateScrapeTask creates a PocketBase record then enqueues an Asynq job. +func (p *Producer) CreateScrapeTask(ctx context.Context, kind, targetURL string, fromChapter, toChapter int) (string, error) { + id, err := p.pb.CreateScrapeTask(ctx, kind, targetURL, fromChapter, toChapter) + if err != nil { + return "", err + } + + payload := ScrapePayload{ + PBTaskID: id, + Kind: kind, + TargetURL: targetURL, + FromChapter: fromChapter, + ToChapter: toChapter, + } + taskType := TypeScrapeBook + if kind == "catalogue" { + taskType = TypeScrapeCatalogue + } + if err := p.enqueue(ctx, taskType, payload); err != nil { + // Non-fatal: PB record exists; runner will pick it up on next poll. + return id, fmt.Errorf("asynq enqueue scrape (task still in PB): %w", err) + } + return id, nil +} + +// CreateAudioTask creates a PocketBase record then enqueues an Asynq job. +func (p *Producer) CreateAudioTask(ctx context.Context, slug string, chapter int, voice string) (string, error) { + id, err := p.pb.CreateAudioTask(ctx, slug, chapter, voice) + if err != nil { + return "", err + } + + payload := AudioPayload{ + PBTaskID: id, + Slug: slug, + Chapter: chapter, + Voice: voice, + } + if err := p.enqueue(ctx, TypeAudioGenerate, payload); err != nil { + return id, fmt.Errorf("asynq enqueue audio (task still in PB): %w", err) + } + return id, nil +} + +// CancelTask delegates to PocketBase; Asynq jobs may already be running and +// cannot be reliably cancelled, so we only update the audit record. +func (p *Producer) CancelTask(ctx context.Context, id string) error { + return p.pb.CancelTask(ctx, id) +} + +// enqueue serialises payload and dispatches it to Asynq. +func (p *Producer) enqueue(_ context.Context, taskType string, payload any) error { + b, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("marshal payload: %w", err) + } + _, err = p.client.Enqueue(asynq.NewTask(taskType, b)) + return err +} diff --git a/backend/internal/asynqqueue/tasks.go b/backend/internal/asynqqueue/tasks.go new file mode 100644 index 0000000..f82344a --- /dev/null +++ b/backend/internal/asynqqueue/tasks.go @@ -0,0 +1,46 @@ +// Package asynqqueue provides Asynq-backed implementations of the +// taskqueue.Producer and taskqueue.Consumer interfaces. +// +// Architecture: +// - Producer: dual-writes — creates a PocketBase record for audit/UI, then +// enqueues an Asynq job so the runner picks it up immediately (sub-ms). +// - Consumer: thin wrapper used only for result write-back (FinishAudioTask, +// FinishScrapeTask, FailTask). ClaimNext*/Heartbeat/Reap are no-ops because +// Asynq owns those responsibilities. +// - Handlers: asynq.HandlerFunc wrappers that decode job payloads and invoke +// the existing runner logic (runScrapeTask / runAudioTask). +// +// Fallback: when REDIS_ADDR is empty the caller should use the plain +// storage.Store (PocketBase-polling) implementation unchanged. +package asynqqueue + +// Queue names — keep all jobs on the default queue for now. +// Add separate queues (e.g. "audio", "scrape") later if you need priority. +const QueueDefault = "default" + +// Task type constants used for Asynq routing. +const ( + TypeAudioGenerate = "audio:generate" + TypeScrapeBook = "scrape:book" + TypeScrapeCatalogue = "scrape:catalogue" +) + +// AudioPayload is the Asynq job payload for audio generation tasks. +type AudioPayload struct { + // PBTaskID is the PocketBase record ID created before enqueueing. + // The handler uses it to write results back via Consumer.FinishAudioTask. + PBTaskID string `json:"pb_task_id"` + Slug string `json:"slug"` + Chapter int `json:"chapter"` + Voice string `json:"voice"` +} + +// ScrapePayload is the Asynq job payload for scrape tasks. +type ScrapePayload struct { + // PBTaskID is the PocketBase record ID created before enqueueing. + PBTaskID string `json:"pb_task_id"` + Kind string `json:"kind"` // "catalogue", "book", or "book_range" + TargetURL string `json:"target_url"` // empty for catalogue tasks + FromChapter int `json:"from_chapter"` // 0 unless Kind=="book_range" + ToChapter int `json:"to_chapter"` // 0 unless Kind=="book_range" +} diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 4ae706b..03457ec 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -86,6 +86,19 @@ type Valkey struct { Addr string } +// Redis holds connection settings for the Asynq task queue Redis instance. +// This is separate from Valkey (presign cache) — it may point to the same +// Redis or a dedicated one. An empty Addr falls back to PocketBase polling. +type Redis struct { + // Addr is the host:port (or rediss://... URL) of the Redis instance. + // Use rediss:// scheme for TLS (e.g. rediss://:password@redis.libnovel.cc:6380). + // An empty string disables Asynq and falls back to PocketBase polling. + Addr string + // Password is the Redis AUTH password. + // Not needed when Addr is a full rediss:// URL that includes the password. + Password string +} + // Runner holds settings specific to the runner/worker binary. type Runner struct { // PollInterval is how often the runner checks PocketBase for pending tasks. @@ -125,6 +138,7 @@ type Config struct { Runner Runner Meilisearch Meilisearch Valkey Valkey + Redis Redis // LogLevel is one of "debug", "info", "warn", "error". LogLevel string } @@ -192,6 +206,11 @@ func Load() Config { Valkey: Valkey{ Addr: envOr("VALKEY_ADDR", ""), }, + + Redis: Redis{ + Addr: envOr("REDIS_ADDR", ""), + Password: envOr("REDIS_PASSWORD", ""), + }, } } diff --git a/backend/internal/runner/asynq_runner.go b/backend/internal/runner/asynq_runner.go new file mode 100644 index 0000000..c1f1b63 --- /dev/null +++ b/backend/internal/runner/asynq_runner.go @@ -0,0 +1,149 @@ +package runner + +// asynq_runner.go — Asynq-based task dispatch for the runner. +// +// When cfg.RedisAddr is set, Run() calls runAsynq() instead of runPoll(). +// The Asynq server replaces the polling loop: it listens on Redis for tasks +// enqueued by the backend Producer and delivers them immediately. +// +// Handlers in this file decode Asynq job payloads and call the existing +// runScrapeTask / runAudioTask methods, keeping all execution logic in one place. + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/hibiken/asynq" + asynqmetrics "github.com/hibiken/asynq/x/metrics" + "github.com/libnovel/backend/internal/asynqqueue" + "github.com/libnovel/backend/internal/domain" +) + +// runAsynq starts an Asynq server that replaces the PocketBase poll loop. +// It also starts the periodic catalogue refresh ticker. +// Blocks until ctx is cancelled. +func (r *Runner) runAsynq(ctx context.Context) error { + redisOpt, err := r.redisConnOpt() + if err != nil { + return fmt.Errorf("runner: parse redis addr: %w", err) + } + + srv := asynq.NewServer(redisOpt, asynq.Config{ + // Allocate concurrency slots for each task type. + // Total concurrency = scrape + audio slots. + Concurrency: r.cfg.MaxConcurrentScrape + r.cfg.MaxConcurrentAudio, + Queues: map[string]int{ + asynqqueue.QueueDefault: 1, + }, + // Let Asynq handle retries with exponential back-off. + RetryDelayFunc: asynq.DefaultRetryDelayFunc, + // Log errors from handlers via the existing structured logger. + ErrorHandler: asynq.ErrorHandlerFunc(func(_ context.Context, task *asynq.Task, err error) { + r.deps.Log.Error("runner: asynq task failed", + "type", task.Type(), + "err", err, + ) + }), + }) + + mux := asynq.NewServeMux() + mux.HandleFunc(asynqqueue.TypeAudioGenerate, r.handleAudioTask) + mux.HandleFunc(asynqqueue.TypeScrapeBook, r.handleScrapeTask) + mux.HandleFunc(asynqqueue.TypeScrapeCatalogue, r.handleScrapeTask) + + // Register Asynq queue metrics with the default Prometheus registry so + // the /metrics endpoint (metrics.go) can expose them. + inspector := asynq.NewInspector(redisOpt) + collector := asynqmetrics.NewQueueMetricsCollector(inspector) + if err := r.metricsRegistry.Register(collector); err != nil { + r.deps.Log.Warn("runner: could not register asynq prometheus collector", "err", err) + } + + // Start the periodic catalogue refresh. + catalogueTick := time.NewTicker(r.cfg.CatalogueRefreshInterval) + defer catalogueTick.Stop() + if !r.cfg.SkipInitialCatalogueRefresh { + go r.runCatalogueRefresh(ctx) + } else { + r.deps.Log.Info("runner: skipping initial catalogue refresh (RUNNER_SKIP_INITIAL_CATALOGUE_REFRESH=true)") + } + + r.deps.Log.Info("runner: asynq mode active", "redis_addr", r.cfg.RedisAddr) + + // Run catalogue refresh ticker in the background. + go func() { + for { + select { + case <-ctx.Done(): + return + case <-catalogueTick.C: + go r.runCatalogueRefresh(ctx) + } + } + }() + + // Start Asynq server (non-blocking). + if err := srv.Start(mux); err != nil { + return fmt.Errorf("runner: asynq server start: %w", err) + } + + // Block until context is cancelled, then gracefully stop. + <-ctx.Done() + r.deps.Log.Info("runner: context cancelled, shutting down asynq server") + srv.Shutdown() + return nil +} + +// redisConnOpt parses cfg.RedisAddr into an asynq.RedisConnOpt. +// Supports full "redis://" / "rediss://" URLs and plain "host:port". +func (r *Runner) redisConnOpt() (asynq.RedisConnOpt, error) { + addr := r.cfg.RedisAddr + // ParseRedisURI handles redis:// and rediss:// schemes. + if len(addr) > 7 && (addr[:8] == "redis://" || addr[:9] == "rediss://") { + return asynq.ParseRedisURI(addr) + } + // Plain "host:port" — use RedisClientOpt directly. + return asynq.RedisClientOpt{ + Addr: addr, + Password: r.cfg.RedisPassword, + }, nil +} + +// handleScrapeTask is the Asynq handler for TypeScrapeBook and TypeScrapeCatalogue. +func (r *Runner) handleScrapeTask(ctx context.Context, t *asynq.Task) error { + var p asynqqueue.ScrapePayload + if err := json.Unmarshal(t.Payload(), &p); err != nil { + return fmt.Errorf("unmarshal scrape payload: %w", err) + } + task := domain.ScrapeTask{ + ID: p.PBTaskID, + Kind: p.Kind, + TargetURL: p.TargetURL, + FromChapter: p.FromChapter, + ToChapter: p.ToChapter, + } + r.tasksRunning.Add(1) + defer r.tasksRunning.Add(-1) + r.runScrapeTask(ctx, task) + return nil +} + +// handleAudioTask is the Asynq handler for TypeAudioGenerate. +func (r *Runner) handleAudioTask(ctx context.Context, t *asynq.Task) error { + var p asynqqueue.AudioPayload + if err := json.Unmarshal(t.Payload(), &p); err != nil { + return fmt.Errorf("unmarshal audio payload: %w", err) + } + task := domain.AudioTask{ + ID: p.PBTaskID, + Slug: p.Slug, + Chapter: p.Chapter, + Voice: p.Voice, + } + r.tasksRunning.Add(1) + defer r.tasksRunning.Add(-1) + r.runAudioTask(ctx, task) + return nil +} diff --git a/backend/internal/runner/metrics.go b/backend/internal/runner/metrics.go index 05bff07..659b113 100644 --- a/backend/internal/runner/metrics.go +++ b/backend/internal/runner/metrics.go @@ -1,21 +1,28 @@ package runner -// metrics.go — lightweight HTTP metrics endpoint for the runner. +// metrics.go — Prometheus metrics HTTP endpoint for the runner. // -// GET /metrics returns a JSON document with live task counters and uptime. -// No external dependency (no Prometheus); plain net/http only. +// GET /metrics returns a Prometheus text/plain scrape response. +// Exposes: +// - Standard Go runtime metrics (via promhttp) +// - Runner task counters (tasks_running, tasks_completed, tasks_failed) +// - Asynq queue metrics (registered in asynq_runner.go when Redis is enabled) +// +// GET /health — simple liveness probe. import ( "context" - "encoding/json" "fmt" "log/slog" "net" "net/http" "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" ) -// metricsServer serves GET /metrics for the runner process. +// metricsServer serves GET /metrics and GET /health for the runner process. type metricsServer struct { addr string r *Runner @@ -23,21 +30,62 @@ type metricsServer struct { } func newMetricsServer(addr string, r *Runner, log *slog.Logger) *metricsServer { - return &metricsServer{addr: addr, r: r, log: log} + ms := &metricsServer{addr: addr, r: r, log: log} + ms.registerCollectors() + return ms +} + +// registerCollectors registers runner-specific Prometheus collectors. +// Called once at construction; Asynq queue collector is registered separately +// in asynq_runner.go after the Redis connection is established. +func (ms *metricsServer) registerCollectors() { + // Runner task gauges / counters backed by the atomic fields on Runner. + ms.r.metricsRegistry.MustRegister(prometheus.NewGaugeFunc( + prometheus.GaugeOpts{ + Namespace: "runner", + Name: "tasks_running", + Help: "Number of tasks currently being processed.", + }, + func() float64 { return float64(ms.r.tasksRunning.Load()) }, + )) + ms.r.metricsRegistry.MustRegister(prometheus.NewCounterFunc( + prometheus.CounterOpts{ + Namespace: "runner", + Name: "tasks_completed_total", + Help: "Total number of tasks completed successfully since startup.", + }, + func() float64 { return float64(ms.r.tasksCompleted.Load()) }, + )) + ms.r.metricsRegistry.MustRegister(prometheus.NewCounterFunc( + prometheus.CounterOpts{ + Namespace: "runner", + Name: "tasks_failed_total", + Help: "Total number of tasks that ended in failure since startup.", + }, + func() float64 { return float64(ms.r.tasksFailed.Load()) }, + )) + ms.r.metricsRegistry.MustRegister(prometheus.NewGaugeFunc( + prometheus.GaugeOpts{ + Namespace: "runner", + Name: "uptime_seconds", + Help: "Seconds since the runner process started.", + }, + func() float64 { return time.Since(ms.r.startedAt).Seconds() }, + )) } // ListenAndServe starts the HTTP server and blocks until ctx is cancelled or // a fatal listen error occurs. func (ms *metricsServer) ListenAndServe(ctx context.Context) error { mux := http.NewServeMux() - mux.HandleFunc("GET /metrics", ms.handleMetrics) + mux.Handle("GET /metrics", promhttp.HandlerFor(ms.r.metricsRegistry, promhttp.HandlerOpts{})) mux.HandleFunc("GET /health", ms.handleHealth) srv := &http.Server{ Addr: ms.addr, Handler: mux, ReadTimeout: 5 * time.Second, - WriteTimeout: 5 * time.Second, + WriteTimeout: 10 * time.Second, BaseContext: func(_ net.Listener) context.Context { return ctx }, } @@ -58,35 +106,8 @@ func (ms *metricsServer) ListenAndServe(ctx context.Context) error { } } -// handleMetrics handles GET /metrics. -// Response shape (JSON): -// -// { -// "tasks_running": N, -// "tasks_completed": N, -// "tasks_failed": N, -// "uptime_seconds": N -// } -func (ms *metricsServer) handleMetrics(w http.ResponseWriter, _ *http.Request) { - uptimeSec := int64(time.Since(ms.r.startedAt).Seconds()) - metricsWriteJSON(w, 0, map[string]int64{ - "tasks_running": ms.r.tasksRunning.Load(), - "tasks_completed": ms.r.tasksCompleted.Load(), - "tasks_failed": ms.r.tasksFailed.Load(), - "uptime_seconds": uptimeSec, - }) -} - -// handleHealth handles GET /health — simple liveness probe for the metrics server. +// handleHealth handles GET /health — simple liveness probe. func (ms *metricsServer) handleHealth(w http.ResponseWriter, _ *http.Request) { - metricsWriteJSON(w, 0, map[string]string{"status": "ok"}) -} - -// metricsWriteJSON writes v as a JSON response with the given status code. -func metricsWriteJSON(w http.ResponseWriter, status int, v any) { w.Header().Set("Content-Type", "application/json") - if status != 0 { - w.WriteHeader(status) - } - _ = json.NewEncoder(w).Encode(v) + _, _ = w.Write([]byte(`{"status":"ok"}`)) } diff --git a/backend/internal/runner/runner.go b/backend/internal/runner/runner.go index eea2378..181c536 100644 --- a/backend/internal/runner/runner.go +++ b/backend/internal/runner/runner.go @@ -34,6 +34,7 @@ import ( "github.com/libnovel/backend/internal/pockettts" "github.com/libnovel/backend/internal/scraper" "github.com/libnovel/backend/internal/taskqueue" + "github.com/prometheus/client_golang/prometheus" ) // Config tunes the runner behaviour. @@ -41,6 +42,7 @@ type Config struct { // WorkerID uniquely identifies this runner instance in PocketBase records. WorkerID string // PollInterval is how often the runner checks for new tasks. + // Only used in PocketBase-polling mode (RedisAddr == ""). PollInterval time.Duration // MaxConcurrentScrape limits simultaneous book-scrape goroutines. MaxConcurrentScrape int @@ -50,9 +52,11 @@ type Config struct { OrchestratorWorkers int // HeartbeatInterval is how often active tasks PATCH their heartbeat_at // timestamp to signal they are still alive. Defaults to 30s when 0. + // Only used in PocketBase-polling mode. HeartbeatInterval time.Duration // StaleTaskThreshold is how old a heartbeat must be (or absent) before the // task is considered orphaned and reset to pending. Defaults to 2m when 0. + // Only used in PocketBase-polling mode. StaleTaskThreshold time.Duration // CatalogueRefreshInterval is how often the runner walks the full catalogue, // scrapes per-book metadata, downloads covers, and re-indexes everything in @@ -66,6 +70,15 @@ type Config struct { // MetricsAddr is the HTTP listen address for the /metrics endpoint. // Defaults to ":9091". Set to "" to disable. MetricsAddr string + // RedisAddr is the address of the Redis instance used for Asynq task + // dispatch. When set the runner switches from PocketBase-polling mode to + // Asynq ServeMux mode (immediate task delivery, no polling). + // Supports plain "host:port" or a full "rediss://..." URL. + // When empty the runner falls back to PocketBase polling. + RedisAddr string + // RedisPassword is the Redis AUTH password. + // Not required when RedisAddr is a full URL that includes credentials. + RedisPassword string } // Dependencies are the external services the runner depends on. @@ -99,6 +112,8 @@ type Runner struct { cfg Config deps Dependencies + metricsRegistry *prometheus.Registry + // Atomic task counters — read by /metrics without locking. tasksRunning atomic.Int64 tasksCompleted atomic.Int64 @@ -139,15 +154,18 @@ func New(cfg Config, deps Dependencies) *Runner { if deps.SearchIndex == nil { deps.SearchIndex = meili.NoopClient{} } - return &Runner{cfg: cfg, deps: deps, startedAt: time.Now()} + return &Runner{cfg: cfg, deps: deps, startedAt: time.Now(), metricsRegistry: prometheus.NewRegistry()} } -// Run starts the poll loop and the metrics HTTP server, blocking until ctx is -// cancelled. +// Run starts the worker loop and the metrics HTTP server, blocking until ctx +// is cancelled. +// +// When cfg.RedisAddr is set the runner uses Asynq (immediate task delivery). +// Otherwise it falls back to PocketBase polling (legacy mode). func (r *Runner) Run(ctx context.Context) error { r.deps.Log.Info("runner: starting", "worker_id", r.cfg.WorkerID, - "poll_interval", r.cfg.PollInterval, + "mode", r.mode(), "max_scrape", r.cfg.MaxConcurrentScrape, "max_audio", r.cfg.MaxConcurrentAudio, "catalogue_refresh_interval", r.cfg.CatalogueRefreshInterval, @@ -164,6 +182,23 @@ func (r *Runner) Run(ctx context.Context) error { }() } + if r.cfg.RedisAddr != "" { + return r.runAsynq(ctx) + } + return r.runPoll(ctx) +} + +// mode returns a short string describing the active dispatch mode. +func (r *Runner) mode() string { + if r.cfg.RedisAddr != "" { + return "asynq" + } + return "poll" +} + +// runPoll is the legacy PocketBase-polling dispatch loop. +// Used when cfg.RedisAddr is empty. +func (r *Runner) runPoll(ctx context.Context) error { scrapeSem := make(chan struct{}, r.cfg.MaxConcurrentScrape) audioSem := make(chan struct{}, r.cfg.MaxConcurrentAudio) var wg sync.WaitGroup @@ -181,6 +216,8 @@ func (r *Runner) Run(ctx context.Context) error { r.deps.Log.Info("runner: skipping initial catalogue refresh (RUNNER_SKIP_INITIAL_CATALOGUE_REFRESH=true)") } + r.deps.Log.Info("runner: poll mode active", "poll_interval", r.cfg.PollInterval) + // Run one poll immediately on startup, then on each tick. for { r.poll(ctx, scrapeSem, audioSem, &wg) diff --git a/caddy/Dockerfile b/caddy/Dockerfile index 8681b3c..d32d510 100644 --- a/caddy/Dockerfile +++ b/caddy/Dockerfile @@ -2,7 +2,8 @@ FROM caddy:2-builder AS builder RUN xcaddy build \ --with github.com/mholt/caddy-ratelimit \ - --with github.com/hslatman/caddy-crowdsec-bouncer/http + --with github.com/hslatman/caddy-crowdsec-bouncer/http \ + --with github.com/mholt/caddy-l4 FROM caddy:2-alpine COPY --from=builder /usr/bin/caddy /usr/bin/caddy diff --git a/docker-compose.yml b/docker-compose.yml index 57b55ee..3a1b496 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -163,6 +163,11 @@ services: GLITCHTIP_DSN: "${GLITCHTIP_DSN}" OTEL_EXPORTER_OTLP_ENDPOINT: "${OTEL_EXPORTER_OTLP_ENDPOINT}" OTEL_SERVICE_NAME: "backend" + # Asynq task queue — backend enqueues jobs to homelab Redis via Caddy TLS proxy. + # Set to "rediss://:password@redis.libnovel.cc:6380" in Doppler prd config. + # Leave empty to fall back to PocketBase polling. + REDIS_ADDR: "${REDIS_ADDR}" + REDIS_PASSWORD: "${REDIS_PASSWORD}" healthcheck: test: ["CMD", "/healthcheck", "http://localhost:8080/health"] interval: 15s @@ -353,8 +358,9 @@ services: # ─── Caddy (reverse proxy + automatic HTTPS) ────────────────────────────────── - # Custom build includes github.com/mholt/caddy-ratelimit and - # github.com/hslatman/caddy-crowdsec-bouncer/http. + # Custom build includes github.com/mholt/caddy-ratelimit, + # github.com/hslatman/caddy-crowdsec-bouncer/http, and + # github.com/mholt/caddy-l4 (TCP layer4 proxy for Redis). caddy: image: kalekber/libnovel-caddy:${GIT_TAG:-latest} build: @@ -372,9 +378,12 @@ services: - "80:80" - "443:443" - "443:443/udp" # HTTP/3 (QUIC) + - "6380:6380" # Redis TCP proxy (TLS) for homelab → Asynq environment: DOMAIN: "${DOMAIN}" CADDY_ACME_EMAIL: "${CADDY_ACME_EMAIL}" + # Homelab Redis address — Caddy TCP-proxies inbound :6380 to this. + HOMELAB_REDIS_ADDR: "${HOMELAB_REDIS_ADDR:?HOMELAB_REDIS_ADDR required for Redis TCP proxy}" env_file: - path: ./crowdsec/.crowdsec.env required: false diff --git a/homelab/otel/grafana/provisioning/alerting/contact-points.yaml b/homelab/otel/grafana/provisioning/alerting/contact-points.yaml new file mode 100644 index 0000000..6852319 --- /dev/null +++ b/homelab/otel/grafana/provisioning/alerting/contact-points.yaml @@ -0,0 +1,16 @@ +# Grafana alerting provisioning — contact points +# Sends all alerts to Gotify (self-hosted push notifications). +apiVersion: 1 + +contactPoints: + - orgId: 1 + name: Gotify + receivers: + - uid: gotify-webhook + type: webhook + settings: + url: "http://gotify/message?token=ABZrZgCY-4ivcmt" + httpMethod: POST + title: "{{ .CommonLabels.alertname }}" + message: "{{ range .Alerts }}{{ .Annotations.summary }}\n{{ .Annotations.description }}{{ end }}" + disableResolveMessage: false diff --git a/homelab/otel/grafana/provisioning/alerting/notification-policy.yaml b/homelab/otel/grafana/provisioning/alerting/notification-policy.yaml new file mode 100644 index 0000000..1876a2b --- /dev/null +++ b/homelab/otel/grafana/provisioning/alerting/notification-policy.yaml @@ -0,0 +1,15 @@ +# Grafana alerting provisioning — notification policies +# Routes all alerts to Gotify by default. +apiVersion: 1 + +policies: + - orgId: 1 + receiver: Gotify + group_by: ["alertname", "service"] + group_wait: 30s + group_interval: 5m + repeat_interval: 4h + routes: + - receiver: Gotify + matchers: + - severity =~ "critical|warning" diff --git a/homelab/otel/grafana/provisioning/alerting/rules.yaml b/homelab/otel/grafana/provisioning/alerting/rules.yaml new file mode 100644 index 0000000..da97ac1 --- /dev/null +++ b/homelab/otel/grafana/provisioning/alerting/rules.yaml @@ -0,0 +1,214 @@ +# Grafana alerting provisioning — alert rules +# Covers: runner down, high task failure rate, audio error spike, backend error spike. +apiVersion: 1 + +groups: + - orgId: 1 + name: LibNovel Runner + folder: LibNovel + interval: 1m + rules: + + - uid: runner-down + title: Runner Down + condition: C + for: 2m + annotations: + summary: "LibNovel runner is not reachable" + description: "The Prometheus scrape of runner:9091 has been failing for >2 minutes. Tasks are not being processed." + labels: + severity: critical + service: runner + data: + - refId: A + datasourceUid: prometheus + relativeTimeRange: { from: 300, to: 0 } + model: + expr: "up{job=\"libnovel-runner\"}" + instant: true + intervalMs: 1000 + maxDataPoints: 43200 + - refId: C + datasourceUid: __expr__ + relativeTimeRange: { from: 300, to: 0 } + model: + type: classic_conditions + conditions: + - evaluator: { params: [1], type: lt } + operator: { type: and } + query: { params: [A] } + reducer: { params: [], type: last } + + - uid: runner-high-failure-rate + title: Runner High Task Failure Rate + condition: C + for: 5m + annotations: + summary: "Runner task failure rate is above 20%" + description: "More than 20% of runner tasks have been failing for the last 5 minutes. Check runner logs." + labels: + severity: warning + service: runner + data: + - refId: A + datasourceUid: prometheus + relativeTimeRange: { from: 600, to: 0 } + model: + expr: "rate(libnovel_runner_tasks_failed_total[5m]) / clamp_min(rate(libnovel_runner_tasks_completed_total[5m]) + rate(libnovel_runner_tasks_failed_total[5m]), 0.001)" + instant: true + intervalMs: 1000 + maxDataPoints: 43200 + - refId: C + datasourceUid: __expr__ + relativeTimeRange: { from: 600, to: 0 } + model: + type: classic_conditions + conditions: + - evaluator: { params: [0.2], type: gt } + operator: { type: and } + query: { params: [A] } + reducer: { params: [], type: last } + + - uid: runner-tasks-stalled + title: Runner Tasks Stalled + condition: C + for: 10m + annotations: + summary: "Runner has tasks running for >10 minutes with no completions" + description: "tasks_running > 0 but rate(tasks_completed) is 0. Tasks may be stuck or the runner is in a crash loop." + labels: + severity: warning + service: runner + data: + - refId: Running + datasourceUid: prometheus + relativeTimeRange: { from: 900, to: 0 } + model: + expr: "libnovel_runner_tasks_running" + instant: true + intervalMs: 1000 + maxDataPoints: 43200 + - refId: Rate + datasourceUid: prometheus + relativeTimeRange: { from: 900, to: 0 } + model: + expr: "rate(libnovel_runner_tasks_completed_total[10m])" + instant: true + intervalMs: 1000 + maxDataPoints: 43200 + - refId: C + datasourceUid: __expr__ + relativeTimeRange: { from: 900, to: 0 } + model: + type: classic_conditions + conditions: + - evaluator: { params: [0], type: gt } + operator: { type: and } + query: { params: [Running] } + reducer: { params: [], type: last } + - evaluator: { params: [0.001], type: lt } + operator: { type: and } + query: { params: [Rate] } + reducer: { params: [], type: last } + + - orgId: 1 + name: LibNovel Backend + folder: LibNovel + interval: 1m + rules: + + - uid: backend-high-error-rate + title: Backend High Error Rate + condition: C + for: 5m + annotations: + summary: "Backend API error rate above 5%" + description: "More than 5% of backend HTTP requests are returning 5xx status codes (as seen from UI OTel instrumentation)." + labels: + severity: warning + service: backend + data: + - refId: A + datasourceUid: prometheus + relativeTimeRange: { from: 600, to: 0 } + model: + expr: "sum(rate(http_client_request_duration_seconds_count{job=\"ui\", server_address=\"backend\", http_response_status_code=~\"5..\"}[5m])) / clamp_min(sum(rate(http_client_request_duration_seconds_count{job=\"ui\", server_address=\"backend\"}[5m])), 0.001)" + instant: true + intervalMs: 1000 + maxDataPoints: 43200 + - refId: C + datasourceUid: __expr__ + relativeTimeRange: { from: 600, to: 0 } + model: + type: classic_conditions + conditions: + - evaluator: { params: [0.05], type: gt } + operator: { type: and } + query: { params: [A] } + reducer: { params: [], type: last } + + - uid: backend-high-p95-latency + title: Backend High p95 Latency + condition: C + for: 5m + annotations: + summary: "Backend p95 latency above 2s" + description: "95th percentile latency of backend spans has exceeded 2 seconds for >5 minutes." + labels: + severity: warning + service: backend + data: + - refId: A + datasourceUid: prometheus + relativeTimeRange: { from: 600, to: 0 } + model: + expr: "histogram_quantile(0.95, sum(rate(traces_spanmetrics_latency_bucket{service=\"backend\"}[5m])) by (le))" + instant: true + intervalMs: 1000 + maxDataPoints: 43200 + - refId: C + datasourceUid: __expr__ + relativeTimeRange: { from: 600, to: 0 } + model: + type: classic_conditions + conditions: + - evaluator: { params: [2], type: gt } + operator: { type: and } + query: { params: [A] } + reducer: { params: [], type: last } + + - orgId: 1 + name: LibNovel OTel Pipeline + folder: LibNovel + interval: 2m + rules: + + - uid: otel-collector-down + title: OTel Collector Down + condition: C + for: 3m + annotations: + summary: "OTel collector is not reachable" + description: "Prometheus cannot scrape otel-collector:8888. Traces and logs may be dropping." + labels: + severity: warning + service: otel-collector + data: + - refId: A + datasourceUid: prometheus + relativeTimeRange: { from: 600, to: 0 } + model: + expr: "up{job=\"otel-collector\"}" + instant: true + intervalMs: 1000 + maxDataPoints: 43200 + - refId: C + datasourceUid: __expr__ + relativeTimeRange: { from: 600, to: 0 } + model: + type: classic_conditions + conditions: + - evaluator: { params: [1], type: lt } + operator: { type: and } + query: { params: [A] } + reducer: { params: [], type: last } diff --git a/homelab/otel/grafana/provisioning/dashboards/backend.json b/homelab/otel/grafana/provisioning/dashboards/backend.json new file mode 100644 index 0000000..50c56b6 --- /dev/null +++ b/homelab/otel/grafana/provisioning/dashboards/backend.json @@ -0,0 +1,338 @@ +{ + "uid": "libnovel-backend", + "title": "Backend API", + "description": "Request rate, error rate, and latency for the LibNovel backend. Powered by Tempo span metrics and UI OTel instrumentation.", + "tags": ["libnovel", "backend", "api"], + "timezone": "browser", + "refresh": "30s", + "time": { "from": "now-3h", "to": "now" }, + "schemaVersion": 39, + "panels": [ + { + "id": 1, + "type": "stat", + "title": "Request Rate (RPS)", + "gridPos": { "x": 0, "y": 0, "w": 4, "h": 4 }, + "options": { + "reduceOptions": { "calcs": ["lastNotNull"] }, + "colorMode": "value", + "graphMode": "area", + "textMode": "auto" + }, + "fieldConfig": { + "defaults": { + "unit": "reqps", + "color": { "mode": "thresholds" }, + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }] } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(traces_spanmetrics_calls_total{service=\"backend\"}[5m]))", + "legendFormat": "rps", + "instant": true + } + ] + }, + { + "id": 2, + "type": "stat", + "title": "Error Rate", + "gridPos": { "x": 4, "y": 0, "w": 4, "h": 4 }, + "options": { + "reduceOptions": { "calcs": ["lastNotNull"] }, + "colorMode": "background", + "graphMode": "none" + }, + "fieldConfig": { + "defaults": { + "unit": "percentunit", + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 0.01 }, + { "color": "red", "value": 0.05 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(traces_spanmetrics_calls_total{service=\"backend\", status_code=\"STATUS_CODE_ERROR\"}[5m])) / clamp_min(sum(rate(traces_spanmetrics_calls_total{service=\"backend\"}[5m])), 0.001)", + "legendFormat": "error rate", + "instant": true + } + ] + }, + { + "id": 3, + "type": "stat", + "title": "p50 Latency", + "gridPos": { "x": 8, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "colorMode": "value", "graphMode": "area" }, + "fieldConfig": { + "defaults": { + "unit": "s", + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 0.2 }, + { "color": "red", "value": 1 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.50, sum(rate(traces_spanmetrics_latency_bucket{service=\"backend\"}[5m])) by (le))", + "legendFormat": "p50", + "instant": true + } + ] + }, + { + "id": 4, + "type": "stat", + "title": "p95 Latency", + "gridPos": { "x": 12, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "colorMode": "value", "graphMode": "area" }, + "fieldConfig": { + "defaults": { + "unit": "s", + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 0.5 }, + { "color": "red", "value": 2 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.95, sum(rate(traces_spanmetrics_latency_bucket{service=\"backend\"}[5m])) by (le))", + "legendFormat": "p95", + "instant": true + } + ] + }, + { + "id": 5, + "type": "stat", + "title": "p99 Latency", + "gridPos": { "x": 16, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "colorMode": "value", "graphMode": "area" }, + "fieldConfig": { + "defaults": { + "unit": "s", + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 1 }, + { "color": "red", "value": 5 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.99, sum(rate(traces_spanmetrics_latency_bucket{service=\"backend\"}[5m])) by (le))", + "legendFormat": "p99", + "instant": true + } + ] + }, + { + "id": 6, + "type": "stat", + "title": "5xx Errors / min", + "gridPos": { "x": 20, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "colorMode": "background", "graphMode": "none" }, + "fieldConfig": { + "defaults": { + "unit": "short", + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 1 }, + { "color": "red", "value": 5 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(http_client_request_duration_seconds_count{job=\"ui\", server_address=\"backend\", http_response_status_code=~\"5..\"}[5m])) * 60", + "legendFormat": "5xx/min", + "instant": true + } + ] + }, + { + "id": 10, + "type": "timeseries", + "title": "Request Rate by Status", + "gridPos": { "x": 0, "y": 4, "w": 12, "h": 8 }, + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { "unit": "reqps", "custom": { "lineWidth": 2, "fillOpacity": 10 } }, + "overrides": [ + { "matcher": { "id": "byFrameRefID", "options": "errors" }, "properties": [{ "id": "color", "value": { "fixedColor": "red", "mode": "fixed" } }] } + ] + }, + "targets": [ + { + "refId": "success", + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(http_client_request_duration_seconds_count{job=\"ui\", server_address=\"backend\", http_response_status_code=~\"2..\"}[5m]))", + "legendFormat": "2xx" + }, + { + "refId": "notfound", + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(http_client_request_duration_seconds_count{job=\"ui\", server_address=\"backend\", http_response_status_code=~\"4..\"}[5m]))", + "legendFormat": "4xx" + }, + { + "refId": "errors", + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(http_client_request_duration_seconds_count{job=\"ui\", server_address=\"backend\", http_response_status_code=~\"5..\"}[5m]))", + "legendFormat": "5xx" + } + ] + }, + { + "id": 11, + "type": "timeseries", + "title": "Latency Percentiles (backend spans)", + "gridPos": { "x": 12, "y": 4, "w": 12, "h": 8 }, + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { "unit": "s", "custom": { "lineWidth": 2, "fillOpacity": 10 } } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.50, sum(rate(traces_spanmetrics_latency_bucket{service=\"backend\"}[5m])) by (le))", + "legendFormat": "p50" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.95, sum(rate(traces_spanmetrics_latency_bucket{service=\"backend\"}[5m])) by (le))", + "legendFormat": "p95" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.99, sum(rate(traces_spanmetrics_latency_bucket{service=\"backend\"}[5m])) by (le))", + "legendFormat": "p99" + } + ] + }, + { + "id": 12, + "type": "timeseries", + "title": "Requests / min by HTTP method (UI → Backend)", + "gridPos": { "x": 0, "y": 12, "w": 12, "h": 8 }, + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { "unit": "short", "custom": { "lineWidth": 2, "fillOpacity": 5 } } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(http_client_request_duration_seconds_count{job=\"ui\", server_address=\"backend\"}[5m])) by (http_request_method) * 60", + "legendFormat": "{{http_request_method}}" + } + ] + }, + { + "id": 13, + "type": "timeseries", + "title": "Requests / min — UI → PocketBase", + "gridPos": { "x": 12, "y": 12, "w": 12, "h": 8 }, + "description": "Traffic from SvelteKit server to PocketBase (auth, collections, etc.).", + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { "unit": "short", "custom": { "lineWidth": 2, "fillOpacity": 5 } } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(http_client_request_duration_seconds_count{job=\"ui\", server_address=\"pocketbase\"}[5m])) by (http_request_method, http_response_status_code) * 60", + "legendFormat": "{{http_request_method}} {{http_response_status_code}}" + } + ] + }, + { + "id": 14, + "type": "timeseries", + "title": "UI → Backend Latency (p50 / p95)", + "gridPos": { "x": 0, "y": 20, "w": 12, "h": 8 }, + "description": "HTTP client latency as seen from the SvelteKit SSR layer calling backend.", + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { "unit": "s", "custom": { "lineWidth": 2, "fillOpacity": 5 } } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.50, sum(rate(http_client_request_duration_seconds_bucket{job=\"ui\", server_address=\"backend\"}[5m])) by (le))", + "legendFormat": "p50" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.95, sum(rate(http_client_request_duration_seconds_bucket{job=\"ui\", server_address=\"backend\"}[5m])) by (le))", + "legendFormat": "p95" + } + ] + }, + { + "id": 20, + "type": "logs", + "title": "Backend Errors", + "gridPos": { "x": 0, "y": 28, "w": 24, "h": 10 }, + "options": { + "showTime": true, + "showLabels": false, + "wrapLogMessage": true, + "prettifyLogMessage": true, + "enableLogDetails": true, + "sortOrder": "Descending", + "dedupStrategy": "none" + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "{service_name=\"backend\"} | json | level =~ `(WARN|ERROR|error|warn)`", + "legendFormat": "" + } + ] + } + ] +} diff --git a/homelab/otel/grafana/provisioning/dashboards/catalogue.json b/homelab/otel/grafana/provisioning/dashboards/catalogue.json new file mode 100644 index 0000000..d17d972 --- /dev/null +++ b/homelab/otel/grafana/provisioning/dashboards/catalogue.json @@ -0,0 +1,275 @@ +{ + "uid": "libnovel-catalogue", + "title": "Catalogue & Content Progress", + "description": "Scraping progress, audio generation coverage, and catalogue health derived from runner structured logs.", + "tags": ["libnovel", "catalogue", "content"], + "timezone": "browser", + "refresh": "1m", + "time": { "from": "now-24h", "to": "now" }, + "schemaVersion": 39, + "panels": [ + { + "id": 1, + "type": "stat", + "title": "Books Scraped (last 24h)", + "description": "Count of unique book slugs appearing in successful scrape task completions.", + "gridPos": { "x": 0, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["sum"] }, "colorMode": "value", "graphMode": "none" }, + "fieldConfig": { + "defaults": { + "color": { "fixedColor": "blue", "mode": "fixed" }, + "thresholds": { "mode": "absolute", "steps": [] } + } + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "sum_over_time({service_name=\"runner\"} | json | msg=`scrape task done` [24h])", + "legendFormat": "books scraped" + } + ] + }, + { + "id": 2, + "type": "stat", + "title": "Chapters Scraped (last 24h)", + "gridPos": { "x": 4, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["sum"] }, "colorMode": "value", "graphMode": "none" }, + "fieldConfig": { + "defaults": { + "color": { "fixedColor": "blue", "mode": "fixed" }, + "thresholds": { "mode": "absolute", "steps": [] } + } + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "sum_over_time({service_name=\"runner\"} | json | unwrap scraped [24h])", + "legendFormat": "chapters scraped" + } + ] + }, + { + "id": 3, + "type": "stat", + "title": "Audio Jobs Completed (last 24h)", + "gridPos": { "x": 8, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["sum"] }, "colorMode": "value", "graphMode": "none" }, + "fieldConfig": { + "defaults": { + "color": { "fixedColor": "green", "mode": "fixed" }, + "thresholds": { "mode": "absolute", "steps": [] } + } + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "sum_over_time({service_name=\"runner\"} | json | msg=`audio task done` [24h])", + "legendFormat": "audio done" + } + ] + }, + { + "id": 4, + "type": "stat", + "title": "Audio Jobs Failed (last 24h)", + "gridPos": { "x": 12, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["sum"] }, "colorMode": "background", "graphMode": "none" }, + "fieldConfig": { + "defaults": { + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 1 }, + { "color": "red", "value": 5 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "sum_over_time({service_name=\"runner\"} | json | msg=`audio task failed` [24h])", + "legendFormat": "audio failed" + } + ] + }, + { + "id": 5, + "type": "stat", + "title": "Scrape Errors (last 24h)", + "gridPos": { "x": 16, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["sum"] }, "colorMode": "background", "graphMode": "none" }, + "fieldConfig": { + "defaults": { + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 1 }, + { "color": "red", "value": 10 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "sum_over_time({service_name=\"runner\"} | json | msg=`scrape task failed` [24h])", + "legendFormat": "scrape errors" + } + ] + }, + { + "id": 6, + "type": "stat", + "title": "Catalogue Refresh — Books Indexed", + "description": "Total books indexed in the last catalogue refresh cycle (from the ok field in the summary log).", + "gridPos": { "x": 20, "y": 0, "w": 4, "h": 4 }, + "options": { "reduceOptions": { "calcs": ["lastNotNull"] }, "colorMode": "value", "graphMode": "none" }, + "fieldConfig": { + "defaults": { + "color": { "fixedColor": "purple", "mode": "fixed" }, + "thresholds": { "mode": "absolute", "steps": [] } + } + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "last_over_time({service_name=\"runner\"} | json | op=`catalogue_refresh` | msg=`catalogue refresh done` | unwrap ok [7d])", + "legendFormat": "indexed" + } + ] + }, + { + "id": 10, + "type": "timeseries", + "title": "Audio Generation Rate (tasks/min)", + "gridPos": { "x": 0, "y": 4, "w": 12, "h": 8 }, + "description": "Rate of audio task completions and failures over time.", + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { "unit": "short", "custom": { "lineWidth": 2, "fillOpacity": 10 } }, + "overrides": [ + { "matcher": { "id": "byName", "options": "failed" }, "properties": [{ "id": "color", "value": { "fixedColor": "red", "mode": "fixed" } }] }, + { "matcher": { "id": "byName", "options": "completed" }, "properties": [{ "id": "color", "value": { "fixedColor": "green", "mode": "fixed" } }] } + ] + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(traces_spanmetrics_calls_total{service=\"runner\", span_name=\"runner.audio_task\", status_code!=\"STATUS_CODE_ERROR\"}[5m])) * 60", + "legendFormat": "completed" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(traces_spanmetrics_calls_total{service=\"runner\", span_name=\"runner.audio_task\", status_code=\"STATUS_CODE_ERROR\"}[5m])) * 60", + "legendFormat": "failed" + } + ] + }, + { + "id": 11, + "type": "timeseries", + "title": "Scraping Rate (tasks/min)", + "gridPos": { "x": 12, "y": 4, "w": 12, "h": 8 }, + "description": "Rate of scrape task completions and failures over time.", + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { "unit": "short", "custom": { "lineWidth": 2, "fillOpacity": 10 } }, + "overrides": [ + { "matcher": { "id": "byName", "options": "failed" }, "properties": [{ "id": "color", "value": { "fixedColor": "red", "mode": "fixed" } }] }, + { "matcher": { "id": "byName", "options": "completed" }, "properties": [{ "id": "color", "value": { "fixedColor": "blue", "mode": "fixed" } }] } + ] + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(traces_spanmetrics_calls_total{service=\"runner\", span_name=\"runner.scrape_task\", status_code!=\"STATUS_CODE_ERROR\"}[5m])) * 60", + "legendFormat": "completed" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(traces_spanmetrics_calls_total{service=\"runner\", span_name=\"runner.scrape_task\", status_code=\"STATUS_CODE_ERROR\"}[5m])) * 60", + "legendFormat": "failed" + } + ] + }, + { + "id": 20, + "type": "logs", + "title": "Scrape Task Events", + "description": "One log line per completed or failed scrape task. Fields: task_id, kind, url, scraped, skipped, errors.", + "gridPos": { "x": 0, "y": 12, "w": 24, "h": 10 }, + "options": { + "showTime": true, + "showLabels": false, + "wrapLogMessage": false, + "prettifyLogMessage": true, + "enableLogDetails": true, + "sortOrder": "Descending", + "dedupStrategy": "none" + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "{service_name=\"runner\"} | json | msg =~ `scrape task (done|failed|starting)`", + "legendFormat": "" + } + ] + }, + { + "id": 21, + "type": "logs", + "title": "Audio Task Events", + "description": "One log line per completed or failed audio task. Fields: task_id, slug, chapter, voice, key (on success), reason (on failure).", + "gridPos": { "x": 0, "y": 22, "w": 24, "h": 10 }, + "options": { + "showTime": true, + "showLabels": false, + "wrapLogMessage": false, + "prettifyLogMessage": true, + "enableLogDetails": true, + "sortOrder": "Descending", + "dedupStrategy": "none" + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "{service_name=\"runner\"} | json | msg =~ `audio task (done|failed|starting)`", + "legendFormat": "" + } + ] + }, + { + "id": 22, + "type": "logs", + "title": "Catalogue Refresh Progress", + "description": "Progress logs from the background catalogue refresh (every 24h). Fields: op=catalogue_refresh, scraped, ok, skipped, errors.", + "gridPos": { "x": 0, "y": 32, "w": 24, "h": 8 }, + "options": { + "showTime": true, + "showLabels": false, + "wrapLogMessage": false, + "prettifyLogMessage": true, + "enableLogDetails": true, + "sortOrder": "Descending", + "dedupStrategy": "none" + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "{service_name=\"runner\"} | json | op=`catalogue_refresh`", + "legendFormat": "" + } + ] + } + ] +} diff --git a/homelab/otel/grafana/provisioning/dashboards/runner.json b/homelab/otel/grafana/provisioning/dashboards/runner.json new file mode 100644 index 0000000..79d94c0 --- /dev/null +++ b/homelab/otel/grafana/provisioning/dashboards/runner.json @@ -0,0 +1,377 @@ +{ + "uid": "libnovel-runner", + "title": "Runner Operations", + "description": "Task queue health, throughput, TTS routing, and live logs for the homelab runner.", + "tags": ["libnovel", "runner"], + "timezone": "browser", + "refresh": "30s", + "time": { "from": "now-3h", "to": "now" }, + "schemaVersion": 39, + "panels": [ + { + "id": 1, + "type": "stat", + "title": "Tasks Running", + "gridPos": { "x": 0, "y": 0, "w": 4, "h": 4 }, + "options": { + "reduceOptions": { "calcs": ["lastNotNull"] }, + "colorMode": "background", + "graphMode": "none", + "textMode": "auto" + }, + "fieldConfig": { + "defaults": { + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 1 }, + { "color": "red", "value": 3 } + ] + }, + "mappings": [] + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "libnovel_runner_tasks_running", + "legendFormat": "running", + "instant": true + } + ] + }, + { + "id": 2, + "type": "stat", + "title": "Tasks Completed (total)", + "gridPos": { "x": 4, "y": 0, "w": 4, "h": 4 }, + "options": { + "reduceOptions": { "calcs": ["lastNotNull"] }, + "colorMode": "background", + "graphMode": "area", + "textMode": "auto" + }, + "fieldConfig": { + "defaults": { + "color": { "fixedColor": "green", "mode": "fixed" }, + "thresholds": { "mode": "absolute", "steps": [] } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "libnovel_runner_tasks_completed_total", + "legendFormat": "completed", + "instant": true + } + ] + }, + { + "id": 3, + "type": "stat", + "title": "Tasks Failed (total)", + "gridPos": { "x": 8, "y": 0, "w": 4, "h": 4 }, + "options": { + "reduceOptions": { "calcs": ["lastNotNull"] }, + "colorMode": "background", + "graphMode": "none", + "textMode": "auto" + }, + "fieldConfig": { + "defaults": { + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 1 }, + { "color": "red", "value": 5 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "libnovel_runner_tasks_failed_total", + "legendFormat": "failed", + "instant": true + } + ] + }, + { + "id": 4, + "type": "stat", + "title": "Runner Uptime", + "gridPos": { "x": 12, "y": 0, "w": 4, "h": 4 }, + "options": { + "reduceOptions": { "calcs": ["lastNotNull"] }, + "colorMode": "value", + "graphMode": "none", + "textMode": "auto" + }, + "fieldConfig": { + "defaults": { + "unit": "s", + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "red", "value": null }, + { "color": "yellow", "value": 60 }, + { "color": "green", "value": 300 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "libnovel_runner_uptime_seconds", + "legendFormat": "uptime", + "instant": true + } + ] + }, + { + "id": 5, + "type": "stat", + "title": "Task Failure Rate", + "gridPos": { "x": 16, "y": 0, "w": 4, "h": 4 }, + "options": { + "reduceOptions": { "calcs": ["lastNotNull"] }, + "colorMode": "background", + "graphMode": "none", + "textMode": "auto" + }, + "fieldConfig": { + "defaults": { + "unit": "percentunit", + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 0.05 }, + { "color": "red", "value": 0.2 } + ] + } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "libnovel_runner_tasks_failed_total / clamp_min(libnovel_runner_tasks_completed_total + libnovel_runner_tasks_failed_total, 1)", + "legendFormat": "failure rate", + "instant": true + } + ] + }, + { + "id": 6, + "type": "stat", + "title": "Runner Alive", + "gridPos": { "x": 20, "y": 0, "w": 4, "h": 4 }, + "options": { + "reduceOptions": { "calcs": ["lastNotNull"] }, + "colorMode": "background", + "graphMode": "none", + "textMode": "auto" + }, + "fieldConfig": { + "defaults": { + "mappings": [ + { "type": "value", "options": { "1": { "text": "UP", "color": "green" }, "0": { "text": "DOWN", "color": "red" } } } + ], + "thresholds": { "mode": "absolute", "steps": [] } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "up{job=\"libnovel-runner\"}", + "legendFormat": "runner", + "instant": true + } + ] + }, + { + "id": 10, + "type": "timeseries", + "title": "Task Throughput (per minute)", + "gridPos": { "x": 0, "y": 4, "w": 12, "h": 8 }, + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { + "unit": "ops", + "custom": { "lineWidth": 2, "fillOpacity": 10 } + }, + "overrides": [ + { "matcher": { "id": "byName", "options": "failed" }, "properties": [{ "id": "color", "value": { "fixedColor": "red", "mode": "fixed" } }] }, + { "matcher": { "id": "byName", "options": "completed" }, "properties": [{ "id": "color", "value": { "fixedColor": "green", "mode": "fixed" } }] } + ] + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "rate(libnovel_runner_tasks_completed_total[5m]) * 60", + "legendFormat": "completed" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "rate(libnovel_runner_tasks_failed_total[5m]) * 60", + "legendFormat": "failed" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "libnovel_runner_tasks_running", + "legendFormat": "running" + } + ] + }, + { + "id": 11, + "type": "timeseries", + "title": "Audio Task Span Latency (p50 / p95 / p99)", + "gridPos": { "x": 12, "y": 4, "w": 12, "h": 8 }, + "description": "End-to-end latency of runner.audio_task spans from Tempo span metrics.", + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { + "unit": "s", + "custom": { "lineWidth": 2, "fillOpacity": 10 } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.50, sum(rate(traces_spanmetrics_latency_bucket{service=\"runner\", span_name=\"runner.audio_task\"}[5m])) by (le))", + "legendFormat": "p50" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.95, sum(rate(traces_spanmetrics_latency_bucket{service=\"runner\", span_name=\"runner.audio_task\"}[5m])) by (le))", + "legendFormat": "p95" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.99, sum(rate(traces_spanmetrics_latency_bucket{service=\"runner\", span_name=\"runner.audio_task\"}[5m])) by (le))", + "legendFormat": "p99" + } + ] + }, + { + "id": 20, + "type": "timeseries", + "title": "Scrape Task Span Latency (p50 / p95 / p99)", + "gridPos": { "x": 0, "y": 12, "w": 12, "h": 8 }, + "description": "End-to-end latency of runner.scrape_task spans from Tempo span metrics.", + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { + "unit": "s", + "custom": { "lineWidth": 2, "fillOpacity": 10 } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.50, sum(rate(traces_spanmetrics_latency_bucket{service=\"runner\", span_name=\"runner.scrape_task\"}[5m])) by (le))", + "legendFormat": "p50" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.95, sum(rate(traces_spanmetrics_latency_bucket{service=\"runner\", span_name=\"runner.scrape_task\"}[5m])) by (le))", + "legendFormat": "p95" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "histogram_quantile(0.99, sum(rate(traces_spanmetrics_latency_bucket{service=\"runner\", span_name=\"runner.scrape_task\"}[5m])) by (le))", + "legendFormat": "p99" + } + ] + }, + { + "id": 21, + "type": "timeseries", + "title": "Audio vs Scrape Task Rate", + "gridPos": { "x": 12, "y": 12, "w": 12, "h": 8 }, + "description": "Relative throughput of audio generation vs book scraping.", + "options": { + "tooltip": { "mode": "multi" }, + "legend": { "displayMode": "list", "placement": "bottom" } + }, + "fieldConfig": { + "defaults": { + "unit": "ops", + "custom": { "lineWidth": 2, "fillOpacity": 10 } + } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(traces_spanmetrics_calls_total{service=\"runner\", span_name=\"runner.audio_task\"}[5m]))", + "legendFormat": "audio tasks/s" + }, + { + "datasource": { "type": "prometheus", "uid": "prometheus" }, + "expr": "sum(rate(traces_spanmetrics_calls_total{service=\"runner\", span_name=\"runner.scrape_task\"}[5m]))", + "legendFormat": "scrape tasks/s" + } + ] + }, + { + "id": 30, + "type": "logs", + "title": "Runner Logs (errors & warnings)", + "gridPos": { "x": 0, "y": 20, "w": 24, "h": 10 }, + "options": { + "showTime": true, + "showLabels": false, + "showCommonLabels": false, + "wrapLogMessage": true, + "prettifyLogMessage": true, + "enableLogDetails": true, + "sortOrder": "Descending", + "dedupStrategy": "none" + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "{service_name=\"runner\"} | json | level =~ `(WARN|ERROR|error|warn)`", + "legendFormat": "" + } + ] + }, + { + "id": 31, + "type": "logs", + "title": "Runner Logs (all)", + "gridPos": { "x": 0, "y": 30, "w": 24, "h": 10 }, + "options": { + "showTime": true, + "showLabels": false, + "showCommonLabels": false, + "wrapLogMessage": true, + "prettifyLogMessage": true, + "enableLogDetails": true, + "sortOrder": "Descending", + "dedupStrategy": "none" + }, + "targets": [ + { + "datasource": { "type": "loki", "uid": "loki" }, + "expr": "{service_name=\"runner\"} | json", + "legendFormat": "" + } + ] + } + ] +} diff --git a/homelab/runner/docker-compose.yml b/homelab/runner/docker-compose.yml index 936e8eb..2eb2991 100644 --- a/homelab/runner/docker-compose.yml +++ b/homelab/runner/docker-compose.yml @@ -1,7 +1,7 @@ # LibNovel homelab runner # # Connects to production PocketBase and MinIO via public subdomains. -# All secrets come from Doppler (project=libnovel, config=prd). +# All secrets come from Doppler (project=libnovel, config=prd_homelab). # Run with: doppler run -- docker compose up -d # # Differs from prod runner: @@ -11,12 +11,31 @@ # - MEILI_URL → https://search.libnovel.cc (Caddy-proxied) # - VALKEY_ADDR → unset (not exposed publicly) # - RUNNER_SKIP_INITIAL_CATALOGUE_REFRESH=true +# - Redis service for Asynq task queue (local to homelab, exposed to prod via Caddy TCP proxy) services: + redis: + image: redis:7-alpine + restart: unless-stopped + volumes: + - redis_data:/data + command: > + redis-server + --appendonly yes + --requirepass "${REDIS_PASSWORD}" + healthcheck: + test: ["CMD", "redis-cli", "-a", "${REDIS_PASSWORD}", "ping"] + interval: 10s + timeout: 5s + retries: 5 + runner: image: kalekber/libnovel-runner:latest restart: unless-stopped stop_grace_period: 135s + depends_on: + redis: + condition: service_healthy environment: # ── PocketBase ────────────────────────────────────────────────────────── POCKETBASE_URL: "https://pb.libnovel.cc" @@ -42,6 +61,14 @@ services: KOKORO_URL: "${KOKORO_URL}" KOKORO_VOICE: "${KOKORO_VOICE}" + # ── Pocket TTS ────────────────────────────────────────────────────────── + POCKET_TTS_URL: "${POCKET_TTS_URL}" + + # ── Asynq / Redis (local service) ─────────────────────────────────────── + # The runner connects to the local Redis sidecar. + REDIS_ADDR: "redis:6379" + REDIS_PASSWORD: "${REDIS_PASSWORD}" + # ── Runner tuning ─────────────────────────────────────────────────────── RUNNER_WORKER_ID: "${RUNNER_WORKER_ID}" RUNNER_POLL_INTERVAL: "${RUNNER_POLL_INTERVAL}" @@ -60,3 +87,6 @@ services: interval: 60s timeout: 5s retries: 3 + +volumes: + redis_data: