From 1627c6969d5bb1e36de98821855984d44ad48e00 Mon Sep 17 00:00:00 2001
From: bhagathkrishnacdac <bhagath.krishna@cdac.in>
Date: Mon, 18 May 2026 17:09:49 +0530
Subject: [PATCH] feat: add per-UE throughput PM counter

---
 Dockerfile_CDAC                 |  2 +-
 pfcpiface/bess.go               | 13 ++++++++++---
 pfcpiface/metrics/interface.go  |  8 ++++++++
 pfcpiface/metrics/prometheus.go | 18 ++++++++++++++++++
 pfcpiface/telemetry.go          |  5 +++++
 5 files changed, 42 insertions(+), 4 deletions(-)

diff --git a/Dockerfile_CDAC b/Dockerfile_CDAC
index b754dbf49..2e2c4b4b3 100644
--- a/Dockerfile_CDAC
+++ b/Dockerfile_CDAC
@@ -3,7 +3,7 @@
 # Copyright 2019-present Intel Corporation
 
 # Stage bess-build: fetch BESS dependencies & pre-reqs
-FROM docker.io/cdac5gc/bess_build:260417 AS bess-build
+FROM docker.io/cdac5gc/bess_build:260518 AS bess-build
 ARG CPU=native
 ARG BESS_COMMIT=cdacmaster
 ENV PLUGINS_DIR=plugins
diff --git a/pfcpiface/bess.go b/pfcpiface/bess.go
index f482a323d..9f4ccdae4 100644
--- a/pfcpiface/bess.go
+++ b/pfcpiface/bess.go
@@ -557,7 +557,7 @@ func (b *bess) SessionStats(pc *PfcpNodeCollector, ch chan<- prometheus.Metric)
 	}
 
 	// Prepare session stats.
-	createStats := func(preResp, postResp *pb.FlowMeasureReadResponse) {
+	createStats := func(preResp, postResp *pb.FlowMeasureReadResponse, direction string) {
 		for i := 0; i < len(postResp.Statistics); i++ {
 			var pre *pb.FlowMeasureReadResponse_Statistic
 
@@ -605,6 +605,13 @@ func (b *bess) SessionStats(pc *PfcpNodeCollector, ch chan<- prometheus.Metric)
 				pdrString,
 				ueIpString,
 			)
+			ch <- prometheus.MustNewConstMetric(
+				pc.ueTrafficBytes,       // New descriptor
+				prometheus.CounterValue, // Counter type
+				float64(post.TotalBytes),
+				ueIpString, // Label: ue_ip
+				direction,  // Label: direction
+			)
 			ch <- prometheus.MustNewConstMetric(
 				pc.sessionRxPackets,
 				prometheus.GaugeValue,
@@ -650,8 +657,8 @@ func (b *bess) SessionStats(pc *PfcpNodeCollector, ch chan<- prometheus.Metric)
 		}
 	}
 
-	createStats(&qosStatsInResp, &postUlQosStatsResp)
-	createStats(&qosStatsInResp, &postDlQosStatsResp)
+	createStats(&qosStatsInResp, &postUlQosStatsResp, "uplink")
+	createStats(&qosStatsInResp, &postDlQosStatsResp, "downlink")
 
 	return
 }
diff --git a/pfcpiface/metrics/interface.go b/pfcpiface/metrics/interface.go
index 3526cf894..b955dc162 100644
--- a/pfcpiface/metrics/interface.go
+++ b/pfcpiface/metrics/interface.go
@@ -37,6 +37,13 @@ type Session struct {
 	Duration  float64
 }
 
+type UETraffic struct {
+	NodeID    string
+	UEIP      string
+	Direction string // "uplink" or "downlink"
+	Bytes     uint64
+}
+
 func NewSession(nodeID string) *Session {
 	return &Session{
 		NodeID:    nodeID,
@@ -51,5 +58,6 @@ func (s *Session) Delete() {
 type InstrumentPFCP interface {
 	SaveMessages(m *Message)
 	SaveSessions(s *Session)
+	SaveUEThroughput(t *UETraffic)
 	Stop() error
 }
diff --git a/pfcpiface/metrics/prometheus.go b/pfcpiface/metrics/prometheus.go
index e07a6c601..23c453fbd 100644
--- a/pfcpiface/metrics/prometheus.go
+++ b/pfcpiface/metrics/prometheus.go
@@ -15,6 +15,8 @@ type Service struct {
 
 	sessions        *prometheus.GaugeVec
 	sessionDuration *prometheus.HistogramVec
+
+	ueThroughput *prometheus.CounterVec
 }
 
 func NewPrometheusService() (*Service, error) {
@@ -68,12 +70,23 @@ func NewPrometheusService() (*Service, error) {
 		return nil, err
 	}
 
+	ueThroughput := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "upf_ue_throughput_bytes",
+		Help: "Total bytes transferred per UE",
+	}, []string{"node_id", "ue_ip", "direction"})
+
+	if err := prometheus.Register(ueThroughput); err != nil {
+		return nil, err
+	}
+
 	s := &Service{
 		msgCount:    msgCount,
 		msgDuration: msgDuration,
 
 		sessions:        sessions,
 		sessionDuration: sessionDuration,
+
+		ueThroughput: ueThroughput,
 	}
 
 	return s, nil
@@ -94,11 +107,16 @@ func (s *Service) SaveSessions(sess *Session) {
 	s.sessionDuration.WithLabelValues(sess.NodeID).Observe(sess.Duration)
 }
 
+func (s *Service) SaveUEThroughput(t *UETraffic) {
+	s.ueThroughput.WithLabelValues(t.NodeID, t.UEIP, t.Direction).Add(float64(t.Bytes))
+}
+
 func (s *Service) Stop() error {
 	prometheus.Unregister(s.msgCount)
 	prometheus.Unregister(s.msgDuration)
 	prometheus.Unregister(s.sessions)
 	prometheus.Unregister(s.sessionDuration)
+	prometheus.Unregister(s.ueThroughput)
 
 	return nil
 }
diff --git a/pfcpiface/telemetry.go b/pfcpiface/telemetry.go
index d06133858..55f99fc31 100644
--- a/pfcpiface/telemetry.go
+++ b/pfcpiface/telemetry.go
@@ -127,6 +127,7 @@ type PfcpNodeCollector struct {
 	sessionRxPackets      *prometheus.Desc
 	sessionDroppedPackets *prometheus.Desc
 	sessionTxBytes        *prometheus.Desc
+	ueTrafficBytes        *prometheus.Desc
 }
 
 func NewPFCPNodeCollector(node *PFCPNode) *PfcpNodeCollector {
@@ -148,6 +149,10 @@ func NewPFCPNodeCollector(node *PFCPNode) *PfcpNodeCollector {
 			"Shows the total number of packets received for a given session in UPF",
 			[]string{"fseid", "pdr", "ue_ip"}, nil,
 		),
+		ueTrafficBytes: prometheus.NewDesc(prometheus.BuildFQName("upf", "ue", "traffic_bytes"),
+			"Total bytes transferred per UE and direction",
+			[]string{"ue_ip", "direction"}, nil,
+		),
 		sessionDroppedPackets: prometheus.NewDesc(prometheus.BuildFQName("upf", "session", "dropped_packets"),
 			"Shows the number of packets dropped for a given session in UPF",
 			[]string{"fseid", "pdr", "ue_ip"}, nil,