feat(pipelined): QFI support in Pipelined and Sessiond

lte/gateway/c/session_manager/SessionStateEnforcer.cpp

@@ -875,7 +875,6 @@ bool SessionStateEnforcer::insert_pdr_from_core(
       ->set_gnb_ipv4_adr(ip_addr);
   rule.mutable_activate_flow_req()->set_downlink_tunnel(teid);
   rule.mutable_deactivate_flow_req()->set_downlink_tunnel(teid);
-
   MLOG(MINFO) << " AMF teid: " << teid << " ip address " << ip_addr
               << " of imsi: " << session->get_imsi()
               << " fteid: " << session->get_upf_local_teid()
@@ -894,6 +893,7 @@ uint32_t SessionStateEnforcer::insert_pdr_from_access(
   rule.mutable_pdi()->set_local_f_teid(upf_teid);
   rule.mutable_activate_flow_req()->set_uplink_tunnel(upf_teid);
   rule.mutable_deactivate_flow_req()->set_uplink_tunnel(upf_teid);
+
   // Insert the PDR along with teid into the session
   session->insert_pdr(&rule, session_uc);
   return upf_teid;
@@ -934,6 +934,13 @@ void SessionStateEnforcer::set_pdr_attributes(
       config.common_context.ue_ipv4());
   rule->mutable_deactivate_flow_req()->set_ipv6_addr(
       config.common_context.ue_ipv6());
+  /*Set the AMBR QCI Value */
+  if (config.rat_specific_context.m5gsm_session_context()
+          .has_subscribed_qos()) {
+    const auto& subscribed_qos =
+        config.rat_specific_context.m5gsm_session_context().subscribed_qos();
+    rule->set_session_qfi(subscribed_qos.qos_class_id());
+  }
 }
 
 std::vector<StaticRuleInstall> SessionStateEnforcer::to_vec(

lte/gateway/python/magma/pipelined/app/classifier.py

@@ -26,7 +26,7 @@
     UESessionSet,
     UESessionState,
 )
-from lte.protos.session_manager_pb2 import UPFPagingInfo
+from lte.protos.session_manager_pb2 import QCI, UPFPagingInfo
 from lte.protos.subscriberdb_pb2 import SubscriberID
 from magma.common.rpc_utils import indicates_connection_error
 from magma.common.sentry import EXCLUDE_FROM_ERROR_MONITORING
@@ -273,10 +273,14 @@ def _paging_msg_sent_callback(self, future):
     def _install_uplink_tunnel_flows(
         self, priority: int, i_teid: int,
         gtp_portno: int, sid: Optional[int], o_teid: int,
+        ng_flag: bool, session_qfi: QCI,
     ):
 
         parser = self._datapath.ofproto_parser
-        match = MagmaMatch(tunnel_id=i_teid, in_port=gtp_portno)
+        if ng_flag and session_qfi:
+            match = MagmaMatch(tunnel_id=i_teid, qfi=session_qfi, in_port=gtp_portno)
+        else:
+            match = MagmaMatch(tunnel_id=i_teid, in_port=gtp_portno)
         actions = [
             parser.OFPActionSetField(eth_src=GTP_PORT_MAC),
             parser.OFPActionSetField(eth_dst="ff:ff:ff:ff:ff:ff"),
@@ -311,7 +315,8 @@ def _install_downlink_tunnel_flows(
         self, priority: int, i_teid: int,
         o_teid: int, in_port: int,
         ue_ip_adr: IPAddress, enodeb_ip_addr: str,
-        gtp_portno: int, sid: Optional[int], ng_flag: bool, ue_ipv6_adr: IPAddress = None,
+        gtp_portno: int, sid: Optional[int], ng_flag: bool,
+        ue_ipv6_adr: IPAddress = None, session_qfi: QCI = 0,
     ):
 
         parser = self._datapath.ofproto_parser
@@ -322,6 +327,8 @@ def _install_downlink_tunnel_flows(
         ]
         if ng_flag:
             actions.append(parser.OFPActionSetField(tun_flags=TUNNEL_OAM_FLAG))
+            if session_qfi:
+                actions.append(parser.OFPActionSetField(qfi=session_qfi))
         if i_teid:
             actions.append(parser.NXActionRegLoad2(dst=INGRESS_TUN_ID_REG, value=i_teid))
         if sid:
@@ -361,7 +368,7 @@ def add_tunnel_flows(
         ng_flag: bool = True,
         ue_ipv6_address: Optional[IPAddress] = None,
         unused_apn: Optional[str] = None, unused_vlan: int = 0,
-        ip_flow_dl: Optional[IPFlowDL] = None,
+        ip_flow_dl: Optional[IPFlowDL] = None, session_qfi: QCI = 0,
     ) -> bool:
 
         priority = Utils.get_of_priority(precedence)
@@ -373,7 +380,10 @@ def add_tunnel_flows(
 
         # Add flow for gtp port for Uplink Tunnel
         if i_teid:
-            self._install_uplink_tunnel_flows(priority, i_teid, gtp_portno, sid, o_teid)
+            self._install_uplink_tunnel_flows(
+                priority, i_teid, gtp_portno, sid,
+                o_teid, ng_flag, session_qfi,
+            )
 
         if ip_flow_dl and ip_flow_dl.set_params:
             self._add_tunnel_ip_flow_dl(
@@ -387,15 +397,15 @@ def add_tunnel_flows(
                     priority, i_teid, o_teid,
                     self._uplink_port, ue_ip_adr,
                     enodeb_ip_addr, gtp_portno,
-                    sid, ng_flag, ue_ipv6_address,
+                    sid, ng_flag, ue_ipv6_address, session_qfi,
                 )
 
                 # Add  Downlink Tunnel flow for mtr port
                 self._install_downlink_tunnel_flows(
                     priority, i_teid, o_teid,
                     self.config.mtr_port,
                     ue_ip_adr, enodeb_ip_addr,
-                    gtp_portno, sid, ng_flag, ue_ipv6_address,
+                    gtp_portno, sid, ng_flag, ue_ipv6_address, session_qfi,
                 )
 
         # Add ARP flow for LOCAL port
@@ -413,9 +423,15 @@ def add_tunnel_flows(
 
         return True
 
-    def _delete_uplink_tunnel_flows(self, i_teid: int, gtp_portno: int):
+    def _delete_uplink_tunnel_flows(
+        self, i_teid: int, gtp_portno: int,
+        ng_flag: bool = True, session_qfi: QCI = 0,
+    ):
 
-        match = MagmaMatch(tunnel_id=i_teid, in_port=gtp_portno)
+        if ng_flag and session_qfi:
+            match = MagmaMatch(tunnel_id=i_teid, qfi=session_qfi, in_port=gtp_portno)
+        else:
+            match = MagmaMatch(tunnel_id=i_teid, in_port=gtp_portno)
 
         flows.delete_flow(self._datapath, self.tbl_num, match)
 
@@ -485,6 +501,7 @@ def delete_tunnel_flows(
         enodeb_ip_addr: Union[None, str, IPAddress] = None,
         ip_flow_dl: Optional[IPFlowDL] = None,
         ue_ipv6_adr: Optional[IPAddress] = None,
+        ng_flag: bool = True, session_qfi: QCI = 0,
     ) -> bool:
 
         # Delete flow for gtp port
@@ -494,7 +511,7 @@ def delete_tunnel_flows(
             gtp_portno = self.config.gtp_port
 
         if i_teid:
-            self._delete_uplink_tunnel_flows(i_teid, gtp_portno)
+            self._delete_uplink_tunnel_flows(i_teid, gtp_portno, ng_flag, session_qfi)
 
         if ip_flow_dl and ip_flow_dl.set_params:
             self._delete_tunnel_ip_flow_dl(ip_flow_dl)
@@ -779,6 +796,7 @@ def gtp_handler(
         sid: int = None, ng_flag: bool = True,
         ue_ipv6_address: str = None, apn: str = None,
         vlan: int = 0, ip_flow_dl: IPFlowDL = None,
+        session_qfi: QCI = 0,
     ):
         ue_ip_adr = None
         ue_ipv6_adr = None
@@ -798,14 +816,17 @@ def gtp_handler(
                 precedence, local_f_teid,
                 o_teid, gnb_ip_addr, ue_ip_adr,
                 sid, ng_flag, ue_ipv6_adr,
-                apn, vlan, ip_flow_dl,
+                apn, vlan, ip_flow_dl, session_qfi,
             )
 
         elif (
             session_state == PdrState.Value('IDLE')
             or session_state == UESessionState.INSTALL_IDLE
         ):
-            self.delete_tunnel_flows(local_f_teid, ue_ip_adr, gnb_ip_addr, ip_flow_dl, ue_ipv6_adr)
+            self.delete_tunnel_flows(
+                local_f_teid, ue_ip_adr, gnb_ip_addr,
+                ip_flow_dl, ue_ipv6_adr, ng_flag, session_qfi,
+            )
             self.install_paging_flow(local_f_teid, ue_ip_adr, ng_flag, ue_ipv6_adr)
 
         elif (
@@ -815,6 +836,7 @@ def gtp_handler(
             self.delete_tunnel_flows(
                 local_f_teid, ue_ip_adr,
                 gnb_ip_addr, ip_flow_dl, ue_ipv6_adr,
+                ng_flag, session_qfi,
             )
             self.remove_paging_flow(ue_ip_adr, ue_ipv6_adr)
 

lte/gateway/python/magma/pipelined/ng_manager/session_state_manager_util.py

@@ -35,6 +35,8 @@ class PDRRuleEntry(NamedTuple):
     add_qos_enforce_rule: ActivateFlowsRequest
     far_action: Optional[FARRuleEntry]
     ue_ipv6_addr: Optional[str]
+    session_qfi: Optional[int]
+
 
 # Create the Named tuple for the FAR entry
 
@@ -64,6 +66,7 @@ def pdr_create_rule_entry(pdr_entry) -> PDRRuleEntry:
     deactivate_flow_req = None
     activate_flow_req = None
     ue_ipv6_addr = None
+    session_qfi = 0
 
     # get local teid
     if pdr_entry.pdi.local_f_teid:
@@ -86,11 +89,14 @@ def pdr_create_rule_entry(pdr_entry) -> PDRRuleEntry:
     if pdr_entry.HasField('activate_flow_req') == True:
         activate_flow_req = pdr_entry.activate_flow_req
 
+    if pdr_entry.session_qfi:
+        session_qfi = pdr_entry.session_qfi
+
     pdr_rule = PDRRuleEntry(
         pdr_entry.pdr_id, pdr_entry.pdr_version,
         pdr_entry.pdr_state, pdr_entry.precedence,
         local_f_teid, ue_ip_addr,
         deactivate_flow_req, activate_flow_req,
-        far_entry, ue_ipv6_addr,
+        far_entry, ue_ipv6_addr, session_qfi,
     )
     return pdr_rule

lte/gateway/python/magma/pipelined/rpc_servicer.py

@@ -993,6 +993,7 @@ def _ng_tunnel_update(self, pdr_entry: PDRRuleEntry, subscriber_id: str) -> bool
             encode_imsi(subscriber_id),
             True,
             pdr_entry.ue_ipv6_addr,
+            session_qfi=pdr_entry.session_qfi,
         )
 
         return ret

lte/gateway/python/magma/pipelined/tests/BUILD.bazel

@@ -876,3 +876,22 @@ pytest_test(
         "//lte/protos:pipelined_python_proto",
     ],
 )
+
+pytest_test(
+    name = "test_qfi_gtp",
+    size = "small",
+    srcs = ["test_qfi_gtp.py"],
+    imports = [
+        LTE_ROOT,
+        ORC8R_ROOT,
+    ],
+    tags = TAG_SUDO_TEST,
+    deps = [
+        ":pipelined_test_util",
+        "//lte/gateway/python/magma/pipelined:bridge_util",
+        "//lte/gateway/python/magma/pipelined/app:testing",
+        "//lte/gateway/python/magma/pipelined/tests/app:start_pipelined",
+        "//lte/protos:mobilityd_python_proto",
+        "//lte/protos:pipelined_python_proto",
+    ],
+)

lte/gateway/python/magma/pipelined/tests/snapshots/test_qfi_gtp.QfigtpTest.test_qfi_delete_tunnel_flows.snapshot

@@ -0,0 +1,3 @@
+ cookie=0x0, table=classifier(main_table), n_packets=3, n_bytes=250, priority=0 actions=resubmit(,ingress(main_table))
+ cookie=0x0, table=classifier(main_table), n_packets=0, n_bytes=0, priority=0,in_port=15578 actions=resubmit(,201)
+ cookie=0x0, table=classifier(main_table), n_packets=0, n_bytes=0, priority=0,in_port=15579 actions=resubmit(,202)

lte/gateway/python/magma/pipelined/tests/snapshots/test_qfi_gtp.QfigtpTest.test_qfi_set_tunnel_flows.snapshot

@@ -0,0 +1,5 @@
+ cookie=0x0, table=classifier(main_table), n_packets=0, n_bytes=0, priority=10,tun_id=0x1,qfi=9,in_port=32768 actions=mod_dl_src:02:00:00:00:00:01,mod_dl_dst:ff:ff:ff:ff:ff:ff,set_field:0x186a0->reg9,load:0x1388->OXM_OF_METADATA[],resubmit(,ingress(main_table))
+ cookie=0x0, table=classifier(main_table), n_packets=0, n_bytes=0, priority=10,ip,in_port=LOCAL,nw_dst=192.168.128.30 actions=load:0x186a0->NXM_NX_TUN_ID[],load:0xc0a83cb2->NXM_NX_TUN_IPV4_DST[],set_field:0x8000->reg8,load:0x1->NXM_NX_TUN_FLAGS[],load:0x9->NXM_NX_QFI[],set_field:0x1->reg9,load:0x1388->OXM_OF_METADATA[],resubmit(,ingress(main_table))
+ cookie=0x0, table=classifier(main_table), n_packets=0, n_bytes=0, priority=10,ip,in_port=15577,nw_dst=192.168.128.30 actions=load:0x186a0->NXM_NX_TUN_ID[],load:0xc0a83cb2->NXM_NX_TUN_IPV4_DST[],set_field:0x8000->reg8,load:0x1->NXM_NX_TUN_FLAGS[],load:0x9->NXM_NX_QFI[],set_field:0x1->reg9,load:0x1388->OXM_OF_METADATA[],resubmit(,ingress(main_table))
+ cookie=0x0, table=classifier(main_table), n_packets=0, n_bytes=0, priority=10,arp,in_port=LOCAL,arp_tpa=192.168.128.30 actions=load:0x1388->OXM_OF_METADATA[],resubmit(,ingress(main_table))
+ cookie=0x0, table=classifier(main_table), n_packets=0, n_bytes=0, priority=10,arp,in_port=15577,arp_tpa=192.168.128.30 actions=load:0x1388->OXM_OF_METADATA[],resubmit(,ingress(main_table))