Add context propagation on child workflow and continue as new (#769)

* Add header propagation to continueasnew
* Add context propagation on ExecuteChildWorkflow
* Add some tests

Author: shreyassrivatsan

internal/error.go

@@ -232,6 +232,7 @@ func NewContinueAsNewError(ctx Context, wfn interface{}, args ...interface{}) *C
 		workflowOptions: *options,
 		workflowType:    workflowType,
 		input:           input,
+		header:          getWorkflowHeader(ctx, options.contextPropagators),
 	}
 	return &ContinueAsNewError{wfn: wfn, args: args, params: params}
 }

internal/error_test.go

@@ -405,7 +405,14 @@ func Test_ContinueAsNewError(t *testing.T) {
 		Name: continueAsNewWfName,
 	})
 
-	s := &WorkflowTestSuite{}
+	header := &shared.Header{
+		Fields: map[string][]byte{"test": []byte("test-data")},
+	}
+
+	s := &WorkflowTestSuite{
+		header:   header,
+		ctxProps: []ContextPropagator{NewStringMapPropagator([]string{"test"})},
+	}
 	wfEnv := s.NewTestWorkflowEnvironment()
 	wfEnv.ExecuteWorkflow(continueAsNewWorkflowFn, 101, "another random string")
 	err := wfEnv.GetWorkflowError()
@@ -422,4 +429,5 @@ func Test_ContinueAsNewError(t *testing.T) {
 	stringArg, ok := args[1].(string)
 	require.True(t, ok)
 	require.Equal(t, a2, stringArg)
+	require.Equal(t, header, continueAsNewErr.params.header)
 }

internal/internal_workflow_testsuite_test.go

@@ -55,6 +55,10 @@ func (s *WorkflowTestSuiteUnitTest) SetupSuite() {
 	s.localActivityOptions = LocalActivityOptions{
 		ScheduleToCloseTimeout: time.Second * 3,
 	}
+	s.header = &shared.Header{
+		Fields: map[string][]byte{"test": []byte("test-data")},
+	}
+	s.ctxProps = []ContextPropagator{NewStringMapPropagator([]string{"test"})}
 	RegisterWorkflowWithOptions(testWorkflowHello, RegisterWorkflowOptions{Name: "testWorkflowHello"})
 	RegisterWorkflow(testWorkflowContext)
 	RegisterWorkflow(testWorkflowHeartbeat)

internal/workflow.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"github.com/uber-go/tally"
+	s "go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/encoded"
 	"go.uber.org/cadence/internal/common"
 	"go.uber.org/zap"
@@ -572,6 +573,7 @@ func ExecuteChildWorkflow(ctx Context, childWorkflow interface{}, args ...interf
 		workflowOptions: *options,
 		input:           input,
 		workflowType:    wfType,
+		header:          getWorkflowHeader(ctx, options.contextPropagators),
 		scheduledTime:   Now(ctx), /* this is needed for test framework, and is not send to server */
 	}
 
@@ -615,6 +617,17 @@ func ExecuteChildWorkflow(ctx Context, childWorkflow interface{}, args ...interf
 	return result
 }
 
+func getWorkflowHeader(ctx Context, ctxProps []ContextPropagator) *s.Header {
+	header := &s.Header{
+		Fields: make(map[string][]byte),
+	}
+	writer := NewHeaderWriter(header)
+	for _, ctxProp := range ctxProps {
+		ctxProp.InjectFromWorkflow(ctx, writer)
+	}
+	return header
+}
+
 // WorkflowInfo information about currently executing workflow
 type WorkflowInfo struct {
 	WorkflowExecution                   WorkflowExecution