diff --git a/cmd/get/main.go b/cmd/get/main.go index cae1f47..c318103 100644 --- a/cmd/get/main.go +++ b/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. diff --git a/cmd/httpstat/conniver_state_test.go b/cmd/httpstat/conniver_state_test.go new file mode 100644 index 0000000..bf070c8 --- /dev/null +++ b/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()) + } +} diff --git a/cmd/httpstat/main.go b/cmd/httpstat/main.go index 2df7e17..2b4d07b 100644 --- a/cmd/httpstat/main.go +++ b/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() diff --git a/pkg/kernel/kernel_unix_test.go b/pkg/kernel/kernel_unix_test.go index fbb67c9..e042251 100644 --- a/pkg/kernel/kernel_unix_test.go +++ b/pkg/kernel/kernel_unix_test.go @@ -1,4 +1,4 @@ -//go:build !windows && !aix +//go:build linux || freebsd || openbsd || darwin || netbsd || dragonfly package kernel diff --git a/pkg/kernel/kernel_windows.go b/pkg/kernel/kernel_windows.go index 879f4f1..2510cf1 100644 --- a/pkg/kernel/kernel_windows.go +++ b/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 } diff --git a/pkg/os/operatingsystem_linux.go b/pkg/os/operatingsystem_linux.go index 3992b96..45561bd 100644 --- a/pkg/os/operatingsystem_linux.go +++ b/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 +} diff --git a/pkg/os/operatingsystem_linux_test.go b/pkg/os/operatingsystem_linux_test.go index c3bedbf..0863d0d 100644 --- a/pkg/os/operatingsystem_linux_test.go +++ b/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 diff --git a/pkg/os/operatingsystem_unix.go b/pkg/os/operatingsystem_unix.go index e46af47..15b1bda 100644 --- a/pkg/os/operatingsystem_unix.go +++ b/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. diff --git a/pkg/tcpinfo/init_linux.go b/pkg/tcpinfo/init_linux.go index 8d1d5a6..bd90db5 100644 --- a/pkg/tcpinfo/init_linux.go +++ b/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}, diff --git a/pkg/tcpinfo/tcpinfo_darwin.go b/pkg/tcpinfo/tcpinfo_darwin.go index 695685c..df1c17f 100644 --- a/pkg/tcpinfo/tcpinfo_darwin.go +++ b/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, } diff --git a/pkg/tcpinfo/tcpinfo_linux.go b/pkg/tcpinfo/tcpinfo_linux.go index f40fb02..4a38477 100644 --- a/pkg/tcpinfo/tcpinfo_linux.go +++ b/pkg/tcpinfo/tcpinfo_linux.go @@ -8,6 +8,7 @@ import ( "strconv" "syscall" "time" + "unsafe" "golang.org/x/sys/unix" ) @@ -173,7 +174,7 @@ type SysInfo struct { Rehash NullableUint32 `tcpi:"name=rehash,prom_type=gauge,prom_help='PLB or timeout triggered rehash attempts.'" json:"rehash,omitempty"` TotalRTO NullableUint16 `tcpi:"name=total_rto,prom_type=counter,prom_help='Total number of RTO timeouts, including SYN/SYN-ACK and recurring timeouts.'" json:"totalRTO,omitempty"` TotalRTORecoveries NullableUint16 `tcpi:"name=total_rto_recoveries,prom_type=counter,prom_help='Total number of RTO recoveries, including any unfinished recovery.'" json:"totalRTORecoveries,omitempty"` - TotalRTOTime NullableUint32 `tcpi:"name=total_rto_time,prom_type=counter,prom_help='Total time spent in RTO recoveries in nanoseconds, including any unfinished recovery.'" json:"totalRTOTime,omitempty"` + TotalRTOTime NullableUint32 `tcpi:"name=total_rto_time,prom_type=counter,prom_help='Total time spent in RTO recoveries in milliseconds, including any unfinished recovery.'" json:"totalRTOTime,omitempty"` CCAlgorithm string `tcpi:"name=cc_algorithm,prom_type=gauge,prom_help='Congestion control algorithm in use for this connection.'" json:"ccAlgorithm,omitempty"` // Vegas CCVegasEnabled NullableUint32 `tcpi:"name=cc_vegas_enabled,prom_type=gauge,prom_help='Whether TCP Vegas is enabled system-wide (true/false).'" json:"ccVegasEnabled,omitempty"` @@ -381,6 +382,14 @@ func (s *SysInfo) MarshalJSON() ([]byte, error) { // timeFieldMultiplier is used to convert fields representing time in microseconds to time.Duration (nanoseconds). var timeFieldMultiplier = time.Microsecond +// nativeIsBigEndian reports the host byte order. The wscale and +// app-limited/fastopen values are C bitfields whose member order flips between +// big- and little-endian, so Unpack has to know which way to read them. +var nativeIsBigEndian = func() bool { + var probe uint16 = 1 + return *(*byte)(unsafe.Pointer(&probe)) == 0 +}() + // Unpack copies fields from RawTCPInfo to TCPInfo, taking care of the bitfields and marking fields not provided // by older kernel versions as null. In the future it may deal with varying lengths of the struct returned by the // system call (i.e., kernels older than 5.4.0). @@ -394,19 +403,38 @@ func (packed *RawTCPInfo) Unpack() *SysInfo { unpacked.Retransmits = packed.retransmits unpacked.Probes = packed.probes unpacked.Backoff = packed.backoff - unpacked.TxWindowScale = packed.bitfield0 & 0x0f - unpacked.RxWindowScale = packed.bitfield0 >> 4 + // The wscale and app-limited/fastopen values are C bitfields, so where they + // sit in the byte depends on host byte order: the first-declared member is + // in the low bits on little-endian and the high bits on big-endian. Read + // them per-endianness so s390x/ppc64/mips agree with amd64 rather than + // getting swapped nibbles and shifted flags. + b0, b1 := packed.bitfield0, packed.bitfield1 + if nativeIsBigEndian { + unpacked.TxWindowScale = b0 >> 4 + unpacked.RxWindowScale = b0 & 0x0f + } else { + unpacked.TxWindowScale = b0 & 0x0f + unpacked.RxWindowScale = b0 >> 4 + } unpacked.DeliveryRateAppLimited = NullableBool{Valid: false} if kernelVersionIsAtLeast_4_9 { unpacked.DeliveryRateAppLimited.Valid = true - unpacked.DeliveryRateAppLimited.Value = packed.bitfield1&1 == 1 // added in v4.9 + if nativeIsBigEndian { + unpacked.DeliveryRateAppLimited.Value = (b1>>7)&1 == 1 // added in v4.9 + } else { + unpacked.DeliveryRateAppLimited.Value = b1&1 == 1 // added in v4.9 + } } unpacked.FastOpenClientFail = NullableUint8{Valid: false} if kernelVersionIsAtLeast_5_5 { // added in v5.5 unpacked.FastOpenClientFail.Valid = true - unpacked.FastOpenClientFail.Value = (packed.bitfield1 >> 1) & 0x3 + if nativeIsBigEndian { + unpacked.FastOpenClientFail.Value = (b1 >> 5) & 0x3 + } else { + unpacked.FastOpenClientFail.Value = (b1 >> 1) & 0x3 + } } unpacked.RTO = time.Duration(packed.rto) * timeFieldMultiplier @@ -418,10 +446,11 @@ func (packed *RawTCPInfo) Unpack() *SysInfo { unpacked.Lost = packed.lost unpacked.Retrans = packed.retrans unpacked.Fackets = packed.fackets - unpacked.LastTxAt = time.Duration(packed.last_data_sent) * timeFieldMultiplier + // the kernel emits tcpi_last_data_* / tcpi_last_ack_recv via jiffies_to_msecs, so these are milliseconds, not microseconds + unpacked.LastTxAt = time.Duration(packed.last_data_sent) * time.Millisecond unpacked.LastTxAckAt = time.Duration(packed.last_ack_sent) * timeFieldMultiplier - unpacked.LastRxAt = time.Duration(packed.last_data_recv) * timeFieldMultiplier - unpacked.LastRxAckAt = time.Duration(packed.last_ack_recv) * timeFieldMultiplier + unpacked.LastRxAt = time.Duration(packed.last_data_recv) * time.Millisecond + unpacked.LastRxAckAt = time.Duration(packed.last_ack_recv) * time.Millisecond unpacked.PMTU = packed.pmtu unpacked.RxSSThreshold = packed.rcv_ssthresh unpacked.RTT = time.Duration(packed.rtt) * timeFieldMultiplier @@ -528,14 +557,17 @@ func (packed *RawTCPInfo) Unpack() *SysInfo { unpacked.RxWindow = NullableUint32{Valid: false} unpacked.Rehash = NullableUint32{Valid: false} - unpacked.TotalRTO = NullableUint16{Valid: false} - unpacked.TotalRTORecoveries = NullableUint16{Valid: false} - unpacked.TotalRTOTime = NullableUint32{Valid: false} if kernelVersionIsAtLeast_6_2 { unpacked.RxWindow.Valid = true unpacked.RxWindow.Value = packed.rcv_wnd unpacked.Rehash.Valid = true unpacked.Rehash.Value = packed.rehash + } + + unpacked.TotalRTO = NullableUint16{Valid: false} + unpacked.TotalRTORecoveries = NullableUint16{Valid: false} + unpacked.TotalRTOTime = NullableUint32{Valid: false} + if kernelVersionIsAtLeast_6_7 { unpacked.TotalRTO.Valid = true unpacked.TotalRTO.Value = packed.total_rto unpacked.TotalRTORecoveries.Valid = true diff --git a/pkg/tcpinfo/tcpinfo_linux_matrix_test.go b/pkg/tcpinfo/tcpinfo_linux_matrix_test.go new file mode 100644 index 0000000..32216e3 --- /dev/null +++ b/pkg/tcpinfo/tcpinfo_linux_matrix_test.go @@ -0,0 +1,146 @@ +//go:build linux + +/** + * Copyright (c) 2022, Xerra Earth Observation Institute + * See LICENSE.TXT in the root directory of this source tree. + */ + +package tcpinfo + +import ( + "testing" + "unsafe" + + "github.com/runZeroInc/conniver/pkg/kernel" +) + +// These tests exist because the version-adaptation logic was, for a long time, +// only ever exercised at the host kernel's version (see the shadowing bug fixed +// alongside them). Once injection actually works, the size table and the +// per-field version gating in Unpack become independently testable across the +// whole matrix, rather than at a single incidental data point. + +// TestSizeTableInvariants pins down two structural properties of tcpInfoSizes +// that hold for any append-only C struct: +// +// 1. Sizes are monotonically non-decreasing with kernel version. A struct that +// only ever gains trailing fields can never shrink. (This alone catches the +// historical v4.9=148 regression, which sat below v4.6=160.) +// 2. The newest row equals the actual size of the Go mirror struct. This ties +// the top of the table to ground truth and catches drift whenever a field +// is appended to RawTCPInfo without a matching table update. +func TestSizeTableInvariants(t *testing.T) { + prev := -1 + var prevV kernel.VersionInfo + for _, row := range tcpInfoSizes { + if row.Size < prev { + t.Errorf("non-monotonic size table: v%s=%d is smaller than v%s=%d; "+ + "an append-only struct cannot shrink across versions", + row.Version.String(), row.Size, prevV.String(), prev) + } + prev = row.Size + prevV = row.Version + } + + want := int(unsafe.Sizeof(RawTCPInfo{})) + last := tcpInfoSizes[len(tcpInfoSizes)-1] + if last.Size != want { + t.Errorf("newest size row v%s=%d does not match sizeof(RawTCPInfo)=%d; "+ + "table and struct have drifted apart", + last.Version.String(), last.Size, want) + } +} + +// fieldIntroducedIn records the first kernel version in which each nullable +// tcp_info field became populated, taken from the commit annotations on the +// RawTCPInfo struct definition. It is the ground truth the Unpack version gating +// must agree with. +var fieldIntroducedIn = []struct { + name string + version kernel.VersionInfo + valid func(*SysInfo) bool +}{ + {"PacingRate", kernel.VersionInfo{Kernel: 3, Major: 15}, func(s *SysInfo) bool { return s.PacingRate.Valid }}, + {"MaxPacingRate", kernel.VersionInfo{Kernel: 3, Major: 15}, func(s *SysInfo) bool { return s.MaxPacingRate.Valid }}, + {"BytesAcked", kernel.VersionInfo{Kernel: 4, Major: 1}, func(s *SysInfo) bool { return s.BytesAcked.Valid }}, + {"BytesReceived", kernel.VersionInfo{Kernel: 4, Major: 1}, func(s *SysInfo) bool { return s.BytesReceived.Valid }}, + {"SegsOut", kernel.VersionInfo{Kernel: 4, Major: 2}, func(s *SysInfo) bool { return s.SegsOut.Valid }}, + {"SegsIn", kernel.VersionInfo{Kernel: 4, Major: 2}, func(s *SysInfo) bool { return s.SegsIn.Valid }}, + {"NotSentBytes", kernel.VersionInfo{Kernel: 4, Major: 6}, func(s *SysInfo) bool { return s.NotSentBytes.Valid }}, + {"MinRTT", kernel.VersionInfo{Kernel: 4, Major: 6}, func(s *SysInfo) bool { return s.MinRTT.Valid }}, + {"DataSegsIn", kernel.VersionInfo{Kernel: 4, Major: 6}, func(s *SysInfo) bool { return s.DataSegsIn.Valid }}, + {"DataSegsOut", kernel.VersionInfo{Kernel: 4, Major: 6}, func(s *SysInfo) bool { return s.DataSegsOut.Valid }}, + {"DeliveryRateAppLimited", kernel.VersionInfo{Kernel: 4, Major: 9}, func(s *SysInfo) bool { return s.DeliveryRateAppLimited.Valid }}, + {"DeliveryRate", kernel.VersionInfo{Kernel: 4, Major: 9}, func(s *SysInfo) bool { return s.DeliveryRate.Valid }}, + {"BusyTime", kernel.VersionInfo{Kernel: 4, Major: 10}, func(s *SysInfo) bool { return s.BusyTime.Valid }}, + {"RxWindowLimited", kernel.VersionInfo{Kernel: 4, Major: 10}, func(s *SysInfo) bool { return s.RxWindowLimited.Valid }}, + {"TxBufferLimited", kernel.VersionInfo{Kernel: 4, Major: 10}, func(s *SysInfo) bool { return s.TxBufferLimited.Valid }}, + {"Delivered", kernel.VersionInfo{Kernel: 4, Major: 18}, func(s *SysInfo) bool { return s.Delivered.Valid }}, + {"DeliveredCE", kernel.VersionInfo{Kernel: 4, Major: 18}, func(s *SysInfo) bool { return s.DeliveredCE.Valid }}, + {"BytesSent", kernel.VersionInfo{Kernel: 4, Major: 19}, func(s *SysInfo) bool { return s.BytesSent.Valid }}, + {"BytesRetrans", kernel.VersionInfo{Kernel: 4, Major: 19}, func(s *SysInfo) bool { return s.BytesRetrans.Valid }}, + {"DSACKDups", kernel.VersionInfo{Kernel: 4, Major: 19}, func(s *SysInfo) bool { return s.DSACKDups.Valid }}, + {"ReordSeen", kernel.VersionInfo{Kernel: 4, Major: 19}, func(s *SysInfo) bool { return s.ReordSeen.Valid }}, + {"RxOutOfOrder", kernel.VersionInfo{Kernel: 5, Major: 4}, func(s *SysInfo) bool { return s.RxOutOfOrder.Valid }}, + {"TxWindow", kernel.VersionInfo{Kernel: 5, Major: 4}, func(s *SysInfo) bool { return s.TxWindow.Valid }}, + {"FastOpenClientFail", kernel.VersionInfo{Kernel: 5, Major: 5}, func(s *SysInfo) bool { return s.FastOpenClientFail.Valid }}, + {"RxWindow", kernel.VersionInfo{Kernel: 6, Major: 2}, func(s *SysInfo) bool { return s.RxWindow.Valid }}, + {"Rehash", kernel.VersionInfo{Kernel: 6, Major: 2}, func(s *SysInfo) bool { return s.Rehash.Valid }}, + {"TotalRTO", kernel.VersionInfo{Kernel: 6, Major: 7}, func(s *SysInfo) bool { return s.TotalRTO.Valid }}, + {"TotalRTORecoveries", kernel.VersionInfo{Kernel: 6, Major: 7}, func(s *SysInfo) bool { return s.TotalRTORecoveries.Valid }}, + {"TotalRTOTime", kernel.VersionInfo{Kernel: 6, Major: 7}, func(s *SysInfo) bool { return s.TotalRTOTime.Valid }}, +} + +// TestVersionFieldGating asserts that, for a spread of kernel versions +// (including the boundary cases that previously hid bugs), Unpack marks each +// field Valid if and only if that field's introduction version is <= the pinned +// kernel. The 6.6.0 case is the direct regression guard for the RTO fields that +// were mis-gated under _6_2: on 6.6 they must be absent, on 6.7 present. +func TestVersionFieldGating(t *testing.T) { + pins := []kernel.VersionInfo{ + {Kernel: 2, Major: 6, Minor: 2}, + {Kernel: 4, Major: 8, Minor: 0}, // just below delivery_rate (4.9) + {Kernel: 4, Major: 9, Minor: 0}, // delivery_rate present, size must be 168 + {Kernel: 5, Major: 4, Minor: 0}, + {Kernel: 6, Major: 2, Minor: 0}, // rcv_wnd/rehash present, RTO fields absent + {Kernel: 6, Major: 6, Minor: 0}, // still below 6.7: RTO fields MUST be absent + {Kernel: 6, Major: 7, Minor: 0}, // RTO fields present + } + + for _, pin := range pins { + pin := pin + t.Run(pin.String(), func(t *testing.T) { + linuxKernelVersion = &pin + adaptToKernelVersion() + got := (&RawTCPInfo{}).Unpack() + for _, f := range fieldIntroducedIn { + wantValid := kernel.CompareKernelVersion(pin, f.version) >= 0 + if f.valid(got) != wantValid { + t.Errorf("at kernel %s: field %s Valid=%v, want %v (introduced in %s)", + pin.String(), f.name, f.valid(got), wantValid, f.version.String()) + } + } + }) + } +} + +// TestInjectionHolds is the direct regression test for the shadowing bug: a +// pinned version must actually determine sizeOfRawTCPInfo and the flags, +// independent of the host kernel that runs the test. +func TestInjectionHolds(t *testing.T) { + linuxKernelVersion = &kernel.VersionInfo{Kernel: 2, Major: 6, Minor: 2} + adaptToKernelVersion() + if sizeOfRawTCPInfo != 104 { + t.Errorf("pinned 2.6.2 but sizeOfRawTCPInfo=%d, want 104 (injection discarded?)", sizeOfRawTCPInfo) + } + if kernelVersionIsAtLeast_3_15 { + t.Error("pinned 2.6.2 but kernelVersionIsAtLeast_3_15 is true (host kernel leaked in)") + } + + linuxKernelVersion = &kernel.VersionInfo{Kernel: 6, Major: 7, Minor: 0} + + adaptToKernelVersion() + if !kernelVersionIsAtLeast_6_7 { + t.Error("pinned 6.7.0 but kernelVersionIsAtLeast_6_7 is false") + } +} diff --git a/pkg/tcpinfo/tcpinfo_linux_test.go b/pkg/tcpinfo/tcpinfo_linux_test.go index 68b7739..d5fa78d 100644 --- a/pkg/tcpinfo/tcpinfo_linux_test.go +++ b/pkg/tcpinfo/tcpinfo_linux_test.go @@ -19,7 +19,7 @@ import ( const ( minKernel int = 6 - minKernelMajor int = 2 + minKernelMajor int = 7 minKernelMinor int = 0 ) @@ -270,6 +270,33 @@ func TestRawTCPInfo_Unpack(t *testing.T) { } } +func TestRawTCPInfo_UnpackLastDataUnits(t *testing.T) { + linuxKernelVersion = &kernel.VersionInfo{Kernel: minKernel, Major: minKernelMajor, Minor: minKernelMinor} + adaptToKernelVersion() + + // tcpi_last_data_* are kernel milliseconds; rtt is genuinely microseconds + raw := RawTCPInfo{ + last_data_sent: 2000, + last_data_recv: 2000, + last_ack_recv: 2000, + rtt: 2000, + } + + got := raw.Unpack() + if got.LastRxAt != 2*time.Second { + t.Errorf("LastRxAt = %v, want %v", got.LastRxAt, 2*time.Second) + } + if got.LastTxAt != 2*time.Second { + t.Errorf("LastTxAt = %v, want %v", got.LastTxAt, 2*time.Second) + } + if got.LastRxAckAt != 2*time.Second { + t.Errorf("LastRxAckAt = %v, want %v", got.LastRxAckAt, 2*time.Second) + } + if got.RTT != 2*time.Millisecond { + t.Errorf("RTT = %v, want %v", got.RTT, 2*time.Millisecond) + } +} + func TestRawTCPInfo_UnpackWindowScaleOptions(t *testing.T) { linuxKernelVersion = &kernel.VersionInfo{Kernel: minKernel, Major: minKernelMajor, Minor: minKernelMinor} adaptToKernelVersion() diff --git a/pkg/tcpinfo/tcpinfo_windows.go b/pkg/tcpinfo/tcpinfo_windows.go index 0d630ca..7836888 100644 --- a/pkg/tcpinfo/tcpinfo_windows.go +++ b/pkg/tcpinfo/tcpinfo_windows.go @@ -220,7 +220,7 @@ func (s *SysInfo) ToInfo() *Info { info := &Info{ State: s.StateName, TxMSS: uint64(s.MSS), - RTT: s.RTTMin, + RTT: s.RTT, RxWindow: uint64(s.RxWindow), TxWindowSegs: uint64(s.TxWindow), Retransmits: uint64(s.SynRetrans), diff --git a/wrap.go b/wrap.go index 2b4b06c..a45cabc 100644 --- a/wrap.go +++ b/wrap.go @@ -347,6 +347,8 @@ func (w *Conn) Read(b []byte) (int, error) { if err != nil { return 0, err } + // defer so a panicking conn still drains inFlight, else Close() deadlocks + defer w.finishIO() n, err := conn.Read(b) w.Lock() @@ -364,7 +366,6 @@ func (w *Conn) Read(b []byte) (int, error) { w.RxErr = err } w.Unlock() - w.finishIO() return n, err } @@ -374,6 +375,8 @@ func (w *Conn) Write(b []byte) (int, error) { if err != nil { return 0, err } + // defer so a panicking conn still drains inFlight, else Close() deadlocks + defer w.finishIO() n, err := conn.Write(b) w.Lock() @@ -391,7 +394,6 @@ func (w *Conn) Write(b []byte) (int, error) { w.TxErr = err } w.Unlock() - w.finishIO() return n, err } diff --git a/wrap_panic_test.go b/wrap_panic_test.go new file mode 100644 index 0000000..31f02f9 --- /dev/null +++ b/wrap_panic_test.go @@ -0,0 +1,43 @@ +package conniver + +import ( + "net" + "testing" + "time" +) + +// panicConn is a net.Conn whose Read panics, modeling a caller-supplied conn +// that blows up mid-read with the panic recovered upstream. +type panicConn struct { + net.Conn +} + +func (c *panicConn) Read(b []byte) (int, error) { panic("read boom") } +func (c *panicConn) Write(b []byte) (int, error) { return len(b), nil } +func (c *panicConn) Close() error { return nil } +func (c *panicConn) LocalAddr() net.Addr { return testAddr("127.0.0.1:12345") } +func (c *panicConn) RemoteAddr() net.Addr { return testAddr("127.0.0.1:443") } +func (c *panicConn) SetDeadline(time.Time) error { return nil } +func (c *panicConn) SetReadDeadline(time.Time) error { return nil } +func (c *panicConn) SetWriteDeadline(time.Time) error { return nil } + +func TestConnCloseAfterPanickingRead(t *testing.T) { + wrapped := WrapConn(&panicConn{}, nil) + + func() { + defer func() { _ = recover() }() + _, _ = wrapped.Read(make([]byte, 8)) + }() + + done := make(chan struct{}) + go func() { + _ = wrapped.Close() + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Close() deadlocked after a recovered panic in Read; inFlight leaked") + } +}