feat: UP initiated association setup

cmd/config/config.go

@@ -11,27 +11,29 @@ import (
 var v = viper.GetViper()
 
 type UpfConfig struct {
-	InterfaceName     []string `mapstructure:"interface_name" json:"interface_name"`
-	XDPAttachMode     string   `mapstructure:"xdp_attach_mode" validate:"oneof=generic native offload" json:"xdp_attach_mode"`
-	ApiAddress        string   `mapstructure:"api_address" validate:"hostname_port" json:"api_address"`
-	PfcpAddress       string   `mapstructure:"pfcp_address" validate:"hostname_port" json:"pfcp_address"`
-	PfcpNodeId        string   `mapstructure:"pfcp_node_id" validate:"hostname|ip" json:"pfcp_node_id"`
-	MetricsAddress    string   `mapstructure:"metrics_address" validate:"hostname_port" json:"metrics_address"`
-	N3Address         string   `mapstructure:"n3_address" validate:"ipv4" json:"n3_address"`
-	GtpPeer           []string `mapstructure:"gtp_peer" validate:"omitempty,dive,hostname_port" json:"gtp_peer"`
-	EchoInterval      uint32   `mapstructure:"echo_interval" validate:"min=1" json:"echo_interval"`
-	QerMapSize        uint32   `mapstructure:"qer_map_size" validate:"min=1" json:"qer_map_size"`
-	FarMapSize        uint32   `mapstructure:"far_map_size" validate:"min=1" json:"far_map_size"`
-	PdrMapSize        uint32   `mapstructure:"pdr_map_size" validate:"min=1" json:"pdr_map_size"`
-	EbpfMapResize     bool     `mapstructure:"resize_ebpf_maps" json:"resize_ebpf_maps"`
-	HeartbeatRetries  uint32   `mapstructure:"heartbeat_retries" json:"heartbeat_retries"`
-	HeartbeatInterval uint32   `mapstructure:"heartbeat_interval" json:"heartbeat_interval"`
-	HeartbeatTimeout  uint32   `mapstructure:"heartbeat_timeout" json:"heartbeat_timeout"`
-	LoggingLevel      string   `mapstructure:"logging_level" validate:"required" json:"logging_level"`
-	UEIPPool          string   `mapstructure:"ueip_pool" validate:"cidr" json:"ueip_pool"`
-	FTEIDPool         uint32   `mapstructure:"teid_pool" json:"teid_pool"`
-	FeatureUEIP       bool     `mapstructure:"feature_ueip" json:"feature_ueip"`
-	FeatureFTUP       bool     `mapstructure:"feature_ftup" json:"feature_ftup"`
+	InterfaceName           []string `mapstructure:"interface_name" json:"interface_name"`
+	XDPAttachMode           string   `mapstructure:"xdp_attach_mode" validate:"oneof=generic native offload" json:"xdp_attach_mode"`
+	ApiAddress              string   `mapstructure:"api_address" validate:"hostname_port" json:"api_address"`
+	PfcpAddress             string   `mapstructure:"pfcp_address" validate:"hostname_port" json:"pfcp_address"`
+	PfcpNodeId              string   `mapstructure:"pfcp_node_id" validate:"hostname|ip" json:"pfcp_node_id"`
+	PfcpRemoteNode          []string `mapstructure:"pfcp_remote_node" validate:"omitempty,dive,hostname|ip" json:"pfcp_node"`
+	AssociationSetupTimeout uint32   `mapstructure:"association_setup_timeout" json:"association_setup_timeout"`
+	MetricsAddress          string   `mapstructure:"metrics_address" validate:"hostname_port" json:"metrics_address"`
+	N3Address               string   `mapstructure:"n3_address" validate:"ipv4" json:"n3_address"`
+	GtpPeer                 []string `mapstructure:"gtp_peer" validate:"omitempty,dive,hostname_port" json:"gtp_peer"`
+	GtpEchoInterval         uint32   `mapstructure:"gtp_echo_interval" validate:"min=1" json:"gtp_echo_interval"`
+	QerMapSize              uint32   `mapstructure:"qer_map_size" validate:"min=1" json:"qer_map_size"`
+	FarMapSize              uint32   `mapstructure:"far_map_size" validate:"min=1" json:"far_map_size"`
+	PdrMapSize              uint32   `mapstructure:"pdr_map_size" validate:"min=1" json:"pdr_map_size"`
+	EbpfMapResize           bool     `mapstructure:"resize_ebpf_maps" json:"resize_ebpf_maps"`
+	HeartbeatRetries        uint32   `mapstructure:"heartbeat_retries" json:"heartbeat_retries"`
+	HeartbeatInterval       uint32   `mapstructure:"heartbeat_interval" json:"heartbeat_interval"`
+	HeartbeatTimeout        uint32   `mapstructure:"heartbeat_timeout" json:"heartbeat_timeout"`
+	LoggingLevel            string   `mapstructure:"logging_level" validate:"required" json:"logging_level"`
+	UEIPPool                string   `mapstructure:"ueip_pool" validate:"cidr" json:"ueip_pool"`
+	FTEIDPool               uint32   `mapstructure:"teid_pool" json:"teid_pool"`
+	FeatureUEIP             bool     `mapstructure:"feature_ueip" json:"feature_ueip"`
+	FeatureFTUP             bool     `mapstructure:"feature_ftup" json:"feature_ftup"`
 }
 
 func init() {
@@ -58,6 +60,10 @@ func init() {
 	pflag.Bool("ftup", false, "Enable or disable FTUP feature")
 	pflag.String("ueippool", "10.60.0.0/24", "IP pool for UEIP feature")
 	pflag.Uint32("teidpool", 65535, "TEID pool for FTUP feature")
+	pflag.StringArray("pfcprnode", []string{}, "Address of remote PFCP node")
+	pflag.StringArray("sxanode", []string{}, "Address of remote Sxa node")
+	pflag.StringArray("sxbnode", []string{}, "Address of remote Sxb node")
+	pflag.Uint32("astimeout", 5, "Association setup timeout in seconds")
 	pflag.Parse()
 
 	// Bind flag errors only when flag is nil, and we ignore empty cli args
@@ -66,10 +72,14 @@ func init() {
 	_ = v.BindPFlag("api_address", pflag.Lookup("aaddr"))
 	_ = v.BindPFlag("pfcp_address", pflag.Lookup("paddr"))
 	_ = v.BindPFlag("pfcp_node_id", pflag.Lookup("nodeid"))
+	_ = v.BindPFlag("pfcp_remote_node", pflag.Lookup("pfcprnode"))
+	_ = v.BindPFlag("sxa_remote_node", pflag.Lookup("sxanode"))
+	_ = v.BindPFlag("sxb_remote_node", pflag.Lookup("sxbnode"))
+	_ = v.BindPFlag("association_setup_timeout", pflag.Lookup("astimeout"))
 	_ = v.BindPFlag("metrics_address", pflag.Lookup("maddr"))
 	_ = v.BindPFlag("n3_address", pflag.Lookup("n3addr"))
 	_ = v.BindPFlag("gtp_peer", pflag.Lookup("peer"))
-	_ = v.BindPFlag("echo_interval", pflag.Lookup("echo"))
+	_ = v.BindPFlag("gtp_echo_interval", pflag.Lookup("echo"))
 	_ = v.BindPFlag("qer_map_size", pflag.Lookup("qersize"))
 	_ = v.BindPFlag("far_map_size", pflag.Lookup("farsize"))
 	_ = v.BindPFlag("pdr_map_size", pflag.Lookup("pdrsize"))
@@ -88,9 +98,10 @@ func init() {
 	v.SetDefault("api_address", ":8080")
 	v.SetDefault("pfcp_address", "127.0.0.1:8805")
 	v.SetDefault("pfcp_node_id", "127.0.0.1")
+	v.SetDefault("association_setup_timeout", 5)
 	v.SetDefault("metrics_address", ":9090")
 	v.SetDefault("n3_address", "127.0.0.1")
-	v.SetDefault("echo_interval", 10)
+	v.SetDefault("gtp_echo_interval", 10)
 	v.SetDefault("qer_map_size", 1024)
 	v.SetDefault("far_map_size", 1024)
 	v.SetDefault("pdr_map_size", 1024)

cmd/core/ie_overwrite_test.go

@@ -2,7 +2,9 @@ package core
 
 import (
 	"net"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/edgecomllc/eupf/cmd/ebpf"
 	"github.com/wmnsk/go-pfcp/ie"
@@ -67,9 +69,11 @@ func TestSessionOverwrite(t *testing.T) {
 		nodeId:           "test-node",
 		mapOperations:    &mapOps,
 		n3Address:        net.ParseIP("127.0.0.1"),
+		associationMutex: &sync.Mutex{},
 	}
 	asReq := message.NewAssociationSetupRequest(0,
 		ie.NewNodeID("", "", "test"),
+		ie.NewRecoveryTimeStamp(time.Now()),
 	)
 	remoteIP := "127.0.0.1"
 	response, err := HandlePfcpAssociationSetupRequest(&pfcpConn, asReq, remoteIP)

cmd/core/node_association.go

@@ -17,7 +17,6 @@ type NodeAssociation struct {
 	NextSequenceID   uint32
 	Sessions         map[uint64]*Session
 	HeartbeatChannel chan uint32
-	FailedHeartbeats uint32
 	HeartbeatsActive bool
 	sync.Mutex
 	// AssociationStart time.Time // Held until propper failure detection is implemented
@@ -41,29 +40,34 @@ func (association *NodeAssociation) NewLocalSEID() uint64 {
 }
 
 func (association *NodeAssociation) NewSequenceID() uint32 {
+	association.Lock()
+	defer association.Unlock()
+
 	association.NextSequenceID += 1
 	return association.NextSequenceID
 }
 
 func (association *NodeAssociation) ScheduleHeartbeat(conn *PfcpConnection) {
-	association.HeartbeatsActive = true
 	ctx := context.Background()
+	failedHeartbeats := uint32(0)
 
 	for {
 		sequence := association.NewSequenceID()
 		SendHeartbeatRequest(conn, sequence, association.Addr)
 
+		heartbeatTimeout := time.NewTimer(time.Duration(config.Conf.HeartbeatTimeout) * time.Second)
 		select {
-		case <-time.After(time.Duration(config.Conf.HeartbeatTimeout) * time.Second):
-			if !association.HandleHeartbeatTimeout() {
+		case <-heartbeatTimeout.C:
+			failedHeartbeats++
+			if failedHeartbeats >= config.Conf.HeartbeatRetries {
 				log.Warn().Msgf("the number of unanswered heartbeats has reached the limit, association deleted: %s", association.Addr)
-				close(association.HeartbeatChannel)
-				conn.DeleteAssociation(association.Addr)
+				conn.heartbeatFailedC <- association.Addr
 				return
 			}
 		case seq := <-association.HeartbeatChannel:
 			if sequence == seq {
-				association.ResetFailedHeartbeats()
+				heartbeatTimeout.Stop()
+				failedHeartbeats = 0
 				<-time.After(time.Duration(config.Conf.HeartbeatInterval) * time.Second)
 			}
 		case <-ctx.Done():
@@ -73,25 +77,6 @@ func (association *NodeAssociation) ScheduleHeartbeat(conn *PfcpConnection) {
 	}
 }
 
-func (association *NodeAssociation) ResetFailedHeartbeats() {
-	association.Lock()
-	association.FailedHeartbeats = 0
-	association.Unlock()
-}
-
-func (association *NodeAssociation) HandleHeartbeatTimeout() bool {
-	association.Lock()
-	defer association.Unlock()
-
-	association.FailedHeartbeats++
-	return association.FailedHeartbeats < config.Conf.HeartbeatRetries
-}
-
 func (association *NodeAssociation) HandleHeartbeat(sequence uint32) {
-	association.Lock()
-	defer association.Unlock()
-
-	if association.HeartbeatChannel != nil {
-		association.HeartbeatChannel <- sequence
-	}
+	association.HeartbeatChannel <- sequence
 }

cmd/core/pfcp_connection.go

@@ -3,27 +3,48 @@ package core
 import (
 	"fmt"
 	"net"
+	"sync"
 	"time"
 
+	"github.com/edgecomllc/eupf/cmd/config"
 	"github.com/edgecomllc/eupf/cmd/core/service"
 
-	"github.com/edgecomllc/eupf/cmd/config"
 	"github.com/edgecomllc/eupf/cmd/ebpf"
 	"github.com/rs/zerolog/log"
 
+	"github.com/wmnsk/go-pfcp/ie"
 	"github.com/wmnsk/go-pfcp/message"
 )
 
+type AssociationConnector interface {
+	getAddress() string
+	sendAssociationSetupRequest(connection *PfcpConnection)
+}
+
+var pfcpHandlers = PfcpHandlerMap{
+	message.MsgTypeHeartbeatRequest:            HandlePfcpHeartbeatRequest,
+	message.MsgTypeHeartbeatResponse:           HandlePfcpHeartbeatResponse,
+	message.MsgTypeAssociationSetupRequest:     HandlePfcpAssociationSetupRequest,
+	message.MsgTypeAssociationSetupResponse:    HandlePfcpAssociationSetupResponse,
+	message.MsgTypeSessionEstablishmentRequest: HandlePfcpSessionEstablishmentRequest,
+	message.MsgTypeSessionDeletionRequest:      HandlePfcpSessionDeletionRequest,
+	message.MsgTypeSessionModificationRequest:  HandlePfcpSessionModificationRequest,
+}
+
 type PfcpConnection struct {
 	udpConn           *net.UDPConn
 	pfcpHandlerMap    PfcpHandlerMap
+	associationMutex  *sync.Mutex
 	NodeAssociations  map[string]*NodeAssociation
 	nodeId            string
 	nodeAddrV4        net.IP
 	n3Address         net.IP
 	mapOperations     ebpf.ForwardingPlaneController
 	RecoveryTimestamp time.Time
+	featuresOctets    []uint8
 	ResourceManager   *service.ResourceManager
+	heartbeatFailedC  chan string
+	nodes             []AssociationConnector
 }
 
 func (connection *PfcpConnection) GetAssociation(assocAddr string) *NodeAssociation {
@@ -33,7 +54,7 @@ func (connection *PfcpConnection) GetAssociation(assocAddr string) *NodeAssociat
 	return nil
 }
 
-func CreatePfcpConnection(addr string, pfcpHandlerMap PfcpHandlerMap, nodeId string, n3Ip string, mapOperations ebpf.ForwardingPlaneController, resourceManager *service.ResourceManager) (*PfcpConnection, error) {
+func NewPfcpConnection(addr string, nodeId string, n3Ip string, mapOperations ebpf.ForwardingPlaneController, resourceManager *service.ResourceManager) (*PfcpConnection, error) {
 	udpAddr, err := net.ResolveUDPAddr("udp", addr)
 	if err != nil {
 		log.Warn().Msgf("Can't resolve UDP address: %s", err.Error())
@@ -51,36 +72,61 @@ func CreatePfcpConnection(addr string, pfcpHandlerMap PfcpHandlerMap, nodeId str
 	}
 	log.Info().Msgf("Starting PFCP connection: %v with Node ID: %v and N3 address: %v", udpAddr, nodeId, n3Addr)
 
+	featuresOctets := []uint8{0, 0, 0}
+	featuresOctets[1] = setBit(featuresOctets[1], 0)
+	if config.Conf.FeatureFTUP {
+		featuresOctets[0] = setBit(featuresOctets[0], 4)
+	}
+	if config.Conf.FeatureUEIP {
+		featuresOctets[2] = setBit(featuresOctets[2], 2)
+	}
+
 	return &PfcpConnection{
 		udpConn:           udpConn,
-		pfcpHandlerMap:    pfcpHandlerMap,
+		pfcpHandlerMap:    pfcpHandlers,
+		associationMutex:  &sync.Mutex{},
 		NodeAssociations:  map[string]*NodeAssociation{},
 		nodeId:            nodeId,
 		nodeAddrV4:        udpAddr.IP,
 		n3Address:         n3Addr,
 		mapOperations:     mapOperations,
 		RecoveryTimestamp: time.Now(),
+		featuresOctets:    featuresOctets,
 		ResourceManager:   resourceManager,
+		heartbeatFailedC:  make(chan string),
+		nodes:             []AssociationConnector{},
 	}, nil
 }
 
+func (connection *PfcpConnection) SetRemoteNodes(nodes []AssociationConnector) {
+	connection.nodes = nodes
+}
+
 func (connection *PfcpConnection) Run() {
-	go func() {
-		for {
-			connection.RefreshAssociations()
-			time.Sleep(time.Duration(config.Conf.HeartbeatInterval) * time.Second)
-		}
-	}()
+
+	ticker := time.NewTicker(time.Duration(config.Conf.AssociationSetupTimeout) * time.Second)
 	buf := make([]byte, 1500)
+
 	for {
-		n, addr, err := connection.Receive(buf)
-		if err != nil {
-			log.Warn().Msgf("Error reading from UDP socket: %s", err.Error())
-			time.Sleep(1 * time.Second)
-			continue
+		select {
+		case <-ticker.C:
+			connection.RefreshAssociations()
+		case associationAddr := <-connection.heartbeatFailedC:
+			connection.DeleteAssociation(associationAddr)
+		default:
+			_ = connection.udpConn.SetReadDeadline(time.Now().Add(time.Second))
+			n, addr, err := connection.Receive(buf)
+			if err != nil {
+				if err.(*net.OpError).Timeout() {
+					continue
+				}
+				log.Warn().Msgf("Error reading from UDP socket: %s", err.Error())
+				time.Sleep(1 * time.Second)
+				continue
+			}
+			log.Debug().Msgf("Received %d bytes from %s", n, addr)
+			connection.Handle(buf[:n], addr)
 		}
-		log.Debug().Msgf("Received %d bytes from %s", n, addr)
-		connection.Handle(buf[:n], addr)
 	}
 }
 
@@ -117,9 +163,9 @@ func (connection *PfcpConnection) SendMessage(msg message.Message, addr *net.UDP
 }
 
 func (connection *PfcpConnection) RefreshAssociations() {
-	for _, assoc := range connection.NodeAssociations {
-		if !assoc.HeartbeatsActive {
-			go assoc.ScheduleHeartbeat(connection)
+	for _, node := range connection.nodes {
+		if connection.GetAssociation(node.getAddress()) == nil {
+			node.sendAssociationSetupRequest(connection)
 		}
 	}
 }
@@ -174,3 +220,37 @@ func (connection *PfcpConnection) ReleaseResources(seID uint64) {
 		connection.ResourceManager.FTEIDM.ReleaseTEID(seID)
 	}
 }
+
+type DefaultAssociationConnector struct {
+	address string
+}
+
+func NewDefaultAssociationConnector(address string) *DefaultAssociationConnector {
+	return &DefaultAssociationConnector{
+		address: address,
+	}
+}
+
+func (connector *DefaultAssociationConnector) getAddress() string {
+	return connector.address
+}
+
+func (connector *DefaultAssociationConnector) sendAssociationSetupRequest(connection *PfcpConnection) {
+
+	associationAddr := connector.getAddress()
+	AssociationSetupRequest := message.NewAssociationSetupRequest(0,
+		newIeNodeID(connection.nodeId),
+		ie.NewRecoveryTimeStamp(connection.RecoveryTimestamp),
+		ie.NewUPFunctionFeatures(connection.featuresOctets[:]...),
+	)
+	log.Info().Msgf("Sent Association Setup Request to: %s", associationAddr)
+
+	udpAddr, err := net.ResolveUDPAddr("udp", associationAddr+":8805")
+	if err != nil {
+		log.Error().Msgf("Failed to resolve udp address from PFCP peer address %s. Error: %s\n", associationAddr, err.Error())
+		return
+	}
+	if err := connection.SendMessage(AssociationSetupRequest, udpAddr); err != nil {
+		log.Info().Msgf("Failed to send Association Setup Request: %s\n", err.Error())
+	}
+}