// Package localmetrics exposes client connection state as a local // Prometheus /metrics endpoint. package localmetrics import ( "context" "errors" "net" "net/http" "net/netip" "sync" "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" dto "github.com/prometheus/client_model/go" log "github.com/sirupsen/logrus" "github.com/netbirdio/netbird/client/internal/peer" ) // DefaultListenAddress is used when local metrics are enabled without an explicit address. const DefaultListenAddress = "127.0.0.1:9191" const ( shutdownTimeout = 3 * time.Second readHeaderTimeout = 5 * time.Second readTimeout = 10 * time.Second writeTimeout = 30 * time.Second idleTimeout = time.Minute ) // statusSource provides the connection state snapshots the collector reads on scrape. type statusSource interface { GetPeerStates() []peer.State GetManagementState() peer.ManagementState GetSignalState() peer.SignalState } // GathererProvider returns the current client metrics gatherer, or nil when // no engine is running. It is called on every scrape. type GathererProvider func() prometheus.Gatherer // Manager runs the local /metrics HTTP endpoint according to the active // client configuration. Reconcile is safe to call on every config change. type Manager struct { status statusSource clientMetrics GathererProvider mu sync.Mutex srv *http.Server addr string } // NewManager creates a manager that serves metrics from status and // clientMetrics and shuts down when ctx is canceled. func NewManager(ctx context.Context, status statusSource, clientMetrics GathererProvider) *Manager { m := &Manager{status: status, clientMetrics: clientMetrics} go func() { <-ctx.Done() m.Stop() }() return m } // Reconcile starts, stops, or restarts the metrics endpoint to match the // desired state. An empty addr falls back to DefaultListenAddress. func (m *Manager) Reconcile(enabled bool, addr string) { if addr == "" { addr = DefaultListenAddress } warnIfNotLoopback(addr) m.mu.Lock() defer m.mu.Unlock() if !enabled { m.stop() return } if m.srv != nil && m.addr == addr { return } m.stop() registry := prometheus.NewRegistry() registry.MustRegister(newCollector(m.status)) gatherers := prometheus.Gatherers{registry, prometheus.GathererFunc(func() ([]*dto.MetricFamily, error) { if m.clientMetrics == nil { return nil, nil } g := m.clientMetrics() if g == nil { return nil, nil } return g.Gather() })} mux := http.NewServeMux() mux.Handle("/metrics", promhttp.HandlerFor(gatherers, promhttp.HandlerOpts{})) srv := &http.Server{ Addr: addr, Handler: mux, ReadHeaderTimeout: readHeaderTimeout, ReadTimeout: readTimeout, WriteTimeout: writeTimeout, IdleTimeout: idleTimeout, } m.srv = srv m.addr = addr log.Infof("serving local metrics on http://%s/metrics", addr) go func() { if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { log.Errorf("failed to serve local metrics on %s: %v", addr, err) m.clear(srv) } }() } // clear drops the reference to srv so a later Reconcile with the same // address restarts it. A newer server may already have replaced it, in // which case the reference must stay. func (m *Manager) clear(srv *http.Server) { m.mu.Lock() defer m.mu.Unlock() if m.srv != srv { return } m.srv = nil m.addr = "" } // Stop shuts down the metrics endpoint if it is running. func (m *Manager) Stop() { m.mu.Lock() defer m.mu.Unlock() m.stop() } // stop shuts down the running server. Callers must hold m.mu. func (m *Manager) stop() { if m.srv == nil { return } ctx, cancel := context.WithTimeout(context.Background(), shutdownTimeout) defer cancel() if err := m.srv.Shutdown(ctx); err != nil { log.Debugf("failed to shut down local metrics server: %v", err) } m.srv = nil m.addr = "" } // collector converts status recorder snapshots into Prometheus metrics at scrape time. type collector struct { status statusSource managementConnected *prometheus.Desc signalConnected *prometheus.Desc peersTotal *prometheus.Desc peersConnected *prometheus.Desc peerLatency *prometheus.Desc } func newCollector(status statusSource) *collector { return &collector{ status: status, managementConnected: prometheus.NewDesc( "netbird_management_connected", "Whether the client is connected to the management service (1 connected, 0 disconnected).", nil, nil, ), signalConnected: prometheus.NewDesc( "netbird_signal_connected", "Whether the client is connected to the signal service (1 connected, 0 disconnected).", nil, nil, ), peersTotal: prometheus.NewDesc( "netbird_peers", "Number of peers known to this client.", nil, nil, ), peersConnected: prometheus.NewDesc( "netbird_peers_connected", "Number of connected peers by connection type.", []string{"connection_type"}, nil, ), peerLatency: prometheus.NewDesc( "netbird_peer_latency_seconds", "Round-trip latency per directly connected peer; relayed connections have no latency measurement.", []string{"peer"}, nil, ), } } // Describe implements prometheus.Collector. func (c *collector) Describe(ch chan<- *prometheus.Desc) { ch <- c.managementConnected ch <- c.signalConnected ch <- c.peersTotal ch <- c.peersConnected ch <- c.peerLatency } // Collect implements prometheus.Collector. func (c *collector) Collect(ch chan<- prometheus.Metric) { ch <- prometheus.MustNewConstMetric(c.managementConnected, prometheus.GaugeValue, boolToFloat(c.status.GetManagementState().Connected)) ch <- prometheus.MustNewConstMetric(c.signalConnected, prometheus.GaugeValue, boolToFloat(c.status.GetSignalState().Connected)) peers := c.status.GetPeerStates() ch <- prometheus.MustNewConstMetric(c.peersTotal, prometheus.GaugeValue, float64(len(peers))) var p2p, relayed float64 for _, p := range peers { if p.ConnStatus != peer.StatusConnected { continue } if p.Relayed { relayed++ continue } p2p++ if latency := p.Latency.Seconds(); latency > 0 { ch <- prometheus.MustNewConstMetric(c.peerLatency, prometheus.GaugeValue, latency, p.FQDN) } } ch <- prometheus.MustNewConstMetric(c.peersConnected, prometheus.GaugeValue, p2p, "p2p") ch <- prometheus.MustNewConstMetric(c.peersConnected, prometheus.GaugeValue, relayed, "relay") } func boolToFloat(b bool) float64 { if b { return 1 } return 0 } // IsLoopback reports whether addr binds the endpoint to the local host only. // An empty address means DefaultListenAddress. It fails closed: an address // that cannot be confirmed loopback, including an unparseable one, is not. func IsLoopback(addr string) bool { if addr == "" { addr = DefaultListenAddress } host, _, err := net.SplitHostPort(addr) if err != nil { return false } if host == "localhost" { return true } ip, err := netip.ParseAddr(host) if err != nil { return false } return ip.Unmap().IsLoopback() } // warnIfNotLoopback logs a warning when the listen address cannot be // confirmed to be local-only, since the endpoint exposes peer and // connectivity details without authentication. func warnIfNotLoopback(addr string) { if IsLoopback(addr) { return } log.Warnf("local metrics endpoint listens on non-loopback address %s and is reachable from the network without authentication", addr) }