REST API

VarMQ ships an HTTP handler that lets you look at your workers, queues, jobs and workflows, and control them from outside the process. It serves JSON. It lists only workers that have a name. /queues lists their in-memory queues, and shared storage bound to a worker in this process.

Routes at a glance

The paths follow the prefix you give to Handler, for example /varmq.

Mounting it

Handler takes a path prefix and returns an http.Handler. It strips the prefix from each request before it looks for a route. Mount it on a mux with the prefix in the pattern, with a trailing slash:

mux := http.NewServeMux()
mux.Handle("/varmq/", varmq.Handler("/varmq"))
http.ListenAndServe("127.0.0.1:8080", mux)

You can also pass the handler to ListenAndServe itself, as examples/api-control does with varmq.Handler("/api"). Then the routes start at /api.

NewServerMux returns the same routes as an *http.ServeMux with no prefix, for mounting them yourself. Handler is NewServerMux behind that prefix, with one addition. It sets the instance header on every response. This includes the 404 and 405 replies that the mux generates for itself. With NewServerMux alone, those two replies do not carry the header.

The program below starts two named workers and an unnamed one, adds jobs to them, and serves the API under /varmq.

12 collapsed lines
package main

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

	"github.com/goptics/varmq"
)

func main() {
	// A string argument names the worker.
	emails := varmq.NewWorker(
		func(j varmq.Job[string]) {
			time.Sleep(2 * time.Second)
		},
		"email-sender",
		2,
	)
	emailQueue := emails.BindQueue()

	// WithName does the same.
	images := varmq.NewWorker(
		func(j varmq.Job[int]) {
			time.Sleep(500 * time.Millisecond)
		},
		varmq.WithName("image-processor"),
		varmq.WithConcurrency(4),
	)
	imageQueue := images.BindQueue()

	// This worker has no name, so the API does not list it.
	varmq.NewWorker(func(j varmq.Job[int]) {}).BindQueue()

	go func() {
		for i := range 20 {
			emailQueue.Add(fmt.Sprintf("mail-%d", i))
			imageQueue.Add(i)
		}
	}()

	mux := http.NewServeMux()
	mux.Handle("/varmq/", varmq.Handler("/varmq"))

	// The API has no authentication, so listen on loopback only.
	log.Println("listening on 127.0.0.1:8080")
	log.Fatal(http.ListenAndServe("127.0.0.1:8080", mux))
}

Run locally

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

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

   Copy

2. Run it with go run ..

3. In another terminal, ask for the workers:

   curl localhost:8080/varmq/health
   curl localhost:8080/varmq/workers
   curl localhost:8080/varmq/workers/email-sender

   Copy

   The first prints {"status":"ok"}. The second prints a JSON array with two workers, email-sender and image-processor, and not the unnamed one. The third prints one worker with its counters, for example:

   {"name":"email-sender","status":"Running","concurrency":2,"num_processing":2,"num_pending":14,"num_idle_workers":0,"queue_ids":["memory:host-a:fifo:1"],"strategy":"priority","sources":[{"queue_id":"memory:host-a:fifo:1","kind":"queue"}],"metrics":{"completed":4,"successful":4,"failed":0}}

   Copy

   The numbers depend on when you ask. The middle of the queue ID is this process’s instance ID, the host name by default.

4. Control a worker:

   curl -i -X PATCH localhost:8080/varmq/workers/email-sender/actions/pause
   curl localhost:8080/varmq/workers/email-sender
   curl -X PATCH localhost:8080/varmq/workers/email-sender/actions/resume
   curl -i -X PATCH localhost:8080/varmq/workers/email-sender/config/concurrency/4

   Copy

   The first prints a 200 OK with no body. While a job is running, the next call shows the worker as Pausing, and later Paused, as Pause does. Run resume once it reads Paused: on a worker that is still Pausing, it is refused (see the status codes). The last call changes the limit to 4.

The repository’s examples/api-control runs four named workers with a steady supply of jobs. vartop is a separate terminal dashboard for VarMQ workers that reads this API, and the example’s last comment suggests pointing it at http://localhost:8080/api.

Which workers it serves

The API serves the workers in WorkerRegistry, a package-level sync.Map from name to worker.

• Only a named worker is registered. WithName and a plain string argument to NewWorker both give a worker a name. A worker with no name is not in the registry and cannot be reached.
• A worker is registered when it starts. A worker created with WithAutoRun(false) is not in the registry until you call Start.
• A name holds one worker. When a second worker with the same name starts, it replaces the first in the registry. The API then reaches the second.
• A stopped worker stays for a while. After it reaches Stopped, the WithStoppedRetention time starts. The default is one minute. When that time has passed, the worker is removed from the registry, unless it was started again. Lifecycle has the details.
• The order of /workers is not defined. It follows the order the map yields entries in.

You can read the registry yourself. varmq.WorkerRegistry.Load("email-sender") returns the worker as an any. A type assertion to varmq.Worker gives you its methods. You can then build your own routes on it.

Routes

Routes at a glance has the table of routes.

On a worker route, the {action} is one of pause, resume, restart, start and stop. Each calls the worker method of the same name, which lifecycle describes. Two behaviours follow from that:

• pause and stop return immediately. A worker that is running jobs reads Pausing or Stopping until they end. Read the worker again to see when it has changed.
• restart waits. Restart stops the worker, waits until it has stopped, and starts it. The request is answered after that, so it can take as long as the running jobs do.

The concurrency route calls TunePool with the number. A number below 1 becomes the CPU count. Concurrency covers what a change does to jobs already running.

What a worker looks like

/workers returns a JSON array of these. /workers/{name} returns one, with strategy, sources and metrics added:

Source fields

Each entry of sources has three fields:

• queue_id is the queue’s /queues ID. It is "" for a topic subscription, or for storage with no queue ID.
• kind is queue or topic.
• priority comes from WithQueuePriority or WithTopicPriority. It is absent when not set, or set to math.MaxInt, which ranks the same.

The list is sorted by priority, lowest first, then the sources with none. Sources with the same priority keep the order you bound them in.

Waiting covers what NumPending and NumProcessing count, and Metrics covers the three counters. For a worker on shared storage, see the overview.

Queues and jobs

/queues lists every queue that can report its state. These are the in-memory queues of named workers. They also include shared storage, such as redisq or sqliteq, bound to a worker in this process. A queue that several workers share is listed once. Each entry looks like this:

{"id":"memory:host-a:fifo:1","counts":{"pending":18,"processing":2,"completed":0,"failed":0},"totals":{"submitted":20,"completed":0,"failed":0},"paused":false}

In Go, an entry is a QueueStats, with JobCounts and JobTotals. A job record is a JobRecord, with one JobError per failed attempt. Its state is a JobState: JobPending, JobProcessing, JobCompleted, JobFailed or JobCancelled. A workflow task can also be JobWaiting.

Pausing a queue. PATCH /queues/{id}/actions/pause and …/resume pause and resume the queue. They answer 200 with the queue’s entry. On shared storage, the pause stops every process that takes from the queue, and it survives a restart.

Listing jobs. GET /queues/{id}/jobs needs a state: pending, processing, completed, failed or cancelled. offset starts at 0. limit is 50 by default and at most 1000. Each record has the job’s id, state, payload, timestamps, attempts and errors. A list leaves out a job’s result and gives its size as result_size.

Reading one job. GET /queues/{id}/jobs/{jobId} returns one record, with its result. The storage must be able to read a job by its ID. The in-memory queues can. redisq and sqliteq cannot yet, so they answer 501.

Cancelling a job. PATCH /queues/{id}/jobs/{jobId}/actions/cancel answers 202 with no body. It works like Cancel on the job’s handle.

• An in-memory queue finds a job with an ID while the job waits in the queue or runs. It answers 404 for any other ID.
• Shared storage cannot tell a waiting job from an unknown ID. So it answers 202 for any ID with no ended record. It marks the ID cancelled for 24 hours, or until a new job is added with that ID through WithJobId.
• A job whose ended record is still kept answers 409.
• A waiting workflow task on shared storage answers 202 but still runs. To stop it, cancel its workflow. A running task’s Context() is cancelled, as for any job.

An in-memory queue keeps no ended jobs by default. Pass WithEndedJobs when you bind it, so the completed, failed and cancelled lists have records. The in-memory job routes work only for a named worker, after you build the handler.

Workflows

The workflow routes read the workflow stores this process knows. These are the in-memory stores you made with NewWorkflowStore, and every store an adapter registered, such as redisq.NewWorkflowStore. A process that only submits workflows, and runs no worker, lists and cancels them too. So protect the handler there as well.

• GET /workflows lists the workflows of every store, newest first. state filters them, for example running or failed. offset is at most 10000, and limit is 50 by default and at most 1000.
• GET /workflows/{id} returns one workflow, with the state of each task.
• PATCH /workflows/{id}/actions/cancel cancels the workflow and answers 200 with its record. Add ?undo=false to skip the undo steps. The API cannot turn on undo steps that the workflow turned off: ?undo=true answers 400.

Responses

Every response with a body sets Content-Type: application/json. A successful worker action, and a job cancel, have no body. An error is a JSON object with one field, error.

The 409 and 500 rows of the worker routes follow the error the worker returns.

• 409: ErrRunningWorker, ErrNotRunningWorker and ErrSameConcurrency. So start on a running worker, or a concurrency change to the same number, is a 409.
• 500: ErrWorkerPausing, ErrWorkerPaused, ErrWorkerStopping and ErrWorkerStopped. So resume on a Pausing worker is a 500 with worker is pausing.

Errors lists which call returns which error. A workflow route also answers 409 when two stores hold the same workflow ID.

Two replies do not come from the routes. A path the mux does not know gets a plain-text 404 page not found. A known path with the wrong method gets a plain-text 405 Method Not Allowed with an Allow header.

Which instance answered

InstanceHeader is the name of a response header, X-Varmq-Instance. Its value is InstanceID. When several instances sit behind one URL, such as a load balancer, the header tells you which process answered.

InstanceID is a package variable. It starts as the host name, which on Kubernetes is the pod name, or as an empty string if the host name is not available. Set it before you start serving when several instances share a host:

varmq.InstanceID = "worker-a"

Every response from the routes carries the header, whether you use Handler or NewServerMux. As described above, only Handler also adds it to the mux’s own 404 and 405 replies.

Next: Prometheus.