[OCTRL-649][core] Corrections after PR review.

Author: claireguyot

core/environment/manager.go

@@ -82,16 +82,22 @@ func NewEnvManager(tm *task.Manager, incomingEventCh <-chan event.Event) *Manage
 
 				case *event.TasksReleasedEvent:
 					// If we got a TasksReleasedEvent, it must be matched with a pending
-					// environment teardown if the task is critical.
+					// environment teardown.
 					if thisEnvCh, ok := instance.pendingTeardownsCh[typedEvent.GetEnvironmentId()]; ok {
 						thisEnvCh <- typedEvent
 						close(thisEnvCh)
 						delete(instance.pendingTeardownsCh, typedEvent.GetEnvironmentId())
 					} else {
+						// If there is no pending environment teardown, it means that the released task stopped
+						// unexpectedly. In that case, the environment should get torn-down only if the task
+						// is critical.
 						var releaseCriticalTask = false
 						for _, v := range typedEvent.GetTaskIds() {
-							if tm.GetTask(v).GetTraits().Critical == true || tm.GetTask(v).GetParent().GetTaskTraits().Critical == true {
-								releaseCriticalTask = true
+							if tm.GetTask(v) != nil {
+								if tm.GetTask(v).GetTraits().Critical == true {
+									//|| tm.GetTask(v).GetParent().GetTaskTraits().Critical == true
+									releaseCriticalTask = true
+								}
 							}
 						}
 						if releaseCriticalTask {
@@ -103,14 +109,19 @@ func NewEnvManager(tm *task.Manager, incomingEventCh <-chan event.Event) *Manage
 
 				case *event.TasksStateChangedEvent:
 					// If we got a TasksStateChangedEvent, it must be matched with a pending
-					// environment transition if the task is critical.
+					// environment transition.
 					if thisEnvCh, ok := instance.pendingStateChangeCh[typedEvent.GetEnvironmentId()]; ok {
 						thisEnvCh <- typedEvent
 					} else {
+						// If there is no pending environment transition, it means that the changed task did so
+						// unexpectedly. In that case, the environment should transition only if the task
+						// is critical.
 						var changeCriticalTask = false
 						for _, v := range typedEvent.GetTaskIds() {
-							if tm.GetTask(v).GetTraits().Critical == true || tm.GetTask(v).GetParent().GetTaskTraits().Critical == true {
-								changeCriticalTask = true
+							if tm.GetTask(v) != nil {
+								if tm.GetTask(v).GetTraits().Critical == true {
+									changeCriticalTask = true
+								}
 							}
 						}
 						if changeCriticalTask {

core/task/errors.go

@@ -82,7 +82,7 @@ type TasksDeploymentError struct {
 }
 
 func (r TasksDeploymentError) Error() string {
-	return fmt.Sprintf("deployment failed for %d critical tasks [%s], and %d non-critical tasks [%s]", len(r.failedCriticalDescriptors), r.failedCriticalDescriptors.String(), len(r.failedNonCriticalDescriptors), r.failedNonCriticalDescriptors.String())
+	return fmt.Sprintf("deployment failed for %d critical tasks, and %d non-critical tasks; critical tasks: [%s]; non-critical tasks: [%s]", len(r.failedCriticalDescriptors), len(r.failedNonCriticalDescriptors), r.failedCriticalDescriptors.String(), r.failedNonCriticalDescriptors.String())
 }
 
 type TaskAlreadyReleasedError taskErrorBase

core/task/manager.go

@@ -690,10 +690,10 @@ func (m *Manager) configureTasks(envId uid.ID, tasks Tasks) error {
 
 		if len(taskNonCriticalErrors) > 0 {
 			log.WithField("partition", envId).
-				Warnf("non-critical task configuration failure, errors: %s", strings.Join(taskNonCriticalErrors, "; "))
+				Warnf("CONFIGURE could not complete for non-critical tasks, errors: %s", strings.Join(taskNonCriticalErrors, "; "))
 		}
 		if len(taskCriticalErrors) > 0 {
-			return fmt.Errorf("CONFIGURE could not complete, errors: %s", strings.Join(taskCriticalErrors, "; "))
+			return fmt.Errorf("CONFIGURE could not complete for critical tasks, errors: %s", strings.Join(taskCriticalErrors, "; "))
 		}
 		return nil
 	} else {
@@ -768,10 +768,10 @@ func (m *Manager) transitionTasks(envId uid.ID, tasks Tasks, src string, event s
 
 		if len(taskNonCriticalErrors) > 0 {
 			log.WithField("partition", envId).
-				Warnf("non-critical task transition failure, errors: %s", strings.Join(taskNonCriticalErrors, "; "))
+				Warnf("%s could not complete for non-critical tasks, errors: %s", event, strings.Join(taskNonCriticalErrors, "; "))
 		}
 		if len(taskCriticalErrors) > 0 {
-			return fmt.Errorf("transition could not complete, errors: %s", strings.Join(taskCriticalErrors, "; "))
+			return fmt.Errorf("%s could not complete for critical tasks, errors: %s", event, strings.Join(taskCriticalErrors, "; "))
 		}
 		return nil
 	} else {

core/workflow/safestate.go

@@ -34,33 +34,30 @@ type SafeState struct {
 	state task.State
 }
 
-func aggregateState(roles []Role, s task.State) (state task.State) {
+func aggregateState(roles []Role) (state task.State) {
+	state = task.INVARIANT
 	if len(roles) == 0 {
-		state = task.INVARIANT
 		return
 	}
-	state = s
-	if len(roles) > 1 {
-		for _, c := range roles[1:] {
-			taskR, isTaskRole := c.(*taskRole)
-			callR, isCallRole := c.(*callRole)
-			if isTaskRole {
-				if !taskR.Critical {
-					continue
-				}
-			} else if isCallRole {
-				if !callR.Critical {
-					continue
-				}
+	for _, c := range roles {
+		taskR, isTaskRole := c.(*taskRole)
+		callR, isCallRole := c.(*callRole)
+		if isTaskRole {
+			if !taskR.Critical {
+				continue
 			}
-			if state == task.MIXED {
-				return
+		} else if isCallRole {
+			if !callR.Critical {
+				continue
 			}
-			if state == task.ERROR {
-				return
-			}
-			state = state.X(c.GetState())
 		}
+		if state == task.MIXED {
+			return
+		}
+		if state == task.ERROR {
+			return
+		}
+		state = state.X(c.GetState())
 	}
 	return
 }
@@ -88,7 +85,7 @@ func (t *SafeState) merge(s task.State, r Role) {
 		return
 	default:
 		allRoles := r.GetRoles()
-		t.state = aggregateState(allRoles, s)
+		t.state = aggregateState(allRoles)
 	}
 }