How to Monitor VarMQ Worker Status: Runtime API and Metrics Guide

VarMQ exposes a built-in runtime API via methods like NumProcessing(), Metrics(), and Status() that report worker health, queue depth, and job counters without requiring external instrumentation.

The goptics/varmq repository provides a lightweight job queue library for Go that includes comprehensive observability hooks. Understanding how to monitor VarMQ worker status is essential for production deployments where visibility into queue depth, processing throughput, and error rates drives scaling decisions. The worker implementation exposes lock-free counters and lifecycle state methods that you can poll directly or export to Prometheus.

Core Runtime Monitoring API

The Worker interface declared in interface.go provides direct access to runtime telemetry. These methods operate without blocking the job processing loop, making them safe to call from health check endpoints or monitoring goroutines.

Queue Depth and Worker Utilization

Four methods report the current capacity and utilization of your worker pool:

• NumProcessing() – Returns the number of jobs currently being processed by active workers. Implemented in worker.go lines 44-46, this queries the internal pool to count busy goroutine nodes.
• NumPending() – Reports the total length of all bound queues (tasks waiting to be picked up). This method delegates to queues.Len() inside the queueManager (see worker.go lines 72-74).
• NumIdleWorkers() – Indicates how many workers in the pool are initialized but not currently processing jobs. Sourced from pool.Len() in worker.go lines 30-32.
• NumConcurrency() – Returns the configured maximum concurrency limit set during worker initialization (worker.go lines 26-28).

Lifecycle State Detection

Monitor the worker's operational state through boolean and string accessors:

• Status() – Returns a human-readable string describing the current lifecycle phase: Initiated, Running, Paused, or Stopped (implemented in worker.go lines 129-134).
• IsPaused(), IsRunning(), IsStopped() – Convenience boolean checks that wrap the internal state machine (see worker.go lines 29-32).

Metrics and Error Streams

For historical tracking and failure detection:

• Metrics() – Returns a Metrics interface exposing lock-free counters for Submitted(), Completed(), Successful(), and Failed() jobs. The underlying struct in metrics.go lines 5-16 uses sync/atomic.Uint64 for high-performance updates.
• Errs() – Provides a read-only channel (<-chan error) that streams runtime errors including panics recovered from job handlers. Referenced in worker.go lines 29-31, this channel enables real-time alerting without polling.

How the Monitoring System Works

The status API leverages several internal components to deliver accurate, low-overhead telemetry.

Job Counters are updated atomically at specific lifecycle points defined in worker.go. The incSubmitted() function triggers when jobs enter the queue via Queue.Add() or PriorityQueue.Add(). After job completion inside initPoolNode().Value.Serve (lines 38-41), the worker calls incCompleted(), incSuccessful(), or incFailed() depending on the outcome.

Queue Aggregation happens through the queueManager struct. When you call NumPending(), the worker delegates to this component, which sums the length of all bound queues managed by the worker instance.

Pool Utilization tracking relies on the internal pool package (internal/pool). The NumIdleWorkers() method reports the cache size of reusable goroutine nodes that are ready but not actively processing.

Practical Monitoring Implementations

Direct Polling for Health Checks

For simple deployments without external dependencies, poll the status methods directly from a goroutine or HTTP handler:

package main

import (
	"fmt"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	// Create a worker with a concurrency of 5
	worker := varmq.NewWorker(func(j varmq.Job[int]) {
		// Simulated work
		time.Sleep(100 * time.Millisecond)
	}, varmq.WithConcurrency(5))

	// Bind a standard queue
	q := worker.BindQueue()

	// Submit a few jobs
	for i := 0; i < 20; i++ {
		q.Add(i)
	}

	// Periodically print worker status
	ticker := time.NewTicker(2 * time.Second)
	for range ticker.C {
		fmt.Printf(
			"Status=%s | Pending=%d | Processing=%d | Idle=%d | Concurrency=%d | Submitted=%d | Completed=%d | Success=%d | Failed=%d\n",
			worker.Status(),
			worker.NumPending(),
			worker.NumProcessing(),
			worker.NumIdleWorkers(),
			worker.NumConcurrency(),
			worker.Metrics().Submitted(),
			worker.Metrics().Completed(),
			worker.Metrics().Successful(),
			worker.Metrics().Failed(),
		)
	}
}

This pattern prints a complete snapshot every two seconds, showing every metric exposed by the API without external instrumentation.

Prometheus Integration

The repository includes a complete Prometheus exporter in examples/prometheus/main.go that registers collectors for all status fields. The implementation wraps VarMQ methods in Prometheus gauge and counter functions:

// registerMetrics registers Prometheus gauges/counters for a VarMQ worker.
// See the full source at: https://github.com/goptics/varmq/blob/main/examples/prometheus/main.go
func registerMetrics(w varmq.Worker, namespace string) {
	// Workers
	prometheus.MustRegister(prometheus.NewGaugeFunc(
		prometheus.GaugeOpts{Namespace: namespace, Name: "workers_processing"},
		func() float64 { return float64(w.NumProcessing()) },
	))
	
	// Queue depth
	prometheus.MustRegister(prometheus.NewGaugeFunc(
		prometheus.GaugeOpts{Namespace: namespace, Name: "waiting_total"},
		func() float64 { return float64(w.NumPending()) },
	))

	// Counters from the internal Metrics interface
	prometheus.MustRegister(prometheus.NewCounterFunc(
		prometheus.CounterOpts{Namespace: namespace, Name: "submitted_total"},
		func() float64 { return float64(w.Metrics().Submitted()) },
	))
}

Run this example with go run ./examples/prometheus to expose metrics on port 8080. Prometheus will scrape gauges for workers_processing, workers_concurrency, workers_idle, and waiting_total, plus counters for submitted_total, completed_total, successful_total, and failed_total.

Consuming Runtime Errors

Attach a goroutine to the error channel to surface failures to your alerting pipeline:

go func() {
	for err := range worker.Errs() {
		// Forward to your alerting pipeline
		fmt.Printf("Worker error: %v\n", err)
	}
}()

The error channel is non-blocking; if no consumer is attached, the worker silently discards errors via the sendError function in worker.go lines 96-103. This design prevents backpressure from slow consumers from blocking job processing.

Summary

• VarMQ worker status is exposed through methods like NumProcessing(), NumPending(), and Status() defined in worker.go, providing real-time visibility without external dependencies.
• Lock-free metrics using sync/atomic.Uint64 track submitted, completed, successful, and failed jobs via the Metrics() interface defined in metrics.go.
• Lifecycle states (Running, Paused, Stopped) are queryable through boolean checks or human-readable strings.
• Error streaming via Errs() captures panics and runtime failures for immediate alerting.
• Production monitoring can be implemented through direct polling, Prometheus exporters (as shown in examples/prometheus/main.go), or custom exporters consuming the error channel.

Frequently Asked Questions

How do I check if a VarMQ worker is currently busy?

Call worker.NumProcessing() to get the count of active jobs, or check worker.NumIdleWorkers() to see available capacity. If NumProcessing() equals NumConcurrency(), the worker pool is saturated. Both methods are implemented in worker.go and read atomic counters updated by the internal pool manager.

Can I export VarMQ metrics to Prometheus without writing custom code?

Yes. The goptics/varmq repository includes a ready-to-run Prometheus exporter in examples/prometheus/main.go. This example registers gauges for queue depth, processing workers, and idle workers, plus counters for job outcomes. Run it with go run ./examples/prometheus to expose metrics on port 8080.

What happens to errors if I don't consume the Errs() channel?

The worker's sendError function (lines 96-103 in worker.go) implements a non-blocking send with a default case. If no goroutine is receiving from Errs(), the error is silently dropped to prevent blocking the job processing loop. Always consume this channel in production to capture panics and job failures.

How accurate are the pending job counts under high concurrency?

NumPending() is eventually consistent but accurate for monitoring purposes. It queries the queueManager to sum all bound queue lengths (see worker.go lines 72-74). Since the method uses the internal queues.Len() operation on each bound queue, it reflects the state at the moment of invocation without locking the entire worker.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}