EvoBGP/internal/observability/metrics.go

// Package observability registers Prometheus metrics for the control plane (prefix aggregates, BGP peers, jobs, HTTP).
package observability

import (
	"context"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"github.com/prometheus/client_golang/prometheus/promhttp"

	"evobgp/internal/store"
)

const namespace = "evobgp"

var (
	metricsStore          atomic.Value // store.Backend
	registerCollectorOnce sync.Once
)

var (
	httpRequests = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: namespace,
			Name:      "http_requests_total",
			Help:      "HTTP requests handled by the API mux (excludes /metrics).",
		},
		[]string{"method", "code", "path_group"},
	)

	jobsFinished = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: namespace,
			Name:      "jobs_finished_total",
			Help:      "Async jobs that reached a terminal state.",
		},
		[]string{"kind", "status"},
	)

	birdBGPEstablished = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace,
		Name:      "bird_bgp_sessions_established",
		Help:      "BGP sessions in Established state from birdc show protocols (0 if scrape disabled or failed).",
	})

	birdProtocolsScrapeSuccess = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace,
		Name:      "bird_protocols_scrape_success",
		Help:      "1 if the last birdc protocols scrape succeeded, else 0.",
	})

	buildInfo = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: namespace,
		Name:      "build_info",
		Help:      "Build metadata (value always 1).",
	}, []string{"version", "git_sha"})

	prefixAggregationDuration = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: namespace,
		Name:      "prefix_aggregation_duration_seconds",
		Help:      "Time spent in smartAggregatePrefixRows during tenant render.",
		Buckets:   prometheus.ExponentialBuckets(0.0001, 2, 16),
	})

	prefixAggregationRawCount = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: namespace,
		Name:      "prefix_aggregation_raw_count",
		Help:      "Prefix row count before CIDR aggregation on tenant render.",
		Buckets:   prometheus.ExponentialBuckets(1, 2, 16),
	})

	prefixAggregationAggregatedCount = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: namespace,
		Name:      "prefix_aggregation_aggregated_count",
		Help:      "Prefix row count after CIDR aggregation on tenant render.",
		Buckets:   prometheus.ExponentialBuckets(1, 2, 16),
	})

	pipelineRefreshDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
		Namespace: namespace,
		Name:      "pipeline_refresh_duration_seconds",
		Help:      "Module refresh ingest duration by module type.",
		Buckets:   prometheus.ExponentialBuckets(0.05, 2, 14),
	}, []string{"module_type"})

	renderPrefixCount = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: namespace,
		Name:      "render_prefix_count",
		Help:      "Materialized prefix count per tenant render.",
		Buckets:   prometheus.ExponentialBuckets(10, 2, 16),
	})

	jobQueueActive = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace,
		Name:      "job_queue_active",
		Help:      "Currently running in-process async jobs.",
	})

	jobQueueCapacity = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: namespace,
		Name:      "job_queue_capacity",
		Help:      "Maximum concurrent in-process async jobs.",
	})

	jobAuditDepth = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: namespace,
		Name:      "job_audit_depth",
		Help:      "Rows in job_audit by status (durable queue).",
	}, []string{"status"})

	upstreamBreakerOpen = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: namespace,
		Name:      "upstream_breaker_open",
		Help:      "1 if per-host HTTP circuit breaker is open.",
	}, []string{"host"})
)

var (
	birdProtocolStatesMu sync.RWMutex
	birdProtocolStates   map[string]string
	birdProtocolStatesAt time.Time
)

// RecordPrefixAggregation records tenant render CIDR aggregation stats.
func RecordPrefixAggregation(rawCount, aggregatedCount int, duration time.Duration) {
	if rawCount < 0 {
		rawCount = 0
	}
	if aggregatedCount < 0 {
		aggregatedCount = 0
	}
	prefixAggregationDuration.Observe(duration.Seconds())
	prefixAggregationRawCount.Observe(float64(rawCount))
	prefixAggregationAggregatedCount.Observe(float64(aggregatedCount))
	renderPrefixCount.Observe(float64(aggregatedCount))
}

// RecordPipelineRefresh records module ingest duration.
func RecordPipelineRefresh(moduleType string, duration time.Duration) {
	if moduleType == "" {
		moduleType = "unknown"
	}
	pipelineRefreshDuration.WithLabelValues(moduleType).Observe(duration.Seconds())
}

// RecordJobQueueDepth updates in-process job worker utilization gauges.
func RecordJobQueueDepth(active, capacity int) {
	if active < 0 {
		active = 0
	}
	if capacity < 0 {
		capacity = 0
	}
	jobQueueActive.Set(float64(active))
	jobQueueCapacity.Set(float64(capacity))
}

// RecordJobTerminal increments jobs_finished_total for terminal statuses.
func RecordJobTerminal(kind, status string) {
	switch status {
	case "succeeded", "failed", "cancelled":
		jobsFinished.WithLabelValues(kind, status).Inc()
	default:
		return
	}
}

// RecordJobAuditDepth updates durable job_audit depth gauges.
func RecordJobAuditDepth(counts map[string]int64) {
	for status, n := range counts {
		if status == "" {
			status = "unknown"
		}
		jobAuditDepth.WithLabelValues(status).Set(float64(n))
	}
}

// SetUpstreamBreakerOpen records circuit breaker state for a host (1=open, 0=closed).
func SetUpstreamBreakerOpen(host string, open bool) {
	if host == "" {
		host = "_"
	}
	v := 0.0
	if open {
		v = 1
	}
	upstreamBreakerOpen.WithLabelValues(host).Set(v)
}

// SetBuildInfo sets evobgp_build_info gauge (idempotent labels).
func SetBuildInfo(version, gitSHA string) {
	if version == "" {
		version = "unknown"
	}
	if gitSHA == "" {
		gitSHA = "unknown"
	}
	buildInfo.WithLabelValues(version, gitSHA).Set(1)
}

type memoryStoreCollector struct {
	prefixMaxDesc *prometheus.Desc
	prefixSumDesc *prometheus.Desc
	peersDesc     *prometheus.Desc
	peerStateDesc *prometheus.Desc
}

func newMemoryStoreCollector() *memoryStoreCollector {
	return &memoryStoreCollector{
		prefixMaxDesc: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, "", "materialized_prefixes_max"),
			"Maximum materialized_prefix_count among all revisions in the store.",
			nil, nil,
		),
		prefixSumDesc: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, "", "materialized_prefixes_sum"),
			"Sum of materialized_prefix_count over revisions (development aggregate).",
			nil, nil,
		),
		peersDesc: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, "", "bgp_peers_configured_total"),
			"BGP peers configured in the control-plane store.",
			nil, nil,
		),
		peerStateDesc: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, "", "bgp_peer_sessions"),
			"Configured BGP peers in the store by session_state (intent / last known, not live BIRD).",
			[]string{"state"}, nil,
		),
	}
}

func (c *memoryStoreCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- c.prefixMaxDesc
	ch <- c.prefixSumDesc
	ch <- c.peersDesc
	ch <- c.peerStateDesc
}

func (c *memoryStoreCollector) Collect(ch chan<- prometheus.Metric) {
	v := metricsStore.Load()
	if v == nil {
		return
	}
	b, ok := v.(store.Backend)
	if !ok || b == nil {
		return
	}
	maxN, sumN := b.MaterializedPrefixStats()
	ch <- prometheus.MustNewConstMetric(c.prefixMaxDesc, prometheus.GaugeValue, float64(maxN))
	ch <- prometheus.MustNewConstMetric(c.prefixSumDesc, prometheus.GaugeValue, float64(sumN))
	ch <- prometheus.MustNewConstMetric(c.peersDesc, prometheus.GaugeValue, float64(b.PeerCount()))
	for state, n := range b.PeerSessionCountsByState() {
		if state == "" {
			state = "unknown"
		}
		ch <- prometheus.MustNewConstMetric(c.peerStateDesc, prometheus.GaugeValue, float64(n), state)
	}
}

// RegisterStoreBackend points Prometheus collectors at any store.Backend (last call wins; safe for tests).
func RegisterStoreBackend(b store.Backend) {
	if b == nil {
		return
	}
	metricsStore.Store(b)
	registerCollectorOnce.Do(func() {
		prometheus.DefaultRegisterer.MustRegister(newMemoryStoreCollector())
	})
}

// RegisterStoreMetrics is a deprecated alias for RegisterStoreBackend (memory-only callers).
func RegisterStoreMetrics(mem *store.Memory) {
	if mem == nil {
		return
	}
	RegisterStoreBackend(mem)
}

// SetBirdSessionMetrics updates gauges from an optional birdc scrape.
func SetBirdSessionMetrics(established int, scrapeOK bool) {
	birdBGPEstablished.Set(float64(established))
	if scrapeOK {
		birdProtocolsScrapeSuccess.Set(1)
	} else {
		birdProtocolsScrapeSuccess.Set(0)
	}
}

// SetBirdProtocolStates caches parsed BGP protocol states from the last birdc scrape.
func SetBirdProtocolStates(states map[string]string) {
	birdProtocolStatesMu.Lock()
	defer birdProtocolStatesMu.Unlock()
	if states == nil {
		birdProtocolStates = map[string]string{}
	} else {
		birdProtocolStates = states
	}
	birdProtocolStatesAt = time.Now()
}

// CachedBirdProtocolStates returns cached protocol states if younger than maxAge.
func CachedBirdProtocolStates(maxAge time.Duration) (map[string]string, bool) {
	if maxAge <= 0 {
		maxAge = 60 * time.Second
	}
	birdProtocolStatesMu.RLock()
	defer birdProtocolStatesMu.RUnlock()
	if birdProtocolStates == nil || time.Since(birdProtocolStatesAt) > maxAge {
		return nil, false
	}
	out := make(map[string]string, len(birdProtocolStates))
	for k, v := range birdProtocolStates {
		out[k] = v
	}
	return out, true
}

// MetricsHandler returns the Prometheus scrape handler.
func MetricsHandler() http.Handler {
	return promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{})
}

// HTTPMiddleware records method, status code, and path group for all wrapped requests.
func HTTPMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		sw := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
		next.ServeHTTP(sw, r)
		httpRequests.WithLabelValues(r.Method, strconv.Itoa(sw.status), pathGroup(r.URL.Path)).Inc()
	})
}

func pathGroup(path string) string {
	if path == "" {
		return "/"
	}
	parts := strings.Split(strings.Trim(path, "/"), "/")
	if len(parts) == 0 || parts[0] == "" {
		return "/"
	}
	if parts[0] == "v1" && len(parts) >= 2 {
		return "/v1/" + parts[1]
	}
	if parts[0] == "metrics" || parts[0] == "version" {
		return "/" + parts[0]
	}
	return "/" + parts[0]
}

type statusRecorder struct {
	http.ResponseWriter
	status int
}

func (s *statusRecorder) WriteHeader(code int) {
	s.status = code
	s.ResponseWriter.WriteHeader(code)
}

// StartBirdProtocolsPoller runs birdc "show protocols" on interval when socket is non-empty.
// Горутина завершается при отмене ctx (корректное завершение вместе с процессом API).
func StartBirdProtocolsPoller(ctx context.Context, socket string, birdcPath string, interval time.Duration, showFn func(ctx context.Context, socket, birdcBin string) (string, error), countFn func(output string) int, parseFn func(output string) map[string]string) {
	socket = trimSpace(socket)
	if ctx == nil || socket == "" || interval <= 0 || showFn == nil || countFn == nil {
		return
	}
	scrape := func() {
		sctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		out, err := showFn(sctx, socket, birdcPath)
		cancel()
		if err != nil {
			SetBirdSessionMetrics(0, false)
			return
		}
		SetBirdSessionMetrics(countFn(out), true)
		if parseFn != nil {
			SetBirdProtocolStates(parseFn(out))
		}
	}
	go func() {
		scrape()
		t := time.NewTicker(interval)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				scrape()
			}
		}
	}()
}

func trimSpace(s string) string {
	for len(s) > 0 && (s[0] == ' ' || s[0] == '\t') {
		s = s[1:]
	}
	for len(s) > 0 {
		last := s[len(s)-1]
		if last != ' ' && last != '\t' {
			break
		}
		s = s[:len(s)-1]
	}
	return s
}