Enable open tracing propagation in workflow lifecycles (#876)

How it works
Context Propagation in Cadence (Customer)

On start workflow API, trace span with context is written into workflow start event attributes, which is persisted in cadence server side.
On workflow-start in client, this span is referenced and activated on execute workflow.
On scheduling child workflows and activities (including local activities), the span is written into child workflow's workflow start event attributes and activity's schedule activity event attributes.
On processing activities/childworkflows, the persisted span is referenced and activated again.

Sample Spans

Notes: Poll + Respond apis spans are omitted here

{traceId:1, spanId:2, parentId:0, operationName:"cadence-RegisterDomain"}
{traceId:1, spanId:3, parentId:2, operationName:"Test Started"}
{traceId:1, spanId:18, parentId:3, operationName:"cadence-StartWorkflowExecution"}
{traceId:1, spanId:19, parentId:18, operationName:"cadence-GetWorkflowExecutionHistory"}
{traceId:1, spanId:21, parentId:18, operationName:"cadence-ExecuteWorkflow"}
{traceId:1, spanId:24, parentId:21, operationName:"cadence-ExecuteActivity"}
{traceId:1, spanId:25, parentId:24, operationName:"cadence-RespondActivityTaskCompleted"}
{traceId:1, spanId:31, parentId:21, operationName:"cadence-ExecuteWorkflow"}
{traceId:1, spanId:32, parentId:31, operationName:"cadence-ExecuteLocalActivity"}

What changed?

added an Propagator entity with tracing extract/inject logic
added trace activation logic in activity and workflow executors
added trace activation on service client (Tchannel + GRPC)
Why?

improve observability

How did you test it?

integration test

Author: shijiesheng

build.gradle

@@ -81,6 +81,7 @@ dependencies {
     testCompile group: 'junit', name: 'junit', version: '4.12'
     testCompile group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'
     testCompile group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'
+    testImplementation 'io.opentracing:opentracing-mock:0.33.0'
 }
 
 license {
@@ -364,4 +365,3 @@ jacocoTestReport {
         }))
     }
 }
-

src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/GrpcServiceStubs.java

@@ -16,6 +16,8 @@
 package com.uber.cadence.internal.compatibility.proto.serviceclient;
 
 import com.google.common.base.Strings;
+import com.google.protobuf.ByteString;
+import com.uber.cadence.api.v1.*;
 import com.uber.cadence.api.v1.DomainAPIGrpc;
 import com.uber.cadence.api.v1.MetaAPIGrpc;
 import com.uber.cadence.api.v1.MetaAPIGrpc.MetaAPIBlockingStub;
@@ -30,6 +32,7 @@
 import com.uber.cadence.api.v1.WorkflowAPIGrpc.WorkflowAPIBlockingStub;
 import com.uber.cadence.api.v1.WorkflowAPIGrpc.WorkflowAPIFutureStub;
 import com.uber.cadence.internal.Version;
+import com.uber.cadence.internal.tracing.TracingPropagator;
 import com.uber.cadence.serviceclient.ClientOptions;
 import io.grpc.CallOptions;
 import io.grpc.Channel;
@@ -48,6 +51,11 @@
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.propagation.TextMapPropagator;
 import io.opentelemetry.context.propagation.TextMapSetter;
+import io.opentracing.Span;
+import io.opentracing.Tracer;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.slf4j.Logger;
@@ -116,7 +124,8 @@ final class GrpcServiceStubs implements IGrpcServiceStubs {
             channel,
             deadlineInterceptor,
             MetadataUtils.newAttachHeadersInterceptor(headers),
-            newOpenTelemetryInterceptor());
+            newOpenTelemetryInterceptor(),
+            newOpenTracingInterceptor(options.getTracer()));
     if (log.isTraceEnabled()) {
       interceptedChannel = ClientInterceptors.intercept(interceptedChannel, tracingInterceptor);
     }
@@ -162,6 +171,80 @@ public void start(Listener<RespT> responseListener, Metadata headers) {
     };
   }
 
+  private ClientInterceptor newOpenTracingInterceptor(Tracer tracer) {
+    return new ClientInterceptor() {
+      private final TracingPropagator tracingPropagator = new TracingPropagator(tracer);
+      private final String OPERATIONFORMAT = "cadence-%s";
+
+      @Override
+      public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+          MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+        return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
+            next.newCall(method, callOptions)) {
+
+          @Override
+          public void start(Listener<RespT> responseListener, Metadata headers) {
+            Span span =
+                tracingPropagator.activateSpanByServiceMethod(
+                    String.format(OPERATIONFORMAT, method.getBareMethodName()));
+            super.start(responseListener, headers);
+            span.finish();
+          }
+
+          @Override
+          public void sendMessage(ReqT message) {
+            if (Objects.equals(method.getBareMethodName(), "StartWorkflowExecution")
+                && message instanceof StartWorkflowExecutionRequest) {
+              StartWorkflowExecutionRequest request = (StartWorkflowExecutionRequest) message;
+              Map<String, byte[]> headers = new HashMap<>();
+              tracingPropagator.inject(headers);
+              Header.Builder headerBuilder = request.getHeader().toBuilder();
+              headers.forEach(
+                  (k, v) -> {
+                    headerBuilder.putFields(
+                        k, Payload.newBuilder().setData(ByteString.copyFrom(v)).build());
+                  });
+
+              // cast should not throw error as we are using the builder
+              message =
+                  (ReqT)
+                      ((StartWorkflowExecutionRequest) message)
+                          .toBuilder()
+                          .setHeader(headerBuilder.build())
+                          .build();
+            }
+            if (Objects.equals(method.getBareMethodName(), "SignalWithStartWorkflowExecution")
+                && message instanceof SignalWithStartWorkflowExecutionRequest) {
+              SignalWithStartWorkflowExecutionRequest request =
+                  (SignalWithStartWorkflowExecutionRequest) message;
+              Map<String, byte[]> headers = new HashMap<>();
+              tracingPropagator.inject(headers);
+              Header.Builder headerBuilder = request.getStartRequest().getHeader().toBuilder();
+              headers.forEach(
+                  (k, v) -> {
+                    headerBuilder.putFields(
+                        k, Payload.newBuilder().setData(ByteString.copyFrom(v)).build());
+                  });
+
+              // cast should not throw error as we are using the builder
+              message =
+                  (ReqT)
+                      ((SignalWithStartWorkflowExecutionRequest) message)
+                          .toBuilder()
+                          .setStartRequest(
+                              request
+                                  .getStartRequest()
+                                  .toBuilder()
+                                  .setHeader(headerBuilder.build()))
+                          .build();
+            }
+            super.sendMessage(message);
+          }
+        };
+      }
+    };
+  }
+
   private ClientInterceptor newTracingInterceptor() {
     return new ClientInterceptor() {
 

src/main/java/com/uber/cadence/internal/sync/POJOWorkflowImplementationFactory.java

@@ -42,6 +42,7 @@
 import com.uber.cadence.workflow.WorkflowInfo;
 import com.uber.cadence.workflow.WorkflowInterceptor;
 import com.uber.cadence.workflow.WorkflowMethod;
+import io.opentracing.Tracer;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Collections;
@@ -61,6 +62,7 @@ final class POJOWorkflowImplementationFactory implements ReplayWorkflowFactory {
       LoggerFactory.getLogger(POJOWorkflowImplementationFactory.class);
   private static final byte[] EMPTY_BLOB = {};
   private final Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory;
+  private final Tracer tracer;
 
   private DataConverter dataConverter;
   private List<ContextPropagator> contextPropagators;
@@ -83,12 +85,14 @@ final class POJOWorkflowImplementationFactory implements ReplayWorkflowFactory {
       ExecutorService threadPool,
       Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
       DeciderCache cache,
-      List<ContextPropagator> contextPropagators) {
+      List<ContextPropagator> contextPropagators,
+      Tracer tracer) {
     this.dataConverter = Objects.requireNonNull(dataConverter);
     this.threadPool = Objects.requireNonNull(threadPool);
     this.interceptorFactory = Objects.requireNonNull(interceptorFactory);
     this.cache = cache;
     this.contextPropagators = contextPropagators;
+    this.tracer = tracer;
   }
 
   void setWorkflowImplementationTypes(
@@ -216,7 +220,8 @@ public ReplayWorkflow getWorkflow(WorkflowType workflowType) {
         threadPool,
         interceptorFactory,
         cache,
-        contextPropagators);
+        contextPropagators,
+        tracer);
   }
 
   @Override

src/main/java/com/uber/cadence/internal/sync/SyncDecisionContext.java

@@ -40,6 +40,7 @@
 import com.uber.cadence.internal.replay.ExecuteLocalActivityParameters;
 import com.uber.cadence.internal.replay.SignalExternalWorkflowParameters;
 import com.uber.cadence.internal.replay.StartChildWorkflowExecutionParameters;
+import com.uber.cadence.internal.tracing.TracingPropagator;
 import com.uber.cadence.worker.WorkflowImplementationOptions;
 import com.uber.cadence.workflow.ActivityException;
 import com.uber.cadence.workflow.ActivityFailureException;
@@ -58,6 +59,9 @@
 import com.uber.cadence.workflow.Workflow;
 import com.uber.cadence.workflow.WorkflowInterceptor;
 import com.uber.m3.tally.Scope;
+import io.opentracing.Span;
+import io.opentracing.Tracer;
+import io.opentracing.noop.NoopTracerFactory;
 import java.lang.reflect.Type;
 import java.time.Duration;
 import java.util.HashMap;
@@ -90,6 +94,7 @@ final class SyncDecisionContext implements WorkflowInterceptor {
   private final Map<String, Functions.Func1<byte[], byte[]>> queryCallbacks = new HashMap<>();
   private final byte[] lastCompletionResult;
   private final WorkflowImplementationOptions workflowImplementationOptions;
+  private final TracingPropagator tracingPropagator;
 
   public SyncDecisionContext(
       DecisionContext context,
@@ -98,6 +103,24 @@ public SyncDecisionContext(
       Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
       byte[] lastCompletionResult,
       WorkflowImplementationOptions workflowImplementationOptions) {
+    this(
+        context,
+        converter,
+        contextPropagators,
+        interceptorFactory,
+        lastCompletionResult,
+        workflowImplementationOptions,
+        NoopTracerFactory.create());
+  }
+
+  public SyncDecisionContext(
+      DecisionContext context,
+      DataConverter converter,
+      List<ContextPropagator> contextPropagators,
+      Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
+      byte[] lastCompletionResult,
+      WorkflowImplementationOptions workflowImplementationOptions,
+      Tracer tracer) {
     this.context = context;
     this.converter = converter;
     this.contextPropagators = contextPropagators;
@@ -109,6 +132,7 @@ public SyncDecisionContext(
     this.headInterceptor = interceptor;
     this.lastCompletionResult = lastCompletionResult;
     this.workflowImplementationOptions = workflowImplementationOptions;
+    this.tracingPropagator = new TracingPropagator(tracer);
   }
 
   /**
@@ -130,9 +154,18 @@ public WorkflowInterceptor getWorkflowInterceptor() {
   @Override
   public byte[] executeWorkflow(
       SyncWorkflowDefinition workflowDefinition, WorkflowExecuteInput input) {
-    return workflowDefinition.execute(input.getInput());
+    Span span = tracingPropagator.activateSpanForExecuteWorkflow(context);
+    try {
+      return workflowDefinition.execute(input.getInput());
+    } finally {
+      span.finish();
+    }
   }
 
+  /**
+   * Schedule an activity task for the provided activity name and input. The activity task is not
+   * necessarily executed in the same thread.
+   */
   @Override
   public <T> Promise<T> executeActivity(
       String activityName,
@@ -466,6 +499,8 @@ private Map<String, byte[]> extractContextsAndConvertToBytes(
     for (ContextPropagator propagator : contextPropagators) {
       result.putAll(propagator.serializeContext(propagator.getCurrentContext()));
     }
+    // inject trace span context
+    tracingPropagator.inject(result);
     return result;
   }
 

src/main/java/com/uber/cadence/internal/sync/SyncWorkflow.java

@@ -32,6 +32,7 @@
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
 import com.uber.cadence.worker.WorkflowImplementationOptions;
 import com.uber.cadence.workflow.WorkflowInterceptor;
+import io.opentracing.Tracer;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
@@ -49,6 +50,7 @@ class SyncWorkflow implements ReplayWorkflow {
   private final SyncWorkflowDefinition workflow;
   WorkflowImplementationOptions workflowImplementationOptions;
   private final Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory;
+  private final Tracer tracer;
   private DeciderCache cache;
   private WorkflowRunnable workflowProc;
   private DeterministicRunner runner;
@@ -60,7 +62,8 @@ public SyncWorkflow(
       ExecutorService threadPool,
       Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
       DeciderCache cache,
-      List<ContextPropagator> contextPropagators) {
+      List<ContextPropagator> contextPropagators,
+      Tracer tracer) {
     this.workflow = Objects.requireNonNull(workflow);
     this.workflowImplementationOptions =
         workflowImplementationOptions == null
@@ -71,6 +74,7 @@ public SyncWorkflow(
     this.interceptorFactory = Objects.requireNonNull(interceptorFactory);
     this.cache = cache;
     this.contextPropagators = contextPropagators;
+    this.tracer = tracer;
   }
 
   @Override
@@ -97,7 +101,8 @@ public void start(HistoryEvent event, DecisionContext context) {
             contextPropagators,
             interceptorFactory,
             event.getWorkflowExecutionStartedEventAttributes().getLastCompletionResult(),
-            workflowImplementationOptions);
+            workflowImplementationOptions,
+            tracer);
 
     workflowProc =
         new WorkflowRunnable(

src/main/java/com/uber/cadence/internal/sync/SyncWorkflowWorker.java

@@ -80,7 +80,8 @@ public SyncWorkflowWorker(
             workflowThreadPool,
             interceptorFactory,
             cache,
-            workflowOptions.getContextPropagators());
+            workflowOptions.getContextPropagators(),
+            workflowOptions.getTracer());
 
     laTaskHandler =
         new POJOActivityTaskHandler(