fix(packetparser): Fix under reporting of TCP flags and packet metrics, improve scalability (#1665)

# Description

Previously `packetparser` in `high` `dataAggregationLevel` would report
(mostly) every single packet since important flags were observed over
the lifetime of the connection.

This changes that behavior to only observe the important flags on
individual packets and report when necessary.

This will mean less packets are reported. However, it also adds back
weighting for bytes, packets, and TCP flags so that metrics remain
accurate versus before.

I also noticed the current docs for the TCP flags metrics are
inaccurate, we only report a subset of the supported flags. Not sure if
this is intentional, however supporting more flags will put more memory
pressure on both conntrack as well as performance pressure on packet
reporting. With sampling in place, this should be more than worth it but
there may be repercussions for the performance of `low`
`dataAggregationLevel`.

## Checklist

- [X] I have read the [contributing
documentation](https://retina.sh/docs/Contributing/overview).
- [X] I signed and signed-off the commits (`git commit -S -s ...`). See
[this
documentation](https://docs.github.com/en/authentication/managing-commit-signature-verification/about-commit-signature-verification)
on signing commits.
- [X] I have correctly attributed the author(s) of the code.
- [X] I have tested the changes locally.
- [X] I have followed the project's style guidelines.
- [X] I have updated the documentation, if necessary.
- [X] I have added tests, if applicable.

## Screenshots (if applicable) or Testing Completed

eBPF objects compile and load as expected.

# `main` Branch

<img width="1463" alt="tcpflags main"
src="https://github.com/user-attachments/assets/167908f0-7c37-4498-a7f6-20a41110d925"
/>
<img width="1463" alt="prometheus packets retina main"
src="https://github.com/user-attachments/assets/de8ad834-ed91-4673-8643-aa9cc51b3451"
/>
<img width="1463" alt="prometheus bytes retina main"
src="https://github.com/user-attachments/assets/420f54dc-fbbe-4a7b-a290-24c5d6666518"
/>

# This Branch

<img width="1463" alt="tcpflags patched"
src="https://github.com/user-attachments/assets/88460ee0-f769-4992-b9be-392b38a64b19"
/>
<img width="1463" alt="prometheus packets retina patched"
src="https://github.com/user-attachments/assets/304bbb97-8c57-477f-9c19-91bb1510b9c7"
/>
<img width="1463" alt="prometheus bytes retina patched"
src="https://github.com/user-attachments/assets/ef917562-487e-45be-8fbb-c9573fa708c1"
/>

## Additional Notes

#1628 will be a follow-up to
this to add additional sampling functionality.

---

Please refer to the [CONTRIBUTING.md](../CONTRIBUTING.md) file for more
information on how to contribute to this project.

Signed-off-by: Matthew McKeen <matthew.mckeen@fastly.com>

Author: mmckeen

pkg/metrics/metrics.go

@@ -188,6 +188,12 @@ func InitializeMetrics() {
 		ConntrackTotalConnectionsDescription,
 	)
 
+	ParsedPacketsCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric(
+		exporter.DefaultRegistry,
+		parsedPacketsCounterName,
+		parsedPacketsCounterDescription,
+	)
+
 	isInitialized = true
 	metricsLogger.Info("Metrics initialized")
 }

pkg/metrics/types.go

@@ -12,6 +12,7 @@ const (
 	// Control plane metrics
 	pluginManagerFailedToReconcileCounterName = "plugin_manager_failed_to_reconcile"
 	lostEventsCounterName                     = "lost_events_counter"
+	parsedPacketsCounterName                  = "parsed_packets_counter"
 
 	// Windows
 	hnsStats            = "windows_hns_stats"
@@ -43,6 +44,7 @@ const (
 	// Control plane metrics
 	pluginManagerFailedToReconcileCounterDescription = "Number of times the plugin manager failed to reconcile the plugins"
 	lostEventsCounterDescription                     = "Number of events lost in control plane"
+	parsedPacketsCounterDescription                  = "Number of packets parsed by the packetparser plugin"
 
 	// Conntrack metrics
 	ConntrackPacketTxDescription         = "Number of tx packets"
@@ -90,6 +92,7 @@ var (
 	// Control Plane Metrics
 	PluginManagerFailedToReconcileCounter CounterVec
 	LostEventsCounter                     CounterVec
+	ParsedPacketsCounter                  CounterVec
 
 	// DNS Metrics.
 	DNSRequestCounter  CounterVec

pkg/module/metrics/forward.go

@@ -170,8 +170,8 @@ func (f *ForwardMetrics) processLocalCtxFlow(flow *v1.Flow) {
 func (f *ForwardMetrics) update(fl *v1.Flow, labels []string) {
 	switch f.metricName {
 	case utils.ForwardPacketsGaugeName:
-		f.forwardMetric.WithLabelValues(labels...).Inc()
+		f.forwardMetric.WithLabelValues(labels...).Add(float64(utils.PreviouslyObservedPackets(fl) + 1))
 	case utils.ForwardBytesGaugeName:
-		f.forwardMetric.WithLabelValues(labels...).Add(float64(utils.PacketSize(fl)))
+		f.forwardMetric.WithLabelValues(labels...).Add(float64(utils.PacketSize(fl) + utils.PreviouslyObservedBytes(fl)))
 	}
 }

pkg/module/metrics/latency_test.go

@@ -124,7 +124,7 @@ func TestProcessFlow(t *testing.T) {
 	f1 := utils.ToFlow(l, t1, apiSeverIp, nodeIp, 80, 443, 6, 3, 0)
 	metaf1 := &utils.RetinaMetadata{}
 	utils.AddTCPID(metaf1, 1234)
-	utils.AddTCPFlags(f1, 1, 0, 0, 0, 0, 0)
+	utils.AddTCPFlags(f1, 1, 0, 0, 0, 0, 0, 0, 0, 0)
 	utils.AddRetinaMetadata(f1, metaf1)
 	f1.Destination = &flow.Endpoint{
 		PodName: "kubernetes-apiserver",
@@ -134,7 +134,7 @@ func TestProcessFlow(t *testing.T) {
 	f2 := utils.ToFlow(l, t2, nodeIp, apiSeverIp, 443, 80, 6, 2, 0)
 	metaf2 := &utils.RetinaMetadata{}
 	utils.AddTCPID(metaf2, 1234)
-	utils.AddTCPFlags(f2, 1, 1, 0, 0, 0, 0)
+	utils.AddTCPFlags(f2, 1, 1, 0, 0, 0, 0, 0, 0, 0)
 	utils.AddRetinaMetadata(f2, metaf2)
 	f2.Source = &flow.Endpoint{
 		PodName: "kubernetes-apiserver",
@@ -147,9 +147,9 @@ func TestProcessFlow(t *testing.T) {
 	 * Test case 2: Existing TCP connection.
 	 */
 	// Node -> Api server.
-	utils.AddTCPFlags(f1, 1, 0, 0, 0, 0, 0)
+	utils.AddTCPFlags(f1, 1, 0, 0, 0, 0, 0, 0, 0, 0)
 	// Api server -> Node.
-	utils.AddTCPFlags(f2, 0, 1, 0, 0, 0, 0)
+	utils.AddTCPFlags(f2, 0, 1, 0, 0, 0, 0, 0, 0, 0)
 	// Process flow.
 	lm.ProcessFlow(f1)
 	lm.ProcessFlow(f2)

pkg/module/metrics/tcpflags.go

@@ -65,6 +65,27 @@ func (t *TCPMetrics) getLabels() []string {
 	return labels
 }
 
+func combineFlagsWithPrevious(flags []string, flow *v1.Flow) map[string]uint32 {
+	var combinedFlags map[string]uint32
+
+	previous := utils.PreviouslyObservedTCPFlags(flow)
+	if previous != nil {
+		combinedFlags = previous
+	} else {
+		combinedFlags = map[string]uint32{}
+	}
+
+	for _, flag := range flags {
+		if _, ok := combinedFlags[flag]; !ok {
+			combinedFlags[flag] = 1
+		} else {
+			combinedFlags[flag]++
+		}
+	}
+
+	return combinedFlags
+}
+
 func (t *TCPMetrics) ProcessFlow(flow *v1.Flow) {
 	if flow == nil {
 		return
@@ -100,11 +121,11 @@ func (t *TCPMetrics) ProcessFlow(flow *v1.Flow) {
 		dstLabels = t.dstCtx.getValues(flow)
 	}
 
-	for _, flag := range flags {
+	for flag, count := range combineFlagsWithPrevious(flags, flow) {
 		labels := append([]string{flag}, srcLabels...)
 		labels = append(labels, dstLabels...)
-		t.tcpFlagsMetrics.WithLabelValues(labels...).Inc()
-		t.l.Debug("TCP flag metric", zap.String("flag", flag), zap.Strings("labels", labels))
+		t.tcpFlagsMetrics.WithLabelValues(labels...).Add(float64(count))
+		t.l.Debug("TCP flag metric", zap.String("flag", flag), zap.Strings("labels", labels), zap.Uint32("count", count))
 	}
 }
 
@@ -113,20 +134,23 @@ func (t *TCPMetrics) processLocalCtxFlow(flow *v1.Flow, flags []string) {
 	if labelValuesMap == nil {
 		return
 	}
+
+	combinedFlags := combineFlagsWithPrevious(flags, flow)
+
 	// Ingress values
 	if l := len(labelValuesMap[ingress]); l > 0 {
-		for _, flag := range flags {
+		for flag, count := range combinedFlags {
 			labels := append([]string{flag}, labelValuesMap[ingress]...)
-			t.tcpFlagsMetrics.WithLabelValues(labels...).Inc()
-			t.l.Debug("TCP flag metric", zap.String("flag", flag), zap.Strings("labels", labels))
+			t.tcpFlagsMetrics.WithLabelValues(labels...).Add(float64(count))
+			t.l.Debug("TCP flag metric", zap.String("flag", flag), zap.Strings("labels", labels), zap.Uint32("count", count))
 		}
 	}
 
 	if l := len(labelValuesMap[egress]); l > 0 {
-		for _, flag := range flags {
+		for flag, count := range combinedFlags {
 			labels := append([]string{flag}, labelValuesMap[egress]...)
-			t.tcpFlagsMetrics.WithLabelValues(labels...).Inc()
-			t.l.Debug("TCP flag metric", zap.String("flag", flag), zap.Strings("labels", labels))
+			t.tcpFlagsMetrics.WithLabelValues(labels...).Add(float64(count))
+			t.l.Debug("TCP flag metric", zap.String("flag", flag), zap.Strings("labels", labels), zap.Uint32("count", count))
 		}
 	}
 }
@@ -171,6 +195,10 @@ func (t *TCPMetrics) getFlagValues(flags *v1.TCPFlags) []string {
 		f = append(f, utils.CWR)
 	}
 
+	if flags.GetNS() {
+		f = append(f, utils.NS)
+	}
+
 	return f
 }