Add opt-in local Prometheus metrics endpoint to the client

client/internal/localmetrics/localmetrics.go

@@ -0,0 +1,210 @@
+// Package localmetrics exposes client connection state as a local
+// Prometheus /metrics endpoint.
+package localmetrics
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"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
+
+// 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
+	}
+
+	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}
+	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)
+		}
+	}()
+}
+
+// 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
+}

client/internal/localmetrics/localmetrics_test.go

@@ -0,0 +1,97 @@
+package localmetrics
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/netbirdio/netbird/client/internal/peer"
+)
+
+type stubStatus struct {
+	peers      []peer.State
+	management peer.ManagementState
+	signal     peer.SignalState
+}
+
+func (s *stubStatus) GetPeerStates() []peer.State              { return s.peers }
+func (s *stubStatus) GetManagementState() peer.ManagementState { return s.management }
+func (s *stubStatus) GetSignalState() peer.SignalState         { return s.signal }
+
+func testStatus() *stubStatus {
+	return &stubStatus{
+		management: peer.ManagementState{Connected: true},
+		signal:     peer.SignalState{Connected: true},
+		peers: []peer.State{
+			{FQDN: "peer-a.netbird.cloud", IP: "100.90.0.1", ConnStatus: peer.StatusConnected, Relayed: false, Latency: 12 * time.Millisecond},
+			{FQDN: "peer-b.netbird.cloud", IP: "100.90.0.2", ConnStatus: peer.StatusConnected, Relayed: false, Latency: 36 * time.Millisecond},
+			{FQDN: "peer-c.netbird.cloud", IP: "100.90.0.3", ConnStatus: peer.StatusConnected, Relayed: true},
+			{FQDN: "peer-d.netbird.cloud", IP: "100.90.0.4", ConnStatus: peer.StatusIdle},
+		},
+	}
+}
+
+func TestCollector(t *testing.T) {
+	c := newCollector(testStatus())
+
+	expected := `
+# HELP netbird_management_connected Whether the client is connected to the management service (1 connected, 0 disconnected).
+# TYPE netbird_management_connected gauge
+netbird_management_connected 1
+# HELP netbird_peer_latency_seconds Round-trip latency per directly connected peer; relayed connections have no latency measurement.
+# TYPE netbird_peer_latency_seconds gauge
+netbird_peer_latency_seconds{peer="peer-a.netbird.cloud"} 0.012
+netbird_peer_latency_seconds{peer="peer-b.netbird.cloud"} 0.036
+# HELP netbird_peers Number of peers known to this client.
+# TYPE netbird_peers gauge
+netbird_peers 4
+# HELP netbird_peers_connected Number of connected peers by connection type.
+# TYPE netbird_peers_connected gauge
+netbird_peers_connected{connection_type="p2p"} 2
+netbird_peers_connected{connection_type="relay"} 1
+# HELP netbird_signal_connected Whether the client is connected to the signal service (1 connected, 0 disconnected).
+# TYPE netbird_signal_connected gauge
+netbird_signal_connected 1
+`
+	require.NoError(t, testutil.CollectAndCompare(c, strings.NewReader(expected)))
+}
+
+func TestServe(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err, "must find a free port")
+	addr := ln.Addr().String()
+	require.NoError(t, ln.Close())
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	m := NewManager(ctx, testStatus(), nil)
+	m.Reconcile(true, addr)
+
+	var body string
+	require.Eventually(t, func() bool {
+		resp, err := http.Get(fmt.Sprintf("http://%s/metrics", addr))
+		if err != nil {
+			return false
+		}
+		defer resp.Body.Close()
+		data, err := io.ReadAll(resp.Body)
+		if err != nil || resp.StatusCode != http.StatusOK {
+			return false
+		}
+		body = string(data)
+		return true
+	}, 2*time.Second, 50*time.Millisecond, "metrics endpoint should come up")
+
+	assert.Contains(t, body, "netbird_peers 4")
+	assert.Contains(t, body, `netbird_peers_connected{connection_type="relay"} 1`)
+	assert.Contains(t, body, `netbird_peer_latency_seconds{peer="peer-a.netbird.cloud"} 0.012`)
+}

client/internal/metrics/influxdb.go

@@ -45,30 +45,13 @@ func (m *influxDBMetrics) RecordConnectionStages(
 	isReconnection bool,
 	timestamps ConnectionStageTimestamps,
 ) {
-	var signalingReceivedToConnection, connectionToWgHandshake, totalDuration float64
-
-	if !timestamps.SignalingReceived.IsZero() && !timestamps.ConnectionReady.IsZero() {
-		signalingReceivedToConnection = timestamps.ConnectionReady.Sub(timestamps.SignalingReceived).Seconds()
-	}
-
-	if !timestamps.ConnectionReady.IsZero() && !timestamps.WgHandshakeSuccess.IsZero() {
-		connectionToWgHandshake = timestamps.WgHandshakeSuccess.Sub(timestamps.ConnectionReady).Seconds()
-	}
-
-	if !timestamps.SignalingReceived.IsZero() && !timestamps.WgHandshakeSuccess.IsZero() {
-		totalDuration = timestamps.WgHandshakeSuccess.Sub(timestamps.SignalingReceived).Seconds()
-	}
-
-	attemptType := "initial"
-	if isReconnection {
-		attemptType = "reconnection"
-	}
+	signalingReceivedToConnection, connectionToWgHandshake, totalDuration := timestamps.Durations()
 
 	connTypeStr := connectionType.String()
 	tags := fmt.Sprintf("deployment_type=%s,connection_type=%s,attempt_type=%s,version=%s,os=%s,arch=%s,peer_id=%s,connection_pair_id=%s",
 		agentInfo.DeploymentType.String(),
 		connTypeStr,
-		attemptType,
+		attemptType(isReconnection),
 		agentInfo.Version,
 		agentInfo.OS,
 		agentInfo.Arch,
@@ -94,7 +77,7 @@ func (m *influxDBMetrics) RecordConnectionStages(
 	m.trimLocked()
 
 	log.Tracef("peer connection metrics [%s, %s, %s]: signalingReceived→connection: %.3fs, connection→wg_handshake: %.3fs, total: %.3fs",
-		agentInfo.DeploymentType.String(), connTypeStr, attemptType, signalingReceivedToConnection, connectionToWgHandshake, totalDuration)
+		agentInfo.DeploymentType.String(), connTypeStr, attemptType(isReconnection), signalingReceivedToConnection, connectionToWgHandshake, totalDuration)
 }
 
 func (m *influxDBMetrics) RecordSyncDuration(_ context.Context, agentInfo AgentInfo, duration time.Duration) {

client/internal/metrics/metrics.go

@@ -89,6 +89,21 @@ type ConnectionStageTimestamps struct {
 	WgHandshakeSuccess time.Time
 }
 
+// Durations returns the stage durations in seconds. A duration is zero when
+// either of its timestamps is missing.
+func (c ConnectionStageTimestamps) Durations() (signalingToConnection, connectionToWgHandshake, total float64) {
+	if !c.SignalingReceived.IsZero() && !c.ConnectionReady.IsZero() {
+		signalingToConnection = c.ConnectionReady.Sub(c.SignalingReceived).Seconds()
+	}
+	if !c.ConnectionReady.IsZero() && !c.WgHandshakeSuccess.IsZero() {
+		connectionToWgHandshake = c.WgHandshakeSuccess.Sub(c.ConnectionReady).Seconds()
+	}
+	if !c.SignalingReceived.IsZero() && !c.WgHandshakeSuccess.IsZero() {
+		total = c.WgHandshakeSuccess.Sub(c.SignalingReceived).Seconds()
+	}
+	return signalingToConnection, connectionToWgHandshake, total
+}
+
 // String returns a human-readable representation of the connection stage timestamps
 func (c ConnectionStageTimestamps) String() string {
 	return fmt.Sprintf("ConnectionStageTimestamps{SignalingReceived=%v, ConnectionReady=%v, WgHandshakeSuccess=%v}",
@@ -279,3 +294,11 @@ func (c *ClientMetrics) stopPushLocked() {
 	c.wg.Wait()
 	c.push.Store(nil)
 }
+
+// attemptType returns the metric label for an initial vs reconnection attempt.
+func attemptType(isReconnection bool) string {
+	if isReconnection {
+		return "reconnection"
+	}
+	return "initial"
+}

client/internal/metrics/metrics_default.go

@@ -2,10 +2,24 @@
 
 package metrics
 
+import "github.com/prometheus/client_golang/prometheus"
+
 // NewClientMetrics creates a new ClientMetrics instance
 func NewClientMetrics(agentInfo AgentInfo) *ClientMetrics {
 	return &ClientMetrics{
-		impl:      newInfluxDBMetrics(),
+		impl:      newPrometheusMetrics(newInfluxDBMetrics()),
 		agentInfo: agentInfo,
 	}
 }
+
+// PrometheusGatherer returns the registry with the mirrored Prometheus
+// metrics, or nil when unavailable.
+func (c *ClientMetrics) PrometheusGatherer() prometheus.Gatherer {
+	if c == nil {
+		return nil
+	}
+	if pm, ok := c.impl.(*prometheusMetrics); ok {
+		return pm.Gatherer()
+	}
+	return nil
+}

client/internal/metrics/prometheus.go

@@ -0,0 +1,119 @@
+//go:build !js
+
+package metrics
+
+import (
+	"context"
+	"io"
+	"strconv"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// prometheusMetrics mirrors recorded client metrics into a Prometheus
+// registry for the local /metrics endpoint, then delegates to the wrapped
+// implementation. Export and Reset pass through untouched: Prometheus
+// metrics are cumulative and pull-based.
+type prometheusMetrics struct {
+	next     metricsImplementation
+	registry *prometheus.Registry
+
+	connectionStages  *prometheus.HistogramVec
+	syncDuration      prometheus.Histogram
+	syncPhaseDuration *prometheus.HistogramVec
+	loginDuration     *prometheus.HistogramVec
+}
+
+func newPrometheusMetrics(next metricsImplementation) *prometheusMetrics {
+	connectionBuckets := []float64{.05, .1, .25, .5, 1, 2.5, 5, 10, 30, 60}
+
+	m := &prometheusMetrics{
+		next:     next,
+		registry: prometheus.NewRegistry(),
+		connectionStages: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "netbird_peer_connection_stage_duration_seconds",
+			Help:    "Duration of peer connection establishment stages.",
+			Buckets: connectionBuckets,
+		}, []string{"stage", "connection_type", "attempt_type"}),
+		syncDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Name:    "netbird_sync_duration_seconds",
+			Help:    "Duration of management sync message processing.",
+			Buckets: prometheus.DefBuckets,
+		}),
+		syncPhaseDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "netbird_sync_phase_duration_seconds",
+			Help:    "Duration of individual sync processing phases.",
+			Buckets: prometheus.DefBuckets,
+		}, []string{"phase"}),
+		loginDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "netbird_login_duration_seconds",
+			Help:    "Duration of logins to the management service.",
+			Buckets: prometheus.DefBuckets,
+		}, []string{"success"}),
+	}
+
+	m.registry.MustRegister(m.connectionStages, m.syncDuration, m.syncPhaseDuration, m.loginDuration)
+	return m
+}
+
+// Gatherer returns the registry holding the mirrored metrics.
+func (m *prometheusMetrics) Gatherer() prometheus.Gatherer {
+	return m.registry
+}
+
+// RecordConnectionStages implements metricsImplementation.
+func (m *prometheusMetrics) RecordConnectionStages(
+	ctx context.Context,
+	agentInfo AgentInfo,
+	connectionPairID string,
+	connectionType ConnectionType,
+	isReconnection bool,
+	timestamps ConnectionStageTimestamps,
+) {
+	attempt := attemptType(isReconnection)
+	connType := connectionType.String()
+
+	signalingToConnection, connectionToWgHandshake, total := timestamps.Durations()
+	if signalingToConnection > 0 {
+		m.connectionStages.WithLabelValues("signaling_to_connection", connType, attempt).Observe(signalingToConnection)
+	}
+	if connectionToWgHandshake > 0 {
+		m.connectionStages.WithLabelValues("connection_to_wg_handshake", connType, attempt).Observe(connectionToWgHandshake)
+	}
+	if total > 0 {
+		m.connectionStages.WithLabelValues("total", connType, attempt).Observe(total)
+	}
+
+	m.next.RecordConnectionStages(ctx, agentInfo, connectionPairID, connectionType, isReconnection, timestamps)
+}
+
+// RecordSyncDuration implements metricsImplementation.
+func (m *prometheusMetrics) RecordSyncDuration(ctx context.Context, agentInfo AgentInfo, duration time.Duration) {
+	m.syncDuration.Observe(duration.Seconds())
+	m.next.RecordSyncDuration(ctx, agentInfo, duration)
+}
+
+// RecordSyncPhase implements metricsImplementation.
+func (m *prometheusMetrics) RecordSyncPhase(ctx context.Context, agentInfo AgentInfo, phase string, duration time.Duration) {
+	m.syncPhaseDuration.WithLabelValues(phase).Observe(duration.Seconds())
+	m.next.RecordSyncPhase(ctx, agentInfo, phase, duration)
+}
+
+// RecordLoginDuration implements metricsImplementation.
+func (m *prometheusMetrics) RecordLoginDuration(ctx context.Context, agentInfo AgentInfo, duration time.Duration, success bool) {
+	m.loginDuration.WithLabelValues(strconv.FormatBool(success)).Observe(duration.Seconds())
+	m.next.RecordLoginDuration(ctx, agentInfo, duration, success)
+}
+
+// Export implements metricsImplementation by delegating to the wrapped
+// implementation; Prometheus metrics are pulled via the registry instead.
+func (m *prometheusMetrics) Export(w io.Writer) error {
+	return m.next.Export(w)
+}
+
+// Reset implements metricsImplementation by delegating to the wrapped
+// implementation; Prometheus metrics must not be cleared on push.
+func (m *prometheusMetrics) Reset() {
+	m.next.Reset()
+}

client/internal/peer/status.go

@@ -1172,6 +1172,18 @@ func (d *Status) GetResolvedDomainsStates() map[domain.Domain]ResolvedDomainInfo
 	return maps.Clone(d.resolvedDomainsStates)
 }
 
+// GetPeerStates returns a snapshot of all known peer states.
+func (d *Status) GetPeerStates() []State {
+	d.mux.RLock()
+	defer d.mux.RUnlock()
+
+	states := make([]State, 0, len(d.peers))
+	for _, state := range d.peers {
+		states = append(states, state)
+	}
+	return states
+}
+
 // GetFullStatus gets full status
 func (d *Status) GetFullStatus() FullStatus {
 	fullStatus := FullStatus{

client/internal/profilemanager/config.go

@@ -102,6 +102,9 @@ type ConfigInput struct {
 	DNSLabels domain.List
 
 	MTU *uint16
+
+	LocalMetricsEnabled *bool
+	LocalMetricsAddress *string
 }
 
 // Config Configuration type
@@ -142,6 +145,11 @@ type Config struct {
 
 	DNSLabels domain.List
 
+	// LocalMetricsEnabled enables the local Prometheus /metrics endpoint.
+	LocalMetricsEnabled bool
+	// LocalMetricsAddress is the listen address of the local /metrics endpoint.
+	LocalMetricsAddress string
+
 	// SSHKey is a private SSH key in a PEM format
 	SSHKey string
 
@@ -386,6 +394,18 @@ func (config *Config) apply(input ConfigInput) (updated bool, err error) {
 		updated = true
 	}
 
+	if input.LocalMetricsEnabled != nil && *input.LocalMetricsEnabled != config.LocalMetricsEnabled {
+		log.Infof("switching local metrics to %t", *input.LocalMetricsEnabled)
+		config.LocalMetricsEnabled = *input.LocalMetricsEnabled
+		updated = true
+	}
+
+	if input.LocalMetricsAddress != nil && *input.LocalMetricsAddress != config.LocalMetricsAddress {
+		log.Infof("switching local metrics address to %s", *input.LocalMetricsAddress)
+		config.LocalMetricsAddress = *input.LocalMetricsAddress
+		updated = true
+	}
+
 	if input.NetworkMonitor != nil && (config.NetworkMonitor == nil || *input.NetworkMonitor != *config.NetworkMonitor) {
 		log.Infof("switching Network Monitor to %t", *input.NetworkMonitor)
 		config.NetworkMonitor = input.NetworkMonitor