[CENNSO-871] Fix sending PFCP datagrams to wrong IPs/ports

test/e2e/pfcp/pfcp.go

@@ -383,6 +383,7 @@ type PFCPConfig struct {
 	// IgnoreHeartbeatRequests makes PFCPConnection ignore incoming
 	// PFCP Heartbeat Requests, thus simulating a faulty CP.
 	IgnoreHeartbeatRequests bool
+	RecoveryTimestamp       time.Time
 }
 
 func (cfg *PFCPConfig) setDefaults() {
@@ -528,6 +529,7 @@ func (pc *PFCPConnection) event(event pfcpEvent) error {
 	}
 
 	if err != nil {
+		pc.log.WithError(err).Error("entering FAILED state")
 		pc.state = pfcpStateFailed
 		pc.cleanAndDone()
 		if event.resultCh != nil {
@@ -605,6 +607,7 @@ LOOP:
 			}
 			if err = pc.simpleEvent(pfcpEventActStop); err != nil {
 				err = errors.Wrapf(err, "stop in state %s", pc.state)
+				pc.log.WithError(err).Error("terminating the event loop")
 				break LOOP
 			}
 		case id := <-pc.timer.Channel():
@@ -618,6 +621,7 @@ LOOP:
 					timerID:   tid,
 				}); err != nil {
 					err = errors.Wrapf(err, "timeout in state %s", pc.state)
+					pc.log.WithError(err).Error("terminating the event loop")
 					break LOOP
 				}
 
@@ -640,6 +644,7 @@ LOOP:
 					msgType = ev.msg.MessageTypeName()
 				}
 				err = errors.Wrapf(err, "error handling event %s / msg type %s in state %s", ev.eventType, msgType, pc.state)
+				pc.log.WithError(err).Error("terminating the event loop")
 				break LOOP
 			}
 		case err = <-pc.listenErrCh:
@@ -914,7 +919,11 @@ func (pc *PFCPConnection) Start(ctx context.Context) error {
 	pc.conn = nil
 	pc.eventCh = make(chan pfcpEvent, 110000)
 	pc.listenErrCh = make(chan error, 1)
-	pc.timestamp = time.Now()
+	if pc.cfg.RecoveryTimestamp.IsZero() {
+		pc.timestamp = time.Now()
+	} else {
+		pc.timestamp = pc.cfg.RecoveryTimestamp
+	}
 	pc.seq = pc.cfg.InitialSeq
 	if pc.seq == 0 {
 		pc.seq = 1
@@ -937,7 +946,7 @@ func (pc *PFCPConnection) Start(ctx context.Context) error {
 		close(pc.startCh)
 		return r.err
 	case <-tch:
-		panic("Association Setup Request over timeout")
+		return errors.New("Association Setup Request over timeout")
 	}
 }
 
@@ -1147,7 +1156,7 @@ func (pc *PFCPConnection) sessionEstablishmentRequest(spec SessionOpSpec) messag
 
 func (pc *PFCPConnection) enqueueRequest(seid SEID, msg message.Message, resultCh chan eventResult) error {
 	if !pc.t.Alive() {
-		return errors.New("PFCPConnection not active")
+		return errors.Errorf("PFCPConnection not active: %s", pc.cfg.NodeID)
 	}
 
 	eventType, ok := requestToEventType(msg)

test/e2e/upg_e2e.go

@@ -1158,6 +1158,59 @@ var _ = ginkgo.Describe("Clearing message queue", func() {
 	})
 })
 
+const numPFCPPeers = 10
+const sessionsPerPeer = 1
+
+var _ = ginkgo.Describe("Multiple PFCP peers", func() {
+	f := framework.NewDefaultFramework(framework.UPGModeTDF, framework.UPGIPModeV4)
+	ginkgo.It("should work correctly", func() {
+		var conns [numPFCPPeers]*pfcp.PFCPConnection
+		hbConfig := &upf.UpfPfcpHeartbeatsSet{
+			Retries: 5,
+			Timeout: 1,
+		}
+		reply := &upf.UpfPfcpHeartbeatsSetReply{}
+		err := f.VPP.ApiChannel.SendRequest(hbConfig).ReceiveReply(reply)
+		gomega.Expect(err).To(gomega.BeNil())
+
+		for i := 0; i < numPFCPPeers; i++ {
+			pfcpCfg := framework.DefaultPFCPConfig(*f.VPPCfg)
+			pfcpCfg.Namespace = f.VPP.GetNS("cp")
+			pfcpCfg.NodeID = fmt.Sprintf("node%d", i)
+			pfcpCfg.CNodeIP = f.AddCNodeIP()
+			pfcpCfg.RecoveryTimestamp = time.Now().Local().Add(time.Duration(-i) * 24 * time.Hour)
+			pc := pfcp.NewPFCPConnection(pfcpCfg)
+			framework.ExpectNoError(pc.Start(f.Context))
+			conns[i] = pc
+
+		}
+		// time.Sleep(40 * time.Second)
+		time.Sleep(10 * time.Second)
+
+		for _, pc := range conns {
+			specs := make([]pfcp.SessionOpSpec, sessionsPerPeer)
+			for i := 0; i < sessionsPerPeer; i++ {
+				scfg := framework.SessionConfig{
+					IdBase: 1,
+					UEIP:   f.AddUEIP(),
+					Mode:   framework.UPGModeTDF,
+					VTime:  2 * time.Hour,
+				}
+				specs[i].IEs = scfg.SessionIEs()
+
+			}
+			_, errs := pc.EstablishSessions(context.Background(), specs[:sessionsPerPeer])
+			for _, err := range errs {
+				framework.ExpectNoError(err)
+			}
+		}
+
+		for _, pc := range conns {
+			pc.Stop()
+		}
+	})
+})
+
 var _ = ginkgo.Describe("[Reporting]", func() {
 	ginkgo.Context("Quota Validity Time", func() {
 		f := framework.NewDefaultFramework(framework.UPGModeTDF, framework.UPGIPModeV4)

vpp.spec

@@ -1 +1 @@
-VPP_IMAGE_BASE=quay.io/travelping/fpp-vpp:v22.02.5
+VPP_IMAGE_BASE=quay.io/travelping/fpp-vpp:v22.02.6