Add timestamp to protobuf

inetdiag/inetdiag.go

@@ -236,6 +236,7 @@ type ParsedMessage struct {
 
 // Parse parsed the NetlinkMessage into a ParsedMessage.  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 Parse(msg *syscall.NetlinkMessage, skipLocal bool) (*ParsedMessage, error) {
 	if msg.Header.Type != 20 {
 		return nil, ErrNotType20

nl-proto/tcpinfo.proto

@@ -306,4 +306,7 @@ message TCPDiagnosticsProto {
   oneof shutdown {
     uint32 shutdown_mask = 10;
   }
+
+  // Timestamp of batch of messages containing this message.
+  int64 timestamp = 11;
 }

nl-proto/tools/convert.go

@@ -7,6 +7,7 @@ import (
 	"bytes"
 	"log"
 	"syscall"
+	"time"
 	"unsafe"
 
 	"github.com/m-lab/tcp-info/inetdiag"
@@ -91,7 +92,7 @@ func AttrToField(all *tcpinfo.TCPDiagnosticsProto, rta *syscall.NetlinkRouteAttr
 
 // CreateProto creates a fully populated TCPDiagnosticsProto from the parsed elements of a type 20 netlink message.
 // This assumes the netlink message is type 20, and behavior is undefined if it is not.
-func CreateProto(header syscall.NlMsghdr, idm *inetdiag.InetDiagMsg, attrs []*syscall.NetlinkRouteAttr) *tcpinfo.TCPDiagnosticsProto {
+func CreateProto(time time.Time, header syscall.NlMsghdr, idm *inetdiag.InetDiagMsg, attrs []*syscall.NetlinkRouteAttr) *tcpinfo.TCPDiagnosticsProto {
 	all := tcpinfo.TCPDiagnosticsProto{}
 	all.InetDiagMsg = HeaderToProto(idm)
 	for i := range attrs {
@@ -100,6 +101,7 @@ func CreateProto(header syscall.NlMsghdr, idm *inetdiag.InetDiagMsg, attrs []*sy
 		}
 	}
 
+	all.Timestamp = time.UnixNano()
 	return &all
 }
 

nl-proto/tools/convert_test.go

@@ -8,6 +8,7 @@ import (
 	"log"
 	"syscall"
 	"testing"
+	"time"
 
 	"github.com/go-test/deep"
 	"github.com/m-lab/tcp-info/inetdiag"
@@ -49,7 +50,7 @@ func convertToProto(msg *syscall.NetlinkMessage, t *testing.T) *tcpinfo.TCPDiagn
 	if err != nil {
 		t.Fatal(err)
 	}
-	return tools.CreateProto(msg.Header, parsedMsg.InetDiagMsg, parsedMsg.Attributes[:])
+	return tools.CreateProto(time.Now(), msg.Header, parsedMsg.InetDiagMsg, parsedMsg.Attributes[:])
 }
 
 func TestReader(t *testing.T) {

play.go

@@ -74,7 +74,7 @@ func Marshal(filename string, marshaler chan *inetdiag.ParsedMessage, wg *sync.W
 		if !ok {
 			break
 		}
-		p := tools.CreateProto(msg.Header, msg.InetDiagMsg, msg.Attributes[:])
+		p := tools.CreateProto(msg.Timestamp, msg.Header, msg.InetDiagMsg, msg.Attributes[:])
 		if false {
 			log.Printf("%+v\n", p.InetDiagMsg)
 			log.Printf("%+v\n", p.TcpInfo)

saver/saver.go

@@ -61,7 +61,7 @@ func runMarshaller(taskChan <-chan Task, wg *sync.WaitGroup) {
 			log.Fatal("Nil writer")
 		}
 		msg := task.Message
-		pb := tools.CreateProto(msg.Header, msg.InetDiagMsg, msg.Attributes[:])
+		pb := tools.CreateProto(msg.Timestamp, msg.Header, msg.InetDiagMsg, msg.Attributes[:])
 		wire, err := proto.Marshal(pb)
 		if err != nil {
 			log.Println(err)