IWF-70: allow headers to be set in Interpreter Config

config/config.go

@@ -68,6 +68,7 @@ type (
 		// default is http://localhost:ApiConfig.Port
 		ApiServiceAddress                  string                              `json:"serviceAddress"`
 		DumpWorkflowInternalActivityConfig *DumpWorkflowInternalActivityConfig `json:"dumpWorkflowInternalActivityConfig"`
+		DefaultHeader                      map[string]string                   `json:"defaultHeader"`
 	}
 
 	DumpWorkflowInternalActivityConfig struct {

integ/config.go

@@ -7,16 +7,19 @@ import (
 const testWorkflowServerPort = "9714"
 const testIwfServerPort = "9715"
 
-func createTestConfig(failAtMemoCompatibility bool, optimizedVersioning *bool) config.Config {
+func createTestConfig(testCfg IwfServiceTestConfig) config.Config {
 	return config.Config{
 		Api: config.ApiConfig{
 			Port:                9715,
 			MaxWaitSeconds:      10, // use 10 so that we can test it in the waiting test
-			OptimizedVersioning: optimizedVersioning,
+			OptimizedVersioning: testCfg.OptimizedVersioning,
 		},
 		Interpreter: config.Interpreter{
 			VerboseDebug:              false,
-			FailAtMemoIncompatibility: failAtMemoCompatibility,
+			FailAtMemoIncompatibility: !testCfg.DisableFailAtMemoIncompatibility,
+			InterpreterActivityConfig: config.InterpreterActivityConfig{
+				DefaultHeader: testCfg.DefaultHeaders,
+			},
 		},
 	}
 }

integ/headers_test.go

@@ -0,0 +1,102 @@
+package integ
+
+import (
+	"context"
+	"github.com/indeedeng/iwf/integ/workflow/headers"
+	"github.com/indeedeng/iwf/service/common/ptr"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/indeedeng/iwf/gen/iwfidl"
+	"github.com/indeedeng/iwf/service"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestHeadersWorkflowTemporal(t *testing.T) {
+	if !*temporalIntegTest {
+		t.Skip()
+	}
+
+	for i := 0; i < *repeatIntegTest; i++ {
+		doTestWorkflowWithHeaders(t, service.BackendTypeTemporal, nil)
+		smallWaitForFastTest()
+
+		doTestWorkflowWithHeaders(t, service.BackendTypeTemporal, minimumContinueAsNewConfig(true))
+		smallWaitForFastTest()
+
+		doTestWorkflowWithHeaders(t, service.BackendTypeTemporal, &iwfidl.WorkflowConfig{
+			DisableSystemSearchAttribute: iwfidl.PtrBool(true),
+		})
+		smallWaitForFastTest()
+	}
+}
+
+func TestHeadersWorkflowCadence(t *testing.T) {
+	if !*cadenceIntegTest {
+		t.Skip()
+	}
+	for i := 0; i < *repeatIntegTest; i++ {
+		doTestWorkflowWithHeaders(t, service.BackendTypeCadence, nil)
+		smallWaitForFastTest()
+		doTestWorkflowWithHeaders(t, service.BackendTypeCadence, minimumContinueAsNewConfig(false))
+		smallWaitForFastTest()
+	}
+}
+
+func doTestWorkflowWithHeaders(t *testing.T, backendType service.BackendType, config *iwfidl.WorkflowConfig) {
+	// start test workflow server
+	wfHandler := headers.NewHandler()
+	closeFunc1 := startWorkflowWorker(wfHandler)
+	defer closeFunc1()
+
+	_, closeFunc2 := startIwfServiceByConfig(IwfServiceTestConfig{
+		BackendType: backendType,
+		DefaultHeaders: map[string]string{
+			headers.TestHeaderKey: headers.TestHeaderValue,
+		},
+	})
+	defer closeFunc2()
+
+	// start a workflow
+	apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
+		Servers: []iwfidl.ServerConfiguration{
+			{
+				URL: "http://localhost:" + testIwfServerPort,
+			},
+		},
+	})
+	wfId := headers.WorkflowType + strconv.Itoa(int(time.Now().UnixNano()))
+	wfInput := &iwfidl.EncodedObject{
+		Encoding: iwfidl.PtrString("json"),
+		Data:     iwfidl.PtrString("test data"),
+	}
+	req := apiClient.DefaultApi.ApiV1WorkflowStartPost(context.Background())
+	startReq := iwfidl.WorkflowStartRequest{
+		WorkflowId:             wfId,
+		IwfWorkflowType:        headers.WorkflowType,
+		WorkflowTimeoutSeconds: 100,
+		IwfWorkerUrl:           "http://localhost:" + testWorkflowServerPort,
+		StartStateId:           ptr.Any(headers.State1),
+		StateInput:             wfInput,
+		WorkflowStartOptions: &iwfidl.WorkflowStartOptions{
+			WorkflowConfigOverride: config,
+		},
+	}
+	_, httpResp, err := req.WorkflowStartRequest(startReq).Execute()
+	panicAtHttpError(err, httpResp)
+
+	req2 := apiClient.DefaultApi.ApiV1WorkflowGetWithWaitPost(context.Background())
+	_, httpResp2, err2 := req2.WorkflowGetRequest(iwfidl.WorkflowGetRequest{
+		WorkflowId: wfId,
+	}).Execute()
+	panicAtHttpError(err2, httpResp2)
+
+	assertions := assert.New(t)
+
+	history, _ := wfHandler.GetTestResult()
+	assertions.Equalf(map[string]int64{
+		"S1_start":  1,
+		"S1_decide": 1,
+	}, history, "headers test fail, %v", history)
+}

integ/util.go

@@ -67,6 +67,7 @@ type IwfServiceTestConfig struct {
 	MemoEncryption                   bool
 	DisableFailAtMemoIncompatibility bool // default to false so that we will fail at test
 	OptimizedVersioning              *bool
+	DefaultHeaders                   map[string]string
 }
 
 func startIwfService(backendType service.BackendType) (closeFunc func()) {
@@ -98,7 +99,6 @@ func startIwfServiceWithClient(backendType service.BackendType) (uclient uclient
 //var integTemporalUclientCached api.UnifiedClient
 
 func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.UnifiedClient, closeFunc func()) {
-	failAtMemoIncompatibility := !config.DisableFailAtMemoIncompatibility
 	if config.BackendType == service.BackendTypeTemporal {
 		dataConverter := converter.GetDefaultDataConverter()
 		if config.MemoEncryption {
@@ -111,7 +111,7 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 			panic(err)
 		}
 		uclient = temporalapi.NewTemporalClient(temporalClient, testNamespace, dataConverter, config.MemoEncryption)
-		iwfService := api.NewService(createTestConfig(failAtMemoIncompatibility, config.OptimizedVersioning), uclient, logger)
+		iwfService := api.NewService(createTestConfig(config), uclient, logger)
 		iwfServer := &http.Server{
 			Addr:    ":" + testIwfServerPort,
 			Handler: iwfService,
@@ -123,7 +123,7 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 		}()
 
 		// start iwf interpreter worker
-		interpreter := temporal.NewInterpreterWorker(createTestConfig(failAtMemoIncompatibility, config.OptimizedVersioning), temporalClient, service.TaskQueue, config.MemoEncryption, dataConverter, uclient)
+		interpreter := temporal.NewInterpreterWorker(createTestConfig(config), temporalClient, service.TaskQueue, config.MemoEncryption, dataConverter, uclient)
 		interpreter.Start()
 		return uclient, func() {
 			iwfServer.Close()
@@ -142,7 +142,7 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 			panic(err)
 		}
 		uclient = cadenceapi.NewCadenceClient(iwf.DefaultCadenceDomain, cadenceClient, serviceClient, encoded.GetDefaultDataConverter(), closeFunc)
-		iwfService := api.NewService(createTestConfig(failAtMemoIncompatibility, config.OptimizedVersioning), uclient, logger)
+		iwfService := api.NewService(createTestConfig(config), uclient, logger)
 		iwfServer := &http.Server{
 			Addr:    ":" + testIwfServerPort,
 			Handler: iwfService,
@@ -154,7 +154,7 @@ func doStartIwfServiceWithClient(config IwfServiceTestConfig) (uclient uclient.U
 		}()
 
 		// start iwf interpreter worker
-		interpreter := cadence.NewInterpreterWorker(createTestConfig(failAtMemoIncompatibility, config.OptimizedVersioning), serviceClient, iwf.DefaultCadenceDomain, service.TaskQueue, closeFunc, uclient)
+		interpreter := cadence.NewInterpreterWorker(createTestConfig(config), serviceClient, iwf.DefaultCadenceDomain, service.TaskQueue, closeFunc, uclient)
 		interpreter.Start()
 		return uclient, func() {
 			iwfServer.Close()

integ/workflow/headers/config.go

@@ -0,0 +1,4 @@
+package headers
+
+const TestHeaderKey = "integration-test-header"
+const TestHeaderValue = "integration-test-value"

integ/workflow/headers/routers.go

@@ -0,0 +1,92 @@
+package headers
+
+import (
+	"github.com/gin-gonic/gin"
+	"github.com/indeedeng/iwf/gen/iwfidl"
+	"github.com/indeedeng/iwf/service"
+	"log"
+	"net/http"
+)
+
+const (
+	WorkflowType = "headers"
+	State1       = "S1"
+)
+
+type handler struct {
+	invokeHistory map[string]int64
+}
+
+func NewHandler() *handler {
+	return &handler{
+		invokeHistory: make(map[string]int64),
+	}
+}
+
+// ApiV1WorkflowStartPost - for a workflow
+func (h *handler) ApiV1WorkflowStateStart(c *gin.Context) {
+	headerValue := c.GetHeader(TestHeaderKey)
+	if headerValue != TestHeaderValue {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "test header not found"})
+	}
+
+	var req iwfidl.WorkflowStateStartRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	log.Println("received state start request, ", req)
+
+	if req.GetWorkflowType() == WorkflowType {
+		// basic workflow go straight to decide methods without any commands
+		if req.GetWorkflowStateId() == State1 {
+			h.invokeHistory[req.GetWorkflowStateId()+"_start"]++
+			c.JSON(http.StatusOK, iwfidl.WorkflowStateStartResponse{
+				CommandRequest: &iwfidl.CommandRequest{
+					DeciderTriggerType: iwfidl.ALL_COMMAND_COMPLETED.Ptr(),
+				},
+			})
+			return
+		}
+	}
+
+	c.JSON(http.StatusBadRequest, struct{}{})
+}
+
+func (h *handler) ApiV1WorkflowStateDecide(c *gin.Context) {
+	headerValue := c.GetHeader(TestHeaderKey)
+	if headerValue != TestHeaderValue {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "test header not found"})
+	}
+
+	var req iwfidl.WorkflowStateDecideRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	log.Println("received state decide request, ", req)
+
+	if req.GetWorkflowType() == WorkflowType {
+		h.invokeHistory[req.GetWorkflowStateId()+"_decide"]++
+		if req.GetWorkflowStateId() == State1 {
+			// go to S2
+			c.JSON(http.StatusOK, iwfidl.WorkflowStateDecideResponse{
+				StateDecision: &iwfidl.StateDecision{
+					NextStates: []iwfidl.StateMovement{
+						{
+							StateId:    service.GracefulCompletingWorkflowStateId,
+							StateInput: req.StateInput,
+						},
+					},
+				},
+			})
+			return
+		}
+	}
+
+	c.JSON(http.StatusBadRequest, struct{}{})
+}
+
+func (h *handler) GetTestResult() (map[string]int64, map[string]interface{}) {
+	return h.invokeHistory, nil
+}

service/interpreter/activityImpl.go

@@ -31,7 +31,9 @@ func StateApiWaitUntil(
 	logger.Info("StateStartActivity", "input", input)
 	iwfWorkerBaseUrl := urlautofix.FixWorkerUrl(input.IwfWorkerUrl)
 
+	svcCfg := env.GetSharedConfig()
 	apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
+		DefaultHeader: svcCfg.Interpreter.InterpreterActivityConfig.DefaultHeader,
 		Servers: []iwfidl.ServerConfiguration{
 			{
 				URL: iwfWorkerBaseUrl,
@@ -86,7 +88,9 @@ func StateApiExecute(
 	logger.Info("StateDecideActivity", "input", input)
 
 	iwfWorkerBaseUrl := urlautofix.FixWorkerUrl(input.IwfWorkerUrl)
+	svcCfg := env.GetSharedConfig()
 	apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
+		DefaultHeader: svcCfg.Interpreter.InterpreterActivityConfig.DefaultHeader,
 		Servers: []iwfidl.ServerConfiguration{
 			{
 				URL: iwfWorkerBaseUrl,
@@ -272,8 +276,9 @@ func DumpWorkflowInternal(
 	logger.Info("DumpWorkflowInternal", "input", req)
 
 	apiAddress := config.GetApiServiceAddressWithDefault(env.GetSharedConfig())
-
+	svcCfg := env.GetSharedConfig()
 	apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
+		DefaultHeader: svcCfg.Interpreter.InterpreterActivityConfig.DefaultHeader,
 		Servers: []iwfidl.ServerConfiguration{
 			{
 				URL: apiAddress,