Allow single-index analysis

Author: DaveCTurner

server/src/internalClusterTest/java/org/elasticsearch/repositories/blobstore/BlobStoreMetadataIntegrityIT.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.admin.cluster.repositories.integrity.VerifyRepositoryIntegrityAction;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
@@ -93,7 +94,7 @@ public void testIntegrityCheck() throws Exception {
             createFullSnapshot(REPOSITORY_NAME, "test-snapshot-" + snapshotIndex);
         }
 
-        final var request = new VerifyRepositoryIntegrityAction.Request(REPOSITORY_NAME, 5, 5, 5, 5, 10000, false);
+        final var request = new VerifyRepositoryIntegrityAction.Request(REPOSITORY_NAME, Strings.EMPTY_ARRAY, 5, 5, 5, 5, 10000, false);
 
         final var response = PlainActionFuture.<VerifyRepositoryIntegrityAction.Response, RuntimeException>get(
             listener -> client().execute(VerifyRepositoryIntegrityAction.INSTANCE, request, listener),

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/integrity/VerifyRepositoryIntegrityAction.java

@@ -42,6 +42,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 public class VerifyRepositoryIntegrityAction extends ActionType<VerifyRepositoryIntegrityAction.Response> {
 
@@ -55,6 +56,7 @@ private VerifyRepositoryIntegrityAction() {
     public static class Request extends MasterNodeReadRequest<Request> {
 
         private final String repository;
+        private final String[] indices;
         private final int threadpoolConcurrency;
         private final int snapshotVerificationConcurrency;
         private final int indexVerificationConcurrency;
@@ -64,6 +66,7 @@ public static class Request extends MasterNodeReadRequest<Request> {
 
         public Request(
             String repository,
+            String[] indices,
             int threadpoolConcurrency,
             int snapshotVerificationConcurrency,
             int indexVerificationConcurrency,
@@ -72,6 +75,7 @@ public Request(
             boolean permitMissingSnapshotDetails
         ) {
             this.repository = repository;
+            this.indices = Objects.requireNonNull(indices, "indices");
             this.threadpoolConcurrency = requireMin("threadpoolConcurrency", 0, threadpoolConcurrency);
             this.snapshotVerificationConcurrency = requireMin("snapshotVerificationConcurrency", 1, snapshotVerificationConcurrency);
             this.indexVerificationConcurrency = requireMin("indexVerificationConcurrency", 1, indexVerificationConcurrency);
@@ -94,6 +98,7 @@ private static int requireMin(String name, int min, int value) {
         public Request(StreamInput in) throws IOException {
             super(in);
             this.repository = in.readString();
+            this.indices = in.readStringArray();
             this.threadpoolConcurrency = in.readVInt();
             this.snapshotVerificationConcurrency = in.readVInt();
             this.indexVerificationConcurrency = in.readVInt();
@@ -106,6 +111,7 @@ public Request(StreamInput in) throws IOException {
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeString(repository);
+            out.writeStringArray(indices);
             out.writeVInt(threadpoolConcurrency);
             out.writeVInt(snapshotVerificationConcurrency);
             out.writeVInt(indexVerificationConcurrency);
@@ -124,6 +130,14 @@ public Task createTask(long id, String type, String action, TaskId parentTaskId,
             return new CancellableTask(id, type, action, getDescription(), parentTaskId, headers);
         }
 
+        public String getRepository() {
+            return repository;
+        }
+
+        public String[] getIndices() {
+            return indices;
+        }
+
         public int getThreadpoolConcurrency() {
             return threadpoolConcurrency;
         }
@@ -152,6 +166,7 @@ public Request withDefaultThreadpoolConcurrency(Settings settings) {
             if (threadpoolConcurrency == 0) {
                 final var request = new Request(
                     repository,
+                    indices,
                     Math.max(1, EsExecutors.allocatedProcessors(settings) / 2),
                     snapshotVerificationConcurrency,
                     indexVerificationConcurrency,
@@ -249,7 +264,7 @@ protected ClusterBlockException checkBlock(Request request, ClusterState state)
         protected void masterOperation(Task task, Request request, ClusterState state, ActionListener<Response> listener) throws Exception {
             // TODO add mechanism to block blob deletions while this is running
             final var cancellableTask = (CancellableTask) task;
-            repositoriesService.repository(request.repository)
+            repositoriesService.repository(request.getRepository())
                 .verifyMetadataIntegrity(
                     request.withDefaultThreadpoolConcurrency(clusterService.getSettings()),
                     listener.map(Response::new),

server/src/main/java/org/elasticsearch/repositories/blobstore/MetadataVerifier.java

@@ -69,6 +69,7 @@ class MetadataVerifier implements Releasable {
     private final ProgressLogger snapshotProgressLogger;
     private final ProgressLogger indexProgressLogger;
     private final ProgressLogger indexSnapshotProgressLogger;
+    private final Set<String> requestedIndices;
 
     MetadataVerifier(
         BlobStoreRepository blobStoreRepository,
@@ -92,6 +93,8 @@ class MetadataVerifier implements Releasable {
         this.snapshotProgressLogger = new ProgressLogger("snapshots", repositoryData.getSnapshotIds().size(), 100);
         this.indexProgressLogger = new ProgressLogger("indices", repositoryData.getIndices().size(), 20);
         this.indexSnapshotProgressLogger = new ProgressLogger("index snapshots", repositoryData.getIndexSnapshotCount(), 1000);
+
+        this.requestedIndices = Set.of(verifyRequest.getIndices());
     }
 
     @Override
@@ -136,7 +139,11 @@ public void run() {
             indexSnapshotProgressLogger.getExpectedMax()
         );
 
-        verifySnapshots();
+        if (requestedIndices.isEmpty()) {
+            verifySnapshots();
+        } else {
+            verifyIndices();
+        }
     }
 
     private void verifySnapshots() {
@@ -231,6 +238,10 @@ private class IndexVerifier {
         }
 
         void run() {
+            if (requestedIndices.isEmpty() == false && requestedIndices.contains(indexId.getName()) == false) {
+                return;
+            }
+
             runThrottled(
                 makeVoidListener(indexRefs, () -> {}),
                 repositoryData.getSnapshots(indexId).iterator(),

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestVerifyRepositoryIntegrityAction.java

@@ -10,6 +10,7 @@
 
 import org.elasticsearch.action.admin.cluster.repositories.integrity.VerifyRepositoryIntegrityAction;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestCancellableNodeClient;
@@ -35,6 +36,7 @@ public String getName() {
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         final var verifyRequest = new VerifyRepositoryIntegrityAction.Request(
             request.param("repository"),
+            request.paramAsStringArray("indices", Strings.EMPTY_ARRAY),
             request.paramAsInt("threadpool_concurrency", 0),
             request.paramAsInt("snapshot_verification_concurrency", 5),
             request.paramAsInt("index_verification_concurrency", 5),