Aether Scaling

consumer/nnrf.go

@@ -10,7 +10,10 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"sort"
+	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/antihax/optional"
@@ -27,6 +30,8 @@ import (
 	"github.com/omec-project/smf/msgtypes/svcmsgtypes"
 )
 
+var SmContextlenMutex sync.Mutex
+
 func SendNFRegistration() (*models.NfProfile, error) {
 	var rep models.NfProfile
 	sNssais := []models.Snssai{}
@@ -51,6 +56,7 @@ func SendNFRegistration() (*models.NfProfile, error) {
 		SNssais:       &sNssais,
 		PlmnList:      smf_context.SmfPlmnConfig(),
 		AllowedPlmns:  smf_context.SmfPlmnConfig(),
+		Load:          smf_context.GetSmContextLen(),
 	}
 
 	var res *http.Response
@@ -250,7 +256,31 @@ func SendNrfForNfInstance(nrfUri string, targetNfType, requestNfType models.NfTy
 	return result, localErr
 }
 
-func SendNFDiscoveryUDM() (*models.ProblemDetails, error) {
+func SendNFDiscoveryUDM(smContext *smf_context.SMContext) (*models.ProblemDetails, error) {
+	// func SendNFDiscoveryUDM(id string) (*models.ProblemDetails, error) {
+	id := smContext.Supi
+
+	value, ok := smf_context.SMF_Self().UdmProfileMap.Load(id)
+	if ok {
+		smContext.SelectedUdmProfile = value.(models.NfProfile)
+		smContext.UdmIP = smContext.SelectedUdmProfile.Ipv4Addresses[0]
+		for _, service := range *smContext.SelectedUdmProfile.NfServices {
+			if service.ServiceName == models.ServiceName_NUDM_SDM {
+				SDMConf := Nudm_SubscriberDataManagement.NewConfiguration()
+				SDMConf.SetBasePath(service.ApiPrefix)
+				smContext.SubscriberDataManagementClient = Nudm_SubscriberDataManagement.NewAPIClient(SDMConf)
+			}
+		}
+		return nil, nil
+	}
+
+	SmContextlenMutex.Lock()
+	smf_context.IncSmContextLen()
+	if _, err := SendNFRegistration(); err != nil {
+		fmt.Println("NRF Registration failure ", err.Error())
+	}
+	SmContextlenMutex.Unlock()
+
 	localVarOptionals := Nnrf_NFDiscovery.SearchNFInstancesParamOpts{}
 
 	var result models.SearchResult
@@ -263,17 +293,48 @@ func SendNFDiscoveryUDM() (*models.ProblemDetails, error) {
 	}
 
 	if localErr == nil {
-		smf_context.SMF_Self().UDMProfile = result.NfInstances[0]
+		// smf_context.SMF_Self().UDMProfile = result.NfInstances[0]
+		nfInstanceIds := make([]string, 0, len(result.NfInstances))
+		for _, nfProfile := range result.NfInstances {
+			nfInstanceIds = append(nfInstanceIds, nfProfile.NfInstanceId)
+		}
+		sort.Strings(nfInstanceIds)
+		nfInstanceIdIndexMap := make(map[string]int)
+		for index, value := range nfInstanceIds {
+			nfInstanceIdIndexMap[value] = index
+		}
 
-		for _, service := range *smf_context.SMF_Self().UDMProfile.NfServices {
+		nfInstanceIndex := 0
+		if smf_context.SMF_Self().EnableScaling {
+			parts := strings.Split(id, "-")
+			imsiNumber, err := strconv.Atoi(parts[1])
+			if err != nil {
+				logger.ConsumerLog.Warnln("imsiNumber conversion failed")
+				return nil, err
+			}
+			nfInstanceIndex = imsiNumber % len(result.NfInstances)
+		}
+
+		for _, nfProfile := range result.NfInstances {
+			if nfInstanceIndex != nfInstanceIdIndexMap[nfProfile.NfInstanceId] {
+				continue
+			}
+			smContext.SelectedUdmProfile = nfProfile
+			smContext.UdmIP = nfProfile.Ipv4Addresses[0]
+			smf_context.SMF_Self().UdmProfileMap.Store(id, nfProfile)
+			logger.ConsumerLog.Warnln("for Ue: ", id, " nfInstanceIndex: ", nfInstanceIndex, " for targetNfType ", string(models.NfType_UDM), " NF is: ", nfProfile.Ipv4Addresses)
+
+			break
+		}
+		for _, service := range *smContext.SelectedUdmProfile.NfServices {
 			if service.ServiceName == models.ServiceName_NUDM_SDM {
 				SDMConf := Nudm_SubscriberDataManagement.NewConfiguration()
 				SDMConf.SetBasePath(service.ApiPrefix)
-				smf_context.SMF_Self().SubscriberDataManagementClient = Nudm_SubscriberDataManagement.NewAPIClient(SDMConf)
+				smContext.SubscriberDataManagementClient = Nudm_SubscriberDataManagement.NewAPIClient(SDMConf)
 			}
 		}
 
-		if smf_context.SMF_Self().SubscriberDataManagementClient == nil {
+		if smContext.SubscriberDataManagementClient == nil {
 			logger.ConsumerLog.Warnln("sdm client failed")
 		}
 	} else {
@@ -334,6 +395,7 @@ func SendNFDiscoveryServingAMF(smContext *smf_context.SMContext) (*models.Proble
 			return nil, openapi.ReportError("NfInstances is nil")
 		}
 		smContext.SubConsumerLog.Info("send NF Discovery Serving AMF Successful")
+		// FIX: FIND AMF based on smContext.Supi
 		smContext.AMFProfile = deepcopy.Copy(result.NfInstances[0]).(models.NfProfile)
 	} else {
 		apiError, ok := localErr.(openapi.GenericOpenAPIError)

context/context.go

@@ -14,6 +14,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/go-redis/redis"
 	"github.com/google/uuid"
 	"github.com/omec-project/openapi/Nnrf_NFDiscovery"
 	"github.com/omec-project/openapi/Nnrf_NFManagement"
@@ -80,10 +81,15 @@ type SMFContext struct {
 	CPNodeID                 NodeID
 	PFCPPort                 int
 	UDMProfile               models.NfProfile
+	UdmProfileMap            sync.Map
+	PcfProfileMap            sync.Map
 	NrfCacheEvictionInterval time.Duration
 	SBIPort                  int
 	LocalSEIDCount           uint64
 	EnableNrfCaching         bool
+	EnableScaling            bool
+	RedisClient              *redis.Client
+	RedisDb                  int
 
 	// For ULCL
 	ULCLSupport bool
@@ -249,7 +255,7 @@ func InitSmfContext(config *factory.Config) *SMFContext {
 	smfContext.UserPlaneInformation = NewUserPlaneInformation(&configuration.UserPlaneInformation)
 
 	smfContext.EnableNrfCaching = configuration.EnableNrfCaching
-
+	smfContext.EnableScaling = configuration.EnableScaling
 	if configuration.EnableNrfCaching {
 		if configuration.NrfCacheEvictionInterval == 0 {
 			smfContext.NrfCacheEvictionInterval = time.Duration(900) // 15 mins
@@ -260,7 +266,13 @@ func InitSmfContext(config *factory.Config) *SMFContext {
 
 	smfContext.PodIp = os.Getenv("POD_IP")
 	SetupNFProfile(config)
-
+	if smfContext.EnableScaling {
+		smfContext.RedisClient = redis.NewClient(&redis.Options{
+			Addr:     "redis:6379",
+			Password: "",
+			DB:       factory.SmfConfig.Configuration.RedisDb,
+		})
+	}
 	return &smfContext
 }
 
@@ -433,13 +445,15 @@ func (smfCtxt *SMFContext) InitDrsm() error {
 	db := drsm.DbInfo{Url: dbUrl, Name: dbName}
 
 	// for local FSEID
+	// if drsmCtxt, err := drsm.InitDRSM("fseid", podId, db, opt, factory.SmfConfig.Configuration.DrsmPunchLiveTime); err == nil {
 	if drsmCtxt, err := drsm.InitDRSM("fseid", podId, db, opt); err == nil {
 		smfCtxt.DrsmCtxts.SeidPool = drsmCtxt
 	} else {
 		return err
 	}
 
 	// for local FTEID
+	// if drsmCtxt, err := drsm.InitDRSM("fteid", podId, db, opt, factory.SmfConfig.Configuration.DrsmPunchLiveTime); err == nil {
 	if drsmCtxt, err := drsm.InitDRSM("fteid", podId, db, opt); err == nil {
 		smfCtxt.DrsmCtxts.TeidPool = drsmCtxt
 	} else {