diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index d0f62df14b..98cbe26df6 100644 --- a/pkg/metrics/metrics.go +++ b/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( diff --git a/pkg/metrics/types.go b/pkg/metrics/types.go index 54f8446bb5..2200ecfb78 100644 --- a/pkg/metrics/types.go +++ b/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 { diff --git a/pkg/plugin/conntrack/_cprog/conntrack.c b/pkg/plugin/conntrack/_cprog/conntrack.c index f5fb2b3729..b09888285a 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 } @@ -494,13 +497,16 @@ 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 } @@ -508,6 +514,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 +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); 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/conntrack/conntrack_linux.go b/pkg/plugin/conntrack/conntrack_linux.go index 4e1eb41cf3..542ee88034 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,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)) 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/_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)) 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)) + } +} 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