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/bess.go b/pfcpiface/bess.go index f482a323d..6539707f1 100644 --- a/pfcpiface/bess.go +++ b/pfcpiface/bess.go @@ -67,6 +67,15 @@ const ( sliceMeterGateUnmeter uint64 = 6 ) +const ( + // SOnarQube define constants. + errGRPCCallFailed = "unable to make GRPC calls" + errMarshalRule = "error marshalling the rule" + errMarshalRequest = "error marshalling request" + errReadFailed = "read failed:" + errInvalidMethodName = "invalid method name:" +) + var intEnc = func(u uint64) *pb.FieldData { return &pb.FieldData{Encoding: &pb.FieldData_ValueInt{ValueInt: u}} } @@ -117,14 +126,15 @@ func (b *bess) AddSliceInfo(sliceInfo *SliceInfo) error { rc := b.GRPCJoin(1, Timeout, done) if !rc { - logger.BessLog.Errorln("unable to make GRPC calls") + logger.BessLog.Errorln(errGRPCCallFailed) } return nil } func (b *bess) SendMsgToUPF( - method upfMsgType, rules PacketForwardingRules, updated PacketForwardingRules) uint8 { + method upfMsgType, rules PacketForwardingRules, updated PacketForwardingRules, +) uint8 { // create context cause := ie.CauseRequestAccepted @@ -189,7 +199,7 @@ func (b *bess) SendMsgToUPF( rc := b.GRPCJoin(calls, Timeout, done) if !rc { - logger.BessLog.Errorln("unable to make GRPC calls") + logger.BessLog.Errorln(errGRPCCallFailed) } return cause @@ -207,7 +217,7 @@ func (b *bess) measureUpf(ifName string, f *pb.MeasureCommandGetSummaryArg) *pb. arg, err := anypb.New(f) if err != nil { - logger.BessLog.Errorln("error marshalling the rule", f, err) + logger.BessLog.Errorln(errMarshalRule, f, err) return nil } @@ -391,7 +401,7 @@ func (b *bess) flipFlowMeasurementBufferFlag(ctx context.Context, module string) arg, err := anypb.New(req) if err != nil { - logger.BessLog.Errorln("error marshalling request", req, err) + logger.BessLog.Errorln(errMarshalRequest, req, err) return } @@ -404,7 +414,7 @@ func (b *bess) flipFlowMeasurementBufferFlag(ctx context.Context, module string) ) if err != nil { - logger.BessLog.Errorln(module, "read failed:", err) + logger.BessLog.Errorln(module, errReadFailed, err) return } @@ -433,7 +443,7 @@ func (b *bess) readFlowMeasurement( arg, err := anypb.New(req) if err != nil { - logger.BessLog.Errorln("error marshalling request", req, err) + logger.BessLog.Errorln(errMarshalRequest, req, err) return } @@ -444,9 +454,8 @@ func (b *bess) readFlowMeasurement( Arg: arg, }, ) - if err != nil { - logger.BessLog.Errorln(module, "read failed:", err) + logger.BessLog.Errorln(module, errReadFailed, err) return } @@ -464,14 +473,15 @@ func (b *bess) readFlowMeasurement( } func (b *bess) readGtpuPathMonitoringStats( - module string, isClear bool) *pb.GtpuPathMonitoringCommandReadResponse { + module string, isClear bool, +) *pb.GtpuPathMonitoringCommandReadResponse { req := &pb.GtpuPathMonitoringCommandReadArg{ Clear: isClear, } arg, err := anypb.New(req) if err != nil { - logger.BessLog.Errorln("error marshalling request", req, err) + logger.BessLog.Errorln(errMarshalRequest, req, err) return nil } @@ -484,9 +494,8 @@ func (b *bess) readGtpuPathMonitoringStats( Arg: arg, }, ) - if err != nil { - logger.BessLog.Errorln(module, "read failed:", err) + logger.BessLog.Errorln(module, errReadFailed, err) return nil } @@ -497,7 +506,6 @@ func (b *bess) readGtpuPathMonitoringStats( var res pb.GtpuPathMonitoringCommandReadResponse err = resp.Data.UnmarshalTo(&res) - if err != nil { logger.BessLog.Errorln(err, resp) return nil @@ -512,33 +520,11 @@ func (b *bess) SessionStats(pc *PfcpNodeCollector, ch chan<- prometheus.Metric) ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) defer cancel() // Flips the buffer flag, automatically waits for in-flight packets to drain. - flip, err := b.flipFlowMeasurementBufferFlag(ctx, PreQosFlowMeasure) + qosStatsInResp, postDlQosStatsResp, postUlQosStatsResp, err := b.readSessionMeasurements(ctx) if err != nil { - logger.BessLog.Errorln(PreQosFlowMeasure, "read failed:", err) - return + return err } - q := []float64{50, 90, 99} - - // Read stats from the now inactive side, and clear if needed. - qosStatsInResp, err := b.readFlowMeasurement(ctx, PreQosFlowMeasure, flip.OldFlag, true, q) - if err != nil { - logger.BessLog.Errorln(PreQosFlowMeasure, "read failed:", err) - return - } - - postDlQosStatsResp, err := b.readFlowMeasurement(ctx, PostDlQosFlowMeasure, flip.OldFlag, true, q) - if err != nil { - logger.BessLog.Errorln(PostDlQosFlowMeasure, "read failed:", err) - return - } - - postUlQosStatsResp, err := b.readFlowMeasurement(ctx, PostUlQosFlowMeasure, flip.OldFlag, true, q) - if err != nil { - logger.BessLog.Errorln(PostUlQosFlowMeasure, "read failed:", err) - return - } - // TODO: pick first connection for now var con *PFCPConn @@ -579,22 +565,9 @@ func (b *bess) SessionStats(pc *PfcpNodeCollector, ch chan<- prometheus.Metric) pdrString := strconv.FormatUint(pre.Pdr, 10) ueIpString := "unknown" - if con != nil { - session, ok := con.store.GetSession(pre.Fseid) - if !ok { - logger.BessLog.Errorln("invalid or unknown FSEID", pre.Fseid) - continue - } - - // Try to find the N6 uplink PDR with the UE IP. - for _, p := range session.pdrs { - if p.IsUplink() && p.ueAddress > 0 { - ueIpString = int2ip(p.ueAddress).String() - logger.BessLog.Debugln(p.fseID, " -> ", ueIpString) - - break - } - } + ueIpString, ok := resolveUEIP(con, pre, ueIpString) + if !ok { + continue } ch <- prometheus.MustNewConstMetric( @@ -656,6 +629,67 @@ func (b *bess) SessionStats(pc *PfcpNodeCollector, ch chan<- prometheus.Metric) return } +func (b *bess) readSessionMeasurements( + ctx context.Context, +) (pre, postDl, postUl pb.FlowMeasureReadResponse, err error) { + // Flip buffer + flip, err := b.flipFlowMeasurementBufferFlag(ctx, PreQosFlowMeasure) + if err != nil { + logger.BessLog.Errorln(PreQosFlowMeasure, errReadFailed, err) + return + } + + q := []float64{50, 90, 99} + + // Read Pre-QoS + pre, err = b.readFlowMeasurement(ctx, PreQosFlowMeasure, flip.OldFlag, true, q) + if err != nil { + logger.BessLog.Errorln(PreQosFlowMeasure, errReadFailed, err) + return + } + + // Read Post DL + postDl, err = b.readFlowMeasurement(ctx, PostDlQosFlowMeasure, flip.OldFlag, true, q) + if err != nil { + logger.BessLog.Errorln(PostDlQosFlowMeasure, errReadFailed, err) + return + } + + // Read Post UL + postUl, err = b.readFlowMeasurement(ctx, PostUlQosFlowMeasure, flip.OldFlag, true, q) + if err != nil { + logger.BessLog.Errorln(PostUlQosFlowMeasure, errReadFailed, err) + return + } + + return +} + +func resolveUEIP( + con *PFCPConn, + pre *pb.FlowMeasureReadResponse_Statistic, + ueIpString string, +) (string, bool) { + if con != nil { + session, ok := con.store.GetSession(pre.Fseid) + if !ok { + logger.BessLog.Errorln("invalid or unknown FSEID", pre.Fseid) + return ueIpString, false + } + + // Try to find the N6 uplink PDR with the UE IP. + for _, p := range session.pdrs { + if p.IsUplink() && p.ueAddress > 0 { + ueIpString = int2ip(p.ueAddress).String() + logger.BessLog.Debugln(p.fseID, " -> ", ueIpString) + break + } + } + } + + return ueIpString, true +} + func (b *bess) endMarkerSendLoop(endMarkerChan chan []byte) { for outPacket := range endMarkerChan { _, err := b.endMarkerSocket.Write(outPacket) @@ -721,7 +755,7 @@ func (b *bess) clearState() { anyWildcardClear, err := anypb.New(clearWildcardCmd) if err != nil { - logger.BessLog.Errorf("error marshalling the rule %v: %v", clearWildcardCmd, err) + logger.BessLog.Errorf("%v %v: %v", errMarshalRule, clearWildcardCmd, err) return } @@ -731,7 +765,7 @@ func (b *bess) clearState() { anyExactClear, err := anypb.New(clearExactCmd) if err != nil { - logger.BessLog.Errorf("error marshalling the rule %v: %v", anyExactClear, err) + logger.BessLog.Errorf("%v %v: %v", errMarshalRule, anyExactClear, err) return } @@ -743,7 +777,7 @@ func (b *bess) clearState() { var anyGtpuPathMonitoringClear *anypb.Any anyGtpuPathMonitoringClear, err = anypb.New(clearGtpuPathMonitoringCmd) if err != nil { - logger.BessLog.Errorf("error marshalling the rule %v: %v", anyGtpuPathMonitoringClear, err) + logger.BessLog.Errorf("%v %v: %v", errMarshalRule, anyGtpuPathMonitoringClear, err) return } @@ -754,7 +788,7 @@ func (b *bess) clearState() { var anyQoSClear *anypb.Any anyQoSClear, err = anypb.New(clearQoSCmd) if err != nil { - logger.BessLog.Errorf("error marshalling the rule %v: %v", anyQoSClear, err) + logger.BessLog.Errorf("%v %v: %v", errMarshalRule, anyQoSClear, err) return } @@ -821,6 +855,14 @@ func (b *bess) SetUpfInfo(u *upf, conf *Conf) { go b.endMarkerSendLoop(b.endMarkerChan) } + b.setupSliceMeter(conf) + + if conf.EnableGtpuPathMonitoring { + enableGtpuPathMonitoring = true + } +} + +func (b *bess) setupSliceMeter(conf *Conf) { if (conf.SliceMeterConfig.N6RateBps > 0) || (conf.SliceMeterConfig.N3RateBps > 0) { ctx, cancel := context.WithTimeout(context.Background(), Timeout) @@ -832,18 +874,14 @@ func (b *bess) SetUpfInfo(u *upf, conf *Conf) { rc := b.GRPCJoin(1, Timeout, done) if !rc { - logger.BessLog.Errorln("unable to make GRPC calls") + logger.BessLog.Errorln(errGRPCCallFailed) } } - - if conf.EnableGtpuPathMonitoring { - enableGtpuPathMonitoring = true - } } func (b *bess) processPDR(ctx context.Context, arg *anypb.Any, method upfMsgType) { if method != upfMsgTypeAdd && method != upfMsgTypeDel && method != upfMsgTypeClear { - logger.BessLog.Infoln("invalid method name:", method) + logger.BessLog.Infoln(errInvalidMethodName, method) return } @@ -920,7 +958,7 @@ func (b *bess) addPDR(ctx context.Context, done chan<- bool, p pdr) { arg, err = anypb.New(f) if err != nil { - logger.BessLog.Infoln("error marshalling the rule", f, err) + logger.BessLog.Infoln(errMarshalRule, f, err) return } @@ -970,7 +1008,7 @@ func (b *bess) delPDR(ctx context.Context, done chan<- bool, p pdr) { arg, err = anypb.New(f) if err != nil { - logger.BessLog.Errorln("error marshalling the rule", f, err) + logger.BessLog.Errorln(errMarshalRule, f, err) return } @@ -982,86 +1020,94 @@ func (b *bess) delPDR(ctx context.Context, done chan<- bool, p pdr) { func (b *bess) addQER(ctx context.Context, done chan<- bool, qer qer) { go func() { - var ( - cir, pir, cbs, ebs, pbs, gate uint64 - srcIface uint8 - ) + // Uplink + b.handleUplinkQER(ctx, qer) - // Uplink QER - srcIface = access + // Downlink + b.handleDownlinkQER(ctx, qer) - // Lookup QCI from QFI, else try default QCI. - qosVal, ok := b.qciQosMap[qer.qfi] - if !ok { - logger.BessLog.Debugf("number of config for qfi/qci: %v using default burst size", qer.qfi) + done <- true + }() +} - qosVal = b.qciQosMap[0] - } +func (b *bess) handleUplinkQER(ctx context.Context, qer qer) { + var ( + cir, pir, cbs, ebs, pbs, gate uint64 + srcIface uint8 + ) - cbs = maxUint64(calcBurstSizeFromRate(qer.ulGbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.cbs)) - ebs = maxUint64(calcBurstSizeFromRate(qer.ulMbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.ebs)) - pbs = maxUint64(calcBurstSizeFromRate(qer.ulMbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.ebs)) - - if qer.ulStatus != ie.GateStatusOpen { - gate = qerGateStatusDrop - } else if qer.ulMbr != 0 || qer.ulGbr != 0 { - /* MBR/GBR is received in Kilobits/sec. - CIR/PIR is sent in bytes */ - cir = maxUint64(((qer.ulGbr * 1000) / 8), 1) - pir = maxUint64(((qer.ulMbr * 1000) / 8), cir) - gate = qerGateMeter - } else { - gate = qerGateUnmeter - } + srcIface = access - switch qer.qosLevel { - case ApplicationQos: - b.addApplicationQER(ctx, gate, srcIface, cir, pir, cbs, pbs, ebs, qer) - case SessionQos: - b.addSessionQER(ctx, gate, srcIface, cir, pir, cbs, pbs, ebs, qer) - } + // Lookup QCI from QFI + qosVal, ok := b.qciQosMap[qer.qfi] + if !ok { + logger.BessLog.Debugf("number of config for qfi/qci: %v using default burst size", qer.qfi) + qosVal = b.qciQosMap[0] + } - // Downlink QER - srcIface = core + cbs = maxUint64(calcBurstSizeFromRate(qer.ulGbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.cbs)) + ebs = maxUint64(calcBurstSizeFromRate(qer.ulMbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.ebs)) + pbs = maxUint64(calcBurstSizeFromRate(qer.ulMbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.ebs)) - // Lookup QCI from QFI, else try default QCI. - qosVal, ok = b.qciQosMap[qer.qfi] - if !ok { - logger.BessLog.Debugf("number of config for qfi/qci: %v using default burst size", qer.qfi) + if qer.ulStatus != ie.GateStatusOpen { + gate = qerGateStatusDrop + } else if qer.ulMbr != 0 || qer.ulGbr != 0 { + cir = maxUint64(((qer.ulGbr * 1000) / 8), 1) + pir = maxUint64(((qer.ulMbr * 1000) / 8), cir) + gate = qerGateMeter + } else { + gate = qerGateUnmeter + } - qosVal = b.qciQosMap[0] - } + switch qer.qosLevel { + case ApplicationQos: + b.addApplicationQER(ctx, gate, srcIface, cir, pir, cbs, pbs, ebs, qer) + case SessionQos: + b.addSessionQER(ctx, gate, srcIface, cir, pir, cbs, pbs, ebs, qer) + } +} - cbs = maxUint64(calcBurstSizeFromRate(qer.dlGbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.cbs)) - ebs = maxUint64(calcBurstSizeFromRate(qer.dlMbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.ebs)) - pbs = maxUint64(calcBurstSizeFromRate(qer.dlMbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.ebs)) - - if qer.dlStatus != ie.GateStatusOpen { - gate = qerGateStatusDrop - } else if qer.dlMbr != 0 || qer.dlGbr != 0 { - /* MBR/GBR is received in Kilobits/sec. - CIR/PIR is sent in bytes */ - cir = maxUint64(((qer.dlGbr * 1000) / 8), 1) - pir = maxUint64(((qer.dlMbr * 1000) / 8), cir) - gate = qerGateMeter - } else { - gate = qerGateUnmeter - } +func (b *bess) handleDownlinkQER(ctx context.Context, qer qer) { + var ( + cir, pir, cbs, ebs, pbs, gate uint64 + srcIface uint8 + ) - switch qer.qosLevel { - case ApplicationQos: - b.addApplicationQER(ctx, gate, srcIface, cir, pir, cbs, pbs, ebs, qer) - case SessionQos: - b.addSessionQER(ctx, gate, srcIface, cir, pir, cbs, pbs, ebs, qer) - } + srcIface = core - done <- true - }() + // Lookup QCI from QFI + qosVal, ok := b.qciQosMap[qer.qfi] + if !ok { + logger.BessLog.Debugf("number of config for qfi/qci: %v using default burst size", qer.qfi) + qosVal = b.qciQosMap[0] + } + + cbs = maxUint64(calcBurstSizeFromRate(qer.dlGbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.cbs)) + ebs = maxUint64(calcBurstSizeFromRate(qer.dlMbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.ebs)) + pbs = maxUint64(calcBurstSizeFromRate(qer.dlMbr, uint64(qosVal.burstDurationMs)), uint64(qosVal.ebs)) + + if qer.dlStatus != ie.GateStatusOpen { + gate = qerGateStatusDrop + } else if qer.dlMbr != 0 || qer.dlGbr != 0 { + cir = maxUint64(((qer.dlGbr * 1000) / 8), 1) + pir = maxUint64(((qer.dlMbr * 1000) / 8), cir) + gate = qerGateMeter + } else { + gate = qerGateUnmeter + } + + switch qer.qosLevel { + case ApplicationQos: + b.addApplicationQER(ctx, gate, srcIface, cir, pir, cbs, pbs, ebs, qer) + case SessionQos: + b.addSessionQER(ctx, gate, srcIface, cir, pir, cbs, pbs, ebs, qer) + } } func (b *bess) addApplicationQER(ctx context.Context, gate uint64, srcIface uint8, cir uint64, pir uint64, cbs uint64, pbs uint64, - ebs uint64, qer qer) { + ebs uint64, qer qer, +) { var ( arg *anypb.Any err error @@ -1086,7 +1132,7 @@ func (b *bess) addApplicationQER(ctx context.Context, gate uint64, srcIface uint arg, err = anypb.New(q) if err != nil { - logger.BessLog.Errorln("error marshalling the rule", q, err) + logger.BessLog.Errorln(errMarshalRule, q, err) return } @@ -1127,7 +1173,8 @@ func (b *bess) delQER(ctx context.Context, done chan<- bool, qer qer) { } func (b *bess) delApplicationQER( - ctx context.Context, srcIface uint8, qer qer) { + ctx context.Context, srcIface uint8, qer qer, +) { var ( arg *anypb.Any err error @@ -1143,7 +1190,7 @@ func (b *bess) delApplicationQER( arg, err = anypb.New(q) if err != nil { - logger.BessLog.Infoln("error marshalling the rule", q, err) + logger.BessLog.Infoln(errMarshalRule, q, err) return } @@ -1157,7 +1204,7 @@ func (b *bess) delApplicationQER( func (b *bess) processFAR(ctx context.Context, arg *anypb.Any, method upfMsgType) { if method != upfMsgTypeAdd && method != upfMsgTypeDel && method != upfMsgTypeClear { - logger.BessLog.Errorln("invalid method name:", method) + logger.BessLog.Errorln(errInvalidMethodName, method) return } @@ -1178,7 +1225,7 @@ func (b *bess) processFAR(ctx context.Context, arg *anypb.Any, method upfMsgType func (b *bess) processGtpuPathMonitoring(ctx context.Context, arg *anypb.Any, method upfMsgType) { if method != upfMsgTypeAdd && method != upfMsgTypeDel && method != upfMsgTypeClear { - logger.BessLog.Errorln("invalid method name:", method) + logger.BessLog.Errorln(errInvalidMethodName, method) return } @@ -1243,7 +1290,7 @@ func (b *bess) addFAR(ctx context.Context, done chan<- bool, far far) { arg, err = anypb.New(f) if err != nil { - logger.BessLog.Infoln("error marshalling the rule", f, err) + logger.BessLog.Infoln(errMarshalRule, f, err) return } @@ -1283,7 +1330,7 @@ func (b *bess) delFAR(ctx context.Context, done chan<- bool, far far) { arg, err = anypb.New(f) if err != nil { - logger.BessLog.Infoln("error marshalling the rule", f, err) + logger.BessLog.Infoln(errMarshalRule, f, err) return } @@ -1309,7 +1356,7 @@ func (b *bess) delFAR(ctx context.Context, done chan<- bool, far far) { func (b *bess) processSliceMeter(ctx context.Context, arg *anypb.Any, method upfMsgType) { if method != upfMsgTypeAdd && method != upfMsgTypeDel && method != upfMsgTypeClear { - logger.BessLog.Errorln("invalid method name:", method) + logger.BessLog.Errorln(errInvalidMethodName, method) return } @@ -1335,93 +1382,118 @@ func (b *bess) addSliceMeter(ctx context.Context, done chan<- bool, meterConfig cir, pir, cbs, ebs, pbs, gate uint64 ) - // Uplink N6 slice meter config - if meterConfig.N6RateBps != 0 { - gate = sliceMeterGateMeter - cir = 1 // Mark all traffic as yellow - pir = meterConfig.N6RateBps / 8 // bit/s to byte/s - } else { - gate = sliceMeterGateUnmeter + // Uplink + arg, err = b.handleUplinkSliceMeter(meterConfig, &cir, &pir, &cbs, &ebs, &pbs, &gate) + if err != nil { + return } + b.processSliceMeter(ctx, arg, upfMsgTypeAdd) - if meterConfig.N6BurstBytes != 0 { - cbs = 1 // Mark all traffic as yellow - pbs = meterConfig.N6BurstBytes - ebs = 0 // Unused - } else { - cbs = 1 // Mark all traffic as yellow - pbs = DefaultBurstSize - ebs = 0 // Unused + // Downlink + arg, err = b.handleDownlinkSliceMeter(meterConfig, &cir, &pir, &cbs, &ebs, &pbs, &gate) + if err != nil { + return } + b.processSliceMeter(ctx, arg, upfMsgTypeAdd) - logger.BessLog.Debugln("uplink slice: cir:", cir, ", pir:", pir, ", cbs:", cbs, ", pbs:", pbs) + done <- true + }() +} - q := &pb.QosCommandAddArg{ - Gate: gate, - Cir: cir, /* committed info rate */ - Pir: pir, /* peak info rate */ - Cbs: cbs, /* committed burst size */ - Pbs: pbs, /* Peak burst size */ - Ebs: ebs, /* Excess burst size */ - OptionalDeductLen: &pb.QosCommandAddArg_DeductLen{DeductLen: 0}, /* Include all headers */ - Fields: []*pb.FieldData{ - intEnc(uint64(farForwardU)), /* Action */ - intEnc(uint64(0)), /* tunnel_out_type */ - }, - } +func (b *bess) handleUplinkSliceMeter( + meterConfig SliceMeterConfig, + cir, pir, cbs, ebs, pbs, gate *uint64, +) (*anypb.Any, error) { + // Uplink N6 slice meter config + if meterConfig.N6RateBps != 0 { + *gate = sliceMeterGateMeter + *cir = 1 + *pir = meterConfig.N6RateBps / 8 + } else { + *gate = sliceMeterGateUnmeter + } - arg, err = anypb.New(q) - if err != nil { - logger.BessLog.Errorln("error marshalling the rule", q, err) - return - } + if meterConfig.N6BurstBytes != 0 { + *cbs = 1 + *pbs = meterConfig.N6BurstBytes + *ebs = 0 + } else { + *cbs = 1 + *pbs = DefaultBurstSize + *ebs = 0 + } - b.processSliceMeter(ctx, arg, upfMsgTypeAdd) + logger.BessLog.Debugln("uplink slice: cir:", *cir, ", pir:", *pir, ", cbs:", *cbs, ", pbs:", *pbs) - // Downlink N3 slice meter config - if meterConfig.N3RateBps != 0 { - gate = sliceMeterGateMeter - cir = 1 // Mark all traffic as yellow - pir = meterConfig.N3RateBps / 8 // bit/s to byte/s - } else { - gate = sliceMeterGateUnmeter - } + q := &pb.QosCommandAddArg{ + Gate: *gate, + Cir: *cir, + Pir: *pir, + Cbs: *cbs, + Pbs: *pbs, + Ebs: *ebs, + OptionalDeductLen: &pb.QosCommandAddArg_DeductLen{DeductLen: 0}, + Fields: []*pb.FieldData{ + intEnc(uint64(farForwardU)), + intEnc(uint64(0)), + }, + } - if meterConfig.N3BurstBytes != 0 { - cbs = 1 // Mark all traffic as yellow - pbs = meterConfig.N3BurstBytes - ebs = 0 // Unused - } else { - cbs = 1 // Mark all traffic as yellow - pbs = DefaultBurstSize - ebs = 0 // Unused - } + arg, err := anypb.New(q) + if err != nil { + logger.BessLog.Errorln(errMarshalRule, q, err) + return nil, err + } - logger.BessLog.Debugln("downlink slice: cir:", cir, ", pir:", pir, ", cbs:", cbs, ", pbs:", pbs) - // TODO: packet deduction should take GTPU extension header into account - q = &pb.QosCommandAddArg{ - Gate: gate, - Cir: cir, /* committed info rate */ - Pir: pir, /* peak info rate */ - Cbs: cbs, /* committed burst size */ - Pbs: pbs, /* Peak burst size */ - Ebs: ebs, /* Excess burst size */ - OptionalDeductLen: &pb.QosCommandAddArg_DeductLen{DeductLen: 50}, /* Exclude Ethernet,IP,UDP,GTP header */ - Fields: []*pb.FieldData{ - intEnc(uint64(farForwardD)), /* Action */ - intEnc(uint64(1)), /* tunnel_out_type */ - }, - } + return arg, nil +} - arg, err = anypb.New(q) - if err != nil { - logger.BessLog.Errorln("error marshalling the rule", q, err) - return - } +func (b *bess) handleDownlinkSliceMeter( + meterConfig SliceMeterConfig, + cir, pir, cbs, ebs, pbs, gate *uint64, +) (*anypb.Any, error) { + // Downlink N3 slice meter config + if meterConfig.N3RateBps != 0 { + *gate = sliceMeterGateMeter + *cir = 1 + *pir = meterConfig.N3RateBps / 8 + } else { + *gate = sliceMeterGateUnmeter + } - b.processSliceMeter(ctx, arg, upfMsgTypeAdd) - done <- true - }() + if meterConfig.N3BurstBytes != 0 { + *cbs = 1 + *pbs = meterConfig.N3BurstBytes + *ebs = 0 + } else { + *cbs = 1 + *pbs = DefaultBurstSize + *ebs = 0 + } + + logger.BessLog.Debugln("downlink slice: cir:", *cir, ", pir:", *pir, ", cbs:", *cbs, ", pbs:", *pbs) + + q := &pb.QosCommandAddArg{ + Gate: *gate, + Cir: *cir, + Pir: *pir, + Cbs: *cbs, + Pbs: *pbs, + Ebs: *ebs, + OptionalDeductLen: &pb.QosCommandAddArg_DeductLen{DeductLen: 50}, + Fields: []*pb.FieldData{ + intEnc(uint64(farForwardD)), + intEnc(uint64(1)), + }, + } + + arg, err := anypb.New(q) + if err != nil { + logger.BessLog.Errorln(errMarshalRule, q, err) + return nil, err + } + + return arg, nil } func (b *bess) processQER(ctx context.Context, arg *anypb.Any, method upfMsgType, qosTableName string) error { @@ -1449,7 +1521,8 @@ func (b *bess) processQER(ctx context.Context, arg *anypb.Any, method upfMsgType func (b *bess) addSessionQER(ctx context.Context, gate uint64, srcIface uint8, cir uint64, pir uint64, cbs uint64, - pbs uint64, ebs uint64, qer qer) { + pbs uint64, ebs uint64, qer qer, +) { var ( arg *anypb.Any err error @@ -1470,7 +1543,7 @@ func (b *bess) addSessionQER(ctx context.Context, gate uint64, srcIface uint8, arg, err = anypb.New(q) if err != nil { - logger.BessLog.Errorln("error marshalling the rule", q, err) + logger.BessLog.Errorln(errMarshalRule, q, err) return } @@ -1497,7 +1570,7 @@ func (b *bess) delSessionQER(ctx context.Context, srcIface uint8, qer qer) { arg, err = anypb.New(q) if err != nil { - logger.BessLog.Errorln("error marshalling the rule", q, err) + logger.BessLog.Errorln(errMarshalRule, q, err) return } diff --git a/pfcpiface/config.go b/pfcpiface/config.go index 0bd1dc981..26ae5a03f 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") } @@ -181,7 +195,6 @@ func validateConf(conf Conf) error { return err } } - return nil } 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..9c84fb61e 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) } @@ -235,7 +239,7 @@ func (pConn *PFCPConn) handleAssociationReleaseRequest(msg message.Message) (mes arres := message.NewAssociationReleaseResponse(arreq.SequenceNumber, pConn.nodeID.localIE, ie.NewCause(ie.CauseMandatoryIEMissing), - //ie.NewOffendingIE(ie.NodeID), + // ie.NewOffendingIE(ie.NodeID), ) return arres, errProcess(errors.New("mandatory IE missing: NodeID")) } @@ -248,7 +252,7 @@ func (pConn *PFCPConn) handleAssociationReleaseRequest(msg message.Message) (mes arres := message.NewAssociationReleaseResponse(arreq.SequenceNumber, pConn.nodeID.localIE, ie.NewCause(ie.CauseMandatoryIEMissing), - //ie.NewOffendingIE(ie.NodeID), // 16 = NodeID + // ie.NewOffendingIE(ie.NodeID), // 16 = NodeID ) return arres, errUnmarshal(err) } @@ -261,7 +265,7 @@ func (pConn *PFCPConn) handleAssociationReleaseRequest(msg message.Message) (mes arres := message.NewAssociationReleaseResponse(arreq.SequenceNumber, pConn.nodeID.localIE, ie.NewCause(ie.CauseRequestRejected), - //ie.NewOffendingIE(ie.NodeID), + // ie.NewOffendingIE(ie.NodeID), ) return arres, nil } @@ -272,14 +276,14 @@ func (pConn *PFCPConn) handleAssociationReleaseRequest(msg message.Message) (mes arres := message.NewAssociationReleaseResponse(arreq.SequenceNumber, pConn.nodeID.localIE, ie.NewCause(ie.CauseMandatoryIEIncorrect), - //ie.NewOffendingIE(16), + // ie.NewOffendingIE(16), ) return arres, nil } // Valid NodeID — send accepted response arres := message.NewAssociationReleaseResponse(arreq.SequenceNumber, - //ie.NewRecoveryTimeStamp(pConn.ts.local), + // ie.NewRecoveryTimeStamp(pConn.ts.local), pConn.nodeID.localIE, ie.NewCause(ie.CauseRequestAccepted), ) 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 10d43b1cd..514b6676f 100644 --- a/pfcpiface/parse_far.go +++ b/pfcpiface/parse_far.go @@ -68,37 +68,7 @@ func (f *far) Forwards() bool { func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error { f.fseID = (fseid) - farID, err := farIE.FARID() - if err != nil { - return err - } - - f.farID = farID - - action, err := farIE.ApplyAction() - if err != nil { - return err - } - - if action[0] == 0 { - return ErrInvalidArgument("FAR Action", action) - } - - f.applyAction = action[0] - - var fwdIEs []*ie.IE - - switch op { - case create: - if (f.applyAction & ActionForward) != 0 { - fwdIEs, err = farIE.ForwardingParameters() - } - case update: - fwdIEs, err = farIE.UpdateForwardingParameters() - default: - return ErrInvalidOperation(op) - } - + fwdIEs, err := f.parseFARBasicFields(farIE, op) if err != nil { return err } @@ -155,3 +125,40 @@ func (f *far) parseFAR(farIE *ie.IE, fseid uint64, upf *upf, op operation) error return nil } + +func (f *far) parseFARBasicFields(farIE *ie.IE, op operation) ([]*ie.IE, error) { + farID, err := farIE.FARID() + if err != nil { + return nil, err + } + f.farID = farID + + action, err := farIE.ApplyAction() + if err != nil { + return nil, err + } + + if action[0] == 0 { + return nil, ErrInvalidArgument("FAR Action", action) + } + f.applyAction = action[0] + + var fwdIEs []*ie.IE + + switch op { + case create: + if (f.applyAction & ActionForward) != 0 { + fwdIEs, err = farIE.ForwardingParameters() + } + case update: + fwdIEs, err = farIE.UpdateForwardingParameters() + default: + return nil, ErrInvalidOperation(op) + } + + if err != nil { + return nil, err + } + + 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 f8ff1e2fe..9db88ce17 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,43 +558,37 @@ 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 { /* reset outerHeaderRemoval to begin with */ outerHeaderRemoval := uint8(0) p.qerIDList = make([]uint32, 0) p.fseID = seid - pdrID, err := ie1.PDRID() - if err != nil { - logger.PfcpLog.Errorln("could not read PDR ID!") - return err - } - - precedence, err := ie1.Precedence() - if err != nil { - logger.PfcpLog.Errorln("could not read Precedence!") - return err - } - - pdi, err := ie1.PDI() - if err != nil { - logger.PfcpLog.Errorln("could not read PDI!") - return err - } - - res, err := ie1.OuterHeaderRemovalDescription() - if res == 0 && err == nil { // 0 == GTP-U/UDP/IPv4 - outerHeaderRemoval = 1 - } - - err = p.parsePDI(pdi, appPFDs, ippool) - if err != nil && !errors.Is(err, errBadFilterDesc) { - return err - } - - farID, err := ie1.FARID() + pdrID, precedence, farID, outerHeaderRemoval, err := p.parseCorePDRFields(ie1, appPFDs, ippool) if err != nil { - logger.PfcpLog.Errorln("could not read FAR ID!") return err } @@ -643,3 +636,47 @@ 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 + } + + precedence, err := ie1.Precedence() + if err != nil { + logger.PfcpLog.Errorln("could not read Precedence!") + return 0, 0, 0, 0, err + } + + pdi, err := ie1.PDI() + if err != nil { + logger.PfcpLog.Errorln("could not read PDI!") + return 0, 0, 0, 0, err + } + + res, err := ie1.OuterHeaderRemovalDescription() + if res == 0 && err == nil { // 0 == GTP-U/UDP/IPv4 + outerHeaderRemoval = 1 + } + + err = p.parsePDI(pdi, appPFDs, ippool) + if err != nil && !errors.Is(err, errBadFilterDesc) { + return 0, 0, 0, 0, err + } + + farID, err := ie1.FARID() + if err != nil { + logger.PfcpLog.Errorln("could not read FAR ID!") + return 0, 0, 0, 0, err + } + + 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..f3767d348 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") @@ -181,7 +182,7 @@ func (m meter) String() string { } func (up4 *UP4) AddSliceInfo(sliceInfo *SliceInfo) error { - //FIXME: UP4 currently supports a single slice meter rate common between UL and DL traffic. For this reason, we + // FIXME: UP4 currently supports a single slice meter rate common between UL and DL traffic. For this reason, we // configure the meter with the largest slice MBR between UL and DL. err := up4.tryConnect() if err != nil { @@ -214,7 +215,6 @@ func (up4 *UP4) AddSliceInfo(sliceInfo *SliceInfo) error { logger.PfcpLog.With("Slice meter entry", sliceMeterEntry).Debugln("installing slice P4 Meter entry") err = up4.p4client.ApplyMeterEntries(p4.Update_MODIFY, sliceMeterEntry) - if err != nil { return err } @@ -890,7 +890,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 +903,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 +914,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 +927,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 +1302,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 +1352,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/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]