131 lines
4.5 KiB
Go
131 lines
4.5 KiB
Go
// Package metrics defines the broker's Prometheus collectors and a small
|
|
// embedded HTTP server that exposes them, both as the standard /metrics
|
|
// exposition endpoint and as a single-page HTML dashboard.
|
|
package metrics
|
|
|
|
import (
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
)
|
|
|
|
// Metrics holds every Prometheus collector the broker exposes. It is safe
|
|
// for concurrent use, and a nil *Metrics is safe to call methods on (all
|
|
// recording methods become no-ops), so callers that construct a Broker
|
|
// without metrics enabled don't need to special-case it.
|
|
type Metrics struct {
|
|
Registry *prometheus.Registry
|
|
|
|
jobsCompleted *prometheus.CounterVec
|
|
jobsFailed *prometheus.CounterVec
|
|
jobsRequeued *prometheus.CounterVec
|
|
jobDuration *prometheus.HistogramVec
|
|
jobsQueued *prometheus.GaugeVec
|
|
databaseCount prometheus.Gauge
|
|
queueCount *prometheus.GaugeVec
|
|
}
|
|
|
|
// New creates a Metrics instance with a fresh (non-global) registry, so
|
|
// multiple brokers can coexist in the same process -- e.g. in tests --
|
|
// without colliding on prometheus.DefaultRegisterer.
|
|
func New() *Metrics {
|
|
registry := prometheus.NewRegistry()
|
|
|
|
m := &Metrics{
|
|
Registry: registry,
|
|
jobsCompleted: prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "broker_jobs_completed_total",
|
|
Help: "Total number of jobs that completed successfully.",
|
|
}, []string{"database", "job_group", "job_name"}),
|
|
jobsFailed: prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "broker_jobs_failed_total",
|
|
Help: "Total number of jobs that were dead-lettered after exhausting retries.",
|
|
}, []string{"database", "job_group", "job_name"}),
|
|
jobsRequeued: prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "broker_jobs_requeued_total",
|
|
Help: "Total number of job attempts that failed and were requeued for retry.",
|
|
}, []string{"database", "job_group", "job_name"}),
|
|
jobDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "broker_job_duration_seconds",
|
|
Help: "Job execution duration in seconds, by group and name.",
|
|
Buckets: prometheus.DefBuckets,
|
|
}, []string{"database", "job_group", "job_name"}),
|
|
jobsQueued: prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "broker_jobs_queued",
|
|
Help: "Current number of pending (not yet claimed) jobs, by database and queue.",
|
|
}, []string{"database", "queue"}),
|
|
databaseCount: prometheus.NewGauge(prometheus.GaugeOpts{
|
|
Name: "broker_databases",
|
|
Help: "Number of database instances managed by this broker process.",
|
|
}),
|
|
queueCount: prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "broker_queues",
|
|
Help: "Number of queues configured for a database instance.",
|
|
}, []string{"database"}),
|
|
}
|
|
|
|
registry.MustRegister(
|
|
m.jobsCompleted,
|
|
m.jobsFailed,
|
|
m.jobsRequeued,
|
|
m.jobDuration,
|
|
m.jobsQueued,
|
|
m.databaseCount,
|
|
m.queueCount,
|
|
)
|
|
|
|
return m
|
|
}
|
|
|
|
// RecordJobCompleted records a successfully completed job attempt.
|
|
func (m *Metrics) RecordJobCompleted(database, group, name string, duration time.Duration) {
|
|
if m == nil {
|
|
return
|
|
}
|
|
m.jobsCompleted.WithLabelValues(database, group, name).Inc()
|
|
m.jobDuration.WithLabelValues(database, group, name).Observe(duration.Seconds())
|
|
}
|
|
|
|
// RecordJobFailed records a job attempt that was dead-lettered (attempts exhausted).
|
|
func (m *Metrics) RecordJobFailed(database, group, name string, duration time.Duration) {
|
|
if m == nil {
|
|
return
|
|
}
|
|
m.jobsFailed.WithLabelValues(database, group, name).Inc()
|
|
m.jobDuration.WithLabelValues(database, group, name).Observe(duration.Seconds())
|
|
}
|
|
|
|
// RecordJobRequeued records a job attempt that failed but was requeued for retry.
|
|
func (m *Metrics) RecordJobRequeued(database, group, name string, duration time.Duration) {
|
|
if m == nil {
|
|
return
|
|
}
|
|
m.jobsRequeued.WithLabelValues(database, group, name).Inc()
|
|
m.jobDuration.WithLabelValues(database, group, name).Observe(duration.Seconds())
|
|
}
|
|
|
|
// SetJobsQueued sets the current pending job count for a database/queue pair.
|
|
func (m *Metrics) SetJobsQueued(database string, queue int, count float64) {
|
|
if m == nil {
|
|
return
|
|
}
|
|
m.jobsQueued.WithLabelValues(database, strconv.Itoa(queue)).Set(count)
|
|
}
|
|
|
|
// SetDatabaseCount sets the number of database instances managed by this process.
|
|
func (m *Metrics) SetDatabaseCount(n int) {
|
|
if m == nil {
|
|
return
|
|
}
|
|
m.databaseCount.Set(float64(n))
|
|
}
|
|
|
|
// SetQueueCount sets the number of queues configured for a database instance.
|
|
func (m *Metrics) SetQueueCount(database string, n int) {
|
|
if m == nil {
|
|
return
|
|
}
|
|
m.queueCount.WithLabelValues(database).Set(float64(n))
|
|
}
|