better handling of panics when executing action plan concurrently

pkg/engine/apply/action/action_plan.go

@@ -1,6 +1,8 @@
 package action
 
 import (
+	"fmt"
+	"runtime/debug"
 	"sync"
 )
 
@@ -30,11 +32,21 @@ func (plan *Plan) GetActionGraphNode(key string) *GraphNode {
 
 // Apply applies the action plan. It may call fn in multiple go routines, executing the plan in parallel
 func (plan *Plan) Apply(fn ApplyFunction, resultUpdater ApplyResultUpdater) *ApplyResult {
+	// make sure we are converting panics into errors
+	fnModified := func(act Base) (errResult error) {
+		defer func() {
+			if err := recover(); err != nil {
+				errResult = fmt.Errorf("panic: %s\n%s", err, string(debug.Stack()))
+			}
+		}()
+		return fn(act)
+	}
+
 	// update total number of actions and start the revision
 	resultUpdater.SetTotal(plan.NumberOfActions())
 
 	// apply the plan and calculate result (success/failed/skipped actions)
-	plan.applyInternal(fn, resultUpdater)
+	plan.applyInternal(fnModified, resultUpdater)
 
 	// tell results updater that we are done and return the results
 	return resultUpdater.Done()

pkg/engine/apply/engine_apply.go

@@ -1,15 +1,13 @@
 package apply
 
 import (
-	"fmt"
 	"github.com/Aptomi/aptomi/pkg/engine/actual"
 	"github.com/Aptomi/aptomi/pkg/engine/apply/action"
 	"github.com/Aptomi/aptomi/pkg/engine/resolve"
 	"github.com/Aptomi/aptomi/pkg/event"
 	"github.com/Aptomi/aptomi/pkg/external"
 	"github.com/Aptomi/aptomi/pkg/lang"
 	"github.com/Aptomi/aptomi/pkg/plugin"
-	"runtime/debug"
 )
 
 // EngineApply executes actions to get from an actual state to desired state
@@ -70,7 +68,7 @@ func (apply *EngineApply) Apply(maxConcurrentActions int) (*resolve.PolicyResolu
 
 	// Note that the action plan will call function in different go routines by apply
 	result := apply.actionPlan.Apply(action.WrapParallelWithLimit(maxConcurrentActions, func(act action.Base) error {
-		err := apply.executeAction(act, context)
+		err := act.Apply(context)
 		if err != nil {
 			apply.eventLog.NewEntry().Errorf("error while applying action '%s': %s", act, err)
 		}
@@ -80,14 +78,3 @@ func (apply *EngineApply) Apply(maxConcurrentActions int) (*resolve.PolicyResolu
 	// No errors occurred
 	return apply.actualState, result
 }
-
-func (apply *EngineApply) executeAction(action action.Base, context *action.Context) (errResult error) {
-	// make sure we are converting panics into errors
-	defer func() {
-		if err := recover(); err != nil {
-			errResult = fmt.Errorf("panic: %s\n%s", err, string(debug.Stack()))
-		}
-	}()
-
-	return action.Apply(context)
-}