add integration test

Author: shijiesheng

test/activity_test.go

@@ -28,6 +28,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/mocktracer"
+
 	"go.uber.org/cadence"
 	"go.uber.org/cadence/activity"
 	"go.uber.org/cadence/worker"
@@ -37,6 +40,7 @@ type Activities struct {
 	mu          sync.Mutex
 	invocations []string
 	activities2 *Activities2
+	tracer      opentracing.Tracer
 }
 
 type Activities2 struct {
@@ -47,7 +51,7 @@ var errFailOnPurpose = cadence.NewCustomError("failing-on-purpose")
 
 func newActivities() *Activities {
 	activities2 := &Activities2{}
-	result := &Activities{activities2: activities2}
+	result := &Activities{activities2: activities2, tracer: mocktracer.New()}
 	activities2.impl = result
 	return result
 }
@@ -146,6 +150,13 @@ func (a *Activities) InspectActivityInfo(ctx context.Context, domain, taskList,
 	return nil
 }
 
+func (a *Activities) InspectActivitySpan(ctx context.Context) (map[string]string, error) {
+	span := opentracing.SpanFromContext(ctx)
+	carrier := make(map[string]string)
+	err := a.tracer.Inject(span.Context(), opentracing.TextMap, opentracing.TextMapCarrier(carrier))
+	return carrier, err
+}
+
 func (a *Activities) register(worker worker.Worker) {
 	// Kept to verify backward compatibility of activity registration.
 	activity.RegisterWithOptions(a, activity.RegisterOptions{Name: "Activities_", DisableAlreadyRegisteredCheck: true})
@@ -155,4 +166,5 @@ func (a *Activities) register(worker worker.Worker) {
 	worker.RegisterActivityWithOptions(a.activities2, activity.RegisterOptions{Name: "Prefix_", DisableAlreadyRegisteredCheck: true})
 	worker.RegisterActivityWithOptions(a.InspectActivityInfo, activity.RegisterOptions{Name: "inspectActivityInfo"})
 	worker.RegisterActivityWithOptions(a.HeartbeatAndSleep, activity.RegisterOptions{Name: "HeartbeatAndSleep", EnableAutoHeartbeat: true})
+	worker.RegisterActivityWithOptions(a.InspectActivitySpan, activity.RegisterOptions{Name: "inspectActivitySpan"})
 }

test/integration_test.go

@@ -38,6 +38,8 @@ import (
 	"go.uber.org/goleak"
 	"go.uber.org/zap/zaptest"
 
+	"github.com/opentracing/opentracing-go/mocktracer"
+
 	"go.uber.org/cadence"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/client"
@@ -98,7 +100,7 @@ func (ts *IntegrationTestSuite) SetupSuite() {
 	ts.Assertions = require.New(ts.T())
 	ts.config = newConfig()
 	ts.activities = newActivities()
-	ts.workflows = &Workflows{}
+	ts.workflows = &Workflows{tracer: mocktracer.New()}
 	ts.Nil(waitForTCP(time.Minute, ts.config.ServiceAddr))
 	var err error
 	if ts.config.EnableGrpcAdapter {
@@ -109,6 +111,7 @@ func (ts *IntegrationTestSuite) SetupSuite() {
 	ts.NoError(err)
 	ts.libClient = client.NewClient(ts.rpcClient.Interface, domainName,
 		&client.Options{
+			Tracer:             ts.workflows.tracer,
 			ContextPropagators: []workflow.ContextPropagator{NewStringMapPropagator([]string{testContextKey})},
 		})
 	ts.domainClient = client.NewDomainClient(ts.rpcClient.Interface, &client.Options{})
@@ -154,6 +157,7 @@ func (ts *IntegrationTestSuite) SetupTest() {
 
 func (ts *IntegrationTestSuite) BeforeTest(suiteName, testName string) {
 	options := worker.Options{
+		Tracer:                            ts.workflows.tracer,
 		DisableStickyExecution:            ts.config.IsStickyOff,
 		Logger:                            zaptest.NewLogger(ts.T()),
 		WorkflowInterceptorChainFactories: []interceptors.WorkflowInterceptorFactory{ts.tracer},
@@ -545,6 +549,13 @@ func (ts *IntegrationTestSuite) TestNonDeterministicWorkflowQuery() {
 	ts.NoError(value.Get(&trace))
 }
 
+func (ts *IntegrationTestSuite) TestOverrideSpanContext() {
+	var result map[string]string
+	err := ts.executeWorkflow("test-override-span-context", ts.workflows.OverrideSpanContext, &result)
+	ts.NoError(err)
+	ts.Equal("some-value", result["mockpfx-baggage-some-key"])
+}
+
 func (ts *IntegrationTestSuite) registerDomain() {
 	ctx, cancel := context.WithTimeout(context.Background(), ctxTimeout)
 	defer cancel()
@@ -587,10 +598,13 @@ func (ts *IntegrationTestSuite) executeWorkflowWithOption(
 	options client.StartWorkflowOptions, wfFunc interface{}, retValPtr interface{}, args ...interface{}) error {
 	ctx, cancel := context.WithTimeout(context.Background(), ctxTimeout)
 	defer cancel()
-	run, err := ts.libClient.ExecuteWorkflow(ctx, options, wfFunc, args...)
+	span := ts.workflows.tracer.StartSpan("test-workflow")
+	defer span.Finish()
+	execution, err := ts.libClient.StartWorkflow(ctx, options, wfFunc, args...)
 	if err != nil {
 		return err
 	}
+	run := ts.libClient.GetWorkflow(ctx, execution.ID, execution.RunID)
 	err = run.Get(ctx, retValPtr)
 	logger := zaptest.NewLogger(ts.T())
 	if ts.config.Debug {

test/workflow_test.go

@@ -27,6 +27,8 @@ import (
 	"math/rand"
 	"time"
 
+	"github.com/opentracing/opentracing-go"
+
 	"go.uber.org/cadence"
 	"go.uber.org/cadence/.gen/go/shared"
 	"go.uber.org/cadence/client"
@@ -41,6 +43,7 @@ const (
 
 type Workflows struct {
 	nonDeterminismSimulatorWorkflowCallCount int
+	tracer                                   opentracing.Tracer
 }
 
 func (w *Workflows) Basic(ctx workflow.Context) ([]string, error) {
@@ -617,6 +620,45 @@ func (w *Workflows) NonDeterminismSimulatorWorkflow(ctx workflow.Context) ([]str
 	return res, nil
 }
 
+func (w *Workflows) OverrideSpanContext(ctx workflow.Context) (map[string]string, error) {
+	type spanActivationResult struct {
+		Carrier map[string]string
+		Err     error
+	}
+	// start a short lived new workflow span within SideEffect to avoid duplicate span creation during replay
+	resultValue := workflow.SideEffect(ctx, func(ctx workflow.Context) interface{} {
+		wSpan := w.tracer.StartSpan("workflow-operation-with-new-span", opentracing.ChildOf(workflow.GetSpanContext(ctx)))
+		defer wSpan.Finish()
+		wSpan.SetBaggageItem("some-key", "some-value")
+		carrier := make(map[string]string)
+		err := w.tracer.Inject(wSpan.Context(), opentracing.TextMap, opentracing.TextMapCarrier(carrier))
+		return spanActivationResult{Carrier: carrier, Err: err}
+	})
+	var activationResult spanActivationResult
+	if err := resultValue.Get(&activationResult); err != nil {
+		return nil, fmt.Errorf("failed to decode span activation result: %v", err)
+	}
+	if activationResult.Err != nil {
+		return nil, fmt.Errorf("failed to activate new span: %v", activationResult.Err)
+	}
+	spanContext, err := w.tracer.Extract(opentracing.TextMap, opentracing.TextMapCarrier(activationResult.Carrier))
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract span context: %v", err)
+	}
+	ctx = workflow.WithSpanContext(ctx, spanContext)
+
+	opts := workflow.ActivityOptions{
+		ScheduleToStartTimeout: time.Second,
+		ScheduleToCloseTimeout: 10 * time.Second,
+		StartToCloseTimeout:    10 * time.Second,
+		HeartbeatTimeout:       2 * time.Second,
+	}
+	aCtx := workflow.WithActivityOptions(ctx, opts)
+	var res map[string]string
+	err = workflow.ExecuteActivity(aCtx, "inspectActivitySpan", "hello").Get(aCtx, &res)
+	return res, err
+}
+
 func (w *Workflows) register(worker worker.Worker) {
 	// Kept to verify backward compatibility of workflow registration.
 	workflow.RegisterWithOptions(w.Basic, workflow.RegisterOptions{DisableAlreadyRegisteredCheck: true})
@@ -646,6 +688,7 @@ func (w *Workflows) register(worker worker.Worker) {
 	worker.RegisterWorkflow(w.ConsistentQueryWorkflow)
 	worker.RegisterWorkflow(w.WorkflowWithLocalActivityCtxPropagation)
 	worker.RegisterWorkflow(w.NonDeterminismSimulatorWorkflow)
+	worker.RegisterWorkflow(w.OverrideSpanContext)
 
 }
 

workflow/context.go

@@ -102,21 +102,33 @@ func GetSpanContext(ctx Context) opentracing.SpanContext {
 //
 //	func goodWorkflow(ctx Context) (string, error) {
 //		// start a short lived new workflow span within SideEffect to avoid duplicate span creation during replay
-//		spanContextValue := SideEffect(ctx, func(ctx Context) interface{} {
-//			wSpan := opentracing.StartSpan("workflow-operation-with-new-span", opentracing.ChildOf(GetSpanContext(ctx)))
+//		type spanActivationResult struct {
+//			Carrier map[string]string // exported field so it's json encoded
+//			Err     error
+//		}
+//		resultValue := workflow.SideEffect(ctx, func(ctx workflow.Context) interface{} {
+//			wSpan := w.tracer.StartSpan("workflow-operation-with-new-span", opentracing.ChildOf(workflow.GetSpanContext(ctx)))
 //			defer wSpan.Finish()
-//			wSpan.SetTag("some-key", "some-value")
-//			return wSpan.Context()
+//			wSpan.SetBaggageItem("some-key", "some-value")
+//			carrier := make(map[string]string)
+//			err := w.tracer.Inject(wSpan.Context(), opentracing.TextMap, opentracing.TextMapCarrier(carrier))
+//			return spanActivationResult{Carrier: carrier, Err: err}
 //		})
-//		var spanContext opentracing.SpanContext
-//		err := spanContextValue.Get(&spanContext)
+//		var activationResult spanActivationResult
+//		if err := resultValue.Get(&activationResult); err != nil {
+//			return nil, fmt.Errorf("failed to decode span activation result: %v", err)
+//		}
+//		if activationResult.Err != nil {
+//			return nil, fmt.Errorf("failed to activate new span: %v", activationResult.Err)
+//		}
+//		spanContext, err := w.tracer.Extract(opentracing.TextMap, opentracing.TextMapCarrier(activationResult.Carrier))
 //		if err != nil {
-//			return "",fmt.Errorf("failed to get span context: %w", err)
+//			return nil, fmt.Errorf("failed to extract span context: %v", err)
 //		}
-//
-//		aCtx := WithSpanContext(ctx, spanContext)
+//		ctx = workflow.WithSpanContext(ctx, spanContext)
 //		var activityFooResult string
-//		err = ExecuteActivity(aCtx, activityFoo).Get(aCtx, &activityFooResult)
+//		aCtx := workflow.WithActivityOptions(ctx, opts)
+//		err = workflow.ExecuteActivity(aCtx, activityFoo).Get(aCtx, &activityFooResult)
 //		return activityFooResult, err
 //	}
 //
@@ -127,7 +139,8 @@ func GetSpanContext(ctx Context) opentracing.SpanContext {
 //		wSpan := opentracing.StartSpan("workflow-operation", opentracing.ChildOf(GetSpanContext(ctx)))
 //		wSpan.SetBaggageItem("some-key", "some-value")
 //		// pass the new span context to activity
-//		aCtx := WithSpanContext(ctx, wSpan.Context())
+//		ctx = WithSpanContext(ctx, wSpan.Context())
+//	    aCtx := workflow.WithActivityOptions(ctx, opts)
 //		var activityFooResult string
 //		err := ExecuteActivity(aCtx, activityFoo).Get(aCtx, &activityFooResult)
 //		wSpan.Finish()