feat(conntrack): measure since-last-report bytes/packets lost on GC reap

pkg/metrics/metrics.go

@@ -192,10 +192,44 @@ 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,
+	)
+
+	ConntrackGCUnreportedBytesCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric(
+		exporter.DefaultRegistry,
+		conntrackGCUnreportedBytesCounterName,
+		conntrackGCUnreportedBytesCounterDescription,
+		utils.Direction,
+	)
+
+	ConntrackGCUnreportedPacketsCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric(
+		exporter.DefaultRegistry,
+		conntrackGCUnreportedPacketsCounterName,
+		conntrackGCUnreportedPacketsCounterDescription,
+		utils.Direction,
 	)
 
 	MetricsExpiredCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric(

pkg/metrics/types.go

@@ -14,6 +14,13 @@ const (
 	lostEventsCounterName                     = "lost_events_counter"
 	parsedPacketsCounterName                  = "parsed_packets_counter"
 	expiredMetricsCounterName                 = "expired_metrics_counter"
+	conntrackGCEntriesCounterName             = "conntrack_gc_entries_counter"
+	conntrackGCUnreportedBytesCounterName     = "conntrack_gc_unreported_bytes_counter"
+	conntrackGCUnreportedPacketsCounterName   = "conntrack_gc_unreported_packets_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"
@@ -49,11 +56,17 @@ const (
 	expiredMetricsCounterDescription                 = "Number of metrics expired due to lack of updates and no longer exported"
 
 	// Conntrack metrics
-	ConntrackPacketTxDescription         = "Number of tx packets"
-	ConntrackPacketRxDescription         = "Number of rx packets"
-	ConntrackBytesTxDescription          = "Number of tx bytes"
-	ConntrackBytesRxDescription          = "Number of rx bytes"
-	ConntrackTotalConnectionsDescription = "Total number of connections"
+	ConntrackPacketTxDescription                   = "Number of tx packets"
+	ConntrackPacketRxDescription                   = "Number of rx packets"
+	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"
+	conntrackGCUnreportedBytesCounterDescription   = "Bytes observed since the last report that were dropped when a conntrack entry was garbage-collected (never reported), by direction"
+	conntrackGCUnreportedPacketsCounterDescription = "Packets observed since the last report that were dropped when a conntrack entry was garbage-collected (never reported), by direction"
+
+	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 +109,9 @@ var (
 	LostEventsCounter                     CounterVec
 	ParsedPacketsCounter                  CounterVec
 	MetricsExpiredCounter                 CounterVec
+	ConntrackGCEntriesCounter             CounterVec
+	ConntrackGCUnreportedBytesCounter     CounterVec
+	ConntrackGCUnreportedPacketsCounter   CounterVec
 
 	// DNS Metrics.
 	DNSRequestCounter  CounterVec
@@ -110,6 +126,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

@@ -11,6 +11,7 @@ import (
 	"runtime"
 	"time"
 
+	"github.com/cilium/cilium/api/v1/flow"
 	"github.com/cilium/ebpf"
 	"github.com/cilium/ebpf/rlimit"
 	"github.com/microsoft/retina/internal/ktime"
@@ -98,6 +99,43 @@ 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
+	// Bytes/packets observed since the last report, captured while the entry's
+	// value is available. These are dropped (never reported) when the entry is
+	// reaped, so they are summed into the GC-unreported counters to quantify the
+	// undercount without emitting per-flow reap events.
+	bytesTxDir, bytesRxDir     uint32
+	packetsTxDir, packetsRxDir uint32
+	// trafficDir is the entry's host-relative direction, used to attribute the
+	// tx/rx residual to ingress/egress: tx is the entry's direction, rx the opposite.
+	trafficDir flow.TrafficDirection
+}
+
+// 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 +160,21 @@ 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
+			// Bytes/packets observed since the last report that are dropped when
+			// reaped entries are deleted below, summed per host-relative direction
+			// (ingress/egress/unknown).
+			unreportedBytes := map[flow.TrafficDirection]uint64{}
+			unreportedPackets := map[flow.TrafficDirection]uint64{}
 
 			// 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 +184,15 @@ 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),
+						bytesTxDir:   value.BytesSeenSinceLastReportTxDir,
+						bytesRxDir:   value.BytesSeenSinceLastReportRxDir,
+						packetsTxDir: value.PacketsSeenSinceLastReportTxDir,
+						packetsRxDir: value.PacketsSeenSinceLastReportRxDir,
+						trafficDir:   flow.TrafficDirection(value.TrafficDirection),
+					})
 				}
 				// Log the conntrack entry
 				srcIP := utils.Int2ip(key.SrcIp).To4()
@@ -154,6 +209,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,17 +241,34 @@ 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()
+					// The reaped entry's since-last-report residual is now lost. tx is
+					// the entry's host-relative direction, rx the opposite; unknown stays
+					// unknown in both.
+					rxDir := utils.OppositeTrafficDirection(d.trafficDir)
+					unreportedBytes[d.trafficDir] += uint64(d.bytesTxDir)
+					unreportedBytes[rxDir] += uint64(d.bytesRxDir)
+					unreportedPackets[d.trafficDir] += uint64(d.packetsTxDir)
+					unreportedPackets[rxDir] += uint64(d.packetsRxDir)
 				}
 			}
+			for _, dir := range []flow.TrafficDirection{flow.TrafficDirection_INGRESS, flow.TrafficDirection_EGRESS, flow.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN} {
+				label := utils.TrafficDirectionString(dir)
+				metrics.ConntrackGCUnreportedBytesCounter.WithLabelValues(label).Add(float64(unreportedBytes[dir]))
+				metrics.ConntrackGCUnreportedPacketsCounter.WithLabelValues(label).Add(float64(unreportedPackets[dir]))
+			}
 			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