net/activesum: network activity summary

Signed-off-by: David Crawshaw <crawshaw@tailscale.com>

net/activesum/activesum.go

@@ -0,0 +1,80 @@
+// Package activesum summarizes network activity into coarse event blocks.
+package activesum
+
+import (
+	"time"
+)
+
+// Event is a coarse (if all is well, at least half-minute) period of
+// network activity. Events end after a Idle time passes, or the network
+// interface changes.
+type Event struct {
+	Start     time.Time     // start of event
+	Duration  time.Duration // duration of event
+	Bytes     uint64        // total rx+tx bytes during event window
+	Interface string        // network interface used for event
+}
+
+// Idle is the amount of time without data that marks the end of an event.
+const Idle = 30 * time.Second
+
+// ActiveSum stores activity summary state and generates Events.
+type ActiveSum struct {
+	// EventFunc is used to deliver complete Events.
+	EventFunc func(ev Event)
+
+	// Current event details.
+	start time.Time     // beginning of current event
+	last  time.Duration // nanos beyond start when last event was recorded
+	bytes uint64        // total rx+tx bytes so far
+	iface string        // network interface of current event
+}
+
+// Variables for testing.
+var timeNow = time.Now
+var timeSince = time.Since
+
+// Record records bytes transferred.
+func (a *ActiveSum) Record(bytes uint64, iface string) {
+	if bytes == 0 {
+		return
+	}
+
+	// The function time.Since is faster than a typical time.Now call
+	// because a.start includes monotonic time, so it uses a fast path
+	// in the runtime that does clock_gettime via VDSO on linux.
+	since := timeSince(a.start)
+
+	// Clear previous event if Idle has passed or interface changed.
+	if a.start.IsZero() || a.iface != iface || (since-a.last) > Idle {
+		a.recordEvent()
+
+		// Calls to time.Now are relatively slow (in per-packet terms), but
+		// we only call it once per event, which lasts at least Idle.
+		a.start = timeNow()
+		a.iface = iface
+		a.bytes = 0
+		since = 0
+	}
+
+	a.bytes += bytes
+	a.last = since
+}
+
+func (a *ActiveSum) recordEvent() {
+	if a.start.IsZero() {
+		return
+	}
+	a.EventFunc(Event{
+		Start:     a.start,
+		Duration:  a.last,
+		Bytes:     a.bytes,
+		Interface: a.iface,
+	})
+}
+
+// Close stops ActiveSum and records any remaining Event.
+func (a *ActiveSum) Close() {
+	a.recordEvent()
+	a.start = time.Time{}
+}

net/activesum/activesum_test.go

@@ -0,0 +1,75 @@
+package activesum
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+type testDatum struct {
+	offset time.Duration
+	bytes  uint64
+	iface  string
+}
+
+var tests = []struct {
+	name string
+	data []testDatum
+	want []Event
+}{
+	{
+		name: "basic",
+		data: []testDatum{
+			{offset: 0, bytes: 128, iface: "eth0"},
+			{offset: time.Millisecond, bytes: 512, iface: "eth0"},
+			{offset: 2 * time.Millisecond, bytes: 256, iface: "eth0"},
+			{offset: time.Second - 3*time.Millisecond, bytes: 128, iface: "eth0"},
+			{offset: 2 * Idle, bytes: 128, iface: "eth0"},
+			{offset: 2 * Idle, bytes: 50, iface: "eth0"},
+			{offset: 0, bytes: 50, iface: "lte0"},
+			{offset: time.Second, bytes: 50, iface: "eth0"},
+			{offset: Idle - 1*time.Second, bytes: 50, iface: "eth0"},
+			{offset: Idle - 1*time.Second, bytes: 50, iface: "eth0"},
+			{offset: Idle - 1*time.Second, bytes: 50, iface: "eth0"},
+			{offset: Idle - 1*time.Second, bytes: 50, iface: "eth0"},
+		},
+		want: []Event{
+			{Start: start, Duration: time.Second, Bytes: 1024, Interface: "eth0"},
+			{Start: start.Add(2*Idle + time.Second), Bytes: 128, Interface: "eth0"},
+			{Start: start.Add(4*Idle + time.Second), Bytes: 50, Interface: "eth0"},
+			{Start: start.Add(4*Idle + time.Second), Bytes: 50, Interface: "lte0"},
+			{Start: start.Add(4*Idle + 2*time.Second), Duration: 1*time.Minute + 56*time.Second, Bytes: 250, Interface: "eth0"},
+		},
+	},
+}
+
+var start = time.Date(1999, time.December, 31, 11, 11, 11, 0, time.UTC)
+
+func TestActiveSum(t *testing.T) {
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			now := start
+			timeNow = func() time.Time { return now }
+			timeSince = func(t time.Time) time.Duration { return now.Sub(t) }
+			t.Cleanup(func() {
+				timeNow = time.Now
+				timeSince = time.Since
+			})
+
+			var got []Event
+			a := &ActiveSum{EventFunc: func(ev Event) {
+				got = append(got, ev)
+			}}
+			for _, d := range test.data {
+				now = now.Add(d.offset)
+				a.Record(d.bytes, d.iface)
+			}
+			a.Close()
+			if !reflect.DeepEqual(got, test.want) {
+				t.Errorf("events mismatch (-got +want):\n%s", cmp.Diff(got, test.want))
+			}
+		})
+	}
+}