From 4597a4d2d0648c027038dadc02db6f9e53fedc55 Mon Sep 17 00:00:00 2001 From: Matthew McKeen Date: Wed, 8 Jul 2026 14:57:00 -0700 Subject: [PATCH 1/6] feat(packetparser): add report_reason label to parsed packets Signed-off-by: Matthew McKeen --- pkg/metrics/metrics.go | 1 + pkg/metrics/types.go | 3 ++ pkg/plugin/conntrack/_cprog/conntrack.c | 18 ++++++++-- pkg/plugin/conntrack/_cprog/conntrack.h | 9 +++++ pkg/plugin/packetparser/_cprog/packetparser.c | 1 + .../packetparser/packetparser_bpfel_arm64.go | 2 +- .../packetparser/packetparser_bpfel_x86.go | 2 +- pkg/plugin/packetparser/packetparser_linux.go | 36 +++++++++++++++++-- 8 files changed, 65 insertions(+), 7 deletions(-) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index d0f62df14b..a3586bc794 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -196,6 +196,7 @@ func InitializeMetrics(logger *slog.Logger) { exporter.DefaultRegistry, parsedPacketsCounterName, parsedPacketsCounterDescription, + reasonLabel, ) MetricsExpiredCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric( diff --git a/pkg/metrics/types.go b/pkg/metrics/types.go index 54f8446bb5..158578b229 100644 --- a/pkg/metrics/types.go +++ b/pkg/metrics/types.go @@ -15,6 +15,9 @@ const ( parsedPacketsCounterName = "parsed_packets_counter" expiredMetricsCounterName = "expired_metrics_counter" + // reasonLabel labels parsed-packet reports by the trigger that produced them. + reasonLabel = "reason" + // Windows hnsStats = "windows_hns_stats" hnsStatsDescription = "Include many different metrics from packets sent/received to closed connections" diff --git a/pkg/plugin/conntrack/_cprog/conntrack.c b/pkg/plugin/conntrack/_cprog/conntrack.c index f5fb2b3729..b855b90a24 100644 --- a/pkg/plugin/conntrack/_cprog/conntrack.c +++ b/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 } @@ -495,12 +498,13 @@ static __always_inline struct packetreport _ct_should_report_packet(struct ct_v4 // 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) && + if ((flags & TCP_ACK) && + !(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 } @@ -508,6 +512,7 @@ static __always_inline struct packetreport _ct_should_report_packet(struct ct_v4 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 +558,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); diff --git a/pkg/plugin/conntrack/_cprog/conntrack.h b/pkg/plugin/conntrack/_cprog/conntrack.h index 0de6018180..c6cedd39e3 100644 --- a/pkg/plugin/conntrack/_cprog/conntrack.h +++ b/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 diff --git a/pkg/plugin/packetparser/_cprog/packetparser.c b/pkg/plugin/packetparser/_cprog/packetparser.c index 5d63edde0a..b30f04cbf7 100644 --- a/pkg/plugin/packetparser/_cprog/packetparser.c +++ b/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 diff --git a/pkg/plugin/packetparser/packetparser_bpfel_arm64.go b/pkg/plugin/packetparser/packetparser_bpfel_arm64.go index 3ce8c9a4c8..9dd8efc686 100644 --- a/pkg/plugin/packetparser/packetparser_bpfel_arm64.go +++ b/pkg/plugin/packetparser/packetparser_bpfel_arm64.go @@ -87,7 +87,7 @@ type packetparserPacket struct { _ [1]byte Flags uint16 IsReply bool - _ [1]byte + ReportReason uint8 PreviouslyObservedPackets uint32 PreviouslyObservedBytes uint32 PreviouslyObservedFlags struct { diff --git a/pkg/plugin/packetparser/packetparser_bpfel_x86.go b/pkg/plugin/packetparser/packetparser_bpfel_x86.go index ce86707cca..61691ff86b 100644 --- a/pkg/plugin/packetparser/packetparser_bpfel_x86.go +++ b/pkg/plugin/packetparser/packetparser_bpfel_x86.go @@ -87,7 +87,7 @@ type packetparserPacket struct { _ [1]byte Flags uint16 IsReply bool - _ [1]byte + ReportReason uint8 PreviouslyObservedPackets uint32 PreviouslyObservedBytes uint32 PreviouslyObservedFlags struct { diff --git a/pkg/plugin/packetparser/packetparser_linux.go b/pkg/plugin/packetparser/packetparser_linux.go index 005f708542..abc3c2830b 100644 --- a/pkg/plugin/packetparser/packetparser_linux.go +++ b/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,8 +826,6 @@ 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 { @@ -803,6 +833,8 @@ func (p *packetParser) processRecord(ctx context.Context, id int) { 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)) From f1d811fe9ebf3411dc1b41cd20286b77c79b0e78 Mon Sep 17 00:00:00 2001 From: Matthew McKeen Date: Wed, 8 Jul 2026 14:57:29 -0700 Subject: [PATCH 2/6] fix(conntrack): detect final ACK using current packet flags Signed-off-by: Matthew McKeen --- pkg/plugin/conntrack/_cprog/conntrack.c | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/pkg/plugin/conntrack/_cprog/conntrack.c b/pkg/plugin/conntrack/_cprog/conntrack.c index b855b90a24..b09888285a 100644 --- a/pkg/plugin/conntrack/_cprog/conntrack.c +++ b/pkg/plugin/conntrack/_cprog/conntrack.c @@ -497,9 +497,11 @@ 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)) && + // (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); From 2156c78278baa85358e62f830eb259934dd13078 Mon Sep 17 00:00:00 2001 From: Matthew McKeen Date: Wed, 8 Jul 2026 19:21:04 -0700 Subject: [PATCH 3/6] feat(conntrack): count GC-removed entries by connection state Signed-off-by: Matthew McKeen --- pkg/metrics/metrics.go | 7 +++++ pkg/metrics/types.go | 6 +++- pkg/plugin/conntrack/conntrack_linux.go | 42 +++++++++++++++++++++---- 3 files changed, 48 insertions(+), 7 deletions(-) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index a3586bc794..dc830ed659 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -199,6 +199,13 @@ func InitializeMetrics(logger *slog.Logger) { reasonLabel, ) + ConntrackGCEntriesCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric( + exporter.DefaultRegistry, + conntrackGCEntriesCounterName, + conntrackGCEntriesCounterDescription, + reasonLabel, + ) + MetricsExpiredCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric( exporter.DefaultRegistry, expiredMetricsCounterName, diff --git a/pkg/metrics/types.go b/pkg/metrics/types.go index 158578b229..9f162d7cb7 100644 --- a/pkg/metrics/types.go +++ b/pkg/metrics/types.go @@ -14,8 +14,10 @@ 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. + // reasonLabel labels parsed-packet reports by the trigger that produced them, + // and conntrack GC entries by the connection state at eviction. reasonLabel = "reason" // Windows @@ -57,6 +59,7 @@ 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" ) // Metric Counters @@ -99,6 +102,7 @@ var ( LostEventsCounter CounterVec ParsedPacketsCounter CounterVec MetricsExpiredCounter CounterVec + ConntrackGCEntriesCounter CounterVec // DNS Metrics. DNSRequestCounter CounterVec diff --git a/pkg/plugin/conntrack/conntrack_linux.go b/pkg/plugin/conntrack/conntrack_linux.go index 4e1eb41cf3..9b0f12c08d 100644 --- a/pkg/plugin/conntrack/conntrack_linux.go +++ b/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,8 +150,8 @@ 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 @@ -137,7 +165,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() @@ -185,12 +213,14 @@ func (ct *Conntrack) Run(ctx context.Context) error { } // 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)) From 46d2d839fc666e01467d9c1da070e625667ccf33 Mon Sep 17 00:00:00 2001 From: Matthew McKeen Date: Wed, 8 Jul 2026 20:12:09 -0700 Subject: [PATCH 4/6] test(conntrack,packetparser): unit test reason classification Signed-off-by: Matthew McKeen --- pkg/plugin/conntrack/conntrack_linux_test.go | 25 +++++++++++++++++++ .../packetparser/packetparser_linux_test.go | 19 ++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/pkg/plugin/conntrack/conntrack_linux_test.go b/pkg/plugin/conntrack/conntrack_linux_test.go index 971ddaead5..e4b0cc1eb0 100644 --- a/pkg/plugin/conntrack/conntrack_linux_test.go +++ b/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) + } + }) + } +} diff --git a/pkg/plugin/packetparser/packetparser_linux_test.go b/pkg/plugin/packetparser/packetparser_linux_test.go index 4fa4a5394c..8e55861959 100644 --- a/pkg/plugin/packetparser/packetparser_linux_test.go +++ b/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)) + } +} From d261916b7e45d41fbbd6fb2b09b39a3619b8fe24 Mon Sep 17 00:00:00 2001 From: Matthew McKeen Date: Thu, 9 Jul 2026 14:55:27 -0700 Subject: [PATCH 5/6] feat(conntrack): add unknown-direction connection and byte gauges Signed-off-by: Matthew McKeen --- pkg/metrics/metrics.go | 12 ++++++++++++ pkg/metrics/types.go | 6 ++++++ pkg/plugin/conntrack/conntrack_linux.go | 10 ++++++++++ pkg/utils/metric_names.go | 3 +++ 4 files changed, 31 insertions(+) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index dc830ed659..98cbe26df6 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -192,6 +192,18 @@ 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, diff --git a/pkg/metrics/types.go b/pkg/metrics/types.go index 9f162d7cb7..2200ecfb78 100644 --- a/pkg/metrics/types.go +++ b/pkg/metrics/types.go @@ -60,6 +60,9 @@ const ( 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 @@ -117,6 +120,9 @@ var ( ConntrackBytesTx GaugeVec ConntrackBytesRx GaugeVec ConntrackTotalConnections GaugeVec + + ConntrackUnknownDirectionConnections GaugeVec + ConntrackUnknownDirectionBytes GaugeVec ) func ToPrometheusType(metric interface{}) prometheus.Collector { diff --git a/pkg/plugin/conntrack/conntrack_linux.go b/pkg/plugin/conntrack/conntrack_linux.go index 9b0f12c08d..542ee88034 100644 --- a/pkg/plugin/conntrack/conntrack_linux.go +++ b/pkg/plugin/conntrack/conntrack_linux.go @@ -156,6 +156,10 @@ func (ct *Conntrack) Run(ctx context.Context) error { // 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) { @@ -182,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", @@ -210,6 +218,8 @@ 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 diff --git a/pkg/utils/metric_names.go b/pkg/utils/metric_names.go index 5bb7f280d3..eb56035b42 100644 --- a/pkg/utils/metric_names.go +++ b/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 From a0110bb8b83dc7b52fb9d4823f3d50aad0ce61e0 Mon Sep 17 00:00:00 2001 From: Matthew McKeen Date: Sun, 19 Jul 2026 17:34:45 -0700 Subject: [PATCH 6/6] feat(conntrack): measure since-last-report bytes/packets lost on GC reap Signed-off-by: Matthew McKeen --- pkg/metrics/metrics.go | 14 +++++++++ pkg/metrics/types.go | 18 ++++++++---- pkg/plugin/conntrack/conntrack_linux.go | 38 ++++++++++++++++++++++++- pkg/utils/flow_utils.go | 30 +++++++++++++++++++ 4 files changed, 93 insertions(+), 7 deletions(-) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 98cbe26df6..4612b58690 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -218,6 +218,20 @@ func InitializeMetrics(logger *slog.Logger) { reasonLabel, ) + ConntrackGCUnreportedBytesCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric( + exporter.DefaultRegistry, + conntrackGCUnreportedBytesCounterName, + conntrackGCUnreportedBytesCounterDescription, + utils.Direction, + ) + + ConntrackGCUnreportedPacketsCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric( + exporter.DefaultRegistry, + conntrackGCUnreportedPacketsCounterName, + conntrackGCUnreportedPacketsCounterDescription, + utils.Direction, + ) + MetricsExpiredCounter = exporter.CreatePrometheusCounterVecForControlPlaneMetric( exporter.DefaultRegistry, expiredMetricsCounterName, diff --git a/pkg/metrics/types.go b/pkg/metrics/types.go index 2200ecfb78..bbdfe3130d 100644 --- a/pkg/metrics/types.go +++ b/pkg/metrics/types.go @@ -15,6 +15,8 @@ const ( 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. @@ -54,12 +56,14 @@ 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" - conntrackGCEntriesCounterDescription = "Number of conntrack entries removed by garbage collection, by connection state" + 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" @@ -106,6 +110,8 @@ var ( ParsedPacketsCounter CounterVec MetricsExpiredCounter CounterVec ConntrackGCEntriesCounter CounterVec + ConntrackGCUnreportedBytesCounter CounterVec + ConntrackGCUnreportedPacketsCounter CounterVec // DNS Metrics. DNSRequestCounter CounterVec diff --git a/pkg/plugin/conntrack/conntrack_linux.go b/pkg/plugin/conntrack/conntrack_linux.go index 542ee88034..e47c77c9f9 100644 --- a/pkg/plugin/conntrack/conntrack_linux.go +++ b/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" @@ -103,6 +104,15 @@ func GenerateDynamic(ctx context.Context, dynamicHeaderPath string, conntrackMet 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 @@ -152,6 +162,11 @@ func (ct *Conntrack) Run(ctx context.Context) error { var noOfCtEntries, entriesDeleted int // 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 @@ -169,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, gcDeletion{key: keyCopy, reason: gcEntryReason(key.Proto, &value)}) + 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() @@ -231,8 +254,21 @@ func (ct *Conntrack) Run(ctx context.Context) error { } 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)) } } diff --git a/pkg/utils/flow_utils.go b/pkg/utils/flow_utils.go index 6d3da9c243..df6b5b4d44 100644 --- a/pkg/utils/flow_utils.go +++ b/pkg/utils/flow_utils.go @@ -39,6 +39,36 @@ const ( TypeUrl string = "retina.sh" //nolint:revive,stylecheck // existing API, renaming would break callers ) +// OppositeTrafficDirection returns the reply-side host-relative direction: the +// reply of an egress connection is ingress and vice versa; unknown has no defined +// opposite and maps to itself. +func OppositeTrafficDirection(d flow.TrafficDirection) flow.TrafficDirection { + switch d { + case flow.TrafficDirection_INGRESS: + return flow.TrafficDirection_EGRESS + case flow.TrafficDirection_EGRESS: + return flow.TrafficDirection_INGRESS + case flow.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN: + return flow.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN + } + return flow.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN +} + +// TrafficDirectionString maps a flow.TrafficDirection to its lowercase metric +// label ("ingress"/"egress"/"unknown"), so callers reporting direction-labelled +// metrics share one vocabulary. +func TrafficDirectionString(d flow.TrafficDirection) string { + switch d { + case flow.TrafficDirection_INGRESS: + return "ingress" + case flow.TrafficDirection_EGRESS: + return "egress" + case flow.TrafficDirection_TRAFFIC_DIRECTION_UNKNOWN: + return "unknown" + } + return "unknown" +} + // ToFlow returns a flow.Flow object. // This sets up a L3/L4 flow object. // sourceIP, destIP are IPv4 or IPv6 addresses; IpVersion is derived from the