ipn/ipnlocal: add metrics for inbound and outbound bytes on Serve connections (#19991)

Adds tailscaled_serve_{inbound,outbound}_bytes_total, labeled by Tailscale
Service name, by wrapping the peer-facing conn in tcpHandlerForVIPService.
Per-service counters persist for the process lifetime rather than being
evicted on serve-config changes.

Fixes #19572

Signed-off-by: Raj Singh <raj@tailscale.com>
Co-authored-by: Ethan Smith <ethan.smith@grafana.com>

ipn/ipnlocal/local.go

@@ -520,6 +520,20 @@ type metrics struct {
 	// approvedRoutes is a metric that reports the number of network routes served by the local node and approved
 	// by the control server.
 	approvedRoutes *usermetric.Gauge
+
+	// serveBytesInbound counts bytes received from peers on Serve connections
+	// for Tailscale Services, labeled by Service name. Plain (non-Service)
+	// serve and funnel traffic is not counted.
+	serveBytesInbound *usermetric.MultiLabelMap[serveLabels]
+
+	// serveBytesOutbound counts bytes sent to peers on Serve connections for
+	// Tailscale Services, labeled by Service name. Plain (non-Service) serve
+	// and funnel traffic is not counted.
+	serveBytesOutbound *usermetric.MultiLabelMap[serveLabels]
+}
+
+type serveLabels struct {
+	Service string `prom:"service"`
 }
 
 // clientGen is a func that creates a control plane client.
@@ -571,6 +585,16 @@ func NewLocalBackend(logf logger.Logf, logID logid.PublicID, sys *tsd.System, lo
 			"tailscaled_advertised_routes", "Number of advertised network routes (e.g. by a subnet router)"),
 		approvedRoutes: sys.UserMetricsRegistry().NewGauge(
 			"tailscaled_approved_routes", "Number of approved network routes (e.g. by a subnet router)"),
+		serveBytesInbound: usermetric.NewMultiLabelMapWithRegistry[serveLabels](
+			sys.UserMetricsRegistry(),
+			"tailscaled_serve_inbound_bytes_total",
+			"counter",
+			"Bytes received from peers on Serve connections for Tailscale Services, labeled by Tailscale Service name."),
+		serveBytesOutbound: usermetric.NewMultiLabelMapWithRegistry[serveLabels](
+			sys.UserMetricsRegistry(),
+			"tailscaled_serve_outbound_bytes_total",
+			"counter",
+			"Bytes sent to peers on Serve connections for Tailscale Services, labeled by Tailscale Service name."),
 	}
 
 	b := &LocalBackend{

ipn/ipnlocal/serve.go

@@ -48,6 +48,7 @@ import (
 	"tailscale.com/util/ctxkey"
 	"tailscale.com/util/mak"
 	"tailscale.com/util/slicesx"
+	"tailscale.com/util/usermetric"
 	"tailscale.com/version"
 )
 
@@ -534,6 +535,56 @@ func (b *LocalBackend) vipServicesFromPrefsLocked(prefs ipn.PrefsView) []*tailcf
 	return servicesList
 }
 
+type serviceMeteredConn struct {
+	net.Conn
+	inbound, outbound *usermetric.MultiLabelMap[serveLabels]
+	key               serveLabels
+}
+
+func (c *serviceMeteredConn) Read(p []byte) (int, error) {
+	n, err := c.Conn.Read(p)
+	if n > 0 {
+		c.inbound.Add(c.key, int64(n))
+	}
+	return n, err
+}
+
+func (c *serviceMeteredConn) Write(p []byte) (int, error) {
+	n, err := c.Conn.Write(p)
+	if n > 0 {
+		c.outbound.Add(c.key, int64(n))
+	}
+	return n, err
+}
+
+// CloseWrite forwards a write-half close to the underlying conn. We only embed
+// the net.Conn interface, which would otherwise hide the underlying conn's
+// CloseWrite; net/http's server relies on it (closeWriteAndWait) to send a FIN
+// and drain gracefully when a connection won't be reused, avoiding a truncating
+// RST on the final response.
+func (c *serviceMeteredConn) CloseWrite() error {
+	if cw, ok := c.Conn.(interface{ CloseWrite() error }); ok {
+		return cw.CloseWrite()
+	}
+	return nil
+}
+
+// meteredConnForService wraps c to count peer bytes against the per-Service
+// Serve counters. The per-Service series is never evicted, so it leaks
+// (intentionally) until tailscaled exits.
+func (b *LocalBackend) meteredConnForService(c net.Conn, svc tailcfg.ServiceName) net.Conn {
+	// Plain (non-Service) serve passes an empty svc; don't meter it.
+	if svc == "" || b.metrics.serveBytesInbound == nil || b.metrics.serveBytesOutbound == nil {
+		return c
+	}
+	return &serviceMeteredConn{
+		Conn:     c,
+		inbound:  b.metrics.serveBytesInbound,
+		outbound: b.metrics.serveBytesOutbound,
+		key:      serveLabels{Service: svc.String()},
+	}
+}
+
 // tcpHandlerForVIPService returns a handler for a TCP connection to a VIP service
 // that is being served via the ipn.ServeConfig. It returns nil if the destination
 // address is not a VIP service or if the VIP service does not have a TCP handler set.
@@ -606,11 +657,13 @@ func (b *LocalBackend) tcpHandlerForServeTCP(tcph ipn.TCPPortHandlerView, dport
 		if tcph.HTTPS() {
 			hs.TLSConfig = b.serveTLSConfig(b.getTLSServeCertForPort(dport, forVIPService), serveTLSNextProtos())
 			return func(c net.Conn) error {
+				c = b.meteredConnForService(c, forVIPService)
 				return hs.ServeTLS(netutil.NewOneConnListener(c, nil), "", "")
 			}
 		}
 
 		return func(c net.Conn) error {
+			c = b.meteredConnForService(c, forVIPService)
 			return hs.Serve(netutil.NewOneConnListener(c, nil))
 		}
 	}
@@ -618,6 +671,7 @@ func (b *LocalBackend) tcpHandlerForServeTCP(tcph ipn.TCPPortHandlerView, dport
 	if backDst := tcph.TCPForward(); backDst != "" {
 		return func(conn net.Conn) error {
 			defer conn.Close()
+			conn = b.meteredConnForService(conn, forVIPService)
 			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 			backConn, err := b.dialer.SystemDial(ctx, "tcp", backDst)
 			cancel()

ipn/ipnlocal/serve_metrics_test.go

@@ -0,0 +1,82 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+//go:build !ts_omit_serve && !ts_omit_usermetrics
+
+package ipnlocal
+
+import (
+	"expvar"
+	"io"
+	"net"
+	"testing"
+
+	"tailscale.com/tailcfg"
+	"tailscale.com/util/usermetric"
+)
+
+func counterValue(m *usermetric.MultiLabelMap[serveLabels], svc string) int64 {
+	v, _ := m.Get(serveLabels{Service: svc}).(*expvar.Int)
+	if v == nil {
+		return -1
+	}
+	return v.Value()
+}
+
+func TestServiceMeteredConn(t *testing.T) {
+	b := newTestBackend(t)
+
+	clientSide, serverSide := net.Pipe()
+	defer clientSide.Close()
+	defer serverSide.Close()
+
+	wrapped := b.meteredConnForService(serverSide, tailcfg.ServiceName("svc:foo"))
+
+	const inboundPayload = "hello from client"
+	writeDone := make(chan struct{})
+	go func() {
+		clientSide.Write([]byte(inboundPayload))
+		close(writeDone)
+	}()
+	buf := make([]byte, len(inboundPayload))
+	if _, err := io.ReadFull(wrapped, buf); err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	<-writeDone
+	if got := counterValue(b.metrics.serveBytesInbound, "svc:foo"); got != int64(len(inboundPayload)) {
+		t.Errorf("inbound = %d; want %d", got, len(inboundPayload))
+	}
+
+	// Deliberately a different length than inboundPayload so a backwards
+	// inbound/outbound wiring can't pass.
+	const outboundPayload = "hello from the server side"
+	writeDone = make(chan struct{})
+	go func() {
+		wrapped.Write([]byte(outboundPayload))
+		close(writeDone)
+	}()
+	buf = make([]byte, len(outboundPayload))
+	if _, err := io.ReadFull(clientSide, buf); err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	<-writeDone
+	if got := counterValue(b.metrics.serveBytesOutbound, "svc:foo"); got != int64(len(outboundPayload)) {
+		t.Errorf("outbound = %d; want %d", got, len(outboundPayload))
+	}
+}
+
+func TestServiceMeteredConnLabelKeepsPrefix(t *testing.T) {
+	b := newTestBackend(t)
+	c1, c2 := net.Pipe()
+	defer c1.Close()
+	defer c2.Close()
+	wrapped := b.meteredConnForService(c1, tailcfg.ServiceName("svc:my-app"))
+	go c2.Write([]byte("x"))
+	io.ReadFull(wrapped, make([]byte, 1))
+	if v := counterValue(b.metrics.serveBytesInbound, "svc:my-app"); v != 1 {
+		t.Errorf("inbound for service=\"svc:my-app\" = %d; want 1", v)
+	}
+	if v := counterValue(b.metrics.serveBytesInbound, "my-app"); v != -1 {
+		t.Errorf("counter unexpectedly present under prefix-stripped name; got %d", v)
+	}
+}