improved graceful termination triggered by sandbox controller (96b0c969) · Commits · MEC - Multi-access Edge Computing / ETSI MEC Sandbox

go-apps/meep-app-enablement/server/app-support/app-support.go

+16 −21

Original line number	Diff line number	Diff line
		@@ -79,7 +79,7 @@ var basePath string
		var baseKey string
		var subMgr *subs.SubscriptionMgr
		var appStore *apps.ApplicationStore
		var gracefulTerminateMap = map[string]*time.Ticker{}
		var gracefulTerminateMap = map[string]chan bool{}

		func notImplemented(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json; charset=UTF-8")
		@@ -283,17 +283,12 @@ func applicationsConfirmTerminationPOST(w http.ResponseWriter, r *http.Request)
		return
		}

		// Check if Confirm Termination was expected
		gracefulTerminateTicker, found := gracefulTerminateMap[appId]
		// Verify that confirmation is epxected
		gracefulTerminateChannel, found := gracefulTerminateMap[appId]
		if !found {
		mutex.Unlock()
		log.Error("Unexpected App Confirmation Termination Notification")
		http.Error(w, "Unexpected App Confirmation Termination Notification", http.StatusBadRequest)
		return
		} else {
		// Stop & delete ticker
		gracefulTerminateTicker.Stop()
		delete(gracefulTerminateMap, appId)
		}

		// Retrieve Termination Confirmation data
		@@ -320,8 +315,9 @@ func applicationsConfirmTerminationPOST(w http.ResponseWriter, r *http.Request)
		return
		}

		// Delete App Instance
		deleteAppInstance(appId)
		// Confirm termination
		gracefulTerminateChannel <- true
		delete(gracefulTerminateMap, appId)

		// Send response
		w.WriteHeader(http.StatusNoContent)
		@@ -830,8 +826,8 @@ func terminateAppInfo(appId string) error {

		// Start graceful timeout timer prior to sending the app termination notification
		mutex.Lock()
		gracefulTerminateTicker := time.NewTicker(time.Duration(DEFAULT_GRACEFUL_TIMEOUT) * time.Second)
		gracefulTerminateMap[appId] = gracefulTerminateTicker
		gracefulTerminateChannel := make(chan bool)
		gracefulTerminateMap[appId] = gracefulTerminateChannel
		mutex.Unlock()

		go func(sub *subs.Subscription) {
		@@ -842,18 +838,17 @@ func terminateAppInfo(appId string) error {
		}

		// Wait for app termination confirmation or timeout
		for range gracefulTerminateTicker.C {
		select {
		case <-gracefulTerminateChannel:
		log.Debug("Termination confirmation received for: ", appId)
		case <-time.After(time.Duration(DEFAULT_GRACEFUL_TIMEOUT) * time.Second):
		mutex.Lock()
		if gracefulTerminateTicker, found := gracefulTerminateMap[appId]; found {
		log.Info("Graceful timeout expiry for ", appId, "---", gracefulTerminateTicker)
		gracefulTerminateTicker.Stop()
		delete(gracefulTerminateMap, appId)
		}
		mutex.Unlock()
		}

		// Delete App instance if timer expires before receiving a termination confirmation
		// Delete App instance
		deleteAppInstance(appId)
		}
		}(sub)
		}

go-apps/meep-sandbox-ctrl/server/app-ctrl.go

+21 −56

Original line number	Diff line number	Diff line
		@@ -48,8 +48,11 @@ type AppCtrl struct {
		const (
		mqFieldAppId = "id"
		mqFieldPersist = "persist"
		mqFieldGracePeriod = "gracePeriod"
		)

		const defaultGracePeriod int = 10

		// App Controller
		var appCtrl *AppCtrl

		@@ -117,33 +120,7 @@ func msgHandler(msg *mq.Msg, userData interface{}) {
		case mq.MsgAppRemoveCnf:
		log.Debug("RX MSG: ", mq.PrintMsg(msg))
		appId := msg.Payload[mqFieldAppId]

		// If process exists, remove it from the active scenario
		activeModel := getActiveModel()
		if activeModel != nil {
		proc, ctx, err := getScenarioProcessById(appId, activeModel)
		if err == nil {
		// Prepare scenario update event
		event := &dataModel.Event{
		Type_: "SCENARIO-UPDATE",
		EventScenarioUpdate: &dataModel.EventScenarioUpdate{
		Action: "REMOVE",
		Node: &dataModel.ScenarioNode{
		Type_: proc.Type_,
		Parent: ctx.Parents[mod.PhyLoc],
		NodeDataUnion: &dataModel.NodeDataUnion{
		Process: proc,
		},
		},
		},
		}
		// Process event to remove node
		_, err = processEvent(event.Type_, event)
		if err != nil {
		log.Error(err.Error())
		}
		}
		}
		removeNodeConfirm(appId)
		default:
		}
		}
		@@ -182,7 +159,7 @@ func setAppInstance(name string, activeModel *mod.Model) error {
		}

		// Set app instance
		err = appCtrl.appStore.Set(app)
		err = appCtrl.appStore.Set(app, nil)
		if err != nil {
		log.Error(err.Error())
		return err
		@@ -190,14 +167,14 @@ func setAppInstance(name string, activeModel *mod.Model) error {
		return nil
		}

		func delAppInstance(id string) error {
		func delAppInstance(id string, gracePeriod int) error {
		// Validate ID
		if id == "" {
		return errors.New("Invalid app instance ID")
		}

		// Delete app instance
		err := appCtrl.appStore.Del(id)
		err := appCtrl.appStore.Del(id, &gracePeriod)
		if err != nil {
		log.Warn(err.Error())
		return err
		@@ -207,7 +184,7 @@ func delAppInstance(id string) error {

		func resetAppInstances(activeModel *mod.Model) error {
		// Flush non-persistent app instances
		appCtrl.appStore.FlushNonPersistent()
		appCtrl.appStore.FlushNonPersistent(nil)

		// Get active scenario app list
		scenarioAppList, err := getScenarioAppInstanceList(activeModel)
		@@ -218,7 +195,7 @@ func resetAppInstances(activeModel *mod.Model) error {

		// Create app instances for scenario apps
		for _, app := range scenarioAppList {
		err := appCtrl.appStore.Set(app)
		err := appCtrl.appStore.Set(app, nil)
		if err != nil {
		log.Error(err.Error())
		}
		@@ -272,24 +249,6 @@ func getScenarioProcess(name string, activeModel mod.Model) (dataModel.Process
		return proc, ctx, nil
		}

		func getScenarioProcessById(id string, activeModel mod.Model) (dataModel.Process, *mod.NodeContext, error) {
		// Get app node
		node := activeModel.GetNodeById(id)
		if node == nil {
		return nil, nil, errors.New("Failed to get app node")
		}
		// Get App Process & context
		proc, ok := node.(*dataModel.Process)
		if !ok {
		return nil, nil, errors.New("Failed to cast node as Process")
		}
		ctx := activeModel.GetNodeContext(proc.Name)
		if ctx == nil {
		return nil, nil, errors.New("Missing node context for " + proc.Name)
		}
		return proc, ctx, nil
		}

		func applicationsPOST(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json; charset=UTF-8")
		log.Info("applicationsPOST")
		@@ -330,7 +289,7 @@ func applicationsPOST(w http.ResponseWriter, r *http.Request) {
		}

		// Create new App instance
		err = appCtrl.appStore.Set(convertApplicationInfoToApp(&appInfo))
		err = appCtrl.appStore.Set(convertApplicationInfoToApp(&appInfo), nil)
		if err != nil {
		log.Error(err.Error())
		http.Error(w, err.Error(), http.StatusInternalServerError)
		@@ -373,7 +332,7 @@ func applicationsAppInstanceIdPUT(w http.ResponseWriter, r *http.Request) {
		}

		// Override entry in DB
		err = appCtrl.appStore.Set(convertApplicationInfoToApp(&appInfo))
		err = appCtrl.appStore.Set(convertApplicationInfoToApp(&appInfo), nil)
		if err != nil {
		log.Error(err.Error())
		http.Error(w, err.Error(), http.StatusInternalServerError)
		@@ -414,7 +373,8 @@ func applicationsAppInstanceIdDELETE(w http.ResponseWriter, r *http.Request) {
		appId := vars["appInstanceId"]

		// Flush App instance data
		err := appCtrl.appStore.Del(appId)
		gracePeriod := defaultGracePeriod
		err := appCtrl.appStore.Del(appId, &gracePeriod)
		if err != nil {
		log.Error(err.Error())
		}
		@@ -644,7 +604,7 @@ func convertApplicationInfoToApp(appInfo dataModel.ApplicationInfo) apps.Appli
		return app
		}

		func appStoreUpdateCb(eventType string, eventData interface{}) {
		func appStoreUpdateCb(eventType string, eventData interface{}, userData interface{}) {
		var msg *mq.Msg

		// Create message to send on MQ
		@@ -655,6 +615,11 @@ func appStoreUpdateCb(eventType string, eventData interface{}) {
		case apps.EventRemove:
		msg = appCtrl.mqLocal.CreateMsg(mq.MsgAppRemove, mq.TargetAll, appCtrl.sandboxName)
		msg.Payload[mqFieldAppId] = eventData.(string)
		gracePeriodStr := "0"
		if gracePeriod, ok := userData.(*int); ok && gracePeriod != nil {
		gracePeriodStr = strconv.Itoa(*gracePeriod)
		}
		msg.Payload[mqFieldGracePeriod] = gracePeriodStr
		case apps.EventFlush:
		msg = appCtrl.mqLocal.CreateMsg(mq.MsgAppFlush, mq.TargetAll, appCtrl.sandboxName)
		msg.Payload[mqFieldPersist] = strconv.FormatBool(eventData.(bool))

go-apps/meep-sandbox-ctrl/server/sandbox-ctrl.go

+99 −18

Original line number	Diff line number	Diff line
		@@ -26,6 +26,7 @@ import (
		"sort"
		"strconv"
		"strings"
		"sync"
		"time"

		"github.com/gorilla/mux"
		@@ -60,6 +61,8 @@ type SandboxCtrl struct {
		replayMgr *replay.ReplayMgr
		pduSessionStore *pss.PduSessionStore
		sandboxStore *ss.SandboxStore
		gracefulRemoveMap map[string]chan bool
		mutex sync.Mutex
		}

		const scenarioDBName string = "scenarios"
		@@ -104,6 +107,7 @@ func Init() (err error) {

		// Create new Sandbox Controller
		sbxCtrl = new(SandboxCtrl)
		sbxCtrl.gracefulRemoveMap = make(map[string]chan bool)

		// Retrieve Sandbox name from environment variable
		sbxCtrl.sandboxName = strings.TrimSpace(os.Getenv("MEEP_SANDBOX_NAME"))
		@@ -1006,6 +1010,7 @@ func sendEventScenarioUpdate(event *dataModel.Event) (int, string, error) {

		// Get node name
		nodeName := getScenarioNodeName(event.EventScenarioUpdate.Node)
		isProc := mod.IsProc(event.EventScenarioUpdate.Node.Type_)

		// Perform necessary action on scenario
		switch event.EventScenarioUpdate.Action {
		@@ -1013,7 +1018,7 @@ func sendEventScenarioUpdate(event *dataModel.Event) (int, string, error) {
		err = sbxCtrl.activeModel.AddScenarioNode(event.EventScenarioUpdate.Node, nodeName)
		if err == nil {
		description = "Added node [" + nodeName + "]"
		if mod.IsProc(event.EventScenarioUpdate.Node.Type_) {
		if isProc {
		_ = setAppInstance(nodeName, sbxCtrl.activeModel)
		}
		}
		@@ -1021,18 +1026,20 @@ func sendEventScenarioUpdate(event *dataModel.Event) (int, string, error) {
		err = sbxCtrl.activeModel.ModifyScenarioNode(event.EventScenarioUpdate.Node, nodeName)
		if err == nil {
		description = "Modified node [" + nodeName + "]"
		if mod.IsProc(event.EventScenarioUpdate.Node.Type_) {
		if isProc {
		_ = setAppInstance(nodeName, sbxCtrl.activeModel)
		}
		}
		case mod.ScenarioRemove:
		nodeId := sbxCtrl.activeModel.GetNodeId(nodeName)
		err = sbxCtrl.activeModel.RemoveScenarioNode(event.EventScenarioUpdate.Node, nodeName)
		gracePeriod := int(event.EventScenarioUpdate.GracePeriod)
		if gracePeriod > 0 {
		err = removeNodeGracefully(nodeId, nodeName, event.EventScenarioUpdate.Node, gracePeriod)
		} else {
		err = removeNode(nodeId, nodeName, event.EventScenarioUpdate.Node)
		}
		if err == nil {
		description = "Removed node [" + nodeName + "]"
		if mod.IsProc(event.EventScenarioUpdate.Node.Type_) {
		_ = delAppInstance(nodeId)
		}
		}
		default:
		err = errors.New("Unsupported scenario update action: " + event.EventScenarioUpdate.Action)
		@@ -1608,6 +1615,80 @@ func activeScenarioUpdateCb(eventType string, userData interface{}) {
		}
		}

		func removeNode(nodeId string, nodeName string, node *dataModel.ScenarioNode) error {
		// Remove scenario node
		err := sbxCtrl.activeModel.RemoveScenarioNode(node, nodeName)
		if err != nil {
		log.Error(err.Error())
		return err
		}

		// Delete app instance
		if mod.IsProc(node.Type_) {
		_ = delAppInstance(nodeId, 0)
		}
		return nil
		}

		func removeNodeGracefully(nodeId string, nodeName string, node *dataModel.ScenarioNode, gracePeriod int) error {
		sbxCtrl.mutex.Lock()
		defer sbxCtrl.mutex.Unlock()

		if mod.IsProc(node.Type_) {
		// Make sure graceful remove is not already in progress
		if _, found := sbxCtrl.gracefulRemoveMap[nodeId]; found {
		return errors.New("Graceful Remove already in progress for nodeId: " + nodeId)
		}

		// Create Graceful remove channel
		gracefulRemoveChannel := make(chan bool)
		sbxCtrl.gracefulRemoveMap[nodeId] = gracefulRemoveChannel

		// Start goroutine to wait for app termination confirmation or timeout
		go func() {
		select {
		case <-gracefulRemoveChannel:
		case <-time.After(time.Duration(gracePeriod) * time.Second):
		sbxCtrl.mutex.Lock()
		delete(sbxCtrl.gracefulRemoveMap, nodeId)
		sbxCtrl.mutex.Unlock()
		}

		// Remove scenario node immediately
		err := sbxCtrl.activeModel.RemoveScenarioNode(node, nodeName)
		if err != nil {
		log.Error(err.Error())
		}
		}()

		// Trigger graceful app termination
		_ = delAppInstance(nodeId, gracePeriod)
		} else {
		// Nothing to wait for, remove scenario node immediately
		err := sbxCtrl.activeModel.RemoveScenarioNode(node, nodeName)
		if err != nil {
		log.Error(err.Error())
		return err
		}
		}

		return nil
		}

		func removeNodeConfirm(appId string) {
		sbxCtrl.mutex.Lock()
		defer sbxCtrl.mutex.Unlock()

		// Process removal confirmation (only if expected)
		if gracefulRemoveChannel, found := sbxCtrl.gracefulRemoveMap[appId]; found {
		gracefulRemoveChannel <- true
		delete(sbxCtrl.gracefulRemoveMap, appId)
		} else {
		log.Warn("Unexpected remove node confirmation")
		return
		}
		}

		func getActiveModel() *mod.Model {
		return sbxCtrl.activeModel
		}

go-packages/meep-applications/application-store.go

+10 −10

Original line number	Diff line number	Diff line
		@@ -57,7 +57,7 @@ type Application struct {
		type ApplicationStoreCfg struct {
		Name string
		Namespace string
		UpdateCb func(eventType string, eventData interface{})
		UpdateCb func(eventType string, eventData interface{}, userData interface{})
		RedisAddr string
		}

		@@ -65,7 +65,7 @@ type ApplicationStore struct {
		apps map[string]*Application
		rc *redis.Connector
		keyRoot string
		updateCb func(eventType string, eventData interface{})
		updateCb func(eventType string, eventData interface{}, userData interface{})
		mutex sync.Mutex
		}

		@@ -100,7 +100,7 @@ func NewApplicationStore(cfg ApplicationStoreCfg) (as ApplicationStore, err er
		}

		// Set - Create or update app entry in DB
		func (as ApplicationStore) Set(app Application) error {
		func (as ApplicationStore) Set(app Application, userData interface{}) error {
		// Validate application
		if app == nil {
		return errors.New("nil application")
		@@ -126,7 +126,7 @@ func (as ApplicationStore) Set(app Application) error {

		// Invoke application update callback
		if as.updateCb != nil {
		as.updateCb(EventAdd, app.Id)
		as.updateCb(EventAdd, app.Id, userData)
		}
		return nil
		}
		@@ -157,7 +157,7 @@ func (as ApplicationStore) GetAll() ([]Application, error) {
		}

		// Del - Remove application with provided id
		func (as *ApplicationStore) Del(id string) error {
		func (as *ApplicationStore) Del(id string, userData interface{}) error {
		// Delete entry
		err := as.deleteEntry(id)
		if err != nil {
		@@ -166,13 +166,13 @@ func (as *ApplicationStore) Del(id string) error {

		// Invoke application update callback
		if as.updateCb != nil {
		as.updateCb(EventRemove, id)
		as.updateCb(EventRemove, id, userData)
		}
		return nil
		}

		// FlushAll - Remove all Application Store entries
		func (as *ApplicationStore) FlushNonPersistent() {
		func (as *ApplicationStore) FlushNonPersistent(userData interface{}) {
		// Get app list
		appList, err := as.GetAll()
		if err != nil {
		@@ -190,19 +190,19 @@ func (as *ApplicationStore) FlushNonPersistent() {
		// Invoke application update callback
		if as.updateCb != nil {
		flushPersistent := false
		as.updateCb(EventFlush, flushPersistent)
		as.updateCb(EventFlush, flushPersistent, userData)
		}
		}

		// FlushAll - Remove all Application Store entries
		func (as *ApplicationStore) Flush() {
		func (as *ApplicationStore) Flush(userData interface{}) {
		// Delete all entries
		_ = as.deleteAllEntries()

		// Invoke application update callback
		if as.updateCb != nil {
		flushPersistent := true
		as.updateCb(EventFlush, flushPersistent)
		as.updateCb(EventFlush, flushPersistent, userData)
		}
		}