diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityPollTask.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityPollTask.java index f0d3e649f0..c92cdb4ecf 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityPollTask.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityPollTask.java @@ -10,7 +10,7 @@ import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; import io.temporal.api.workflowservice.v1.PollActivityTaskQueueRequest; import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; -import io.temporal.internal.common.ProtobufTimeUtils; + import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.worker.MetricsType; @@ -120,11 +120,6 @@ public ActivityTask poll() { metricsScope.counter(MetricsType.ACTIVITY_POLL_NO_TASK_COUNTER).inc(1); return null; } - metricsScope - .timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY) - .record( - ProtobufTimeUtils.toM3Duration( - response.getStartedTime(), response.getCurrentAttemptScheduledTime())); isSuccessful = true; pollerTracker.pollSucceeded(); return new ActivityTask( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java index 1e8791bd02..f8766c1109 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java @@ -12,7 +12,7 @@ import io.temporal.api.workflowservice.v1.PollActivityTaskQueueRequest; import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; import io.temporal.internal.common.GrpcUtils; -import io.temporal.internal.common.ProtobufTimeUtils; + import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.worker.MetricsType; @@ -121,11 +121,6 @@ public CompletableFuture poll(SlotPermit permit) { return null; } pollerTracker.pollSucceeded(); - metricsScope - .timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY) - .record( - ProtobufTimeUtils.toM3Duration( - r.getStartedTime(), r.getCurrentAttemptScheduledTime())); return new ActivityTask( r, permit, diff --git a/temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java b/temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java index 6e03278164..00a2f47fdf 100644 --- a/temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java +++ b/temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java @@ -89,6 +89,38 @@ public synchronized void assertTimer(String name, Map tags) { } } + /** + * Asserts that a timer metric was reported exactly {@code expectedCount} times for the given + * tags. Useful for verifying that a metric is not accidentally recorded more than once (e.g. + * duplicated by multiple code paths). + */ + public synchronized void assertTimer(String name, Map tags, long expectedCount) { + String metricName = getMetricName(name, tags); + StatsAccumulator value = timers.get(metricName); + if (value == null) { + fail( + "No metric '" + + metricName + + "', reported metrics: \n " + + String.join("\n ", timers.keySet())); + } + assertEquals( + "Timer '" + metricName + "' was reported an unexpected number of times", + expectedCount, + value.count()); + } + + public synchronized void assertNoTimer(String name, Map tags) { + String metricName = getMetricName(name, tags); + if (timers.containsKey(metricName)) { + fail( + "Timer metric '" + + metricName + + "' was reported, but was not expected. Recorded count: " + + timers.get(metricName).count()); + } + } + public synchronized void assertTimerMinDuration( String name, Map tags, Duration minDuration) { String metricName = getMetricName(name, tags); diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java index 17178b1996..4cc1de3327 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java @@ -120,6 +120,17 @@ public class MetricsTest { .put(MetricsTag.POLLER_TYPE, PollerTypeMetricsTag.PollerType.ACTIVITY_TASK.getValue()) .build(); + // Tags for the activity_schedule_to_start_latency metric, which is recorded with the real + // activity_type/workflow_type once per activity task in ActivityWorker.handleActivity (see + // #2733 - previously it was also incorrectly recorded a second/third time at poll time with + // activity_type defaulting to "none"). + private static final Map TAGS_ACTIVITY_SCHEDULE_TO_START = + new ImmutableMap.Builder() + .putAll(TAGS_ACTIVITY_WORKER) + .put(MetricsTag.ACTIVITY_TYPE, "Execute") + .put(MetricsTag.WORKFLOW_TYPE, "NoArgsWorkflow") + .build(); + @Rule public TestWatcher watchman = new TestWatcher() { @@ -183,7 +194,11 @@ public void testWorkerMetrics() throws InterruptedException { "temporal_workflow_task_queue_poll_succeed", TAGS_STICKY_WORKFLOW_WORKER); reporter.assertCounter("temporal_workflow_task_queue_poll_succeed", TAGS_WORKFLOW_WORKER); // We ran some workflow and activity tasks, so we should have some timers here. - reporter.assertTimer("temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_WORKER); + // activity_schedule_to_start_latency is recorded once per activity task, tagged with the + // real activity_type (see ActivityWorker.handleActivity / #2733), not the poller-level + // scope's default activity_type="none". + reporter.assertTimer( + "temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_SCHEDULE_TO_START); reporter.assertTimer("temporal_workflow_task_schedule_to_start_latency", TAGS_WORKFLOW_WORKER); reporter.assertTimer( "temporal_workflow_task_schedule_to_start_latency", TAGS_STICKY_WORKFLOW_WORKER); @@ -228,7 +243,11 @@ public void testWorkerMetricsAutoPoller() throws InterruptedException { "temporal_workflow_task_queue_poll_succeed", TAGS_STICKY_WORKFLOW_WORKER); reporter.assertCounter("temporal_workflow_task_queue_poll_succeed", TAGS_WORKFLOW_WORKER); // We ran some workflow and activity tasks, so we should have some timers here. - reporter.assertTimer("temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_WORKER); + // activity_schedule_to_start_latency is recorded once per activity task, tagged with the + // real activity_type (see ActivityWorker.handleActivity / #2733), not the poller-level + // scope's default activity_type="none". + reporter.assertTimer( + "temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_SCHEDULE_TO_START); reporter.assertTimer("temporal_workflow_task_schedule_to_start_latency", TAGS_WORKFLOW_WORKER); reporter.assertTimer( "temporal_workflow_task_schedule_to_start_latency", TAGS_STICKY_WORKFLOW_WORKER); @@ -239,6 +258,47 @@ public void testWorkerMetricsAutoPoller() throws InterruptedException { reporter.assertGauge("temporal_num_pollers", TAGS_ACTIVITY_POLLER, 5); } + /** + * Regression test for #2733: activity_schedule_to_start_latency must be recorded exactly once per + * activity task, and only with the correct activity_type tag. + * + *

Previously the metric was recorded up to three times per activity task: once in {@code + * ActivityPollTask}/{@code AsyncActivityPollTask} at poll time (using the poller-level scope, + * where activity_type defaults to "none"), and once in {@code ActivityWorker.handleActivity} + * using a scope tagged with the real activity_type. This test verifies the duplicate poll-time + * recording(s) were removed and only the correctly-tagged recording from {@code handleActivity} + * remains. + */ + @Test + public void testActivityScheduleToStartLatencyMetricRecordedOnce() throws InterruptedException { + setUp(WorkerFactoryOptions.getDefaultInstance()); + + Worker worker = testEnvironment.newWorker(TASK_QUEUE); + worker.registerWorkflowImplementationTypes(TestScheduleToStartMetricWorkflow.class); + worker.registerActivitiesImplementations(new TestActivityImpl()); + testEnvironment.start(); + + WorkflowClient workflowClient = testEnvironment.getWorkflowClient(); + WorkflowOptions options = + WorkflowOptions.newBuilder() + .setWorkflowRunTimeout(Duration.ofSeconds(1000)) + .setTaskQueue(TASK_QUEUE) + .build(); + NoArgsWorkflow workflow = workflowClient.newWorkflowStub(NoArgsWorkflow.class, options); + workflow.execute(); + + Thread.sleep(REPORTING_FLUSH_TIME); + + // The metric must be recorded exactly once with the correct activity_type/workflow_type + // tags (i.e. the recording made by ActivityWorker.handleActivity). + reporter.assertTimer( + "temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_SCHEDULE_TO_START, 1); + + // The metric must NOT additionally be recorded under the poller-level scope, where + // activity_type defaults to "none" - that was the duplicate recording removed by #2733. + reporter.assertNoTimer("temporal_activity_schedule_to_start_latency", TAGS_ACTIVITY_WORKER); + } + @Test public void testWorkflowMetrics() throws InterruptedException { setUp(WorkerFactoryOptions.getDefaultInstance()); @@ -593,6 +653,20 @@ public int execute(int input) { } } + public static class TestScheduleToStartMetricWorkflow implements NoArgsWorkflow { + + @Override + public void execute() { + ActivityOptions activityOptions = + ActivityOptions.newBuilder() + .setTaskQueue(TASK_QUEUE) + .setScheduleToCloseTimeout(Duration.ofSeconds(100)) + .build(); + TestActivity3 activity = Workflow.newActivityStub(TestActivity3.class, activityOptions); + activity.execute(1); + } + } + public static class TestMetricsInChildWorkflow implements TestChildWorkflow { @Override