Merge branch 'main' of https://github.com/marcschaeferger/newt

internal/telemetry/metrics.go

@@ -3,6 +3,8 @@ package telemetry
 import (
 	"context"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
@@ -37,9 +39,9 @@ var (
 
 	// Config/Restart
 	mConfigReloads     metric.Int64Counter
-	mRestartCount      metric.Int64Counter
 	mConfigApply       metric.Float64Histogram
 	mCertRotationTotal metric.Int64Counter
+	mProcessStartTime  metric.Float64ObservableGauge
 
 	// Build info
 	mBuildInfo metric.Int64ObservableGauge
@@ -50,6 +52,8 @@ var (
 	mWSDisconnects      metric.Int64Counter
 	mWSKeepaliveFailure metric.Int64Counter
 	mWSSessionDuration  metric.Float64Histogram
+	mWSConnected        metric.Int64ObservableGauge
+	mWSReconnects       metric.Int64Counter
 
 	// Proxy
 	mProxyActiveConns      metric.Int64ObservableGauge
@@ -58,16 +62,28 @@ var (
 	mProxyDropsTotal       metric.Int64Counter
 	mProxyAcceptsTotal     metric.Int64Counter
 	mProxyConnDuration     metric.Float64Histogram
+	mProxyConnectionsTotal metric.Int64Counter
 
-	buildVersion string
-	buildCommit  string
+	buildVersion     string
+	buildCommit      string
+	processStartUnix = float64(time.Now().UnixNano()) / 1e9
+	wsConnectedState atomic.Int64
 )
 
-// attrsWithSite appends global site/region labels when present.
+// Proxy connection lifecycle events.
+const (
+	ProxyConnectionOpened = "opened"
+	ProxyConnectionClosed = "closed"
+)
+
+// attrsWithSite appends site/region labels only when explicitly enabled to keep
+// label cardinality low by default.
 func attrsWithSite(extra ...attribute.KeyValue) []attribute.KeyValue {
-	attrs := make([]attribute.KeyValue, 0, len(extra)+2)
-	attrs = append(attrs, extra...)
-	attrs = append(attrs, siteAttrs()...)
+	attrs := make([]attribute.KeyValue, len(extra))
+	copy(attrs, extra)
+	if ShouldIncludeSiteLabels() {
+		attrs = append(attrs, siteAttrs()...)
+	}
 	return attrs
 }
 
@@ -111,8 +127,9 @@ func registerSiteInstruments() error {
 	if err != nil {
 		return err
 	}
-	mSiteLastHeartbeat, err = meter.Float64ObservableGauge("newt_site_last_heartbeat_seconds",
-		metric.WithDescription("Seconds since last site heartbeat"))
+	mSiteLastHeartbeat, err = meter.Float64ObservableGauge("newt_site_last_heartbeat_timestamp_seconds",
+		metric.WithDescription("Unix timestamp of the last site heartbeat"),
+		metric.WithUnit("s"))
 	if err != nil {
 		return err
 	}
@@ -164,13 +181,22 @@ func registerConnInstruments() error {
 func registerConfigInstruments() error {
 	mConfigReloads, _ = meter.Int64Counter("newt_config_reloads_total",
 		metric.WithDescription("Configuration reloads"))
-	mRestartCount, _ = meter.Int64Counter("newt_restart_count_total",
-		metric.WithDescription("Process restart count (incremented on start)"))
 	mConfigApply, _ = meter.Float64Histogram("newt_config_apply_seconds",
 		metric.WithDescription("Configuration apply duration in seconds"),
 		metric.WithUnit("s"))
 	mCertRotationTotal, _ = meter.Int64Counter("newt_cert_rotation_total",
 		metric.WithDescription("Certificate rotation events (success/failure)"))
+	mProcessStartTime, _ = meter.Float64ObservableGauge("process_start_time_seconds",
+		metric.WithDescription("Unix timestamp of the process start time"),
+		metric.WithUnit("s"))
+	if mProcessStartTime != nil {
+		if _, err := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error {
+			o.ObserveFloat64(mProcessStartTime, processStartUnix)
+			return nil
+		}, mProcessStartTime); err != nil {
+			otel.Handle(err)
+		}
+	}
 	return nil
 }
 
@@ -191,6 +217,10 @@ func registerBuildWSProxyInstruments() error {
 	mWSSessionDuration, _ = meter.Float64Histogram("newt_websocket_session_duration_seconds",
 		metric.WithDescription("Duration of established WebSocket sessions"),
 		metric.WithUnit("s"))
+	mWSConnected, _ = meter.Int64ObservableGauge("newt_websocket_connected",
+		metric.WithDescription("WebSocket connection state (1=connected, 0=disconnected)"))
+	mWSReconnects, _ = meter.Int64Counter("newt_websocket_reconnects_total",
+		metric.WithDescription("WebSocket reconnect attempts by reason"))
 	// Proxy
 	mProxyActiveConns, _ = meter.Int64ObservableGauge("newt_proxy_active_connections",
 		metric.WithDescription("Proxy active connections per tunnel and protocol"))
@@ -207,6 +237,8 @@ func registerBuildWSProxyInstruments() error {
 	mProxyConnDuration, _ = meter.Float64Histogram("newt_proxy_connection_duration_seconds",
 		metric.WithDescription("Duration of completed proxy connections"),
 		metric.WithUnit("s"))
+	mProxyConnectionsTotal, _ = meter.Int64Counter("newt_proxy_connections_total",
+		metric.WithDescription("Proxy connection lifecycle events by protocol"))
 	// Register a default callback for build info if version/commit set
 	reg, e := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error {
 		if buildVersion == "" && buildCommit == "" {
@@ -219,7 +251,9 @@ func registerBuildWSProxyInstruments() error {
 		if buildCommit != "" {
 			attrs = append(attrs, attribute.String("commit", buildCommit))
 		}
-		attrs = append(attrs, siteAttrs()...)
+		if ShouldIncludeSiteLabels() {
+			attrs = append(attrs, siteAttrs()...)
+		}
 		o.ObserveInt64(mBuildInfo, 1, metric.WithAttributes(attrs...))
 		return nil
 	}, mBuildInfo)
@@ -229,6 +263,17 @@ func registerBuildWSProxyInstruments() error {
 		// Provide a functional stopper that unregisters the callback
 		obsStopper = func() { _ = reg.Unregister() }
 	}
+	if mWSConnected != nil {
+		if regConn, err := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error {
+			val := wsConnectedState.Load()
+			o.ObserveInt64(mWSConnected, val, metric.WithAttributes(attrsWithSite()...))
+			return nil
+		}, mWSConnected); err != nil {
+			otel.Handle(err)
+		} else {
+			wsConnStopper = func() { _ = regConn.Unregister() }
+		}
+	}
 	return nil
 }
 
@@ -237,10 +282,11 @@ func registerBuildWSProxyInstruments() error {
 // heartbeat seconds, and active sessions.
 
 var (
-	obsOnce      sync.Once
-	obsStopper   func()
-	proxyObsOnce sync.Once
-	proxyStopper func()
+	obsOnce       sync.Once
+	obsStopper    func()
+	proxyObsOnce  sync.Once
+	proxyStopper  func()
+	wsConnStopper func()
 )
 
 // SetObservableCallback registers a single callback that will be invoked
@@ -251,7 +297,7 @@ var (
 //
 //	telemetry.SetObservableCallback(func(ctx context.Context, o metric.Observer) error {
 //	    o.ObserveInt64(mSiteOnline, 1)
-//	    o.ObserveFloat64(mSiteLastHeartbeat, time.Since(lastHB).Seconds())
+//	    o.ObserveFloat64(mSiteLastHeartbeat, float64(lastHB.Unix()))
 //	    o.ObserveInt64(mTunnelSessions, int64(len(activeSessions)))
 //	    return nil
 //	})
@@ -290,8 +336,6 @@ func SetProxyObservableCallback(cb func(context.Context, metric.Observer) error)
 func RegisterBuildInfo(version, commit string) {
 	buildVersion = version
 	buildCommit = commit
-	// Increment restart count on boot
-	mRestartCount.Add(context.Background(), 1)
 }
 
 // Config reloads
@@ -358,6 +402,25 @@ func IncWSKeepaliveFailure(ctx context.Context, reason string) {
 	)...))
 }
 
+// SetWSConnectionState updates the backing gauge for the WebSocket connected state.
+func SetWSConnectionState(connected bool) {
+	if connected {
+		wsConnectedState.Store(1)
+	} else {
+		wsConnectedState.Store(0)
+	}
+}
+
+// IncWSReconnect increments the WebSocket reconnect counter with a bounded reason label.
+func IncWSReconnect(ctx context.Context, reason string) {
+	if reason == "" {
+		reason = "unknown"
+	}
+	mWSReconnects.Add(ctx, 1, metric.WithAttributes(attrsWithSite(
+		attribute.String("reason", reason),
+	)...))
+}
+
 func ObserveWSSessionDuration(ctx context.Context, seconds float64, result string) {
 	mWSSessionDuration.Record(ctx, seconds, metric.WithAttributes(attrsWithSite(
 		attribute.String("result", result),
@@ -413,6 +476,21 @@ func ObserveProxyConnectionDuration(ctx context.Context, tunnelID, protocol, res
 	mProxyConnDuration.Record(ctx, seconds, metric.WithAttributes(attrsWithSite(attrs...)...))
 }
 
+// IncProxyConnectionEvent records proxy connection lifecycle events (opened/closed).
+func IncProxyConnectionEvent(ctx context.Context, tunnelID, protocol, event string) {
+	if event == "" {
+		event = "unknown"
+	}
+	attrs := []attribute.KeyValue{
+		attribute.String("protocol", protocol),
+		attribute.String("event", event),
+	}
+	if ShouldIncludeTunnelID() && tunnelID != "" {
+		attrs = append(attrs, attribute.String("tunnel_id", tunnelID))
+	}
+	mProxyConnectionsTotal.Add(ctx, 1, metric.WithAttributes(attrsWithSite(attrs...)...))
+}
+
 // --- Config/PKI helpers ---
 
 func ObserveConfigApply(ctx context.Context, phase, result string, seconds float64) {

internal/telemetry/metrics_test_helper.go

@@ -0,0 +1,59 @@
+package telemetry
+
+import (
+	"sync"
+	"time"
+)
+
+func resetMetricsForTest() {
+	initOnce = sync.Once{}
+	obsOnce = sync.Once{}
+	proxyObsOnce = sync.Once{}
+	obsStopper = nil
+	proxyStopper = nil
+	if wsConnStopper != nil {
+		wsConnStopper()
+	}
+	wsConnStopper = nil
+	meter = nil
+
+	mSiteRegistrations = nil
+	mSiteOnline = nil
+	mSiteLastHeartbeat = nil
+
+	mTunnelSessions = nil
+	mTunnelBytes = nil
+	mTunnelLatency = nil
+	mReconnects = nil
+
+	mConnAttempts = nil
+	mConnErrors = nil
+
+	mConfigReloads = nil
+	mConfigApply = nil
+	mCertRotationTotal = nil
+	mProcessStartTime = nil
+
+	mBuildInfo = nil
+
+	mWSConnectLatency = nil
+	mWSMessages = nil
+	mWSDisconnects = nil
+	mWSKeepaliveFailure = nil
+	mWSSessionDuration = nil
+	mWSConnected = nil
+	mWSReconnects = nil
+
+	mProxyActiveConns = nil
+	mProxyBufferBytes = nil
+	mProxyAsyncBacklogByte = nil
+	mProxyDropsTotal = nil
+	mProxyAcceptsTotal = nil
+	mProxyConnDuration = nil
+	mProxyConnectionsTotal = nil
+
+	processStartUnix = float64(time.Now().UnixNano()) / 1e9
+	wsConnectedState.Store(0)
+	includeTunnelIDVal.Store(false)
+	includeSiteLabelVal.Store(false)
+}

internal/telemetry/state_view.go

@@ -62,8 +62,8 @@ func observeSiteOnlineFor(o metric.Observer, sv StateView, siteID string) {
 
 func observeLastHeartbeatFor(o metric.Observer, sv StateView, siteID string) {
 	if t, ok := sv.LastHeartbeat(siteID); ok {
-		secs := time.Since(t).Seconds()
-		o.ObserveFloat64(mSiteLastHeartbeat, secs, metric.WithAttributes(
+		ts := float64(t.UnixNano()) / 1e9
+		o.ObserveFloat64(mSiteLastHeartbeat, ts, metric.WithAttributes(
 			attribute.String("site_id", siteID),
 		))
 	}

internal/telemetry/telemetry.go

@@ -118,6 +118,11 @@ func Init(ctx context.Context, cfg Config) (*Setup, error) {
 	} else {
 		includeTunnelIDVal.Store(false)
 	}
+        if getenv("NEWT_METRICS_INCLUDE_SITE_LABELS", "true") == "true" {
+                includeSiteLabelVal.Store(true)
+        } else {
+                includeSiteLabelVal.Store(false)
+        }
 	res := buildResource(ctx, cfg)
 	UpdateSiteInfo(cfg.SiteID, cfg.Region)
 
@@ -294,7 +299,10 @@ func parseResourceAttributes(s string) map[string]string {
 // Global site/region used to enrich metric labels.
 var siteIDVal atomic.Value
 var regionVal atomic.Value
-var includeTunnelIDVal atomic.Value // bool; default true
+var (
+	includeTunnelIDVal  atomic.Value // bool; default true
+	includeSiteLabelVal atomic.Value // bool; default false
+)
 
 // UpdateSiteInfo updates the global site_id and region used for metric labels.
 // Thread-safe via atomic.Value: subsequent metric emissions will include
@@ -335,7 +343,12 @@ func siteAttrs() []attribute.KeyValue {
 }
 
 // SiteLabelKVs exposes site label KVs for other packages (e.g., proxy manager).
-func SiteLabelKVs() []attribute.KeyValue { return siteAttrs() }
+func SiteLabelKVs() []attribute.KeyValue {
+	if !ShouldIncludeSiteLabels() {
+		return nil
+	}
+	return siteAttrs()
+}
 
 // ShouldIncludeTunnelID returns whether tunnel_id labels should be emitted.
 func ShouldIncludeTunnelID() bool {
@@ -345,6 +358,15 @@ func ShouldIncludeTunnelID() bool {
 	return true
 }
 
+// ShouldIncludeSiteLabels returns whether site_id/region should be emitted as
+// metric labels in addition to resource attributes.
+func ShouldIncludeSiteLabels() bool {
+	if v, ok := includeSiteLabelVal.Load().(bool); ok {
+		return v
+	}
+	return false
+}
+
 func getenv(k, d string) string {
 	if v := os.Getenv(k); v != "" {
 		return v

internal/telemetry/telemetry_attrfilter_test.go

@@ -14,17 +14,23 @@ import (
 
 // Test that disallowed attributes are filtered from the exposition.
 func TestAttributeFilterDropsUnknownKeys(t *testing.T) {
-	ctx := context.Background()
-cfg := Config{ServiceName: "newt", PromEnabled: true, AdminAddr: "127.0.0.1:0"}
+        ctx := context.Background()
+        resetMetricsForTest()
+        t.Setenv("NEWT_METRICS_INCLUDE_SITE_LABELS", "true")
+        cfg := Config{ServiceName: "newt", PromEnabled: true, AdminAddr: "127.0.0.1:0"}
 	tel, err := Init(ctx, cfg)
-	if err != nil { t.Fatalf("init: %v", err) }
+	if err != nil {
+		t.Fatalf("init: %v", err)
+	}
 	defer func() { _ = tel.Shutdown(context.Background()) }()
 
-	if tel.PrometheusHandler == nil { t.Fatalf("prom handler nil") }
+	if tel.PrometheusHandler == nil {
+		t.Fatalf("prom handler nil")
+	}
 	ts := httptest.NewServer(tel.PrometheusHandler)
 	defer ts.Close()
 
-// Add samples with disallowed attribute keys
+	// Add samples with disallowed attribute keys
 	for _, k := range []string{"forbidden", "site_id", "host"} {
 		set := attribute.NewSet(attribute.String(k, "x"))
 		AddTunnelBytesSet(ctx, 123, set)
@@ -32,7 +38,9 @@ cfg := Config{ServiceName: "newt", PromEnabled: true, AdminAddr: "127.0.0.1:0"}
 	time.Sleep(50 * time.Millisecond)
 
 	resp, err := http.Get(ts.URL)
-	if err != nil { t.Fatalf("GET: %v", err) }
+	if err != nil {
+		t.Fatalf("GET: %v", err)
+	}
 	defer resp.Body.Close()
 	b, _ := io.ReadAll(resp.Body)
 	body := string(b)
@@ -43,4 +51,3 @@ cfg := Config{ServiceName: "newt", PromEnabled: true, AdminAddr: "127.0.0.1:0"}
 		t.Fatalf("expected allowed attribute site_id to be present in metrics, got: %s", body)
 	}
 }
-

internal/telemetry/telemetry_golden_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -15,36 +16,61 @@ import (
 // Golden test that /metrics contains expected metric names.
 func TestMetricsGoldenContains(t *testing.T) {
 	ctx := context.Background()
-cfg := Config{ServiceName: "newt", PromEnabled: true, AdminAddr: "127.0.0.1:0", BuildVersion: "test"}
+	resetMetricsForTest()
+	t.Setenv("NEWT_METRICS_INCLUDE_SITE_LABELS", "true")
+	cfg := Config{ServiceName: "newt", PromEnabled: true, AdminAddr: "127.0.0.1:0", BuildVersion: "test"}
 	tel, err := Init(ctx, cfg)
-	if err != nil { t.Fatalf("telemetry init error: %v", err) }
+	if err != nil {
+		t.Fatalf("telemetry init error: %v", err)
+	}
 	defer func() { _ = tel.Shutdown(context.Background()) }()
 
-	if tel.PrometheusHandler == nil { t.Fatalf("prom handler nil") }
+	if tel.PrometheusHandler == nil {
+		t.Fatalf("prom handler nil")
+	}
 	ts := httptest.NewServer(tel.PrometheusHandler)
 	defer ts.Close()
 
-	// Trigger a counter
+	// Trigger counters to ensure they appear in the scrape
 	IncConnAttempt(ctx, "websocket", "success")
+	IncWSReconnect(ctx, "io_error")
+	IncProxyConnectionEvent(ctx, "", "tcp", ProxyConnectionOpened)
+	if tel.MeterProvider != nil {
+		_ = tel.MeterProvider.ForceFlush(ctx)
+	}
 	time.Sleep(100 * time.Millisecond)
 
-	resp, err := http.Get(ts.URL)
-	if err != nil { t.Fatalf("GET metrics failed: %v", err) }
-	defer resp.Body.Close()
-	b, _ := io.ReadAll(resp.Body)
-	body := string(b)
+	var body string
+	for i := 0; i < 5; i++ {
+		resp, err := http.Get(ts.URL)
+		if err != nil {
+			t.Fatalf("GET metrics failed: %v", err)
+		}
+		b, _ := io.ReadAll(resp.Body)
+		_ = resp.Body.Close()
+		body = string(b)
+		if strings.Contains(body, "newt_connection_attempts_total") {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
 
-	f, err := os.Open("internal/telemetry/testdata/expected_contains.golden")
-	if err != nil { t.Fatalf("read golden: %v", err) }
+	f, err := os.Open(filepath.Join("testdata", "expected_contains.golden"))
+	if err != nil {
+		t.Fatalf("read golden: %v", err)
+	}
 	defer f.Close()
 	s := bufio.NewScanner(f)
 	for s.Scan() {
 		needle := strings.TrimSpace(s.Text())
-		if needle == "" { continue }
+		if needle == "" {
+			continue
+		}
 		if !strings.Contains(body, needle) {
 			t.Fatalf("expected metrics body to contain %q. body=\n%s", needle, body)
 		}
 	}
-	if err := s.Err(); err != nil { t.Fatalf("scan golden: %v", err) }
+	if err := s.Err(); err != nil {
+		t.Fatalf("scan golden: %v", err)
+	}
 }
-

internal/telemetry/telemetry_smoke_test.go

@@ -13,13 +13,15 @@ import (
 // Smoke test that /metrics contains at least one newt_* metric when Prom exporter is enabled.
 func TestMetricsSmoke(t *testing.T) {
 	ctx := context.Background()
+	resetMetricsForTest()
+	t.Setenv("NEWT_METRICS_INCLUDE_SITE_LABELS", "true")
 	cfg := Config{
-		ServiceName:    "newt",
-		PromEnabled:    true,
-		OTLPEnabled:    false,
-		AdminAddr:      "127.0.0.1:0",
-		BuildVersion:   "test",
-		BuildCommit:    "deadbeef",
+		ServiceName:          "newt",
+		PromEnabled:          true,
+		OTLPEnabled:          false,
+		AdminAddr:            "127.0.0.1:0",
+		BuildVersion:         "test",
+		BuildCommit:          "deadbeef",
 		MetricExportInterval: 5 * time.Second,
 	}
 	tel, err := Init(ctx, cfg)
@@ -37,18 +39,27 @@ func TestMetricsSmoke(t *testing.T) {
 
 	// Record a simple metric and then fetch /metrics
 	IncConnAttempt(ctx, "websocket", "success")
+	if tel.MeterProvider != nil {
+		_ = tel.MeterProvider.ForceFlush(ctx)
+	}
 	// Give the exporter a tick to collect
 	time.Sleep(100 * time.Millisecond)
 
-	resp, err := http.Get(ts.URL)
-	if err != nil {
-		t.Fatalf("GET /metrics failed: %v", err)
+	var body string
+	for i := 0; i < 5; i++ {
+		resp, err := http.Get(ts.URL)
+		if err != nil {
+			t.Fatalf("GET /metrics failed: %v", err)
+		}
+		b, _ := io.ReadAll(resp.Body)
+		_ = resp.Body.Close()
+		body = string(b)
+		if strings.Contains(body, "newt_connection_attempts_total") {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
 	}
-	defer resp.Body.Close()
-	b, _ := io.ReadAll(resp.Body)
-	body := string(b)
 	if !strings.Contains(body, "newt_connection_attempts_total") {
 		t.Fatalf("expected newt_connection_attempts_total in metrics, got:\n%s", body)
 	}
 }
-

internal/telemetry/testdata/expected_contains.golden

@@ -1,3 +1,7 @@
 newt_connection_attempts_total
+newt_websocket_connected
+newt_websocket_reconnects_total
+newt_proxy_connections_total
 newt_build_info
 
+process_start_time_seconds