125 lines
4.5 KiB
Go
125 lines
4.5 KiB
Go
// Package observability also exposes Prometheus metrics for push-gateway.
|
|
//
|
|
// The metrics cover the full surface area described in 02 §6.4: active
|
|
// connections, messages pushed/dropped, heartbeats, disconnects, Redis Pub/Sub
|
|
// latency, Kafka consumption and Redis SET rebuilds. All metrics are
|
|
// registered with the global prometheus.DefaultRegisterer on package init.
|
|
package observability
|
|
|
|
import (
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
"github.com/prometheus/client_golang/prometheus/promauto"
|
|
)
|
|
|
|
// Metrics holds all push-gateway Prometheus metric handles.
|
|
type Metrics struct {
|
|
ActiveConnections prometheus.Gauge
|
|
MessagesPushed *prometheus.CounterVec
|
|
MessagesDropped *prometheus.CounterVec
|
|
Heartbeats prometheus.Counter
|
|
Disconnects *prometheus.CounterVec
|
|
RedisPubSubLatency *prometheus.HistogramVec
|
|
KafkaConsumed *prometheus.CounterVec
|
|
RedisSetRebuild prometheus.Counter
|
|
ConnectionsPerUser prometheus.Gauge
|
|
}
|
|
|
|
// metrics is the process-wide metric set, initialized once by NewMetrics.
|
|
var metrics *Metrics
|
|
|
|
// NewMetrics registers and returns the push-gateway metric set. Subsequent
|
|
// calls return the same instance to avoid duplicate-registration panics.
|
|
func NewMetrics() *Metrics {
|
|
if metrics != nil {
|
|
return metrics
|
|
}
|
|
metrics = &Metrics{
|
|
ActiveConnections: promauto.NewGauge(prometheus.GaugeOpts{
|
|
Namespace: "push_gateway",
|
|
Name: "active_connections",
|
|
Help: "Current number of live WebSocket connections.",
|
|
}),
|
|
MessagesPushed: promauto.NewCounterVec(prometheus.CounterOpts{
|
|
Namespace: "push_gateway",
|
|
Name: "messages_pushed_total",
|
|
Help: "Total messages pushed, partitioned by event and result.",
|
|
}, []string{"event", "result"}),
|
|
MessagesDropped: promauto.NewCounterVec(prometheus.CounterOpts{
|
|
Namespace: "push_gateway",
|
|
Name: "messages_dropped_total",
|
|
Help: "Messages dropped, partitioned by reason.",
|
|
}, []string{"reason"}),
|
|
Heartbeats: promauto.NewCounter(prometheus.CounterOpts{
|
|
Namespace: "push_gateway",
|
|
Name: "heartbeat_total",
|
|
Help: "Total WebSocket Ping heartbeats received.",
|
|
}),
|
|
Disconnects: promauto.NewCounterVec(prometheus.CounterOpts{
|
|
Namespace: "push_gateway",
|
|
Name: "disconnect_total",
|
|
Help: "Disconnects, partitioned by reason.",
|
|
}, []string{"reason"}),
|
|
RedisPubSubLatency: promauto.NewHistogramVec(prometheus.HistogramOpts{
|
|
Namespace: "push_gateway",
|
|
Name: "redis_pubsub_latency_seconds",
|
|
Help: "Redis Pub/Sub round-trip latency in seconds.",
|
|
Buckets: prometheus.ExponentialBuckets(0.001, 2, 12),
|
|
}, []string{"direction"}),
|
|
KafkaConsumed: promauto.NewCounterVec(prometheus.CounterOpts{
|
|
Namespace: "push_gateway",
|
|
Name: "kafka_consumed_total",
|
|
Help: "Kafka messages consumed, partitioned by topic and partition.",
|
|
}, []string{"topic", "partition"}),
|
|
RedisSetRebuild: promauto.NewCounter(prometheus.CounterOpts{
|
|
Namespace: "push_gateway",
|
|
Name: "redis_set_rebuild_total",
|
|
Help: "Total Redis online-presence SET rebuilds (startup or recovery).",
|
|
}),
|
|
ConnectionsPerUser: promauto.NewGauge(prometheus.GaugeOpts{
|
|
Namespace: "push_gateway",
|
|
Name: "connections_per_user_max",
|
|
Help: "Maximum concurrent connections held by a single user.",
|
|
}),
|
|
}
|
|
return metrics
|
|
}
|
|
|
|
// Metrics returns the process-wide metric set. When NewMetrics has not been
|
|
// called a new set is registered on demand.
|
|
func MetricsInstance() *Metrics {
|
|
if metrics == nil {
|
|
return NewMetrics()
|
|
}
|
|
return metrics
|
|
}
|
|
|
|
// IncMessageDropped is a convenience helper for the channel-full case.
|
|
func (m *Metrics) IncMessageDropped(reason string) {
|
|
m.MessagesDropped.WithLabelValues(reason).Inc()
|
|
}
|
|
|
|
// IncPushed is a convenience helper for the push result counter.
|
|
func (m *Metrics) IncPushed(event, result string) {
|
|
m.MessagesPushed.WithLabelValues(event, result).Inc()
|
|
}
|
|
|
|
// IncDisconnect is a convenience helper for the disconnect counter.
|
|
func (m *Metrics) IncDisconnect(reason string) {
|
|
m.Disconnects.WithLabelValues(reason).Inc()
|
|
}
|
|
|
|
// ObservePubSubLatency records a Redis Pub/Sub round-trip duration.
|
|
func (m *Metrics) ObservePubSubLatency(direction string, seconds float64) {
|
|
m.RedisPubSubLatency.WithLabelValues(direction).Observe(seconds)
|
|
}
|
|
|
|
// IncKafkaConsumed increments the Kafka consumed counter.
|
|
func (m *Metrics) IncKafkaConsumed(topic, partition string) {
|
|
m.KafkaConsumed.WithLabelValues(topic, partition).Inc()
|
|
}
|
|
|
|
// IncRedisSetRebuild increments the SET rebuild counter (ISSUE-058 metric).
|
|
func (m *Metrics) IncRedisSetRebuild() {
|
|
m.RedisSetRebuild.Inc()
|
|
}
|