First step in expanding pcap parser (#1031)

* add panic handling for active polling mode

* add top level PCAP packet parsing

* simplify very noisy FileCount metric label

* add packet decoding benchmark, reorg pcap.go, add metrics

* improve test coverage

* improve slice allocation

* go mod tidy

* various PR comment fixes

* revert to len(data)/1500

Author: gfr10598

active/poller.go

@@ -16,6 +16,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"golang.org/x/sync/errgroup"
+	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
 
 	job "github.com/m-lab/etl-gardener/client"
@@ -95,13 +96,19 @@ func postAndIgnoreResponse(ctx context.Context, url url.URL) error {
 // or the context is canceled.
 func (g *GardenerAPI) RunAll(ctx context.Context, rSrc RunnableSource, job tracker.Job) (*errgroup.Group, error) {
 	eg := &errgroup.Group{}
+	count := 0
 	for {
 		run, err := rSrc.Next(ctx)
 		if err != nil {
-			metrics.BackendFailureCount.WithLabelValues(
-				job.Datatype, "rSrc.Next").Inc()
-			log.Println(err)
-			return eg, err
+			if err == iterator.Done {
+				debug.Printf("Dispatched total of %d archives for %s\n", count, job.String())
+				return eg, nil
+			} else {
+				metrics.BackendFailureCount.WithLabelValues(
+					job.Datatype, "rSrc.Next").Inc()
+				log.Println(err, "processing", job.String())
+				return eg, err
+			}
 		}
 
 		heartbeat := tracker.HeartbeatURL(g.trackerBase, job)
@@ -111,20 +118,28 @@ func (g *GardenerAPI) RunAll(ctx context.Context, rSrc RunnableSource, job track
 
 		debug.Println("Starting func")
 
-		f := func() error {
+		f := func() (err error) {
 			metrics.ActiveTasks.WithLabelValues(rSrc.Label()).Inc()
 			defer metrics.ActiveTasks.WithLabelValues(rSrc.Label()).Dec()
 
-			err := run.Run(ctx)
+			// Capture any panic and convert it to an error.
+			defer func(tag string) {
+				if err2 := metrics.PanicToErr(err, recover(), "Runall.f: "+tag); err2 != nil {
+					err = err2
+				}
+			}(run.Info())
+
+			err = run.Run(ctx)
 			if err == nil {
 				update := tracker.UpdateURL(g.trackerBase, job, tracker.Parsing, run.Info())
 				if postErr := postAndIgnoreResponse(ctx, *update); postErr != nil {
 					log.Println(postErr, "on update for", job.Path())
 				}
 			}
-			return err
+			return
 		}
 
+		count++
 		eg.Go(f)
 	}
 }
@@ -171,12 +186,14 @@ func (g *GardenerAPI) pollAndRun(ctx context.Context,
 	toRunnable func(o *storage.ObjectAttrs) Runnable, tokens TokenSource) error {
 	job, err := g.NextJob(ctx)
 	if err != nil {
+		log.Println(err, "on Gardener client.NextJob()")
 		return err
 	}
 
 	log.Println(job, "filter:", job.Filter)
 	gcsSource, err := g.JobFileSource(ctx, job.Job, toRunnable)
 	if err != nil {
+		log.Println(err, "on JobFileSource")
 		return err
 	}
 	src := Throttle(gcsSource, tokens)
@@ -189,13 +206,20 @@ func (g *GardenerAPI) pollAndRun(ctx context.Context,
 	}
 
 	eg, err := g.RunAll(ctx, src, job.Job)
+	if err != nil {
+		log.Println(err)
+	}
 
 	// Once all are dispatched, we want to wait until all have completed
 	// before posting the state change.
 	go func() {
 		log.Println("all tasks dispatched for", job.Path())
-		eg.Wait()
-		log.Println("finished", job.Path())
+		err := eg.Wait()
+		if err != nil {
+			log.Println(err, "on wait for", job.Path())
+		} else {
+			log.Println("finished", job.Path())
+		}
 		update := tracker.UpdateURL(g.trackerBase, job.Job, tracker.ParseComplete, "")
 		// TODO - should this have a retry?
 		if postErr := postAndIgnoreResponse(ctx, *update); postErr != nil {

active/poller_test.go

@@ -12,10 +12,18 @@ import (
 
 	"github.com/m-lab/etl-gardener/tracker"
 	"github.com/m-lab/etl/active"
+	"github.com/m-lab/go/logx"
 	"github.com/m-lab/go/rtx"
 	"google.golang.org/api/iterator"
 )
 
+func init() {
+	// Always prepend the filename and line number.
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+	logx.LogxDebug.Set("true")
+	logx.Debug.SetFlags(log.LstdFlags | log.Lshortfile)
+}
+
 type fakeGardener struct {
 	t *testing.T // for logging
 
@@ -129,7 +137,7 @@ func TestGardenerAPI_RunAll(t *testing.T) {
 	src, err := g.JobFileSource(ctx, job.Job, p.toRunnable)
 	rtx.Must(err, "file source")
 	eg, err := g.RunAll(ctx, src, job.Job)
-	if err != iterator.Done {
+	if err != nil {
 		t.Fatal(err)
 	}
 	err = eg.Wait()

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/go-test/deep v1.0.7
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/google/go-jsonnet v0.17.0
+	github.com/google/gopacket v1.1.19
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/googleapis/google-cloud-go-testing v0.0.0-20210719221736-1c9a4c676720
 	github.com/kr/pretty v0.2.1

go.sum

@@ -184,6 +184,8 @@ github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-jsonnet v0.17.0 h1:/9NIEfhK1NQRKl3sP2536b2+x5HnZMdql7x3yK/l8JY=
 github.com/google/go-jsonnet v0.17.0/go.mod h1:sOcuej3UW1vpPTZOr8L7RQimqai1a57bt5j22LzGZCw=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
+github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
 github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=

metrics/metrics.go

@@ -582,6 +582,36 @@ var (
 		},
 		[]string{"table", "kind", "group"},
 	)
+
+	PcapPacketCount = promauto.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Name: "etl_pcap_packet_count",
+			Help: "Distribution of PCAP packet counts",
+			Buckets: []float64{
+				1, 2, 3, 5,
+				10, 18, 32, 56,
+				100, 178, 316, 562,
+				1000, 1780, 3160, 5620,
+				10000, 17800, 31600, 56200, math.Inf(1),
+			},
+		},
+		[]string{"port"},
+	)
+
+	PcapConnectionDuration = promauto.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Name: "etl_pcap_connection_duration",
+			Help: "Distribution of PCAP connection duration",
+			Buckets: []float64{
+				.1, .2, .3, .5,
+				1, 1.8, 3.2, 5.6,
+				10, 18, 32, 56,
+				100, 178, 316, 562,
+				1000, 1780, 3160, 5620, math.Inf(1),
+			},
+		},
+		[]string{"port"},
+	)
 )
 
 // catchStatus wraps the native http.ResponseWriter and captures any written HTTP

parser/pcap.go

@@ -1,19 +1,116 @@
 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/go/logx"
 )
 
+var (
+	info         = log.New(os.Stdout, "info: ", log.LstdFlags|log.Lshortfile)
+	sparseLogger = log.New(os.Stdout, "sparse: ", log.LstdFlags|log.Lshortfile)
+	sparse20     = logx.NewLogEvery(sparseLogger, 50*time.Millisecond)
+
+	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
 //=====================================================================================
@@ -73,6 +170,10 @@ func (p *PCAPParser) ParseAndInsert(fileMetadata map[string]bigquery.Value, test
 	row.Date = fileMetadata["date"].(civil.Date)
 	row.ID = p.GetUUID(testName)
 
+	// Parse top level PCAP data and update metrics.
+	// TODO - add schema fields here.
+	_, _ = GetPackets(rawContent)
+
 	// Insert the row.
 	if err := p.Put(&row); err != nil {
 		return err
@@ -84,6 +185,10 @@ func (p *PCAPParser) ParseAndInsert(fileMetadata map[string]bigquery.Value, test
 	return nil
 }
 
+//=====================================================================================
+//                       Implementation of the etl.Parser interface
+//=====================================================================================
+
 // GetUUID extracts the UUID from the filename.
 // For example, for filename 2021/07/22/ndt-4c6fb_1625899199_00000000013A4623.pcap.gz,
 // it returns ndt-4c6fb_1625899199_00000000013A4623.