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.

A longer version of the program below is in examples/prometheus.

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.

14 collapsed lines
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("jobs_waiting", "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)
}

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

   Copy

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"

   Copy

   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_jobs_waiting 800
   foo_worker_workers_concurrency 10
   foo_worker_workers_idle 0
   foo_worker_workers_processing 10

   Copy

4. Ask again after about fifteen seconds. All 1000 jobs have completed, jobs_waiting 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.

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.

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 it does not count a job that another process added. 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

• 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.

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.