app lifecyle updates (ce8ef233) · Commits · MEC - Multi-access Edge Computing / ETSI MEC Sandbox

go-apps/meep-mg-manager/server/mg-manager.go

+134 −85

Original line number	Diff line number	Diff line
		@@ -85,6 +85,9 @@ const lbAlgoHopCount = "HOP-COUNT"
		// const lbAlgoDistance = "DISTANCE"
		// const lbAlgoNone = "NONE"

		// MQ payload fields
		const fieldEventType = "event-type"

		type mgInfo struct {
		mg mgModel.MobilityGroup
		appInfoMap map[string]*appInfo
		@@ -151,10 +154,6 @@ type MgManager struct {
		svcInfoMap map[string]*serviceInfo
		mgSvcInfoMap map[string]*mgServiceInfo

		// mapping from element name to svc name for usercharts
		svcToElemMap map[string]string
		elemToSvcMap map[string]string

		// Network Element Info mapping
		netElemInfoMap map[string]*netElemInfo

		@@ -170,8 +169,6 @@ func Init() (err error) {
		mgm.netLocList = make([]string, 0)
		mgm.svcInfoMap = make(map[string]*serviceInfo)
		mgm.mgSvcInfoMap = make(map[string]*mgServiceInfo)
		mgm.svcToElemMap = make(map[string]string)
		mgm.elemToSvcMap = make(map[string]string)
		mgm.netElemInfoMap = make(map[string]*netElemInfo)
		mgm.mgInfoMap = make(map[string]*mgInfo)

		@@ -222,7 +219,7 @@ func Init() (err error) {
		_ = mgm.lbRulesStore.rc.DBFlush(mgm.baseKey)

		// Initialize Edge-LB rules with current active scenario
		processActiveScenarioUpdate()
		processScenarioActivate()

		return nil
		}
		@@ -246,35 +243,67 @@ func msgHandler(msg *mq.Msg, userData interface{}) {
		switch msg.Message {
		case mq.MsgScenarioActivate:
		log.Debug("RX MSG: ", mq.PrintMsg(msg))
		processActiveScenarioUpdate()
		processScenarioActivate()
		case mq.MsgScenarioUpdate:
		log.Debug("RX MSG: ", mq.PrintMsg(msg))
		processActiveScenarioUpdate()
		eventType := msg.Payload[fieldEventType]
		processScenarioUpdate(eventType)
		case mq.MsgScenarioTerminate:
		log.Debug("RX MSG: ", mq.PrintMsg(msg))
		processActiveScenarioUpdate()
		processScenarioTerminate()
		default:
		log.Trace("Ignoring unsupported message: ", mq.PrintMsg(msg))
		}
		}

		func processActiveScenarioUpdate() {
		func processScenarioActivate() {

		// Sync with active scenario store
		mgm.activeModel.UpdateScenario()

		scenarioName := mgm.activeModel.GetScenarioName()
		if mgm.scenarioName != scenarioName {
		mgm.scenarioName = scenarioName
		_ = httpLog.ReInit(moduleName, mgm.sandboxName, scenarioName, redisAddr, influxAddr)
		// Get scenario name
		mgm.scenarioName = mgm.activeModel.GetScenarioName()
		if mgm.scenarioName == "" {
		log.Error("Failed to find active scenario")
		return
		}

		// Handle empty/missing scenario
		if scenarioName == "" {
		clearScenario()
		// Initialize HTTP metrics logger with scenario name
		_ = httpLog.ReInit(moduleName, mgm.sandboxName, mgm.scenarioName, redisAddr, influxAddr)

		// Parse scenario
		err := processScenario(mgm.activeModel)
		if err != nil {
		log.Error("Failed to process scenario with error: ", err.Error())
		return
		}

		// Set Default Edge-LB mapping
		refreshDefaultNetLocAppMaps()

		// Re-evaluate MG Service mapping
		refreshMgSvcMapping()

		// Store & Apply latest MG Service mappings
		applyMgSvcMapping()

		// Inform TC Engine of LB rules updatge
		publishLbRulesUpdate()
		}

		func processScenarioUpdate(eventType string) {

		// Ignore unsupported update types
		switch eventType {
		case mod.EventMobility, mod.EventPoaInRange, mod.EventAddNode, mod.EventModifyNode, mod.EventRemoveNode:
		break
		default:
		return
		}

		// Sync with active scenario store
		mgm.activeModel.UpdateScenario()

		// Parse scenario
		err := processScenario(mgm.activeModel)
		if err != nil {
		@@ -290,22 +319,40 @@ func processActiveScenarioUpdate() {

		// Store & Apply latest MG Service mappings
		applyMgSvcMapping()

		// Inform TC Engine of LB rules updatge
		publishLbRulesUpdate()
		}

		func processScenarioTerminate() {

		// Sync with active scenario store
		mgm.activeModel.UpdateScenario()

		// Clear scenario data
		clearScenario()

		// Inform TC Engine of LB rules updatge
		publishLbRulesUpdate()
		}

		func clearScenario() {
		log.Debug("clearScenario() -- Resetting all variables")

		mgm.scenarioName = ""
		mgm.networkGraph = nil
		mgm.netLocList = make([]string, 0)
		mgm.svcInfoMap = make(map[string]*serviceInfo)
		mgm.mgSvcInfoMap = make(map[string]*mgServiceInfo)
		mgm.svcToElemMap = make(map[string]string)
		mgm.elemToSvcMap = make(map[string]string)
		mgm.netElemInfoMap = make(map[string]*netElemInfo)
		mgm.mgInfoMap = make(map[string]*mgInfo)

		// Flush module data and send update
		_ = mgm.lbRulesStore.rc.DBFlush(mgm.baseKey)
		}

		// publishLbRulesUpdate - Inform TC Engine of LB rules update
		func publishLbRulesUpdate() {

		// Send LB Rules Update message
		msg := mgm.mqLocal.CreateMsg(mq.MsgMgLbRulesUpdate, moduleTcEngine, mgm.sandboxName)
		@@ -325,6 +372,19 @@ func processScenario(model *mod.Model) error {

		// Get list of processes
		procNames := model.GetNodeNames("CLOUD-APP", "EDGE-APP", "UE-APP")
		procNamesMap := make(map[string]bool)
		for _, procName := range procNames {
		procNamesMap[procName] = true
		}

		// Clear existing network element svc mappings & remove elements no longer in scenario
		for procName, netElem := range mgm.netElemInfoMap {
		if _, found := procNamesMap[procName]; found {
		netElem.mgSvcMap = map[string]*svcMapInfo{}
		} else {
		delete(mgm.netElemInfoMap, procName)
		}
		}

		// Get network graph from model
		mgm.networkGraph = model.GetNetworkGraph()
		@@ -433,8 +493,6 @@ func addServiceInfo(svcName string, mgSvcName string, nodeName string) {

		// Add service instance to service info map
		mgm.svcInfoMap[svcInfo.name] = svcInfo
		mgm.svcToElemMap[svcInfo.name] = svcInfo.name
		mgm.elemToSvcMap[svcInfo.name] = svcInfo.name
		}

		func getNetElem(name string) *netElemInfo {
		@@ -452,19 +510,25 @@ func getNetElem(name string) *netElemInfo {
		return netElem
		}

		func setDefaultNetLocAppMaps() {
		log.Debug("setDefaultNetLocAppMaps")
		func refreshDefaultNetLocAppMaps() {
		log.Debug("refreshDefaultNetLocAppMaps")

		// For each MG Service & net location in scenario, use Group App instances from scenario and
		// default LB algorithm to determine which App instance is best for net location
		for _, mgInfo := range mgm.mgInfoMap {
		// Only set on first pass
		if len(mgInfo.defaultNetLocAppMap) == 0 {
		mgInfo.defaultNetLocAppMap = make(map[string]string)
		for _, netLoc := range mgm.netLocList {
		mgInfo.defaultNetLocAppMap[netLoc] = runLbAlgoHopCount(mgm.mgSvcInfoMap[mgInfo.mg.Name].services, netLoc)
		}
		}
		}

		func refreshNetLocAppMaps() {
		log.Debug("refreshNetLocAppMaps")

		for _, mgInfo := range mgm.mgInfoMap {
		refreshNetLocAppMap(mgInfo)
		}
		}

		func refreshNetLocAppMap(mgInfo *mgInfo) {
		@@ -477,10 +541,6 @@ func refreshNetLocAppMap(mgInfo *mgInfo) {
		var mgApps = map[string]*serviceInfo{}
		for _, appInfo := range mgInfo.appInfoMap {
		mgApps[appInfo.app.Id] = mgm.svcInfoMap[appInfo.app.Id]
		if mgApps[appInfo.app.Id] == nil {

		mgApps[appInfo.app.Id] = mgm.svcInfoMap[mgm.svcToElemMap[appInfo.app.Id]]
		}
		}

		// For each net location in scenario, use Group LB algorithm to determine which
		@@ -533,7 +593,7 @@ func refreshMgSvcMapping() {
		// notification and update mapping
		if bestApp != currentApp {
		log.Info("Best App: " + bestApp + " != Current App: " + currentApp)
		completeStateTransfer(mgInfo, netElemInfo, ueInfo, mgm.elemToSvcMap[currentApp])
		completeStateTransfer(mgInfo, netElemInfo, ueInfo, currentApp)
		setSvcMap(netElemInfo, mgInfo.mg.Name, bestApp)
		}

		@@ -557,7 +617,7 @@ func refreshMgSvcMapping() {
		// notification and update mapping
		if bestApp != currentApp {
		log.Info("Best App: " + bestApp + " != Current App: " + currentApp)
		completeStateTransfer(mgInfo, netElemInfo, ueInfo, mgm.elemToSvcMap[currentApp])
		completeStateTransfer(mgInfo, netElemInfo, ueInfo, currentApp)
		setSvcMap(netElemInfo, mgInfo.mg.Name, bestApp)
		}

		@@ -723,22 +783,12 @@ func applyMgSvcMapping() {
		log.Error(err.Error())
		return
		}

		// Send LB Rules Update message
		msg := mgm.mqLocal.CreateMsg(mq.MsgMgLbRulesUpdate, moduleTcEngine, mgm.sandboxName)
		log.Debug("TX MSG: ", mq.PrintMsg(msg))
		err = mgm.mqLocal.SendMsg(msg)
		if err != nil {
		log.Error("Failed to send message. Error: ", err.Error())
		}
		}

		func mgCreate(mg *mgModel.MobilityGroup) error {
		// Make sure group does not already exist
		if mgm.mgInfoMap[mg.Name] != nil {
		log.Warn("Mobility group already exists: ", mg.Name)
		err := errors.New("Mobility group already exists")
		return err
		return errors.New("Mobility group already exists")
		}

		// Create new Mobility Group & copy data
		@@ -760,8 +810,8 @@ func mgUpdate(mg *mgModel.MobilityGroup) error {
		// Make sure group exists
		mgInfo := mgm.mgInfoMap[mg.Name]
		if mgInfo == nil {
		log.Error("Mobility group does not exist: ", mg.Name)
		err := errors.New("Mobility group does not exist")
		err := errors.New("Mobility group does not exist: " + mg.Name)
		log.Error(err.Error())
		return err
		}

		@@ -775,8 +825,8 @@ func mgUpdate(mg *mgModel.MobilityGroup) error {
		func mgDelete(mgName string) error {
		// Make sure group exists
		if mgm.mgInfoMap[mgName] == nil {
		log.Error("Mobility group does not exist: ", mgName)
		err := errors.New("Mobility group does not exist")
		err := errors.New("Mobility group does not exist: " + mgName)
		log.Error(err.Error())
		return err
		}

		@@ -791,14 +841,20 @@ func mgAppCreate(mgName string, mgApp *mgModel.MobilityGroupApp) error {
		// Make sure group exists
		mgInfo := mgm.mgInfoMap[mgName]
		if mgInfo == nil {
		log.Error("Mobility group does not exist: ", mgName)
		err := errors.New("Mobility group does not exist")
		err := errors.New("Mobility group does not exist: " + mgName)
		log.Error(err.Error())
		return err
		}
		// Make sure App does not already exist
		if mgInfo.appInfoMap[mgApp.Id] != nil {
		log.Error("Mobility group App already exists: ", mgApp.Id)
		err := errors.New("Mobility group App already exists")
		err := errors.New("Mobility group App already exists: " + mgApp.Id)
		log.Error(err.Error())
		return err
		}
		// Make sure App ID is equal to a service instance
		if mgm.svcInfoMap[mgApp.Id] == nil {
		err := errors.New("MG App ID not equal to service instance: " + mgApp.Id)
		log.Error(err.Error())
		return err
		}

		@@ -819,6 +875,8 @@ func mgAppCreate(mgName string, mgApp *mgModel.MobilityGroupApp) error {
		// Add to MG App map & App client map
		mgInfo.appInfoMap[mgApp.Id] = mgAppInfo
		log.Info("Created new MG App: " + mgApp.Id + " in group: " + mgName)

		// Re-evaluate MG best app instance for each scenario network location
		refreshNetLocAppMap(mgInfo)

		// Re-evaluate MG Service mapping
		@@ -834,15 +892,15 @@ func mgAppUpdate(mgName string, mgApp *mgModel.MobilityGroupApp) error {
		// Make sure group exists
		mgInfo := mgm.mgInfoMap[mgName]
		if mgInfo == nil {
		log.Error("Mobility group does not exist: ", mgName)
		err := errors.New("Mobility group does not exist")
		err := errors.New("Mobility group does not exist: " + mgName)
		log.Error(err.Error())
		return err
		}
		// Make sure App exists
		mgAppInfo := mgInfo.appInfoMap[mgApp.Id]
		if mgAppInfo == nil {
		log.Error("Mobility group App does not exist: ", mgApp.Id)
		err := errors.New("Mobility group App does not exist")
		err := errors.New("Mobility group App does not exist: " + mgApp.Id)
		log.Error(err.Error())
		return err
		}

		@@ -854,8 +912,8 @@ func mgAppUpdate(mgName string, mgApp *mgModel.MobilityGroupApp) error {
		mgAppClientCfg.BasePath = mgApp.Url
		mgAppInfo.appClient = mga.NewAPIClient(mgAppClientCfg)
		if mgAppInfo.appClient == nil {
		log.Error("Failed to create MG App REST API client: ", mgAppClientCfg.BasePath)
		err := errors.New("Failed to create MG App REST API client")
		err := errors.New("Failed to create MG App REST API client: " + mgAppClientCfg.BasePath)
		log.Error(err.Error())
		return err
		}

		@@ -867,20 +925,22 @@ func mgAppDelete(mgName string, appID string) error {
		// Make sure group exists
		mgInfo := mgm.mgInfoMap[mgName]
		if mgInfo == nil {
		log.Error("Mobility group does not exist: ", mgName)
		err := errors.New("Mobility group does not exist")
		err := errors.New("Mobility group does not exist: " + mgName)
		log.Error(err.Error())
		return err
		}
		// Make sure App exists
		if mgInfo.appInfoMap[appID] == nil {
		log.Error("Mobility group App does not exist: ", appID)
		err := errors.New("Mobility group App does not exist")
		err := errors.New("Mobility group App does not exist: " + appID)
		log.Error(err.Error())
		return err
		}

		// Remove entry from App map & App Client map
		delete(mgInfo.appInfoMap, appID)
		log.Info("Deleted MG App: " + appID + " in group: " + mgName)

		// Re-evaluate MG best app instance for each scenario network location
		refreshNetLocAppMap(mgInfo)

		return nil
		@@ -890,14 +950,14 @@ func mgUeCreate(mgName string, appID string, mgUe *mgModel.MobilityGroupUe) erro
		// Make sure group exists
		mgInfo := mgm.mgInfoMap[mgName]
		if mgInfo == nil {
		log.Error("Mobility group does not exist: ", mgName)
		err := errors.New("Mobility group does not exist")
		err := errors.New("Mobility group does not exist: " + mgName)
		log.Error(err.Error())
		return err
		}
		// Make sure App exists
		if mgInfo.appInfoMap[appID] == nil {
		log.Error("Mobility group App does not exist: ", appID)
		err := errors.New("Mobility group App does not exist")
		err := errors.New("Mobility group App does not exist: " + appID)
		log.Error(err.Error())
		return err
		}

		@@ -925,23 +985,22 @@ func processAppState(mgName string, appID string, mgAppState *mgModel.MobilityGr
		// Retrieve MG info
		mgInfo := mgm.mgInfoMap[mgName]
		if mgInfo == nil {
		log.Error("Mobility group does not exist: ", mgName)
		err := errors.New("Mobility group does not exist")
		err := errors.New("Mobility group does not exist: " + mgName)
		log.Error(err.Error())
		return err
		}
		// Retrieve App info
		appInfo := mgInfo.appInfoMap[appID]

		if appInfo == nil {
		log.Error("Mobility group App does not exist: ", appID)
		err := errors.New("Mobility group App does not exist")
		err := errors.New("Mobility group App does not exist: " + appID)
		log.Error(err.Error())
		return err
		}
		// Retrieve UE Info
		ueInfo := mgInfo.ueInfoMap[mgAppState.UeId]
		if ueInfo == nil {
		log.Error("Mobility group UE does not exist: ", mgAppState.UeId)
		err := errors.New("Mobility group UE does not exist")
		err := errors.New("Mobility group UE does not exist: " + mgAppState.UeId)
		log.Error(err.Error())
		return err
		}

		@@ -955,8 +1014,6 @@ func processAppState(mgName string, appID string, mgAppState *mgModel.MobilityGr

		mgm.mutex.Lock()
		for appName := range ueInfo.appsInRange {
		appName = mgm.elemToSvcMap[appName]

		if appName != appID {
		appInfo := mgInfo.appInfoMap[appName]
		if appInfo == nil {
		@@ -1432,14 +1489,6 @@ func mgTransferAppState(w http.ResponseWriter, r *http.Request) {
		// log.Debug(" " + k)
		// }
		// }
		// log.Debug("+++ svcToElemMap:")
		// for k, v := range mgm.svcToElemMap {
		// log.Debug(" " + k + ":" + v)
		// }
		// log.Debug("+++ elemToSvcMap:")
		// for k, v := range mgm.elemToSvcMap {
		// log.Debug(" " + k + ":" + v)
		// }
		// log.Debug("+++ netElemInfoMap:")
		// for netElemName, netElemInfo := range mgm.netElemInfoMap {
		// log.Debug(" " + netElemName + ":")