From 6f981b05f8dd9e1e2c373799f65d0be98ce1583a Mon Sep 17 00:00:00 2001 From: ReshmaJS Date: Fri, 15 May 2026 14:33:37 +0530 Subject: [PATCH 1/3] fixes except nc and complicated --- conf/parser.py | 3 + conf/ports.py | 145 ++++++++-------- conf/test_route_control.py | 22 +-- pfcpiface/config.go | 28 ++- pfcpiface/conn.go | 5 +- pfcpiface/ip_pool_test.go | 18 +- pfcpiface/messages_conn.go | 16 +- pfcpiface/node.go | 8 +- pfcpiface/p4rt_translator.go | 31 ++-- pfcpiface/parse_far.go | 172 ++++++++----------- pfcpiface/parse_far_test.go | 7 +- pfcpiface/parse_pdr.go | 264 ++++++++++++++++------------- pfcpiface/parse_pdr_test.go | 140 ++++++++++----- pfcpiface/parse_sdf.go | 32 +++- pfcpiface/parse_sdf_test.go | 216 +++++++++++++++-------- pfcpiface/session_qer.go | 35 ++-- pfcpiface/up4.go | 13 +- pfcpiface/upf.go | 23 ++- pkg/fake_bess/fake_bess.go | 18 +- pkg/fake_bess/fake_bess_service.go | 250 ++++++++++++++++----------- ptf/lib/trex_utils.py | 67 +++++--- test/integration/basic_test.go | 27 +-- test/integration/framework.go | 19 ++- test/integration/verify_up4.go | 48 ++++-- 24 files changed, 960 insertions(+), 647 deletions(-) diff --git a/conf/parser.py b/conf/parser.py index 7b98f6953..332670500 100644 --- a/conf/parser.py +++ b/conf/parser.py @@ -179,7 +179,10 @@ def parse(self, ifaces): "/tmp/notifycp" ) ) + self._parse_post_unix_config() + def _parse_post_unix_config(self): + """Parses Unix socket paths, feature flags, table sizes, and monitoring config.""" # UnixPort Paths try: self.endmarker_sockaddr = self.conf["endmarker_sockaddr"] diff --git a/conf/ports.py b/conf/ports.py index f7a2dd9bf..e790ba51f 100644 --- a/conf/ports.py +++ b/conf/ports.py @@ -150,76 +150,9 @@ def init_port(self, idx, conf_mode): self.num_q = num_q print("Setting up port {} on worker ids {}".format(name, self.workers)) - # Detect the mode of this interface - DPDK/AF_XDP/AF_PACKET - if conf_mode is None: - conf_mode = self.detect_mode() - - if conf_mode not in ["af_xdp", "linux", "dpdk", "af_packet", "sim", "cndp"]: - raise Exception("Invalid mode: {} selected.".format(conf_mode)) - - if conf_mode in ["af_xdp", "linux"]: - try: - # Initialize kernel datapath. - # AF_XDP requires that num_rx_qs == num_tx_qs - kwargs = { - "vdev": "net_af_xdp{},iface={},start_queue=0,queue_count={}".format( - idx, name, num_q - ), - "num_out_q": num_q, - "num_inc_q": num_q, - } - self.init_datapath(**kwargs) - except: - if conf_mode == "linux": - print( - "Failed to create AF_XDP socket for {}. Retrying with AF_PACKET socket...".format( - name - ) - ) - conf_mode = "af_packet" - else: - print( - "Failed to create AF_XDP socket for {}. Exiting...".format(name) - ) - sys.exit() + conf_mode = self._setup_initial_mode(idx, name, num_q, conf_mode) - if conf_mode == "cndp": - try: - # Initialize kernel fastpath. - self.init_datapath(cndp=True) - except: - print( - "Failed to create CNDP/AF_XDP socket for {}. Exiting...".format( - name - ) - ) - sys.exit() - - if conf_mode == "af_packet": - try: - # Initialize kernel datapath - kwargs = { - "vdev": "net_af_packet{},iface={},qpairs={}".format( - idx, name, num_q - ), - "num_out_q": num_q, - "num_inc_q": num_q, - } - self.init_datapath(**kwargs) - except: - print( - "Failed to create AF_PACKET socket for {}. Exiting...".format(name) - ) - sys.exit() - - if conf_mode == "sim": - self.fpi = Source(name="{}_source".format(name)) - self.fpo = Sink(name="{}_out".format(name)) - self.bpf = BPF(name="{}FastBPF".format(name)) - self.bpf.clear() - - # Attach datapath to worker's root TC - self.fpi.attach_task(wid=0) + self._setup_kernel_modes(idx, name, num_q, conf_mode) if conf_mode == "dpdk": kwargs = None @@ -312,6 +245,80 @@ def init_port(self, idx, conf_mode): # Finall set conf mode self.mode = conf_mode + def _setup_initial_mode(self, idx, name, num_q, conf_mode): + """Detects mode and initializes AF_XDP/Linux datapath.""" + # Detect the mode of this interface - DPDK/AF_XDP/AF_PACKET + if conf_mode is None: + conf_mode = self.detect_mode() + + if conf_mode not in ["af_xdp", "linux", "dpdk", "af_packet", "sim", "cndp"]: + raise Exception("Invalid mode: {} selected.".format(conf_mode)) + + if conf_mode in ["af_xdp", "linux"]: + try: + # Initialize kernel datapath. + # AF_XDP requires that num_rx_qs == num_tx_qs + kwargs = { + "vdev": "net_af_xdp{},iface={},start_queue=0,queue_count={}".format( + idx, name, num_q + ), + "num_out_q": num_q, + "num_inc_q": num_q, + } + self.init_datapath(**kwargs) + except: + if conf_mode == "linux": + print( + "Failed to create AF_XDP socket for {}. Retrying with AF_PACKET socket...".format( + name + ) + ) + conf_mode = "af_packet" + else: + print( + "Failed to create AF_XDP socket for {}. Exiting...".format(name) + ) + sys.exit() + return conf_mode + def _setup_kernel_modes(self, idx, name, num_q, conf_mode): + """Handles CNDP, AF_PACKET and SIM modes.""" + if conf_mode == "cndp": + try: + # Initialize kernel fastpath. + self.init_datapath(cndp=True) + except: + print( + "Failed to create CNDP/AF_XDP socket for {}. Exiting...".format( + name + ) + ) + sys.exit() + + if conf_mode == "af_packet": + try: + # Initialize kernel datapath + kwargs = { + "vdev": "net_af_packet{},iface={},qpairs={}".format( + idx, name, num_q + ), + "num_out_q": num_q, + "num_inc_q": num_q, + } + self.init_datapath(**kwargs) + except: + print( + "Failed to create AF_PACKET socket for {}. Exiting...".format(name) + ) + sys.exit() + + if conf_mode == "sim": + self.fpi = Source(name="{}_source".format(name)) + self.fpo = Sink(name="{}_out".format(name)) + self.bpf = BPF(name="{}FastBPF".format(name)) + self.bpf.clear() + + # Attach datapath to worker's root TC + self.fpi.attach_task(wid=0) def setup_port( self, diff --git a/conf/test_route_control.py b/conf/test_route_control.py index ca45065b8..ab34f7371 100644 --- a/conf/test_route_control.py +++ b/conf/test_route_control.py @@ -12,7 +12,7 @@ fetch_mac, mac_to_hex, mac_to_int, validate_ipv4) - +DEFAULT_MAC_ADDRESS = "00:1a:2b:3c:4d:5e" class BessControllerMock(object): """Mock of BessController to avoid using BESS from pybess.bess""" @@ -71,7 +71,7 @@ def test_given_invalid_ip_when_validate_ipv6_then_returns_false(self): self.assertFalse(validate_ipv4("")) def test_given_valid_mac_when_mac_to_int_then_returns_int_representation(self): - self.assertEqual(mac_to_int("00:1a:2b:3c:4d:5e"), 112394521950) + self.assertEqual(mac_to_int(DEFAULT_MAC_ADDRESS ), 112394521950) def test_given_invalid_mac_when_mac_to_int_then_raises_exception(self): with self.assertRaises(ValueError): @@ -80,17 +80,17 @@ def test_given_invalid_mac_when_mac_to_int_then_raises_exception(self): def test_given_valid_mac_when_mac_to_hex_then_return_hex_string_representation( self, ): - self.assertEqual(mac_to_hex("00:1a:2b:3c:4d:5e"), "001A2B3C4D5E") + self.assertEqual(mac_to_hex(DEFAULT_MAC_ADDRESS ), "001A2B3C4D5E") def test_given_known_destination_when_fetch_mac_then_returns_mac(self): ndb = Mock() kwargs = { "ifindex": 1, "dst": "192.168.1.1", - "lladdr": "00:1a:2b:3c:4d:5e" + "lladdr": DEFAULT_MAC_ADDRESS } ndb.neighbours.dump.return_value = [kwargs] - self.assertEqual(fetch_mac(ndb, "192.168.1.1"), "00:1a:2b:3c:4d:5e") + self.assertEqual(fetch_mac(ndb, "192.168.1.1"), DEFAULT_MAC_ADDRESS) def test_given_unknown_destination_when_fetch_mac_then_returns_none(self): ndb = Mock() @@ -132,13 +132,13 @@ def add_route_entry( kwargs = { "ifindex": 1, "dst": "192.168.1.1", - "lladdr": "00:1a:2b:3c:4d:5e" + "lladdr": DEFAULT_MAC_ADDRESS } self.ndb.neighbours.dump.return_value = [kwargs] mock_get_update_module_name.return_value = "merge_module" mock_get_route_module_name.return_value = "route_module" mock_get_merge_module_name.return_value = "update_module" - mock_fetch_mac.return_value = "00:1a:2b:3c:4d:5e" + mock_fetch_mac.return_value = DEFAULT_MAC_ADDRESS self.route_controller.add_new_route_entry(route_entry) return route_entry @@ -257,7 +257,7 @@ def test_given_valid_new_route_when_add_new_route_entry_and_mac_known_then_route kwargs = { "ifindex": 1, "dst": "192.168.1.1", - "lladdr": "00:1a:2b:3c:4d:5e" + "lladdr": DEFAULT_MAC_ADDRESS } self.ndb.neighbours.dump.return_value = [kwargs] mock_routes = [{"event": "RTM_NEWROUTE"}, {"event": "OTHER_ACTION"}] @@ -279,7 +279,7 @@ def test_given_valid_new_route_when_add_new_route_entry_and_mac_known_and_neighb kwargs = { "ifindex": 1, "dst": "1.2.3.4", - "lladdr": "00:1a:2b:3c:4d:5e" + "lladdr": DEFAULT_MAC_ADDRESS } self.ndb.neighbours.dump.return_value = [kwargs] mock_routes = [{"event": "RTM_NEWROUTE"}, {"event": "OTHER_ACTION"}] @@ -481,13 +481,13 @@ def test_given_new_neighbor_in_unresolved_when_add_unresolved_new_neighbor_then_ kwargs = { "ifindex": 1, "dst": "192.168.1.1", - "lladdr": "00:1a:2b:3c:4d:5e" + "lladdr": DEFAULT_MAC_ADDRESS } self.ndb.neighbours.dump.return_value = [kwargs] mock_netlink_msg = { "attrs": { "NDA_DST": "1.2.3.4", - "NDA_LLADDR": "00:1a:2b:3c:4d:5e", + "NDA_LLADDR": DEFAULT_MAC_ADDRESS , } } mock_routes = [{"event": "RTM_NEWROUTE"}, {"event": "OTHER_ACTION"}] diff --git a/pfcpiface/config.go b/pfcpiface/config.go index 0bd1dc981..8fd73832b 100644 --- a/pfcpiface/config.go +++ b/pfcpiface/config.go @@ -4,16 +4,15 @@ package pfcpiface import ( - "github.com/omec-project/upf-epc/internal/p4constants" - "go.uber.org/zap" - "go.uber.org/zap/zapcore" - - "net" - "time" - "encoding/json" + "net" "os" "regexp" + "time" + + "github.com/omec-project/upf-epc/internal/p4constants" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" ) const ( @@ -148,6 +147,18 @@ func validateConf(conf Conf) error { } } + if err := validateUEIPPoolAndPeers(conf); err != nil { + return err + } + + if err := validateTimeouts(conf); err != nil { + return err + } + + return nil +} + +func validateUEIPPoolAndPeers(conf Conf) error { if conf.CPIface.EnableUeIPAlloc { for _, dnn := range conf.CPIface.DnnList { _, _, err := net.ParseCIDR(dnn.UEIPPool) @@ -163,7 +174,10 @@ func validateConf(conf Conf) error { return ErrInvalidArgumentWithReason("conf.CPIface.Peers", peer, "invalid IP") } } + return nil +} +func validateTimeouts(conf Conf) error { if _, err := time.ParseDuration(conf.RespTimeout); err != nil { return ErrInvalidArgumentWithReason("conf.RespTimeout", conf.RespTimeout, "invalid duration") } diff --git a/pfcpiface/conn.go b/pfcpiface/conn.go index 4153fe563..f80b104d9 100644 --- a/pfcpiface/conn.go +++ b/pfcpiface/conn.go @@ -122,7 +122,7 @@ func (node *PFCPNode) NewPFCPConn(lAddr, rAddr string, buf []byte) *PFCPConn { rng := rand.New(rand.NewSource(time.Now().UnixNano())) // #nosec G404 - var p = &PFCPConn{ + p := &PFCPConn{ ctx: node.ctx, Conn: conn, ts: ts, @@ -213,7 +213,10 @@ func (pConn *PFCPConn) Serve() { }(connTimeout) // TODO: Sender goroutine + pConn.waitForShutdown(connTimeout) +} +func (pConn *PFCPConn) waitForShutdown(connTimeout chan struct{}) { for { select { case <-connTimeout: diff --git a/pfcpiface/ip_pool_test.go b/pfcpiface/ip_pool_test.go index 35bf56887..4a3347d3c 100644 --- a/pfcpiface/ip_pool_test.go +++ b/pfcpiface/ip_pool_test.go @@ -4,22 +4,24 @@ package pfcpiface import ( - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - "math" "net" "sync" "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) +const ipSubnetCIDR = "10.0.0.0/24" + func TestNewIPPool(t *testing.T) { tests := []struct { name string poolSubnet string wantErr bool }{ - {name: "normal pool", poolSubnet: "10.0.0.0/24", wantErr: false}, + {name: "normal pool", poolSubnet: ipSubnetCIDR, wantErr: false}, {name: "smallest allowed pool", poolSubnet: "10.0.0.0/30", wantErr: false}, {name: "IPv6 pool", poolSubnet: "2001::/124", wantErr: false}, {name: "too small pool", poolSubnet: "10.0.0.0/32", wantErr: true}, @@ -53,7 +55,7 @@ func TestIPPool_LookupOrAllocIP(t *testing.T) { }) t.Run("repeated SEID lookups return same IP", func(t *testing.T) { - const poolSubnet = "10.0.0.0/24" + const poolSubnet = ipSubnetCIDR const seid = 1234 pool, err := NewIPPool(poolSubnet) require.NoError(t, err) @@ -66,7 +68,7 @@ func TestIPPool_LookupOrAllocIP(t *testing.T) { }) t.Run("full subnet allocation", func(t *testing.T) { - const poolSubnet = "10.0.0.0/24" + const poolSubnet = ipSubnetCIDR const usableAddresses = 256 - 2 // Account for network and broadcast addresses const baseSeid = 1000 _, ipnet, err := net.ParseCIDR(poolSubnet) @@ -118,7 +120,7 @@ func TestIPPool_LookupOrAllocIP(t *testing.T) { func TestIPPool_DeallocIP(t *testing.T) { t.Run("plain alloc into dealloc", func(t *testing.T) { - const poolSubnet = "10.0.0.0/24" + const poolSubnet = ipSubnetCIDR const seid = 1234 pool, err := NewIPPool(poolSubnet) require.NoError(t, err) @@ -129,7 +131,7 @@ func TestIPPool_DeallocIP(t *testing.T) { }) t.Run("dealloc non-existent SEIDs fails", func(t *testing.T) { - pool, err := NewIPPool("10.0.0.0/24") + pool, err := NewIPPool(ipSubnetCIDR) require.NoError(t, err) err = pool.DeallocIP(1234) assert.Error(t, err) diff --git a/pfcpiface/messages_conn.go b/pfcpiface/messages_conn.go index e2226947f..1194e1149 100644 --- a/pfcpiface/messages_conn.go +++ b/pfcpiface/messages_conn.go @@ -12,9 +12,13 @@ import ( "github.com/wmnsk/go-pfcp/message" ) -var errFlowDescAbsent = errors.New("flow description not present") -var errDatapathDown = errors.New("datapath down") -var errReqRejected = errors.New("request rejected") +const msgAssociationSetupResponseFrom = "association Setup Response from" + +var ( + errFlowDescAbsent = errors.New("flow description not present") + errDatapathDown = errors.New("datapath down") + errReqRejected = errors.New("request rejected") +) func (pConn *PFCPConn) sendAssociationRequest() { // Build request message @@ -189,7 +193,7 @@ func (pConn *PFCPConn) handleAssociationSetupResponse(msg message.Message) error } if cause != ie.CauseRequestAccepted { - logger.PfcpLog.Errorln("association Setup Response from", addr, + logger.PfcpLog.Errorln(msgAssociationSetupResponseFrom, addr, "with Cause:", cause) return errReqRejected } @@ -206,12 +210,12 @@ func (pConn *PFCPConn) handleAssociationSetupResponse(msg message.Message) error if pConn.ts.remote.IsZero() { pConn.ts.remote = ts - logger.PfcpLog.Infoln("association Setup Response from", addr, + logger.PfcpLog.Infoln(msgAssociationSetupResponseFrom, addr, "with recovery timestamp:", ts) } else if ts.After(pConn.ts.remote) { old := pConn.ts.remote pConn.ts.remote = ts - logger.PfcpLog.Warnln("association Setup Response from", addr, + logger.PfcpLog.Warnln(msgAssociationSetupResponseFrom, addr, "with newer recovery timestamp:", ts, "older:", old) } diff --git a/pfcpiface/node.go b/pfcpiface/node.go index 11a2b6ade..b8feb2bd5 100644 --- a/pfcpiface/node.go +++ b/pfcpiface/node.go @@ -15,6 +15,8 @@ import ( "github.com/omec-project/upf-epc/pfcpiface/metrics" ) +const msgRemovedConnectionTo = "removed connection to" + // PFCPNode represents a PFCP endpoint of the UPF. type PFCPNode struct { ctx context.Context @@ -126,7 +128,7 @@ func (node *PFCPNode) Serve() { }) case rAddr := <-node.pConnDone: node.pConns.Delete(rAddr) - logger.PfcpLog.Infoln("removed connection to", rAddr) + logger.PfcpLog.Infoln(msgRemovedConnectionTo, rAddr) case <-node.ctx.Done(): shutdown = true @@ -148,7 +150,7 @@ func (node *PFCPNode) Serve() { break clearLoop } node.pConns.Delete(rAddr) - logger.PfcpLog.Infoln("removed connection to", rAddr) + logger.PfcpLog.Infoln(msgRemovedConnectionTo, rAddr) } default: // nothing to read from channel @@ -159,7 +161,7 @@ func (node *PFCPNode) Serve() { if len(node.pConnDone) > 0 { for rAddr := range node.pConnDone { node.pConns.Delete(rAddr) - logger.PfcpLog.Infoln("removed connection to", rAddr) + logger.PfcpLog.Infoln(msgRemovedConnectionTo, rAddr) } } diff --git a/pfcpiface/p4rt_translator.go b/pfcpiface/p4rt_translator.go index 0459c91b2..ec956f788 100644 --- a/pfcpiface/p4rt_translator.go +++ b/pfcpiface/p4rt_translator.go @@ -47,6 +47,12 @@ const ( DefaultApplicationID = 0 ) +const ( + fieldEntryTableID = "entry.TableId" // only if used as string + msgFindMatchField = "find match field" + msgFieldName = "field name" +) + type tunnelParams struct { tunnelIP4Src uint32 tunnelIP4Dst uint32 @@ -185,7 +191,7 @@ func (t *P4rtTranslator) getActionParamByName(action *p4ConfigV1.Action, paramNa // TODO: find a way to use *p4.TableEntry as receiver func (t *P4rtTranslator) withExactMatchField(entry *p4.TableEntry, name string, value interface{}) error { if entry.TableId == 0 { - return ErrInvalidArgumentWithReason("entry.TableId", entry.TableId, "no table name for entry defined, set table name before adding match fields") + return ErrInvalidArgumentWithReason(fieldEntryTableID, entry.TableId, "no table name for entry defined, set table name before adding match fields") } p4Table, err := t.getTableByID(entry.TableId) @@ -195,7 +201,7 @@ func (t *P4rtTranslator) withExactMatchField(entry *p4.TableEntry, name string, p4MatchField := t.getMatchFieldByName(p4Table, name) if p4MatchField == nil { - return ErrOperationFailedWithParam("find match field", "name", name) + return ErrOperationFailedWithParam(msgFindMatchField, "name", name) } matchField := &p4.FieldMatch{ @@ -218,11 +224,11 @@ func (t *P4rtTranslator) withExactMatchField(entry *p4.TableEntry, name string, } func (t *P4rtTranslator) withLPMField(entry *p4.TableEntry, name string, value uint32, prefixLen uint8) error { - lpmFieldLog := logger.P4Log.With("entry", entry.String(), "field name", name) + lpmFieldLog := logger.P4Log.With("entry", entry.String(), msgFieldName, name) lpmFieldLog.Debugln("adding LPM match field to the entry") if entry.TableId == 0 { - return ErrInvalidArgumentWithReason("entry.TableId", entry.TableId, "no table ID for entry defined, set table ID before adding match fields") + return ErrInvalidArgumentWithReason(fieldEntryTableID, entry.TableId, "no table ID for entry defined, set table ID before adding match fields") } p4Table, err := t.getTableByID(entry.TableId) @@ -232,7 +238,7 @@ func (t *P4rtTranslator) withLPMField(entry *p4.TableEntry, name string, value u p4MatchField := t.getMatchFieldByName(p4Table, name) if p4MatchField == nil { - return ErrOperationFailedWithParam("find match field", "name", name) + return ErrOperationFailedWithParam(msgFindMatchField, "name", name) } byteVal, err := convertValueToBinary(value) @@ -257,11 +263,11 @@ func (t *P4rtTranslator) withLPMField(entry *p4.TableEntry, name string, value u } func (t *P4rtTranslator) withRangeMatchField(entry *p4.TableEntry, name string, low interface{}, high interface{}) error { - rangeFieldLog := logger.P4Log.With("entry", entry.String(), "field name", name) + rangeFieldLog := logger.P4Log.With("entry", entry.String(), msgFieldName, name) rangeFieldLog.Debugln("adding range match field to the entry") if entry.TableId == 0 { - return ErrInvalidArgumentWithReason("entry.TableId", entry.TableId, "no table ID for entry defined, set table ID before adding match fields") + return ErrInvalidArgumentWithReason(fieldEntryTableID, entry.TableId, "no table ID for entry defined, set table ID before adding match fields") } lowByteVal, err := convertValueToBinary(low) @@ -281,7 +287,7 @@ func (t *P4rtTranslator) withRangeMatchField(entry *p4.TableEntry, name string, p4MatchField := t.getMatchFieldByName(p4Table, name) if p4MatchField == nil { - return ErrOperationFailedWithParam("find match field", "name", name) + return ErrOperationFailedWithParam(msgFindMatchField, "name", name) } matchField := &p4.FieldMatch{ @@ -301,11 +307,11 @@ func (t *P4rtTranslator) withRangeMatchField(entry *p4.TableEntry, name string, } func (t *P4rtTranslator) withTernaryMatchField(entry *p4.TableEntry, name string, value interface{}, mask interface{}) error { - ternaryFieldLog := logger.P4Log.With("entry", entry.String(), "field name", name) + ternaryFieldLog := logger.P4Log.With("entry", entry.String(), msgFieldName, name) ternaryFieldLog.Debugln("adding ternary match field to the entry") if entry.TableId == 0 { - return ErrInvalidArgumentWithReason("entry.TableId", entry.TableId, "no table name for entry defined, set table name before adding match fields") + return ErrInvalidArgumentWithReason(fieldEntryTableID, entry.TableId, "no table name for entry defined, set table name before adding match fields") } byteVal, err := convertValueToBinary(value) @@ -330,7 +336,7 @@ func (t *P4rtTranslator) withTernaryMatchField(entry *p4.TableEntry, name string p4MatchField := t.getMatchFieldByName(p4Table, name) if p4MatchField == nil { - return ErrOperationFailedWithParam("find match field", "name", name) + return ErrOperationFailedWithParam(msgFindMatchField, "name", name) } matchField := &p4.FieldMatch{ @@ -682,7 +688,8 @@ func (t *P4rtTranslator) buildUplinkTerminationsEntry(pdr pdr, appMeterIdx uint3 } func (t *P4rtTranslator) buildDownlinkTerminationsEntry(pdr pdr, appMeterIdx uint32, relatedFAR far, - internalAppID uint8, qfi uint8, tc uint8, relatedQER qer) (*p4.TableEntry, error) { + internalAppID uint8, qfi uint8, tc uint8, relatedQER qer, +) (*p4.TableEntry, error) { builderLog := logger.P4Log.With("pdr", pdr, "appMeterIndex", appMeterIdx, "tc", tc, "related-far", relatedFAR) builderLog.Debugln("building P4rt table entry for UP4 terminations_downlink table") diff --git a/pfcpiface/parse_far.go b/pfcpiface/parse_far.go index 32ca99ed4..fff1b920c 100644 --- a/pfcpiface/parse_far.go +++ b/pfcpiface/parse_far.go @@ -67,91 +67,13 @@ func (f *far) Forwards() bool { func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error { logger.PfcpLog.Debugf("[parseFAR][Enter] F-SEID=%d Operation=%v", fseid, op) + f.fseID = (fseid) - f.fseID = fseid - - farID, err := farIE.FARID() - if err != nil { - logger.PfcpLog.Errorf("[parseFAR] Failed to read FAR ID: %v", err) - return err - } - - f.farID = farID - logger.PfcpLog.Debugf("[parseFAR] FAR ID=%d", farID) - - action, err := farIE.ApplyAction() - if err != nil { - logger.PfcpLog.Errorf("[parseFAR] Failed to read ApplyAction: %v", err) - return err - } - - logger.PfcpLog.Debugf("[parseFAR] ApplyAction Raw=%v", action) - - if action[0] == 0 { - logger.PfcpLog.Errorf("[parseFAR] Invalid FAR Action=%v", action) - return ErrInvalidArgument("FAR Action", action) - } - - f.applyAction = action[0] - - logger.PfcpLog.Debugf( - "[parseFAR] FARID=%d ApplyAction=%02x", - f.farID, - f.applyAction, - ) - - var fwdIEs []*ie.IE - - switch op { - case create: - logger.PfcpLog.Debugf( - "[parseFAR] Processing CREATE FAR FARID=%d", - f.farID, - ) - - if (f.applyAction & ActionForward) != 0 { - logger.PfcpLog.Debugf( - "[parseFAR] Fetching ForwardingParameters FARID=%d", - f.farID, - ) - - fwdIEs, err = farIE.ForwardingParameters() - } - - case update: - logger.PfcpLog.Debugf( - "[parseFAR] Processing UPDATE FAR FARID=%d", - f.farID, - ) - - fwdIEs, err = farIE.UpdateForwardingParameters() - - default: - logger.PfcpLog.Errorf( - "[parseFAR] Invalid operation=%v FARID=%d", - op, - f.farID, - ) - - return ErrInvalidOperation(op) - } - + fwdIEs, err := f.parseFARBasicFields(farIE, op) if err != nil { - logger.PfcpLog.Errorf( - "[parseFAR] Failed to parse forwarding parameters FARID=%d Error=%v", - f.farID, - err, - ) - return err } - logger.PfcpLog.Debugf( - "[parseFAR] FARID=%d Forwarding IE Count=%d", - f.farID, - len(fwdIEs), - ) - f.sendEndMarker = false var fields Bits @@ -167,12 +89,10 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error switch fwdIE.Type { case ie.OuterHeaderCreation: fields = Set(fields, FwdIEOuterHeaderCreation) - logger.PfcpLog.Debugf( "[parseFAR] Parsing OuterHeaderCreation FARID=%d", f.farID, ) - ohcFields, err = fwdIE.OuterHeaderCreation() if err != nil { logger.PfcpLog.Errorf( @@ -180,7 +100,6 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error f.farID, err, ) - continue } @@ -188,7 +107,6 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error f.tunnelIP4Dst = ip2int(ohcFields.IPv4Address) f.tunnelType = uint8(1) f.tunnelPort = tunnelGTPUPort - logger.PfcpLog.Debugf( "[parseFAR] FARID=%d TEID=%d DstIP=%v TunnelPort=%d", f.farID, @@ -198,12 +116,10 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error ) case ie.DestinationInterface: fields = Set(fields, FwdIEDestinationIntf) - logger.PfcpLog.Debugf( "[parseFAR] Parsing DestinationInterface FARID=%d", f.farID, ) - f.dstIntf, err = fwdIE.DestinationInterface() if err != nil { logger.PfcpLog.Errorf( @@ -211,29 +127,24 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error f.farID, err, ) - continue } switch f.dstIntf { case ie.DstInterfaceAccess: f.tunnelIP4Src = ip2int(upf.accessIP) - logger.PfcpLog.Debugf( "[parseFAR] FARID=%d DestinationInterface=ACCESS SrcIP=%v", f.farID, upf.accessIP, ) - case ie.DstInterfaceCore: f.tunnelIP4Src = ip2int(upf.coreIP) - logger.PfcpLog.Debugf( "[parseFAR] FARID=%d DestinationInterface=CORE SrcIP=%v", f.farID, upf.coreIP, ) - default: logger.PfcpLog.Warnf( "[parseFAR] FARID=%d Unknown DestinationInterface=%d", @@ -243,12 +154,10 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error } case ie.PFCPSMReqFlags: fields = Set(fields, FwdIEPfcpSMReqFlags) - logger.PfcpLog.Debugf( "[parseFAR] Parsing PFCPSMReqFlags FARID=%d", f.farID, ) - smReqFlags, err := fwdIE.PFCPSMReqFlags() if err != nil { logger.PfcpLog.Errorf( @@ -256,10 +165,8 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error f.farID, err, ) - continue } - logger.PfcpLog.Debugf( "[parseFAR] FARID=%d PFCPSMReqFlags=%08b", f.farID, @@ -268,7 +175,6 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error if has2ndBit(smReqFlags) { f.sendEndMarker = true - logger.PfcpLog.Debugf( "[parseFAR] FARID=%d EndMarker enabled", f.farID, @@ -282,7 +188,6 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error ) } } - logger.PfcpLog.Debugf( "[parseFAR][Exit] F-SEID=%d FARID=%d Action=%02x SrcIP=%d DstIP=%d TEID=%d EndMarker=%v", f.fseID, @@ -293,6 +198,77 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error f.tunnelTEID, f.sendEndMarker, ) - return nil } + +func (f *far) parseFARBasicFields(farIE *ie.IE, op operation) ([]*ie.IE, error) { + farID, err := farIE.FARID() + if err != nil { + logger.PfcpLog.Errorf("[parseFAR] Failed to read FAR ID: %v", err) + return nil, err + } + f.farID = farID + logger.PfcpLog.Debugf("[parseFAR] FAR ID=%d", farID) + + action, err := farIE.ApplyAction() + if err != nil { + logger.PfcpLog.Errorf("[parseFAR] Failed to read ApplyAction: %v", err) + return nil, err + } + logger.PfcpLog.Debugf("[parseFAR] ApplyAction Raw=%v", action) + if action[0] == 0 { + logger.PfcpLog.Errorf("[parseFAR] Invalid FAR Action=%v", action) + return nil, ErrInvalidArgument("FAR Action", action) + } + f.applyAction = action[0] + logger.PfcpLog.Debugf( + "[parseFAR] FARID=%d ApplyAction=%02x", + f.farID, + f.applyAction, + ) + + var fwdIEs []*ie.IE + + switch op { + case create: + logger.PfcpLog.Debugf( + "[parseFAR] Processing CREATE FAR FARID=%d", + f.farID, + ) + if (f.applyAction & ActionForward) != 0 { + logger.PfcpLog.Debugf( + "[parseFAR] Fetching ForwardingParameters FARID=%d", + f.farID, + ) + fwdIEs, err = farIE.ForwardingParameters() + } + case update: + logger.PfcpLog.Debugf( + "[parseFAR] Processing UPDATE FAR FARID=%d", + f.farID, + ) + fwdIEs, err = farIE.UpdateForwardingParameters() + default: + logger.PfcpLog.Errorf( + "[parseFAR] Invalid operation=%v FARID=%d", + op, + f.farID, + ) + return nil, ErrInvalidOperation(op) + } + + if err != nil { + logger.PfcpLog.Errorf( + "[parseFAR] Failed to parse forwarding parameters FARID=%d Error=%v", + f.farID, + err, + ) + return nil, err + } + logger.PfcpLog.Debugf( + "[parseFAR] FARID=%d Forwarding IE Count=%d", + f.farID, + len(fwdIEs), + ) + return fwdIEs, nil +} diff --git a/pfcpiface/parse_far_test.go b/pfcpiface/parse_far_test.go index 8e73a056b..34c1d8b49 100644 --- a/pfcpiface/parse_far_test.go +++ b/pfcpiface/parse_far_test.go @@ -23,6 +23,7 @@ type farTestCase struct { const ( defaultGTPProtocolPort = 2152 ) +const ipAddressIn = "10.0.0.1" func TestParseFAR(t *testing.T) { createOp, updateOp := create, update @@ -115,7 +116,7 @@ func TestParseFARShouldError(t *testing.T) { ie.NewApplyAction(0), ie.NewUpdateForwardingParameters( ie.NewDestinationInterface(ie.DstInterfaceAccess), - ie.NewOuterHeaderCreation(0x100, 100, "10.0.0.1", "", 0, 0, 0), + ie.NewOuterHeaderCreation(0x100, 100, ipAddressIn, "", 0, 0, 0), ), ), expected: &far{ @@ -130,7 +131,7 @@ func TestParseFARShouldError(t *testing.T) { ie.NewApplyAction(ActionDrop), ie.NewUpdateForwardingParameters( ie.NewDestinationInterface(ie.DstInterfaceAccess), - ie.NewOuterHeaderCreation(0x100, 100, "10.0.0.1", "", 0, 0, 0), + ie.NewOuterHeaderCreation(0x100, 100, ipAddressIn, "", 0, 0, 0), ), ), expected: &far{ @@ -143,7 +144,7 @@ func TestParseFARShouldError(t *testing.T) { mockFar := &far{} mockUpf := &upf{ accessIP: net.ParseIP("192.168.0.1"), - coreIP: net.ParseIP("10.0.0.1"), + coreIP: net.ParseIP(ipAddressIn), } err := mockFar.parseFAR(scenario.input, 101, mockUpf, scenario.op) diff --git a/pfcpiface/parse_pdr.go b/pfcpiface/parse_pdr.go index d6cf2326a..f244ac803 100644 --- a/pfcpiface/parse_pdr.go +++ b/pfcpiface/parse_pdr.go @@ -129,42 +129,7 @@ func (pr portRange) asComplexTernaryMatches(strategy RangeConversionStrategy) ([ rules = append(rules, portRangeTernaryRule{uint16(port), math.MaxUint16}) } } else if strategy == Ternary { - // Adapted from https://stackoverflow.com/a/66959276 - const limit = math.MaxUint16 - maxPort := func(port, mask uint16) uint16 { - xid := limit - mask - nid := port & mask - return nid + xid - } - - portMask := func(port, end uint16) uint16 { - bit := uint16(1) - mask := uint16(limit) - testMask := uint16(limit) - netPort := port & limit - maximumPort := maxPort(netPort, limit) - - for netPort > 0 && maximumPort < end { - netPort = port & testMask - if netPort < port { - break - } - maximumPort = maxPort(netPort, testMask) - if maximumPort <= end { - mask = testMask - } - testMask -= bit - bit <<= 1 - } - return mask - } - - port := uint32(pr.low) // Promote to higher bit width for greater-equals check. - for port <= uint32(pr.high) { - mask := portMask(uint16(port), pr.high) - rules = append(rules, portRangeTernaryRule{uint16(port), mask}) - port = uint32(maxPort(uint16(port), mask)) + 1 - } + return pr.buildTernaryRules(), nil } else { return nil, ErrInvalidArgument("asComplexTernaryMatches", strategy) } @@ -172,6 +137,46 @@ func (pr portRange) asComplexTernaryMatches(strategy RangeConversionStrategy) ([ return rules, nil } +func (pr portRange) buildTernaryRules() []portRangeTernaryRule { + // Adapted from https://stackoverflow.com/a/66959276 + const limit = math.MaxUint16 + maxPort := func(port, mask uint16) uint16 { + xid := limit - mask + nid := port & mask + return nid + xid + } + + portMask := func(port, end uint16) uint16 { + bit := uint16(1) + mask := uint16(limit) + testMask := uint16(limit) + netPort := port & limit + maximumPort := maxPort(netPort, limit) + + for netPort > 0 && maximumPort < end { + netPort = port & testMask + if netPort < port { + break + } + maximumPort = maxPort(netPort, testMask) + if maximumPort <= end { + mask = testMask + } + testMask -= bit + bit <<= 1 + } + return mask + } + rules := make([]portRangeTernaryRule, 0) + port := uint32(pr.low) // Promote to higher bit width for greater-equals check. + for port <= uint32(pr.high) { + mask := portMask(uint16(port), pr.high) + rules = append(rules, portRangeTernaryRule{uint16(port), mask}) + port = uint32(maxPort(uint16(port), mask)) + 1 + } + return rules +} + type portRangeTernaryRule struct { port, mask uint16 } @@ -208,14 +213,7 @@ func CreatePortRangeCartesianProduct(src, dst portRange) ([]portRangeTernaryCart if err != nil { return nil, err } - - for _, r := range srcTernaryRules { - p := portRangeTernaryCartesianProduct{ - srcPort: r.port, srcMask: r.mask, - dstPort: dstTernary.port, dstMask: dstTernary.mask, - } - rules = append(rules, p) - } + rules = buildRulesFromRange(srcTernaryRules, dstTernary, true) } else if dst.isRangeMatch() { dstTernaryRules, err := dst.asComplexTernaryMatches(Exact) if err != nil { @@ -227,13 +225,7 @@ func CreatePortRangeCartesianProduct(src, dst portRange) ([]portRangeTernaryCart return nil, err } - for _, r := range dstTernaryRules { - p := portRangeTernaryCartesianProduct{ - srcPort: srcTernary.port, srcMask: srcTernary.mask, - dstPort: r.port, dstMask: r.mask, - } - rules = append(rules, p) - } + rules = buildRulesFromRange(dstTernaryRules, srcTernary, false) } else { // Neither is range. Only one rule needed. srcTernary, err := src.asTrivialTernaryMatch() @@ -252,10 +244,33 @@ func CreatePortRangeCartesianProduct(src, dst portRange) ([]portRangeTernaryCart } rules = append(rules, p) } - return rules, nil } +func buildRulesFromRange( + rangeRules []portRangeTernaryRule, + fixedRule portRangeTernaryRule, + isSrcRange bool, +) []portRangeTernaryCartesianProduct { + rules := make([]portRangeTernaryCartesianProduct, 0, len(rangeRules)) + + for _, r := range rangeRules { + if isSrcRange { + rules = append(rules, portRangeTernaryCartesianProduct{ + srcPort: r.port, srcMask: r.mask, + dstPort: fixedRule.port, dstMask: fixedRule.mask, + }) + } else { + rules = append(rules, portRangeTernaryCartesianProduct{ + srcPort: fixedRule.port, srcMask: fixedRule.mask, + dstPort: r.port, dstMask: r.mask, + }) + } + } + + return rules +} + type applicationFilter struct { srcIP uint32 dstIP uint32 @@ -510,24 +525,8 @@ func (p *pdr) parseSDFFilter(ie *ie.IE) error { } func (p *pdr) parsePDI(pdiIEs []*ie.IE, appPFDs map[string]appPFD, ippool *IPPool) error { - for _, pdiIE := range pdiIEs { - switch pdiIE.Type { - case ie.UEIPAddress: - if err := p.parseUEAddressIE(pdiIE, ippool); err != nil { - logger.PfcpLog.Errorf("failed to parse UE Address IE: %v", err) - return err - } - case ie.SourceInterface: - if err := p.parseSourceInterfaceIE(pdiIE); err != nil { - logger.PfcpLog.Errorf("failed to parse Source Interface IE: %v", err) - return err - } - case ie.FTEID: - if err := p.parseFTEID(pdiIE); err != nil { - logger.PfcpLog.Errorf("failed to parse F-TEID IE: %v", err) - return err - } - } + if err := p.processBasicPDI(pdiIEs, ippool); err != nil { + return err } // initialize application filter with UE address; @@ -559,60 +558,41 @@ func (p *pdr) parsePDI(pdiIEs []*ie.IE, appPFDs map[string]appPFD, ippool *IPPoo return nil } +func (p *pdr) processBasicPDI(pdiIEs []*ie.IE, ippool *IPPool) error { + for _, pdiIE := range pdiIEs { + switch pdiIE.Type { + case ie.UEIPAddress: + if err := p.parseUEAddressIE(pdiIE, ippool); err != nil { + logger.PfcpLog.Errorf("failed to parse UE Address IE: %v", err) + return err + } + case ie.SourceInterface: + if err := p.parseSourceInterfaceIE(pdiIE); err != nil { + logger.PfcpLog.Errorf("failed to parse Source Interface IE: %v", err) + return err + } + case ie.FTEID: + if err := p.parseFTEID(pdiIE); err != nil { + logger.PfcpLog.Errorf("failed to parse F-TEID IE: %v", err) + return err + } + } + } + return nil +} + func (p *pdr) parsePDR(ie1 *ie.IE, seid uint64, appPFDs map[string]appPFD, ippool *IPPool) error { logger.PfcpLog.Infof("[parsePDR][Enter] SEID=%d IEType=%d", seid, ie1.Type) - /* reset outerHeaderRemoval to begin with */ outerHeaderRemoval := uint8(0) p.qerIDList = make([]uint32, 0) p.fseID = seid - pdrID, err := ie1.PDRID() + pdrID, precedence, farID, outerHeaderRemoval, err := p.parseCorePDRFields(ie1, appPFDs, ippool) if err != nil { logger.PfcpLog.Errorln("could not read PDR ID!") return err } - logger.PfcpLog.Infof("[parsePDR] Parsed PDR ID=%d", pdrID) - - precedence, err := ie1.Precedence() - if err != nil { - logger.PfcpLog.Errorln("could not read Precedence!") - return err - } - logger.PfcpLog.Infof("[parsePDR] Parsed Precedence=%d", precedence) - - pdi, err := ie1.PDI() - if err != nil { - logger.PfcpLog.Errorln("could not read PDI!") - return err - } - logger.PfcpLog.Debugln("[parsePDR] Successfully parsed PDI") - - res, err := ie1.OuterHeaderRemovalDescription() - if res == 0 && err == nil { // 0 == GTP-U/UDP/IPv4 - outerHeaderRemoval = 1 - logger.PfcpLog.Infof("[parsePDR] OuterHeaderRemoval enabled") - } else { - logger.PfcpLog.Debugf("[parsePDR] OuterHeaderRemoval not present or unsupported, res=%d err=%v", res, err) - } - - err = p.parsePDI(pdi, appPFDs, ippool) - if err != nil { - if errors.Is(err, errBadFilterDesc) { - logger.PfcpLog.Warnf("[parsePDR] parsePDI returned bad filter description: %v", err) - } else { - logger.PfcpLog.Errorf("[parsePDR] parsePDI failed: %v", err) - return err - } - } - logger.PfcpLog.Debugln("[parsePDR] PDI parsed successfully") - - farID, err := ie1.FARID() - if err != nil { - logger.PfcpLog.Errorln("could not read FAR ID!") - return err - } - logger.PfcpLog.Infof("[parsePDR] Parsed FAR ID=%d", farID) /* Multiple instances of QERID can be present in CreatePDR/UpdatePDR go-pfcp currently support API to return list of QERIDs. So, we @@ -629,7 +609,6 @@ func (p *pdr) parsePDR(ie1 *ie.IE, seid uint64, appPFDs map[string]appPFD, ippoo logger.PfcpLog.Errorf("[parsePDR] CreatePDR decode failed: %v", errin) return errin } - case ie.UpdatePDR: logger.PfcpLog.Debugln("[parsePDR] Processing UpdatePDR IE") ies, errin = ie1.UpdatePDR() @@ -637,9 +616,8 @@ func (p *pdr) parsePDR(ie1 *ie.IE, seid uint64, appPFDs map[string]appPFD, ippoo logger.PfcpLog.Errorf("[parsePDR] UpdatePDR decode failed: %v", errin) return errin } - default: - logger.PfcpLog.Warnf("[parsePDR] Unsupported IE type=%d", ie1.Type) + logger.PfcpLog.Errorf("[parsePDR] Unsupported IE type: %v", ie1.Type) } for _, x := range ies { @@ -649,7 +627,6 @@ func (p *pdr) parsePDR(ie1 *ie.IE, seid uint64, appPFDs map[string]appPFD, ippoo logger.PfcpLog.Errorln("[parsePDR] qerID read failed") continue } - logger.PfcpLog.Infof("[parsePDR] Parsed QER ID=%d", qerID) p.qerIDList = append(p.qerIDList, qerID) } @@ -672,3 +649,56 @@ func (p *pdr) parsePDR(ie1 *ie.IE, seid uint64, appPFDs map[string]appPFD, ippoo return nil } + +func (p *pdr) parseCorePDRFields( + ie1 *ie.IE, + appPFDs map[string]appPFD, + ippool *IPPool, +) (uint16, uint32, uint32, uint8, error) { + var outerHeaderRemoval uint8 + + pdrID, err := ie1.PDRID() + if err != nil { + logger.PfcpLog.Errorln("could not read PDR ID!") + return 0, 0, 0, 0, err + } + logger.PfcpLog.Infof("[parsePDR] Parsed PDR ID=%d", pdrID) + precedence, err := ie1.Precedence() + if err != nil { + logger.PfcpLog.Errorln("could not read Precedence!") + return 0, 0, 0, 0, err + } + logger.PfcpLog.Infof("[parsePDR] Parsed Precedence=%d", precedence) + pdi, err := ie1.PDI() + if err != nil { + logger.PfcpLog.Errorln("could not read PDI!") + return 0, 0, 0, 0, err + } + logger.PfcpLog.Debugln("[parsePDR] Successfully parsed PDI") + res, err := ie1.OuterHeaderRemovalDescription() + if res == 0 && err == nil { // 0 == GTP-U/UDP/IPv4 + outerHeaderRemoval = 1 + logger.PfcpLog.Infof("[parsePDR] OuterHeaderRemoval enabled") + } else { + logger.PfcpLog.Debugf("[parsePDR] OuterHeaderRemoval not present or unsupported, res=%d err=%v", res, err) + } + + err = p.parsePDI(pdi, appPFDs, ippool) + if err != nil { + if errors.Is(err, errBadFilterDesc) { + logger.PfcpLog.Warnf("[parsePDR] parsePDI returned bad filter description: %v", err) + } else { + logger.PfcpLog.Errorf("[parsePDR] parsePDI failed: %v", err) + return 0, 0, 0, 0, err + } + } + logger.PfcpLog.Debugln("[parsePDR] PDI parsed successfully") + + farID, err := ie1.FARID() + if err != nil { + logger.PfcpLog.Errorln("could not read FAR ID!") + return 0, 0, 0, 0, err + } + logger.PfcpLog.Infof("[parsePDR] Parsed FAR ID=%d", farID) + return pdrID, precedence, farID, outerHeaderRemoval, nil +} diff --git a/pfcpiface/parse_pdr_test.go b/pfcpiface/parse_pdr_test.go index 5ba28451c..8fcd1614b 100644 --- a/pfcpiface/parse_pdr_test.go +++ b/pfcpiface/parse_pdr_test.go @@ -14,6 +14,11 @@ import ( "github.com/wmnsk/go-pfcp/ie" ) +const ( + ipAddrPrimary = "17.0.0.1" + ipAddrSecondary = "192.168.1.1" +) + type pdrTestCase struct { input *ie.IE expected *pdr @@ -180,7 +185,8 @@ func TestCreatePortRangeCartesianProduct(t *testing.T) { want []portRangeTernaryCartesianProduct wantErr bool }{ - {name: "exact ranges", + { + name: "exact ranges", args: args{src: newExactMatchPortRange(5000), dst: newExactMatchPortRange(80)}, want: []portRangeTernaryCartesianProduct{{ srcPort: 5000, @@ -188,8 +194,10 @@ func TestCreatePortRangeCartesianProduct(t *testing.T) { dstPort: 80, dstMask: math.MaxUint16, }}, - wantErr: false}, - {name: "wildcard dst range", + wantErr: false, + }, + { + name: "wildcard dst range", args: args{src: newExactMatchPortRange(10), dst: newWildcardPortRange()}, want: []portRangeTernaryCartesianProduct{{ srcPort: 10, @@ -197,8 +205,10 @@ func TestCreatePortRangeCartesianProduct(t *testing.T) { dstPort: 0, dstMask: 0, }}, - wantErr: false}, - {name: "true range src range", + wantErr: false, + }, + { + name: "true range src range", args: args{src: newRangeMatchPortRange(1, 3), dst: newExactMatchPortRange(80)}, want: []portRangeTernaryCartesianProduct{ { @@ -218,12 +228,16 @@ func TestCreatePortRangeCartesianProduct(t *testing.T) { srcMask: 0xffff, dstPort: 80, dstMask: math.MaxUint16, - }}, - wantErr: false}, - {name: "invalid double range", + }, + }, + wantErr: false, + }, + { + name: "invalid double range", args: args{src: newRangeMatchPortRange(10, 20), dst: newRangeMatchPortRange(80, 85)}, want: nil, - wantErr: true}, + wantErr: true, + }, } for _, tt := range tests { @@ -369,7 +383,8 @@ func Test_portRange_asComplexTernaryMatches(t *testing.T) { wantErr bool want []portRangeTernaryRule }{ - {name: "Exact match port range", + { + name: "Exact match port range", pr: portRange{ low: 8888, high: 8888, @@ -377,8 +392,10 @@ func Test_portRange_asComplexTernaryMatches(t *testing.T) { want: []portRangeTernaryRule{ {port: 8888, mask: 0xffff}, }, - wantErr: false}, - {name: "wildcard port range", + wantErr: false, + }, + { + name: "wildcard port range", pr: portRange{ low: 0, high: math.MaxUint16, @@ -386,8 +403,10 @@ func Test_portRange_asComplexTernaryMatches(t *testing.T) { want: []portRangeTernaryRule{ {port: 0, mask: 0}, }, - wantErr: false}, - {name: "Simplest port range", + wantErr: false, + }, + { + name: "Simplest port range", pr: portRange{ low: 0b0, // 0 high: 0b1, // 1 @@ -395,8 +414,10 @@ func Test_portRange_asComplexTernaryMatches(t *testing.T) { //want: []portRangeTernaryRule{ // {port: 0b0, mask: 0xfffe}, //}, - wantErr: false}, - {name: "Simplest port range2", + wantErr: false, + }, + { + name: "Simplest port range2", pr: portRange{ low: 0b01, // 1 high: 0b10, // 2 @@ -405,8 +426,10 @@ func Test_portRange_asComplexTernaryMatches(t *testing.T) { // {port: 0b01, mask: 0xffff}, // {port: 0b10, mask: 0xffff}, //}, - wantErr: false}, - {name: "Trivial ternary port range", + wantErr: false, + }, + { + name: "Trivial ternary port range", pr: portRange{ low: 0x0100, // 256 high: 0x01ff, // 511 @@ -415,8 +438,10 @@ func Test_portRange_asComplexTernaryMatches(t *testing.T) { //want: []portRangeTernaryRule{ // {port: 0x0100, mask: 0xff00}, //}, - wantErr: false}, - {name: "one to three range", + wantErr: false, + }, + { + name: "one to three range", pr: portRange{ low: 0b01, // 1 high: 0b11, // 3 @@ -425,33 +450,42 @@ func Test_portRange_asComplexTernaryMatches(t *testing.T) { // {port: 0b01, mask: 0xffff}, // {port: 0b10, mask: 0xfffe}, //}, - wantErr: false}, - {name: "True port range", + wantErr: false, + }, + { + name: "True port range", pr: portRange{ low: 0b00010, // 2 high: 0b11101, // 29 }, - wantErr: false}, - {name: "Worst case port range", + wantErr: false, + }, + { + name: "Worst case port range", pr: portRange{ low: 1, high: 65534, }, strategy: Ternary, - wantErr: false}, - {name: "low port filter", + wantErr: false, + }, + { + name: "low port filter", pr: portRange{ low: 0, high: 1023, }, strategy: Ternary, - wantErr: false}, - {name: "some small app filter", + wantErr: false, + }, + { + name: "some small app filter", pr: portRange{ low: 8080, high: 8084, }, - wantErr: false}, + wantErr: false, + }, } for _, tt := range tests { t.Run( @@ -464,22 +498,34 @@ func Test_portRange_asComplexTernaryMatches(t *testing.T) { if tt.want != nil && !reflect.DeepEqual(got, tt.want) { t.Errorf("asComplexTernaryMatches() got = %v, want %v", got, tt.want) } - // Do exhaustive test over entire value range. - for port := 0; port <= math.MaxUint16; port++ { - expectMatch := port >= int(tt.pr.low) && port <= int(tt.pr.high) - if matchesTernary(uint16(port), got) != expectMatch { - mod := " " - if !expectMatch { - mod = " not " - } - t.Errorf("Expected port %v to%vmatch against rules %v from range %+v", port, mod, got, tt.pr) - } - } + verifyPortRangeMatch(t, tt.pr, got) }, ) } } +func verifyPortRangeMatch( + t *testing.T, + pr portRange, + got []portRangeTernaryRule, +) { + // Do exhaustive test over entire value range. + for port := 0; port <= math.MaxUint16; port++ { + expectMatch := port >= int(pr.low) && port <= int(pr.high) + + if matchesTernary(uint16(port), got) != expectMatch { + mod := " " + if !expectMatch { + mod = " not " + } + t.Errorf( + "Expected port %v to%vmatch against rules %v from range %+v", + port, mod, got, pr, + ) + } + } +} + func Test_portRange_asTrivialTernaryMatch(t *testing.T) { tests := []struct { name string @@ -546,7 +592,7 @@ func Test_portRange_Width(t *testing.T) { } func Test_pdr_parseSDFFilter(t *testing.T) { - ueAddress := "17.0.0.1" + ueAddress := ipAddrPrimary newFilter := func(flowDesc string) *ie.IE { return ie.NewSDFFilter(flowDesc, "", "", "", 1) @@ -564,7 +610,7 @@ func Test_pdr_parseSDFFilter(t *testing.T) { sdfIE: newFilter("permit out udp from 192.168.1.1/32 to assigned 80-400"), direction: core, wantAppFilter: applicationFilter{ - srcIP: ip2int(net.ParseIP("192.168.1.1")), + srcIP: ip2int(net.ParseIP(ipAddrSecondary)), dstIP: ip2int(net.ParseIP(ueAddress)), srcPortRange: newRangeMatchPortRange(80, 400), dstPortRange: newWildcardPortRange(), @@ -581,7 +627,7 @@ func Test_pdr_parseSDFFilter(t *testing.T) { direction: access, wantAppFilter: applicationFilter{ srcIP: ip2int(net.ParseIP(ueAddress)), - dstIP: ip2int(net.ParseIP("192.168.1.1")), + dstIP: ip2int(net.ParseIP(ipAddrSecondary)), srcPortRange: newWildcardPortRange(), dstPortRange: newRangeMatchPortRange(80, 400), proto: 17, @@ -596,7 +642,7 @@ func Test_pdr_parseSDFFilter(t *testing.T) { sdfIE: newFilter("permit out udp from 192.168.1.1/32 80-400 to assigned"), direction: core, wantAppFilter: applicationFilter{ - srcIP: ip2int(net.ParseIP("192.168.1.1")), + srcIP: ip2int(net.ParseIP(ipAddrSecondary)), dstIP: ip2int(net.ParseIP(ueAddress)), srcPortRange: newRangeMatchPortRange(80, 400), dstPortRange: newWildcardPortRange(), @@ -613,7 +659,7 @@ func Test_pdr_parseSDFFilter(t *testing.T) { direction: access, wantAppFilter: applicationFilter{ srcIP: ip2int(net.ParseIP(ueAddress)), - dstIP: ip2int(net.ParseIP("192.168.1.1")), + dstIP: ip2int(net.ParseIP(ipAddrSecondary)), srcPortRange: newWildcardPortRange(), dstPortRange: newRangeMatchPortRange(80, 400), proto: 17, @@ -637,7 +683,7 @@ func Test_pdr_parseSDFFilter(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { p := &pdr{ - ueAddress: ip2int(net.ParseIP("17.0.0.1")), + ueAddress: ip2int(net.ParseIP(ipAddrPrimary)), srcIface: tt.direction, } if err := p.parseSDFFilter(tt.sdfIE); (err != nil) != tt.wantErr { @@ -652,7 +698,7 @@ func Test_pdr_parseSDFFilter(t *testing.T) { } func Test_pdr_parsePDI(t *testing.T) { - ueAddress := "17.0.0.1" + ueAddress := ipAddrPrimary type args struct { pdiIEs []*ie.IE diff --git a/pfcpiface/parse_sdf.go b/pfcpiface/parse_sdf.go index 117e6eddf..9ba14dde3 100644 --- a/pfcpiface/parse_sdf.go +++ b/pfcpiface/parse_sdf.go @@ -135,6 +135,24 @@ func parseFlowDesc(flowDesc, ueIP string) (*ipFilterRule, error) { } } + if err := processFlowFields(fields, ipf, parseLog, xform); err != nil { + return nil, err + } + + parseLog = parseLog.With("ip-filter", ipf) + parseLog.Debugln("flow description parsed successfully") + + return ipf, nil +} + +func processFlowFields( + fields []string, + ipf *ipFilterRule, + parseLog interface { + Errorln(...interface{}) + }, + xform func(int), +) error { for i := 3; i < len(fields); i++ { switch fields[i] { case "from": @@ -144,7 +162,7 @@ func parseFlowDesc(flowDesc, ueIP string) (*ipFilterRule, error) { err := ipf.src.parseNet(fields[i]) if err != nil { parseLog.Errorln(err) - return nil, err + return err } if fields[i+1] != "to" { @@ -153,7 +171,7 @@ func parseFlowDesc(flowDesc, ueIP string) (*ipFilterRule, error) { err = ipf.src.parsePort(fields[i]) if err != nil { parseLog.Errorln("src port parse failed", err) - return nil, err + return err } } case "to": @@ -163,7 +181,7 @@ func parseFlowDesc(flowDesc, ueIP string) (*ipFilterRule, error) { err := ipf.dst.parseNet(fields[i]) if err != nil { parseLog.Errorln(err) - return nil, err + return err } if i < len(fields)-1 { @@ -172,16 +190,12 @@ func parseFlowDesc(flowDesc, ueIP string) (*ipFilterRule, error) { err = ipf.dst.parsePort(fields[i]) if err != nil { parseLog.Errorln("dst port parse failed", err) - return nil, err + return err } } } } - - parseLog = parseLog.With("ip-filter", ipf) - parseLog.Debugln("flow description parsed successfully") - - return ipf, nil + return nil } func parseAction(action string) error { diff --git a/pfcpiface/parse_sdf_test.go b/pfcpiface/parse_sdf_test.go index b787419cc..8cf9611be 100644 --- a/pfcpiface/parse_sdf_test.go +++ b/pfcpiface/parse_sdf_test.go @@ -4,11 +4,17 @@ package pfcpiface import ( + "net" + "testing" + "github.com/omec-project/upf-epc/logger" "github.com/stretchr/testify/require" +) - "net" - "testing" +const ( + ipRouteCIDR = "10.0.0.1/32" + ipGateway = "60.60.0.1" + ipPeerAddress = "60.60.0.102" ) func mustParseCIDRNet(s string) *net.IPNet { @@ -35,34 +41,50 @@ func Test_endpoint_parseNet(t *testing.T) { want endpoint wantErr bool }{ - {name: "single IP", + { + name: "single IP", args: "10.0.0.1", - want: endpoint{IPNet: mustParseCIDRNet("10.0.0.1/32")}, - wantErr: false}, - {name: "single IP with /32 net", - args: "10.0.0.1/32", - want: endpoint{IPNet: mustParseCIDRNet("10.0.0.1/32")}, - wantErr: false}, - {name: "single IP with net", + want: endpoint{IPNet: mustParseCIDRNet(ipRouteCIDR)}, + wantErr: false, + }, + { + name: "single IP with /32 net", + args: ipRouteCIDR, + want: endpoint{IPNet: mustParseCIDRNet(ipRouteCIDR)}, + wantErr: false, + }, + { + name: "single IP with net", args: "10.0.0.1/24", want: endpoint{IPNet: mustParseCIDRNet("10.0.0.1/24")}, - wantErr: false}, - {name: "single IPv6", + wantErr: false, + }, + { + name: "single IPv6", args: "2001:db8:a0b:12f0::1/32", want: endpoint{IPNet: mustParseCIDRNet("2001:db8:a0b:12f0::1/32")}, - wantErr: false}, - {name: "invalid empty arg", + wantErr: false, + }, + { + name: "invalid empty arg", args: "", - wantErr: true}, - {name: "malformed IP missing octet", + wantErr: true, + }, + { + name: "malformed IP missing octet", args: "10.0.1/24", - wantErr: true}, - {name: "malformed IP", + wantErr: true, + }, + { + name: "malformed IP", args: "100", - wantErr: true}, - {name: "malformed IP double slash", + wantErr: true, + }, + { + name: "malformed IP double slash", args: "10.0.0.1/32/24", - wantErr: true}, + wantErr: true, + }, } for _, tt := range tests { t.Run( @@ -84,39 +106,59 @@ func Test_endpoint_parsePort(t *testing.T) { want endpoint wantErr bool }{ - {name: "single port", + { + name: "single port", args: "8080", want: endpoint{ports: newExactMatchPortRange(8080)}, - wantErr: false}, - {name: "single port range", + wantErr: false, + }, + { + name: "single port range", args: "8080-8080", want: endpoint{ports: newExactMatchPortRange(8080)}, - wantErr: false}, - {name: "normal port range", + wantErr: false, + }, + { + name: "normal port range", args: "8080-8084", want: endpoint{ports: newRangeMatchPortRange(8080, 8084)}, - wantErr: false}, - {name: "invalid empty port range", + wantErr: false, + }, + { + name: "invalid empty port range", args: "", - wantErr: true}, - {name: "invalid inverted port range", + wantErr: true, + }, + { + name: "invalid inverted port range", args: "100-90", - wantErr: true}, - {name: "malformed double dash port range", + wantErr: true, + }, + { + name: "malformed double dash port range", args: "100-200-300", - wantErr: true}, - {name: "missing high port range", + wantErr: true, + }, + { + name: "missing high port range", args: "100-", - wantErr: true}, - {name: "missing low port range", + wantErr: true, + }, + { + name: "missing low port range", args: "-100", - wantErr: true}, - {name: "wrong separator", + wantErr: true, + }, + { + name: "wrong separator", args: "200,300", - wantErr: true}, - {name: "malformed non-decimal number format", + wantErr: true, + }, + { + name: "malformed non-decimal number format", args: "0x0000-0xffff", - wantErr: true}, + wantErr: true, + }, } for _, tt := range tests { t.Run( @@ -240,12 +282,16 @@ func Test_parseFlowDesc(t *testing.T) { want *ipFilterRule wantErr bool }{ - {name: "empty", + { + name: "empty", args: args{ flowDesc: "", - ueIP: ""}, - wantErr: true}, - {name: "catch-all", + ueIP: "", + }, + wantErr: true, + }, + { + name: "catch-all", args: args{ flowDesc: "permit out ip from any to assigned", ueIP: ueIpString, @@ -262,8 +308,10 @@ func Test_parseFlowDesc(t *testing.T) { IPNet: newIpv4AddrAsNet(ueIpString), ports: newWildcardPortRange(), }, - }, wantErr: false}, - {name: "from IPv4 host TCP to don't care", + }, wantErr: false, + }, + { + name: "from IPv4 host TCP to don't care", args: args{ flowDesc: "permit out tcp from 60.60.0.102/32 to assigned", ueIP: ueIpString, @@ -280,8 +328,10 @@ func Test_parseFlowDesc(t *testing.T) { IPNet: newIpv4AddrAsNet(ueIpString), ports: newWildcardPortRange(), }, - }, wantErr: false}, - {name: "from don't care UDP to IPv4 host", + }, wantErr: false, + }, + { + name: "from don't care UDP to IPv4 host", args: args{ flowDesc: "permit out udp from any to 60.60.0.102", ueIP: ueIpString, @@ -295,11 +345,13 @@ func Test_parseFlowDesc(t *testing.T) { ports: newWildcardPortRange(), }, dst: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.102"), + IPNet: newIpv4AddrAsNet(ipPeerAddress), ports: newWildcardPortRange(), }, - }, wantErr: false}, - {name: "from IPv4 net to IPv4 host", + }, wantErr: false, + }, + { + name: "from IPv4 net to IPv4 host", args: args{ flowDesc: "permit out ip from 60.60.0.1/26 to 60.60.0.102", ueIP: ueIpString, @@ -313,11 +365,13 @@ func Test_parseFlowDesc(t *testing.T) { ports: newWildcardPortRange(), }, dst: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.102"), + IPNet: newIpv4AddrAsNet(ipPeerAddress), ports: newWildcardPortRange(), }, - }, wantErr: false}, - {name: "from single port", + }, wantErr: false, + }, + { + name: "from single port", args: args{ flowDesc: "permit out ip from 60.60.0.1 8888 to 60.60.0.102/26", ueIP: ueIpString, @@ -327,15 +381,17 @@ func Test_parseFlowDesc(t *testing.T) { direction: "out", proto: reservedProto, src: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.1"), + IPNet: newIpv4AddrAsNet(ipGateway), ports: newExactMatchPortRange(8888), }, dst: endpoint{ IPNet: mustParseCIDRNet("60.60.0.102/26"), ports: newWildcardPortRange(), }, - }, wantErr: false}, - {name: "from single port range", + }, wantErr: false, + }, + { + name: "from single port range", args: args{ flowDesc: "permit out ip from 60.60.0.1 8888-8888 to 60.60.0.102/26", ueIP: ueIpString, @@ -345,15 +401,17 @@ func Test_parseFlowDesc(t *testing.T) { direction: "out", proto: reservedProto, src: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.1"), + IPNet: newIpv4AddrAsNet(ipGateway), ports: newExactMatchPortRange(8888), }, dst: endpoint{ IPNet: mustParseCIDRNet("60.60.0.102/26"), ports: newWildcardPortRange(), }, - }, wantErr: false}, - {name: "to single port", + }, wantErr: false, + }, + { + name: "to single port", args: args{ flowDesc: "permit out ip from 60.60.0.1 to 60.60.0.102 9999", ueIP: ueIpString, @@ -363,15 +421,17 @@ func Test_parseFlowDesc(t *testing.T) { direction: "out", proto: reservedProto, src: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.1"), + IPNet: newIpv4AddrAsNet(ipGateway), ports: newWildcardPortRange(), }, dst: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.102"), + IPNet: newIpv4AddrAsNet(ipPeerAddress), ports: newExactMatchPortRange(9999), }, - }, wantErr: false}, - {name: "from single port to single port", + }, wantErr: false, + }, + { + name: "from single port to single port", args: args{ flowDesc: "permit out ip from 60.60.0.1 8888 to 60.60.0.102 9999", ueIP: ueIpString, @@ -381,15 +441,17 @@ func Test_parseFlowDesc(t *testing.T) { direction: "out", proto: reservedProto, src: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.1"), + IPNet: newIpv4AddrAsNet(ipGateway), ports: newExactMatchPortRange(8888), }, dst: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.102"), + IPNet: newIpv4AddrAsNet(ipPeerAddress), ports: newExactMatchPortRange(9999), }, - }, wantErr: false}, - {name: "from single port range to single port range", + }, wantErr: false, + }, + { + name: "from single port range to single port range", args: args{ flowDesc: "permit out ip from 60.60.0.1 8888-8888 to 60.60.0.102 9999-9999", ueIP: ueIpString, @@ -399,18 +461,21 @@ func Test_parseFlowDesc(t *testing.T) { direction: "out", proto: reservedProto, src: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.1"), + IPNet: newIpv4AddrAsNet(ipGateway), ports: newExactMatchPortRange(8888), }, dst: endpoint{ - IPNet: newIpv4AddrAsNet("60.60.0.102"), + IPNet: newIpv4AddrAsNet(ipPeerAddress), ports: newExactMatchPortRange(9999), }, - }, wantErr: false}, - {name: "to unknown assigned UE IP (uplink)", + }, wantErr: false, + }, + { + name: "to unknown assigned UE IP (uplink)", args: args{ flowDesc: "permit out udp from 60.60.0.1/32 to assigned", - ueIP: "0.0.0.0"}, + ueIP: "0.0.0.0", + }, want: &ipFilterRule{ action: "permit", direction: "out", @@ -423,7 +488,8 @@ func Test_parseFlowDesc(t *testing.T) { IPNet: newIpv4WildcardNet(), ports: newWildcardPortRange(), }, - }, wantErr: false}, + }, wantErr: false, + }, } for _, tt := range tests { t.Run( diff --git a/pfcpiface/session_qer.go b/pfcpiface/session_qer.go index 869e6ab2d..35379147a 100644 --- a/pfcpiface/session_qer.go +++ b/pfcpiface/session_qer.go @@ -102,6 +102,27 @@ func (s *PFCPSession) MarkSessionQer(qers []qer) { // if len(sessQerIDList) = 0 : no session QER // if len(sessQerIDList) = 3 : TBD (UE level QER handling). // Currently handle same as len = 2 + var ( + sessionIdx int + sessQerID uint32 + ) + sessionIdx, sessQerID = findSessionQER(qers, sessQerIDList) + + logger.PfcpLog.Infoln("session QER found. QER ID:", sessQerID) + + qers[sessionIdx].qosLevel = SessionQos + + for i := range s.pdrs { + // remove common qerID from pdr's qer list + idx := findItemIndex(s.pdrs[i].qerIDList, sessQerID) + if idx != len(s.pdrs[i].qerIDList) { + s.pdrs[i].qerIDList = append(s.pdrs[i].qerIDList[:idx], s.pdrs[i].qerIDList[idx+1:]...) + s.pdrs[i].qerIDList = append(s.pdrs[i].qerIDList, sessQerID) + } + } +} + +func findSessionQER(qers []qer, sessQerIDList []uint32) (int, uint32) { var ( sessionIdx int sessionMbr uint64 @@ -126,19 +147,7 @@ func (s *PFCPSession) MarkSessionQer(qers []qer) { } } } - - logger.PfcpLog.Infoln("session QER found. QER ID:", sessQerID) - - qers[sessionIdx].qosLevel = SessionQos - - for i := range s.pdrs { - // remove common qerID from pdr's qer list - idx := findItemIndex(s.pdrs[i].qerIDList, sessQerID) - if idx != len(s.pdrs[i].qerIDList) { - s.pdrs[i].qerIDList = append(s.pdrs[i].qerIDList[:idx], s.pdrs[i].qerIDList[idx+1:]...) - s.pdrs[i].qerIDList = append(s.pdrs[i].qerIDList, sessQerID) - } - } + return sessionIdx, sessQerID } // RemoveQER removes qer from existing list of QERs in the session. diff --git a/pfcpiface/up4.go b/pfcpiface/up4.go index 9e6c2ee96..ed7ab0ab8 100644 --- a/pfcpiface/up4.go +++ b/pfcpiface/up4.go @@ -41,6 +41,7 @@ const ( // TODO: we might want to make it configurable in future. DefaultQFI = 9 ) +const msgNumberOfFreeIDs = "number of free IDs" var ( p4RtcServerIP = flag.String("p4RtcServerIP", "", "P4 Server ip") @@ -890,7 +891,7 @@ func (up4 *UP4) allocateAppMeterCellID() (uint32, error) { "no free AppMeter Cell IDs available") } - logger.PfcpLog.With("allocated ID", allocated, "number of free IDs", up4.appMeterCellIDsPool.Cardinality()).Debugln("application meter cell ID allocated") + logger.PfcpLog.With("allocated ID", allocated, msgNumberOfFreeIDs, up4.appMeterCellIDsPool.Cardinality()).Debugln("application meter cell ID allocated") return allocated.(uint32), nil } @@ -903,7 +904,7 @@ func (up4 *UP4) releaseAppMeterCellID(allocated uint32) { up4.appMeterCellIDsPool.Add(allocated) - logger.PfcpLog.With("released ID", allocated, "number of free IDs", up4.appMeterCellIDsPool.Cardinality()).Debugln("application meter cell ID released") + logger.PfcpLog.With("released ID", allocated, msgNumberOfFreeIDs, up4.appMeterCellIDsPool.Cardinality()).Debugln("application meter cell ID released") } func (up4 *UP4) allocateSessionMeterCellID() (uint32, error) { @@ -914,7 +915,7 @@ func (up4 *UP4) allocateSessionMeterCellID() (uint32, error) { "no free SessionMeter Cell IDs available") } - logger.PfcpLog.With("allocated ID", allocated, "number of free IDs", up4.sessMeterCellIDsPool.Cardinality()).Debugln("session meter cell ID allocated") + logger.PfcpLog.With("allocated ID", allocated, msgNumberOfFreeIDs, up4.sessMeterCellIDsPool.Cardinality()).Debugln("session meter cell ID allocated") return allocated.(uint32), nil } @@ -927,7 +928,7 @@ func (up4 *UP4) releaseSessionMeterCellID(allocated uint32) { up4.sessMeterCellIDsPool.Add(allocated) - logger.PfcpLog.With("released ID", allocated, "number of free IDs", up4.sessMeterCellIDsPool.Cardinality()).Debugln("session meter cell ID released") + logger.PfcpLog.With("released ID", allocated, msgNumberOfFreeIDs, up4.sessMeterCellIDsPool.Cardinality()).Debugln("session meter cell ID released") } func (up4 *UP4) updateUEAddrAndFSEIDMappings(pdr pdr) { @@ -1302,7 +1303,7 @@ func (up4 *UP4) modifyUP4ForwardingConfiguration(pdrs []pdr, allFARs []far, qers return ErrNotFoundWithParam("allocated GTP tunnel peer ID", "tunnel params", tunnelParameters) } - var sessMeter = meter{meterTypeSession, 0, 0} + sessMeter := meter{meterTypeSession, 0, 0} if len(pdr.qerIDList) == 2 { // if 2 QERs are provided, the second one is Session QER sessMeter = up4.meters[meterID{ @@ -1352,7 +1353,7 @@ func (up4 *UP4) modifyUP4ForwardingConfiguration(pdrs []pdr, allFARs []far, qers } } - var appMeter = meter{meterTypeApplication, 0, 0} + appMeter := meter{meterTypeApplication, 0, 0} if len(pdr.qerIDList) != 0 { // if only 1 QER provided, it's an application QER // if 2 QERs provided, the first one is an application QER diff --git a/pfcpiface/upf.go b/pfcpiface/upf.go index 811ded329..da8012106 100644 --- a/pfcpiface/upf.go +++ b/pfcpiface/upf.go @@ -145,7 +145,18 @@ func NewUPF(conf *Conf, fp datapath) *upf { readTimeout: time.Second * time.Duration(conf.ReadTimeout), n4addr: conf.N4Addr, } + if !setupPeersAndInterfaces(u, conf) { + return nil + } + initTimersAndIPPool(u, conf) + + u.SetUpfInfo(u, conf) + + return u +} +func setupPeersAndInterfaces(u *upf, conf *Conf) bool { + var err error if len(conf.CPIface.Peers) > 0 { u.peers = make([]string, len(conf.CPIface.Peers)) nc := copy(u.peers, conf.CPIface.Peers) @@ -159,16 +170,20 @@ func NewUPF(conf *Conf, fp datapath) *upf { u.accessIP, err = GetUnicastAddressFromInterface(conf.AccessIface.IfName) if err != nil { logger.PfcpLog.Errorln(err) - return nil + return false } u.coreIP, err = GetUnicastAddressFromInterface(conf.CoreIface.IfName) if err != nil { logger.PfcpLog.Errorln(err) - return nil + return false } } + return true +} +func initTimersAndIPPool(u *upf, conf *Conf) { + var err error u.respTimeout, err = time.ParseDuration(conf.RespTimeout) if err != nil { logger.PfcpLog.Fatalln("unable to parse resp_timeout") @@ -189,8 +204,4 @@ func NewUPF(conf *Conf, fp datapath) *upf { logger.PfcpLog.Fatalln("ip pool init failed", err) } } - - u.SetUpfInfo(u, conf) - - return u } diff --git a/pkg/fake_bess/fake_bess.go b/pkg/fake_bess/fake_bess.go index 821d83fd4..5ca29e5c8 100644 --- a/pkg/fake_bess/fake_bess.go +++ b/pkg/fake_bess/fake_bess.go @@ -10,6 +10,8 @@ import ( "google.golang.org/grpc" ) +const msgUnexpectedMessageType = "unexpected message type" + type FakeBESS struct { grpcServer *grpc.Server service *fakeBessService @@ -53,13 +55,13 @@ func (b *FakeBESS) GetPdrTableEntries() (entries map[uint32][]FakePdr) { for _, m := range msgs { e, ok := m.(*bess_pb.WildcardMatchCommandAddArg) if !ok { - panic("unexpected message type") + panic(msgUnexpectedMessageType) } pdr := UnmarshalPdr(e) entries[pdr.PdrID] = append(entries[pdr.PdrID], pdr) } - return + return entries } func (b *FakeBESS) GetFarTableEntries() (entries map[uint32]FakeFar) { @@ -68,12 +70,12 @@ func (b *FakeBESS) GetFarTableEntries() (entries map[uint32]FakeFar) { for _, m := range msgs { e, ok := m.(*bess_pb.ExactMatchCommandAddArg) if !ok { - panic("unexpected message type") + panic(msgUnexpectedMessageType) } far := UnmarshalFar(e) entries[far.FarID] = far } - return + return entries } // Session QERs are missing a QerID and are therefore returned as a slice, not map. @@ -82,11 +84,11 @@ func (b *FakeBESS) GetSessionQerTableEntries() (entries []FakeQer) { for _, m := range msgs { e, ok := m.(*bess_pb.QosCommandAddArg) if !ok { - panic("unexpected message type") + panic(msgUnexpectedMessageType) } entries = append(entries, UnmarshalSessionQer(e)) } - return + return entries } func (b *FakeBESS) GetAppQerTableEntries() (entries []FakeQer) { @@ -94,9 +96,9 @@ func (b *FakeBESS) GetAppQerTableEntries() (entries []FakeQer) { for _, m := range msgs { e, ok := m.(*bess_pb.QosCommandAddArg) if !ok { - panic("unexpected message type") + panic(msgUnexpectedMessageType) } entries = append(entries, UnmarshalAppQer(e)) } - return + return entries } diff --git a/pkg/fake_bess/fake_bess_service.go b/pkg/fake_bess/fake_bess_service.go index 75c1a8a2c..4618f992a 100644 --- a/pkg/fake_bess/fake_bess_service.go +++ b/pkg/fake_bess/fake_bess_service.go @@ -29,6 +29,14 @@ const ( deleteCmd = "delete" ) +const ( + msgUpdatedEntry = "updated existing entry %v" + msgAddedEntry = "added new entry %v" + msgEntryNotFound = "entry not found: %v" + msgDeletedEntry = "deleted existing entry %v" + msgShouldNotHappen = "should not happen" +) + type FakePdr struct { srcIface uint8 srcIfaceMask uint8 @@ -255,7 +263,7 @@ func UnmarshalPdr(wc *bess_pb.WildcardMatchCommandAddArg) (p FakePdr) { p.qerID = uint32(wc.Valuesv[3].GetValueInt()) p.farID = uint32(wc.Valuesv[4].GetValueInt()) - return + return p } func UnmarshalFar(em *bess_pb.ExactMatchCommandAddArg) (f FakeFar) { @@ -271,7 +279,7 @@ func UnmarshalFar(em *bess_pb.ExactMatchCommandAddArg) (f FakeFar) { f.tunnelTEID = uint32(em.Values[4].GetValueInt()) f.tunnelPort = uint16(em.Values[5].GetValueInt()) - return + return f } func UnmarshalSessionQer(qc *bess_pb.QosCommandAddArg) (q FakeQer) { @@ -279,7 +287,7 @@ func UnmarshalSessionQer(qc *bess_pb.QosCommandAddArg) (q FakeQer) { // srcIface = uint32(qc.Fields[0].GetValueInt()) q.fseID = qc.Fields[1].GetValueInt() - return + return q } func UnmarshalAppQer(qc *bess_pb.QosCommandAddArg) (q FakeQer) { @@ -291,7 +299,7 @@ func UnmarshalAppQer(qc *bess_pb.QosCommandAddArg) (q FakeQer) { // Values q.qfi = uint8(qc.Values[0].GetValueInt()) - return + return q } // Fake BESS module @@ -314,7 +322,7 @@ func (b *baseModule) HandleRequest(cmd string, arg *anypb.Any) (err error) { return status.Errorf(codes.InvalidArgument, "invalid command: %v", cmd) } - return + return err } type wildcardModule struct { @@ -331,33 +339,14 @@ func (w *wildcardModule) GetState() (msgs []proto.Message) { func (w *wildcardModule) HandleRequest(cmd string, arg *anypb.Any) (err error) { if err = w.baseModule.HandleRequest(cmd, arg); err != nil { - return + return err } log := logger.BessLog.With("module", w.Name(), "cmd", cmd) switch cmd { case addCmd: - wc := &bess_pb.WildcardMatchCommandAddArg{} - err = arg.UnmarshalTo(wc) - if err != nil { - return err - } - var existing *bess_pb.WildcardMatchCommandAddArg - for _, e := range w.entries { - if fieldsAreEqual(e.GetValues(), wc.GetValues()) && - fieldsAreEqual(e.GetMasks(), wc.GetMasks()) { - existing = e - } - } - if existing != nil { - log.Debugf("updated existing entry %v", existing) - existing.Reset() - proto.Merge(existing, wc) - } else { - log.Debugf("added new entry %v", wc) - w.entries = append(w.entries, wc) - } + return w.handleAddEntry(arg) case deleteCmd: wc := &bess_pb.WildcardMatchCommandDeleteArg{} err = arg.UnmarshalTo(wc) @@ -372,9 +361,9 @@ func (w *wildcardModule) HandleRequest(cmd string, arg *anypb.Any) (err error) { } } if idx == -1 { - return status.Errorf(codes.NotFound, "entry not found: %v", wc) + return status.Errorf(codes.NotFound, msgEntryNotFound, wc) } else { - log.Debugf("deleted existing entry %v", w.entries[idx]) + log.Debugf(msgDeletedEntry, w.entries[idx]) w.entries = append(w.entries[:idx], w.entries[idx+1:]...) } case clearCmd: @@ -386,7 +375,36 @@ func (w *wildcardModule) HandleRequest(cmd string, arg *anypb.Any) (err error) { // clear all rules w.entries = nil default: - panic("should not happen") + panic(msgShouldNotHappen) + } + + return nil +} + +func (w *wildcardModule) handleAddEntry(arg *anypb.Any) error { + log := logger.BessLog.With("module", w.Name(), "cmd", addCmd) + + wc := &bess_pb.WildcardMatchCommandAddArg{} + if err := arg.UnmarshalTo(wc); err != nil { + return err + } + + var existing *bess_pb.WildcardMatchCommandAddArg + for _, e := range w.entries { + if fieldsAreEqual(e.GetValues(), wc.GetValues()) && + fieldsAreEqual(e.GetMasks(), wc.GetMasks()) { + existing = e + break + } + } + + if existing != nil { + log.Debugf(msgUpdatedEntry, existing) + existing.Reset() + proto.Merge(existing, wc) + } else { + log.Debugf(msgAddedEntry, wc) + w.entries = append(w.entries, wc) } return nil @@ -401,37 +419,19 @@ func (e *exactMatchModule) GetState() (msgs []proto.Message) { for _, em := range e.entries { msgs = append(msgs, em) } - return + return msgs } func (e *exactMatchModule) HandleRequest(cmd string, arg *anypb.Any) (err error) { if err = e.baseModule.HandleRequest(cmd, arg); err != nil { - return + return err } log := logger.BessLog.With("module", e.Name(), "cmd", cmd) switch cmd { case addCmd: - em := &bess_pb.ExactMatchCommandAddArg{} - err = arg.UnmarshalTo(em) - if err != nil { - return err - } - var existing *bess_pb.ExactMatchCommandAddArg - for _, et := range e.entries { - if fieldsAreEqual(et.GetFields(), em.GetFields()) { - existing = et - } - } - if existing != nil { - log.Debugf("updated existing entry %v", em) - existing.Reset() - proto.Merge(existing, em) - } else { - log.Debugf("added new entry %v", em) - e.entries = append(e.entries, em) - } + return e.handleAddEntry(arg) case deleteCmd: em := &bess_pb.ExactMatchCommandDeleteArg{} err = arg.UnmarshalTo(em) @@ -445,9 +445,9 @@ func (e *exactMatchModule) HandleRequest(cmd string, arg *anypb.Any) (err error) } } if idx == -1 { - return status.Errorf(codes.NotFound, "entry not found: %v", em) + return status.Errorf(codes.NotFound, msgEntryNotFound, em) } else { - log.Debugf("deleted existing entry %v", e.entries[idx]) + log.Debugf(msgDeletedEntry, e.entries[idx]) e.entries = append(e.entries[:idx], e.entries[idx+1:]...) } case clearCmd: @@ -459,7 +459,35 @@ func (e *exactMatchModule) HandleRequest(cmd string, arg *anypb.Any) (err error) // clear all rules e.entries = nil default: - panic("should not happen") + panic(msgShouldNotHappen) + } + + return nil +} + +func (e *exactMatchModule) handleAddEntry(arg *anypb.Any) error { + log := logger.BessLog.With("module", e.Name(), "cmd", addCmd) + + em := &bess_pb.ExactMatchCommandAddArg{} + if err := arg.UnmarshalTo(em); err != nil { + return err + } + + var existing *bess_pb.ExactMatchCommandAddArg + for _, et := range e.entries { + if fieldsAreEqual(et.GetFields(), em.GetFields()) { + existing = et + break + } + } + + if existing != nil { + log.Debugf(msgUpdatedEntry, em) + existing.Reset() + proto.Merge(existing, em) + } else { + log.Debugf(msgAddedEntry, em) + e.entries = append(e.entries, em) } return nil @@ -477,39 +505,21 @@ func (q *qosModule) GetState() (msgs []proto.Message) { for _, em := range q.entries { msgs = append(msgs, em) } - return + return msgs } func (q *qosModule) HandleRequest(cmd string, arg *anypb.Any) (err error) { q.mutex.Lock() defer q.mutex.Unlock() if err = q.baseModule.HandleRequest(cmd, arg); err != nil { - return + return err } log := logger.BessLog.With("module", q.Name(), "cmd", cmd) switch cmd { case addCmd: - wc := &bess_pb.QosCommandAddArg{} - err = arg.UnmarshalTo(wc) - if err != nil { - return err - } - var existing *bess_pb.QosCommandAddArg - for _, e := range q.entries { - if fieldsAreEqual(e.GetFields(), wc.GetFields()) { - existing = e - } - } - if existing != nil { - log.Debugf("updated existing entry %v", existing) - existing.Reset() - proto.Merge(existing, wc) - } else { - log.Debugf("added new entry %v", wc) - q.entries = append(q.entries, wc) - } + return q.handleAddEntry(arg) case deleteCmd: qc := &bess_pb.QosCommandDeleteArg{} err = arg.UnmarshalTo(qc) @@ -523,9 +533,9 @@ func (q *qosModule) HandleRequest(cmd string, arg *anypb.Any) (err error) { } } if idx == -1 { - return status.Errorf(codes.NotFound, "entry not found: %v", qc) + return status.Errorf(codes.NotFound, msgEntryNotFound, qc) } else { - log.Debugf("deleted existing entry %v", q.entries[idx]) + log.Debugf(msgDeletedEntry, q.entries[idx]) q.entries = append(q.entries[:idx], q.entries[idx+1:]...) } case clearCmd: @@ -537,7 +547,35 @@ func (q *qosModule) HandleRequest(cmd string, arg *anypb.Any) (err error) { // clear all rules q.entries = nil default: - panic("should not happen") + panic(msgShouldNotHappen) + } + + return nil +} + +func (q *qosModule) handleAddEntry(arg *anypb.Any) error { + log := logger.BessLog.With("module", q.Name(), "cmd", addCmd) + + wc := &bess_pb.QosCommandAddArg{} + if err := arg.UnmarshalTo(wc); err != nil { + return err + } + + var existing *bess_pb.QosCommandAddArg + for _, e := range q.entries { + if fieldsAreEqual(e.GetFields(), wc.GetFields()) { + existing = e + break + } + } + + if existing != nil { + log.Debugf(msgUpdatedEntry, existing) + existing.Reset() + proto.Merge(existing, wc) + } else { + log.Debugf(msgAddedEntry, wc) + q.entries = append(q.entries, wc) } return nil @@ -552,37 +590,19 @@ func (q *gtpuPathMonitoringModule) GetState() (msgs []proto.Message) { for _, em := range q.entries { msgs = append(msgs, em) } - return + return msgs } func (q *gtpuPathMonitoringModule) HandleRequest(cmd string, arg *anypb.Any) (err error) { if err = q.baseModule.HandleRequest(cmd, arg); err != nil { - return + return err } log := logger.BessLog.With("module", q.Name(), "cmd", cmd) switch cmd { case addCmd: - wc := &bess_pb.GtpuPathMonitoringCommandAddDeleteArg{} - err = arg.UnmarshalTo(wc) - if err != nil { - return err - } - var existing *bess_pb.GtpuPathMonitoringCommandAddDeleteArg - for _, e := range q.entries { - if e.GetGnbIp() == wc.GetGnbIp() { - existing = e - } - } - if existing != nil { - log.Debugf("updated existing entry %v", existing) - existing.Reset() - proto.Merge(existing, wc) - } else { - log.Debugf("added new entry %v", wc) - q.entries = append(q.entries, wc) - } + return q.handleAddEntry(arg) case deleteCmd: qc := &bess_pb.GtpuPathMonitoringCommandAddDeleteArg{} err = arg.UnmarshalTo(qc) @@ -596,9 +616,9 @@ func (q *gtpuPathMonitoringModule) HandleRequest(cmd string, arg *anypb.Any) (er } } if idx == -1 { - return status.Errorf(codes.NotFound, "entry not found: %v", qc) + return status.Errorf(codes.NotFound, msgEntryNotFound, qc) } else { - log.Debugf("deleted existing entry %v", q.entries[idx]) + log.Debugf(msgDeletedEntry, q.entries[idx]) q.entries = append(q.entries[:idx], q.entries[idx+1:]...) } case clearCmd: @@ -610,7 +630,35 @@ func (q *gtpuPathMonitoringModule) HandleRequest(cmd string, arg *anypb.Any) (er // clear all rules q.entries = nil default: - panic("should not happen") + panic(msgShouldNotHappen) + } + + return nil +} + +func (q *gtpuPathMonitoringModule) handleAddEntry(arg *anypb.Any) error { + log := logger.BessLog.With("module", q.Name(), "cmd", addCmd) + + wc := &bess_pb.GtpuPathMonitoringCommandAddDeleteArg{} + if err := arg.UnmarshalTo(wc); err != nil { + return err + } + + var existing *bess_pb.GtpuPathMonitoringCommandAddDeleteArg + for _, e := range q.entries { + if e.GetGnbIp() == wc.GetGnbIp() { + existing = e + break + } + } + + if existing != nil { + log.Debugf(msgUpdatedEntry, existing) + existing.Reset() + proto.Merge(existing, wc) + } else { + log.Debugf(msgAddedEntry, wc) + q.entries = append(q.entries, wc) } return nil diff --git a/ptf/lib/trex_utils.py b/ptf/lib/trex_utils.py index 4a30eed72..993c25852 100644 --- a/ptf/lib/trex_utils.py +++ b/ptf/lib/trex_utils.py @@ -268,30 +268,59 @@ def start_and_monitor_port_stats( f"{' *' if bad_sample else ''}" ) - if bad_sample: - if elapsed > ramp_up_timeout: - client.stop(ports=[tx_port]) - raise Exception( - f"TX port ({tx_port}) did not reach or sustain " - f"min sending rate ({to_readable(min_tx_bps)})" - ) - else: - # Discard last sample - samples.tx_bps.pop() - samples.tx_pps.pop() - samples.rx_bps.pop() - samples.rx_pps.pop() - - if len(samples.tx_bps) == num_samples: - # We have enough samples. - client.stop(ports=[tx_port]) - client.wait_on_traffic(ports=[tx_port], timeout=2) - break + if _handle_sampling_state( + client=client, + samples=samples, + tx_port=tx_port, + min_tx_bps=min_tx_bps, + tx_bps=tx_bps, + elapsed=elapsed, + ramp_up_timeout=ramp_up_timeout, + num_samples=num_samples, + ): time.sleep(interval - elapsed % interval) return samples +def _handle_sampling_state( + client: STLClient, + samples: RateSamples, + tx_port: int, + min_tx_bps: int, + tx_bps: float, + elapsed: float, + ramp_up_timeout: int, + num_samples: int, +) -> bool: + """ + Handles bad sample logic and completion condition. + :return: True if sampling should stop, False otherwise + """ + + bad_sample = tx_bps < min_tx_bps + + if bad_sample: + if elapsed > ramp_up_timeout: + client.stop(ports=[tx_port]) + raise Exception( + f"TX port ({tx_port}) did not reach or sustain " + f"min sending rate ({to_readable(min_tx_bps)})" + ) + else: + # Discard last sample + samples.tx_bps.pop() + samples.tx_pps.pop() + samples.rx_bps.pop() + samples.rx_pps.pop() + return False + + if len(samples.tx_bps) == num_samples: + client.stop(ports=[tx_port]) + client.wait_on_traffic(ports=[tx_port], timeout=2) + return True + + return False def get_port_stats(port: int, stats) -> PortStats: port_stats = stats.get(port) diff --git a/test/integration/basic_test.go b/test/integration/basic_test.go index aac67c9ba..60967550b 100644 --- a/test/integration/basic_test.go +++ b/test/integration/basic_test.go @@ -17,6 +17,11 @@ import ( "github.com/wmnsk/go-pfcp/message" ) +const ( + aclPermitHTTPUDP = "permit out udp from any 80-80 to assigned" + anyIPv4AddressZ = "0.0.0.0" +) + func TestUPFBasedUeIPAllocation(t *testing.T) { // TODO: verify if UEIP bit is set in the UP Function Features of PFCP Association Response setup(t, ConfigUPFBasedIPAllocation) @@ -27,7 +32,7 @@ func TestUPFBasedUeIPAllocation(t *testing.T) { sliceID: 1, nbAddress: nodeBAddress, upfN3Address: upfN3Address, - sdfFilter: "permit out udp from any 80-80 to assigned", + sdfFilter: aclPermitHTTPUDP, ulTEID: 15, dlTEID: 16, QFI: 0x9, @@ -37,7 +42,7 @@ func TestUPFBasedUeIPAllocation(t *testing.T) { ueAddress: "10.250.0.1", appFilter: appFilter{ proto: 0x11, - appIP: net.ParseIP("0.0.0.0"), + appIP: net.ParseIP(anyIPv4AddressZ), appPrefixLen: 0, appPort: portRange{ 80, 80, @@ -210,7 +215,7 @@ func TestSingleUEAttachAndDetach(t *testing.T) { nbAddress: nodeBAddress, ueAddress: ueAddress, upfN3Address: upfN3Address, - sdfFilter: "permit out udp from any 80-80 to assigned", + sdfFilter: aclPermitHTTPUDP, ulTEID: 15, dlTEID: 16, QFI: 0x9, @@ -218,7 +223,7 @@ func TestSingleUEAttachAndDetach(t *testing.T) { expected: p4RtValues{ appFilter: appFilter{ proto: 0x11, - appIP: net.ParseIP("0.0.0.0"), + appIP: net.ParseIP(anyIPv4AddressZ), appPrefixLen: 0, appPort: portRange{ 80, 80, @@ -291,7 +296,7 @@ func TestSingleUEAttachAndDetach(t *testing.T) { expected: p4RtValues{ appFilter: appFilter{ proto: 0x11, - appIP: net.ParseIP("0.0.0.0"), + appIP: net.ParseIP(anyIPv4AddressZ), appPrefixLen: 0, appPort: portRange{ 80, 80, @@ -323,7 +328,7 @@ func TestSingleUEAttachAndDetach(t *testing.T) { expected: p4RtValues{ appFilter: appFilter{ proto: 0x11, - appIP: net.ParseIP("0.0.0.0"), + appIP: net.ParseIP(anyIPv4AddressZ), appPrefixLen: 0, appPort: portRange{ 80, 80, @@ -355,7 +360,7 @@ func TestSingleUEAttachAndDetach(t *testing.T) { expected: p4RtValues{ appFilter: appFilter{ proto: 0x11, - appIP: net.ParseIP("0.0.0.0"), + appIP: net.ParseIP(anyIPv4AddressZ), appPrefixLen: 0, appPort: portRange{ 80, 80, @@ -388,7 +393,7 @@ func TestSingleUEAttachAndDetach(t *testing.T) { expected: p4RtValues{ appFilter: appFilter{ proto: 0x11, - appIP: net.ParseIP("0.0.0.0"), + appIP: net.ParseIP(anyIPv4AddressZ), appPrefixLen: 0, appPort: portRange{ 80, 80, @@ -421,7 +426,7 @@ func TestSingleUEAttachAndDetach(t *testing.T) { expected: p4RtValues{ appFilter: appFilter{ proto: 0x11, - appIP: net.ParseIP("0.0.0.0"), + appIP: net.ParseIP(anyIPv4AddressZ), appPrefixLen: 0, appPort: portRange{ 80, 80, @@ -450,7 +455,7 @@ func TestUEBuffering(t *testing.T) { nbAddress: nodeBAddress, ueAddress: ueAddress, upfN3Address: upfN3Address, - sdfFilter: "permit out udp from any 80-80 to assigned", + sdfFilter: aclPermitHTTPUDP, ulTEID: 15, dlTEID: 16, QFI: 0x9, @@ -458,7 +463,7 @@ func TestUEBuffering(t *testing.T) { expected: p4RtValues{ appFilter: appFilter{ proto: 0x11, - appIP: net.ParseIP("0.0.0.0"), + appIP: net.ParseIP(anyIPv4AddressZ), appPrefixLen: 0, appPort: portRange{ 80, 80, diff --git a/test/integration/framework.go b/test/integration/framework.go index 766ca537d..b6b2c6794 100644 --- a/test/integration/framework.go +++ b/test/integration/framework.go @@ -64,6 +64,11 @@ const ( deviceConfigPath = "../../conf/p4/bin/bmv2.json" ) +const ( + localLoopbackAltIP = "127.0.0.8" + localhostIPIn = "127.0.0.1" +) + type UEState uint8 const ( @@ -227,11 +232,11 @@ func waitForPFCPAssociationSetup(pfcpClient *pfcpsim.PFCPClient) error { } func waitForMockUP4ToStart() error { - return waitForPortOpen("tcp", "127.0.0.1", "50001") + return waitForPortOpen("tcp", localhostIPIn, "50001") } func waitForBESSFakeToStart() error { - return waitForPortOpen("tcp", "127.0.0.1", "10514") + return waitForPortOpen("tcp", localhostIPIn, "10514") } func isDatapathUP4() bool { @@ -281,7 +286,7 @@ func mustInitCountersWithDummyValue() { } func MustStartMockUP4() { - providers.MustRunDockerContainer(ContainerNameMockUP4, ImageNameMockUP4, "--topo single", "127.0.0.1", []string{"50001/tcp"}, "", DockerTestNetwork) + providers.MustRunDockerContainer(ContainerNameMockUP4, ImageNameMockUP4, "--topo single", localhostIPIn, []string{"50001/tcp"}, "", DockerTestNetwork) err := waitForMockUP4ToStart() if err != nil { panic(err) @@ -296,7 +301,7 @@ func MustStopMockUP4() { func MustStartPFCPAgent() { providers.MustRunDockerContainer(ContainerNamePFCPAgent, ImageNamePFCPAgent, "-config /config/upf.jsonc", - "127.0.0.8", []string{"8805/udp", "8080/tcp"}, "/tmp:/config", DockerTestNetwork) + localLoopbackAltIP, []string{"8805/udp", "8080/tcp"}, "/tmp:/config", DockerTestNetwork) } func MustStopPFCPAgent() { @@ -331,15 +336,15 @@ func setup(t *testing.T, configType uint32) { MustStartPFCPAgent() case ModeNative: upfConf := GetConfig(os.Getenv(EnvDatapath), configType) - upfConf.N4Addr = "127.0.0.8" + upfConf.N4Addr = localLoopbackAltIP pfcpAgent = pfcpiface.NewPFCPIface(upfConf) go pfcpAgent.Run() default: t.Fatal("Unexpected test mode") } - pfcpClient = pfcpsim.NewPFCPClient("127.0.0.1") - err := pfcpClient.ConnectN4("127.0.0.8") + pfcpClient = pfcpsim.NewPFCPClient(localhostIPIn) + err := pfcpClient.ConnectN4(localLoopbackAltIP) require.NoErrorf(t, err, "failed to connect to UPF") // wait for PFCP Agent to initialize, blocking diff --git a/test/integration/verify_up4.go b/test/integration/verify_up4.go index 7dbb5afa6..2491ebf1c 100644 --- a/test/integration/verify_up4.go +++ b/test/integration/verify_up4.go @@ -8,6 +8,7 @@ import ( "math" "testing" + "github.com/antoninbas/p4runtime-go-client/pkg/client" p4rtc "github.com/antoninbas/p4runtime-go-client/pkg/client" "github.com/antoninbas/p4runtime-go-client/pkg/util/conversion" "github.com/omec-project/upf-epc/internal/p4constants" @@ -45,6 +46,11 @@ const ( maxApplicationID uint8 = 255 ) +const ( + p4RuntimeAddr = "127.0.0.1:50001" + errP4RuntimeConnect = "failed to connect to P4Runtime server" +) + var ( tablesNames = p4constants.GetTableIDToNameMap() actionNames = p4constants.GetActionIDToNameMap() @@ -290,8 +296,8 @@ func buildExpectedSliceTcMeter(expectedValues sliceMeter) (*p4_v1.MeterEntry, er // TODO: we should pass a list of pfcpSessionData if we will test multiple UEs func verifyP4RuntimeEntries(t *testing.T, testdata *pfcpSessionData, expectedValues p4RtValues, ueState UEState) { - p4rtClient, err := providers.ConnectP4rt("127.0.0.1:50001", false) - require.NoErrorf(t, err, "failed to connect to P4Runtime server") + p4rtClient, err := providers.ConnectP4rt(p4RuntimeAddr, false) + require.NoErrorf(t, err, errP4RuntimeConnect) defer providers.DisconnectP4rt() var ( @@ -305,8 +311,8 @@ func verifyP4RuntimeEntries(t *testing.T, testdata *pfcpSessionData, expectedVal } // FIXME: uncomment once pfcpiface properly removes all the state, see SDFAB-960 - //allInstalledEntries, _ := p4rtClient.ReadTableEntryWildcard("") - //require.Equal(t, expectedNumberOfAllEntries, len(allInstalledEntries), + // allInstalledEntries, _ := p4rtClient.ReadTableEntryWildcard("") + // require.Equal(t, expectedNumberOfAllEntries, len(allInstalledEntries), // fmt.Sprintf("UP4 should have exactly %v p4RtEntries installed", expectedNumberOfAllEntries), // allInstalledEntries) @@ -381,6 +387,28 @@ func verifyP4RuntimeEntries(t *testing.T, testdata *pfcpSessionData, expectedVal require.Equal(t, applicationID, entries[0].Match[1].GetExact().Value[0]) // check UE address require.Equal(t, expected.Match[0], entries[0].Match[0], "PreQosPipe.terminations_downlink match fields do not equal expected") + verifyPostDLAndMeters( + t, + p4rtClient, + testdata, + expectedValues, + ueState, + expected, + entries, + tunnelPeerSessionsDownlink, + ) +} + +func verifyPostDLAndMeters( + t *testing.T, + p4rtClient *client.Client, + testdata *pfcpSessionData, + expectedValues p4RtValues, + ueState UEState, + expected *p4_v1.TableEntry, + entries []*p4_v1.TableEntry, + tunnelPeerSessionsDownlink []byte, +) { if ueState == UEStateAttached && !testdata.dlGateClosed { // ignore counter ID as it is random number generated by pfcpiface require.Equal(t, expected.Action.GetAction().Params[1], entries[0].Action.GetAction().Params[1], @@ -456,8 +484,8 @@ func verifyP4RuntimeEntries(t *testing.T, testdata *pfcpSessionData, expectedVal } func verifyNumberOfEntries(t *testing.T, tableID uint32, expectedNoOfEntries int) { - p4rtClient, err := providers.ConnectP4rt("127.0.0.1:50001", false) - require.NoErrorf(t, err, "failed to connect to P4Runtime server") + p4rtClient, err := providers.ConnectP4rt(p4RuntimeAddr, false) + require.NoErrorf(t, err, errP4RuntimeConnect) defer providers.DisconnectP4rt() entries, err := p4rtClient.ReadTableEntryWildcard(p4constants.GetTableIDToNameMap()[tableID]) @@ -467,8 +495,8 @@ func verifyNumberOfEntries(t *testing.T, tableID uint32, expectedNoOfEntries int } func verifyNoP4RuntimeEntries(t *testing.T) { - p4rtClient, err := providers.ConnectP4rt("127.0.0.1:50001", false) - require.NoErrorf(t, err, "failed to connect to P4Runtime server") + p4rtClient, err := providers.ConnectP4rt(p4RuntimeAddr, false) + require.NoErrorf(t, err, errP4RuntimeConnect) defer providers.DisconnectP4rt() meters, _ := p4rtClient.ReadMeterEntryWildcard(MeterSession) @@ -514,8 +542,8 @@ func verifyNoP4RuntimeEntries(t *testing.T) { } func verifyP4RuntimeSliceMeter(t *testing.T, expectedValues p4RtValues) { - p4rtClient, err := providers.ConnectP4rt("127.0.0.1:50001", false) - require.NoErrorf(t, err, "failed to connect to P4Runtime server") + p4rtClient, err := providers.ConnectP4rt(p4RuntimeAddr, false) + require.NoErrorf(t, err, errP4RuntimeConnect) defer providers.DisconnectP4rt() sliceTcMeter := p4constants.GetMeterIDToNameMap()[p4constants.MeterPreQosPipeSliceTcMeter] From c2be4d36ecfd7e9f9e9ba06f3fa28ca6319cfa03 Mon Sep 17 00:00:00 2001 From: ReshmaJS Date: Thu, 21 May 2026 10:40:48 +0530 Subject: [PATCH 2/3] py files fixed --- conf/parser.py | 6 +- conf/ports.py | 147 +++++++++++++++++++++++-------------------------- 2 files changed, 71 insertions(+), 82 deletions(-) diff --git a/conf/parser.py b/conf/parser.py index 332670500..042048004 100644 --- a/conf/parser.py +++ b/conf/parser.py @@ -179,10 +179,7 @@ def parse(self, ifaces): "/tmp/notifycp" ) ) - self._parse_post_unix_config() - def _parse_post_unix_config(self): - """Parses Unix socket paths, feature flags, table sizes, and monitoring config.""" # UnixPort Paths try: self.endmarker_sockaddr = self.conf["endmarker_sockaddr"] @@ -220,5 +217,4 @@ def _parse_post_unix_config(self): try: self.enable_gtpu_path_monitoring = bool(self.conf["enable_gtpu_path_monitoring"]) except KeyError: - print("GTPu Path monitoring disabled") - + print("GTPu Path monitoring disabled") \ No newline at end of file diff --git a/conf/ports.py b/conf/ports.py index e790ba51f..d56e4adbb 100644 --- a/conf/ports.py +++ b/conf/ports.py @@ -150,9 +150,76 @@ def init_port(self, idx, conf_mode): self.num_q = num_q print("Setting up port {} on worker ids {}".format(name, self.workers)) - conf_mode = self._setup_initial_mode(idx, name, num_q, conf_mode) + # Detect the mode of this interface - DPDK/AF_XDP/AF_PACKET + if conf_mode is None: + conf_mode = self.detect_mode() + + if conf_mode not in ["af_xdp", "linux", "dpdk", "af_packet", "sim", "cndp"]: + raise Exception("Invalid mode: {} selected.".format(conf_mode)) + + if conf_mode in ["af_xdp", "linux"]: + try: + # Initialize kernel datapath. + # AF_XDP requires that num_rx_qs == num_tx_qs + kwargs = { + "vdev": "net_af_xdp{},iface={},start_queue=0,queue_count={}".format( + idx, name, num_q + ), + "num_out_q": num_q, + "num_inc_q": num_q, + } + self.init_datapath(**kwargs) + except: + if conf_mode == "linux": + print( + "Failed to create AF_XDP socket for {}. Retrying with AF_PACKET socket...".format( + name + ) + ) + conf_mode = "af_packet" + else: + print( + "Failed to create AF_XDP socket for {}. Exiting...".format(name) + ) + sys.exit() - self._setup_kernel_modes(idx, name, num_q, conf_mode) + if conf_mode == "cndp": + try: + # Initialize kernel fastpath. + self.init_datapath(cndp=True) + except: + print( + "Failed to create CNDP/AF_XDP socket for {}. Exiting...".format( + name + ) + ) + sys.exit() + + if conf_mode == "af_packet": + try: + # Initialize kernel datapath + kwargs = { + "vdev": "net_af_packet{},iface={},qpairs={}".format( + idx, name, num_q + ), + "num_out_q": num_q, + "num_inc_q": num_q, + } + self.init_datapath(**kwargs) + except: + print( + "Failed to create AF_PACKET socket for {}. Exiting...".format(name) + ) + sys.exit() + + if conf_mode == "sim": + self.fpi = Source(name="{}_source".format(name)) + self.fpo = Sink(name="{}_out".format(name)) + self.bpf = BPF(name="{}FastBPF".format(name)) + self.bpf.clear() + + # Attach datapath to worker's root TC + self.fpi.attach_task(wid=0) if conf_mode == "dpdk": kwargs = None @@ -245,80 +312,6 @@ def init_port(self, idx, conf_mode): # Finall set conf mode self.mode = conf_mode - def _setup_initial_mode(self, idx, name, num_q, conf_mode): - """Detects mode and initializes AF_XDP/Linux datapath.""" - # Detect the mode of this interface - DPDK/AF_XDP/AF_PACKET - if conf_mode is None: - conf_mode = self.detect_mode() - - if conf_mode not in ["af_xdp", "linux", "dpdk", "af_packet", "sim", "cndp"]: - raise Exception("Invalid mode: {} selected.".format(conf_mode)) - - if conf_mode in ["af_xdp", "linux"]: - try: - # Initialize kernel datapath. - # AF_XDP requires that num_rx_qs == num_tx_qs - kwargs = { - "vdev": "net_af_xdp{},iface={},start_queue=0,queue_count={}".format( - idx, name, num_q - ), - "num_out_q": num_q, - "num_inc_q": num_q, - } - self.init_datapath(**kwargs) - except: - if conf_mode == "linux": - print( - "Failed to create AF_XDP socket for {}. Retrying with AF_PACKET socket...".format( - name - ) - ) - conf_mode = "af_packet" - else: - print( - "Failed to create AF_XDP socket for {}. Exiting...".format(name) - ) - sys.exit() - return conf_mode - def _setup_kernel_modes(self, idx, name, num_q, conf_mode): - """Handles CNDP, AF_PACKET and SIM modes.""" - if conf_mode == "cndp": - try: - # Initialize kernel fastpath. - self.init_datapath(cndp=True) - except: - print( - "Failed to create CNDP/AF_XDP socket for {}. Exiting...".format( - name - ) - ) - sys.exit() - - if conf_mode == "af_packet": - try: - # Initialize kernel datapath - kwargs = { - "vdev": "net_af_packet{},iface={},qpairs={}".format( - idx, name, num_q - ), - "num_out_q": num_q, - "num_inc_q": num_q, - } - self.init_datapath(**kwargs) - except: - print( - "Failed to create AF_PACKET socket for {}. Exiting...".format(name) - ) - sys.exit() - - if conf_mode == "sim": - self.fpi = Source(name="{}_source".format(name)) - self.fpo = Sink(name="{}_out".format(name)) - self.bpf = BPF(name="{}FastBPF".format(name)) - self.bpf.clear() - - # Attach datapath to worker's root TC - self.fpi.attach_task(wid=0) def setup_port( self, @@ -416,4 +409,4 @@ def setup_port( merge.connect(update) if self.mode == "sim": - self.rtr = merge + self.rtr = merge \ No newline at end of file From b9221d3463f7d3546b211cd501d58e43a05e1df5 Mon Sep 17 00:00:00 2001 From: ReshmaJS Date: Thu, 21 May 2026 14:13:14 +0530 Subject: [PATCH 3/3] parser file updated --- conf/parser.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/conf/parser.py b/conf/parser.py index 042048004..332670500 100644 --- a/conf/parser.py +++ b/conf/parser.py @@ -179,7 +179,10 @@ def parse(self, ifaces): "/tmp/notifycp" ) ) + self._parse_post_unix_config() + def _parse_post_unix_config(self): + """Parses Unix socket paths, feature flags, table sizes, and monitoring config.""" # UnixPort Paths try: self.endmarker_sockaddr = self.conf["endmarker_sockaddr"] @@ -217,4 +220,5 @@ def parse(self, ifaces): try: self.enable_gtpu_path_monitoring = bool(self.conf["enable_gtpu_path_monitoring"]) except KeyError: - print("GTPu Path monitoring disabled") \ No newline at end of file + print("GTPu Path monitoring disabled") +