Repository: honeydipper/honeydipper
Pull request #520 (2023-02-28 00:47:01)

chore(release): release
### Highlights

* fix: commandHandler protecting the return labels
* feat: allow export on error




Patch:
diff --git a/internal/config/schema.go b/internal/config/schema.go
index 046b1a8..4f19a51 100644
--- a/internal/config/schema.go
+++ b/internal/config/schema.go
@@ -103,6 +103,7 @@ type Workflow struct {
 	Export          map[string]interface{}
 	ExportOnSuccess map[string]interface{} `json:"export_on_success" mapstructure:"export_on_success"`
 	ExportOnFailure map[string]interface{} `json:"export_on_failure" mapstructure:"export_on_failure"`
+	ExportOnError   map[string]interface{} `json:"export_on_error" mapstructure:"export_on_error"`
 	NoExport        []string               `json:"no_export" mapstructure:"no_export"`
 }
 
diff --git a/internal/workflow/continue.go b/internal/workflow/continue.go
index 68fd775..3d6da35 100644
--- a/internal/workflow/continue.go
+++ b/internal/workflow/continue.go
@@ -137,6 +137,8 @@ func (w *Session) processExport(msg *dipper.Message) {
 	}
 	if status != SessionStatusError {
 		w.postWorkflowExport(w.workflow.Export, envData)
+	} else {
+		w.postWorkflowExport(w.workflow.ExportOnError, envData)
 	}
 	if status == SessionStatusSuccess {
 		w.postWorkflowExport(w.workflow.ExportOnSuccess, envData)
diff --git a/internal/workflow/session.go b/internal/workflow/session.go
index e793353..7b408e9 100644
--- a/internal/workflow/session.go
+++ b/internal/workflow/session.go
@@ -320,6 +320,7 @@ func (w *Session) interpolateWorkflow(msg *dipper.Message) {
 	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
diff --git a/pkg/dipper/commandHandler.go b/pkg/dipper/commandHandler.go
index e8e96e8..fefbe11 100644
--- a/pkg/dipper/commandHandler.go
+++ b/pkg/dipper/commandHandler.go
@@ -89,6 +89,69 @@ func (p *CommandProvider) Return(call *Message, retval *Message) {
 	SendMessage(p.ReturnWriter, retMsg)
 }
 
+type commandWrapper struct {
+	msg      *Message
+	method   string
+	provider *CommandProvider
+	f        MessageHandler
+	retry    int
+	timeout  time.Duration
+	backoff  time.Duration
+}
+
+func (w *commandWrapper) attempt(replyChannel chan Message) {
+	w.msg.Reply = replyChannel
+	m := *w.msg
+	// make a copy of Labels so the function call won't
+	// tamper with the original one.
+	m.Labels = map[string]string{}
+	for k, v := range w.msg.Labels {
+		m.Labels[k] = v
+	}
+
+	go func() {
+		defer func() {
+			close(replyChannel)
+			replyChannel = nil
+		}()
+
+		Logger.Debugf("[operaotr] cmd labels %+v", m.Labels)
+
+		select {
+		case reply := <-replyChannel:
+			if _, ok := reply.Labels["no-timeout"]; ok {
+				reply = <-m.Reply
+			}
+
+			_, hasError := reply.Labels["error"]
+			if status, ok := reply.Labels["status"]; (hasError || (ok && status != SUCCESS)) && w.retry > 0 {
+				Logger.Debugf("[operaotr] %d retry left for method %s", w.retry, w.method)
+				w.retry--
+				time.Sleep(w.backoff * time.Millisecond)
+				w.backoff *= 2
+				go w.attempt(make(chan Message, 1))
+			} else {
+				w.provider.Return(w.msg, &reply)
+			}
+		case <-time.After(time.Second * w.timeout):
+			_ = w.provider.ReturnError(w.msg, "timeout")
+		}
+	}()
+
+	defer func() {
+		if r := recover(); r != nil && replyChannel != nil {
+			Logger.Warningf("Resuming after command error: %v", r)
+			Logger.Warning(errors.Wrap(r, 1).ErrorStack())
+			replyChannel <- Message{
+				Labels: map[string]string{
+					"error": fmt.Sprintf("%+v", r),
+				},
+			}
+		}
+	}()
+	w.f(&m)
+}
+
 // Router : route the message to rpc handlers.
 func (p *CommandProvider) Router(msg *Message) {
 	method := msg.Labels["method"]
@@ -98,56 +161,16 @@ func (p *CommandProvider) Router(msg *Message) {
 	}
 
 	retry, timeout, backoff := p.UnpackLabels(msg)
-
-	var attempt func(chan Message)
-	attempt = func(replyChannel chan Message) {
-		msg.Reply = replyChannel
-		m := *msg
-
-		go func() {
-			defer func() {
-				close(replyChannel)
-				replyChannel = nil
-			}()
-
-			Logger.Debugf("[operaotr] cmd labels %+v", m.Labels)
-
-			select {
-			case reply := <-replyChannel:
-				if _, ok := reply.Labels["no-timeout"]; ok {
-					reply = <-m.Reply
-				}
-
-				_, hasError := reply.Labels["error"]
-				if status, ok := reply.Labels["status"]; (hasError || (ok && status != SUCCESS)) && retry > 0 {
-					Logger.Debugf("[operaotr] %d retry left for method %s", retry, method)
-					retry--
-					time.Sleep(backoff * time.Millisecond)
-					backoff *= 2
-					go attempt(make(chan Message, 1))
-				} else {
-					p.Return(msg, &reply)
-				}
-			case <-time.After(time.Second * timeout):
-				_ = p.ReturnError(msg, "timeout")
-			}
-		}()
-
-		defer func() {
-			if r := recover(); r != nil && replyChannel != nil {
-				Logger.Warningf("Resuming after command error: %v", r)
-				Logger.Warning(errors.Wrap(r, 1).ErrorStack())
-				replyChannel <- Message{
-					Labels: map[string]string{
-						"error": fmt.Sprintf("%+v", r),
-					},
-				}
-			}
-		}()
-		f(&m)
+	w := &commandWrapper{
+		msg:      msg,
+		f:        f,
+		provider: p,
+		retry:    retry,
+		timeout:  timeout,
+		backoff:  backoff,
 	}
 
-	attempt(make(chan Message, 1))
+	w.attempt(make(chan Message, 1))
 }
 
 // UnpackLabels loads necessary variables out of the labels.


Test patch:
diff --git a/internal/workflow/session_context_test.go b/internal/workflow/session_context_test.go
index fc3bde4..4c29623 100644
--- a/internal/workflow/session_context_test.go
+++ b/internal/workflow/session_context_test.go
@@ -98,6 +98,17 @@ contexts:
 	exported = map[string]interface{}{"data1": "testdata"}
 	w.processNoExport(exported)
 	assert.Empty(t, exported, "remove all items from exported data")
+
+	w = s.newSession("", "uuid10", &config.Workflow{Name: "workflow1", ExportOnError: map[string]interface{}{"foo": "bar"}, Export: map[string]interface{}{"foo2": "bar2"}}).(*Session)
+	w.prepare(&dipper.Message{}, nil, map[string]interface{}{})
+	w.processExport(&dipper.Message{
+		Labels: map[string]string{
+			"status": "error",
+			"reason": "test",
+		},
+	})
+	assert.Equal(t, "bar", w.exported[0]["foo"], "export_on_error should be used if on error")
+	assert.NotContains(t, w.exported[0], "foo2", "export should not be used if on error")
 }
 
 var configStrWithEventContexts = `