Add Prometheus metrics for jobs and queues

README.md

@@ -102,6 +102,7 @@ Starting the server runs a set of workers listening on a default job queue. It a
 | GET    | /groups/{groupID}      | Get the status of a job group and its jobs      |
 | DELETE | /jobs/{jobID}          | Deletes a pending job from the queue and immediately cancels its execution and frees the thread. Send a query param "purge=true" to delete completed jobs. Only the Go PostgreSQL driver cancels queries mid execution. MySQL server will keep continuing to execute the query. For MySQL, it's important to set `max_execution_time`.   |
 | DELETE | /groups/{groupID}          | Deletes a pending job from the queue and immediately cancels its execution and frees the thread. Send a query param "purge=true" to delete completed jobs. Only the Go PostgreSQL driver cancels queries mid execution. MySQL server will keep continuing to execute the query. For MySQL, it's important to set `max_execution_time`  |
+| GET    | /metrics               | Exposes Prometheus metrics for jobs and queues |
 
 POST requests accept raw JSON bodies. The JSON params are listed below.
 
@@ -173,6 +174,33 @@ $ curl localhost:6060/tasks/get_profit_entries_by_date/jobs -H "Content-Type: ap
 $ curl localhost:6060/tasks/get_profit_entries_by_date/jobs -H "Content-Type: application/json" --data '{"job_id": "myjob", "queue": "low_priority"}'
 ```
 
+## Metrics
+
+Job queue metrics are exposed in the Prometheus exposition format at `/metrics` on the same HTTP interface.
+
+| Metric | Type | Description |
+|--------|------|-------------|
+| `dbtl_jobs_enqueued_total{task_name}` | counter | Jobs enqueued by this instance, including jobs in groups. |
+| `dbtl_jobs_running{task_name}` | gauge | Jobs currently executing on this instance. |
+| `dbtl_job_attempts_total{task_name,outcome}` | counter | Job execution attempts, with `outcome` as `success`, `error` or `canceled`. Every retry is a separate attempt. |
+| `dbtl_job_execution_duration_seconds{task_name}` | histogram | Time taken to run the query, read the rows, and write the results, including failed attempts. |
+| `dbtl_job_queue_wait_duration_seconds{task_name,queue}` | histogram | Time a job waited in the queue before its first attempt. ETA delays are not counted. |
+| `dbtl_queue_pending_jobs{queue}` | gauge | Runnable jobs waiting in the queue. Read from the broker on every scrape. |
+| `dbtl_worker_concurrency{queue}` | gauge | Configured worker concurrency for the queue on this instance. |
+
+```
+$ curl localhost:6060/metrics
+dbtl_jobs_enqueued_total{task_name="get_profit_entries_by_date"} 42
+dbtl_jobs_running{task_name="get_profit_entries_by_date"} 2
+dbtl_job_attempts_total{task_name="get_profit_entries_by_date",outcome="success"} 38
+dbtl_job_attempts_total{task_name="get_profit_entries_by_date",outcome="error"} 2
+dbtl_queue_pending_jobs{queue="default"} 5
+dbtl_worker_concurrency{queue="default"} 10
+...
+```
+
+Metrics are tracked per instance. Instances started with `--worker-only` don't run the HTTP server, so their metrics can't be scraped.
+
 ## Go API client
 `github.com/zerodha/dungbeetle/client` package can be used as a Go HTTP API client for DungBeetle.
 

cmd/http.go

@@ -168,6 +168,12 @@ func (a *Handlers) handleCancelGroupJob(w http.ResponseWriter, r *http.Request)
 	a.sendResponse(w, true)
 }
 
+// handleGetMetrics exposes Prometheus/VictoriaMetrics metrics for scraping.
+func (a *Handlers) handleGetMetrics(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Content-Type", "text/plain")
+	a.co.Metrics().WritePrometheus(w)
+}
+
 // sendResponse sends a JSON envelope to the HTTP response.
 func (a *Handlers) sendResponse(w http.ResponseWriter, data interface{}) {
 	w.Header().Set("Content-Type", "application/json; charset=utf-8")

cmd/init.go

@@ -139,6 +139,7 @@ func initHTTP(h *Handlers, ko *koanf.Koanf) {
 	r.Get("/jobs/queue/{queue}", h.handleGetPendingJobs)
 	r.Post("/groups", h.handlePostJobGroup)
 	r.Get("/groups/{groupID}", h.handleGetGroupStatus)
+	r.Get("/metrics", h.handleGetMetrics)
 
 	log.Info("starting HTTP server", "address", ko.String("server"))
 	if err := http.ListenAndServe(ko.String("server"), r); err != nil {

go.mod

@@ -4,6 +4,7 @@ go 1.25.0
 
 require (
 	github.com/ClickHouse/clickhouse-go v1.5.4
+	github.com/VictoriaMetrics/metrics v1.40.2
 	github.com/go-chi/chi/v5 v5.2.4
 	github.com/go-sql-driver/mysql v1.9.3
 	github.com/gofrs/uuid/v5 v5.3.2
@@ -39,6 +40,8 @@ require (
 	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/robfig/cron/v3 v3.0.1 // indirect
+	github.com/valyala/fastrand v1.1.0 // indirect
+	github.com/valyala/histogram v1.2.0 // indirect
 	github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect

go.sum

@@ -2,6 +2,8 @@ filippo.io/edwards25519 v1.1.1 h1:YpjwWWlNmGIDyXOn8zLzqiD+9TyIlPhGFG96P39uBpw=
 filippo.io/edwards25519 v1.1.1/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
 github.com/ClickHouse/clickhouse-go v1.5.4 h1:cKjXeYLNWVJIx2J1K6H2CqyRmfwVJVY1OV1coaaFcI0=
 github.com/ClickHouse/clickhouse-go v1.5.4/go.mod h1:EaI/sW7Azgz9UATzd5ZdZHRUhHgv5+JMS9NSr2smCJI=
+github.com/VictoriaMetrics/metrics v1.40.2 h1:OVSjKcQEx6JAwGeu8/KQm9Su5qJ72TMEW4xYn5vw3Ac=
+github.com/VictoriaMetrics/metrics v1.40.2/go.mod h1:XE4uudAAIRaJE614Tl5HMrtoEU6+GDZO4QTnNSsZRuA=
 github.com/bkaradzic/go-lz4 v1.0.0 h1:RXc4wYsyz985CkXXeX04y4VnZFGG8Rd43pRaHsOXAKk=
 github.com/bkaradzic/go-lz4 v1.0.0/go.mod h1:0YdlkowM3VswSROI7qDxhRvJ3sLhlFrRRwjwegp5jy4=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
@@ -95,6 +97,10 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G8=
+github.com/valyala/fastrand v1.1.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ=
+github.com/valyala/histogram v1.2.0 h1:wyYGAZZt3CpwUiIb9AU/Zbllg1llXyrtApRS815OLoQ=
+github.com/valyala/histogram v1.2.0/go.mod h1:Hb4kBwb4UxsaNbbbh+RRz8ZR6pdodR57tzWUS3BUzXY=
 github.com/vmihailenco/msgpack v4.0.4+incompatible h1:dSLoQfGFAo3F6OoNhwUmLwVgaUXK79GlxNBwueZn0xI=
 github.com/vmihailenco/msgpack v4.0.4+incompatible/go.mod h1:fy3FlTQTDXWkZ7Bh6AcGMlsjHatGryHQYUTf1ShIgkk=
 github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=

internal/core/core.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
 	uuid "github.com/gofrs/uuid/v5"
 	"github.com/kalbhor/tasqueue/v2"
 	"github.com/vmihailenco/msgpack"
@@ -47,6 +48,9 @@ type Core struct {
 	jobCtx map[string]context.CancelFunc
 	mu     sync.RWMutex
 
+	// Prometheus/VictoriaMetrics metrics.
+	metrics *metrics.Set
+
 	lo *slog.Logger
 }
 
@@ -59,10 +63,16 @@ func New(o Opt, srcDBs dbpool.Pool, res ResultBackends, lo *slog.Logger) *Core {
 		resultBackends: res,
 		jobCtx:         make(map[string]context.CancelFunc),
 		mu:             sync.RWMutex{},
+		metrics:        metrics.NewSet(),
 		lo:             lo,
 	}
 }
 
+// Metrics returns the core's Prometheus/VictoriaMetrics metrics set.
+func (co *Core) Metrics() *metrics.Set {
+	return co.metrics
+}
+
 // Start is a blocking function that spawns the queue workers and starts processing jobs.
 // This should be invoked last after all other initializations are done.
 func (co *Core) Start(ctx context.Context, workerName string, concurrency int) error {
@@ -94,6 +104,7 @@ func (co *Core) NewJob(j models.JobReq, taskName string) (models.JobResp, error)
 	if err != nil {
 		return models.JobResp{}, err
 	}
+	co.metrics.GetOrCreateCounter(metricName(metricJobsEnqueued, "task_name", taskName)).Inc()
 
 	return models.JobResp{
 		JobID:    uuid,
@@ -131,6 +142,8 @@ func (co *Core) NewJobGroup(req models.GroupReq) (models.GroupResp, error) {
 
 	jobs := make([]models.JobResp, len(sigs))
 	for i, r := range sigs {
+		co.metrics.GetOrCreateCounter(metricName(metricJobsEnqueued, "task_name", r.Task)).Inc()
+
 		jobs[i] = models.JobResp{
 			JobID:    r.Opts.ID,
 			TaskName: r.Task,
@@ -377,10 +390,16 @@ func (co *Core) makeJob(j models.JobReq, taskName string) (tasqueue.Job, error)
 		args[i] = j.Args[i]
 	}
 
+	readyAt := time.Now()
+	if eta.After(readyAt) {
+		readyAt = eta
+	}
+
 	b, err := msgpack.Marshal(taskMeta{
-		Args: args,
-		DB:   j.DB,
-		TTL:  int(ttl),
+		Args:    args,
+		DB:      j.DB,
+		TTL:     int(ttl),
+		ReadyAt: readyAt,
 	})
 	if err != nil {
 		return tasqueue.Job{}, err
@@ -398,6 +417,9 @@ type taskMeta struct {
 	Args []interface{} `json:"args"`
 	DB   string        `json:"db"`
 	TTL  int           `json:"ttl"`
+
+	// ReadyAt is when the job became runnable (enqueue time or ETA). Zero for jobs enqueued by older versions.
+	ReadyAt time.Time `json:"ready_at"`
 }
 
 // initQueue creates and returns a distributed queue system (Tasqueue) and registers
@@ -418,17 +440,53 @@ func (co *Core) initQueue() (*tasqueue.Server, error) {
 	// Register every SQL ready tasks in the queue system as a job function.
 	for name := range co.tasks {
 		query := co.tasks[name]
-		err := qs.RegisterTask(string(name), func(b []byte, jctx tasqueue.JobCtx) error {
-			if _, err := qs.GetJob(context.Background(), jctx.Meta.ID); err != nil {
+		co.registerQueueMetrics(query.Queue, query.Conc)
+
+		var (
+			running   = co.metrics.GetOrCreateGauge(metricName(metricJobsRunning, "task_name", name), nil)
+			succeeded = co.metrics.GetOrCreateCounter(metricName(metricJobAttempts, "task_name", name, "outcome", "success"))
+			errored   = co.metrics.GetOrCreateCounter(metricName(metricJobAttempts, "task_name", name, "outcome", "error"))
+			canceled  = co.metrics.GetOrCreateCounter(metricName(metricJobAttempts, "task_name", name, "outcome", "canceled"))
+			execTime  = co.metrics.GetOrCreatePrometheusHistogramExt(metricName(metricJobExecDuration, "task_name", name), durationBuckets)
+		)
+
+		err := qs.RegisterTask(string(name), func(b []byte, jctx tasqueue.JobCtx) (err error) {
+			running.Inc()
+			defer func() {
+				running.Dec()
+				switch {
+				case err == nil:
+					succeeded.Inc()
+				// A deleted job's state is gone from the store, so GetJob returns ErrNotFound.
+				case errors.Is(err, errJobCanceled), errors.Is(err, tasqueue.ErrNotFound):
+					canceled.Inc()
+				default:
+					errored.Inc()
+				}
+			}()
+
+			if _, err = qs.GetJob(context.Background(), jctx.Meta.ID); err != nil {
 				return err
 			}
 
 			var args taskMeta
-			if err := msgpack.Unmarshal(b, &args); err != nil {
+			if err = msgpack.Unmarshal(b, &args); err != nil {
 				return fmt.Errorf("could not unmarshal args : %w", err)
 			}
 
-			count, err := co.execJob(jctx.Meta.ID, name, args.DB, time.Duration(args.TTL)*time.Second, args.Args, query)
+			// Retries are re-enqueued right after a failed attempt, so only the first attempt measures queue wait.
+			if jctx.Meta.Retried == 0 && !args.ReadyAt.IsZero() {
+				wait := max(time.Since(args.ReadyAt), 0)
+				co.metrics.GetOrCreatePrometheusHistogramExt(
+					metricName(metricJobQueueWait, "task_name", name, "queue", jctx.Meta.Queue),
+					durationBuckets,
+				).Update(wait.Seconds())
+			}
+
+			var count int64
+			start := time.Now()
+			count, err = co.execJob(jctx.Meta.ID, name, args.DB, time.Duration(args.TTL)*time.Second, args.Args, query)
+			execTime.UpdateDuration(start)
 			if err != nil {
 				return fmt.Errorf("could not execute job : %w", err)
 			}
@@ -446,11 +504,14 @@ func (co *Core) initQueue() (*tasqueue.Server, error) {
 	return qs, nil
 }
 
+// errJobCanceled is returned when a job is deleted or canceled before or during execution.
+var errJobCanceled = errors.New("the job was canceled")
+
 // execJob executes an SQL statement job and inserts the results into the result backend.
 func (co *Core) execJob(jobID, taskName, dbName string, ttl time.Duration, args []interface{}, task Task) (int64, error) {
 	// If the job's deleted, stop.
 	if _, err := co.GetJobStatus(jobID); err != nil {
-		return 0, errors.New("the job was canceled")
+		return 0, errJobCanceled
 	}
 
 	// Execute the query.
@@ -495,8 +556,9 @@ func (co *Core) execJob(jobID, taskName, dbName string, ttl time.Duration, args
 
 	rows, err = db.QueryContext(ctx, task.Raw, args...)
 	if err != nil {
-		if err == context.Canceled {
-			return 0, errors.New("the job was canceled")
+		// Drivers report cancellation differently (eg: pq returns its own error), so check the context instead.
+		if ctx.Err() != nil {
+			return 0, errJobCanceled
 		}
 
 		return 0, fmt.Errorf("task execution failed: %s: %v", task.Name, err)

internal/core/metrics.go

@@ -0,0 +1,70 @@
+package core
+
+import (
+	"context"
+	"math"
+	"runtime"
+	"strings"
+)
+
+// Metric names, without the dbtl_ prefix.
+const (
+	metricJobsEnqueued     = "jobs_enqueued_total"
+	metricJobsRunning      = "jobs_running"
+	metricJobAttempts      = "job_attempts_total"
+	metricJobExecDuration  = "job_execution_duration_seconds"
+	metricJobQueueWait     = "job_queue_wait_duration_seconds"
+	metricQueuePendingJobs = "queue_pending_jobs"
+	metricWorkerConc       = "worker_concurrency"
+)
+
+// durationBuckets are histogram upper bounds in seconds, sized for SQL jobs
+// that can take anywhere from milliseconds to an hour.
+var durationBuckets = []float64{0.1, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, 1800, 3600}
+
+// labelEscaper escapes label values as required by the Prometheus text format.
+var labelEscaper = strings.NewReplacer(`\`, `\\`, `"`, `\"`, "\n", `\n`)
+
+// metricName returns the prefixed metric name with labels given as key, value pairs.
+// E.g. metricName("jobs_running", "task_name", "a") -> `dbtl_jobs_running{task_name="a"}`
+func metricName(name string, labels ...string) string {
+	var b strings.Builder
+	b.WriteString("dbtl_")
+	b.WriteString(name)
+	b.WriteByte('{')
+	for i := 0; i+1 < len(labels); i += 2 {
+		if i > 0 {
+			b.WriteByte(',')
+		}
+		b.WriteString(labels[i])
+		b.WriteString(`="`)
+		labelEscaper.WriteString(&b, labels[i+1])
+		b.WriteByte('"')
+	}
+	b.WriteByte('}')
+
+	return b.String()
+}
+
+// registerQueueMetrics registers the pending jobs and worker concurrency gauges for a queue.
+// Repeat calls for the same queue are no-ops.
+func (co *Core) registerQueueMetrics(queue string, conc int) {
+	// Mirror tasqueue's default concurrency for tasks that don't set one.
+	if conc <= 0 {
+		conc = runtime.GOMAXPROCS(0)
+	}
+	co.metrics.GetOrCreateGauge(metricName(metricWorkerConc, "queue", queue), func() float64 {
+		return float64(conc)
+	})
+
+	// Evaluated on every scrape. The broker only lists runnable jobs, so ETA-scheduled ones aren't counted.
+	co.metrics.GetOrCreateGauge(metricName(metricQueuePendingJobs, "queue", queue), func() float64 {
+		jobs, err := co.opt.Broker.GetPending(context.Background(), queue)
+		if err != nil {
+			co.lo.Error("error fetching pending jobs for metrics", "queue", queue, "error", err)
+			return math.NaN()
+		}
+
+		return float64(len(jobs))
+	})
+}