Emit DimensionalTimeSliceCrawler worker partition process latency metrics (#6091)

**What?**

This commit emits two additional metrics for DimensionalTimeSliceCrawler

metric WorkerPartitionWaitTime measures the wait time before a worker partition
starts being processed after creation.

metric WorkerPartitionProcessLatency measures the duration between pulling data
from source to writing to buffer in a worker partition.

**Why?**

These two metrics will provide visibility how long it takes to process a
partition after being created.

Signed-off-by: Wenjie Yao <wjyao@amazon.com>
Co-authored-by: Wenjie Yao <wjyao@amazon.com>

Author: wjyao0316

data-prepper-plugins/saas-source-plugins/source-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/source_crawler/base/DimensionalTimeSliceCrawler.java

@@ -1,6 +1,7 @@
 package org.opensearch.dataprepper.plugins.source.source_crawler.base;
 
 import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Timer;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSet;
 import org.opensearch.dataprepper.model.buffer.Buffer;
@@ -36,17 +37,23 @@
 public class DimensionalTimeSliceCrawler implements Crawler<DimensionalTimeSliceWorkerProgressState> {
     private static final Logger log = LoggerFactory.getLogger(DimensionalTimeSliceCrawler.class);
     private static final String DIMENSIONAL_TIME_SLICE_WORKER_PARTITIONS_CREATED = "DimensionalTimeSliceWorkerPartitionsCreated";
+    private static final String WORKER_PARTITION_WAIT_TIME = "WorkerPartitionWaitTime";
+    private static final String WORKER_PARTITION_PROCESS_LATENCY = "WorkerPartitionProcessLatency";
     private static final Duration HOUR_DURATION = Duration.ofHours(1);
 
     private final CrawlerClient client;
     private final Counter partitionsCreatedCounter;
+    private final Timer partitionWaitTimeTimer;
+    private final Timer partitionProcessLatencyTimer;
     private List<String> dimensionTypes;
     private static final String LAST_UPDATED_KEY = "last_updated|";
 
     public DimensionalTimeSliceCrawler(CrawlerClient client,
                                        PluginMetrics pluginMetrics) {
         this.client = client;
         this.partitionsCreatedCounter = pluginMetrics.counter(DIMENSIONAL_TIME_SLICE_WORKER_PARTITIONS_CREATED);
+        this.partitionWaitTimeTimer = pluginMetrics.timer(WORKER_PARTITION_WAIT_TIME);
+        this.partitionProcessLatencyTimer = pluginMetrics.timer(WORKER_PARTITION_PROCESS_LATENCY);
     }
 
     /**
@@ -78,7 +85,8 @@ public Instant crawl(LeaderPartition leaderPartition, EnhancedSourceCoordinator
 
     @Override
     public void executePartition(DimensionalTimeSliceWorkerProgressState state, Buffer<Record<Event>> buffer, AcknowledgementSet acknowledgementSet) {
-        client.executePartition(state, buffer, acknowledgementSet);
+        partitionWaitTimeTimer.record(Duration.between(state.getPartitionCreationTime(), Instant.now()));
+        partitionProcessLatencyTimer.record(() -> client.executePartition(state, buffer, acknowledgementSet));
     }
 
     private void createPartitionsForDimensionTypes(LeaderPartition leaderPartition,
@@ -149,6 +157,7 @@ private void createPartitionForIncrementalSync(LeaderPartition leaderPartition,
     void createWorkerPartition(Instant startTime, Instant endTime,
                                String dimensionType, EnhancedSourceCoordinator coordinator) {
         DimensionalTimeSliceWorkerProgressState workerState = new DimensionalTimeSliceWorkerProgressState();
+        workerState.setPartitionCreationTime(Instant.now());
         workerState.setStartTime(startTime);
         workerState.setEndTime(endTime);
         workerState.setDimensionType(dimensionType);

data-prepper-plugins/saas-source-plugins/source-crawler/src/main/java/org/opensearch/dataprepper/plugins/source/source_crawler/coordination/state/DimensionalTimeSliceWorkerProgressState.java

@@ -17,4 +17,7 @@ public class DimensionalTimeSliceWorkerProgressState implements SaasWorkerProgre
 
     @JsonProperty("dimensionType")
     private String dimensionType;
+
+    @JsonProperty("partitionCreationTime")
+    private Instant partitionCreationTime;
 }

data-prepper-plugins/saas-source-plugins/source-crawler/src/test/java/org/opensearch/dataprepper/plugins/source/source_crawler/base/DimensionalTimeSliceCrawlerTest.java

@@ -1,6 +1,7 @@
 package org.opensearch.dataprepper.plugins.source.source_crawler.base;
 
 import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Timer;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -37,6 +38,8 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
 
 @ExtendWith(MockitoExtension.class)
 public class DimensionalTimeSliceCrawlerTest {
@@ -50,6 +53,12 @@ public class DimensionalTimeSliceCrawlerTest {
     @Mock
     private Counter partitionsCreatedCounter;
 
+    @Mock
+    private Timer partitionWaitTimeTimer;
+
+    @Mock
+    private Timer partitionProcessLatencyTimer;
+
     @Mock
     private EnhancedSourceCoordinator coordinator;
 
@@ -64,6 +73,8 @@ public class DimensionalTimeSliceCrawlerTest {
     @BeforeEach
     void setUp() {
         when(pluginMetrics.counter(anyString())).thenReturn(partitionsCreatedCounter);
+        when(pluginMetrics.timer("WorkerPartitionWaitTime")).thenReturn(partitionWaitTimeTimer);
+        when(pluginMetrics.timer("WorkerPartitionProcessLatency")).thenReturn(partitionProcessLatencyTimer);
         crawler = new DimensionalTimeSliceCrawler(client, pluginMetrics);
         crawler.initialize(LOG_TYPES);
     }
@@ -144,12 +155,22 @@ void testCreateWorkerPartition() {
     @Test
     void testExecutePartition() {
         DimensionalTimeSliceWorkerProgressState state = new DimensionalTimeSliceWorkerProgressState();
+        state.setPartitionCreationTime(Instant.now().minusSeconds(1));
         Buffer<Record<Event>> buffer = mock(Buffer.class);
         AcknowledgementSet ackSet = mock(AcknowledgementSet.class);
 
+        doAnswer(invocation -> {
+            Runnable runnable = invocation.getArgument(0);
+            runnable.run();
+            return null;
+        }).when(partitionProcessLatencyTimer).record(any(Runnable.class));
+        doNothing().when(partitionWaitTimeTimer).record(any(Duration.class));
+
         crawler.executePartition(state, buffer, ackSet);
 
         verify(client).executePartition(eq(state), eq(buffer), eq(ackSet));
+        verify(partitionProcessLatencyTimer).record(any(Runnable.class));
+        verify(partitionWaitTimeTimer).record(any(Duration.class));
     }
 
     @Test

data-prepper-plugins/saas-source-plugins/source-crawler/src/test/java/org/opensearch/dataprepper/plugins/source/source_crawler/coordination/state/DimensionalTimeSliceWorkerProgressStateTest.java

@@ -32,6 +32,7 @@ void testDefaultValues() throws JsonProcessingException {
         assertNull(state.getStartTime());
         assertNull(state.getEndTime());
         assertNull(state.getDimensionType());
+        assertNull(state.getPartitionCreationTime());
     }
 
     @Test
@@ -40,12 +41,14 @@ void testDeserializeDimensionalTimeSliceWorkerProgressState_withTypeInfo() throw
                 "  \"@class\": \"org.opensearch.dataprepper.plugins.source.source_crawler.coordination.state.DimensionalTimeSliceWorkerProgressState\",\n" +
                 "  \"startTime\": \"2024-10-20T02:27:15.717Z\",\n" +
                 "  \"endTime\": \"2024-10-20T03:27:15.717Z\",\n" +
+                "  \"partitionCreationTime\": \"2024-11-21T04:29:15.719Z\",\n" +
                 "  \"dimensionType\": \"Exchange\"\n" +
                 "}";
 
         DimensionalTimeSliceWorkerProgressState state = objectMapper.readValue(json, DimensionalTimeSliceWorkerProgressState.class);
         assertEquals(Instant.parse("2024-10-20T02:27:15.717Z"), state.getStartTime());
         assertEquals(Instant.parse("2024-10-20T03:27:15.717Z"), state.getEndTime());
+        assertEquals(Instant.parse("2024-11-21T04:29:15.719Z"), state.getPartitionCreationTime());
         assertEquals("Exchange", state.getDimensionType());
     }
 }