broad-announce/internal/observability/metrics.go

package observability

import (
	"github.com/prometheus/client_golang/prometheus"
)

// NewRegistry returns a fresh Prometheus registry. Each service gets
// its own so the metric labels are scoped correctly.
func NewRegistry(serviceName string) (*prometheus.Registry, *IngestdMetrics) {
	reg := prometheus.NewRegistry()
	reg.MustRegister(
		prometheus.NewGoCollector(),
		prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}),
	)
	return reg, NewIngestdMetrics(reg, serviceName)
}

// IngestdMetrics groups the counters/histograms declared in SPEC §22
// for the ingest tier. Other tiers get their own metric groups.
type IngestdMetrics struct {
	AlertsReceived *prometheus.CounterVec   // result=accepted|invalid|rate_limited|payload_too_large|quarantined|circuit_open
	PayloadBytes   prometheus.Histogram
	RateLimitHits  *prometheus.CounterVec   // scope=source|company
	Quarantines    *prometheus.CounterVec
	CBState        *prometheus.GaugeVec
	PublishLatency prometheus.Histogram
	// MQTTMessages is the M4 per-message counter; labels mirror
	// the same result taxonomy as AlertsReceived (accepted,
	// deduped, bad_topic, bad_signature, unknown_source,
	// rate_limited_source, rate_limited_company, invalid,
	// invalid_json, broker_unavailable) plus a "received" label
	// for every message that survived parseIncomingTopic.
	MQTTMessages *prometheus.CounterVec
	// WSMessages is the M5 per-message counter; same result
	// taxonomy as MQTTMessages plus a "received" label for every
	// message that survived the post-auth frame read.
	WSMessages *prometheus.CounterVec
	// WSConnections tracks WS endpoint lifecycle. state ∈
	// {open, closed_clean, closed_protocol_error, closed_unauth,
	// closed_rate_limited, closed_per_ip_cap}.
	WSConnections *prometheus.CounterVec
	// ConnectionRejected tracks per-IP concurrency cap rejections.
	// transport ∈ {http, ws}.
	ConnectionRejected *prometheus.CounterVec
	// TailSubscribers is the current number of /v1/tail/ws
	// clients (gauge, not counter).
	TailSubscribers prometheus.Gauge
	// TailDropped tracks tail events dropped because a
	// subscriber's channel was full. The hub increments this.
	TailDropped *prometheus.CounterVec
}

// NewIngestdMetrics registers and returns the ingestd metrics.
func NewIngestdMetrics(reg prometheus.Registerer, serviceName string) *IngestdMetrics {
	m := &IngestdMetrics{
		AlertsReceived: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "alerts_received_total",
			Help:      "Number of inbound alerts by result.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"result"}),
		PayloadBytes: prometheus.NewHistogram(prometheus.HistogramOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "payload_bytes",
			Help:      "Accepted alert payload size in bytes.",
			Buckets:   prometheus.ExponentialBuckets(64, 4, 8), // 64..1MB
			ConstLabels: prometheus.Labels{"service": serviceName},
		}),
		RateLimitHits: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "rate_limited_total",
			Help:      "Rate-limit rejections by scope.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"scope"}),
		Quarantines: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "source_quarantined_total",
			Help:      "Source quarantines triggered.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"source_id", "company_id"}),
		CBState: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "circuit_breaker_state",
			Help:      "0=closed, 1=half_open, 2=open.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"component"}),
		PublishLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "publish_latency_seconds",
			Help:      "Time to publish an accepted alert to NATS.",
			Buckets:   prometheus.DefBuckets,
			ConstLabels: prometheus.Labels{"service": serviceName},
		}),
		MQTTMessages: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "mqtt_messages_total",
			Help:      "Inbound MQTT messages by result (M4).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"result"}),
		WSMessages: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "ws_messages_total",
			Help:      "Inbound WebSocket messages by result (M5).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"result"}),
		WSConnections: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "ws_connections_total",
			Help:      "WebSocket connection lifecycle events (M5).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"state"}),
		ConnectionRejected: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "connection_rejected_total",
			Help:      "Per-IP concurrency cap rejections (M5, SPEC §22 layer 2).",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"transport"}),
		TailSubscribers: prometheus.NewGauge(prometheus.GaugeOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "tail_subscribers",
			Help:      "Current number of /v1/tail/ws clients.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}),
		TailDropped: prometheus.NewCounterVec(prometheus.CounterOpts{
			Namespace: "ba",
			Subsystem: "ingestd",
			Name:      "tail_dropped_total",
			Help:      "Tail events dropped because a subscriber was too slow.",
			ConstLabels: prometheus.Labels{"service": serviceName},
		}, []string{"reason"}),
	}
	reg.MustRegister(
		m.AlertsReceived,
		m.PayloadBytes,
		m.RateLimitHits,
		m.Quarantines,
		m.CBState,
		m.PublishLatency,
		m.MQTTMessages,
		m.WSMessages,
		m.WSConnections,
		m.ConnectionRejected,
		m.TailSubscribers,
		m.TailDropped,
	)
	m.AlertsReceived.WithLabelValues("accepted")
	return m
}