prehistoric-systems/omnivore
1
0
Fork 0
mirror of https://github.com/omnivore-app/omnivore.git synced 2026-03-11 08:54:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Use prometheus/client_golang for metrics endpoint

Browse source 
Replace hand-rolled Prometheus text output in /metrics with the official
prometheus/client_golang library. Adds an internal/metrics package that
registers five GaugeVec collectors (active, failed, completed, prioritized,
oldest_job_age_seconds) and refreshes them from Redis on each request via
a thin wrapper around promhttp.Handler.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Aliaksei Karneyeu 2026-03-04 14:55:19 +01:00
parent 0647484597
commit 25cbcddf01
4 changed files with 115 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
packages/content-fetch-go/go.mod
View file

@ -23,6 +23,7 @@ require (
github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0 // indirect
github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.55.0 // indirect
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.55.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/chromedp/sysutil v1.1.0 // indirect
github.com/cncf/xds/go v0.0.0-20251022180443-0feb69152e9f // indirect
@ -43,7 +44,12 @@ require (
github.com/googleapis/enterprise-certificate-proxy v0.3.11 // indirect
github.com/googleapis/gax-go/v2 v2.17.0 // indirect
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // 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/spiffe/go-spiffe/v2 v2.6.0 // indirect
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
go.opentelemetry.io/contrib/detectors/gcp v1.38.0 // indirect
@ -55,6 +61,7 @@ require (
go.opentelemetry.io/otel/sdk/metric v1.39.0 // indirect
go.opentelemetry.io/otel/trace v1.39.0 // indirect
go.uber.org/atomic v1.11.0 // indirect
go.yaml.in/yaml/v2 v2.4.2 // indirect
golang.org/x/crypto v0.47.0 // indirect
golang.org/x/net v0.49.0 // indirect
golang.org/x/oauth2 v0.35.0 // indirect

15
packages/content-fetch-go/go.sum
View file

@ -28,6 +28,8 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.55.0/go.mod h1:vB2GH9GAYYJTO3mEn8oYwzEdhlayZIdQz6zdzgUIRvA=
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.55.0 h1:0s6TxfCu2KHkkZPnBfsQ2y5qia0jl3MMrmBhu3nCOYk=
github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.55.0/go.mod h1:Mf6O40IAyB9zR/1J8nGDDPirZQQPbYJni8Yisy7NTMc=
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/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
@ -95,6 +97,8 @@ github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBF
github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80 h1:6Yzfa6GP0rIo/kULo2bwGEkFvCePZ3qHDDTC3/J9Swo=
github.com/ledongthuc/pdf v0.0.0-20220302134840-0c2507a12d80/go.mod h1:imJHygn/1yfhB7XSJJKlFZKl/J+dCPAknuiaGOshXAs=
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/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde h1:x0TT0RDC7UhAVbbWWBzr41ElhJx5tXPWkIHA2HWPRuw=
github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0=
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo=
@ -103,6 +107,14 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRI
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/posthog/posthog-go v1.10.0 h1:wfoy7Jfb4LigCoHYyMZoiJmmEoCLOkSaYfDxM/NtCqY=
github.com/posthog/posthog-go v1.10.0/go.mod h1:wB3/9Q7d9gGb1P/yf/Wri9VBlbP8oA8z++prRzL5OcY=
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.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
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.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/spiffe/go-spiffe/v2 v2.6.0 h1:l+DolpxNWYgruGQVV0xsfeya3CsC7m8iBzDnMpsbLuo=
@ -133,6 +145,8 @@ go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6
go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
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=
golang.org/x/crypto v0.47.0 h1:V6e3FRj+n4dbpw86FJ8Fv7XVOql7TEwpHapKoMJ/GO8=
golang.org/x/crypto v0.47.0/go.mod h1:ff3Y9VzzKbwSSEzWqJsJVBnWmRwRSHt/6Op5n9bQc4A=
golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o=
@ -162,5 +176,6 @@ google.golang.org/grpc v1.78.0 h1:K1XZG/yGDJnzMdd/uZHAkVqJE+xIDOcmdSFZkBUicNc=
google.golang.org/grpc v1.78.0/go.mod h1:I47qjTo4OKbMkjA/aOOwxDIiPSBofUtQUI5EfpWvW7U=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

90
packages/content-fetch-go/internal/metrics/metrics.go Normal file
View file

@ -0,0 +1,90 @@
// Package metrics registers and exposes Prometheus gauges for the content-fetch
// queue, matching the metric names produced by the original TypeScript service.
package metrics
import (
"context"
"log"
"net/http"
"github.com/omnivore-app/omnivore/content-fetch-go/internal/bullmq"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/redis/go-redis/v9"
)
const queueLabel = "queue"
var (
activeGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Name: "omnivore_queue_messages_active",
Help: "Number of active jobs in the queue.",
}, []string{queueLabel})
failedGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Name: "omnivore_queue_messages_failed",
Help: "Number of failed jobs in the queue.",
}, []string{queueLabel})
completedGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Name: "omnivore_queue_messages_completed",
Help: "Number of completed jobs in the queue.",
}, []string{queueLabel})
prioritizedGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Name: "omnivore_queue_messages_prioritized",
Help: "Number of prioritized (waiting) jobs in the queue.",
}, []string{queueLabel})
oldestJobAgeGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Name: "omnivore_queue_messages_oldest_job_age_seconds",
Help: "Age in seconds of the oldest prioritized job in the queue.",
}, []string{queueLabel})
)
func init() {
prometheus.MustRegister(
activeGauge,
failedGauge,
completedGauge,
prioritizedGauge,
oldestJobAgeGauge,
)
}
// Handler returns an http.Handler that refreshes queue metrics from Redis on
// every request and then delegates to the standard promhttp handler.
func Handler(rdb *redis.Client, queueName string) http.Handler {
inner := promhttp.Handler()
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if err := refresh(r.Context(), rdb, queueName); err != nil {
log.Printf("Error refreshing queue metrics: %v", err)
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
inner.ServeHTTP(w, r)
})
}
// refresh pulls the current queue counts from Redis and updates the gauges.
func refresh(ctx context.Context, rdb *redis.Client, queueName string) error {
counts, err := bullmq.GetQueueCounts(ctx, rdb, queueName)
if err != nil {
return err
}
labels := prometheus.Labels{queueLabel: queueName}
activeGauge.With(labels).Set(float64(counts["active"]))
failedGauge.With(labels).Set(float64(counts["failed"]))
completedGauge.With(labels).Set(float64(counts["completed"]))
prioritizedGauge.With(labels).Set(float64(counts["prioritized"]))
age, err := bullmq.OldestPrioritizedJobAge(ctx, rdb, queueName)
if err != nil {
log.Printf("Error getting oldest job age: %v", err)
age = 0
}
oldestJobAgeGauge.With(labels).Set(age)
return nil
}

38
packages/content-fetch-go/internal/server/server.go
View file

@ -4,15 +4,15 @@ package server
import (
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"strconv"
"github.com/omnivore-app/omnivore/content-fetch-go/internal/browser"
"github.com/omnivore-app/omnivore/content-fetch-go/internal/bullmq"
"github.com/omnivore-app/omnivore/content-fetch-go/internal/config"
"github.com/omnivore-app/omnivore/content-fetch-go/internal/handler"
"github.com/omnivore-app/omnivore/content-fetch-go/internal/bullmq"
"github.com/omnivore-app/omnivore/content-fetch-go/internal/metrics"
"github.com/omnivore-app/omnivore/content-fetch-go/internal/redisutil"
)
@ -34,7 +34,7 @@ func New(cfg *config.Config, rds *redisutil.RedisDataSource, br *browser.Browser
m := &mux{cfg: cfg, rds: rds, br: br, worker: w}
m.HandleFunc("GET /_ah/health", m.health)
m.HandleFunc("GET /lifecycle/prestop", m.prestop)
m.HandleFunc("GET /metrics", m.metrics)
m.Handle("GET /metrics", metrics.Handler(rds.MQClient, bullmq.ContentFetchQueue))
m.HandleFunc("/", m.root) // GET and POST
return m
}
@ -48,43 +48,11 @@ func (m *mux) health(w http.ResponseWriter, r *http.Request) {
// It signals the worker to stop and waits for in-flight jobs to finish.
func (m *mux) prestop(w http.ResponseWriter, r *http.Request) {
log.Println("Prestop lifecycle hook called.")
// Worker shutdown is handled by the caller (main) via context cancellation;
// here we just wait until it's done.
m.worker.Wait()
log.Println("Worker drained on prestop")
w.WriteHeader(http.StatusOK)
}
// metrics returns Prometheus-format queue metrics, matching the original /metrics endpoint.
func (m *mux) metrics(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
counts, err := bullmq.GetQueueCounts(ctx, m.rds.MQClient, bullmq.ContentFetchQueue)
if err != nil {
log.Printf("Error getting queue counts: %v", err)
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
age, err := bullmq.OldestPrioritizedJobAge(ctx, m.rds.MQClient, bullmq.ContentFetchQueue)
if err != nil {
log.Printf("Error getting oldest job age: %v", err)
}
output := ""
for _, metric := range []string{"active", "failed", "completed", "prioritized"} {
val, _ := counts[metric]
output += fmt.Sprintf("# TYPE omnivore_queue_messages_%s gauge\n", metric)
output += fmt.Sprintf("omnivore_queue_messages_%s{queue=%q} %d\n", metric, bullmq.ContentFetchQueue, val)
}
output += "# TYPE omnivore_queue_messages_oldest_job_age_seconds gauge\n"
output += fmt.Sprintf("omnivore_queue_messages_oldest_job_age_seconds{queue=%q} %s\n",
bullmq.ContentFetchQueue, strconv.FormatFloat(age, 'f', -1, 64))
w.Header().Set("Content-Type", "text/plain")
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(output))
}
// root handles the primary job-processing endpoint (GET or POST /?token=...).
func (m *mux) root(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/" {