Merge 72b3d21 into 756c71a

cache/cache.go

@@ -32,14 +32,18 @@ func NewCache() *Cache {
 }
 
 // Update swaps msg with the cache contents, and returns the evicted value.
-func (c *Cache) Update(msg *inetdiag.ParsedMessage) *inetdiag.ParsedMessage {
-	cookie := msg.InetDiagMsg.ID.Cookie()
+func (c *Cache) Update(msg *inetdiag.ParsedMessage) (*inetdiag.ParsedMessage, error) {
+	idm, err := msg.RawIDM.Parse()
+	if err != nil {
+		return nil, err
+	}
+	cookie := idm.ID.Cookie()
 	c.current[cookie] = msg
 	evicted, ok := c.previous[cookie]
 	if ok {
 		delete(c.previous, cookie)
 	}
-	return evicted
+	return evicted, nil
 }
 
 // EndCycle marks the completion of updates from one set of netlink messages.

cache/cache_test.go

@@ -1,29 +1,61 @@
 package cache_test
 
 import (
-	"encoding/binary"
+	"encoding/json"
+	"log"
+	"syscall"
 	"testing"
-	"time"
 
 	"github.com/m-lab/tcp-info/cache"
 	"github.com/m-lab/tcp-info/inetdiag"
 )
 
-func fakeMsg(cookie uint64) inetdiag.ParsedMessage {
-	pm := inetdiag.ParsedMessage{Timestamp: time.Now(), InetDiagMsg: &inetdiag.InetDiagMsg{}}
-	binary.BigEndian.PutUint64(pm.InetDiagMsg.ID.IDiagCookie[:], cookie)
-	return pm
+func init() {
+	// Always prepend the filename and line number.
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+}
+
+func testFatal(t *testing.T, err error) {
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func fakeMsg(t *testing.T, cookie uint64, dport uint16) inetdiag.ParsedMessage {
+	var json1 = `{"Header":{"Len":356,"Type":20,"Flags":2,"Seq":1,"Pid":148940},"Data":"CgEAAOpWE6cmIAAAEAMEFbM+nWqBv4ehJgf4sEANDAoAAAAAAAAAgQAAAAAdWwAAAAAAAAAAAAAAAAAAAAAAAAAAAAC13zIBBQAIAAAAAAAFAAUAIAAAAAUABgAgAAAAFAABAAAAAAAAAAAAAAAAAAAAAAAoAAcAAAAAAICiBQAAAAAAALQAAAAAAAAAAAAAAAAAAAAAAAAAAAAArAACAAEAAAAAB3gBQIoDAECcAABEBQAAuAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAUCEAAAAAAAAgIQAAQCEAANwFAACsywIAJW8AAIRKAAD///9/CgAAAJQFAAADAAAALMkAAIBwAAAAAAAALnUOAAAAAAD///////////ayBAAAAAAASfQPAAAAAADMEQAANRMAAAAAAABiNQAAxAsAAGMIAABX5AUAAAAAAAoABABjdWJpYwAAAA=="}`
+	nm := syscall.NetlinkMessage{}
+	err := json.Unmarshal([]byte(json1), &nm)
+	if err != nil {
+		t.Fatal(err)
+	}
+	mp, err := inetdiag.Parse(&nm, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	idm, err := mp.RawIDM.Parse()
+	for i := 0; i < 8; i++ {
+		idm.ID.IDiagCookie[i] = byte(cookie & 0x0FF)
+		cookie >>= 8
+	}
+	for i := 0; i < 2; i++ {
+		idm.ID.IDiagDPort[i] = byte(dport & 0x0FF)
+		dport >>= 8
+	}
+	log.Printf("Cookie: %x\n", idm.ID.Cookie())
+	return *mp
 }
 
 func TestUpdate(t *testing.T) {
 	c := cache.NewCache()
-	pm1 := fakeMsg(1234)
-	old := c.Update(&pm1)
+	pm1 := fakeMsg(t, 0x1234, 1)
+	old, err := c.Update(&pm1)
+	testFatal(t, err)
 	if old != nil {
 		t.Error("old should be nil")
 	}
-	pm2 := fakeMsg(4321)
-	old = c.Update(&pm2)
+	pm2 := fakeMsg(t, 4321, 1)
+	old, err = c.Update(&pm2)
+	testFatal(t, err)
 	if old != nil {
 		t.Error("old should be nil")
 	}
@@ -36,8 +68,9 @@ func TestUpdate(t *testing.T) {
 		t.Error("Should be empty")
 	}
 
-	pm3 := fakeMsg(4321)
-	old = c.Update(&pm3)
+	pm3 := fakeMsg(t, 4321, 1)
+	old, err = c.Update(&pm3)
+	testFatal(t, err)
 	if old == nil {
 		t.Error("old should NOT be nil")
 	}
@@ -47,8 +80,8 @@ func TestUpdate(t *testing.T) {
 		t.Error("Should not be empty", len(leftover))
 	}
 	for k := range leftover {
-		if *leftover[k] != pm1 {
-			t.Error("Should have found pm1")
+		if k != 0x1234 {
+			t.Errorf("Should have found pm1 %x\n", k)
 		}
 	}
 	if c.CycleCount() != 2 {

collector/collector.go

@@ -10,13 +10,20 @@ import (
 	"github.com/m-lab/tcp-info/saver"
 )
 
+func init() {
+	// Always prepend the filename and line number.
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+}
+
 var (
 	errCount   = 0
 	localCount = 0
 )
 
 func appendAll(all []*inetdiag.ParsedMessage, msgs []*syscall.NetlinkMessage, skipLocal bool) []*inetdiag.ParsedMessage {
-	ts := time.Now()
+	// 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)
 	for i := range msgs {
 		pm, err := inetdiag.Parse(msgs[i], skipLocal)
 		if err != nil {

collector/collector_test.go

@@ -13,6 +13,17 @@ import (
 	"github.com/m-lab/tcp-info/inetdiag"
 )
 
+func init() {
+	// Always prepend the filename and line number.
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+}
+
+func testFatal(t *testing.T, err error) {
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
 type testCacheLogger struct{}
 
 func (t *testCacheLogger) LogCacheStats(_, _ int) {}
@@ -91,8 +102,13 @@ func TestRun(t *testing.T) {
 			if m == nil {
 				continue
 			}
-			if m.InetDiagMsg != nil && m.InetDiagMsg.ID.SPort() == uint16(port) {
-				if prev == nil || prev.Compare(m) > inetdiag.NoMajorChange {
+			idm, err := m.RawIDM.Parse()
+			testFatal(t, err)
+			if idm != nil && idm.ID.SPort() == uint16(port) {
+				change, err := m.Compare(prev)
+				if err != nil {
+					log.Println(err)
+				} else if change > inetdiag.NoMajorChange {
 					prev = m
 					changed = true
 				}

inetdiag/export_test.go

@@ -0,0 +1,3 @@
+package inetdiag
+
+var SplitInetDiagMsg = splitInetDiagMsg