Add latency calculation tests & logs (#103)

Added witness latency calculation tests. Specifically the following
scenarios are now being tested:

- the request and the response have 8 ms of latency
- the request and the response have 0 ms of latency
- the request and the response overlap by -1 ms of latency
- the request and the response fully overlap each other have 0 ms of
latency
- the request and the response are in the wrong order have -4 ms of
latency
- the request and the response timestamps are the zero value have 0 ms
of latency
- a stream with two interleaved request and response each with 3 ms of
latency
- two pipelined requests and responses have 0 ms of latency

Added 2 validations in the stream processing layer:
- if the first or last packet timestamps are the default zero value,
override to `time.Now()`.
- if the last packet timestamp is before the first packet timestamp,
override last packet timestamp to first packet timestamp.

^These are being logged and printed as part of the `PrintWarnings`
report at the end of a session.

Added 3 validations in the backend collector layer:
- if the collector matches 2 requests or 2 responses together (due to
HTTP/1.1 pipelining and matching on `ACK`s), log it and leave latency
default value of `0.0`.
- if the collector sees timestamp zero values, log it and leave latency
default value of `0.0`
- if the collector sees negative latency calculation, log and report it.

Author: cesher

apidump/summary.go

@@ -265,11 +265,17 @@ func (s *Summary) printHostHighlights(top *client_telemetry.PacketCountSummary)
 // any packets, capturing packets but failing to parse them, etc.
 func (s *Summary) PrintWarnings() {
 	// Report on recoverable error counts during trace
-	if pcap.CountNilAssemblerContext > 0 || pcap.CountNilAssemblerContextAfterParse > 0 || pcap.CountBadAssemblerContextType > 0 {
-		printer.Stderr.Infof("Detected packet assembly context problems during capture: %v empty, %v bad type, %v empty after parse. ",
+	if pcap.CountNilAssemblerContext > 0 ||
+		pcap.CountNilAssemblerContextAfterParse > 0 ||
+		pcap.CountBadAssemblerContextType > 0 ||
+		pcap.CountZeroValuePacketTimestamp > 0 ||
+		pcap.CountLastPacketBeforeFirstPacket > 0 {
+		printer.Stderr.Infof("Detected packet assembly context problems during capture: %v empty, %v bad type, %v empty after parse, %v zero timestamp, %v last before first timestamp. ",
 			pcap.CountNilAssemblerContext,
 			pcap.CountBadAssemblerContextType,
-			pcap.CountNilAssemblerContextAfterParse)
+			pcap.CountNilAssemblerContextAfterParse,
+			pcap.CountZeroValuePacketTimestamp,
+			pcap.CountLastPacketBeforeFirstPacket)
 		printer.Stderr.Infof("These errors may cause some packets to be missing from the trace.\n")
 	}
 
@@ -383,5 +389,4 @@ func DumpPacketCounters(logf func(f string, args ...interface{}), interfaces map
 	}
 
 	logf("==================================================\n")
-
 }

pcap/stream.go

@@ -31,6 +31,14 @@ var CountNilAssemblerContextAfterParse uint64
 // happen at all.
 var CountBadAssemblerContextType uint64
 
+// Number of times we got parsed network traffic where either the firstPacket or lastPacket
+// timestamp is the default zero value.
+var CountZeroValuePacketTimestamp uint64
+
+// Number of times we got parsed network traffic where the firstPacket timestamp
+// is before the lastPacket timestamp.
+var CountLastPacketBeforeFirstPacket uint64
+
 // tcpFlow represents a uni-directional flow of TCP segments along with a
 // bidirectional ID that identifies the tcpFlow in the opposite direction.
 // Writes come from TCP assembler via tcpStream, while reads come from users
@@ -219,11 +227,24 @@ func (f *tcpFlow) reassemblyComplete() {
 }
 
 func (f *tcpFlow) toPNT(firstPacketTime time.Time, lastPacketTime time.Time,
-	c akinet.ParsedNetworkContent) akinet.ParsedNetworkTraffic {
-	if firstPacketTime.IsZero() {
-		firstPacketTime = f.clock.Now()
+	c akinet.ParsedNetworkContent,
+) akinet.ParsedNetworkTraffic {
+	if firstPacketTime.IsZero() || lastPacketTime.IsZero() {
+		now := f.clock.Now()
+		printer.V(6).Infof("ParsedNetworkTraffic with zero value packet timestamps. first: %v last: %v now: %v", firstPacketTime, lastPacketTime, now)
+		atomic.AddUint64(&CountZeroValuePacketTimestamp, 1)
+
+		if firstPacketTime.IsZero() {
+			firstPacketTime = now
+		}
+		if lastPacketTime.IsZero() {
+			lastPacketTime = now
+		}
 	}
-	if lastPacketTime.IsZero() {
+	if lastPacketTime.Before(firstPacketTime) {
+		printer.V(6).Infof("ParsedNetworkTraffic with last packet before first packet. first: %v last: %v", firstPacketTime, lastPacketTime)
+		atomic.AddUint64(&CountLastPacketBeforeFirstPacket, 1)
+
 		lastPacketTime = firstPacketTime
 	}
 

trace/backend_collector.go

@@ -52,9 +52,9 @@ type witnessWithInfo struct {
 	dstIP           net.IP // The HTTP server's IP address.
 	dstPort         uint16 // The HTTP server's port number.
 	observationTime time.Time
+	finalPacketTime time.Time
 	id              akid.WitnessID
-	requestEnd      time.Time
-	responseStart   time.Time
+	isRequest       bool
 
 	// Mutex protecting witness while it is being processed and/or flushed.
 	witnessMutex sync.Mutex
@@ -89,36 +89,41 @@ func (r *witnessWithInfo) toReport() (*kgxapi.WitnessReport, error) {
 	}, nil
 }
 
-func (w *witnessWithInfo) recordTimestamp(isRequest bool, t akinet.ParsedNetworkTraffic) {
-	if isRequest {
-		w.requestEnd = t.FinalPacketTime
-	} else {
-		w.responseStart = t.ObservationTime
+func (w *witnessWithInfo) computeProcessingLatency(isRequest bool, t akinet.ParsedNetworkTraffic) {
+	if w.isRequest == isRequest {
+		if isRequest {
+			printer.Debugln("Skipping latency calculation. Matched 2 requests together.")
+		} else {
+			printer.Debugln("Skipping latency calculation. Matched 2 responses together.")
+		}
+		return
 	}
 
-}
-
-func (w *witnessWithInfo) computeProcessingLatency(isRequest bool, t akinet.ParsedNetworkTraffic) {
 	// Processing latency is the time from the last packet of the request,
 	// to the first packet of the response.
-	requestEnd := w.requestEnd
-	responseStart := t.ObservationTime
-
-	// handle arrival in opposite order
-	if isRequest {
+	var requestEnd, responseStart time.Time
+	if w.isRequest {
+		requestEnd = w.finalPacketTime
+		responseStart = t.ObservationTime
+	} else {
 		requestEnd = t.FinalPacketTime
-		responseStart = w.responseStart
+		responseStart = w.observationTime
 	}
 
 	// Missing data, leave as default value in protobuf
 	if requestEnd.IsZero() || responseStart.IsZero() {
+		printer.Debugf("Skipping latency calculation. Zero value timestamps. RequestEnd: %v ResponseStart %v.\n", requestEnd, responseStart)
 		return
 	}
 
+	latency := float32(responseStart.Sub(requestEnd).Microseconds()) / 1000.0
+	if latency < 0.0 {
+		printer.Debugf("Negative latency calculation: %v\n", latency)
+	}
+
 	// HTTPMethodMetadata only for now
 	if meta := spec_util.HTTPMetaFromMethod(w.witness.Method); meta != nil {
-		latency := responseStart.Sub(requestEnd)
-		meta.ProcessingLatency = float32(latency.Microseconds()) / 1000.0
+		meta.ProcessingLatency = latency
 	}
 }
 
@@ -251,10 +256,10 @@ func (c *BackendCollector) Process(t akinet.ParsedNetworkTraffic) error {
 			dstPort:         uint16(t.DstPort),
 			witness:         partial.Witness,
 			observationTime: t.ObservationTime,
+			finalPacketTime: t.FinalPacketTime,
 			id:              partial.PairKey,
+			isRequest:       isRequest,
 		}
-		// Store whichever timestamp brackets the processing interval.
-		w.recordTimestamp(isRequest, t)
 		c.pairCache.Store(partial.PairKey, w)
 		printer.Debugf("Partial witness %v request=%v at %v -- %v\n",
 			partial.PairKey, isRequest, t.ObservationTime, t.FinalPacketTime)
@@ -299,8 +304,10 @@ func (c *BackendCollector) processTLSHandshake(tls akinet.TLSHandshakeMetadata)
 	return nil
 }
 
-var cloudAPIEnvironmentsPathRE = regexp.MustCompile(`^/environments/[^/]+$`)
-var cloudAPIHostnames = sets.NewSet[string]()
+var (
+	cloudAPIEnvironmentsPathRE = regexp.MustCompile(`^/environments/[^/]+$`)
+	cloudAPIHostnames          = sets.NewSet[string]()
+)
 
 func init() {
 	cloudAPIHostnames.Insert("api.getpostman-stage.com")
@@ -317,7 +324,6 @@ func init() {
 // XXX This is a stop-gap hack to exclude certain endpoints for Cloud API from
 // Repro Mode.
 func excludeWitnessFromReproMode(w *pb.Witness) bool {
-
 	httpMeta := w.GetMethod().GetMeta().GetHttp()
 	if httpMeta == nil {
 		return false