Add consistency checker for stuck snapshot debugging (elastic#124616)

Checks the local cluster state after marking a shard snapshot as
complete, and logs a message if the completion is not reflected in this
cluster state.

Author: DaveCTurner

server/src/internalClusterTest/java/org/elasticsearch/snapshots/SnapshotShardsServiceIT.java

@@ -9,17 +9,29 @@
 
 package org.elasticsearch.snapshots;
 
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionTestUtils;
+import org.elasticsearch.action.support.SubscribableListener;
+import org.elasticsearch.cluster.coordination.Coordinator;
 import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.snapshots.mockstore.MockRepository;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.MockLog;
 import org.elasticsearch.test.disruption.NetworkDisruption;
+import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.test.transport.MockTransportService;
+import org.elasticsearch.transport.TestTransportChannel;
+import org.elasticsearch.transport.TransportResponse;
 
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.equalTo;
@@ -89,4 +101,71 @@ public void testRetryPostingSnapshotStatusMessages() throws Exception {
             assertThat(snapshotInfo.successfulShards(), equalTo(shards));
         }, 30L, TimeUnit.SECONDS);
     }
+
+    @TestLogging(
+        reason = "verifying debug logging",
+        value = "org.elasticsearch.snapshots.SnapshotShardsService.ShardStatusConsistencyChecker:DEBUG"
+    )
+    public void testConsistencyCheckerLogging() {
+        final var masterNode = internalCluster().startMasterOnlyNode();
+        final var dataNode = internalCluster().startDataOnlyNode();
+
+        final var repoName = randomIdentifier();
+        createRepository(repoName, "fs");
+
+        final var indexName = randomIdentifier();
+        assertAcked(prepareCreate(indexName, 0, indexSettingsNoReplicas(between(1, 5))));
+        indexRandomDocs(indexName, scaledRandomIntBetween(50, 100));
+
+        // allow cluster states to go through normally until we see a shard snapshot update
+        final var shardUpdateSeen = new AtomicBoolean();
+        MockTransportService.getInstance(masterNode)
+            .addRequestHandlingBehavior(SnapshotsService.UPDATE_SNAPSHOT_STATUS_ACTION_NAME, (handler, request, channel, task) -> {
+                shardUpdateSeen.set(true);
+                handler.messageReceived(request, channel, task);
+            });
+
+        // then, after the shard snapshot update, delay cluster state update commits on the data node until we see the
+        // ShardStatusConsistencyChecker log an inconsistency.
+        final var logMessageSeenListener = new SubscribableListener<Void>();
+        MockTransportService.getInstance(dataNode)
+            .addRequestHandlingBehavior(Coordinator.COMMIT_STATE_ACTION_NAME, (handler, request, channel, task) -> {
+                if (shardUpdateSeen.get() == false || logMessageSeenListener.isDone()) {
+                    handler.messageReceived(request, channel, task);
+                } else {
+                    // delay the commit ...
+                    logMessageSeenListener.addListener(
+                        ActionTestUtils.assertNoFailureListener(
+                            ignored -> ActionListener.run(
+                                ActionListener.<TransportResponse>noop(),
+                                l -> handler.messageReceived(request, new TestTransportChannel(l), task)
+                            )
+                        )
+                    );
+                    // ... and send a failure straight back to the master so it applies and acks the state update anyway
+                    channel.sendResponse(new ElasticsearchException("simulated"));
+                }
+            });
+
+        MockLog.awaitLogger(
+            () -> createSnapshot(repoName, randomIdentifier(), List.of(indexName)),
+            SnapshotShardsService.ShardStatusConsistencyChecker.class,
+            new MockLog.SeenEventExpectation(
+                "debug log",
+                SnapshotShardsService.ShardStatusConsistencyChecker.class.getCanonicalName(),
+                Level.DEBUG,
+                "*unexpectedly still in state [*INIT*] after notifying master"
+            ) {
+                @Override
+                public boolean innerMatch(LogEvent event) {
+                    if (event.getMessage().getFormattedMessage().contains("after notifying master")) {
+                        // now that the inconsistency was logged, release the cluster state commits on the data node
+                        logMessageSeenListener.onResponse(null);
+                    }
+                    return true;
+                }
+            }
+        );
+
+    }
 }

server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java

@@ -58,9 +58,14 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 
 import static java.util.Collections.emptyMap;
@@ -841,6 +846,11 @@ public void onResponse(Void aVoid) {
                     Strings.format("successfully sent shard snapshot state [%s] update to the master node", status.state())
                 );
                 logger.trace("[{}][{}] updated snapshot state to [{}]", shardId, snapshot, status);
+                if (status.state().completed()) {
+                    shardStatusConsistencyChecker.ensureShardComplete(snapshot, shardId);
+                } else {
+                    assert status.state() == ShardState.PAUSED_FOR_NODE_REMOVAL : status;
+                }
             }
 
             @Override
@@ -868,4 +878,89 @@ public void onFailure(Exception e) {
             )
         );
     }
+
+    private final ShardStatusConsistencyChecker shardStatusConsistencyChecker = new ShardStatusConsistencyChecker();
+
+    /**
+     * After receiving an ack from the master confirming that it marked a shard snapshot as complete, checks the local cluster state to
+     * confirm that it was updated to reflect this and logs a message if not. Not 100% watertight because this node might be lagging behind
+     * and hasn't received the cluster state yet, or else it might have been removed from the cluster, but both of these things will also
+     * be visible in the logs. See e.g. o.e.c.c.Publication#logIncompleteNodes and o.e.c.c.NodeLeftExecutor.
+     */
+    // visible for testing
+    class ShardStatusConsistencyChecker {
+
+        // dedicated logger so we can separate these logs from other SnapshotShardsService DEBUG logs
+        private static final Logger CONSISTENCY_CHECKER_LOGGER = LogManager.getLogger(ShardStatusConsistencyChecker.class);
+
+        private record CheckTask(Snapshot snapshot, ShardId shardId) {}
+
+        private final AtomicInteger queuedTaskCount = new AtomicInteger(0); // atomic to ensure max one thread processing the queue
+        private final Queue<CheckTask> queue = new ConcurrentLinkedQueue<>();
+
+        /**
+         * Adds a {@link CheckTask} with the snapshot info to the queue. If the queue was previously empty, then {@link #runCheck()} will be
+         * started on a GENERIC thread.
+         */
+        void ensureShardComplete(Snapshot snapshot, ShardId shardId) {
+            if (CONSISTENCY_CHECKER_LOGGER.isDebugEnabled() == false) {
+                return;
+            }
+
+            if (queuedTaskCount.get() > 1000) {
+                // racy check, we only need an approximate limit
+                return;
+            }
+
+            queue.add(new CheckTask(snapshot, shardId));
+            if (queuedTaskCount.getAndIncrement() == 0) {
+                threadPool.generic().execute(this::runCheck);
+            } // else a runCheck is already running somewhere and will pick up the task we just added
+        }
+
+        /**
+         * Empties the {@link #queue} and verifies that the cluster state shows a successful shard snapshot for each {@link CheckTask}.
+         * A debug level message is logged if the cluster state does not show a completed shard snapshot: this could be a bug, or a race
+         * with a cluster state update not being applied yet.
+         */
+        private void runCheck() {
+            while (true) {
+                final var taskCount = queuedTaskCount.get();
+                final var clusterStateSnapshotsInProgress = SnapshotsInProgress.get(clusterService.state());
+                final var shardsBySnapshot = new HashMap<Snapshot, Set<ShardId>>();
+                for (int i = 0; i < taskCount; i++) {
+                    final var task = queue.poll();
+                    assert task != null;
+                    shardsBySnapshot.computeIfAbsent(task.snapshot(), ignored -> new HashSet<>()).add(task.shardId());
+                }
+                for (final var snapshotShards : shardsBySnapshot.entrySet()) {
+                    final var snapshot = snapshotShards.getKey();
+                    final var clusterStateSnapshotEntry = clusterStateSnapshotsInProgress.snapshot(snapshot);
+                    if (clusterStateSnapshotEntry != null) {
+                        for (final var shardId : snapshotShards.getValue()) {
+                            final var clusterStateShardStatus = clusterStateSnapshotEntry.shards().get(shardId);
+                            if (clusterStateShardStatus == null) {
+                                CONSISTENCY_CHECKER_LOGGER.debug(
+                                    "shard [{}] in snapshot [{}] unexpectedly not found (should be impossible)",
+                                    shardId,
+                                    snapshot
+                                );
+                            } else if (clusterStateShardStatus.state().completed() == false) {
+                                CONSISTENCY_CHECKER_LOGGER.debug(
+                                    "shard [{}] in snapshot [{}] unexpectedly still in state [{}] after notifying master",
+                                    shardId,
+                                    snapshot,
+                                    clusterStateShardStatus
+                                );
+                            } // else shard is marked complete as expected
+                        }
+                    } // else snapshot already completed & removed from cluster state
+                }
+
+                if (queuedTaskCount.addAndGet(-taskCount) == 0) {
+                    return;
+                } // else someone added some more tasks, so keep trying
+            }
+        }
+    }
 }