Repository: honeydipper/honeydipper
Pull request #558 (2023-07-09 03:04:30)

feat: detach a child workflow to make it independent
<!--
Thanks for contributing!

See https://github.com/honeydipper/honeydipper/blob/master/CODE_OF_CONDUCT.md for information on our contributor code of conduct, and https://github.com/honeydipper/honeydipper/tree/master/docs for more information on developing on Honeydipper.

If possible, please follow these guidelines for commit messages:
https://github.com/angular/angular.js/blob/master/DEVELOPERS.md#-git-commit-guidelines

-->
#### Description
<!--
Add a description of your pull request.
-->
Making a child workflow independent so it can show up during get events API call. Also the child workflow will no longer followed by the parent workflow hooks.


#### This PR fixes the following issues
<!--
Replace this comment with fixed issues in the following format:
Fixes #123
Fixes #124
-->



Patch:
diff --git a/internal/config/schema.go b/internal/config/schema.go
index 4f11c5a..939b034 100644
--- a/internal/config/schema.go
+++ b/internal/config/schema.go
@@ -95,6 +95,7 @@ type Workflow struct {
 	Steps        []Workflow
 	Threads      []Workflow
 	Wait         string
+	Detach       bool
 
 	Switch  string
 	Cases   map[string]interface{}
diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go
index d6cab76..725d7f6 100644
--- a/internal/daemon/daemon.go
+++ b/internal/daemon/daemon.go
@@ -46,3 +46,12 @@ func Run(cfg *config.Config) {
 	OnStart()
 	cfg.Watch()
 }
+
+// Go launches a go routine with necessary prep and cleanup in the wrapper.
+func Go(f func()) {
+	Children.Add(1)
+	go func() {
+		defer Children.Done()
+		f()
+	}()
+}
diff --git a/internal/workflow/execute.go b/internal/workflow/execute.go
index 982182b..f6b1528 100644
--- a/internal/workflow/execute.go
+++ b/internal/workflow/execute.go
@@ -17,6 +17,12 @@ import (
 	"github.com/mitchellh/mapstructure"
 )
 
+var Success = &dipper.Message{
+	Labels: map[string]string{
+		"status": SessionStatusSuccess,
+	},
+}
+
 // execute is the entry point of the workflow.
 func (w *Session) execute(msg *dipper.Message) {
 	w.fireHook("on_session", msg)
@@ -67,11 +73,7 @@ func (w *Session) execute(msg *dipper.Message) {
 // noop continues the workflow as doing nothing.
 func (w *Session) noop(msg *dipper.Message) {
 	if msg.Labels["status"] != "success" {
-		msg = &dipper.Message{
-			Labels: map[string]string{
-				"status": SessionStatusSuccess,
-			},
-		}
+		msg = Success
 	}
 	if w.ID != "" {
 		w.continueExec(msg, nil)
@@ -262,8 +264,8 @@ func (w *Session) executeSwitch(msg *dipper.Message) {
 	w.noop(msg)
 }
 
-// executeAction takes actions for a single iteration in a single loop round.
-func (w *Session) executeAction(msg *dipper.Message) {
+// processActionHooks process the action hooks.
+func (w *Session) processActionHooks(msg *dipper.Message) bool {
 	switch {
 	case w.workflow.Workflow != "":
 		fallthrough
@@ -284,22 +286,38 @@ func (w *Session) executeAction(msg *dipper.Message) {
 		if w.currentHook == "" {
 			w.fireHook("on_action", msg)
 		}
-		if w.currentHook != "" {
-			return
-		}
 	}
 	// no action hooks if workflow is noop
 
+	return w.currentHook != ""
+}
+
+// executeAction takes actions for a single iteration in a single loop round.
+func (w *Session) executeAction(msg *dipper.Message) {
+	if w.processActionHooks(msg) { // hook in progress
+		return
+	}
+
 	switch {
 	case w.workflow.Workflow != "":
 		envData := w.buildEnvData(msg)
 		work := dipper.InterpolateStr(w.workflow.Workflow, envData)
 		w.performing = work
 		if !w.isHook && w.workflow.Name == "" {
-			w.ctx["_meta_name"] = work
+			w.ctx["_meta_name"] = "calling " + work
 		}
 		child := w.createChildSessionWithName(work, msg)
-		child.execute(msg)
+		if w.workflow.Detach {
+			child.parent = ""
+			delete(child.ctx, "resume_token")
+			daemon.Go(func() {
+				defer dipper.SafeExitOnError("Failed in execute detached workflow %+v", w.workflow.Workflow)
+				child.execute(msg)
+			})
+			w.continueExec(Success, nil)
+		} else {
+			child.execute(msg)
+		}
 	case w.isFunction():
 		w.performing = "function"
 		f := w.interpolateFunction(&w.workflow.Function, msg)
@@ -325,11 +343,7 @@ func (w *Session) executeAction(msg *dipper.Message) {
 		w.performing = "switch"
 		w.executeSwitch(msg)
 	default:
-		w.continueExec(&dipper.Message{
-			Labels: map[string]string{
-				"status": SessionStatusSuccess,
-			},
-		}, nil)
+		w.continueExec(Success, nil)
 	}
 }
 
diff --git a/internal/workflow/session.go b/internal/workflow/session.go
index 6b30533..6fcacc0 100644
--- a/internal/workflow/session.go
+++ b/internal/workflow/session.go
@@ -299,7 +299,7 @@ func (w *Session) injectLocalCTX(msg *dipper.Message) {
 func (w *Session) interpolateWorkflow(msg *dipper.Message) {
 	v := w.workflow
 	envData := w.buildEnvData(msg)
-	ret := config.Workflow{}
+	ret := *v
 
 	ret.Name = dipper.InterpolateStr(v.Name, envData)
 	ret.Description = dipper.InterpolateStr(v.Description, envData)
@@ -324,32 +324,33 @@ func (w *Session) interpolateWorkflow(msg *dipper.Message) {
 		ret.IterateParallel = []interface{}{}
 	}
 
-	ret.While = v.While                     // repeatedly interpolated later
-	ret.WhileAny = v.WhileAny               // repeatedly interpolated later
-	ret.WhileMatch = v.WhileMatch           // repeatedly interpolated later
-	ret.Until = v.Until                     // repeatedly interpolated later
-	ret.UntilAll = v.UntilAll               // repeatedly interpolated later
-	ret.UntilMatch = v.UntilMatch           // repeatedly interpolated later
-	ret.Else = v.Else                       // delayed
-	ret.Workflow = v.Workflow               // delayed
-	ret.Function = v.Function               // delayed
-	ret.Steps = v.Steps                     // delayed
-	ret.Threads = v.Threads                 // delayed
-	ret.Export = v.Export                   // delayed
-	ret.ExportOnSuccess = v.ExportOnSuccess // delayed
-	ret.ExportOnFailure = v.ExportOnFailure // delayed
-	ret.ExportOnError = v.ExportOnError     // delayed
-	ret.Switch = v.Switch                   // delayed
-	ret.Cases = v.Cases                     // delayed
-	ret.Default = v.Default                 // delayed
-
-	ret.Context = v.Context     // interpolated in initCTX
-	ret.Contexts = v.Contexts   // interpolated in initCTX
-	ret.NoExport = v.NoExport   // no interpolation
-	ret.IterateAs = v.IterateAs // no interpolation
-	ret.OnError = v.OnError     // no interpolation
-	ret.OnFailure = v.OnFailure // no interpolation
-	ret.Local = v.Local         // no interpolation
+	// ret.While = v.While                     // repeatedly interpolated later
+	// ret.WhileAny = v.WhileAny               // repeatedly interpolated later
+	// ret.WhileMatch = v.WhileMatch           // repeatedly interpolated later
+	// ret.Until = v.Until                     // repeatedly interpolated later
+	// ret.UntilAll = v.UntilAll               // repeatedly interpolated later
+	// ret.UntilMatch = v.UntilMatch           // repeatedly interpolated later
+	// ret.Else = v.Else                       // delayed
+	// ret.Workflow = v.Workflow               // delayed
+	// ret.Function = v.Function               // delayed
+	// ret.Steps = v.Steps                     // delayed
+	// ret.Threads = v.Threads                 // delayed
+	// ret.Export = v.Export                   // delayed
+	// ret.ExportOnSuccess = v.ExportOnSuccess // delayed
+	// ret.ExportOnFailure = v.ExportOnFailure // delayed
+	// ret.ExportOnError = v.ExportOnError     // delayed
+	// ret.Switch = v.Switch                   // delayed
+	// ret.Cases = v.Cases                     // delayed
+	// ret.Default = v.Default                 // delayed
+
+	// ret.Context = v.Context     // interpolated in initCTX
+	// ret.Contexts = v.Contexts   // interpolated in initCTX
+	// ret.NoExport = v.NoExport   // no interpolation
+	// ret.IterateAs = v.IterateAs // no interpolation
+	// ret.OnError = v.OnError     // no interpolation
+	// ret.OnFailure = v.OnFailure // no interpolation
+	// ret.Local = v.Local         // no interpolation
+	// ret.Detach = v.Detach       // no interpolation
 
 	w.workflow = &ret
 }
@@ -389,7 +390,7 @@ func (w *Session) setPerforming(performing string) string {
 	case wf.CallDriver != "":
 		w.performing = wf.CallDriver
 	case wf.Workflow != "":
-		w.performing = wf.Workflow
+		w.performing = "calling " + wf.Workflow
 	case w.isIteration():
 		w.performing = "iterating"
 	case w.isLoop():


Test patch:
diff --git a/internal/workflow/store_test.go b/internal/workflow/store_test.go
index 4469b5b..f3b32df 100644
--- a/internal/workflow/store_test.go
+++ b/internal/workflow/store_test.go
@@ -70,7 +70,7 @@ func TestNewSession(t *testing.T) {
 				Workflow: "test",
 			},
 			"asserts": func() {
-				assert.Equal(t, "test", session.performing)
+				assert.Equal(t, "calling test", session.performing)
 				assert.Equal(t, "0", session.parent)
 				session.save()
 				assert.Equal(t, "1", session.ID)