Queue Depth Exporter

A custom Prometheus exporter that implements the prometheus.Collector interface to scrape metrics from an external service API. It translates queue depth, job counts, latency, and worker status into Prometheus-compatible metrics at scrape time.

Input: An external service endpoint that exposes a /stats JSON API. Output: Prometheus metrics on :9101/metrics (queue depth, processed jobs, average latency, and active workers).

Setup

mkdir queue-depth-exporter
cd queue-depth-exporter
go mod init github.com/yourorg/queue-depth-exporter
go get github.com/prometheus/client_golang/prometheus@latest
go get github.com/prometheus/client_golang/prometheus/promhttp@latest

main.go

package main

import (
	"encoding/json"
	"log"
	"net/http"
	"os"
	"time"

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

// ServiceStats represents the response from our external service /stats endpoint.
type ServiceStats struct {
	QueueDepth    int     `json:"queue_depth"`
	ProcessedJobs int     `json:"processed_jobs"`
	AvgLatencyMs  float64 `json:"avg_latency_ms"`
	WorkersActive int     `json:"workers_active"`
}

type ServiceCollector struct {
	endpoint     string
	client       *http.Client
	queueDepth   *prometheus.Desc
	jobsTotal    *prometheus.Desc
	latency      *prometheus.Desc
	workersUp    *prometheus.Desc
	scrapeErrors *prometheus.Desc
}

// NewServiceCollector creates a Prometheus Collector that fetches metrics from
// endpoint at each scrape. The HTTP client has a 5s timeout to prevent
// a slow external service from blocking the entire /metrics endpoint.
func NewServiceCollector(endpoint string) *ServiceCollector {
	return &ServiceCollector{
		endpoint: endpoint,
		client:   &http.Client{Timeout: 5 * time.Second},
		queueDepth: prometheus.NewDesc(
			"service_queue_depth",
			"Number of items waiting in the queue.",
			nil, nil,
		),
		jobsTotal: prometheus.NewDesc(
			"service_processed_jobs_total",
			"Total number of processed jobs.",
			nil, nil,
		),
		latency: prometheus.NewDesc(
			"service_avg_latency_seconds",
			"Average processing latency in seconds.",
			nil, nil,
		),
		workersUp: prometheus.NewDesc(
			"service_workers_active",
			"Number of active workers.",
			nil, nil,
		),
		scrapeErrors: prometheus.NewDesc(
			"service_scrape_errors_total",
			"Total scrape errors.",
			nil, nil,
		),
	}
}

// Describe sends all metric descriptors to ch. Required by prometheus.Collector.
func (c *ServiceCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- c.queueDepth
	ch <- c.jobsTotal
	ch <- c.latency
	ch <- c.workersUp
	ch <- c.scrapeErrors
}

// Collect fetches fresh stats from the external service and sends metrics to ch.
// Called by Prometheus on each scrape. Records a scrape error metric on failure.
func (c *ServiceCollector) Collect(ch chan<- prometheus.Metric) {
	resp, err := c.client.Get(c.endpoint + "/stats")
	if err != nil {
		ch <- prometheus.MustNewConstMetric(c.scrapeErrors, prometheus.CounterValue, 1)
		return
	}
	defer resp.Body.Close()

	var stats ServiceStats
	if err := json.NewDecoder(resp.Body).Decode(&stats); err != nil {
		ch <- prometheus.MustNewConstMetric(c.scrapeErrors, prometheus.CounterValue, 1)
		return
	}

	ch <- prometheus.MustNewConstMetric(c.queueDepth, prometheus.GaugeValue, float64(stats.QueueDepth))
	ch <- prometheus.MustNewConstMetric(c.jobsTotal, prometheus.CounterValue, float64(stats.ProcessedJobs))
	ch <- prometheus.MustNewConstMetric(c.latency, prometheus.GaugeValue, stats.AvgLatencyMs/1000.0)
	ch <- prometheus.MustNewConstMetric(c.workersUp, prometheus.GaugeValue, float64(stats.WorkersActive))
}

func main() {
	endpoint := os.Getenv("SERVICE_ENDPOINT")
	if endpoint == "" {
		endpoint = "http://localhost:9000"
	}

	collector := NewServiceCollector(endpoint)
	prometheus.MustRegister(collector)

	http.Handle("/metrics", promhttp.Handler())
	log.Printf("exporter listening on :9101/metrics (scraping %s/stats)", endpoint)
	log.Fatal(http.ListenAndServe(":9101", nil))
}

Running It

The exporter scrapes an external service at /stats. To test locally without a real service, start a mock stats server in a second terminal:

# mock stats server on port 9000
cat > mock_server.go << 'EOF'
package main

import (
	"encoding/json"
	"log"
	"net/http"
)

func main() {
	http.HandleFunc("/stats", func(w http.ResponseWriter, r *http.Request) {
		json.NewEncoder(w).Encode(map[string]interface{}{
			"queue_depth":    42,
			"processed_jobs": 1234,
			"avg_latency_ms": 85.5,
			"workers_active": 4,
		})
	})
	log.Println("mock service on :9000")
	log.Fatal(http.ListenAndServe(":9000", nil))
}
EOF
go run mock_server.go

In another terminal, start the exporter:

SERVICE_ENDPOINT=http://localhost:9000 go run main.go
# exporter listening on :9101/metrics (scraping http://localhost:9000/stats)

Fetch the metrics:

curl http://localhost:9101/metrics | grep service_

Expected output:

# HELP service_avg_latency_seconds Average processing latency in seconds.
# TYPE service_avg_latency_seconds gauge
service_avg_latency_seconds 0.0855
# HELP service_processed_jobs_total Total number of processed jobs.
# TYPE service_processed_jobs_total counter
service_processed_jobs_total 1234
# HELP service_queue_depth Number of items waiting in the queue.
# TYPE service_queue_depth gauge
service_queue_depth 42
# HELP service_scrape_errors_total Total scrape errors.
# TYPE service_scrape_errors_total counter
service_scrape_errors_total 0
# HELP service_workers_active Number of active workers.
# TYPE service_workers_active gauge
service_workers_active 4

Stop the mock server and re-fetch — you'll see service_scrape_errors_total 1 increment as the exporter records the failed scrape instead of crashing.