Skip to content

refactor(packetparser): reduce allocations in the hot path #2508

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Matthew McKeen (mmckeen) wants to merge 2 commits into
base: main
Choose a base branch
from
Open

refactor(packetparser): reduce allocations in the hot path #2508

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
79ee3a5
perf(metrics): decode flow extensions once per flow
mmckeen Jul 9, 2026
91fe5fd
perf(conntrack,cache): gate hot-loop debug logs behind level check
mmckeen Jul 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 18 additions & 6 deletions pkg/controllers/cache/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
}
Expand Down
27 changes: 14 additions & 13 deletions pkg/module/metrics/dns.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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
}
Expand All @@ -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
}
Expand All @@ -139,23 +140,23 @@ 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))
}
return labels, nil
}

func (d *DNSMetrics) ProcessFlow(flow *v1.Flow) {
func (d *DNSMetrics) ProcessFlow(flow *v1.Flow, ext *structpb.Struct) {
if flow == nil {
return
}
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
10 changes: 5 additions & 5 deletions pkg/module/metrics/dns_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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")
Expand All @@ -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)
Expand Down
23 changes: 12 additions & 11 deletions pkg/module/metrics/drops.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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 {
Expand All @@ -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
}

Expand All @@ -134,31 +135,31 @@ 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))
}

if l := len(labelValuesMap[egress]); l > 0 {
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))
}
}
Expand All @@ -174,15 +175,15 @@ 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:
updated = true
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)
Expand Down
8 changes: 5 additions & 3 deletions pkg/module/metrics/drops_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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)
Expand All @@ -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)

Expand Down
21 changes: 11 additions & 10 deletions pkg/module/metrics/forward.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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 {
Expand All @@ -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
}

Expand All @@ -123,7 +124,7 @@ func (f *ForwardMetrics) ProcessFlow(flow *v1.Flow) {
}

if !f.isAdvanced() {
f.update(flow, labels)
f.update(ext, labels)
return
}

Expand All @@ -145,26 +146,26 @@ 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
}
// 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))
}
}
Expand All @@ -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)
Expand Down
Loading
Loading