Add metrics tracking shard time from unassigned to initialized/started

server/src/main/java/org/elasticsearch/cluster/ClusterModule.java

@@ -197,7 +197,8 @@ public ClusterModule(
             shardsAllocator,
             clusterInfoService,
             snapshotsInfoService,
-            shardRoutingRoleStrategy
+            shardRoutingRoleStrategy,
+            telemetryProvider.getMeterRegistry()
         );
         this.allocationService.addAllocFailuresResetListenerTo(clusterService);
         this.metadataDeleteIndexService = new MetadataDeleteIndexService(settings, clusterService, allocationService);

server/src/main/java/org/elasticsearch/cluster/routing/allocation/AllocationService.java

@@ -57,6 +57,7 @@
 import org.elasticsearch.index.Index;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.snapshots.SnapshotsInfoService;
+import org.elasticsearch.telemetry.metric.MeterRegistry;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -90,6 +91,7 @@ public class AllocationService {
     private final ClusterInfoService clusterInfoService;
     private final SnapshotsInfoService snapshotsInfoService;
     private final ShardRoutingRoleStrategy shardRoutingRoleStrategy;
+    private final ShardChangesObserver shardChangesObserver;
 
     // only for tests that use the GatewayAllocator as the unique ExistingShardsAllocator
     @SuppressWarnings("this-escape")
@@ -101,7 +103,7 @@ public AllocationService(
         SnapshotsInfoService snapshotsInfoService,
         ShardRoutingRoleStrategy shardRoutingRoleStrategy
     ) {
-        this(allocationDeciders, shardsAllocator, clusterInfoService, snapshotsInfoService, shardRoutingRoleStrategy);
+        this(allocationDeciders, shardsAllocator, clusterInfoService, snapshotsInfoService, shardRoutingRoleStrategy, MeterRegistry.NOOP);
         setExistingShardsAllocators(Collections.singletonMap(GatewayAllocator.ALLOCATOR_NAME, gatewayAllocator));
     }
 
@@ -110,13 +112,15 @@ public AllocationService(
         ShardsAllocator shardsAllocator,
         ClusterInfoService clusterInfoService,
         SnapshotsInfoService snapshotsInfoService,
-        ShardRoutingRoleStrategy shardRoutingRoleStrategy
+        ShardRoutingRoleStrategy shardRoutingRoleStrategy,
+        MeterRegistry meterRegistry
     ) {
         this.allocationDeciders = allocationDeciders;
         this.shardsAllocator = shardsAllocator;
         this.clusterInfoService = clusterInfoService;
         this.snapshotsInfoService = snapshotsInfoService;
         this.shardRoutingRoleStrategy = shardRoutingRoleStrategy;
+        this.shardChangesObserver = new ShardChangesObserver(meterRegistry);
     }
 
     /**
@@ -769,7 +773,9 @@ private RoutingAllocation createRoutingAllocation(ClusterState clusterState, lon
             clusterState,
             clusterInfoService.getClusterInfo(),
             snapshotsInfoService.snapshotShardSizes(),
-            currentNanoTime
+            currentNanoTime,
+            false,
+            shardChangesObserver
         );
     }
 

server/src/main/java/org/elasticsearch/cluster/routing/allocation/RoutingAllocation.java

@@ -95,15 +95,15 @@ public RoutingAllocation(
         this(deciders, null, clusterState, clusterInfo, shardSizeInfo, currentNanoTime);
     }
 
-    /**
-     * Creates a new {@link RoutingAllocation}
-     * @param deciders {@link AllocationDeciders} to used to make decisions for routing allocations
-     * @param routingNodes Routing nodes in the current cluster or {@code null} if using those in the given cluster state
-     * @param clusterState cluster state before rerouting
-     * @param clusterInfo information about node disk usage and shard disk usage
-     * @param shardSizeInfo information about snapshot shard sizes
-     * @param currentNanoTime the nano time to use for all delay allocation calculation (typically {@link System#nanoTime()})
-     */
+    /// Creates a new [RoutingAllocation]
+    ///
+    /// @param deciders [AllocationDeciders] to use to make decisions for routing allocations
+    /// @param routingNodes routing nodes in the current cluster or `null` if using those in the given cluster state
+    /// @param clusterState cluster state before rerouting
+    /// @param clusterInfo information about node disk usage and shard disk usage
+    /// @param shardSizeInfo information about snapshot shard sizes
+    /// @param currentNanoTime the nano time to use for all delay allocation calculation (typically `System#nanoTime()`)
+    ///
     public RoutingAllocation(
         AllocationDeciders deciders,
         @Nullable RoutingNodes routingNodes,
@@ -112,25 +112,29 @@ public RoutingAllocation(
         SnapshotShardSizeInfo shardSizeInfo,
         long currentNanoTime
     ) {
-        this(deciders, routingNodes, clusterState, clusterInfo, shardSizeInfo, currentNanoTime, false);
-    }
-
-    /**
-     * Creates a new {@link RoutingAllocation}
-     * @param deciders {@link AllocationDeciders} to used to make decisions for routing allocations
-     * @param routingNodes Routing nodes in the current cluster or {@code null} if using those in the given cluster state
-     * @param clusterState cluster state before rerouting
-     * @param currentNanoTime the nano time to use for all delay allocation calculation (typically {@link System#nanoTime()})
-     * @param isSimulating {@code true} if "transient" deciders should be ignored because we are simulating the final allocation
-     */
+        this(deciders, routingNodes, clusterState, clusterInfo, shardSizeInfo, currentNanoTime, false, RoutingChangesObserver.NOOP);
+    }
+
+    /// Creates a new [RoutingAllocation]
+    ///
+    /// @param deciders [AllocationDeciders] to use to make decisions for routing allocations
+    /// @param routingNodes routing nodes in the current cluster or `null` if using those in the given cluster state
+    /// @param clusterState cluster state before rerouting
+    /// @param clusterInfo information about node disk usage and shard disk usage
+    /// @param shardSizeInfo information about snapshot shard sizes
+    /// @param currentNanoTime the nano time to use for all delay allocation calculation (typically `System#nanoTime()`)
+    /// @param isSimulating `true` if "transient" deciders should be ignored because we are simulating the final allocation
+    /// @param shardChangesObserver observer that records shard state transition timing metrics
+    ///
     public RoutingAllocation(
         AllocationDeciders deciders,
         @Nullable RoutingNodes routingNodes,
         ClusterState clusterState,
         ClusterInfo clusterInfo,
         SnapshotShardSizeInfo shardSizeInfo,
         long currentNanoTime,
-        boolean isSimulating
+        boolean isSimulating,
+        RoutingChangesObserver shardChangesObserver
     ) {
         this.deciders = deciders;
         this.routingNodes = routingNodes;
@@ -154,7 +158,7 @@ public RoutingAllocation(
                     indexMetadataUpdater,
                     restoreInProgressUpdater,
                     resizeSourceIndexUpdater,
-                    new ShardChangesObserver() }
+                    shardChangesObserver }
         );
     }
 
@@ -465,7 +469,8 @@ public RoutingAllocation mutableCloneForSimulation() {
             clusterInfo,
             shardSizeInfo,
             currentNanoTime,
-            true
+            true,
+            RoutingChangesObserver.NOOP
         );
     }
 

server/src/main/java/org/elasticsearch/cluster/routing/allocation/ShardChangesObserver.java

@@ -14,11 +14,51 @@
 import org.elasticsearch.cluster.routing.RoutingChangesObserver;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.UnassignedInfo;
+import org.elasticsearch.telemetry.metric.LongHistogram;
+import org.elasticsearch.telemetry.metric.MeterRegistry;
 
-public class ShardChangesObserver implements RoutingChangesObserver {
+import java.util.Arrays;
+import java.util.Map;
+import java.util.function.LongSupplier;
+import java.util.stream.Collectors;
 
+/// Observes shard state transitions during allocation rounds, logging them and emitting APM timing metrics.
+class ShardChangesObserver implements RoutingChangesObserver {
     private static final Logger logger = LogManager.getLogger(ShardChangesObserver.class);
 
+    static final String UNASSIGNED_TO_INITIALIZING_METRIC = "es.allocator.shards.unassigned_to_initializing.duration.histogram";
+    static final String UNASSIGNED_TO_STARTED_METRIC = "es.allocator.shards.unassigned_to_started.duration.histogram";
+
+    private static final Map<UnassignedInfo.Reason, Map<String, Object>> PRIMARY_ATTRIBUTES = buildAttributesByReason(true);
+    private static final Map<UnassignedInfo.Reason, Map<String, Object>> REPLICA_ATTRIBUTES = buildAttributesByReason(false);
+
+    private static Map<UnassignedInfo.Reason, Map<String, Object>> buildAttributesByReason(boolean primary) {
+        return Arrays.stream(UnassignedInfo.Reason.values())
+            .collect(Collectors.toUnmodifiableMap(r -> r, r -> Map.of("es_shard_primary", primary, "es_shard_reason", r.name())));
+    }
+
+    private final LongHistogram unassignedToInitializingDuration;
+    private final LongHistogram unassignedToStartedDuration;
+    private final LongSupplier currentTimeMillisSupplier;
+
+    ShardChangesObserver(MeterRegistry meterRegistry) {
+        this(meterRegistry, System::currentTimeMillis);
+    }
+
+    ShardChangesObserver(MeterRegistry meterRegistry, LongSupplier currentTimeMillisSupplier) {
+        this.unassignedToInitializingDuration = meterRegistry.registerLongHistogram(
+            UNASSIGNED_TO_INITIALIZING_METRIC,
+            "Duration a shard spent in UNASSIGNED state before being assigned to a node",
+            "ms"
+        );
+        this.unassignedToStartedDuration = meterRegistry.registerLongHistogram(
+            UNASSIGNED_TO_STARTED_METRIC,
+            "Total duration from when a shard became UNASSIGNED to when it became STARTED",
+            "ms"
+        );
+        this.currentTimeMillisSupplier = currentTimeMillisSupplier;
+    }
+
     @Override
     public void shardInitialized(ShardRouting unassignedShard, ShardRouting initializedShard) {
         logger.trace(
@@ -27,6 +67,11 @@ public void shardInitialized(ShardRouting unassignedShard, ShardRouting initiali
             initializedShard.recoverySource().getType(),
             initializedShard.currentNodeId()
         );
+        UnassignedInfo info = unassignedShard.unassignedInfo();
+        if (info != null) {
+            long durationMillis = currentTimeMillisSupplier.getAsLong() - info.unassignedTimeMillis();
+            unassignedToInitializingDuration.record(Math.max(0, durationMillis), attributes(info, initializedShard));
+        }
     }
 
     @Override
@@ -37,6 +82,12 @@ public void shardStarted(ShardRouting initializingShard, ShardRouting startedSha
             initializingShard.recoverySource().getType(),
             startedShard.currentNodeId()
         );
+        // Relocation target shards have no unassignedInfo
+        UnassignedInfo info = initializingShard.unassignedInfo();
+        if (info != null) {
+            long durationMillis = currentTimeMillisSupplier.getAsLong() - info.unassignedTimeMillis();
+            unassignedToStartedDuration.record(Math.max(0, durationMillis), attributes(info, startedShard));
+        }
     }
 
     @Override
@@ -63,4 +114,8 @@ public void replicaPromoted(ShardRouting replicaShard) {
     private static String shardIdentifier(ShardRouting shardRouting) {
         return shardRouting.shardId().toString() + '[' + (shardRouting.primary() ? 'P' : 'R') + ']';
     }
+
+    private static Map<String, Object> attributes(UnassignedInfo info, ShardRouting shard) {
+        return (shard.primary() ? PRIMARY_ATTRIBUTES : REPLICA_ATTRIBUTES).get(info.reason());
+    }
 }

server/src/test/java/org/elasticsearch/cluster/routing/allocation/AllocationServiceTests.java

@@ -44,6 +44,7 @@
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.snapshots.EmptySnapshotsInfoService;
+import org.elasticsearch.telemetry.metric.MeterRegistry;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.gateway.TestGatewayAllocator;
 
@@ -154,7 +155,8 @@ public ShardAllocationDecision explainShardAllocation(ShardRouting shard, Routin
             },
             new EmptyClusterInfoService(),
             EmptySnapshotsInfoService.INSTANCE,
-            TestShardRoutingRoleStrategies.DEFAULT_ROLE_ONLY
+            TestShardRoutingRoleStrategies.DEFAULT_ROLE_ONLY,
+            MeterRegistry.NOOP
         );
 
         final String unrealisticAllocatorName = "unrealistic";
@@ -268,7 +270,8 @@ public void testExplainsNonAllocationOfShardWithUnknownAllocator() {
             null,
             null,
             null,
-            TestShardRoutingRoleStrategies.DEFAULT_ROLE_ONLY
+            TestShardRoutingRoleStrategies.DEFAULT_ROLE_ONLY,
+            MeterRegistry.NOOP
         );
         allocationService.setExistingShardsAllocators(
             Collections.singletonMap(GatewayAllocator.ALLOCATOR_NAME, new TestGatewayAllocator())
@@ -390,7 +393,8 @@ public void testAutoExpandReplicas() throws Exception {
             null,
             new EmptyClusterInfoService(),
             EmptySnapshotsInfoService.INSTANCE,
-            TestShardRoutingRoleStrategies.DEFAULT_ROLE_ONLY
+            TestShardRoutingRoleStrategies.DEFAULT_ROLE_ONLY,
+            MeterRegistry.NOOP
         );
 
         final ProjectId project1 = randomUniqueProjectId();

server/src/test/java/org/elasticsearch/cluster/routing/allocation/ShardChangesObserverTests.java

@@ -23,21 +23,30 @@
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.ShardRoutingState;
+import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.telemetry.InstrumentType;
+import org.elasticsearch.telemetry.Measurement;
+import org.elasticsearch.telemetry.RecordingMeterRegistry;
 import org.elasticsearch.test.MockLog;
 import org.elasticsearch.test.junit.annotations.TestLogging;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static org.elasticsearch.cluster.routing.TestShardRouting.shardRoutingBuilder;
 import static org.elasticsearch.test.MockLog.assertThatLogger;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
 
 @TestLogging(value = "org.elasticsearch.cluster.routing.allocation.ShardChangesObserver:TRACE", reason = "verifies debug level logging")
 public class ShardChangesObserverTests extends ESAllocationTestCase {
 
     public void testLogShardStarting() {
-
         var indexName = randomIdentifier();
         var indexMetadata = IndexMetadata.builder(indexName).settings(indexSettings(IndexVersion.current(), 1, 0)).build();
 
@@ -66,7 +75,6 @@ public void testLogShardStarting() {
     }
 
     public void testLogShardMovement() {
-
         var allocationId = randomUUID();
         var indexName = randomIdentifier();
         var indexMetadata = IndexMetadata.builder(indexName)
@@ -143,4 +151,78 @@ public void testLogShardFailureAndPromotion() {
             )
         );
     }
+
+    public void testUnassignedMetrics() {
+        final var meterRegistry = new RecordingMeterRegistry();
+        final long unassignedAtMillis = randomLongBetween(0, System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1));
+        final AtomicLong nowMillis = new AtomicLong(unassignedAtMillis);
+        final var observer = new ShardChangesObserver(meterRegistry, nowMillis::get);
+
+        final var reason = randomFrom(UnassignedInfo.Reason.values());
+        // ALLOCATION_FAILED needs failedAllocations > 0
+        final int failedAllocations = reason == UnassignedInfo.Reason.ALLOCATION_FAILED ? randomIntBetween(1, 5) : 0;
+        // NODE_RESTARTING needs lastAllocatedNodeId
+        final String lastAllocatedNodeId = reason == UnassignedInfo.Reason.NODE_RESTARTING ? randomIdentifier() : null;
+        final var unassignedInfo = new UnassignedInfo(
+            reason,
+            null,
+            null,
+            failedAllocations,
+            System.nanoTime(),
+            unassignedAtMillis,
+            false,
+            UnassignedInfo.AllocationStatus.NO_ATTEMPT,
+            Collections.emptySet(),
+            lastAllocatedNodeId
+        );
+        final var shardId = new ShardId("test-index", "_na_", 0);
+        final var primary = randomBoolean();
+        final var unassignedShard = shardRoutingBuilder(shardId, null, primary, ShardRoutingState.UNASSIGNED).withUnassignedInfo(
+            unassignedInfo
+        ).build();
+
+        final var recorder = meterRegistry.getRecorder();
+        assertThat(
+            recorder.getMeasurements(InstrumentType.LONG_HISTOGRAM, ShardChangesObserver.UNASSIGNED_TO_INITIALIZING_METRIC).size(),
+            equalTo(0)
+        );
+        assertThat(
+            recorder.getMeasurements(InstrumentType.LONG_HISTOGRAM, ShardChangesObserver.UNASSIGNED_TO_STARTED_METRIC).size(),
+            equalTo(0)
+        );
+
+        // replica shards must recover from primary
+        final var recoverySource = primary ? RecoverySource.EmptyStoreRecoverySource.INSTANCE : RecoverySource.PeerRecoverySource.INSTANCE;
+        final var initializedShard = shardRoutingBuilder(shardId, "node-1", primary, ShardRoutingState.INITIALIZING).withRecoverySource(
+            recoverySource
+        ).withUnassignedInfo(unassignedInfo).build();
+        final long initializedTimeMillis = System.currentTimeMillis();
+        nowMillis.set(initializedTimeMillis);
+        observer.shardInitialized(unassignedShard, initializedShard);
+
+        final var startedShard = shardRoutingBuilder(shardId, "node-1", primary, ShardRoutingState.STARTED).build();
+        final long startedTimeMillis = initializedTimeMillis + randomLongBetween(0, 1000L);
+        nowMillis.set(startedTimeMillis);
+        observer.shardStarted(initializedShard, startedShard);
+
+        final List<Measurement> initializedMetrics = recorder.getMeasurements(
+            InstrumentType.LONG_HISTOGRAM,
+            ShardChangesObserver.UNASSIGNED_TO_INITIALIZING_METRIC
+        );
+        assertThat(initializedMetrics, hasSize(1));
+        final var initializedMetricValue = initializedMetrics.getFirst();
+        assertThat(initializedMetricValue.getLong(), equalTo(Math.max(0, initializedTimeMillis - unassignedAtMillis)));
+        assertThat(initializedMetricValue.attributes().get("es_shard_primary"), equalTo(primary));
+        assertThat(initializedMetricValue.attributes().get("es_shard_reason"), equalTo(reason.name()));
+
+        final List<Measurement> startedMetrics = recorder.getMeasurements(
+            InstrumentType.LONG_HISTOGRAM,
+            ShardChangesObserver.UNASSIGNED_TO_STARTED_METRIC
+        );
+        assertThat(startedMetrics, hasSize(1));
+        final var startedMetricValue = startedMetrics.getFirst();
+        assertThat(startedMetricValue.getLong(), equalTo(Math.max(0, startedTimeMillis - unassignedAtMillis)));
+        assertThat(startedMetricValue.attributes().get("es_shard_primary"), equalTo(primary));
+        assertThat(startedMetricValue.attributes().get("es_shard_reason"), equalTo(reason.name()));
+    }
 }