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/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/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))
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) {