Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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() != "")
Comment thread
gitar-bot[bot] marked this conversation as resolved.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
.withTag(TAG_IS_CRON, attributes.getCronSchedule() != "")
.withTag(TAG_IS_CRON, attributes.getCronSchedule() != null && !attributes.getCronSchedule().isEmpty())

!= on strings compares object references

.start();
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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);

Expand Down Expand Up @@ -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;
}
}
Loading