Sandbox gfr new jitter

active/active.go

@@ -101,7 +101,8 @@ type GCSSource struct {
 
 	pendingChan chan Runnable
 
-	label string
+	label    string
+	datatype string
 }
 
 // NewGCSSource creates a new source for active processing.
@@ -113,6 +114,7 @@ func NewGCSSource(ctx context.Context, job tracker.Job, fl FileLister, toRunnabl
 
 		pendingChan: make(chan Runnable, 0),
 		label:       job.Path(),
+		datatype:    job.Datatype,
 	}
 
 	go src.streamToPending(ctx, job)
@@ -125,6 +127,10 @@ func (src *GCSSource) Label() string {
 	return src.label
 }
 
+func (src *GCSSource) Datatype() string {
+	return src.datatype
+}
+
 // CancelStreaming terminates the streaming goroutine context.
 func (src *GCSSource) CancelStreaming() {
 	src.ctx.cancel()

active/poller.go

@@ -119,8 +119,8 @@ func (g *GardenerAPI) RunAll(ctx context.Context, rSrc RunnableSource, job track
 		debug.Println("Starting func")
 
 		f := func() (err error) {
-			metrics.ActiveTasks.WithLabelValues(rSrc.Label()).Inc()
-			defer metrics.ActiveTasks.WithLabelValues(rSrc.Label()).Dec()
+			metrics.ActiveTasks.WithLabelValues(rSrc.Datatype()).Inc()
+			defer metrics.ActiveTasks.WithLabelValues(rSrc.Datatype()).Dec()
 
 			// Capture any panic and convert it to an error.
 			defer func(tag string) {

active/runnable.go

@@ -23,4 +23,7 @@ type RunnableSource interface {
 
 	// Label returns a string for use in metrics and debug logs'
 	Label() string
+
+	// Datatype returns the datatype for use in metrics
+	Datatype() string
 }

active/throttle_test.go

@@ -73,6 +73,9 @@ func (s *source) Next(ctx context.Context) (active.Runnable, error) {
 func (s *source) Label() string {
 	return "label"
 }
+func (s *source) Datatype() string {
+	return "datatype"
+}
 
 func (sr *statsRunnable) Run(ctx context.Context) error {
 	now := sr.stats.add()

cmd/update-schema/update.go

@@ -30,6 +30,7 @@ import (
 
 // CreateOrUpdateTCPInfo will update existing TCPInfo table, or create new table if update fails.
 func CreateOrUpdateTCPInfo(project string, dataset string, table string) error {
+	return nil // HACK to avoid collision with other sandbox work.
 	row := schema.TCPRow{}
 	schema, err := row.Schema()
 	rtx.Must(err, "TCPRow.Schema")
@@ -40,6 +41,7 @@ func CreateOrUpdateTCPInfo(project string, dataset string, table string) error {
 }
 
 func CreateOrUpdatePT(project string, dataset string, table string) error {
+	return nil // HACK to avoid collision with other sandbox work.
 	row := schema.PTTest{}
 	schema, err := row.Schema()
 	rtx.Must(err, "PTTest.Schema")
@@ -108,6 +110,7 @@ func CreateOrUpdateSwitchStats(project string, dataset string, table string) err
 }
 
 func CreateOrUpdatePCAPRow(project string, dataset string, table string) error {
+	return nil // HACK to avoid collision with other sandbox work.
 	row := schema.PCAPRow{}
 	schema, err := row.Schema()
 	rtx.Must(err, "PCAPRow.Schema")

parser/pcap.go

@@ -3,22 +3,19 @@ package parser
 import (
 	"fmt"
 	"log"
-	"net"
 	"os"
 	"path/filepath"
 	"strings"
 	"time"
 
 	"cloud.google.com/go/bigquery"
 	"cloud.google.com/go/civil"
-	"github.com/google/gopacket"
-	"github.com/google/gopacket/layers"
-	"github.com/google/gopacket/pcapgo"
 	v2as "github.com/m-lab/annotation-service/api/v2"
 	"github.com/m-lab/etl/etl"
 	"github.com/m-lab/etl/metrics"
 	"github.com/m-lab/etl/row"
 	"github.com/m-lab/etl/schema"
+	"github.com/m-lab/etl/tcpip"
 	"github.com/m-lab/go/logx"
 )
 
@@ -30,87 +27,6 @@ var (
 	ErrNoIPLayer = fmt.Errorf("no IP layer")
 )
 
-// Packet struct contains the packet data and metadata.
-type Packet struct {
-	// If we use a pointer here, for some reason we get zero value timestamps.
-	Ci   gopacket.CaptureInfo
-	Data []byte
-	Err  error
-}
-
-// GetIP decodes the IP layers and returns some basic information.
-// It is a bit slow and does memory allocation.
-func (p *Packet) GetIP() (net.IP, net.IP, uint8, uint16, error) {
-	// Decode a packet.
-	pkt := gopacket.NewPacket(p.Data, layers.LayerTypeEthernet, gopacket.DecodeOptions{
-		Lazy:                     true,
-		NoCopy:                   true,
-		SkipDecodeRecovery:       true,
-		DecodeStreamsAsDatagrams: false,
-	})
-
-	if ipLayer := pkt.Layer(layers.LayerTypeIPv4); ipLayer != nil {
-		ip, _ := ipLayer.(*layers.IPv4)
-		// For IPv4, the TTL length is the ip.Length adjusted for the header length.
-		return ip.SrcIP, ip.DstIP, ip.TTL, ip.Length - uint16(4*ip.IHL), nil
-	} else if ipLayer := pkt.Layer(layers.LayerTypeIPv6); ipLayer != nil {
-		ip, _ := ipLayer.(*layers.IPv6)
-		// In IPv6, the Length field is the payload length.
-		return ip.SrcIP, ip.DstIP, ip.HopLimit, ip.Length, nil
-	} else {
-		return nil, nil, 0, 0, ErrNoIPLayer
-	}
-}
-
-func GetPackets(data []byte) ([]Packet, error) {
-	pcap, err := pcapgo.NewReader(strings.NewReader(string(data)))
-	if err != nil {
-		log.Print(err)
-		return nil, err
-	}
-
-	// TODO: len(data)/18 provides much better estimate of number of packets.
-	// len(data)/18 was determined by looking at bytes/packet in a few pcaps files.
-	// The number seems too small, but perhaps the data is still compressed at this point.
-	// However, it seems to cause mysterious crashes in sandbox, so
-	// reverting to /1500 for now.
-	packets := make([]Packet, 0, len(data)/1500)
-
-	for data, ci, err := pcap.ZeroCopyReadPacketData(); err == nil; data, ci, err = pcap.ReadPacketData() {
-		packets = append(packets, Packet{Ci: ci, Data: data, Err: err})
-	}
-
-	if err != nil {
-		metrics.WarningCount.WithLabelValues("pcap", "ip_layer_failure").Inc()
-		metrics.PcapPacketCount.WithLabelValues("IP error").Observe(float64(len(packets)))
-		return packets, err
-	} else if len(packets) > 0 {
-		srcIP, _, _, _, err := packets[0].GetIP()
-		// TODO - eventually we should identify key local ports, like 443 and 3001.
-		if err != nil {
-			metrics.WarningCount.WithLabelValues("pcap", "ip_layer_failure").Inc()
-			metrics.PcapPacketCount.WithLabelValues("IP error").Observe(float64(len(packets)))
-		} else {
-			start := packets[0].Ci.Timestamp
-			end := packets[len(packets)-1].Ci.Timestamp
-			duration := end.Sub(start)
-			// TODO add TCP layer, so we can label the stats based on local port value.
-			if len(srcIP) == 4 {
-				metrics.PcapPacketCount.WithLabelValues("ipv4").Observe(float64(len(packets)))
-				metrics.PcapConnectionDuration.WithLabelValues("ipv4").Observe(duration.Seconds())
-			} else {
-				metrics.PcapPacketCount.WithLabelValues("ipv6").Observe(float64(len(packets)))
-				metrics.PcapConnectionDuration.WithLabelValues("ipv6").Observe(duration.Seconds())
-			}
-		}
-	} else {
-		// No packets.
-		metrics.PcapPacketCount.WithLabelValues("unknown").Observe(float64(len(packets)))
-	}
-
-	return packets, nil
-}
-
 //=====================================================================================
 //                       PCAP Parser
 //=====================================================================================
@@ -172,7 +88,7 @@ func (p *PCAPParser) ParseAndInsert(fileMetadata map[string]bigquery.Value, test
 
 	// Parse top level PCAP data and update metrics.
 	// TODO - add schema fields here.
-	_, _ = GetPackets(rawContent)
+	_, _ = tcpip.ProcessPackets(row.Parser.ArchiveURL, testName, rawContent)
 
 	// Insert the row.
 	if err := p.Put(&row); err != nil {

parser/pcap_test.go

@@ -1,13 +1,10 @@
 package parser_test
 
 import (
-	"io"
 	"io/ioutil"
-	"os"
 	"path"
 	"strings"
 	"testing"
-	"time"
 
 	"cloud.google.com/go/bigquery"
 	"cloud.google.com/go/civil"
@@ -134,111 +131,3 @@ func TestPCAPParser_GetUUID(t *testing.T) {
 		})
 	}
 }
-
-func TestIPLayer(t *testing.T) {
-	type test struct {
-		name         string
-		fn           string
-		packets      int64
-		duration     time.Duration
-		srcIP, dstIP string
-		TTL          uint8
-	}
-	tests := []test{
-		{name: "retransmits", fn: "testdata/PCAP/ndt-nnwk2_1611335823_00000000000C2DFE.pcap.gz",
-			packets: 336, duration: 15409174000, srcIP: "173.49.19.128"},
-		{name: "ipv6", fn: "testdata/PCAP/ndt-nnwk2_1611335823_00000000000C2DA8.pcap.gz",
-			packets: 15, duration: 134434000, srcIP: "2a0d:5600:24:a71::1d"},
-		{name: "protocolErrors2", fn: "testdata/PCAP/ndt-nnwk2_1611335823_00000000000C2DA9.pcap.gz",
-			packets: 5180, duration: 13444117000, srcIP: "2a0d:5600:24:a71::1d"},
-	}
-	for _, tt := range tests {
-		f, err := os.Open(tt.fn)
-		if err != nil {
-			t.Fatal(err)
-		}
-		data, err := ioutil.ReadAll(f)
-		if err != nil {
-			t.Fatal(err)
-		}
-		packets, err := parser.GetPackets(data)
-		if err != nil {
-			t.Fatal(err)
-		}
-		start := packets[0].Ci.Timestamp
-		end := packets[len(packets)-1].Ci.Timestamp
-		duration := end.Sub(start)
-		if duration != tt.duration {
-			t.Errorf("%s: duration = %v, want %v", tt.name, duration, tt.duration)
-		}
-		if len(packets) != int(tt.packets) {
-			t.Errorf("%s: expected %d packets, got %d", tt.name, tt.packets, len(packets))
-		}
-		srcIP, _, _, _, err := packets[0].GetIP()
-		if err != nil {
-			t.Fatal(err)
-		}
-		if srcIP.String() != tt.srcIP {
-			t.Errorf("%s: expected srcIP %s, got %s", tt.name, tt.srcIP, srcIP.String())
-		}
-	}
-}
-
-func TestPCAPGarbage(t *testing.T) {
-	data := []byte{0xd4, 0xc3, 0xb2, 0xa1, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12}
-	_, err := parser.GetPackets(data)
-	if err != io.ErrUnexpectedEOF {
-		t.Fatal(err)
-	}
-
-	data = append(data, data...)
-	_, err = parser.GetPackets(data)
-	if err == nil || !strings.Contains(err.Error(), "Unknown major") {
-		t.Fatal(err)
-	}
-}
-
-func getTestFile(b *testing.B, name string) []byte {
-	f, err := os.Open(path.Join(`testdata/PCAP/`, name))
-	if err != nil {
-		b.Fatal(err)
-	}
-	data, err := ioutil.ReadAll(f)
-	if err != nil {
-		b.Fatal(err)
-	}
-	return data
-}
-
-// Original single file RunParallel:
-// Just packet decoding: BenchmarkGetPackets-8   	    8678	    128426 ns/op	  165146 B/op	     381 allocs/op
-// With IP decoding:     BenchmarkGetPackets-8   	    4279	    285547 ns/op	  376125 B/op	    1729 allocs/op
-
-// Enhanced RunParallel: BenchmarkGetPackets-8   	    2311	    514898 ns/op	 1181138 B/op	    1886 allocs/op
-func BenchmarkGetPackets(b *testing.B) {
-	type tt struct {
-		data    []byte
-		numPkts int
-	}
-	tests := []tt{
-		{getTestFile(b, "ndt-nnwk2_1611335823_00000000000C2DFE.pcap.gz"), 336},
-		{getTestFile(b, "ndt-nnwk2_1611335823_00000000000C2DA8.pcap.gz"), 15},
-		{getTestFile(b, "ndt-nnwk2_1611335823_00000000000C2DA9.pcap.gz"), 5180},
-	}
-	b.ResetTimer()
-
-	i := 0
-	b.RunParallel(func(pb *testing.PB) {
-		for pb.Next() {
-			test := tests[i%len(tests)]
-			i++
-			pkts, err := parser.GetPackets(test.data)
-			if err != nil {
-				b.Fatal(err)
-			}
-			if len(pkts) != test.numPkts {
-				b.Errorf("expected %d packets, got %d", test.numPkts, len(pkts))
-			}
-		}
-	})
-}

schema/pcap.go

@@ -1,17 +1,62 @@
 package schema
 
 import (
+	"time"
+
 	"cloud.google.com/go/bigquery"
 	"cloud.google.com/go/civil"
 	"github.com/m-lab/etl/row"
 	"github.com/m-lab/go/cloud/bqx"
 )
 
+type TcpStats struct {
+	Packets   int64
+	Truncated int64
+
+	OptionCounts []int64 // 16 counts, indicating how often each option type occurred.
+
+	RetransmitPackets int64
+	RetransmitBytes   int64
+
+	Sacks int64
+
+	ECECount      int64
+	WindowChanges int64
+
+	// Errors and anomalies
+	BadSacks              int64 // Number of sacks with bad boundaries
+	BadDeltas             int64 // Number of seqs and acks that were more than 1<<30 off from previous value.
+	MissingPackets        int64 // Observations of packet sequence numbers that didn't match previous payload length.
+	SendNextExceededLimit int64 // Number of times SendNext() returned a value that exceeded the receiver window limit.
+	TTLChanges            int64 // Observed number of TTL values that don't match first IP header.
+	SrcPortErrors         int64 // Observed number of source ports that don't match first IP header.
+	DstPortErrors         int64 // Observed number of dest ports that don't match tcp.DstPort
+	OtherErrors           int64 // Number of other errors that occurred.
+
+}
+
+type AlphaFields struct {
+	TruncatedPackets int64     `bigquery:"truncated_packets"`
+	SynPacket        int64     `bigquery:"syn_packet" json:"syn_packet"`
+	SynTime          time.Time `bigquery:"syn_time" json:"syn_time"`
+	SynAckPacket     int64     `bigquery:"syn_ack_packet" json:"syn_ack_packet"`
+	SynAckTime       time.Time `bigquery:"syn_ack_time" json:"syn_ack_time"`
+	Packets          int64     `bigquery:"packets" json:"packets"`
+	Sacks            int64     `bigquery:"sacks" json:"sacks"`
+	IPAddrErrors     int64     `bigquery:"ip_addr_errors" json:"ip_addr_errors"` // Number of packets with IP addresses that don't match first IP header at all.
+	WithoutTCPLayer  int64     `bigquery:"no_tcp_layer" json:"no_tcp_layer"`     // Number of packets with no TCP layer.
+
+	LeftStats  TcpStats
+	RightStats TcpStats
+}
+
 // PCAPRow describes a single BQ row of pcap (packet capture) data.
 type PCAPRow struct {
-	ID     string     `bigquery:"id"`
-	Parser ParseInfo  `bigquery:"parser"`
-	Date   civil.Date `bigquery:"date"`
+	ID     string     `bigquery:"id" json:"id"`
+	Parser ParseInfo  `bigquery:"parser" json:"parser"`
+	Date   civil.Date `bigquery:"date" json:"date"`
+
+	// Alpha *AlphaFields `bigquery:"alpha" json:"alpha"`
 
 	// NOT part of struct schema. Included only to provide a fake annotator interface.
 	row.NullAnnotator `bigquery:"-"`