[core] New DEPLOY transition in env state machine

Author: teo

coconut/protos/o2control.pb.go

@@ -114,13 +114,14 @@ func newEnvironment(userVars map[string]string) (env *Environment, err error) {
 	env.Sm = fsm.NewFSM(
 		"STANDBY",
 		fsm.Events{
-			{Name: "CONFIGURE",      Src: []string{"STANDBY"},                   Dst: "CONFIGURED"},
-			{Name: "RESET",          Src: []string{"CONFIGURED"},                Dst: "STANDBY"},
+			{Name: "DEPLOY",         Src: []string{"STANDBY"},                   Dst: "DEPLOYED"},
+			{Name: "CONFIGURE",      Src: []string{"DEPLOYED"},                  Dst: "CONFIGURED"},
+			{Name: "RESET",          Src: []string{"CONFIGURED"},                Dst: "DEPLOYED"},
 			{Name: "START_ACTIVITY", Src: []string{"CONFIGURED"},                Dst: "RUNNING"},
 			{Name: "STOP_ACTIVITY",  Src: []string{"RUNNING"},                   Dst: "CONFIGURED"},
-			{Name: "EXIT",           Src: []string{"CONFIGURED", "STANDBY"},     Dst: "DONE"},
-			{Name: "GO_ERROR",       Src: []string{"CONFIGURED", "RUNNING"},     Dst: "ERROR"},
-			{Name: "RECOVER",        Src: []string{"ERROR"},                     Dst: "STANDBY"},
+			{Name: "EXIT",           Src: []string{"CONFIGURED", "DEPLOYED", "STANDBY"},     Dst: "DONE"},
+			{Name: "GO_ERROR",       Src: []string{"CONFIGURED", "DEPLOYED", "RUNNING"},     Dst: "ERROR"},
+			{Name: "RECOVER",        Src: []string{"ERROR"},                     Dst: "DEPLOYED"},
 		},
 		fsm.Callbacks{
 			"before_event": func(e *fsm.Event) {

core/environment/environment.go

@@ -133,11 +133,13 @@ func (envs *Manager) CreateEnvironment(workflowPath string, userVars map[string]
 	envs.m[env.id] = env
 	envs.pendingStateChangeCh[env.id] = env.stateChangedCh
 
-	err = env.TryTransition(NewConfigureTransition(
+	err = env.TryTransition(NewDeployTransition(
 		envs.taskman,
 		nil, //roles,
-		nil,
-		true),
+		nil),
+	)
+	err = env.TryTransition(NewConfigureTransition(
+		envs.taskman),
 	)
 	envs.mu.Unlock()
 
@@ -457,11 +459,13 @@ func (envs *Manager) CreateAutoEnvironment(workflowPath string, userVars map[str
 	envs.pendingStateChangeCh[env.id] = env.stateChangedCh
 	envs.mu.Unlock()
 
-	err = env.TryTransition(NewConfigureTransition(
+	err = env.TryTransition(NewDeployTransition(
 		envs.taskman,
 		nil, //roles,
-		nil,
-		true	))
+		nil))
+	err = env.TryTransition(NewConfigureTransition(
+		envs.taskman))
+
 	if err != nil {
 		envState := env.CurrentState()
 		env.sendEnvironmentEvent(&event.EnvironmentEvent{EnvironmentID: env.Id().String(), Error: err})

core/environment/manager.go

@@ -27,6 +27,7 @@ package environment
 
 import (
 	"errors"
+
 	"github.com/AliceO2Group/Control/core/protos"
 	"github.com/AliceO2Group/Control/core/task"
 )
@@ -39,8 +40,10 @@ type Transition interface {
 
 func MakeTransition(taskman *task.Manager, optype pb.ControlEnvironmentRequest_Optype) Transition {
 	switch optype {
+	case pb.ControlEnvironmentRequest_DEPLOY:
+		return NewDeployTransition(taskman, nil, nil)
 	case pb.ControlEnvironmentRequest_CONFIGURE:
-		return NewConfigureTransition(taskman, nil, nil, true)
+		return NewConfigureTransition(taskman)
 	case pb.ControlEnvironmentRequest_START_ACTIVITY:
 		return NewStartActivityTransition(taskman)
 	case pb.ControlEnvironmentRequest_STOP_ACTIVITY:

core/environment/transition.go

@@ -26,34 +26,23 @@ package environment
 
 import (
 	"errors"
-	"fmt"
-	"time"
 
 	"github.com/AliceO2Group/Control/common/event"
 	"github.com/AliceO2Group/Control/core/task"
 	"github.com/AliceO2Group/Control/core/task/taskop"
-	"github.com/AliceO2Group/Control/core/workflow"
-	"github.com/pborman/uuid"
-	"github.com/sirupsen/logrus"
 )
 
-func NewConfigureTransition(taskman *task.Manager, addRoles []string, removeRoles []string, reconfigureAll bool) Transition {
+func NewConfigureTransition(taskman *task.Manager) Transition {
 	return &ConfigureTransition{
 		baseTransition: baseTransition{
 			name: "CONFIGURE",
 			taskman: taskman,
 		},
-		addRoles: addRoles,
-		removeRoles: removeRoles,
-		reconfigureAll: reconfigureAll,
 	}
 }
 
 type ConfigureTransition struct {
 	baseTransition
-	addRoles		[]string
-	removeRoles		[]string
-	reconfigureAll	bool
 }
 
 func (t ConfigureTransition) do(env *Environment) (err error) {
@@ -63,161 +52,6 @@ func (t ConfigureTransition) do(env *Environment) (err error) {
 
 	wf := env.Workflow()
 
-
-	// Role tree operations go here, and afterwards we'll generally get a role tree which
-	// has
-	// - some TaskRoles already deployed with Tasks
-	// - some TaskRoles with no Tasks but with matching Tasks in the roster
-	// - some TaskRoles with no Tasks and no matching running Tasks in the roster
-
-/*
-	// First we free the relevant roles, if any
-	if len(t.removeRoles) != 0 {
-		rolesThatStay := env.roles[:0]
-		rolesToRelease := make([]string, 0)
-
-		for _, role := range env.roles {
-			for _, removeRole := range t.removeRoles {
-				if role == removeRole {
-					rolesToRelease = append(rolesToRelease, role)
-					break
-				}
-				rolesThatStay = append(rolesThatStay, role)
-			}
-		}
-		err = t.roleman.ReleaseRoles(env.Id().Array(), rolesToRelease)
-		if err != nil {
-			return
-		}
-		env.roles = rolesThatStay
-	}
-	// IDEA: instead of passing around m.state or roleman, pass around one or more of
-	// roleman's channels. This way roleman could potentially be lockless, and we just pipe
-	// him a list of rolenames to remove/add, or even a function or a struct that does so.
-	// This struct would implement an interface of the type of his channel, and he could
-	// use type assertion to check whether he needs to add, remove or do something else.
-
-	// Alright, so now we have freed some roles (if required).
-	// We proceed by deduplicating and attempting an acquire.
-	if len(t.addRoles) != 0 {
-		rolesToAcquire := make([]string, 0)
-
-		for _, addRole := range t.addRoles {
-			alreadyInEnv := false
-			for _, role := range env.roles {
-				if role == addRole {
-					alreadyInEnv = true
-					break
-				}
-			}
-			if !alreadyInEnv {
-				rolesToAcquire = append(rolesToAcquire, addRole)
-			}
-		}
-		err = t.roleman.AcquireRoles(env.Id().Array(), rolesToAcquire)
-		if err != nil {
-			return
-		}
-
-		// We complete a move to CONFIGURED for all roles and we're done.
-		err = t.roleman.ConfigureRoles(env.Id().Array(), rolesToAcquire)
-		if err != nil {
-			return
-		}
-
-		env.roles = append(env.roles, rolesToAcquire...)
-	}
-
-	// Finally, we configure.
-	if t.reconfigureAll {
-		err = t.roleman.ConfigureRoles(env.Id().Array(), env.roles)
-		if err != nil {
-			return
-		}
-	}
-
-	return*/
-
-	notifyStatus := make(chan task.Status)
-	subscriptionId := uuid.NewUUID().String()
-	env.wfAdapter.SubscribeToStatusChange(subscriptionId, notifyStatus)
-	defer env.wfAdapter.UnsubscribeFromStatusChange(subscriptionId)
-
-	// listen to workflow State changes
-	notifyState := make(chan task.State)
-	env.wfAdapter.SubscribeToStateChange(subscriptionId, notifyState)
-	defer env.wfAdapter.UnsubscribeFromStateChange(subscriptionId)
-
-	taskDescriptors := wf.GenerateTaskDescriptors()
-	if len(taskDescriptors) != 0 {
-		// err = t.taskman.AcquireTasks(env.Id().Array(), taskDescriptors)
-		taskmanMessage := task.NewEnvironmentMessage(taskop.AcquireTasks, env.Id(), nil, taskDescriptors)
-		t.taskman.MessageChannel <- taskmanMessage
-	}
-	if err != nil {
-		return
-	}
-
-	// We set all callRoles to ACTIVE right now, because there's no task activation for them.
-	// This is the callRole equivalent of AcquireTasks, which only pushes updates to taskRoles.
-	allHooks := wf.GetHooksForTrigger("")	// no trigger = all hooks
-	callHooks := allHooks.FilterCalls()							// get the calls
-	if len(callHooks) > 0 {
-		for _, h := range callHooks {
-			pr, ok := h.GetParentRole().(workflow.PublicUpdatable)
-			if !ok {
-				continue
-			}
-			go pr.UpdateStatus(task.ACTIVE)
-		}
-	}
-
-	deploymentTimeout := 90 * time.Second
-	wfStatus := wf.GetStatus()
-	if wfStatus != task.ACTIVE {
-		WORKFLOW_ACTIVE_LOOP:
-		for {
-			log.Debug("waiting for workflow to become active")
-			select {
-			case wfStatus = <-notifyStatus:
-				log.WithField("status", wfStatus.String()).
-				    Debug("workflow status change")
-				if wfStatus == task.ACTIVE {
-					break WORKFLOW_ACTIVE_LOOP
-				}
-				continue
-			case <-time.After(deploymentTimeout):
-				err = errors.New(fmt.Sprintf("workflow deployment timed out. timeout: %s",deploymentTimeout.String()))
-				break WORKFLOW_ACTIVE_LOOP
-			// This is needed for when the workflow fails during the STAGING state(mesos status),mesos responds with the `REASON_COMMAND_EXECUTOR_FAILED`,
-			// By listening to workflow state ERROR we can break the loop before reaching the timeout (1m30s), we can trigger the cleanup faster
-			// in the CreateEnvironment (environment/manager.go) and the lock in the `envman` is reserved for a sorter period, which allows operations like
-			// `environment list` to be done almost immediatelly after mesos informs with TASK_FAILED.
-			case wfState := <-notifyState:
-				if wfState == task.ERROR {
-					workflow.LeafWalk(wf, func(role workflow.Role) {
-						if st := role.GetState();  st == task.ERROR {
-							log.WithField("state", st).
-								WithField("role", role.GetPath()).
-								WithField("environment", role.GetEnvironmentId().String()).
-								Error("environment reached invalid state")
-						}
-					})
-					log.WithField("state", wfState.String()).
-				    	Debug("workflow state change")
-					err = errors.New("workflow deployment failed, aborting and cleaning up")
-					break WORKFLOW_ACTIVE_LOOP
-				}
-			}
-		}
-	}
-
-	if err != nil {
-		log.WithFields(logrus.Fields{"error": err.Error()}).
-			Error("workflow deployment error")
-		return
-	}
-
 	tasks := wf.GetTasks()
 
 	if len(tasks) != 0 {