Darwin refactor (#84)

* many darwin refactoring changes

* Add collector.Run for darwin

* collector stuff

* restore IP4 IP6 tests

cache/cache_test.go

@@ -3,7 +3,6 @@ package cache_test
 import (
 	"encoding/json"
 	"log"
-	"syscall"
 	"testing"
 
 	"github.com/m-lab/tcp-info/cache"
@@ -23,7 +22,7 @@ func testFatal(t *testing.T, err error) {
 
 func fakeMsg(t *testing.T, cookie uint64, dport uint16) netlink.ArchivalRecord {
 	var json1 = `{"Header":{"Len":356,"Type":20,"Flags":2,"Seq":1,"Pid":148940},"Data":"CgEAAOpWE6cmIAAAEAMEFbM+nWqBv4ehJgf4sEANDAoAAAAAAAAAgQAAAAAdWwAAAAAAAAAAAAAAAAAAAAAAAAAAAAC13zIBBQAIAAAAAAAFAAUAIAAAAAUABgAgAAAAFAABAAAAAAAAAAAAAAAAAAAAAAAoAAcAAAAAAICiBQAAAAAAALQAAAAAAAAAAAAAAAAAAAAAAAAAAAAArAACAAEAAAAAB3gBQIoDAECcAABEBQAAuAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAUCEAAAAAAAAgIQAAQCEAANwFAACsywIAJW8AAIRKAAD///9/CgAAAJQFAAADAAAALMkAAIBwAAAAAAAALnUOAAAAAAD///////////ayBAAAAAAASfQPAAAAAADMEQAANRMAAAAAAABiNQAAxAsAAGMIAABX5AUAAAAAAAoABABjdWJpYwAAAA=="}`
-	nm := syscall.NetlinkMessage{}
+	nm := netlink.NetlinkMessage{}
 	err := json.Unmarshal([]byte(json1), &nm)
 	if err != nil {
 		t.Fatal(err)

collector/collector_darwin.go

@@ -0,0 +1,15 @@
+package collector
+
+import (
+	"context"
+
+	"github.com/m-lab/tcp-info/netlink"
+	"github.com/m-lab/tcp-info/saver"
+)
+
+// No code, but needed for compiling.
+
+func Run(ctx context.Context, reps int, svrChan chan<- []*netlink.ArchivalRecord, cl saver.CacheLogger, skipLocal bool) (localCount, errCount int) {
+	// Does notihg in Darwin
+	return 0, 0
+}

collector/collector.go → collector/collector_linux.go

@@ -18,7 +18,7 @@ var (
 	localCount = 0
 )
 
-func appendAll(all []*netlink.ArchivalRecord, msgs []*syscall.NetlinkMessage, skipLocal bool) []*netlink.ArchivalRecord {
+func appendAll(all []*netlink.ArchivalRecord, msgs []*netlink.NetlinkMessage, skipLocal bool) []*netlink.ArchivalRecord {
 	// We use UTC, and truncate to millisecond to improve compression.
 	// Since the syscall to collect the data takes multiple milliseconds, this truncation seems reasonable.
 	ts := time.Now().UTC().Truncate(time.Millisecond)

collector/socket-monitor.go → collector/socket-monitor_linux.go

@@ -1,5 +1,7 @@
 package collector
 
+// This package is only meaningful in Linux.
+
 import (
 	"log"
 	"syscall"

inetdiag/inetdiag.go

@@ -1,5 +1,3 @@
-//go:generate ffjson $GOFILE
-
 // Package inetdiag provides basic structs and utilities for INET_DIAG messaages.
 // Based on uapi/linux/inet_diag.h.
 package inetdiag
@@ -31,10 +29,15 @@ expressed in host-byte order"
 */
 
 import (
-	"syscall"
+        "golang.org/x/sys/unix"
 )
 
 // inet_diag.h
+
+// NOTE: darwin unix.AF_INET6 and syscall.AF_INET6 are incorrect for
+// our purposes (0x1e), so, we set this explicitly.
+const AF_INET6 = 0x0a
+
 const (
 	INET_DIAG_NONE = iota
 	INET_DIAG_MEMINFO
@@ -82,8 +85,8 @@ var InetDiagType = map[int32]string{
 }
 
 var diagFamilyMap = map[uint8]string{
-	syscall.AF_INET:  "tcp",
-	syscall.AF_INET6: "tcp6",
+	unix.AF_INET: "tcp",
+	AF_INET6:     "tcp6", 	// because darwin values for AF_INET6 are incorrect.
 }
 
 // Protocol defines the type corresponding to INET_DIAG_PROTOCOL 8 bit field.

inetdiag/structs.go

@@ -28,7 +28,9 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"log"
 	"net"
+	"runtime"
 	"unsafe"
 )
 
@@ -218,6 +220,45 @@ type InetDiagMsg struct {
 	IDiagInode   uint32 `csv:"IDM.Inode"`
 }
 
+const (
+	// This previously came from syscall, but explicit here to work on Darwin.
+	RTA_ALIGNTO = 4
+)
+
+// rtaAlignOf rounds the length of a netlink route attribute up to align it properly.
+func rtaAlignOf(attrlen int) int {
+	return (attrlen + RTA_ALIGNTO - 1) & ^(RTA_ALIGNTO - 1)
+}
+
+// RawInetDiagMsg holds the []byte representation of an InetDiagMsg
+type RawInetDiagMsg []byte
+
+func SplitInetDiagMsg(data []byte) (RawInetDiagMsg, []byte) {
+	// TODO - why using rtaAlign on InetDiagMsg ???
+	align := rtaAlignOf(int(unsafe.Sizeof(InetDiagMsg{})))
+	if len(data) < align {
+		log.Println("Wrong length", len(data), "<", align)
+		_, file, line, _ := runtime.Caller(2)
+		log.Println(file, line, data)
+		return nil, nil
+	}
+	return RawInetDiagMsg(data[:align]), data[align:]
+}
+
+var ErrParseFailed = errors.New("Unable to parse InetDiagMsg")
+
+// Parse returns the InetDiagMsg itself
+// Modified from original to also return attribute data array.
+func (raw RawInetDiagMsg) Parse() (*InetDiagMsg, error) {
+	// TODO - why using rtaAlign on InetDiagMsg ???
+
+	align := rtaAlignOf(int(unsafe.Sizeof(InetDiagMsg{})))
+	if len(raw) < align {
+		return nil, ErrParseFailed
+	}
+	return (*InetDiagMsg)(unsafe.Pointer(&raw[0])), nil
+}
+
 // SocketMemInfo implements the struct associated with INET_DIAG_SKMEMINFO
 // Haven't found a corresponding linux struct, but the message is described
 // in https://manpages.debian.org/stretch/manpages/sock_diag.7.en.html

inetdiag/structs_test.go

@@ -2,10 +2,14 @@ package inetdiag
 
 import (
 	"bytes"
+	"fmt"
+	"syscall"
 	"testing"
+	"unsafe"
 
 	"github.com/gocarina/gocsv"
 	"github.com/m-lab/go/rtx"
+	"github.com/m-lab/tcp-info/tcp"
 )
 
 // This needs to be a whitebox test because it tests unexported types.
@@ -60,3 +64,100 @@ func TestStructAndCSVExport(t *testing.T) {
 		t.Error(sid.Cookie(), "!= 255")
 	}
 }
+
+func toString(id SockID) string {
+	return fmt.Sprintf("%s:%d -> %s:%d", id.SrcIP().String(), id.SPort(), id.DstIP().String(), id.DPort())
+}
+
+func TestParseInetDiagMsg(t *testing.T) {
+	var data [100]byte
+	for i := range data {
+		data[i] = byte(i + 2)
+	}
+	raw, value := SplitInetDiagMsg(data[:])
+	hdr, err := raw.Parse()
+	rtx.Must(err, "")
+
+	if hdr.ID.Interface() == 0 || hdr.ID.Cookie() == 0 || hdr.ID.DPort() == 0 || toString(hdr.ID) == "" {
+		t.Errorf("None of the accessed values should be zero")
+	}
+	if hdr.IDiagFamily != syscall.AF_INET {
+		t.Errorf("Failed %+v\n", hdr)
+	}
+	if tcp.State(hdr.IDiagState) != tcp.SYN_RECV {
+		t.Errorf("Failed %+v\n", hdr)
+	}
+
+	if len(value) != 28 {
+		t.Error("Len", len(value))
+	}
+
+	raw, value = SplitInetDiagMsg(data[:1])
+	if raw != nil || value != nil {
+		t.Error("This should fail, the data is too small.")
+	}
+}
+
+func TestID4(t *testing.T) {
+	var data [unsafe.Sizeof(InetDiagMsg{})]byte
+	srcIPOffset := unsafe.Offsetof(InetDiagMsg{}.ID) + unsafe.Offsetof(InetDiagMsg{}.ID.IDiagSrc)
+	data[srcIPOffset] = 127
+	data[srcIPOffset+1] = 0
+	data[srcIPOffset+2] = 0
+	data[srcIPOffset+3] = 1
+
+	srcPortOffset := unsafe.Offsetof(InetDiagMsg{}.ID) + unsafe.Offsetof(InetDiagMsg{}.ID.IDiagSPort)
+	// netlink uses host byte ordering, which may or may not be network byte ordering.  So no swapping should be
+	// done.
+	*(*uint16)(unsafe.Pointer(&data[srcPortOffset])) = 0x1234
+
+	dstIPOffset := unsafe.Offsetof(InetDiagMsg{}.ID) + unsafe.Offsetof(InetDiagMsg{}.ID.IDiagDst)
+	data[dstIPOffset] = 1
+	data[dstIPOffset+1] = 0
+	data[dstIPOffset+2] = 0
+	data[dstIPOffset+3] = 127 // Looks like localhost, but its reversed.
+
+	raw, _ := SplitInetDiagMsg(data[:])
+	hdr, err := raw.Parse()
+	rtx.Must(err, "")
+	if !hdr.ID.SrcIP().IsLoopback() {
+		t.Errorf("Should be loopback but isn't")
+	}
+	if hdr.ID.DstIP().IsLoopback() {
+		t.Errorf("Shouldn't be loopback but is")
+	}
+	if hdr.ID.SPort() != 0x3412 {
+		t.Errorf("SPort should be 0x3412 %+v\n", hdr.ID)
+	}
+
+	if !hdr.ID.SrcIP().IsLoopback() {
+		t.Errorf("Should be identified as loopback")
+	}
+	if hdr.ID.DstIP().IsLoopback() {
+		t.Errorf("Should not be identified as loopback") // Yeah I know this is not self-consistent. :P
+	}
+}
+
+func TestID6(t *testing.T) {
+	var data [unsafe.Sizeof(InetDiagMsg{})]byte
+	srcIPOffset := unsafe.Offsetof(InetDiagMsg{}.ID) + unsafe.Offsetof(InetDiagMsg{}.ID.IDiagSrc)
+	for i := 0; i < 8; i++ {
+		data[srcIPOffset] = byte(0x0A + i)
+	}
+
+	dstIPOffset := unsafe.Offsetof(InetDiagMsg{}.ID) + unsafe.Offsetof(InetDiagMsg{}.ID.IDiagDst)
+	for i := 0; i < 8; i++ {
+		data[dstIPOffset] = byte(i + 1)
+	}
+
+	raw, _ := SplitInetDiagMsg(data[:])
+	hdr, err := raw.Parse()
+	rtx.Must(err, "")
+
+	if hdr.ID.SrcIP().IsLoopback() {
+		t.Errorf("Should not be identified as loopback")
+	}
+	if hdr.ID.DstIP().IsLoopback() {
+		t.Errorf("Should not be identified as loopback")
+	}
+}

netlink/archival-record.go

@@ -9,13 +9,11 @@ import (
 	"io"
 	"log"
 	"net"
-	"syscall"
 	"time"
 	"unsafe"
 
-	"github.com/m-lab/tcp-info/tcp"
-
 	"github.com/m-lab/tcp-info/inetdiag"
+	"github.com/m-lab/tcp-info/tcp"
 )
 
 /*********************************************************************************************
@@ -37,7 +35,7 @@ type ArchivalRecord struct {
 
 	// Storing the RawIDM instead of the parsed InetDiagMsg reduces Marshalling by 2.6 usec, and
 	// typical compressed size by 3-4 bytes/record
-	RawIDM RawInetDiagMsg `json:",omitempty"` // RawInetDiagMsg within NLMsg
+	RawIDM inetdiag.RawInetDiagMsg `json:",omitempty"` // RawInetDiagMsg within NLMsg
 	// Saving just the .Value fields reduces Marshalling by 1.9 usec.
 	Attributes [][]byte `json:",omitempty"` // byte slices from RouteAttr.Value, backed by NLMsg
 
@@ -46,14 +44,30 @@ type ArchivalRecord struct {
 	Metadata *Metadata `json:",omitempty"`
 }
 
+// ParseRouteAttr parses a byte array into slice of NetlinkRouteAttr struct.
+// Derived from "github.com/vishvananda/netlink/nl/nl_linux.go"
+func ParseRouteAttr(b []byte) ([]NetlinkRouteAttr, error) {
+	var attrs []NetlinkRouteAttr
+	for len(b) >= SizeofRtAttr {
+		a, vbuf, alen, err := netlinkRouteAttrAndValue(b)
+		if err != nil {
+			return nil, err
+		}
+		ra := NetlinkRouteAttr{Attr: RtAttr(*a), Value: vbuf[:int(a.Len)-SizeofRtAttr]}
+		attrs = append(attrs, ra)
+		b = b[alen:]
+	}
+	return attrs, nil
+}
+
 // MakeArchivalRecord parses the NetlinkMessage into a ArchivalRecord.  If skipLocal is true, it will return nil for
 // loopback, local unicast, multicast, and unspecified connections.
 // Note that Parse does not populate the Timestamp field, so caller should do so.
-func MakeArchivalRecord(msg *syscall.NetlinkMessage, skipLocal bool) (*ArchivalRecord, error) {
+func MakeArchivalRecord(msg *NetlinkMessage, skipLocal bool) (*ArchivalRecord, error) {
 	if msg.Header.Type != 20 {
 		return nil, ErrNotType20
 	}
-	raw, attrBytes := splitInetDiagMsg(msg.Data)
+	raw, attrBytes := inetdiag.SplitInetDiagMsg(msg.Data)
 	if raw == nil {
 		return nil, ErrParseFailed
 	}
@@ -119,8 +133,8 @@ const (
 
 // Useful offsets for Compare
 const (
-	lastDataSentOffset = unsafe.Offsetof(syscall.TCPInfo{}.Last_data_sent)
-	pmtuOffset         = unsafe.Offsetof(syscall.TCPInfo{}.Pmtu)
+	lastDataSentOffset = unsafe.Offsetof(tcp.LinuxTCPInfo{}.LastDataSent)
+	pmtuOffset         = unsafe.Offsetof(tcp.LinuxTCPInfo{}.PMTU)
 	busytimeOffset     = unsafe.Offsetof(tcp.LinuxTCPInfo{}.BusyTime)
 )
 
@@ -233,8 +247,8 @@ func (pm *ArchivalRecord) Compare(previous *ArchivalRecord) (ChangeType, error)
 // LoadRawNetlinkMessage is a simple utility to read the next NetlinkMessage from a source reader,
 // e.g. from a file of naked binary netlink messages.
 // NOTE: This is a bit fragile if there are any bit errors in the message headers.
-func LoadRawNetlinkMessage(rdr io.Reader) (*syscall.NetlinkMessage, error) {
-	var header syscall.NlMsghdr
+func LoadRawNetlinkMessage(rdr io.Reader) (*NetlinkMessage, error) {
+	var header NlMsghdr
 	// TODO - should we pass in LittleEndian as a parameter?
 	err := binary.Read(rdr, binary.LittleEndian, &header)
 	if err != nil {
@@ -247,7 +261,7 @@ func LoadRawNetlinkMessage(rdr io.Reader) (*syscall.NetlinkMessage, error) {
 		return nil, err
 	}
 
-	return &syscall.NetlinkMessage{Header: header, Data: data}, nil
+	return &NetlinkMessage{Header: header, Data: data}, nil
 }
 
 // ArchiveReader produces ArchivedRecord structs from some source.