Report activity schedule to start latency once per task without the activity type

CHANGELOG.md

@@ -86,6 +86,9 @@ to docs, or any other relevant information.
   ignored and retries were scheduled past them instead of failing with `RETRY_STATE_TIMEOUT`.
 - Worker heartbeats now list each plugin name once. Previously a plugin registered on both the
   `WorkflowClient` and the `WorkerFactory` was listed twice.
+- Activity workers now report `activity_schedule_to_start_latency` once per activity task, measured up to the time the
+  worker receives the task. Previously each task reported it twice, once with the activity type and workflow type tags,
+  and measured it up to the server start time.
 
 ## Previous releases
 

temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityPollTask.java

@@ -10,7 +10,6 @@
 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 +119,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(

temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java

@@ -333,6 +333,13 @@ public void handle(ActivityTask task) throws Exception {
       ActivityTaskHandler.Result result = null;
       boolean taskFailed = false;
       try {
+        // Schedule to start latency does not depend on the activity type, so it is reported with
+        // the worker scope.
+        workerMetricsScope
+            .timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY)
+            .record(
+                ProtobufTimeUtils.toM3DurationSinceNow(
+                    pollResponse.getCurrentAttemptScheduledTime()));
         result = handleActivity(task, metricsScope);
         if (result.getTaskFailed() != null
             && !io.temporal.internal.common.FailureUtils.isBenignApplicationFailure(
@@ -382,12 +389,6 @@ private ActivityTaskHandler.Result handleActivity(ActivityTask task, Scope metri
             false);
       }
       PollActivityTaskQueueResponseOrBuilder pollResponse = task.getResponse();
-      metricsScope
-          .timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY)
-          .record(
-              ProtobufTimeUtils.toM3Duration(
-                  pollResponse.getStartedTime(), pollResponse.getCurrentAttemptScheduledTime()));
-
       ActivityTaskHandler.Result result;
 
       Stopwatch sw = metricsScope.timer(MetricsType.ACTIVITY_EXEC_LATENCY).start();

temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java

@@ -12,7 +12,6 @@
 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 +120,6 @@ public CompletableFuture<ActivityTask> 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,

temporal-sdk/src/test/java/io/temporal/common/reporter/TestStatsReporter.java

@@ -39,6 +39,9 @@ public synchronized void assertNoMetric(String name, Map<String, String> tags) {
               + counters.get(metricName).get()
               + "'");
     }
+    if (timers.containsKey(metricName)) {
+      fail("Timer '" + metricName + "' was reported");
+    }
   }
 
   public synchronized void assertCounter(String name, Map<String, String> tags, long expected) {
@@ -89,6 +92,20 @@ public synchronized void assertTimer(String name, Map<String, String> tags) {
     }
   }
 
+  public synchronized void assertTimer(
+      String name, Map<String, String> tags, Predicate<StatsAccumulator> isExpected) {
+    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()));
+    }
+    assertTrue(metricName + " count: " + value.count(), isExpected.test(value));
+  }
+
   public synchronized void assertTimerMinDuration(
       String name, Map<String, String> tags, Duration minDuration) {
     String metricName = getMetricName(name, tags);

temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java

@@ -307,6 +307,18 @@ public void testWorkflowMetrics() throws InterruptedException {
     reporter.assertCounter(TEMPORAL_REQUEST, workflowTags, 1);
     reporter.assertTimer(TEMPORAL_REQUEST_LATENCY, workflowTags);
 
+    // The workflow ran a single activity task. Schedule to start latency is independent of the
+    // activity type, so it must be reported exactly once and without the activity type tags.
+    reporter.assertTimer(
+        ACTIVITY_SCHEDULE_TO_START_LATENCY, TAGS_ACTIVITY_WORKER, stats -> stats.count() == 1);
+    reporter.assertNoMetric(
+        ACTIVITY_SCHEDULE_TO_START_LATENCY,
+        new ImmutableMap.Builder<String, String>()
+            .putAll(TAGS_ACTIVITY_WORKER)
+            .put(MetricsTag.ACTIVITY_TYPE, "Execute")
+            .put(MetricsTag.WORKFLOW_TYPE, "NoArgsWorkflow")
+            .build());
+
     Map<String, String> workflowTaskCompletionTags =
         new ImmutableMap.Builder<String, String>()
             .putAll(TAGS_WORKFLOW_WORKER)