Metrics

Every worker keeps four counters over its lifetime. Metrics() returns them. They tell you how much work the worker has taken in and how it went, without tracking jobs yourself.

This page covers workers bound to in-memory queues. For persistent and distributed queues, see persistence.

The four counters

Metrics has four read methods, each returning a uint64:

A job that runs counts once in Completed and once in Successful or Failed. After a worker has gone idle, Completed equals Successful plus Failed, unless you called Reset partway.

A job that never runs counts in Submitted and in none of the others. That happens to a job that is closed before it starts: closed through its handle, or removed by Purge. It also happens to a topic payload that is dropped unread, as below. Only a job that ran increments Completed.

A topic counts too: each Publish adds one to Submitted of every worker subscribed to it at that moment. The subscriber’s worker turns a payload into a job only when it takes the payload. So an unread payload is counted as submitted, but it is not a job yet. Unsubscribing drops such payloads.

Reading them

The program below adds three jobs to an error worker. Job 2 returns an error. It waits until the worker is idle and reads the four counters.

package main

import (
	"fmt"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewErrWorker(func(j varmq.Job[int]) error {
		if j.Data() == 2 {
			return fmt.Errorf("job %d failed", j.Data())
		}
		return nil
	})
	queue := worker.BindQueue()

	for i := 1; i <= 3; i++ {
		queue.Add(i)
	}
	worker.WaitUntilIdle()

	m := worker.Metrics()
	fmt.Println("completed", m.Completed())
	fmt.Println("successful", m.Successful())
	fmt.Println("failed", m.Failed())
}

All three jobs were submitted and completed. Two succeeded, and job 2 is the one failed job, so Completed equals Successful plus Failed.

• The counters are cumulative. They start at zero when the worker is created, and nothing but Reset clears them. Pausing, stopping, restarting and purging do not change them.
• They are safe to read from any goroutine. Each method is one atomic read.
• Four reads are not one snapshot. A job can finish between two of them. So four values read while jobs run can be off by as many jobs as finish between the reads.
• A job’s own Wait can return a moment before Completed counts it. The worker closes a finished job, which releases Wait. Just after that, the worker counts the job as completed. Successful and Failed are already counted by then. To read settled values, wait for the worker with WaitUntilIdle or Wait first. Waiting compares them.

Submitted, NumPending, NumProcessing and Completed

On a worker with no job waiting or running, Submitted == Completed. Three things break this: a job that was closed before it ran, storage shared with another worker, and a Reset partway. A closed job is the most common case: you purged it, closed it through its handle, or a topic dropped it unread.

The rule comes from a general equation. A job is waiting, running or done, so for a worker with nothing running you might expect:

Submitted == NumPending + NumProcessing + Completed

NumPending is the jobs that have not started, plus the finished jobs of a batch that has not ended. NumProcessing is the jobs running right now, counted as busy goroutines. With no job waiting or running, both are 0, so the equation becomes Submitted == Completed.

When the equation does not hold

The equation does not hold in these cases:

• A job closed before it ran breaks it. Purge, and closing a queued job through its handle, leave Submitted higher than the sum. Such a job was counted when it was added, and it never reaches Completed, because only a job that ran does. Once the worker has dropped it, it is not pending either, so nothing on the right counts it. A job closed through its handle still counts in NumPending until the worker reaches it, so the gap opens then.
• While jobs run, the sum can be too high. A goroutine can take several jobs from an in-memory FIFO queue or a topic into a batch (batched dispatch). NumProcessing counts that goroutine as one. NumPending keeps counting the finished jobs of the batch until the whole batch has ended. But Completed counts each job as it finishes. For that time the sum can run above Submitted.
• Unsubscribing with unread payloads breaks it. Unsubscribe, Close on the subscription, and Close on the topic all end a subscription. They drop the payloads it has not read. Publish counted them in Submitted. They never run, so they never reach Completed. The worker no longer counts them as pending, because their subscription is gone. Purge on a subscription drops the unread payloads the same way.
• Storage shared between workers breaks it. Add counts in Submitted on the worker whose handle you call. The worker that takes the job counts it in Completed. A queue you wrote and bound to two workers with WithQueue splits the counters between them, so the equation holds for neither.
• Reset breaks it. It zeroes Submitted and Completed. The jobs still waiting or running count on the right side of the equation, and no longer on the left.

So use the equation on a worker with nothing running, where none of these cases applies. If Submitted == Completed does not hold on a worker with no job waiting or running, look at the cases above first. Do not use the equation as a live check while jobs run.

The program below checks the equation on an idle worker, then on a worker paused with PauseAndWait with two jobs waiting, then after Purge.

package main

import (
	"fmt"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewWorker(func(j varmq.Job[int]) {
		fmt.Println("ran job", j.Data())
	})
	queue := worker.BindQueue()
	m := worker.Metrics()

	show := func(label string) {
		fmt.Printf("%s: submitted %d, pending %d + processing %d + completed %d\n",
			label, m.Submitted(), worker.NumPending(), worker.NumProcessing(), m.Completed())
	}

	queue.Add(1)
	worker.WaitUntilIdle()
	show("idle")

	worker.PauseAndWait()
	queue.Add(2)
	queue.Add(3)
	show("paused")

	queue.Purge()
	show("purged")
}

The equation holds on the idle worker: 1 = 0 + 0 + 1. It also holds on the paused one: 3 = 2 + 0 + 1. A paused worker starts nothing, and its two jobs wait in the queue. After Purge the right side is 1 and Submitted is still 3: the two purged jobs never ran, so they never reach Completed.

Reset

Reset sets all four counters to zero. It does not touch the queue or the jobs that are running.

• The four counters are cleared one after another, not in one step.
• A job that was waiting or running when you call it is counted in Completed when it finishes, though its Submitted is gone. After a Reset in the middle of a run, Completed can be higher than Submitted.

Call Reset when the worker is idle, as the program below does.

package main

import (
	"fmt"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewWorker(func(j varmq.Job[int]) {
		fmt.Println("ran job", j.Data())
	})
	queue := worker.BindQueue()
	m := worker.Metrics()

	queue.Add(1)
	queue.Add(2)
	worker.WaitUntilIdle()
	fmt.Println("before Reset: completed", m.Completed())

	m.Reset()
	fmt.Println("after Reset: completed", m.Completed())
}

Both jobs ran before the Reset, so the counters read 2 before it and 0 after.

Where else the counters appear

A named worker’s counters are also what the REST handler reports, and the Prometheus example exports the same four. See REST API and Prometheus.