Prometheus

varmq imports no metrics library. A worker exposes what Prometheus needs as plain methods: how many jobs are running and waiting, and four counters of what has happened. You wrap those methods in collectors from the Prometheus Go client. Its GaugeFunc and CounterFunc call a function each time Prometheus scrapes, so nothing of yours has to poll the worker.

This page follows examples/prometheus, which is its own Go module and uses client_golang v1.23.2.

A worker with metrics

The program below runs a worker with concurrency 10, registers eight collectors for it, and serves them at /metrics. It adds 1000 jobs that each take 100 milliseconds.

package main

import (
  "fmt"
  "log"
  "net/http"
  "time"

  "github.com/goptics/varmq"
  "github.com/prometheus/client_golang/prometheus"
  "github.com/prometheus/client_golang/prometheus/promhttp"
)

func main() {
  worker := varmq.NewWorker(
    func(j varmq.Job[string]) {
      time.Sleep(100 * time.Millisecond)
    },
    varmq.WithConcurrency(10),
  )
  queue := worker.BindQueue()

  registerMetrics(worker, "foo_worker")
  http.Handle("/metrics", promhttp.Handler())

  go func() {
    for i := range 1000 {
      queue.Add(fmt.Sprintf("task-%d", i))
    }
  }()

  log.Println("metrics on http://localhost:8080/metrics")
  log.Fatal(http.ListenAndServe(":8080", nil))
}

// Each collector calls its function on every scrape, so a value is read when
// Prometheus asks for it.
func registerMetrics(w varmq.Worker, namespace string) {
  gauge := func(name, help string, read func() int) {
    prometheus.MustRegister(prometheus.NewGaugeFunc(
      prometheus.GaugeOpts{Namespace: namespace, Name: name, Help: help},
      func() float64 { return float64(read()) },
    ))
  }
  counter := func(name, help string, read func() uint64) {
    prometheus.MustRegister(prometheus.NewCounterFunc(
      prometheus.CounterOpts{Namespace: namespace, Name: name, Help: help},
      func() float64 { return float64(read()) },
    ))
  }

  m := w.Metrics()

  gauge("workers_processing", "Jobs being processed right now", w.NumProcessing)
  gauge("workers_concurrency", "Maximum number of jobs processed at once", w.NumConcurrency)
  gauge("workers_idle", "Idle goroutines in the pool", w.NumIdleWorkers)
  gauge("waiting_total", "Jobs not started yet; can read high while batches run", w.NumPending)

  counter("submitted_total", "Jobs the queue accepted", m.Submitted)
  counter("completed_total", "Jobs whose function has returned", m.Completed)
  counter("successful_total", "Jobs that returned without an error or a panic", m.Successful)
  counter("failed_total", "Jobs that returned an error or panicked", m.Failed)
}

The repository’s example registers collectors with the same eight names, one MustRegister call each, on a worker with the same concurrency. The helpers above shorten that, and their help texts are shorter.

Run locally

1. Make a module, save the program as main.go and add the dependencies:

   mkdir prometheus-demo && cd prometheus-demo
   go mod init prometheus-demo
   go get github.com/goptics/varmq github.com/prometheus/client_golang

2. Run it with go run ..

3. In another terminal, read the metrics while the jobs run:

   curl -s localhost:8080/metrics | grep "^foo_worker"

   The output looks like this a moment after the start. The numbers depend on when you ask:

   foo_worker_completed_total 190
   foo_worker_failed_total 0
   foo_worker_submitted_total 1000
   foo_worker_successful_total 190
   foo_worker_waiting_total 800
   foo_worker_workers_concurrency 10
   foo_worker_workers_idle 0
   foo_worker_workers_processing 10

4. Ask again after about fifteen seconds. All 1000 jobs have completed, waiting_total and workers_processing read 0, and workers_idle reads 1.

Leave out grep to see the # HELP and # TYPE lines too, and the Go runtime and process metrics. promhttp.Handler() serves the default registry, which has those as well.

The metrics

With the namespace foo_worker, each name starts with foo_worker_. The scene below replays the scrape of the program above. Move the clock, press a metric to read what it is, and see the PromQL for it. The numbers follow a real run of the program: 1000 jobs, 100 milliseconds each, 10 at a time.

Scrape replay · simulation

A replay of what a Prometheus scrape of /metrics returns while the example program runs 1000 jobs with a concurrency of 10, each job taking 100 milliseconds. The left side lists the eight metrics in the text format, each with its HELP line, its TYPE line and its value. A slider sets the time from 0 to 11 seconds, and the values change with it. At 2 seconds the scrape shows completed 190, waiting 800 and processing 10. Press a metric to see its type, the varmq method behind it, what it means, a chart of its value over the run, and PromQL queries for it. A checkbox restarts the process at 5 seconds, which zeroes the counters.

PlayTime

2.0 s

# HELP foo_worker_completed_total Jobs whose function has returned# TYPE foo_worker_completed_total counterfoo_worker_completed_total 190# HELP foo_worker_failed_total Jobs that returned an error or panicked# TYPE foo_worker_failed_total counterfoo_worker_failed_total 0# HELP foo_worker_submitted_total Jobs the queue accepted# TYPE foo_worker_submitted_total counterfoo_worker_submitted_total 1000# HELP foo_worker_successful_total Jobs that returned without an error or a panic# TYPE foo_worker_successful_total counterfoo_worker_successful_total 190# HELP foo_worker_waiting_total Jobs not started yet; can read high while batches run# TYPE foo_worker_waiting_total gaugefoo_worker_waiting_total 800# HELP foo_worker_workers_concurrency Maximum number of jobs processed at once# TYPE foo_worker_workers_concurrency gaugefoo_worker_workers_concurrency 10# HELP foo_worker_workers_idle Idle goroutines in the pool# TYPE foo_worker_workers_idle gaugefoo_worker_workers_idle 0# HELP foo_worker_workers_processing Jobs being processed right now# TYPE foo_worker_workers_processing gaugefoo_worker_workers_processing 10

curl -s localhost:8080/metrics | grep '^foo_worker'

Restart the process at 5 s

Numbers follow a real run of this program.

Name	Type	From	What it is
workers_processing	gauge	NumProcessing	jobs being processed right now
workers_concurrency	gauge	NumConcurrency	the concurrency limit
workers_idle	gauge	NumIdleWorkers	idle goroutines in the pool
waiting_total	gauge	NumPending	jobs that have not started. While a batch runs it can read high: see waiting
submitted_total	counter	Submitted	jobs the queue accepted
completed_total	counter	Completed	jobs whose function has returned
successful_total	counter	Successful	jobs that returned without an error or a panic
failed_total	counter	Failed	jobs that returned an error or panicked

The four counters are the four of Metrics(). Metrics explains what each counts. Waiting explains what NumPending and NumProcessing count.

Queries to start with

These use the namespace foo_worker. Change it to the namespace you pass to each collector.

Question	PromQL
How many jobs finish each second?	rate(foo_worker_completed_total[5m])
What share of finished jobs fail?	rate(foo_worker_failed_total[5m]) / rate(foo_worker_completed_total[5m])
How full is the pool? (1 means every slot runs a job)	foo_worker_workers_processing / foo_worker_workers_concurrency
How large is the backlog?	foo_worker_waiting_total
Do jobs arrive faster than they finish?	rate(foo_worker_submitted_total[5m]) - rate(foo_worker_completed_total[5m])

The failure share is NaN while no job finishes. The pool can read above 1 just after TunePool lowers the limit, while the jobs already running finish. The last row is above 0 while jobs arrive faster than they finish. On a persistent queue it can mislead: Submitted counts only this process’s Add calls, so a job another process added is not counted, and Completed can then be higher than Submitted. On a distributed queue the counters may differ from what you expect. The overview says what each counter counts there.

What to watch for

• waiting_total is a gauge. It goes up and down. The _total suffix is the Prometheus convention for counters, so you may want another name for it in your own code.
• The reads are not one snapshot. A scrape reads eight values at slightly different moments, and a job can finish between two of them. A set read while jobs run can be off by a few. Metrics says more.
• The counters start at zero when the worker is created. They are totals. Use rate() or increase() in Prometheus to see how many jobs ran in a period. A restart of the process zeroes them, and Prometheus functions such as rate() read a drop as a counter reset.
• Reset makes a counter fall. It zeroes the four counters, and Prometheus reads that as a reset, like a restart. Do not call it on a worker you export unless you want that.
• One namespace for each worker. MustRegister panics when a collector with the same metric name is registered twice, with duplicate metrics collector registration attempted. Register each worker under its own namespace.
• A worker on a persistent or distributed queue asks its storage. NumPending reads the length from Redis or SQLite, so every scrape makes that call. For the counters of such a worker, see the overview. The Submitted count of a distributed queue may differ from what you expect, and these docs do not cover how a distributed queue’s counters count yet.

Scraping it

Point Prometheus at the address the program listens on. A minimal scrape configuration is:

scrape_configs:
  - job_name: varmq
    static_configs:
      - targets: ["localhost:8080"]

Prometheus scrapes /metrics by default, which is where the program serves them. If you serve varmq’s REST API as well, it can sit on the same server: mount it with http.Handle("/varmq/", varmq.Handler("/varmq")) next to the metrics route. The REST API lists only named workers, so give the worker a name. The REST API returns the same worker numbers as JSON, which suits a dashboard, and Prometheus suits history and alerts.