tcp_info version handling (#5) plus a batch of decode, cli, and platform fixes

cmd/get/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log"
 	"net"
 	"net/http"
@@ -60,6 +61,10 @@ func main() {
 	if err != nil {
 		log.Fatalf("get: %v", err)
 	}
+	// drain the body before closing so the conn can go back in the pool;
+	// otherwise it is never reused and the CloseIdleConnections call below has
+	// nothing to close.
+	_, _ = io.Copy(io.Discard, resp.Body)
 	_ = resp.Body.Close()
 
 	// Use client.CloseIdleConnections() to trigger the closed events for all wrapped connections.

cmd/httpstat/conniver_state_test.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"testing"
+	"time"
+
+	"github.com/runZeroInc/conniver"
+)
+
+// TestConniverPerHopReset checks that a hop never reads stats left from a
+// prior hop: reset clears the recorded conn and the wait only observes the
+// callback fired after the reset.
+func TestConniverPerHopReset(t *testing.T) {
+	// hop A records a conn.
+	a := &conniver.Conn{}
+	setConniver(a)
+	if getConniver() != a {
+		t.Fatalf("hop A: want recorded conn, got %v", getConniver())
+	}
+
+	// hop B starts: reset must drop hop A's stats.
+	setConniver(nil)
+	if getConniver() != nil {
+		t.Fatalf("hop B start: want nil, got stale %v", getConniver())
+	}
+
+	// with no callback yet, wait returns via timeout and conn stays nil.
+	start := time.Now()
+	waitConniver(50 * time.Millisecond)
+	if getConniver() != nil {
+		t.Fatalf("hop B before callback: want nil, got %v", getConniver())
+	}
+	if time.Since(start) < 40*time.Millisecond {
+		t.Fatalf("wait returned too early, ready channel not honored")
+	}
+
+	// hop B's callback fires: wait now unblocks promptly with hop B's conn.
+	setConniver(nil)
+	b := &conniver.Conn{}
+	go func() {
+		time.Sleep(10 * time.Millisecond)
+		setConniver(b)
+	}()
+	start = time.Now()
+	waitConniver(2 * time.Second)
+	if time.Since(start) > time.Second {
+		t.Fatalf("wait did not unblock on callback")
+	}
+	if getConniver() != b {
+		t.Fatalf("hop B: want %v, got %v", b, getConniver())
+	}
+}

cmd/httpstat/main.go

@@ -231,6 +231,7 @@ func dialContext(network string) func(ctx context.Context, network, addr string)
 var (
 	conniverM sync.Mutex
 	connConn  *conniver.Conn
+	connReady chan struct{}
 )
 
 func getConniver() *conniver.Conn {
@@ -239,15 +240,46 @@ func getConniver() *conniver.Conn {
 	return connConn
 }
 
+// setConniver records the closed conn for the current hop. Passing nil resets
+// for a new hop and arms a fresh ready channel, so a hop never prints stats
+// left from a prior hop.
 func setConniver(c *conniver.Conn) {
 	conniverM.Lock()
 	defer conniverM.Unlock()
 	connConn = c
+	if c == nil {
+		connReady = make(chan struct{})
+		return
+	}
+	if connReady != nil {
+		close(connReady)
+		connReady = nil
+	}
+}
+
+// waitConniver blocks until this hop's Closed callback has fired, or the
+// timeout elapses. DisableKeepAlives means the conn is never pooled, so
+// CloseIdleConnections can't force the callback and the read would otherwise
+// race the transport's readLoop.
+func waitConniver(timeout time.Duration) {
+	conniverM.Lock()
+	ready := connReady
+	conniverM.Unlock()
+	if ready == nil {
+		return
+	}
+	select {
+	case <-ready:
+	case <-time.After(timeout):
+	}
 }
 
 // visit visits a url and times the interaction.
 // If the response is a 30x, visit follows the redirect.
 func visit(url *url.URL) {
+	// drop any stats left from a prior hop before this hop opens its conn.
+	setConniver(nil)
+
 	req := newRequest(httpMethod, url, postBody)
 
 	var t0, t1, t2, t3, t4, t5, t6 time.Time
@@ -262,8 +294,11 @@ func visit(url *url.URL) {
 			}
 		},
 		ConnectDone: func(net, addr string, err error) {
+			// fires once per candidate address; a single failure is not fatal,
+			// the dialer still has other addresses to try. let client.Do below
+			// report the total failure once every address is exhausted.
 			if err != nil {
-				log.Fatalf("unable to connect to host %v: %v", addr, err)
+				return
 			}
 			t2 = time.Now()
 
@@ -375,8 +410,10 @@ func visit(url *url.URL) {
 		return strings.Join(v, "\n")
 	}
 
-	// Trigger connection close to gather final tcpinfo
+	// Trigger connection close to gather final tcpinfo, then wait for the
+	// Closed callback so the stats below belong to this hop.
 	client.CloseIdleConnections()
+	waitConniver(2 * time.Second)
 
 	fmt.Println()
 

pkg/kernel/kernel_unix_test.go

@@ -1,4 +1,4 @@
-//go:build !windows && !aix
+//go:build linux || freebsd || openbsd || darwin || netbsd || dragonfly
 
 package kernel
 

pkg/kernel/kernel_windows.go

@@ -11,10 +11,10 @@ import (
 
 // VersionInfo holds information about the kernel.
 type VersionInfo struct {
-	kvi   string // Version of the kernel (e.g. 6.1.7601.17592 -> 6)
-	major int    // Major part of the kernel version (e.g. 6.1.7601.17592 -> 1)
-	minor int    // Minor part of the kernel version (e.g. 6.1.7601.17592 -> 7601)
-	build int    // Build number of the kernel version (e.g. 6.1.7601.17592 -> 17592)
+	kvi   string // BuildLabEx registry string (e.g. 7601.17592.amd64fre.win7sp1_gdr.110408-1631)
+	major int    // major version, low byte of GetVersion (e.g. 6.1.7601 -> 6)
+	minor int    // minor version, second byte of GetVersion (e.g. 6.1.7601 -> 1)
+	build int    // build number, high word of GetVersion (e.g. 6.1.7601 -> 7601)
 }
 
 func (k *VersionInfo) String() string {
@@ -37,16 +37,13 @@ func GetKernelVersion() (*VersionInfo, error) {
 	}
 	KVI.kvi = blex
 
-	// Important - dockerd.exe MUST be manifested for this API to return
-	// the correct information.
-	dwVersion, err := windows.GetVersion()
-	if err != nil {
-		return KVI, err
-	}
+	// RtlGetVersion ignores the manifest shim that caps GetVersion at 6.2.9200
+	// for unmanifested processes, so it reports the real version on Win 8.1+.
+	ver := windows.RtlGetVersion()
 
-	KVI.major = int(dwVersion & 0xFF)
-	KVI.minor = int((dwVersion & 0xFF00) >> 8)
-	KVI.build = int((dwVersion & 0xFFFF0000) >> 16)
+	KVI.major = int(ver.MajorVersion)
+	KVI.minor = int(ver.MinorVersion)
+	KVI.build = int(ver.BuildNumber)
 
 	return KVI, nil
 }

pkg/os/operatingsystem_linux.go

@@ -18,8 +18,23 @@ var (
 
 	// used by stateless systems like Clear Linux
 	altOsRelease = "/usr/lib/os-release"
+
+	// marker files the runtime drops; on a cgroup v2 host /proc/1/cgroup is a
+	// bare "0::/" and can't be told from the host by suffix alone
+	dockerEnv    = "/.dockerenv"
+	containerEnv = "/run/.containerenv"
 )
 
+// cgroup path fragments naming a container runtime, for the cgroup v2 case
+// where the host check by suffix does not fire
+var containerCgroupMarkers = [][]byte{
+	[]byte("/docker"),
+	[]byte("/lxc"),
+	[]byte("/kubepods"),
+	[]byte("/containerd"),
+	[]byte("/machine"),
+}
+
 // GetOperatingSystem gets the name of the current operating system.
 func GetOperatingSystem() (string, error) {
 	if prettyName, err := getValueFromOsRelease("PRETTY_NAME"); err != nil {
@@ -62,19 +77,42 @@ func getValueFromOsRelease(key string) (string, error) {
 		}
 	}
 
+	// a stopped scan (oversized line, mid-read i/o error) must not pass as an empty value
+	if err := scanner.Err(); err != nil {
+		return "", err
+	}
+
 	return value, nil
 }
 
 // IsContainerized returns true if we are running inside a container.
 func IsContainerized() (bool, error) {
+	if fileExists(dockerEnv) || fileExists(containerEnv) {
+		return true, nil
+	}
 	b, err := os.ReadFile(proc1Cgroup)
 	if err != nil {
 		return false, err
 	}
 	for line := range bytes.SplitSeq(b, []byte{'\n'}) {
-		if len(line) > 0 && !bytes.HasSuffix(line, []byte(":/")) && !bytes.HasSuffix(line, []byte(":/init.scope")) {
+		if len(line) == 0 {
+			continue
+		}
+		// cgroup v1: a non-root path means containerized
+		if !bytes.HasSuffix(line, []byte(":/")) && !bytes.HasSuffix(line, []byte(":/init.scope")) {
 			return true, nil
 		}
+		// cgroup v2: single line whose path names a runtime
+		for _, marker := range containerCgroupMarkers {
+			if bytes.Contains(line, marker) {
+				return true, nil
+			}
+		}
 	}
 	return false, nil
 }
+
+func fileExists(path string) bool {
+	_, err := os.Stat(path)
+	return err == nil
+}

pkg/os/operatingsystem_linux_test.go

@@ -5,6 +5,7 @@ package os
 import (
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 )
 
@@ -156,6 +157,32 @@ func runEtcReleaseParsingTests(t *testing.T, tests []EtcReleaseParsingTest, pars
 	}
 }
 
+// a line past bufio's 64 KiB token limit stops the scan; the value must not
+// come back empty-with-nil so GetOperatingSystem quietly reports "Linux"
+func TestOsReleaseScanError(t *testing.T) {
+	backup := etcOsRelease
+	altBackup := altOsRelease
+	dir := os.TempDir()
+	etcOsRelease = filepath.Join(dir, "etcOsRelease")
+	altOsRelease = filepath.Join(dir, "altOsRelease")
+
+	defer func() {
+		os.Remove(etcOsRelease)
+		etcOsRelease = backup
+		altOsRelease = altBackup
+	}()
+
+	content := strings.Repeat("A", 65*1024) + "\nPRETTY_NAME=\"Ubuntu 14.04 LTS\"\n"
+	if err := os.WriteFile(etcOsRelease, []byte(content), 0o600); err != nil {
+		t.Fatalf("failed to write to %s: %v", etcOsRelease, err)
+	}
+
+	s, err := GetOperatingSystem()
+	if err == nil {
+		t.Fatalf("expected scan error, got %q with nil error", s)
+	}
+}
+
 func TestIsContainerized(t *testing.T) {
 	var (
 		backup                                = proc1Cgroup
@@ -250,6 +277,38 @@ func TestIsContainerized(t *testing.T) {
 	}
 }
 
+// cgroup v2 PID 1 reads a bare "0::/" that ends in ":/", so the suffix check
+// alone treats it as host; the marker file is what disambiguates
+func TestIsContainerizedCgroupV2(t *testing.T) {
+	backupCgroup := proc1Cgroup
+	backupDocker := dockerEnv
+	dir := os.TempDir()
+	proc1Cgroup = filepath.Join(dir, "proc1CgroupV2")
+	dockerEnv = filepath.Join(dir, "dockerenv")
+
+	defer func() {
+		os.Remove(proc1Cgroup)
+		os.Remove(dockerEnv)
+		proc1Cgroup = backupCgroup
+		dockerEnv = backupDocker
+	}()
+
+	if err := os.WriteFile(proc1Cgroup, []byte("0::/\n"), 0o600); err != nil {
+		t.Fatalf("failed to write to %s: %v", proc1Cgroup, err)
+	}
+	if err := os.WriteFile(dockerEnv, []byte{}, 0o600); err != nil {
+		t.Fatalf("failed to write to %s: %v", dockerEnv, err)
+	}
+
+	inContainer, err := IsContainerized()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !inContainer {
+		t.Fatal("Wrongly assuming non-containerized on cgroup v2 with docker marker")
+	}
+}
+
 func TestOsReleaseFallback(t *testing.T) {
 	backup := etcOsRelease
 	altBackup := altOsRelease

pkg/os/operatingsystem_unix.go

@@ -14,7 +14,8 @@ func GetOperatingSystem() (string, error) {
 	if err := unix.Uname(utsname); err != nil {
 		return "", err
 	}
-	return unix.ByteSliceToString(utsname.Machine[:]), nil
+	// Sysname is the OS name (e.g. "Darwin"); Machine is the CPU arch
+	return unix.ByteSliceToString(utsname.Sysname[:]), nil
 }
 
 // GetOperatingSystemVersion gets the version of the current operating system, as a string.

pkg/tcpinfo/init_linux.go

@@ -46,7 +46,7 @@ var tcpInfoSizes = []VersionedStructSize{
 	{Version: kernel.VersionInfo{Kernel: 4, Major: 1, Minor: 0}, Size: 136, Flag: &kernelVersionIsAtLeast_4_1},
 	{Version: kernel.VersionInfo{Kernel: 4, Major: 2, Minor: 0}, Size: 144, Flag: &kernelVersionIsAtLeast_4_2},
 	{Version: kernel.VersionInfo{Kernel: 4, Major: 6, Minor: 0}, Size: 160, Flag: &kernelVersionIsAtLeast_4_6},
-	{Version: kernel.VersionInfo{Kernel: 4, Major: 9, Minor: 0}, Size: 148, Flag: &kernelVersionIsAtLeast_4_9},
+	{Version: kernel.VersionInfo{Kernel: 4, Major: 9, Minor: 0}, Size: 168, Flag: &kernelVersionIsAtLeast_4_9},
 	{Version: kernel.VersionInfo{Kernel: 4, Major: 10, Minor: 0}, Size: 192, Flag: &kernelVersionIsAtLeast_4_10},
 	{Version: kernel.VersionInfo{Kernel: 4, Major: 18, Minor: 0}, Size: 200, Flag: &kernelVersionIsAtLeast_4_18},
 	{Version: kernel.VersionInfo{Kernel: 4, Major: 19, Minor: 0}, Size: 224, Flag: &kernelVersionIsAtLeast_4_19},

pkg/tcpinfo/tcpinfo_darwin.go

@@ -196,6 +196,7 @@ func (packed *RawInfo) Unpack() *SysInfo {
 	unpacked.TxRetransmitPackets = packed.TxRetransmitPackets
 
 	unpacked.TxOptions = []Option{}
+	unpacked.RxOptions = []Option{}
 	for _, flag := range tcpOptions {
 		if packed.Options&flag == 0 {
 			continue
@@ -226,7 +227,7 @@ func (s *SysInfo) ToInfo() *Info {
 		RxWindow:      uint64(s.RxWindow),
 		TxSSThreshold: uint64(s.TxSSThreshold),
 		TxWindowBytes: uint64(s.TxCWindow),
-		TxWindowSegs:  uint64(s.TxWindow),
+		// no TxWindowSegs on darwin: xnu has no per-segment cwnd, and snd_wnd is bytes not segments
 		Retransmits:   s.TxRetransmitPackets,
 		Sys:           s,
 	}