feat(packetparser): break down parsed packets and conntrack GCed entries by reason

pkg/metrics/metrics.go

@@ -192,10 +192,30 @@ func InitializeMetrics(logger *slog.Logger) {
 		ConntrackTotalConnectionsDescription,
 	)
 
+	ConntrackUnknownDirectionConnections = exporter.CreatePrometheusGaugeVecForMetric(
+		exporter.DefaultRegistry,
+		utils.ConntrackUnknownDirectionConnectionsName,
+		ConntrackUnknownDirectionConnectionsDescription,
+	)
+
+	ConntrackUnknownDirectionBytes = exporter.CreatePrometheusGaugeVecForMetric(
+		exporter.DefaultRegistry,
+		utils.ConntrackUnknownDirectionBytesName,
+		ConntrackUnknownDirectionBytesDescription,
+	)
+
 	ParsedPacketsCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric(
 		exporter.DefaultRegistry,
 		parsedPacketsCounterName,
 		parsedPacketsCounterDescription,
+		reasonLabel,
+	)
+
+	ConntrackGCEntriesCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric(
+		exporter.DefaultRegistry,
+		conntrackGCEntriesCounterName,
+		conntrackGCEntriesCounterDescription,
+		reasonLabel,
 	)
 
 	MetricsExpiredCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric(

pkg/metrics/types.go

@@ -14,6 +14,11 @@ const (
 	lostEventsCounterName                     = "lost_events_counter"
 	parsedPacketsCounterName                  = "parsed_packets_counter"
 	expiredMetricsCounterName                 = "expired_metrics_counter"
+	conntrackGCEntriesCounterName             = "conntrack_gc_entries_counter"
+
+	// reasonLabel labels parsed-packet reports by the trigger that produced them,
+	// and conntrack GC entries by the connection state at eviction.
+	reasonLabel = "reason"
 
 	// Windows
 	hnsStats            = "windows_hns_stats"
@@ -54,6 +59,10 @@ const (
 	ConntrackBytesTxDescription          = "Number of tx bytes"
 	ConntrackBytesRxDescription          = "Number of rx bytes"
 	ConntrackTotalConnectionsDescription = "Total number of connections"
+	conntrackGCEntriesCounterDescription = "Number of conntrack entries removed by garbage collection, by connection state"
+
+	ConntrackUnknownDirectionConnectionsDescription = "Number of live connections whose direction is unknown (SYN not observed, e.g. established before tracking or after LRU eviction)"
+	ConntrackUnknownDirectionBytesDescription       = "Cumulative bytes of live unknown-direction connections"
 )
 
 // Metric Counters
@@ -96,6 +105,7 @@ var (
 	LostEventsCounter                     CounterVec
 	ParsedPacketsCounter                  CounterVec
 	MetricsExpiredCounter                 CounterVec
+	ConntrackGCEntriesCounter             CounterVec
 
 	// DNS Metrics.
 	DNSRequestCounter  CounterVec
@@ -110,6 +120,9 @@ var (
 	ConntrackBytesTx          GaugeVec
 	ConntrackBytesRx          GaugeVec
 	ConntrackTotalConnections GaugeVec
+
+	ConntrackUnknownDirectionConnections GaugeVec
+	ConntrackUnknownDirectionBytes       GaugeVec
 )
 
 func ToPrometheusType(metric interface{}) prometheus.Collector {

pkg/plugin/conntrack/_cprog/conntrack.c

@@ -62,6 +62,7 @@ struct packet
     __u8 proto;
     __u16 flags; // For TCP packets, this is the TCP flags. For UDP packets, this is will always be 1 for conntrack purposes.
     bool is_reply;
+    __u8 report_reason; // Why this packet was reported to userspace. One of REPORT_REASON_*.
     __u32 previously_observed_packets; // When sampling, this is the number of observed packets since the last report.
     __u32 previously_observed_bytes; // When sampling, this is the number of observed bytes since the last report.
     struct tcpflagscount previously_observed_flags; // When sampling, this is the previously observed TCP flags since the last report.
@@ -77,6 +78,7 @@ struct packetreport
     __u32 previously_observed_bytes;
     struct tcpflagscount previously_observed_flags;
     bool report;
+    __u8 report_reason; // One of REPORT_REASON_*; valid when report is true.
 };
 
 /**
@@ -477,6 +479,7 @@ static __always_inline struct packetreport _ct_should_report_packet(struct ct_v4
     if (now >= eviction_time) {
         bpf_map_delete_elem(&retina_conntrack, key);
         report.report = true;
+        report.report_reason = REPORT_REASON_TIMEOUT;
         return report; // Report the last packet received before deletion
     }
 
@@ -494,20 +497,24 @@ static __always_inline struct packetreport _ct_should_report_packet(struct ct_v4
         // Handle TCP connection termination states
 
         // Check if this is the final ACK in TCP connection teardown
-        // (Both directions have seen FIN, and this is just an ACK without other control flags)
-        if ((flags & TCP_ACK) && 
-            !(flags & (TCP_FIN | TCP_SYN | TCP_RST)) && 
-            (entry->flags_seen_tx_dir & TCP_FIN) && 
+        // (Both directions have seen FIN, and the current packet is just an ACK without other
+        // control flags). Test packet_flags, not the seen-flags-OR'd flags, since this direction
+        // has already recorded its FIN and would otherwise always fail the FIN check.
+        if ((packet_flags & TCP_ACK) &&
+            !(packet_flags & (TCP_FIN | TCP_SYN | TCP_RST)) &&
+            (entry->flags_seen_tx_dir & TCP_FIN) &&
             (entry->flags_seen_rx_dir & TCP_FIN)) {
             bpf_map_delete_elem(&retina_conntrack, key);
             report.report = true;
+            report.report_reason = REPORT_REASON_FINAL_ACK;
             return report; // Report final ACK before connection removal
         }
 
         // If RST is seen, delete connection immediately
         if (flags & TCP_RST) {
             bpf_map_delete_elem(&retina_conntrack, key);
             report.report = true;
+            report.report_reason = REPORT_REASON_RST;
             return report; // Report RST before connection removal
         }
 
@@ -553,6 +560,13 @@ static __always_inline struct packetreport _ct_should_report_packet(struct ct_v4
     // 3. Reporting interval has elapsed
     if (should_report || (sampled && flags != seen_flags) || now - last_report >= CT_REPORT_INTERVAL) {
         report.report = true;
+        if (should_report) {
+            report.report_reason = REPORT_REASON_TCP_FLAGS;
+        } else if (sampled && flags != seen_flags) {
+            report.report_reason = REPORT_REASON_FLAG_CHANGE;
+        } else {
+            report.report_reason = REPORT_REASON_INTERVAL;
+        }
         // Update the connection's state
         if (direction == CT_PACKET_DIR_TX) {
             WRITE_ONCE(entry->last_report_tx_dir, now);

pkg/plugin/conntrack/_cprog/conntrack.h

@@ -51,3 +51,12 @@
 #define OBSERVATION_POINT_TO_ENDPOINT 0x01
 #define OBSERVATION_POINT_FROM_NETWORK 0x02
 #define OBSERVATION_POINT_TO_NETWORK 0x03
+
+// Reasons a packet was reported to userspace.
+#define REPORT_REASON_NEW_CONNECTION 0x00
+#define REPORT_REASON_INTERVAL 0x01
+#define REPORT_REASON_FLAG_CHANGE 0x02
+#define REPORT_REASON_TCP_FLAGS 0x03
+#define REPORT_REASON_FINAL_ACK 0x04
+#define REPORT_REASON_RST 0x05
+#define REPORT_REASON_TIMEOUT 0x06

pkg/plugin/conntrack/conntrack_linux.go

@@ -98,6 +98,34 @@ func GenerateDynamic(ctx context.Context, dynamicHeaderPath string, conntrackMet
 	return nil
 }
 
+// gcDeletion is an entry queued for garbage collection, carrying the reason classified
+// while its value was still available (the delete loop only has the key).
+type gcDeletion struct {
+	key    conntrackCtV4Key
+	reason string
+}
+
+// gcEntryReason classifies a conntrack entry removed by garbage collection, from its
+// protocol and the TCP flags seen across both directions.
+func gcEntryReason(proto uint8, value *conntrackCtEntry) string {
+	switch proto {
+	case 17: //nolint:gomnd // UDP
+		return "udp_idle"
+	case 6: //nolint:gomnd // TCP
+		flags := value.FlagsSeenTxDir | value.FlagsSeenRxDir
+		switch {
+		case flags&TCP_RST != 0:
+			return "tcp_rst"
+		case flags&TCP_FIN != 0:
+			return "tcp_fin"
+		default:
+			return "tcp_idle"
+		}
+	default:
+		return "other"
+	}
+}
+
 // Run starts the Conntrack garbage collection loop.
 func (ct *Conntrack) Run(ctx context.Context) error {
 	ticker := time.NewTicker(ct.gcFrequency)
@@ -122,12 +150,16 @@ func (ct *Conntrack) Run(ctx context.Context) error {
 			var value conntrackCtEntry
 
 			var noOfCtEntries, entriesDeleted int
-			// List of keys to be deleted
-			var keysToDelete []conntrackCtV4Key
+			// List of entries to be deleted, with the reason captured while the value is available.
+			var keysToDelete []gcDeletion
 
 			// metrics counters
 			var packetsCountTx, packetsCountRx, totConnections uint32
 			var bytesCountTx, bytesCountRx uint64
+			// Live unknown-direction population (SYN never observed), e.g. connections
+			// established before tracking or recreated after LRU eviction.
+			var unknownDirectionEntries uint32
+			var unknownDirectionBytes uint64
 
 			iter := ct.ctMap.Iterate()
 			for iter.Next(&key, &value) {
@@ -137,7 +169,7 @@ func (ct *Conntrack) Run(ctx context.Context) error {
 					// Iterating a hash map from which keys are being deleted is not safe.
 					// So, we store the keys to be deleted in a list and delete them after the iteration.
 					keyCopy := key // Copy the key to avoid using the same key in the next iteration
-					keysToDelete = append(keysToDelete, keyCopy)
+					keysToDelete = append(keysToDelete, gcDeletion{key: keyCopy, reason: gcEntryReason(key.Proto, &value)})
 				}
 				// Log the conntrack entry
 				srcIP := utils.Int2ip(key.SrcIp).To4()
@@ -154,6 +186,10 @@ func (ct *Conntrack) Run(ctx context.Context) error {
 					bytesCountRx += ctMeta.BytesRxCount
 					packetsCountTx += ctMeta.PacketsTxCount
 					packetsCountRx += ctMeta.PacketsRxCount
+					if value.IsDirectionUnknown {
+						unknownDirectionEntries++
+						unknownDirectionBytes += ctMeta.BytesTxCount + ctMeta.BytesRxCount
+					}
 				}
 
 				ct.l.Debug("conntrack entry",
@@ -182,15 +218,19 @@ func (ct *Conntrack) Run(ctx context.Context) error {
 				metrics.ConntrackPacketsRx.WithLabelValues().Set(float64(packetsCountRx))
 				metrics.ConntrackBytesRx.WithLabelValues().Set(float64(bytesCountRx))
 				metrics.ConntrackTotalConnections.WithLabelValues().Set(float64(totConnections))
+				metrics.ConntrackUnknownDirectionConnections.WithLabelValues().Set(float64(unknownDirectionEntries))
+				metrics.ConntrackUnknownDirectionBytes.WithLabelValues().Set(float64(unknownDirectionBytes))
 			}
 
 			// Delete the conntrack entries
-			for _, key := range keysToDelete {
-				if err := ct.ctMap.Delete(key); err != nil {
-					// Should only happen in a high connection churn scenario
+			for _, d := range keysToDelete {
+				if err := ct.ctMap.Delete(d.key); err != nil {
+					// Should only happen in a high connection churn scenario, e.g. the entry was
+					// already deleted in-kernel on connection teardown between iteration and now.
 					ct.l.Debug("Delete failed", zap.Error(err))
 				} else {
 					entriesDeleted++
+					metrics.ConntrackGCEntriesCounter.WithLabelValues(d.reason).Inc()
 				}
 			}
 			ct.l.Debug("conntrack GC completed", zap.Int("number_of_entries", noOfCtEntries), zap.Int("entries_deleted", entriesDeleted))

pkg/plugin/conntrack/conntrack_linux_test.go

@@ -90,3 +90,28 @@ func getCurrentFilePath(t *testing.T) string {
 	}
 	return filename
 }
+
+func TestGCEntryReason(t *testing.T) {
+	tests := []struct {
+		name             string
+		proto            uint8
+		txFlags, rxFlags uint8
+		want             string
+	}{
+		{"udp idle", 17, 0, 0, "udp_idle"},
+		{"tcp rst", 6, TCP_RST, 0, "tcp_rst"},
+		{"tcp fin tx", 6, TCP_FIN, 0, "tcp_fin"},
+		{"tcp fin rx", 6, 0, TCP_FIN, "tcp_fin"},
+		{"tcp idle", 6, TCP_ACK, TCP_ACK, "tcp_idle"},
+		{"tcp rst takes precedence over fin", 6, TCP_FIN | TCP_RST, 0, "tcp_rst"},
+		{"other proto", 1, 0, 0, "other"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			v := conntrackCtEntry{FlagsSeenTxDir: tt.txFlags, FlagsSeenRxDir: tt.rxFlags}
+			if got := gcEntryReason(tt.proto, &v); got != tt.want {
+				t.Errorf("gcEntryReason(%d, tx=%#x, rx=%#x) = %q, want %q", tt.proto, tt.txFlags, tt.rxFlags, got, tt.want)
+			}
+		})
+	}
+}

pkg/plugin/packetparser/_cprog/packetparser.c

@@ -233,6 +233,7 @@ static void parse(struct __sk_buff *skb, __u8 obs)
 	// Process the packet in ct
 	struct packetreport report __attribute__((unused));
 	report = ct_process_packet(&p, obs, sampled);
+	p.report_reason = report.report_reason;
 
 	// If the data aggregation level is low, always send the packet to the perf buffer.
 	#if DATA_AGGREGATION_LEVEL == DATA_AGGREGATION_LEVEL_LOW

pkg/plugin/packetparser/packetparser_linux.go

@@ -777,6 +777,38 @@ func (p *packetParser) run(ctx context.Context) error {
 	return nil
 }
 
+// Report reasons, matching REPORT_REASON_* in conntrack.h.
+const (
+	reportReasonNewConnection uint8 = iota
+	reportReasonInterval
+	reportReasonFlagChange
+	reportReasonTCPFlags
+	reportReasonFinalACK
+	reportReasonRST
+	reportReasonTimeout
+)
+
+func reportReasonString(r uint8) string {
+	switch r {
+	case reportReasonNewConnection:
+		return "new_connection"
+	case reportReasonInterval:
+		return "interval"
+	case reportReasonFlagChange:
+		return "flag_change"
+	case reportReasonTCPFlags:
+		return "tcp_flags"
+	case reportReasonFinalACK:
+		return "final_ack"
+	case reportReasonRST:
+		return "rst"
+	case reportReasonTimeout:
+		return "timeout"
+	default:
+		return "unknown"
+	}
+}
+
 // This is the data consumer.
 // There will more than one of these.
 func (p *packetParser) processRecord(ctx context.Context, id int) {
@@ -794,15 +826,15 @@ func (p *packetParser) processRecord(ctx context.Context, id int) {
 				zap.Int("worker_id", id),
 			)
 
-			metrics.ParsedPacketsCounter.WithLabelValues().Inc()
-
 			var bpfEvent packetparserPacket
 			err := binary.Read(bytes.NewReader(record.RawSample), binary.LittleEndian, &bpfEvent)
 			if err != nil {
 				p.l.Error("Error reading bpfEvent", zap.Error(err))
 				continue
 			}
 
+			metrics.ParsedPacketsCounter.WithLabelValues(reportReasonString(bpfEvent.ReportReason)).Inc()
+
 			// Post processing of the bpfEvent.
 			// Anything after this is required only for Pod level metrics.
 			sourcePortShort := uint32(utils.HostToNetShort(bpfEvent.SrcPort))

pkg/plugin/packetparser/packetparser_linux_test.go

@@ -835,3 +835,22 @@ func restoreBackup() {
 		}
 	}
 }
+
+func TestReportReasonString(t *testing.T) {
+	cases := []struct {
+		r    uint8
+		want string
+	}{
+		{reportReasonNewConnection, "new_connection"},
+		{reportReasonInterval, "interval"},
+		{reportReasonFlagChange, "flag_change"},
+		{reportReasonTCPFlags, "tcp_flags"},
+		{reportReasonFinalACK, "final_ack"},
+		{reportReasonRST, "rst"},
+		{reportReasonTimeout, "timeout"},
+		{99, "unknown"},
+	}
+	for _, tt := range cases {
+		assert.Equal(t, tt.want, reportReasonString(tt.r))
+	}
+}

pkg/utils/metric_names.go

@@ -41,6 +41,9 @@ const (
 	ConntrackBytesTxGaugeName     = "conntrack_bytes_tx"
 	ConntrackBytesRxGaugeName     = "conntrack_bytes_rx"
 	ConntrackTotalConnectionsName = "conntrack_total_connections"
+
+	ConntrackUnknownDirectionConnectionsName = "conntrack_unknown_direction_connections"
+	ConntrackUnknownDirectionBytesName       = "conntrack_unknown_direction_bytes"
 )
 
 // IsAdvancedMetric is a helper function to determine if a name is an advanced metric