398 lines
17 KiB
Go
398 lines
17 KiB
Go
package metrics
|
|
|
|
import (
|
|
"expvar"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
)
|
|
|
|
var (
|
|
Registry = prometheus.NewRegistry()
|
|
|
|
HTTPRequestsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "http_requests_total",
|
|
Help: "Total number of HTTP requests by method, normalized path, and status.",
|
|
}, []string{"method", "path", "status"})
|
|
|
|
HTTPRequestDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "http_request_duration_seconds",
|
|
Help: "HTTP request duration by method, normalized path, and status.",
|
|
Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
|
|
}, []string{"method", "path", "status"})
|
|
|
|
DBQueryDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "db_query_duration_seconds",
|
|
Help: "Database operation duration by driver, operation, and status.",
|
|
Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5},
|
|
}, []string{"driver", "operation", "status"})
|
|
|
|
RealtimeFanoutEventsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "realtime_fanout_events_total",
|
|
Help: "Total number of realtime fanout events by stage, kind, target type, and status.",
|
|
}, []string{"stage", "kind", "target_type", "status"})
|
|
|
|
RealtimeFanoutDeliveryDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "realtime_fanout_delivery_duration_seconds",
|
|
Help: "Realtime fanout delivery age measured from message timestamp to local dispatch.",
|
|
Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
|
|
}, []string{"kind", "target_type"})
|
|
|
|
WSConnectionsActive = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
|
|
Name: "ws_connections_active",
|
|
Help: "Current active WebSocket connections.",
|
|
}, func() float64 {
|
|
return float64(expvarIntValue("LiveSessions"))
|
|
})
|
|
IMWSConnectionsActive = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
|
|
Name: "im_ws_connections_active",
|
|
Help: "Current active WebSocket connections, using the IM metric namespace.",
|
|
ConstLabels: prometheus.Labels{"pod": metricPodName()},
|
|
}, func() float64 {
|
|
return float64(expvarIntValue("LiveSessions"))
|
|
})
|
|
IMWSWriteTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_ws_write_total",
|
|
Help: "WebSocket frame write and pre-write rejection outcomes by pod, frame type, result, and bounded reason.",
|
|
}, []string{"pod", "frame", "result", "reason"})
|
|
IMWSWriteDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_ws_write_duration_seconds",
|
|
Help: "WebSocket frame write duration by pod, frame type, and result.",
|
|
Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
|
|
}, []string{"pod", "frame", "result"})
|
|
IMWSWriteBytesTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_ws_write_bytes_total",
|
|
Help: "WebSocket payload bytes submitted to the socket by pod and frame type.",
|
|
}, []string{"pod", "frame"})
|
|
IMWSDisconnectTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_ws_disconnect_total",
|
|
Help: "WebSocket read-loop disconnects by pod and bounded reason.",
|
|
}, []string{"pod", "reason"})
|
|
IMWSHeartbeatTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_ws_heartbeat_total",
|
|
Help: "WebSocket heartbeat events by pod and event.",
|
|
}, []string{"pod", "event"})
|
|
IMWSHeartbeatSeconds = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_ws_heartbeat_seconds",
|
|
Help: "Effective WebSocket heartbeat timing by pod and setting.",
|
|
}, []string{"pod", "setting"})
|
|
IMWSSendQueueCurrent = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_ws_send_queue_current",
|
|
Help: "Current number of queued WebSocket outbound channel entries across sessions on a pod.",
|
|
}, []string{"pod"})
|
|
IMWSSendQueueCapacity = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_ws_send_queue_capacity",
|
|
Help: "Configured outbound channel capacity of one WebSocket session by pod.",
|
|
}, []string{"pod"})
|
|
IMWSSendQueueEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_ws_send_queue_enqueue_total",
|
|
Help: "WebSocket outbound channel enqueue attempts by pod and result.",
|
|
}, []string{"pod", "result"})
|
|
IMWSSendQueueDepth = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_ws_send_queue_depth",
|
|
Help: "Per-session WebSocket outbound channel depth observed after enqueue attempts.",
|
|
Buckets: []float64{0, 1, 4, 8, 16, 32, 64, 96, 128, 144, 160},
|
|
}, []string{"pod", "result"})
|
|
IMWSSendQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_ws_send_queue_wait_duration_seconds",
|
|
Help: "Time a WebSocket outbound channel entry waited before dequeue or purge.",
|
|
Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120},
|
|
}, []string{"pod", "result"})
|
|
IMClusterLivePeers = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
|
|
Name: "im_cluster_live_peers",
|
|
Help: "Current connected cluster peers.",
|
|
ConstLabels: prometheus.Labels{"pod": metricPodName()},
|
|
}, func() float64 {
|
|
return float64(expvarIntValue("LiveClusterNodes"))
|
|
})
|
|
IMClusterExpectedPeers = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
|
|
Name: "im_cluster_expected_peers",
|
|
Help: "Expected cluster peers derived from configured total nodes.",
|
|
ConstLabels: prometheus.Labels{"pod": metricPodName()},
|
|
}, func() float64 {
|
|
total := expvarIntValue("TotalClusterNodes")
|
|
if total > 0 {
|
|
total--
|
|
}
|
|
return float64(total)
|
|
})
|
|
|
|
ClusterP2MQueueLength = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_cluster_p2m_queue_length",
|
|
Help: "Current proxy-to-master queue depth by pod and peer.",
|
|
}, []string{"pod", "peer"})
|
|
ClusterP2MQueueCapacity = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_cluster_p2m_queue_capacity",
|
|
Help: "Configured proxy-to-master queue capacity by pod and peer.",
|
|
}, []string{"pod", "peer"})
|
|
ClusterP2MEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_cluster_p2m_enqueue_total",
|
|
Help: "Proxy-to-master enqueue attempts by pod, peer, and result.",
|
|
}, []string{"pod", "peer", "result"})
|
|
ClusterRPCInflight = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_cluster_rpc_inflight",
|
|
Help: "Current synchronous cluster RPC calls by pod, peer, and method.",
|
|
}, []string{"pod", "peer", "method"})
|
|
ClusterRPCDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_cluster_rpc_duration_seconds",
|
|
Help: "Cluster RPC duration by pod, peer, method, and result.",
|
|
Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5},
|
|
}, []string{"pod", "peer", "method", "result"})
|
|
ClusterRouteTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_cluster_route_total",
|
|
Help: "Cluster route attempts by pod, peer, and result.",
|
|
}, []string{"pod", "peer", "result"})
|
|
ClusterProxyResponseTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_cluster_proxy_response_total",
|
|
Help: "Cluster proxy responses by pod and result.",
|
|
}, []string{"pod", "result"})
|
|
InternalQueueEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_internal_queue_enqueue_total",
|
|
Help: "Internal message queue enqueue attempts by pod, queue, and result.",
|
|
}, []string{"pod", "queue", "result"})
|
|
InternalQueueDepth = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_internal_queue_depth",
|
|
Help: "Internal message queue depth observed after enqueue attempts by pod and queue.",
|
|
Buckets: []float64{0, 1, 4, 16, 32, 64, 128, 192, 256, 512, 1024, 2048, 4096, 8192},
|
|
}, []string{"pod", "queue"})
|
|
InternalQueueCapacity = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_internal_queue_capacity",
|
|
Help: "Configured internal message queue capacity by pod and queue.",
|
|
}, []string{"pod", "queue"})
|
|
InternalQueueCurrent = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_internal_queue_current",
|
|
Help: "Current entries in an internal deferred queue by pod and queue.",
|
|
}, []string{"pod", "queue"})
|
|
MessagePersistDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_message_persist_duration_seconds",
|
|
Help: "Kafka message persistence duration by message kind and result.",
|
|
Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
|
|
}, []string{"kind", "result"})
|
|
MessagePublishStageDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_message_publish_stage_duration_seconds",
|
|
Help: "Message publish hot-path duration by message kind, stage, and result.",
|
|
Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
|
|
}, []string{"kind", "stage", "result"})
|
|
TopicQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_topic_queue_wait_duration_seconds",
|
|
Help: "Time a client publish waited in a topic queue before processing by topic kind.",
|
|
Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
|
|
}, []string{"pod", "kind"})
|
|
PushDispatchQueueEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_push_dispatch_queue_enqueue_total",
|
|
Help: "Asynchronous push dispatch queue enqueue attempts by pod and result.",
|
|
}, []string{"pod", "result"})
|
|
PushDispatchQueueDepth = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_push_dispatch_queue_depth",
|
|
Help: "Current asynchronous push dispatch queue depth by pod.",
|
|
}, []string{"pod"})
|
|
PushDispatchQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_push_dispatch_queue_wait_duration_seconds",
|
|
Help: "Time a push dispatch job waited before a worker started processing it.",
|
|
Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
|
|
}, []string{"pod"})
|
|
PushDispatchDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_push_dispatch_duration_seconds",
|
|
Help: "Asynchronous push presence resolution and dispatch duration by pod and result.",
|
|
Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60},
|
|
}, []string{"pod", "result"})
|
|
PushRecipientCount = prometheus.NewHistogramVec(prometheus.HistogramOpts{
|
|
Name: "im_push_recipient_count",
|
|
Help: "Recipients in an asynchronously dispatched push receipt by pod.",
|
|
Buckets: []float64{1, 2, 10, 50, 100, 300, 600, 900, 1200, 2000},
|
|
}, []string{"pod"})
|
|
KafkaMessagePersistTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_kafka_message_persist_total",
|
|
Help: "Kafka message persistence outcomes by message kind and result.",
|
|
}, []string{"kind", "result"})
|
|
KafkaConsumerLag = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_kafka_consumer_lag",
|
|
Help: "Kafka reader lag reported by kafka-go by pod, topic, and consumer group.",
|
|
}, []string{"pod", "topic", "group"})
|
|
KafkaBacklogOldestAge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
|
|
Name: "im_kafka_backlog_oldest_age_seconds",
|
|
Help: "Age of the latest fetched Kafka message by pod, topic, and consumer group.",
|
|
}, []string{"pod", "topic", "group"})
|
|
KafkaDeadLettersTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
|
|
Name: "im_kafka_dead_letters_total",
|
|
Help: "Kafka messages moved to dead letter by message kind and reason.",
|
|
}, []string{"kind", "reason"})
|
|
)
|
|
|
|
func init() {
|
|
Registry.MustRegister(WSConnectionsActive, IMWSConnectionsActive, IMWSWriteTotal, IMWSWriteDuration,
|
|
IMWSWriteBytesTotal, IMWSDisconnectTotal, IMWSHeartbeatTotal, IMWSHeartbeatSeconds,
|
|
IMWSSendQueueCurrent, IMWSSendQueueCapacity,
|
|
IMWSSendQueueEnqueueTotal, IMWSSendQueueDepth, IMWSSendQueueWaitDuration,
|
|
IMClusterLivePeers, IMClusterExpectedPeers,
|
|
HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration,
|
|
RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration, ClusterP2MQueueLength,
|
|
ClusterP2MQueueCapacity, ClusterP2MEnqueueTotal, ClusterRPCInflight, ClusterRPCDuration,
|
|
ClusterRouteTotal, ClusterProxyResponseTotal, InternalQueueEnqueueTotal, InternalQueueDepth,
|
|
InternalQueueCapacity, InternalQueueCurrent, MessagePersistDuration, MessagePublishStageDuration, TopicQueueWaitDuration,
|
|
PushDispatchQueueEnqueueTotal, PushDispatchQueueDepth, PushDispatchQueueWaitDuration,
|
|
PushDispatchDuration, PushRecipientCount,
|
|
KafkaMessagePersistTotal)
|
|
Registry.MustRegister(KafkaConsumerLag, KafkaBacklogOldestAge, KafkaDeadLettersTotal)
|
|
}
|
|
|
|
func metricPodName() string {
|
|
if value := strings.TrimSpace(os.Getenv("HOSTNAME")); value != "" {
|
|
return value
|
|
}
|
|
return "standalone"
|
|
}
|
|
|
|
func ObserveDB(driverName, operation string, started time.Time, err error) {
|
|
status := "ok"
|
|
if err != nil {
|
|
status = "error"
|
|
}
|
|
DBQueryDuration.WithLabelValues(driverName, operation, status).Observe(time.Since(started).Seconds())
|
|
}
|
|
|
|
func ObserveMessagePublishStage(kind, stage string, started time.Time, err error) {
|
|
status := "ok"
|
|
if err != nil {
|
|
status = "error"
|
|
}
|
|
MessagePublishStageDuration.WithLabelValues(normalizeLabel(kind), normalizeLabel(stage), status).
|
|
Observe(time.Since(started).Seconds())
|
|
}
|
|
|
|
func ObserveTopicQueueWait(kind string, queuedAt time.Time) {
|
|
if queuedAt.IsZero() {
|
|
return
|
|
}
|
|
TopicQueueWaitDuration.WithLabelValues(metricPodName(), normalizeLabel(kind)).Observe(time.Since(queuedAt).Seconds())
|
|
}
|
|
|
|
func ObservePushDispatchEnqueue(depth int, accepted bool) {
|
|
result := "ok"
|
|
if !accepted {
|
|
result = "full"
|
|
}
|
|
PushDispatchQueueEnqueueTotal.WithLabelValues(metricPodName(), result).Inc()
|
|
PushDispatchQueueDepth.WithLabelValues(metricPodName()).Set(float64(depth))
|
|
}
|
|
|
|
func SetPushDispatchQueueDepth(depth int) {
|
|
PushDispatchQueueDepth.WithLabelValues(metricPodName()).Set(float64(depth))
|
|
}
|
|
|
|
func ObservePushDispatch(queuedAt time.Time, recipients int, started time.Time, err error) {
|
|
pod := metricPodName()
|
|
if !queuedAt.IsZero() {
|
|
PushDispatchQueueWaitDuration.WithLabelValues(pod).Observe(started.Sub(queuedAt).Seconds())
|
|
}
|
|
result := "ok"
|
|
if err != nil {
|
|
result = "error"
|
|
}
|
|
PushDispatchDuration.WithLabelValues(pod, result).Observe(time.Since(started).Seconds())
|
|
PushRecipientCount.WithLabelValues(pod).Observe(float64(recipients))
|
|
}
|
|
|
|
func IncRealtimeFanout(stage, kind, targetType, status string) {
|
|
RealtimeFanoutEventsTotal.WithLabelValues(normalizeLabel(stage), normalizeLabel(kind),
|
|
normalizeLabel(targetType), normalizeLabel(status)).Inc()
|
|
}
|
|
|
|
func ObserveRealtimeFanoutDelivery(kind, targetType string, timestamp time.Time) {
|
|
if timestamp.IsZero() {
|
|
return
|
|
}
|
|
RealtimeFanoutDeliveryDuration.WithLabelValues(normalizeLabel(kind), normalizeLabel(targetType)).
|
|
Observe(time.Since(timestamp).Seconds())
|
|
}
|
|
|
|
func ObserveInternalQueueEnqueue(queue string, depth, capacity int, accepted bool) {
|
|
result := "ok"
|
|
if !accepted {
|
|
result = "full"
|
|
}
|
|
pod := metricPodName()
|
|
queue = normalizeLabel(queue)
|
|
InternalQueueEnqueueTotal.WithLabelValues(pod, queue, result).Inc()
|
|
InternalQueueDepth.WithLabelValues(pod, queue).Observe(float64(depth))
|
|
InternalQueueCapacity.WithLabelValues(pod, queue).Set(float64(capacity))
|
|
}
|
|
|
|
func SetInternalQueueCurrent(queue string, depth int) {
|
|
InternalQueueCurrent.WithLabelValues(metricPodName(), normalizeLabel(queue)).Set(float64(depth))
|
|
}
|
|
|
|
func ObserveWSWrite(frame, result, reason string, started time.Time, payloadBytes int) {
|
|
pod := metricPodName()
|
|
frame = normalizeLabel(frame)
|
|
result = normalizeLabel(result)
|
|
reason = normalizeLabel(reason)
|
|
IMWSWriteTotal.WithLabelValues(pod, frame, result, reason).Inc()
|
|
IMWSWriteDuration.WithLabelValues(pod, frame, result).Observe(time.Since(started).Seconds())
|
|
if payloadBytes > 0 {
|
|
IMWSWriteBytesTotal.WithLabelValues(pod, frame).Add(float64(payloadBytes))
|
|
}
|
|
}
|
|
|
|
func IncWSDisconnect(reason string) {
|
|
IMWSDisconnectTotal.WithLabelValues(metricPodName(), normalizeLabel(reason)).Inc()
|
|
}
|
|
|
|
func IncWSHeartbeat(event string) {
|
|
IMWSHeartbeatTotal.WithLabelValues(metricPodName(), normalizeLabel(event)).Inc()
|
|
}
|
|
|
|
func SetWSHeartbeatConfig(pingPeriod, pongWait time.Duration) {
|
|
pod := metricPodName()
|
|
IMWSHeartbeatSeconds.WithLabelValues(pod, "ping_period").Set(pingPeriod.Seconds())
|
|
IMWSHeartbeatSeconds.WithLabelValues(pod, "pong_wait").Set(pongWait.Seconds())
|
|
}
|
|
|
|
func ObserveWSSendQueueEnqueue(depth, capacity int, accepted bool) {
|
|
result := "ok"
|
|
if !accepted {
|
|
result = "full"
|
|
}
|
|
pod := metricPodName()
|
|
IMWSSendQueueEnqueueTotal.WithLabelValues(pod, result).Inc()
|
|
IMWSSendQueueDepth.WithLabelValues(pod, result).Observe(float64(depth))
|
|
IMWSSendQueueCapacity.WithLabelValues(pod).Set(float64(capacity))
|
|
if accepted {
|
|
IMWSSendQueueCurrent.WithLabelValues(pod).Inc()
|
|
}
|
|
}
|
|
|
|
func ObserveWSSendQueueDequeue(queuedAt time.Time, purged bool) {
|
|
result := "dequeued"
|
|
if purged {
|
|
result = "purged"
|
|
}
|
|
pod := metricPodName()
|
|
IMWSSendQueueCurrent.WithLabelValues(pod).Dec()
|
|
if !queuedAt.IsZero() {
|
|
IMWSSendQueueWaitDuration.WithLabelValues(pod, result).Observe(time.Since(queuedAt).Seconds())
|
|
}
|
|
}
|
|
|
|
func normalizeLabel(value string) string {
|
|
if value == "" {
|
|
return "unknown"
|
|
}
|
|
return value
|
|
}
|
|
|
|
func expvarIntValue(name string) int64 {
|
|
value := expvar.Get(name)
|
|
if value == nil {
|
|
return 0
|
|
}
|
|
n, err := strconv.ParseInt(value.String(), 10, 64)
|
|
if err != nil {
|
|
return 0
|
|
}
|
|
return n
|
|
}
|