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
}

// 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"}),
	}
	reg.MustRegister(
		m.AlertsReceived,
		m.PayloadBytes,
		m.RateLimitHits,
		m.Quarantines,
		m.CBState,
		m.PublishLatency,
		m.MQTTMessages,
	)
	m.AlertsReceived.WithLabelValues("accepted")
	return m
}