feat(trace): add cadenceIsCron tag in the execute workflow span (#1071)

What changed?

Add cadenceIsCron tag on execute workflow span. (In Go, we add it in start workflow span)

Why?
cadence-workflow/cadence-go-client#1518

How did you test it?

Unit Test / Integration Test

Author: shijiesheng

src/main/java/com/uber/cadence/internal/tracing/TracingPropagator.java

@@ -45,6 +45,7 @@ public class TracingPropagator {
   private static final String TAG_WORKFLOW_TYPE = "cadenceWorkflowType";
   private static final String TAG_WORKFLOW_RUN_ID = "cadenceRunID";
   private static final String TAG_ACTIVITY_TYPE = "cadenceActivityType";
+  private static final String TAG_IS_CRON = "cadenceIsCron";
 
   private final Tracer tracer;
 
@@ -70,6 +71,9 @@ public Span spanForExecuteWorkflow(DecisionContext context) {
         .withTag(TAG_WORKFLOW_TYPE, context.getWorkflowType().getName())
         .withTag(TAG_WORKFLOW_ID, context.getWorkflowId())
         .withTag(TAG_WORKFLOW_RUN_ID, context.getRunId())
+        .withTag(
+            TAG_IS_CRON,
+            attributes.getCronSchedule() != null && !attributes.getCronSchedule().isEmpty())
         .start();
   }
 

src/test/java/com/uber/cadence/internal/tracing/StartWorkflowTest.java

@@ -137,6 +137,18 @@ public Integer Double(Integer n) {
     }
   }
 
+  public interface CronWorkflow {
+    @WorkflowMethod(executionStartToCloseTimeoutSeconds = 120, taskList = TASK_LIST)
+    String execute();
+  }
+
+  public static class CronWorkflowImpl implements CronWorkflow {
+    @Override
+    public String execute() {
+      return "done";
+    }
+  }
+
   private static final boolean useDockerService = TestEnvironment.isUseDockerService();
   private static final Logger logger = LoggerFactory.getLogger(StartWorkflowTest.class);
   private static final String DOMAIN = "test-domain";
@@ -259,6 +271,93 @@ public void testSignalStartWorkflowGRPCNoPropagation() {
     testSignalWithStartWorkflowHelper(service, mockTracer, false);
   }
 
+  @Test
+  public void testCronWorkflowSetsIsCronSpanTagGRPC() {
+    Assume.assumeTrue(useDockerService);
+    MockTracer mockTracer = new MockTracer();
+    IWorkflowService service =
+        new WorkflowServiceGrpc(
+            ClientOptions.newBuilder().setTracer(mockTracer).setPort(7833).build());
+    testCronWorkflowHelper(service, mockTracer);
+  }
+
+  private void testCronWorkflowHelper(IWorkflowService service, MockTracer mockTracer) {
+    try {
+      service.RegisterDomain(new RegisterDomainRequest().setName(DOMAIN));
+    } catch (DomainAlreadyExistsError e) {
+      logger.info("domain already registered");
+    } catch (Exception e) {
+      fail("fail to register domain: " + e);
+    }
+
+    WorkflowClient client =
+        WorkflowClient.newInstance(
+            service, WorkflowClientOptions.newBuilder().setDomain(DOMAIN).build());
+
+    WorkerFactory workerFactory =
+        WorkerFactory.newInstance(client, WorkerFactoryOptions.newBuilder().build());
+    Worker worker = workerFactory.newWorker(TASK_LIST, WorkerOptions.newBuilder().build());
+    worker.registerWorkflowImplementationTypes(CronWorkflowImpl.class);
+    workerFactory.start();
+
+    Span rootSpan = mockTracer.buildSpan("Test Started").start();
+    mockTracer.activateSpan(rootSpan);
+
+    WorkflowStub wf =
+        client.newUntypedWorkflowStub(
+            "CronWorkflow::execute",
+            new WorkflowOptions.Builder()
+                .setExecutionStartToCloseTimeout(Duration.ofMinutes(2))
+                .setTaskList(TASK_LIST)
+                .setCronSchedule("* * * * *")
+                .build());
+    try {
+      wf.start();
+
+      // Cron's minimum interval is 1 minute, so wait for the first scheduled run to execute and
+      // produce a finished cadence-ExecuteWorkflow span.
+      MockSpan executeWorkflowSpan = awaitExecuteWorkflowSpan(mockTracer, Duration.ofSeconds(150));
+      assertNotNull("cadence-ExecuteWorkflow span not found", executeWorkflowSpan);
+      assertEquals(
+          "cadenceIsCron tag should be true for cron workflows",
+          Boolean.TRUE,
+          executeWorkflowSpan.tags().get("cadenceIsCron"));
+    } catch (Exception e) {
+      fail("workflow failure: " + e);
+    } finally {
+      try {
+        wf.cancel();
+      } catch (Exception ignored) {
+        // best effort: stop further cron runs
+      }
+      rootSpan.finish();
+      workerFactory.shutdown();
+    }
+  }
+
+  private MockSpan awaitExecuteWorkflowSpan(MockTracer mockTracer, Duration timeout) {
+    long deadline = System.currentTimeMillis() + timeout.toMillis();
+    while (System.currentTimeMillis() < deadline) {
+      MockSpan span =
+          mockTracer
+              .finishedSpans()
+              .stream()
+              .filter(s -> s.operationName().equals("cadence-ExecuteWorkflow"))
+              .findFirst()
+              .orElse(null);
+      if (span != null) {
+        return span;
+      }
+      try {
+        Thread.sleep(1000);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        break;
+      }
+    }
+    return null;
+  }
+
   private void testStartWorkflowHelper(
       IWorkflowService service, MockTracer mockTracer, boolean shouldPropagate) {
     try {

src/test/java/com/uber/cadence/internal/tracing/TracingPropagatorTest.java

@@ -19,10 +19,15 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import com.google.common.collect.ImmutableMap;
 import com.uber.cadence.Header;
 import com.uber.cadence.PollForActivityTaskResponse;
+import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
+import com.uber.cadence.WorkflowType;
+import com.uber.cadence.internal.replay.DecisionContext;
 import io.opentracing.Span;
 import io.opentracing.mock.MockSpan;
 import io.opentracing.mock.MockTracer;
@@ -31,6 +36,8 @@
 
 public class TracingPropagatorTest {
 
+  private static final String CADENCE_IS_CRON = "cadenceIsCron";
+
   private final MockTracer mockTracer = new MockTracer();
   private final TracingPropagator propagator = new TracingPropagator(mockTracer);
 
@@ -58,4 +65,34 @@ public void testSpanForExecuteActivity_allowReusingHeaders() {
       assertEquals("follows_from", from.getReferenceType());
     }
   }
+
+  @Test
+  public void testSpanForExecuteWorkflow_withCronSchedule_setsIsCronTagTrue() {
+    Span span = propagator.spanForExecuteWorkflow(newDecisionContext("0 * * * *"));
+    span.finish();
+
+    MockSpan mockSpan = mockTracer.finishedSpans().get(0);
+    assertEquals(Boolean.TRUE, mockSpan.tags().get(CADENCE_IS_CRON));
+  }
+
+  @Test
+  public void testSpanForExecuteWorkflow_withoutCronSchedule_setsIsCronTagFalse() {
+    Span span = propagator.spanForExecuteWorkflow(newDecisionContext(""));
+    span.finish();
+
+    MockSpan mockSpan = mockTracer.finishedSpans().get(0);
+    assertEquals(Boolean.FALSE, mockSpan.tags().get(CADENCE_IS_CRON));
+  }
+
+  private static DecisionContext newDecisionContext(String cronSchedule) {
+    WorkflowExecutionStartedEventAttributes attributes =
+        new WorkflowExecutionStartedEventAttributes().setCronSchedule(cronSchedule);
+
+    DecisionContext context = mock(DecisionContext.class);
+    when(context.getWorkflowExecutionStartedEventAttributes()).thenReturn(attributes);
+    when(context.getWorkflowType()).thenReturn(new WorkflowType().setName("TestWorkflow"));
+    when(context.getWorkflowId()).thenReturn("workflow-id");
+    when(context.getRunId()).thenReturn("run-id");
+    return context;
+  }
 }