[POA-4078] Add agent metrics (#134)

Adds metrics to the agent letting us know:
- How long is the buffer of pcap capture packets
- How long is the buffer of unprocessed packets
- How long is the buffer of parsed network traffic
- How long is the buffer of unpaired partial witnesses

Author: cesher

pcap/net_parse.go

@@ -140,6 +140,9 @@ func (p *NetworkTrafficParser) ParseFromInterface(
 		// Signal caller that we're done on exit
 		defer close(out)
 
+		startTime := time.Now()
+		bufferTimeSum := 0 * time.Second
+		intervalLength := 1 * time.Minute
 		for {
 			select {
 			// packets channel is going to read until EOF or when signalClose is
@@ -158,6 +161,16 @@ func (p *NetworkTrafficParser) ParseFromInterface(
 
 					return
 				}
+
+				now := time.Now()
+				if now.Sub(startTime) >= intervalLength {
+					bufferLength := float64(bufferTimeSum.Nanoseconds()) / float64(intervalLength.Nanoseconds())
+					printer.Debugf("Approximate unprocessed-packets buffer length: %v", bufferLength)
+					bufferTimeSum = 0 * time.Second
+					startTime = now
+				}
+				bufferTimeSum += now.Sub(packet.Metadata().Timestamp)
+
 				p.observer(packet)
 				p.packetToParsedNetworkTraffic(out, assembler, packet)
 			case <-ticker.C:

pcap/pcap.go

@@ -55,20 +55,32 @@ func (p *pcapImpl) capturePackets(done <-chan struct{}, interfaceName, bpfFilter
 		}()
 
 		startTime := time.Now()
-		count := 0
+		firstPacket := true
+		bufferTimeSum := 0 * time.Second
+		intervalLength := 1 * time.Minute
 		for {
 			select {
 			case <-done:
 				return
 			case pkt, ok := <-pktChan:
 				if ok {
+					now := time.Now()
+					if now.Sub(startTime) >= intervalLength {
+						bufferLength := float64(bufferTimeSum.Nanoseconds()) / float64(intervalLength.Nanoseconds())
+						printer.Debugf("Aproximate captured-packets buffer length: %v", bufferLength)
+						bufferTimeSum = 0 * time.Second
+						startTime = now
+					}
+					bufferTimeSum += now.Sub(pkt.Metadata().Timestamp)
+
 					wrappedChan <- pkt
 
-					if count == 0 {
+					if firstPacket {
+						firstPacket = false
 						ttfp := time.Since(startTime)
 						printer.Debugf("Time to first packet on %s: %s\n", interfaceName, ttfp)
 					}
-					count += 1
+
 				} else {
 					return
 				}

pcap/run.go

@@ -1,6 +1,8 @@
 package pcap
 
 import (
+	"time"
+
 	"github.com/akitasoftware/akita-libs/akinet"
 	akihttp "github.com/akitasoftware/akita-libs/akinet/http"
 	akihttp2 "github.com/akitasoftware/akita-libs/akinet/http2"
@@ -11,6 +13,7 @@ import (
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
 	"github.com/pkg/errors"
+	"github.com/postmanlabs/postman-insights-agent/printer"
 	"github.com/postmanlabs/postman-insights-agent/trace"
 )
 
@@ -50,7 +53,19 @@ func Collect(
 		return errors.Wrap(err, "couldn't start parsing from interface")
 	}
 
+	startTime := time.Now()
+	bufferTimeSum := 0 * time.Second
+	intervalLength := 1 * time.Minute
 	for t := range parsedChan {
+		now := time.Now()
+		if now.Sub(startTime) >= intervalLength {
+			bufferLength := float64(bufferTimeSum.Nanoseconds()) / float64(intervalLength.Nanoseconds())
+			printer.Debugf("Aproximate parsed-network-traffic buffer length: %v", bufferLength)
+			bufferTimeSum = 0 * time.Second
+			startTime = now
+		}
+		bufferTimeSum += now.Sub(t.ObservationTime)
+
 		t.Interface = intf
 		err := proc.Process(t)
 		t.Content.ReleaseBuffers()

trace/backend_collector.go

@@ -425,6 +425,8 @@ func (c *BackendCollector) periodicFlush() {
 }
 
 func (c *BackendCollector) flushPairCache(cutoffTime time.Time) {
+	totalWitnesses := 0
+	flushedWitnesses := 0
 	c.pairCache.Range(func(k, v interface{}) bool {
 		e := v.(*witnessWithInfo)
 		if e.observationTime.Before(cutoffTime) {
@@ -435,7 +437,11 @@ func (c *BackendCollector) flushPairCache(cutoffTime time.Time) {
 
 			c.queueUpload(e)
 			c.pairCache.Delete(k)
+
+			flushedWitnesses += 1
 		}
+		totalWitnesses += 1
 		return true
 	})
+	printer.Debugf("flushed-witnesses in cache: %v, total-witnesses in cache: %v", flushedWitnesses, totalWitnesses)
 }