Align workflow_task_schedule_to_start_latency (#2669)

Align workflow_task_schedule_to_start_latency

Author: Quinn-With-Two-Ns

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

@@ -3,7 +3,6 @@
 import static io.temporal.serviceclient.MetricsTag.METRICS_TAGS_CALL_OPTIONS_KEY;
 
 import com.google.protobuf.DoubleValue;
-import com.google.protobuf.Timestamp;
 import com.uber.m3.tally.Scope;
 import io.grpc.Context;
 import io.temporal.api.common.v1.WorkerVersionCapabilities;
@@ -113,10 +112,11 @@ public CompletableFuture<ActivityTask> poll(SlotPermit permit) {
                 metricsScope.counter(MetricsType.ACTIVITY_POLL_NO_TASK_COUNTER).inc(1);
                 return null;
               }
-              Timestamp startedTime = ProtobufTimeUtils.getCurrentProtoTime();
               metricsScope
                   .timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY)
-                  .record(ProtobufTimeUtils.toM3Duration(startedTime, r.getScheduledTime()));
+                  .record(
+                      ProtobufTimeUtils.toM3Duration(
+                          r.getStartedTime(), r.getCurrentAttemptScheduledTime()));
               return new ActivityTask(
                   r,
                   permit,

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

@@ -2,7 +2,6 @@
 
 import static io.temporal.serviceclient.MetricsTag.METRICS_TAGS_CALL_OPTIONS_KEY;
 
-import com.google.protobuf.Timestamp;
 import com.uber.m3.tally.Scope;
 import com.uber.m3.util.ImmutableMap;
 import io.grpc.Context;
@@ -154,10 +153,9 @@ public CompletableFuture<WorkflowTask> poll(SlotPermit permit)
               pollerMetricScope
                   .counter(MetricsType.WORKFLOW_TASK_QUEUE_POLL_SUCCEED_COUNTER)
                   .inc(1);
-              Timestamp startedTime = ProtobufTimeUtils.getCurrentProtoTime();
               pollerMetricScope
                   .timer(MetricsType.WORKFLOW_TASK_SCHEDULE_TO_START_LATENCY)
-                  .record(ProtobufTimeUtils.toM3Duration(startedTime, r.getScheduledTime()));
+                  .record(ProtobufTimeUtils.toM3Duration(r.getStartedTime(), r.getScheduledTime()));
               return new WorkflowTask(r, (reason) -> slotSupplier.releaseSlot(reason, permit));
             })
         .whenComplete(