From 812bb41626afd78c7380370db8a38e2d5e55a37d Mon Sep 17 00:00:00 2001 From: Shijie Sheng Date: Wed, 8 Jul 2026 15:29:00 -0700 Subject: [PATCH] feat(trace): add cadenceIsCron tag in the start workflow span Signed-off-by: Shijie Sheng --- .../internal/tracing/TracingPropagator.java | 2 + .../internal/tracing/StartWorkflowTest.java | 99 +++++++++++++++++++ .../tracing/TracingPropagatorTest.java | 37 +++++++ 3 files changed, 138 insertions(+) diff --git a/src/main/java/com/uber/cadence/internal/tracing/TracingPropagator.java b/src/main/java/com/uber/cadence/internal/tracing/TracingPropagator.java index d3800f995..e53a80acd 100644 --- a/src/main/java/com/uber/cadence/internal/tracing/TracingPropagator.java +++ b/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,7 @@ 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() != "") .start(); } diff --git a/src/test/java/com/uber/cadence/internal/tracing/StartWorkflowTest.java b/src/test/java/com/uber/cadence/internal/tracing/StartWorkflowTest.java index 17c78e2ae..8c5ddd180 100644 --- a/src/test/java/com/uber/cadence/internal/tracing/StartWorkflowTest.java +++ b/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 { diff --git a/src/test/java/com/uber/cadence/internal/tracing/TracingPropagatorTest.java b/src/test/java/com/uber/cadence/internal/tracing/TracingPropagatorTest.java index 9c50b5f48..284d469b8 100644 --- a/src/test/java/com/uber/cadence/internal/tracing/TracingPropagatorTest.java +++ b/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; + } }