[tunnel] add agent metrics, overlay scraper, and re-export collector

Add Prometheus metrics to the tunnel agent (info, uptime, per-protocol
packet/byte counters, BFD heartbeats, DNS query/cache stats, mirror
reconciler counters) and a scraper + re-export system for tunnelproxy.

Each connected agent gets a dedicated scrape goroutine that polls its
/metrics endpoint through the overlay network. Results flow through a
buffered channel to a single aggregator goroutine, avoiding concurrent
map writes. The ReexportCollector serves the aggregated metrics on the
tunnelproxy's /metrics endpoint with tunnel_node/agent/project_id
labels injected.

The agent advertises its metrics port to the server via a connection
label (CONNECT-IP) or ConnectRequest field (relay API), so the scraper
knows which port to hit regardless of the configured --metrics-addr.

Per-protocol counters on the packet hot path use pre-resolved
prometheus.Counter objects to avoid WithLabelValues map lookups on
every packet.

Author: dilyevsky

pkg/cmd/tunnel/run.go

@@ -6,10 +6,12 @@ import (
 	"io"
 	"log/slog"
 	"math/rand"
-	"slices"
+	"net"
 	"net/http"
 	"net/netip"
 	"os"
+	"slices"
+	"strconv"
 	"sync"
 	"time"
 
@@ -476,6 +478,13 @@ func (t *tunnelNodeReconciler) reconcile(ctx context.Context, req ctrl.Request)
 	if t.cfg.IsLocalMode || insecureSkipVerify {
 		cOpts = append(cOpts, tunnel.WithInsecureSkipVerify(true))
 	}
+	// Advertise the metrics port so the server can scrape agent metrics
+	// through the overlay network.
+	if _, portStr, err := net.SplitHostPort(metricsAddr); err == nil {
+		if p, err := strconv.Atoi(portStr); err == nil && p > 0 {
+			cOpts = append(cOpts, tunnel.WithClientMetricsPort(p))
+		}
+	}
 
 	t.dialMu.Lock()
 	t.tunnelUID = tnUUID

pkg/kube-controller/controllers/mirror.go

@@ -286,18 +286,23 @@ func (r *MirrorReconciler) syncGateway(ctx context.Context, gw *gwapiv1.Gateway)
 	if apierrors.IsNotFound(err) {
 		log.Infof("Mirror: creating Gateway %s (from %s/%s)", apoxyName, gw.Namespace, gw.Name)
 		if _, err := r.apoxyClient.GatewayV1().Gateways().Create(ctx, apoxy, metav1.CreateOptions{}); err != nil {
+			MirrorSyncErrors.WithLabelValues("Gateway").Inc()
 			return reconcile.Result{}, fmt.Errorf("creating Apoxy Gateway %s: %w", apoxyName, err)
 		}
+		MirrorSyncedResources.WithLabelValues("Gateway").Inc()
 		return reconcile.Result{}, nil
 	} else if err != nil {
+		MirrorSyncErrors.WithLabelValues("Gateway").Inc()
 		return reconcile.Result{}, fmt.Errorf("getting Apoxy Gateway %s: %w", apoxyName, err)
 	}
 
 	apoxy.ResourceVersion = existing.ResourceVersion
 	log.Infof("Mirror: updating Gateway %s (from %s/%s)", apoxyName, gw.Namespace, gw.Name)
 	if _, err := r.apoxyClient.GatewayV1().Gateways().Update(ctx, apoxy, metav1.UpdateOptions{}); err != nil {
+		MirrorSyncErrors.WithLabelValues("Gateway").Inc()
 		return reconcile.Result{}, fmt.Errorf("updating Apoxy Gateway %s: %w", apoxyName, err)
 	}
+	MirrorSyncedResources.WithLabelValues("Gateway").Inc()
 	return reconcile.Result{}, nil
 }
 
@@ -348,18 +353,23 @@ func (r *MirrorReconciler) syncHTTPRoute(ctx context.Context, route *gwapiv1.HTT
 	if apierrors.IsNotFound(err) {
 		log.Infof("Mirror: creating HTTPRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 		if _, err := r.apoxyClient.GatewayV1().HTTPRoutes().Create(ctx, apoxy, metav1.CreateOptions{}); err != nil {
+			MirrorSyncErrors.WithLabelValues("HTTPRoute").Inc()
 			return reconcile.Result{}, fmt.Errorf("creating Apoxy HTTPRoute %s: %w", apoxyName, err)
 		}
+		MirrorSyncedResources.WithLabelValues("HTTPRoute").Inc()
 		return reconcile.Result{}, nil
 	} else if err != nil {
+		MirrorSyncErrors.WithLabelValues("HTTPRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("getting Apoxy HTTPRoute %s: %w", apoxyName, err)
 	}
 
 	apoxy.ResourceVersion = existing.ResourceVersion
 	log.Infof("Mirror: updating HTTPRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 	if _, err := r.apoxyClient.GatewayV1().HTTPRoutes().Update(ctx, apoxy, metav1.UpdateOptions{}); err != nil {
+		MirrorSyncErrors.WithLabelValues("HTTPRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("updating Apoxy HTTPRoute %s: %w", apoxyName, err)
 	}
+	MirrorSyncedResources.WithLabelValues("HTTPRoute").Inc()
 	return reconcile.Result{}, nil
 }
 
@@ -410,18 +420,23 @@ func (r *MirrorReconciler) syncGRPCRoute(ctx context.Context, route *gwapiv1.GRP
 	if apierrors.IsNotFound(err) {
 		log.Infof("Mirror: creating GRPCRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 		if _, err := r.apoxyClient.GatewayV1().GRPCRoutes().Create(ctx, apoxy, metav1.CreateOptions{}); err != nil {
+			MirrorSyncErrors.WithLabelValues("GRPCRoute").Inc()
 			return reconcile.Result{}, fmt.Errorf("creating Apoxy GRPCRoute %s: %w", apoxyName, err)
 		}
+		MirrorSyncedResources.WithLabelValues("GRPCRoute").Inc()
 		return reconcile.Result{}, nil
 	} else if err != nil {
+		MirrorSyncErrors.WithLabelValues("GRPCRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("getting Apoxy GRPCRoute %s: %w", apoxyName, err)
 	}
 
 	apoxy.ResourceVersion = existing.ResourceVersion
 	log.Infof("Mirror: updating GRPCRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 	if _, err := r.apoxyClient.GatewayV1().GRPCRoutes().Update(ctx, apoxy, metav1.UpdateOptions{}); err != nil {
+		MirrorSyncErrors.WithLabelValues("GRPCRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("updating Apoxy GRPCRoute %s: %w", apoxyName, err)
 	}
+	MirrorSyncedResources.WithLabelValues("GRPCRoute").Inc()
 	return reconcile.Result{}, nil
 }
 
@@ -472,18 +487,23 @@ func (r *MirrorReconciler) syncTCPRoute(ctx context.Context, route *gwapiv1alpha
 	if apierrors.IsNotFound(err) {
 		log.Infof("Mirror: creating TCPRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 		if _, err := r.apoxyClient.GatewayV1alpha2().TCPRoutes().Create(ctx, apoxy, metav1.CreateOptions{}); err != nil {
+			MirrorSyncErrors.WithLabelValues("TCPRoute").Inc()
 			return reconcile.Result{}, fmt.Errorf("creating Apoxy TCPRoute %s: %w", apoxyName, err)
 		}
+		MirrorSyncedResources.WithLabelValues("TCPRoute").Inc()
 		return reconcile.Result{}, nil
 	} else if err != nil {
+		MirrorSyncErrors.WithLabelValues("TCPRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("getting Apoxy TCPRoute %s: %w", apoxyName, err)
 	}
 
 	apoxy.ResourceVersion = existing.ResourceVersion
 	log.Infof("Mirror: updating TCPRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 	if _, err := r.apoxyClient.GatewayV1alpha2().TCPRoutes().Update(ctx, apoxy, metav1.UpdateOptions{}); err != nil {
+		MirrorSyncErrors.WithLabelValues("TCPRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("updating Apoxy TCPRoute %s: %w", apoxyName, err)
 	}
+	MirrorSyncedResources.WithLabelValues("TCPRoute").Inc()
 	return reconcile.Result{}, nil
 }
 
@@ -534,18 +554,23 @@ func (r *MirrorReconciler) syncTLSRoute(ctx context.Context, route *gwapiv1alpha
 	if apierrors.IsNotFound(err) {
 		log.Infof("Mirror: creating TLSRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 		if _, err := r.apoxyClient.GatewayV1alpha2().TLSRoutes().Create(ctx, apoxy, metav1.CreateOptions{}); err != nil {
+			MirrorSyncErrors.WithLabelValues("TLSRoute").Inc()
 			return reconcile.Result{}, fmt.Errorf("creating Apoxy TLSRoute %s: %w", apoxyName, err)
 		}
+		MirrorSyncedResources.WithLabelValues("TLSRoute").Inc()
 		return reconcile.Result{}, nil
 	} else if err != nil {
+		MirrorSyncErrors.WithLabelValues("TLSRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("getting Apoxy TLSRoute %s: %w", apoxyName, err)
 	}
 
 	apoxy.ResourceVersion = existing.ResourceVersion
 	log.Infof("Mirror: updating TLSRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 	if _, err := r.apoxyClient.GatewayV1alpha2().TLSRoutes().Update(ctx, apoxy, metav1.UpdateOptions{}); err != nil {
+		MirrorSyncErrors.WithLabelValues("TLSRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("updating Apoxy TLSRoute %s: %w", apoxyName, err)
 	}
+	MirrorSyncedResources.WithLabelValues("TLSRoute").Inc()
 	return reconcile.Result{}, nil
 }
 
@@ -596,18 +621,23 @@ func (r *MirrorReconciler) syncUDPRoute(ctx context.Context, route *gwapiv1alpha
 	if apierrors.IsNotFound(err) {
 		log.Infof("Mirror: creating UDPRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 		if _, err := r.apoxyClient.GatewayV1alpha2().UDPRoutes().Create(ctx, apoxy, metav1.CreateOptions{}); err != nil {
+			MirrorSyncErrors.WithLabelValues("UDPRoute").Inc()
 			return reconcile.Result{}, fmt.Errorf("creating Apoxy UDPRoute %s: %w", apoxyName, err)
 		}
+		MirrorSyncedResources.WithLabelValues("UDPRoute").Inc()
 		return reconcile.Result{}, nil
 	} else if err != nil {
+		MirrorSyncErrors.WithLabelValues("UDPRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("getting Apoxy UDPRoute %s: %w", apoxyName, err)
 	}
 
 	apoxy.ResourceVersion = existing.ResourceVersion
 	log.Infof("Mirror: updating UDPRoute %s (from %s/%s)", apoxyName, route.Namespace, route.Name)
 	if _, err := r.apoxyClient.GatewayV1alpha2().UDPRoutes().Update(ctx, apoxy, metav1.UpdateOptions{}); err != nil {
+		MirrorSyncErrors.WithLabelValues("UDPRoute").Inc()
 		return reconcile.Result{}, fmt.Errorf("updating Apoxy UDPRoute %s: %w", apoxyName, err)
 	}
+	MirrorSyncedResources.WithLabelValues("UDPRoute").Inc()
 	return reconcile.Result{}, nil
 }
 
@@ -636,6 +666,7 @@ func (r *MirrorReconciler) RunHeartbeat(ctx context.Context, namespace string) e
 	for {
 		if err := r.renewLease(ctx, namespace, leaseName, durationSecs); err != nil {
 			log.Errorf("Mirror heartbeat: failed to renew lease %s: %v", leaseName, err)
+			MirrorHeartbeatFailures.Inc()
 		}
 
 		select {

pkg/kube-controller/controllers/mirror_metrics.go

@@ -0,0 +1,40 @@
+package controllers
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
+)
+
+var (
+	// MirrorSyncedResources counts successful resource sync operations.
+	MirrorSyncedResources = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "tunnel_mirror_synced_resources_total",
+			Help: "Total mirror resource sync operations that succeeded.",
+		},
+		[]string{"resource_type"},
+	)
+	// MirrorSyncErrors counts failed resource sync operations.
+	MirrorSyncErrors = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "tunnel_mirror_sync_errors_total",
+			Help: "Total mirror resource sync operations that failed.",
+		},
+		[]string{"resource_type"},
+	)
+	// MirrorHeartbeatFailures counts heartbeat lease renewal failures.
+	MirrorHeartbeatFailures = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "tunnel_mirror_heartbeat_failures_total",
+			Help: "Total mirror heartbeat lease renewal failures.",
+		},
+	)
+)
+
+func init() {
+	metrics.Registry.MustRegister(
+		MirrorSyncedResources,
+		MirrorSyncErrors,
+		MirrorHeartbeatFailures,
+	)
+}

pkg/tunnel/api/client.go

@@ -19,12 +19,13 @@ import (
 )
 
 type Client struct {
-	http       *http.Client
-	h3         *http3.Transport
-	baseURL    *url.URL
-	tunnelName string
-	token      string
-	agent      string
+	http        *http.Client
+	h3          *http3.Transport
+	baseURL     *url.URL
+	tunnelName  string
+	token       string
+	agent       string
+	metricsPort int
 }
 
 type ClientOptions struct {
@@ -44,6 +45,9 @@ type ClientOptions struct {
 	// PacketConn is an optional UDP PacketConn to use for QUIC connections.
 	// If nil, a new UDP socket will be created for each connection.
 	PacketConn net.PacketConn
+	// MetricsPort is the port the agent's Prometheus metrics server listens on.
+	// Advertised to the relay so it can scrape metrics through the overlay.
+	MetricsPort int
 }
 
 func NewClient(opts ClientOptions) (*Client, error) {
@@ -104,12 +108,13 @@ func NewClient(opts ClientOptions) (*Client, error) {
 	}
 
 	return &Client{
-		http:       hc,
-		h3:         t,
-		baseURL:    u,
-		tunnelName: opts.TunnelName,
-		token:      opts.Token,
-		agent:      opts.Agent,
+		http:        hc,
+		h3:          t,
+		baseURL:     u,
+		tunnelName:  opts.TunnelName,
+		token:       opts.Token,
+		agent:       opts.Agent,
+		metricsPort: opts.MetricsPort,
 	}, nil
 }
 
@@ -119,7 +124,7 @@ func (c *Client) Close() error {
 
 // Connect to the relay and establish a new tunnel connection.
 func (c *Client) Connect(ctx context.Context) (*ConnectResponse, error) {
-	reqBody := ConnectRequest{Agent: c.agent}
+	reqBody := ConnectRequest{Agent: c.agent, MetricsPort: c.metricsPort}
 	var resp ConnectResponse
 	if err := c.doJSON(ctx, http.MethodPost, c.path("/v1/tunnel/"+c.tunnelName), reqBody, &resp, http.StatusCreated); err != nil {
 		return nil, err

pkg/tunnel/api/types.go

@@ -16,6 +16,9 @@ type Request struct {
 type ConnectRequest struct {
 	// Agent is the name of the agent.
 	Agent string `json:"agent"`
+	// MetricsPort is the port the agent's Prometheus metrics server listens on.
+	// 0 means the agent does not expose metrics.
+	MetricsPort int `json:"metricsPort,omitempty"`
 }
 
 type ConnectResponse struct {

pkg/tunnel/bfdl/metrics.go

@@ -67,6 +67,16 @@ var (
 		},
 		[]string{"role", "direction"},
 	)
+
+	// BFDHeartbeatsReceived counts valid BFD heartbeat packets received.
+	// Labels: "role" (server|client).
+	BFDHeartbeatsReceived = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "tunnel_bfd_heartbeats_received_total",
+			Help: "Total valid BFD heartbeat packets received.",
+		},
+		[]string{"role"},
+	)
 )
 
 func init() {
@@ -77,5 +87,6 @@ func init() {
 		BFDStateTransitions,
 		BFDDetectTimeouts,
 		BFDPacketErrors,
+		BFDHeartbeatsReceived,
 	)
 }

pkg/tunnel/bfdl/server.go

@@ -211,8 +211,11 @@ func (s *Server) Start(ctx context.Context) error {
 			continue
 		}
 
-		if connID != "" && s.onAlive != nil {
-			s.onAlive(ctx, connID)
+		if connID != "" {
+			BFDHeartbeatsReceived.WithLabelValues("server").Inc()
+			if s.onAlive != nil {
+				s.onAlive(ctx, connID)
+			}
 		}
 	}
 }

pkg/tunnel/client.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"log/slog"
 	"net"
+	"strconv"
 	"net/http"
 	"net/netip"
 	"net/url"
@@ -26,6 +27,7 @@ import (
 	alog "github.com/apoxy-dev/apoxy/pkg/log"
 	"github.com/apoxy-dev/apoxy/pkg/tunnel/bfdl"
 	tunnelconn "github.com/apoxy-dev/apoxy/pkg/tunnel/connection"
+	"github.com/apoxy-dev/apoxy/pkg/tunnel/metrics"
 	"github.com/apoxy-dev/apoxy/pkg/tunnel/router"
 )
 
@@ -73,6 +75,9 @@ type tunnelClientOptions struct {
 	packetObserver tunnelconn.PacketObserver
 	// Labels to send on tunnel connections.
 	labels map[string]string
+	// metricsPort is the port the agent's metrics server listens on.
+	// Advertised to the server for overlay scraping.
+	metricsPort int
 }
 
 func defaultClientOptions() *tunnelClientOptions {
@@ -163,6 +168,14 @@ func WithLabels(labels map[string]string) TunnelClientOption {
 	}
 }
 
+// WithClientMetricsPort advertises the agent's metrics port to the server
+// so the server can scrape metrics from this agent through the overlay.
+func WithClientMetricsPort(port int) TunnelClientOption {
+	return func(o *tunnelClientOptions) {
+		o.metricsPort = port
+	}
+}
+
 // BuildClientRouter builds a router for the client tunnel side using provided
 // options and sane defaults.
 func BuildClientRouter(opts ...TunnelClientOption) (router.Router, error) {
@@ -288,6 +301,9 @@ func (d *TunnelDialer) Dial(
 	for k, v := range options.labels {
 		q.Add("label."+k, v)
 	}
+	if options.metricsPort > 0 {
+		q.Add("label."+metrics.LabelMetricsPort, strconv.Itoa(options.metricsPort))
+	}
 	addrUrl.RawQuery = q.Encode()
 
 	tmpl, err := uritemplate.New(addrUrl.String())