ipn,log: add logger for sockstat deltas

Signed-off-by: Will Norris <will@tailscale.com>
Co-authored-by: Melanie Warrick <warrick@tailscale.com>

cmd/tailscaled/depaware.txt

@@ -219,10 +219,11 @@ tailscale.com/cmd/tailscaled dependencies: (generated by github.com/tailscale/de
    L    tailscale.com/kube                                           from tailscale.com/ipn/store/kubestore
         tailscale.com/log/filelogger                                 from tailscale.com/logpolicy
         tailscale.com/log/logheap                                    from tailscale.com/control/controlclient
+        tailscale.com/log/sockstatlog                                from tailscale.com/ipn/ipnlocal
         tailscale.com/logpolicy                                      from tailscale.com/cmd/tailscaled+
         tailscale.com/logtail                                        from tailscale.com/control/controlclient+
         tailscale.com/logtail/backoff                                from tailscale.com/control/controlclient+
-        tailscale.com/logtail/filch                                  from tailscale.com/logpolicy
+        tailscale.com/logtail/filch                                  from tailscale.com/logpolicy+
         tailscale.com/metrics                                        from tailscale.com/derp+
         tailscale.com/net/connstats                                  from tailscale.com/net/tstun+
         tailscale.com/net/dns                                        from tailscale.com/ipn/ipnlocal+

ipn/ipnlocal/local.go

@@ -11,6 +11,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"net"
 	"net/http"
 	"net/http/httputil"
@@ -43,6 +44,8 @@ import (
 	"tailscale.com/ipn/ipnauth"
 	"tailscale.com/ipn/ipnstate"
 	"tailscale.com/ipn/policy"
+	"tailscale.com/log/sockstatlog"
+	"tailscale.com/logpolicy"
 	"tailscale.com/net/dns"
 	"tailscale.com/net/dnscache"
 	"tailscale.com/net/dnsfallback"
@@ -149,6 +152,7 @@ type LocalBackend struct {
 	sshAtomicBool         atomic.Bool
 	shutdownCalled        bool // if Shutdown has been called
 	debugSink             *capture.Sink
+	sockstatLogger        *sockstatlog.Logger
 
 	// getTCPHandlerForFunnelFlow returns a handler for an incoming TCP flow for
 	// the provided srcAddr and dstPort if one exists.
@@ -304,6 +308,14 @@ func NewLocalBackend(logf logger.Logf, logid string, store ipn.StateStore, diale
 		loginFlags:     loginFlags,
 	}
 
+	// for now, only log sockstats on unstable builds
+	if version.IsUnstableBuild() {
+		b.sockstatLogger, err = sockstatlog.NewLogger(logpolicy.LogsDir(logf))
+		if err != nil {
+			log.Printf("error setting up sockstat logger: %v", err)
+		}
+	}
+
 	// Default filter blocks everything and logs nothing, until Start() is called.
 	b.setFilter(filter.NewAllowNone(logf, &netipx.IPSet{}))
 
@@ -541,6 +553,10 @@ func (b *LocalBackend) Shutdown() {
 	}
 	b.mu.Unlock()
 
+	if b.sockstatLogger != nil {
+		b.sockstatLogger.Shutdown()
+	}
+
 	b.unregisterLinkMon()
 	b.unregisterHealthWatch()
 	if cc != nil {
@@ -1432,6 +1448,10 @@ func (b *LocalBackend) Start(opts ipn.Options) error {
 
 	b.e.SetNetInfoCallback(b.setNetInfo)
 
+	if b.sockstatLogger != nil {
+		b.sockstatLogger.Start()
+	}
+
 	blid := b.backendLogID
 	b.logf("Backend: logs: be:%v fe:%v", blid, opts.FrontendLogID)
 	b.send(ipn.Notify{BackendLogID: &blid})

log/sockstatlog/logger.go

@@ -0,0 +1,124 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+// Package sockstatlog provides a logger for capturing and storing network socket stats.
+package sockstatlog
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"time"
+
+	"tailscale.com/logtail/filch"
+	"tailscale.com/net/sockstats"
+	"tailscale.com/util/mak"
+)
+
+// pollPeriod specifies how often to poll for socket stats.
+const pollPeriod = time.Second / 10
+
+// Logger logs statistics about network sockets.
+type Logger struct {
+	ticker    time.Ticker
+	logbuffer *filch.Filch
+}
+
+// deltaStat represents the bytes transferred during a time period.
+// The first element is transmitted bytes, the second element is received bytes.
+type deltaStat [2]uint64
+
+// event represents the socket stats on a specific interface during a time period.
+type event struct {
+	// Time is when the event started as a Unix timestamp in milliseconds.
+	Time int64 `json:"t"`
+
+	// Duration is the duration of this event in milliseconds.
+	Duration int64 `json:"d"`
+
+	// IsCellularInterface is set to 1 if the traffic was sent over a cellular interface.
+	IsCellularInterface int `json:"c,omitempty"`
+
+	// Stats records the stats for each Label during the time period.
+	Stats map[sockstats.Label]deltaStat `json:"s"`
+}
+
+// NewLogger returns a new Logger that will store stats in logdir.
+// On platforms that do not support sockstat logging, a nil Logger will be returned.
+// The returned Logger is not yet running.
+func NewLogger(logdir string) (*Logger, error) {
+	if !sockstats.IsAvailable {
+		return nil, nil
+	}
+
+	if err := os.MkdirAll(logdir, 0755); err != nil && !os.IsExist(err) {
+		return nil, err
+	}
+	filchPrefix := filepath.Join(logdir, "sockstats")
+	filch, err := filch.New(filchPrefix, filch.Options{ReplaceStderr: false})
+	if err != nil {
+		return nil, err
+	}
+
+	return &Logger{
+		ticker:    *time.NewTicker(pollPeriod),
+		logbuffer: filch,
+	}, nil
+}
+
+func (l *Logger) Start() {
+	go l.poll()
+}
+
+// poll fetches the current socket stats at the configured time interval,
+// calculates the delta since the last poll, and logs any non-zero values.
+// This method does not return.
+func (l *Logger) poll() {
+	// last is the last set of socket stats we saw.
+	var lastStats *sockstats.SockStats
+	var lastTime time.Time
+
+	enc := json.NewEncoder(l.logbuffer)
+	for t := range l.ticker.C {
+		stats := sockstats.Get()
+		if lastStats != nil {
+			diffstats := delta(lastStats, stats)
+			if len(diffstats) > 0 {
+				e := event{
+					Time:     lastTime.UnixMilli(),
+					Duration: t.Sub(lastTime).Milliseconds(),
+					Stats:    diffstats,
+				}
+				if stats.CurrentInterfaceCellular {
+					e.IsCellularInterface = 1
+				}
+				enc.Encode(e)
+			}
+		}
+		lastTime = t
+		lastStats = stats
+	}
+}
+
+func (l *Logger) Shutdown() {
+	l.ticker.Stop()
+	l.logbuffer.Close()
+}
+
+// delta calculates the delta stats between two SockStats snapshots.
+// b is assumed to have occurred after a.
+// Zero values are omitted from the returned map, and an empty map is returned if no bytes were transferred.
+func delta(a, b *sockstats.SockStats) (stats map[sockstats.Label]deltaStat) {
+	if a == nil || b == nil {
+		return nil
+	}
+	for label, bs := range b.Stats {
+		as := a.Stats[label]
+		if as.TxBytes == bs.TxBytes && as.RxBytes == bs.RxBytes {
+			// fast path for unchanged stats
+			continue
+		}
+		mak.Set(&stats, label, deltaStat{bs.TxBytes - as.TxBytes, bs.RxBytes - as.RxBytes})
+	}
+	return stats
+}

log/sockstatlog/logger_test.go

@@ -0,0 +1,119 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+package sockstatlog
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"tailscale.com/net/sockstats"
+)
+
+func TestDelta(t *testing.T) {
+	tests := []struct {
+		name      string
+		a, b      *sockstats.SockStats
+		wantStats map[sockstats.Label]deltaStat
+	}{
+		{
+			name:      "nil a stat",
+			a:         nil,
+			b:         &sockstats.SockStats{},
+			wantStats: nil,
+		},
+		{
+			name:      "nil b stat",
+			a:         &sockstats.SockStats{},
+			b:         nil,
+			wantStats: nil,
+		},
+		{
+			name: "no change",
+			a: &sockstats.SockStats{
+				Stats: map[sockstats.Label]sockstats.SockStat{
+					sockstats.LabelDERPHTTPClient: {
+						TxBytes: 10,
+						TxBytesByInterface: map[string]uint64{
+							"en0": 10,
+						},
+					},
+				},
+			},
+			b: &sockstats.SockStats{
+				Stats: map[sockstats.Label]sockstats.SockStat{
+					sockstats.LabelDERPHTTPClient: {
+						TxBytes: 10,
+						TxBytesByInterface: map[string]uint64{
+							"en0": 10,
+						},
+					},
+				},
+			},
+			wantStats: nil,
+		},
+		{
+			name: "tx after empty stat",
+			a:    &sockstats.SockStats{},
+			b: &sockstats.SockStats{
+				Stats: map[sockstats.Label]sockstats.SockStat{
+					sockstats.LabelDERPHTTPClient: {
+						TxBytes: 10,
+						TxBytesByInterface: map[string]uint64{
+							"en0": 10,
+						},
+					},
+				},
+				Interfaces: []string{"en0"},
+			},
+			wantStats: map[sockstats.Label]deltaStat{
+				sockstats.LabelDERPHTTPClient: {10, 0},
+			},
+		},
+		{
+			name: "rx after non-empty stat",
+			a: &sockstats.SockStats{
+				Stats: map[sockstats.Label]sockstats.SockStat{
+					sockstats.LabelDERPHTTPClient: {
+						TxBytes: 10,
+						RxBytes: 10,
+						TxBytesByInterface: map[string]uint64{
+							"en0": 10,
+						},
+						RxBytesByInterface: map[string]uint64{
+							"en0": 10,
+						},
+					},
+				},
+				Interfaces: []string{"en0"},
+			},
+			b: &sockstats.SockStats{
+				Stats: map[sockstats.Label]sockstats.SockStat{
+					sockstats.LabelDERPHTTPClient: {
+						TxBytes: 10,
+						RxBytes: 30,
+						TxBytesByInterface: map[string]uint64{
+							"en0": 10,
+						},
+						RxBytesByInterface: map[string]uint64{
+							"en0": 30,
+						},
+					},
+				},
+				Interfaces: []string{"en0"},
+			},
+			wantStats: map[sockstats.Label]deltaStat{
+				sockstats.LabelDERPHTTPClient: {0, 20},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotStats := delta(tt.a, tt.b)
+			if !cmp.Equal(gotStats, tt.wantStats) {
+				t.Errorf("gotStats = %v, want %v", gotStats, tt.wantStats)
+			}
+		})
+	}
+}

logpolicy/logpolicy.go

@@ -192,9 +192,9 @@ func (l logWriter) Write(buf []byte) (int, error) {
 	return len(buf), nil
 }
 
-// logsDir returns the directory to use for log configuration and
+// LogsDir returns the directory to use for log configuration and
 // buffer storage.
-func logsDir(logf logger.Logf) string {
+func LogsDir(logf logger.Logf) string {
 	if d := os.Getenv("TS_LOGS_DIR"); d != "" {
 		fi, err := os.Stat(d)
 		if err == nil && fi.IsDir() {
@@ -478,7 +478,7 @@ func NewWithConfigPath(collection, dir, cmdName string) *Policy {
 	}
 
 	if dir == "" {
-		dir = logsDir(earlyLogf)
+		dir = LogsDir(earlyLogf)
 	}
 	if cmdName == "" {
 		cmdName = version.CmdName()

net/sockstats/sockstats.go

@@ -18,8 +18,9 @@ import (
 // WithSockStats() function, along with the interfaces that we have
 // per-interface statistics for.
 type SockStats struct {
-	Stats      map[Label]SockStat
+	Stats                    map[Label]SockStat
-	Interfaces []string
+	Interfaces               []string
+	CurrentInterfaceCellular bool
 }
 
 // SockStat contains the sent and received bytes for a socket instrumented with

net/sockstats/sockstats_noop.go

@@ -9,6 +9,8 @@ import (
 	"context"
 )
 
+const IsAvailable = false
+
 func withSockStats(ctx context.Context, label Label) context.Context {
 	return ctx
 }

net/sockstats/sockstats_tsgo.go

@@ -19,6 +19,8 @@ import (
 	"tailscale.com/util/clientmetric"
 )
 
+const IsAvailable = true
+
 type sockStatCounters struct {
 	txBytes, rxBytes                       atomic.Uint64
 	rxBytesByInterface, txBytesByInterface map[int]*atomic.Uint64
@@ -155,8 +157,9 @@ func get() *SockStats {
 	defer sockStats.mu.Unlock()
 
 	r := &SockStats{
-		Stats:      make(map[Label]SockStat),
+		Stats:                    make(map[Label]SockStat),
-		Interfaces: make([]string, 0, len(sockStats.usedInterfaces)),
+		Interfaces:               make([]string, 0, len(sockStats.usedInterfaces)),
+		CurrentInterfaceCellular: sockStats.currentInterfaceCellular.Load(),
 	}
 	for iface := range sockStats.usedInterfaces {
 		r.Interfaces = append(r.Interfaces, sockStats.knownInterfaces[iface])