Reroute when new repository is registered (#73761)

Today we fail to allocate searchable snapshot shards if the repository
containing their underlying data is not registered with the cluster.
This failure is somewhat messy, we allocate them and let the recovery
fail, and furthermore we don't automatically retry the allocation if the
repository is subsequently registered.

This commit introduces an allocation decider to prevent the allocation
of such shards, and explain more clearly why, and also a cluster state
listener that performs a reroute when a new repository is registered.

Relates #73669
Relates #73714

Author: DaveCTurner

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsIntegTests.java

@@ -9,6 +9,7 @@
 import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ResourceNotFoundException;
+import org.elasticsearch.action.admin.cluster.allocation.ClusterAllocationExplanation;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotIndexShardStatus;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotStats;
@@ -20,11 +21,16 @@
 import org.elasticsearch.action.admin.indices.stats.ShardStats;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.RestoreInProgress;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.cluster.routing.allocation.AllocateUnassignedDecision;
+import org.elasticsearch.cluster.routing.allocation.AllocationDecision;
+import org.elasticsearch.cluster.routing.allocation.NodeAllocationResult;
+import org.elasticsearch.cluster.routing.allocation.decider.Decision;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
@@ -971,6 +977,123 @@ public void testSnapshotOfSearchableSnapshotIncludesNoDataButCanBeRestored() thr
         );
     }
 
+    public void testSnapshotOfSearchableSnapshotCanBeRestoredBeforeRepositoryRegistered() throws Exception {
+        final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        createAndPopulateIndex(
+            indexName,
+            Settings.builder().put(INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1).put(INDEX_SOFT_DELETES_SETTING.getKey(), true)
+        );
+
+        final TotalHits originalAllHits = internalCluster().client()
+            .prepareSearch(indexName)
+            .setTrackTotalHits(true)
+            .get()
+            .getHits()
+            .getTotalHits();
+        final TotalHits originalBarHits = internalCluster().client()
+            .prepareSearch(indexName)
+            .setTrackTotalHits(true)
+            .setQuery(matchQuery("foo", "bar"))
+            .get()
+            .getHits()
+            .getTotalHits();
+        logger.info("--> [{}] in total, of which [{}] match the query", originalAllHits, originalBarHits);
+
+        // Take snapshot containing the actual data to one repository
+        final String dataRepoName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        createRepository(dataRepoName, "fs");
+
+        final SnapshotId dataSnapshot = createSnapshot(dataRepoName, "data-snapshot", List.of(indexName)).snapshotId();
+        assertAcked(client().admin().indices().prepareDelete(indexName));
+
+        final String restoredIndexName = randomValueOtherThan(indexName, () -> randomAlphaOfLength(10).toLowerCase(Locale.ROOT));
+        mountSnapshot(dataRepoName, dataSnapshot.getName(), indexName, restoredIndexName, Settings.EMPTY);
+        ensureGreen(restoredIndexName);
+
+        if (randomBoolean()) {
+            logger.info("--> closing index before snapshot");
+            assertAcked(client().admin().indices().prepareClose(restoredIndexName));
+        }
+
+        // Back up the cluster to a different repo
+        final String backupRepoName = randomValueOtherThan(dataRepoName, () -> randomAlphaOfLength(10).toLowerCase(Locale.ROOT));
+        createRepository(backupRepoName, "fs");
+        final SnapshotId backupSnapshot = createSnapshot(backupRepoName, "backup-snapshot", List.of(restoredIndexName)).snapshotId();
+
+        // Clear out data & the repo that contains it
+        final RepositoryMetadata dataRepoMetadata = client().admin()
+            .cluster()
+            .prepareGetRepositories(dataRepoName)
+            .get()
+            .repositories()
+            .get(0);
+        assertAcked(client().admin().indices().prepareDelete(restoredIndexName));
+        assertAcked(client().admin().cluster().prepareDeleteRepository(dataRepoName));
+
+        // Restore the backup snapshot
+        assertThat(
+            client().admin()
+                .cluster()
+                .prepareRestoreSnapshot(backupRepoName, backupSnapshot.getName())
+                .setIndices(restoredIndexName)
+                .get()
+                .status(),
+            equalTo(RestStatus.ACCEPTED)
+        );
+
+        assertBusy(() -> {
+            final ClusterAllocationExplanation clusterAllocationExplanation = client().admin()
+                .cluster()
+                .prepareAllocationExplain()
+                .setIndex(restoredIndexName)
+                .setShard(0)
+                .setPrimary(true)
+                .get()
+                .getExplanation();
+
+            final String description = Strings.toString(clusterAllocationExplanation);
+            final AllocateUnassignedDecision allocateDecision = clusterAllocationExplanation.getShardAllocationDecision()
+                .getAllocateDecision();
+            assertTrue(description, allocateDecision.isDecisionTaken());
+            assertThat(description, allocateDecision.getAllocationDecision(), equalTo(AllocationDecision.NO));
+            for (NodeAllocationResult nodeAllocationResult : allocateDecision.getNodeDecisions()) {
+                for (Decision decision : nodeAllocationResult.getCanAllocateDecision().getDecisions()) {
+                    final String explanation = decision.getExplanation();
+                    if (explanation.contains("this index is backed by a searchable snapshot")
+                        && explanation.contains("no such repository is registered")
+                        && explanation.contains("the required repository was originally named [" + dataRepoName + "]")) {
+                        return;
+                    }
+                }
+            }
+
+            fail(description);
+        });
+
+        assertBusy(() -> {
+            final RestoreInProgress restoreInProgress = client().admin()
+                .cluster()
+                .prepareState()
+                .clear()
+                .setCustoms(true)
+                .get()
+                .getState()
+                .custom(RestoreInProgress.TYPE, RestoreInProgress.EMPTY);
+            assertTrue(Strings.toString(restoreInProgress, true, true), restoreInProgress.isEmpty());
+        });
+
+        // Re-register the repository containing the actual data & verify that the shards are now allocated
+        final String newRepositoryName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        final Settings.Builder settings = Settings.builder().put(dataRepoMetadata.settings());
+        if (randomBoolean()) {
+            settings.put(READONLY_SETTING_KEY, "true");
+        }
+        assertAcked(clusterAdmin().preparePutRepository(newRepositoryName).setType("fs").setSettings(settings));
+
+        ensureGreen(restoredIndexName);
+        assertTotalHits(restoredIndexName, originalAllHits, originalBarHits);
+    }
+
     private void assertSearchableSnapshotStats(String indexName, boolean cacheEnabled, List<String> nonCachedExtensions) {
         final SearchableSnapshotsStatsResponse statsResponse = client().execute(
             SearchableSnapshotsStatsAction.INSTANCE,

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshots.java

@@ -9,11 +9,20 @@
 import org.apache.lucene.store.BufferedIndexInput;
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.cluster.ClusterChangedEvent;
+import org.elasticsearch.cluster.ClusterStateListener;
+import org.elasticsearch.cluster.metadata.RepositoriesMetadata;
+import org.elasticsearch.cluster.metadata.RepositoryMetadata;
+import org.elasticsearch.cluster.routing.RerouteService;
+import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.repositories.RepositoryData;
 import org.elasticsearch.xpack.searchablesnapshots.action.cache.TransportSearchableSnapshotsNodeCachesStatsAction;
 import org.elasticsearch.xpack.searchablesnapshots.allocation.decider.DedicatedFrozenNodeAllocationDecider;
+import org.elasticsearch.xpack.searchablesnapshots.allocation.decider.SearchableSnapshotRepositoryExistsAllocationDecider;
 import org.elasticsearch.xpack.searchablesnapshots.cache.blob.BlobStoreCacheService;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
@@ -108,12 +117,15 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.index.mapper.MapperService.SINGLE_MAPPING_NAME;
@@ -344,7 +356,10 @@ public Collection<Object> createComponents(
         this.allocator.set(new SearchableSnapshotAllocator(client, clusterService.getRerouteService(), frozenCacheInfoService));
         components.add(new FrozenCacheServiceSupplier(frozenCacheService.get()));
         components.add(new CacheServiceSupplier(cacheService.get()));
-        new SearchableSnapshotIndexMetadataUpgrader(clusterService, threadPool).initialize();
+        if (DiscoveryNode.isMasterNode(settings)) {
+            new SearchableSnapshotIndexMetadataUpgrader(clusterService, threadPool).initialize();
+            clusterService.addListener(new RepositoryUuidWatcher(clusterService.getRerouteService()));
+        }
         return Collections.unmodifiableList(components);
     }
 
@@ -519,6 +534,7 @@ protected XPackLicenseState getLicenseState() {
     public Collection<AllocationDecider> createAllocationDeciders(Settings settings, ClusterSettings clusterSettings) {
         return List.of(
             new SearchableSnapshotAllocationDecider(() -> getLicenseState().isAllowed(XPackLicenseState.Feature.SEARCHABLE_SNAPSHOTS)),
+            new SearchableSnapshotRepositoryExistsAllocationDecider(),
             new SearchableSnapshotEnableAllocationDecider(settings, clusterSettings),
             new HasFrozenCacheAllocationDecider(frozenCacheInfoService),
             new DedicatedFrozenNodeAllocationDecider()
@@ -697,4 +713,34 @@ public FrozenCacheService get() {
             return frozenCacheService;
         }
     }
+
+    private static final class RepositoryUuidWatcher implements ClusterStateListener {
+
+        private final RerouteService rerouteService;
+        private final HashSet<String> knownUuids = new HashSet<>();
+
+        RepositoryUuidWatcher(RerouteService rerouteService) {
+            this.rerouteService = rerouteService;
+        }
+
+        @Override
+        public void clusterChanged(ClusterChangedEvent event) {
+            final RepositoriesMetadata repositoriesMetadata = event.state().metadata().custom(RepositoriesMetadata.TYPE);
+            if (repositoriesMetadata == null) {
+                knownUuids.clear();
+                return;
+            }
+
+            final Set<String> newUuids = repositoriesMetadata.repositories()
+                .stream()
+                .map(RepositoryMetadata::uuid)
+                .filter(s -> s.equals(RepositoryData.MISSING_UUID) == false)
+                .collect(Collectors.toSet());
+            if (knownUuids.addAll(newUuids)) {
+                rerouteService.reroute("repository UUIDs changed", Priority.NORMAL, ActionListener.wrap((() -> {})));
+            }
+            knownUuids.retainAll(newUuids);
+            assert knownUuids.equals(newUuids) : knownUuids + " vs " + newUuids;
+        }
+    }
 }