From 79ee3a50eb6f9fc169f2f9174cf19712364dc973 Mon Sep 17 00:00:00 2001 From: Matthew McKeen Date: Thu, 9 Jul 2026 00:55:58 -0700 Subject: [PATCH 1/2] perf(metrics): decode flow extensions once per flow The metrics dispatch and each metric module independently re-unmarshaled the flow's structpb extensions on every accessor call, so a single flow's extensions were decoded ~4-5 times across the forward/tcpflags/latency/dns/drops modules. Decode the extensions once in the dispatch loop and thread the *structpb.Struct through ProcessFlow, reading fields via new utils.*FromStruct accessors. The existing flow-based accessors delegate to these, so behavior is unchanged; only the redundant per-flow unmarshals are removed. Signed-off-by: Matthew McKeen --- pkg/module/metrics/dns.go | 27 ++++++++-------- pkg/module/metrics/dns_test.go | 10 +++--- pkg/module/metrics/drops.go | 23 ++++++------- pkg/module/metrics/drops_test.go | 8 +++-- pkg/module/metrics/forward.go | 21 ++++++------ pkg/module/metrics/forward_test.go | 8 +++-- pkg/module/metrics/latency.go | 17 ++++++---- pkg/module/metrics/latency_test.go | 10 +++--- pkg/module/metrics/metrics_module.go | 3 +- pkg/module/metrics/mock_types.go | 9 +++--- pkg/module/metrics/tcpflags.go | 15 +++++---- pkg/module/metrics/tcpflags_test.go | 8 +++-- pkg/module/metrics/tcpretrans.go | 3 +- pkg/module/metrics/types.go | 5 ++- pkg/utils/flow_utils.go | 48 ++++++++++++++++------------ pkg/utils/utils_linux_test.go | 4 +-- 16 files changed, 124 insertions(+), 95 deletions(-) diff --git a/pkg/module/metrics/dns.go b/pkg/module/metrics/dns.go index 96f4886951..f7ec3c3978 100644 --- a/pkg/module/metrics/dns.go +++ b/pkg/module/metrics/dns.go @@ -17,6 +17,7 @@ import ( "github.com/microsoft/retina/pkg/utils" "github.com/pkg/errors" "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" ) const ( @@ -100,8 +101,8 @@ func (d *DNSMetrics) getResponseLabels() []string { return labels } -func (d *DNSMetrics) requestValues(flow *v1.Flow) []string { - flowDNS, dnsType, _ := utils.GetDNS(flow) +func (d *DNSMetrics) requestValues(flow *v1.Flow, ext *structpb.Struct) []string { + flowDNS, dnsType, _ := utils.GetDNSFromStruct(flow, ext) if flowDNS == nil { return nil } @@ -118,8 +119,8 @@ func (d *DNSMetrics) requestValues(flow *v1.Flow) []string { return labels } -func (d *DNSMetrics) responseValues(flow *v1.Flow) []string { - flowDNS, dnsType, numResponses := utils.GetDNS(flow) +func (d *DNSMetrics) responseValues(flow *v1.Flow, ext *structpb.Struct) []string { + flowDNS, dnsType, numResponses := utils.GetDNSFromStruct(flow, ext) if flowDNS == nil { return nil } @@ -139,15 +140,15 @@ func (d *DNSMetrics) responseValues(flow *v1.Flow) []string { return labels } -func (d *DNSMetrics) getLabelsForProcessFlow(flow *v1.Flow) ([]string, error) { +func (d *DNSMetrics) getLabelsForProcessFlow(flow *v1.Flow, ext *structpb.Struct) ([]string, error) { var labels []string // Get the DNS query type - _, dnsType, _ := utils.GetDNS(flow) + _, dnsType, _ := utils.GetDNSFromStruct(flow, ext) switch dnsType { case utils.DNSType_QUERY: - labels = d.requestValues(flow) + labels = d.requestValues(flow, ext) case utils.DNSType_RESPONSE: - labels = d.responseValues(flow) + labels = d.responseValues(flow, ext) case utils.DNSType_UNKNOWN: default: return labels, errors.Errorf("invalid DNS type %d", int32(dnsType)) @@ -155,7 +156,7 @@ func (d *DNSMetrics) getLabelsForProcessFlow(flow *v1.Flow) ([]string, error) { return labels, nil } -func (d *DNSMetrics) ProcessFlow(flow *v1.Flow) { +func (d *DNSMetrics) ProcessFlow(flow *v1.Flow, ext *structpb.Struct) { if flow == nil { return } @@ -167,11 +168,11 @@ func (d *DNSMetrics) ProcessFlow(flow *v1.Flow) { if d.isLocalContext() { // when localcontext is enabled, we do not need the context options for both src and dst // metrics aggregation will be on a single pod basis and not the src/dst pod combination basis. - d.processLocalCtxFlow(flow) + d.processLocalCtxFlow(flow, ext) return } - labels, err := d.getLabelsForProcessFlow(flow) + labels, err := d.getLabelsForProcessFlow(flow, ext) if err != nil { d.getLogger().Error("Failed to get labels for process flow", zap.Error(err)) return @@ -199,13 +200,13 @@ func (d *DNSMetrics) ProcessFlow(flow *v1.Flow) { d.getLogger().Debug("Update dns metric in remote ctx", zap.Any("metric", d.dnsMetrics), zap.Any("labels", labels)) } -func (d *DNSMetrics) processLocalCtxFlow(flow *v1.Flow) { +func (d *DNSMetrics) processLocalCtxFlow(flow *v1.Flow, ext *structpb.Struct) { labelValuesMap := d.sourceCtx().getLocalCtxValues(flow) if labelValuesMap == nil { return } - labels, err := d.getLabelsForProcessFlow(flow) + labels, err := d.getLabelsForProcessFlow(flow, ext) if err != nil { d.getLogger().Error("Failed to get labels for process flow", zap.Error(err)) return diff --git a/pkg/module/metrics/dns_test.go b/pkg/module/metrics/dns_test.go index 7bf582139c..fa9233fc1c 100644 --- a/pkg/module/metrics/dns_test.go +++ b/pkg/module/metrics/dns_test.go @@ -172,15 +172,15 @@ func TestValues(t *testing.T) { t.Run(tt.name, func(t *testing.T) { switch tt.l7Type { case flow.L7FlowType_REQUEST: - if got := tt.d.requestValues(tt.input); !reflect.DeepEqual(got, tt.want) { + if got := tt.d.requestValues(tt.input, utils.GetExtensionsStruct(tt.input)); !reflect.DeepEqual(got, tt.want) { t.Errorf("RequestValues() = %v, want %v", got, tt.want) } case flow.L7FlowType_RESPONSE: - if got := tt.d.responseValues(tt.input); !reflect.DeepEqual(got, tt.want) { + if got := tt.d.responseValues(tt.input, utils.GetExtensionsStruct(tt.input)); !reflect.DeepEqual(got, tt.want) { t.Errorf("ResponseValues() = %v, want %v", got, tt.want) } case flow.L7FlowType_UNKNOWN_L7_TYPE: - if got := tt.d.responseValues(tt.input); !reflect.DeepEqual(got, tt.want) { + if got := tt.d.responseValues(tt.input, utils.GetExtensionsStruct(tt.input)); !reflect.DeepEqual(got, tt.want) { t.Errorf("ResponseValues() = %v, want %v", got, tt.want) } case flow.L7FlowType_SAMPLE: @@ -300,7 +300,7 @@ func TestProcessLocalCtx(t *testing.T) { d := NewDNSMetrics(ctxOptions, l, localContext, 0) d.dnsMetrics = mockCV - d.ProcessFlow(tt.input) + d.ProcessFlow(tt.input, utils.GetExtensionsStruct(tt.input)) // There should be no tracked metrics when TTL is infinite assert.Equal(t, 0, len(d.trackedMetricLabels()), "there should be no tracked metrics when TTL is infinite") @@ -316,7 +316,7 @@ func TestProcessLocalCtx(t *testing.T) { mockCV.EXPECT().WithLabelValues(tt.expectedLabels).Return(c).Times(1) } - d.ProcessFlow(tt.input) + d.ProcessFlow(tt.input, utils.GetExtensionsStruct(tt.input)) if tt.metricsUpdate { mockCV.EXPECT().DeleteLabelValues(tt.expectedLabels).Return(true).Times(1) diff --git a/pkg/module/metrics/drops.go b/pkg/module/metrics/drops.go index f1ed9bc345..6a9a7ad438 100644 --- a/pkg/module/metrics/drops.go +++ b/pkg/module/metrics/drops.go @@ -14,6 +14,7 @@ import ( "github.com/microsoft/retina/pkg/metrics" "github.com/microsoft/retina/pkg/utils" "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" ) const ( @@ -90,7 +91,7 @@ func (d *DropCountMetrics) Clean() { // TODO: update ProcessFlow with bytes metrics. We are only accounting for count. // bytes metrics needs some additional work in ebpf and in this func to get the skb length -func (d *DropCountMetrics) ProcessFlow(flow *v1.Flow) { +func (d *DropCountMetrics) ProcessFlow(flow *v1.Flow, ext *structpb.Struct) { // Flow does not have bytes section at the moment, // so we will update only packet count if flow == nil { @@ -104,17 +105,17 @@ func (d *DropCountMetrics) ProcessFlow(flow *v1.Flow) { if d.isLocalContext() { // when localcontext is enabled, we do not need the context options for both src and dst // metrics aggregation will be on a single pod basis and not the src/dst pod combination basis. - d.processLocalCtxFlow(flow) + d.processLocalCtxFlow(flow, ext) return } labels := []string{ - utils.DropReasonDescription(flow), + utils.DropReasonDescriptionFromStruct(ext), flow.TrafficDirection.String(), } if !d.isAdvanced() { - d.update(flow, labels) + d.update(ext, labels) return } @@ -134,23 +135,23 @@ func (d *DropCountMetrics) ProcessFlow(flow *v1.Flow) { // No additional context options - d.update(flow, labels) + d.update(ext, labels) d.getLogger().Debug("drop count metric is added", zap.Any("labels", labels)) } -func (d *DropCountMetrics) processLocalCtxFlow(flow *v1.Flow) { +func (d *DropCountMetrics) processLocalCtxFlow(flow *v1.Flow, ext *structpb.Struct) { labelValuesMap := d.sourceCtx().getLocalCtxValues(flow) if labelValuesMap == nil { return } - dropReason := utils.DropReasonDescription(flow) + dropReason := utils.DropReasonDescriptionFromStruct(ext) // Ingress values if l := len(labelValuesMap[ingress]); l > 0 { labels := make([]string, 0, l+2) labels = append(labels, dropReason, ingress) labels = append(labels, labelValuesMap[ingress]...) - d.update(flow, labels) + d.update(ext, labels) d.getLogger().Debug("drop count metric is added in INGRESS in local ctx", zap.Any("labels", labels)) } @@ -158,7 +159,7 @@ func (d *DropCountMetrics) processLocalCtxFlow(flow *v1.Flow) { labels := make([]string, 0, l+2) labels = append(labels, dropReason, egress) labels = append(labels, labelValuesMap[egress]...) - d.update(flow, labels) + d.update(ext, labels) d.getLogger().Debug("drop count metric is added in EGRESS in local ctx", zap.Any("labels", labels)) } } @@ -174,7 +175,7 @@ func (d *DropCountMetrics) expire(labels []string) bool { return del } -func (d *DropCountMetrics) update(fl *v1.Flow, labels []string) { +func (d *DropCountMetrics) update(ext *structpb.Struct, labels []string) { var updated bool switch d.metricName { case utils.DroppedPacketsGaugeName: @@ -182,7 +183,7 @@ func (d *DropCountMetrics) update(fl *v1.Flow, labels []string) { d.dropMetric.WithLabelValues(labels...).Inc() case utils.DropBytesGaugeName: updated = true - d.dropMetric.WithLabelValues(labels...).Add(float64(utils.PacketSize(fl))) + d.dropMetric.WithLabelValues(labels...).Add(float64(utils.PacketSizeFromStruct(ext))) } if updated { d.updated(labels) diff --git a/pkg/module/metrics/drops_test.go b/pkg/module/metrics/drops_test.go index b6774fae04..f1b608e596 100644 --- a/pkg/module/metrics/drops_test.go +++ b/pkg/module/metrics/drops_test.go @@ -8,15 +8,17 @@ import ( "testing" "time" + "log/slog" + "github.com/cilium/cilium/api/v1/flow" "github.com/microsoft/retina/crd/api/v1alpha1" "github.com/microsoft/retina/pkg/log" metricsinit "github.com/microsoft/retina/pkg/metrics" + "github.com/microsoft/retina/pkg/utils" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" "go.uber.org/mock/gomock" "go.uber.org/zap" - "log/slog" ) func TestNewDrop(t *testing.T) { @@ -303,7 +305,7 @@ func TestNewDrop(t *testing.T) { assert.Equal(t, tc.exepectedLabels, f.getLabels(), "labels should be equal Test Name: %s", tc.name) f.metricName = metricName - f.ProcessFlow(tc.f) + f.ProcessFlow(tc.f, utils.GetExtensionsStruct(tc.f)) // There should be no tracked metrics when TTL is infinite assert.Equal(t, 0, len(f.trackedMetricLabels()), "there should be no tracked metrics when TTL is infinite Test Name: %s", tc.name) @@ -318,7 +320,7 @@ func TestNewDrop(t *testing.T) { dropMock.EXPECT().WithLabelValues(gomock.Any()).Return(testmetric).Times(tc.metricCall) f.metricName = metricName - f.ProcessFlow(tc.f) + f.ProcessFlow(tc.f, utils.GetExtensionsStruct(tc.f)) dropMock.EXPECT().DeleteLabelValues(gomock.Any()).Return(true).Times(tc.trackedMetrics) diff --git a/pkg/module/metrics/forward.go b/pkg/module/metrics/forward.go index e52e621779..541bac797c 100644 --- a/pkg/module/metrics/forward.go +++ b/pkg/module/metrics/forward.go @@ -16,6 +16,7 @@ import ( metricsinit "github.com/microsoft/retina/pkg/metrics" "github.com/microsoft/retina/pkg/utils" "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" ) const ( @@ -100,7 +101,7 @@ func (f *ForwardMetrics) Clean() { // TODO: update ProcessFlow with bytes metrics. We are only accounting for count. // bytes metrics needs some additional work in ebpf and in this func to get the skb length -func (f *ForwardMetrics) ProcessFlow(flow *v1.Flow) { +func (f *ForwardMetrics) ProcessFlow(flow *v1.Flow, ext *structpb.Struct) { // Flow does not have bytes section at the moment, // so we will update only packet count if flow == nil { @@ -114,7 +115,7 @@ func (f *ForwardMetrics) ProcessFlow(flow *v1.Flow) { if f.isLocalContext() { // when localcontext is enabled, we do not need the context options for both src and dst // metrics aggregation will be on a single pod basis and not the src/dst pod combination basis. - f.processLocalCtxFlow(flow) + f.processLocalCtxFlow(flow, ext) return } @@ -123,7 +124,7 @@ func (f *ForwardMetrics) ProcessFlow(flow *v1.Flow) { } if !f.isAdvanced() { - f.update(flow, labels) + f.update(ext, labels) return } @@ -145,11 +146,11 @@ func (f *ForwardMetrics) ProcessFlow(flow *v1.Flow) { labels = append(labels, strconv.FormatBool(flow.GetIsReply().GetValue())) } - f.update(flow, labels) + f.update(ext, labels) f.getLogger().Debug("forward count metric is added", zap.Any("labels", labels)) } -func (f *ForwardMetrics) processLocalCtxFlow(flow *v1.Flow) { +func (f *ForwardMetrics) processLocalCtxFlow(flow *v1.Flow, ext *structpb.Struct) { labelValuesMap := f.sourceCtx().getLocalCtxValues(flow) if labelValuesMap == nil { return @@ -157,14 +158,14 @@ func (f *ForwardMetrics) processLocalCtxFlow(flow *v1.Flow) { // Ingress values. if len(labelValuesMap[ingress]) > 0 { labels := append([]string{ingress}, labelValuesMap[ingress]...) - f.update(flow, labels) + f.update(ext, labels) f.getLogger().Debug("forward count metric in INGRESS in local ctx", zap.Any("labels", labels)) } // Egress values. if len(labelValuesMap[egress]) > 0 { labels := append([]string{egress}, labelValuesMap[egress]...) - f.update(flow, labels) + f.update(ext, labels) f.getLogger().Debug("forward count metric in EGRESS in local ctx", zap.Any("labels", labels)) } } @@ -180,15 +181,15 @@ func (f *ForwardMetrics) expire(labels []string) bool { return d } -func (f *ForwardMetrics) update(fl *v1.Flow, labels []string) { +func (f *ForwardMetrics) update(ext *structpb.Struct, labels []string) { var updated bool switch f.metricName { case utils.ForwardPacketsGaugeName: updated = true - f.forwardMetric.WithLabelValues(labels...).Add(float64(utils.PreviouslyObservedPackets(fl) + 1)) + f.forwardMetric.WithLabelValues(labels...).Add(float64(utils.PreviouslyObservedPacketsFromStruct(ext) + 1)) case utils.ForwardBytesGaugeName: updated = true - f.forwardMetric.WithLabelValues(labels...).Add(float64(utils.PacketSize(fl) + utils.PreviouslyObservedBytes(fl))) + f.forwardMetric.WithLabelValues(labels...).Add(float64(utils.PacketSizeFromStruct(ext) + utils.PreviouslyObservedBytesFromStruct(ext))) } if updated { f.updated(labels) diff --git a/pkg/module/metrics/forward_test.go b/pkg/module/metrics/forward_test.go index e1a94c0314..6b4f6d56cf 100644 --- a/pkg/module/metrics/forward_test.go +++ b/pkg/module/metrics/forward_test.go @@ -8,15 +8,17 @@ import ( "testing" "time" + "log/slog" + "github.com/cilium/cilium/api/v1/flow" "github.com/microsoft/retina/crd/api/v1alpha1" "github.com/microsoft/retina/pkg/log" metricsinit "github.com/microsoft/retina/pkg/metrics" + "github.com/microsoft/retina/pkg/utils" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" "go.uber.org/mock/gomock" "go.uber.org/zap" - "log/slog" ) type TestMetrics struct { @@ -341,7 +343,7 @@ func TestNewForward(t *testing.T) { assert.Equal(t, tc.exepectedLabels, f.getLabels(), "labels should be equal Test Name: %s", tc.name) f.metricName = metricName - f.ProcessFlow(tc.f) + f.ProcessFlow(tc.f, utils.GetExtensionsStruct(tc.f)) // There should be no tracked metrics when TTL is infinite assert.Equal(t, 0, len(f.trackedMetricLabels()), "there should be no tracked metrics when TTL is infinite Test Name: %s", tc.name) @@ -356,7 +358,7 @@ func TestNewForward(t *testing.T) { forwardMock.EXPECT().WithLabelValues(gomock.Any()).Return(testmetric).Times(tc.metricCall) f.metricName = metricName - f.ProcessFlow(tc.f) + f.ProcessFlow(tc.f, utils.GetExtensionsStruct(tc.f)) forwardMock.EXPECT().DeleteLabelValues(gomock.Any()).Return(true).Times(tc.trackedMetrics) diff --git a/pkg/module/metrics/latency.go b/pkg/module/metrics/latency.go index 0fb88ec7b9..6f89599275 100644 --- a/pkg/module/metrics/latency.go +++ b/pkg/module/metrics/latency.go @@ -22,6 +22,7 @@ import ( "github.com/microsoft/retina/pkg/pubsub" "github.com/microsoft/retina/pkg/utils" "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" ) const ( @@ -186,8 +187,12 @@ func (lm *LatencyMetrics) Clean() { } } -func (lm *LatencyMetrics) ProcessFlow(f *flow.Flow) { - if f == nil || f.GetL4() == nil || f.GetL4().GetTCP() == nil || utils.GetTCPID(f) == 0 || f.GetIP() == nil { +func (lm *LatencyMetrics) ProcessFlow(f *flow.Flow, ext *structpb.Struct) { + if f == nil || f.GetL4() == nil || f.GetL4().GetTCP() == nil || f.GetIP() == nil { + return + } + tcpID := utils.TCPIDFromStruct(ext) + if tcpID == 0 { return } @@ -206,7 +211,7 @@ func (lm *LatencyMetrics) ProcessFlow(f *flow.Flow) { _, ipInDestination := apiServerIps[destinationIP] if ipInSource || ipInDestination { - lm.calculateLatency(f) + lm.calculateLatency(f, tcpID) } } } @@ -263,7 +268,7 @@ func (lm *LatencyMetrics) ProcessFlow(f *flow.Flow) { | | +-------------------------------------------------+ */ -func (lm *LatencyMetrics) calculateLatency(f *flow.Flow) { +func (lm *LatencyMetrics) calculateLatency(f *flow.Flow, tcpID uint64) { // Ignore all packets observed at endpoint. // We only care about node-apiserver packets observed at eth0. // TO_NETWORK: Packets leaving node via eth0. @@ -274,7 +279,7 @@ func (lm *LatencyMetrics) calculateLatency(f *flow.Flow) { dstIP: f.IP.Destination, srcP: f.GetL4().GetTCP().GetSourcePort(), dstP: f.GetL4().GetTCP().GetDestinationPort(), - id: utils.GetTCPID(f), + id: tcpID, } // There will be multiple identical packets with same ID. Store only the first one. if item := lm.cache.Get(k); item == nil { @@ -289,7 +294,7 @@ func (lm *LatencyMetrics) calculateLatency(f *flow.Flow) { dstIP: f.IP.Source, srcP: f.GetL4().GetTCP().GetDestinationPort(), dstP: f.GetL4().GetTCP().GetSourcePort(), - id: utils.GetTCPID(f), + id: tcpID, } if item := lm.cache.Get(k); item != nil { // Calculate latency in milliseconds. diff --git a/pkg/module/metrics/latency_test.go b/pkg/module/metrics/latency_test.go index 1a78fcb37a..33519e6626 100644 --- a/pkg/module/metrics/latency_test.go +++ b/pkg/module/metrics/latency_test.go @@ -140,8 +140,8 @@ func TestProcessFlow(t *testing.T) { PodName: "kubernetes-apiserver", } // Process flow. - lm.ProcessFlow(f1) - lm.ProcessFlow(f2) + lm.ProcessFlow(f1, utils.GetExtensionsStruct(f1)) + lm.ProcessFlow(f2, utils.GetExtensionsStruct(f2)) /* * Test case 2: Existing TCP connection. @@ -151,13 +151,13 @@ func TestProcessFlow(t *testing.T) { // Api server -> Node. utils.AddTCPFlags(f2, 0, 1, 0, 0, 0, 0, 0, 0, 0) // Process flow. - lm.ProcessFlow(f1) - lm.ProcessFlow(f2) + lm.ProcessFlow(f1, utils.GetExtensionsStruct(f1)) + lm.ProcessFlow(f2, utils.GetExtensionsStruct(f2)) /* * Test case 3: No reply from apiserver. */ - lm.ProcessFlow(f1) + lm.ProcessFlow(f1, utils.GetExtensionsStruct(f1)) // Sleep for TTL. time.Sleep(1 * time.Second) // Check dropped packet. diff --git a/pkg/module/metrics/metrics_module.go b/pkg/module/metrics/metrics_module.go index 6d473f1d49..1f81c27db8 100644 --- a/pkg/module/metrics/metrics_module.go +++ b/pkg/module/metrics/metrics_module.go @@ -294,8 +294,9 @@ func (m *Module) run(newCtx context.Context) { m.RLock() f := ev.Event.(*flow.Flow) m.l.Debug("converted flow object", zap.Any("flow l4", f.IP)) + ext := utils.GetExtensionsStruct(f) for _, metricObj := range m.registry { - metricObj.ProcessFlow(f) + metricObj.ProcessFlow(f, ext) } m.RUnlock() case *flow.LostEvent: diff --git a/pkg/module/metrics/mock_types.go b/pkg/module/metrics/mock_types.go index d25658da8c..65fe0759a9 100644 --- a/pkg/module/metrics/mock_types.go +++ b/pkg/module/metrics/mock_types.go @@ -15,6 +15,7 @@ import ( flow "github.com/cilium/cilium/api/v1/flow" v1alpha1 "github.com/microsoft/retina/crd/api/v1alpha1" gomock "go.uber.org/mock/gomock" + structpb "google.golang.org/protobuf/types/known/structpb" ) // MockIModule is a mock of IModule interface. @@ -102,15 +103,15 @@ func (mr *MockAdvMetricsInterfaceMockRecorder) Init(metricName any) *gomock.Call } // ProcessFlow mocks base method. -func (m *MockAdvMetricsInterface) ProcessFlow(f *flow.Flow) { +func (m *MockAdvMetricsInterface) ProcessFlow(f *flow.Flow, ext *structpb.Struct) { m.ctrl.T.Helper() - m.ctrl.Call(m, "ProcessFlow", f) + m.ctrl.Call(m, "ProcessFlow", f, ext) } // ProcessFlow indicates an expected call of ProcessFlow. -func (mr *MockAdvMetricsInterfaceMockRecorder) ProcessFlow(f any) *gomock.Call { +func (mr *MockAdvMetricsInterfaceMockRecorder) ProcessFlow(f, ext any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ProcessFlow", reflect.TypeOf((*MockAdvMetricsInterface)(nil).ProcessFlow), f) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ProcessFlow", reflect.TypeOf((*MockAdvMetricsInterface)(nil).ProcessFlow), f, ext) } // MockContextOptionsInterface is a mock of ContextOptionsInterface interface. diff --git a/pkg/module/metrics/tcpflags.go b/pkg/module/metrics/tcpflags.go index 9114ca4d13..0f2de1721c 100644 --- a/pkg/module/metrics/tcpflags.go +++ b/pkg/module/metrics/tcpflags.go @@ -14,6 +14,7 @@ import ( metricsinit "github.com/microsoft/retina/pkg/metrics" "github.com/microsoft/retina/pkg/utils" "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" ) const ( @@ -66,10 +67,10 @@ func (t *TCPMetrics) getLabels() []string { return labels } -func combineFlagsWithPrevious(flags []string, flow *v1.Flow) map[string]uint32 { +func combineFlagsWithPrevious(flags []string, ext *structpb.Struct) map[string]uint32 { var combinedFlags map[string]uint32 - previous := utils.PreviouslyObservedTCPFlags(flow) + previous := utils.PreviouslyObservedTCPFlagsFromStruct(ext) if previous != nil { combinedFlags = previous } else { @@ -87,7 +88,7 @@ func combineFlagsWithPrevious(flags []string, flow *v1.Flow) map[string]uint32 { return combinedFlags } -func (t *TCPMetrics) ProcessFlow(flow *v1.Flow) { +func (t *TCPMetrics) ProcessFlow(flow *v1.Flow, ext *structpb.Struct) { if flow == nil { return } @@ -109,7 +110,7 @@ func (t *TCPMetrics) ProcessFlow(flow *v1.Flow) { if t.isLocalContext() { // when localcontext is enabled, we do not need the context options for both src and dst // metrics aggregation will be on a single pod basis and not the src/dst pod combination basis. - t.processLocalCtxFlow(flow, flags) + t.processLocalCtxFlow(flow, flags, ext) return } @@ -122,7 +123,7 @@ func (t *TCPMetrics) ProcessFlow(flow *v1.Flow) { dstLabels = t.destinationCtx().getValues(flow) } - for flag, count := range combineFlagsWithPrevious(flags, flow) { + for flag, count := range combineFlagsWithPrevious(flags, ext) { labels := append([]string{flag}, srcLabels...) labels = append(labels, dstLabels...) t.update(labels, count) @@ -130,13 +131,13 @@ func (t *TCPMetrics) ProcessFlow(flow *v1.Flow) { } } -func (t *TCPMetrics) processLocalCtxFlow(flow *v1.Flow, flags []string) { +func (t *TCPMetrics) processLocalCtxFlow(flow *v1.Flow, flags []string, ext *structpb.Struct) { labelValuesMap := t.sourceCtx().getLocalCtxValues(flow) if labelValuesMap == nil { return } - combinedFlags := combineFlagsWithPrevious(flags, flow) + combinedFlags := combineFlagsWithPrevious(flags, ext) // Ingress values if l := len(labelValuesMap[ingress]); l > 0 { diff --git a/pkg/module/metrics/tcpflags_test.go b/pkg/module/metrics/tcpflags_test.go index 9be4e15b00..24a4ca788e 100644 --- a/pkg/module/metrics/tcpflags_test.go +++ b/pkg/module/metrics/tcpflags_test.go @@ -8,15 +8,17 @@ import ( "testing" "time" + "log/slog" + "github.com/cilium/cilium/api/v1/flow" "github.com/microsoft/retina/crd/api/v1alpha1" "github.com/microsoft/retina/pkg/log" metricsinit "github.com/microsoft/retina/pkg/metrics" + "github.com/microsoft/retina/pkg/utils" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" "go.uber.org/mock/gomock" "go.uber.org/zap" - "log/slog" ) func TestNewTCPMetrics(t *testing.T) { @@ -499,7 +501,7 @@ func TestNewTCPMetrics(t *testing.T) { assert.Equal(t, tc.checkIsAdvance, tcp.isAdvanced(), "IsAdvance should be %v Test Name: %s", tc.checkIsAdvance, tc.name) assert.Equal(t, tc.exepectedLabels, tcp.getLabels(), "labels should be %v Test Name: %s", tc.exepectedLabels, tc.name) - tcp.ProcessFlow(tc.f) + tcp.ProcessFlow(tc.f, utils.GetExtensionsStruct(tc.f)) assert.Equal(t, 0, len(tcp.trackedMetricLabels()), "there should be no tracked metrics when TTL is infinite Test Name: %s", tc.name) @@ -512,7 +514,7 @@ func TestNewTCPMetrics(t *testing.T) { tcpFlagMockMetrics.EXPECT().WithLabelValues(gomock.Any()).Return(testmetric).Times(tc.metricCall) - tcp.ProcessFlow(tc.f) + tcp.ProcessFlow(tc.f, utils.GetExtensionsStruct(tc.f)) tcpFlagMockMetrics.EXPECT().DeleteLabelValues(gomock.Any()).Return(true).Times(tc.trackedMetrics) diff --git a/pkg/module/metrics/tcpretrans.go b/pkg/module/metrics/tcpretrans.go index 623fbcf933..681b7bcdfe 100644 --- a/pkg/module/metrics/tcpretrans.go +++ b/pkg/module/metrics/tcpretrans.go @@ -14,6 +14,7 @@ import ( metricsinit "github.com/microsoft/retina/pkg/metrics" "github.com/microsoft/retina/pkg/utils" "go.uber.org/zap" + "google.golang.org/protobuf/types/known/structpb" ) const ( @@ -66,7 +67,7 @@ func (t *TCPRetransMetrics) getLabels() []string { return labels } -func (t *TCPRetransMetrics) ProcessFlow(flow *v1.Flow) { +func (t *TCPRetransMetrics) ProcessFlow(flow *v1.Flow, _ *structpb.Struct) { if flow == nil { return } diff --git a/pkg/module/metrics/types.go b/pkg/module/metrics/types.go index 5887c1ab5d..7f2c67ca41 100644 --- a/pkg/module/metrics/types.go +++ b/pkg/module/metrics/types.go @@ -11,6 +11,7 @@ import ( api "github.com/microsoft/retina/crd/api/v1alpha1" "github.com/microsoft/retina/pkg/common" "github.com/microsoft/retina/pkg/utils" + "google.golang.org/protobuf/types/known/structpb" ) const ( @@ -81,7 +82,9 @@ type AdvMetricsInterface interface { Init(metricName string) // This func is used to clean up old metrics on reconcile. Clean() - ProcessFlow(f *flow.Flow) + // ext is the flow's extensions decoded once by the caller so modules can + // read fields via utils.*FromStruct without re-unmarshaling per flow. + ProcessFlow(f *flow.Flow, ext *structpb.Struct) } type ContextOptionsInterface interface { diff --git a/pkg/utils/flow_utils.go b/pkg/utils/flow_utils.go index 6d3da9c243..a3122d4749 100644 --- a/pkg/utils/flow_utils.go +++ b/pkg/utils/flow_utils.go @@ -215,8 +215,9 @@ func AddPreviouslyObservedTCPFlags(s *structpb.Struct, syn, ack, fin, rst, psh, s.GetFields()[ExtKeyPrevObservedTCPFlags] = structpb.NewStructValue(tcpFlags) } -func PreviouslyObservedTCPFlags(f *flow.Flow) map[string]uint32 { - s := GetExtensionsStruct(f) +// PreviouslyObservedTCPFlagsFromStruct reads the field from an already-decoded +// extensions struct, avoiding a repeat unmarshal when several fields are read. +func PreviouslyObservedTCPFlagsFromStruct(s *structpb.Struct) map[string]uint32 { if s == nil { return nil } @@ -239,8 +240,9 @@ func AddPreviouslyObservedBytes(s *structpb.Struct, bytes uint32) { s.GetFields()[ExtKeyPrevObservedBytes] = structpb.NewNumberValue(float64(bytes)) } -func PreviouslyObservedBytes(f *flow.Flow) uint32 { - s := GetExtensionsStruct(f) +// PreviouslyObservedBytesFromStruct reads the field from an already-decoded +// extensions struct, avoiding a repeat unmarshal when several fields are read. +func PreviouslyObservedBytesFromStruct(s *structpb.Struct) uint32 { if s == nil { return 0 } @@ -259,8 +261,9 @@ func AddPreviouslyObservedPackets(s *structpb.Struct, packets uint32) { s.GetFields()[ExtKeyPrevObservedPackets] = structpb.NewNumberValue(float64(packets)) } -func PreviouslyObservedPackets(f *flow.Flow) uint32 { - s := GetExtensionsStruct(f) +// PreviouslyObservedPacketsFromStruct reads the field from an already-decoded +// extensions struct, avoiding a repeat unmarshal when several fields are read. +func PreviouslyObservedPacketsFromStruct(s *structpb.Struct) uint32 { if s == nil { return 0 } @@ -296,11 +299,9 @@ func AddTCPID(s *structpb.Struct, id uint64) { s.GetFields()[ExtKeyTCPID] = structpb.NewNumberValue(float64(id)) } -func GetTCPID(f *flow.Flow) uint64 { - if f.GetL4() == nil || f.GetL4().GetTCP() == nil { - return 0 - } - s := GetExtensionsStruct(f) +// TCPIDFromStruct reads the field from an already-decoded extensions struct, +// avoiding a repeat unmarshal when it is read alongside other fields. +func TCPIDFromStruct(s *structpb.Struct) uint64 { if s == nil { return 0 } @@ -352,12 +353,13 @@ func AddDNSInfo( } } -func GetDNS(f *flow.Flow) (*flow.DNS, DNSType, uint32) { +// GetDNSFromStruct reads the DNS type and answer count from an already-decoded +// extensions struct, avoiding a repeat unmarshal when several fields are read. +func GetDNSFromStruct(f *flow.Flow, s *structpb.Struct) (*flow.DNS, DNSType, uint32) { if f == nil || f.L7 == nil || f.L7.GetDns() == nil { return nil, DNSType_UNKNOWN, 0 } dns := f.L7.GetDns() - s := GetExtensionsStruct(f) if s == nil { return dns, DNSType_UNKNOWN, 0 } @@ -411,8 +413,9 @@ func AddPacketSize(s *structpb.Struct, packetSize uint32) { s.GetFields()[ExtKeyBytes] = structpb.NewNumberValue(float64(packetSize)) } -func PacketSize(f *flow.Flow) uint32 { - s := GetExtensionsStruct(f) +// PacketSizeFromStruct reads the field from an already-decoded extensions +// struct, avoiding a repeat unmarshal when several fields are read. +func PacketSizeFromStruct(s *structpb.Struct) uint32 { if s == nil { return 0 } @@ -446,11 +449,9 @@ func AddDropReason(f *flow.Flow, s *structpb.Struct, dropReason uint16) { } } -func DropReasonDescription(f *flow.Flow) string { - if f == nil { - return "" - } - s := GetExtensionsStruct(f) +// DropReasonDescriptionFromStruct reads the field from an already-decoded +// extensions struct, avoiding a repeat unmarshal when several fields are read. +func DropReasonDescriptionFromStruct(s *structpb.Struct) string { if s == nil { return "" } @@ -461,6 +462,13 @@ func DropReasonDescription(f *flow.Flow) string { return v.GetStringValue() } +func DropReasonDescription(f *flow.Flow) string { + if f == nil { + return "" + } + return DropReasonDescriptionFromStruct(GetExtensionsStruct(f)) +} + func decodeTime(nanoseconds int64) (pbTime *timestamppb.Timestamp, err error) { goTime, err := time.Parse(time.RFC3339Nano, time.Unix(0, nanoseconds).Format(time.RFC3339Nano)) if err != nil { diff --git a/pkg/utils/utils_linux_test.go b/pkg/utils/utils_linux_test.go index 1fecf2f7a4..6d6edbce65 100644 --- a/pkg/utils/utils_linux_test.go +++ b/pkg/utils/utils_linux_test.go @@ -90,7 +90,7 @@ func TestAddPacketSize(t *testing.T) { AddPacketSize(ext, uint32(100)) SetExtensions(fl, ext) - res := PacketSize(fl) + res := PacketSizeFromStruct(GetExtensionsStruct(fl)) assert.EqualValues(t, res, uint32(100)) } @@ -113,7 +113,7 @@ func TestTcpID(t *testing.T) { ext := NewExtensions() AddTCPID(ext, uint64(1234)) SetExtensions(fl, ext) - assert.EqualValues(t, GetTCPID(fl), uint64(1234)) + assert.EqualValues(t, TCPIDFromStruct(GetExtensionsStruct(fl)), uint64(1234)) } func TestAddDropReason(t *testing.T) { From 91fe5fd8de28a2b6b162cb9d8dce794a7cf3082a Mon Sep 17 00:00:00 2001 From: Matthew McKeen Date: Thu, 9 Jul 2026 00:56:11 -0700 Subject: [PATCH 2/2] perf(conntrack,cache): gate hot-loop debug logs behind level check The conntrack GC loop logged every map entry (up to CT_MAP_SIZE) on each tick, and the cache IP lookup logged on every flow. Both built the zap field slice and formatted IPs/strings eagerly at the call site even when debug logging is disabled, allocating on the hottest paths in production. Guard both with logger.Check(zap.DebugLevel) so the field construction and formatting are skipped entirely unless debug logging is enabled. Signed-off-by: Matthew McKeen --- pkg/controllers/cache/cache.go | 24 +++++++++++---- pkg/plugin/conntrack/conntrack_linux.go | 41 +++++++++++++------------ 2 files changed, 39 insertions(+), 26 deletions(-) diff --git a/pkg/controllers/cache/cache.go b/pkg/controllers/cache/cache.go index 431269597b..302d263dce 100644 --- a/pkg/controllers/cache/cache.go +++ b/pkg/controllers/cache/cache.go @@ -112,37 +112,49 @@ func (c *Cache) getObjByIPType(ip string, t objectType) interface{} { case TypeEndpoint: podKey, ok := c.ipToEpKey[ip] if !ok { - c.l.Debug("pod not found for IP", zap.String("ip", ip)) + if ce := c.l.Check(zap.DebugLevel, "pod not found for IP"); ce != nil { + ce.Write(zap.String("ip", ip)) + } return nil } ep, ok := c.epMap[podKey] if ok { - c.l.Debug("pod found for IP", zap.String("ip", ip), zap.String("pod", podKey)) + if ce := c.l.Check(zap.DebugLevel, "pod found for IP"); ce != nil { + ce.Write(zap.String("ip", ip), zap.String("pod", podKey)) + } return ep } case TypeSvc: svcKey, ok := c.ipToSvcKey[ip] if !ok { - c.l.Debug("service not found for IP", zap.String("ip", ip)) + if ce := c.l.Check(zap.DebugLevel, "service not found for IP"); ce != nil { + ce.Write(zap.String("ip", ip)) + } return nil } svc, ok := c.svcMap[svcKey] if ok { - c.l.Debug("service found for IP", zap.String("ip", ip), zap.String("svc", svcKey)) + if ce := c.l.Check(zap.DebugLevel, "service found for IP"); ce != nil { + ce.Write(zap.String("ip", ip), zap.String("svc", svcKey)) + } return svc } case TypeNode: nodeName, ok := c.ipToNodeName[ip] if !ok { - c.l.Debug("node not found for IP", zap.String("ip", ip)) + if ce := c.l.Check(zap.DebugLevel, "node not found for IP"); ce != nil { + ce.Write(zap.String("ip", ip)) + } return nil } node, ok := c.nodeMap[nodeName] if ok { - c.l.Debug("node found for IP", zap.String("ip", ip), zap.String("node", nodeName)) + if ce := c.l.Check(zap.DebugLevel, "node found for IP"); ce != nil { + ce.Write(zap.String("ip", ip), zap.String("node", nodeName)) + } return node } } diff --git a/pkg/plugin/conntrack/conntrack_linux.go b/pkg/plugin/conntrack/conntrack_linux.go index 4e1eb41cf3..7029dfcea7 100644 --- a/pkg/plugin/conntrack/conntrack_linux.go +++ b/pkg/plugin/conntrack/conntrack_linux.go @@ -139,12 +139,6 @@ func (ct *Conntrack) Run(ctx context.Context) error { keyCopy := key // Copy the key to avoid using the same key in the next iteration keysToDelete = append(keysToDelete, keyCopy) } - // Log the conntrack entry - srcIP := utils.Int2ip(key.SrcIp).To4() - dstIP := utils.Int2ip(key.DstIp).To4() - sourcePortShort := uint32(utils.HostToNetShort(key.SrcPort)) - destinationPortShort := uint32(utils.HostToNetShort(key.DstPort)) - // Add conntrack metrics. if conntrackMetricsEnabled { // Basic metrics, node-level @@ -156,20 +150,27 @@ func (ct *Conntrack) Run(ctx context.Context) error { packetsCountRx += ctMeta.PacketsRxCount } - ct.l.Debug("conntrack entry", - zap.String("src_ip", srcIP.String()), - zap.Uint32("src_port", sourcePortShort), - zap.String("dst_ip", dstIP.String()), - zap.Uint32("dst_port", destinationPortShort), - zap.String("proto", decodeProto(key.Proto)), - zap.Uint32("eviction_time", value.EvictionTime), - zap.Uint8("traffic_direction", value.TrafficDirection), - zap.String("flags_seen_tx_dir", decodeFlags(value.FlagsSeenTxDir)), - zap.String("flags_seen_rx_dir", decodeFlags(value.FlagsSeenRxDir)), - zap.Uint32("last_reported_tx_dir", value.LastReportTxDir), - zap.Uint32("last_reported_rx_dir", value.LastReportRxDir), - zap.Bool("is_direction_unknown", value.IsDirectionUnknown), - ) + // Gate the per-entry log behind a level check so the IP/string + // formatting and field allocation are skipped in this hot loop + // when debug logging is disabled. + if ce := ct.l.Check(zap.DebugLevel, "conntrack entry"); ce != nil { + srcIP := utils.Int2ip(key.SrcIp).To4() + dstIP := utils.Int2ip(key.DstIp).To4() + ce.Write( + zap.String("src_ip", srcIP.String()), + zap.Uint32("src_port", uint32(utils.HostToNetShort(key.SrcPort))), + zap.String("dst_ip", dstIP.String()), + zap.Uint32("dst_port", uint32(utils.HostToNetShort(key.DstPort))), + zap.String("proto", decodeProto(key.Proto)), + zap.Uint32("eviction_time", value.EvictionTime), + zap.Uint8("traffic_direction", value.TrafficDirection), + zap.String("flags_seen_tx_dir", decodeFlags(value.FlagsSeenTxDir)), + zap.String("flags_seen_rx_dir", decodeFlags(value.FlagsSeenRxDir)), + zap.Uint32("last_reported_tx_dir", value.LastReportTxDir), + zap.Uint32("last_reported_rx_dir", value.LastReportRxDir), + zap.Bool("is_direction_unknown", value.IsDirectionUnknown), + ) + } } if err := iter.Err(); err != nil { ct.l.Error("Iterate failed", zap.Error(err))