diff --git a/packages/content-fetch-go/go.mod b/packages/content-fetch-go/go.mod index 63e90053b..5bd49897a 100644 --- a/packages/content-fetch-go/go.mod +++ b/packages/content-fetch-go/go.mod @@ -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 diff --git a/packages/content-fetch-go/go.sum b/packages/content-fetch-go/go.sum index faf93d16d..ed90d038e 100644 --- a/packages/content-fetch-go/go.sum +++ b/packages/content-fetch-go/go.sum @@ -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= diff --git a/packages/content-fetch-go/internal/metrics/metrics.go b/packages/content-fetch-go/internal/metrics/metrics.go new file mode 100644 index 000000000..04ae461a8 --- /dev/null +++ b/packages/content-fetch-go/internal/metrics/metrics.go @@ -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 +} diff --git a/packages/content-fetch-go/internal/server/server.go b/packages/content-fetch-go/internal/server/server.go index ad7499758..6778a9a06 100644 --- a/packages/content-fetch-go/internal/server/server.go +++ b/packages/content-fetch-go/internal/server/server.go @@ -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 != "/" {