add additional metrics

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
@@ -160,6 +163,15 @@ func (p *NetworkTrafficParser) ParseFromInterface(
 				}
 				p.observer(packet)
 				p.packetToParsedNetworkTraffic(out, assembler, packet)
+
+				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)
 			case <-ticker.C:
 				// The assembler stops reassembly for streams older than streamFlushTimeout.
 				// This means the corresponding tcpFlow readers will return EOF.

pcap/pcap.go

@@ -75,7 +75,7 @@ func (p *pcapImpl) capturePackets(done <-chan struct{}, interfaceName, bpfFilter
 					now := time.Now()
 					if now.Sub(startTime) >= intervalLength {
 						bufferLength := float64(bufferTimeSum.Nanoseconds()) / float64(intervalLength.Nanoseconds())
-						printer.Debugf("Aproximate pcap buffer length: %v", bufferLength)
+						printer.Debugf("Aproximate captured-packets buffer length: %v", bufferLength)
 						bufferTimeSum = 0 * time.Second
 						startTime = now
 					}

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,13 +53,25 @@ 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 {
 		t.Interface = intf
 		err := proc.Process(t)
 		t.Content.ReleaseBuffers()
 		if err != nil {
 			return err
 		}
+
+		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)
 	}
 
 	return nil

trace/backend_collector.go

@@ -425,6 +425,8 @@ func (c *BackendCollector) periodicFlush() {
 }
 
 func (c *BackendCollector) flushPairCache(cutoffTime time.Time) {
+	now := time.Now()
+	bufferTimeSum := 0 * time.Second
 	c.pairCache.Range(func(k, v interface{}) bool {
 		e := v.(*witnessWithInfo)
 		if e.observationTime.Before(cutoffTime) {
@@ -435,7 +437,13 @@ func (c *BackendCollector) flushPairCache(cutoffTime time.Time) {
 
 			c.queueUpload(e)
 			c.pairCache.Delete(k)
+
+			if !e.witnessFlushed {
+				bufferTimeSum += now.Sub(e.observationTime)
+			}
 		}
 		return true
 	})
+	bufferLength := float64(bufferTimeSum.Nanoseconds()) / float64(pairCacheCleanupInterval.Nanoseconds())
+	printer.Debugf("Approximate unpaired-witness-cache buffer length: %v", bufferLength)
 }