Introduce searchable snapshots index setting for cascade deletion of snapshots (#74977)

Today there is no relationship between the lifecycle of a 
snapshot mounted as an index and the lifecycle of index 
itself. This lack of relationship makes it possible to delete 
a snapshot in a repository while the mounted index still 
exists, causing the shards to fail in the future.

On the other hand creating a snapshot that contains a 
single index to later mount it as a searchable snapshot 
index becomes more and more natural for users and 
ILM; but it comes with the risk to forget to delete the
 snapshot when the searchable snapshot index is 
deleted from the cluster, "leaking" snapshots in 
repositories.

We'd like to improve the situation and provide a way 
to automatically delete the snapshot when the mounted 
index is deleted. To opt in for this behaviour, a user has 
to enable a specific index setting when mounting the 
snapshot (the proposed name for the setting is 
index.store.snapshot.delete_searchable_snapshot). 

Elasticsearch then verifies that the snapshot to mount 
contains only 1 snapshotted index and that the snapshot 
is not used by another mounted index with a different 
value for the opt in setting, and mounts the snapshot 
with the new private index setting. 

This is the part implemented in this commit.

In follow-up pull requests this index setting will be used 
when the last mounted index is deleted and removed 
from the cluster state in order to add the searchable 
snapshot id to a list of snapshots to delete in the 
repository metadata. Snapshots that are marked as 
"to delete" will not be able to be restored or cloned, and 
Elasticsearch will take care of deleting the snapshot as 
soon as possible. Then ILM will be changed to use this 
setting when mounting a snapshot as a cold or frozen 
index and delete_searchable_snapshot option in ILM 
will be removed. Finally, deleting a snapshot that is 
still used by mounted indices will be prevented.

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

@@ -51,6 +51,7 @@ import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.logging.DeprecationCategory;
@@ -82,6 +83,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -100,6 +102,11 @@ import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF
 import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_SHARDS;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_VERSION_CREATED;
 import static org.elasticsearch.common.util.set.Sets.newHashSet;
+import static org.elasticsearch.index.IndexModule.INDEX_STORE_TYPE_SETTING;
+import static org.elasticsearch.snapshots.SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION;
+import static org.elasticsearch.snapshots.SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOTS_REPOSITORY_NAME_SETTING_KEY;
+import static org.elasticsearch.snapshots.SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOTS_REPOSITORY_UUID_SETTING_KEY;
+import static org.elasticsearch.snapshots.SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOTS_SNAPSHOT_UUID_SETTING_KEY;
 import static org.elasticsearch.snapshots.SnapshotUtils.filterIndices;
 import static org.elasticsearch.snapshots.SnapshotsService.NO_FEATURE_STATES_VALUE;
 
@@ -1015,11 +1022,12 @@ public class RestoreService implements ClusterStateApplier {
         Settings changeSettings,
         String[] ignoreSettings
     ) {
+        final Settings settings = indexMetadata.getSettings();
         Settings normalizedChangeSettings = Settings.builder()
             .put(changeSettings)
             .normalizePrefix(IndexMetadata.INDEX_SETTING_PREFIX)
             .build();
-        if (IndexSettings.INDEX_SOFT_DELETES_SETTING.get(indexMetadata.getSettings())
+        if (IndexSettings.INDEX_SOFT_DELETES_SETTING.get(settings)
             && IndexSettings.INDEX_SOFT_DELETES_SETTING.exists(changeSettings)
             && IndexSettings.INDEX_SOFT_DELETES_SETTING.get(changeSettings) == false) {
             throw new SnapshotRestoreException(
@@ -1027,8 +1035,26 @@ public class RestoreService implements ClusterStateApplier {
                 "cannot disable setting [" + IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey() + "] on restore"
             );
         }
+        if ("snapshot".equals(INDEX_STORE_TYPE_SETTING.get(settings))) {
+            final Boolean changed = changeSettings.getAsBoolean(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION, null);
+            if (changed != null) {
+                final Boolean previous = settings.getAsBoolean(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION, null);
+                if (Objects.equals(previous, changed) == false) {
+                    throw new SnapshotRestoreException(
+                        snapshot,
+                        String.format(
+                            Locale.ROOT,
+                            "cannot change value of [%s] when restoring searchable snapshot [%s:%s] as index %s",
+                            SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION,
+                            snapshot.getRepository(),
+                            snapshot.getSnapshotId().getName(),
+                            indexMetadata.getIndex()
+                        )
+                    );
+                }
+            }
+        }
         IndexMetadata.Builder builder = IndexMetadata.builder(indexMetadata);
-        Settings settings = indexMetadata.getSettings();
         Set<String> keyFilters = new HashSet<>();
         List<String> simpleMatchPatterns = new ArrayList<>();
         for (String ignoredSetting : ignoreSettings) {
@@ -1147,6 +1173,9 @@ public class RestoreService implements ClusterStateApplier {
                 resolveSystemIndicesToDelete(currentState, featureStatesToRestore)
             );
 
+            // List of searchable snapshots indices to restore
+            final Set<Index> searchableSnapshotsIndices = new HashSet<>();
+
             // Updating cluster state
             final Metadata.Builder mdBuilder = Metadata.builder(currentState.metadata());
             final ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
@@ -1236,6 +1265,10 @@ public class RestoreService implements ClusterStateApplier {
                             : new ShardRestoreStatus(localNodeId)
                     );
                 }
+
+                if ("snapshot".equals(INDEX_STORE_TYPE_SETTING.get(updatedIndexMetadata.getSettings()))) {
+                    searchableSnapshotsIndices.add(updatedIndexMetadata.getIndex());
+                }
             }
 
             final ClusterState.Builder builder = ClusterState.builder(currentState);
@@ -1273,10 +1306,11 @@ public class RestoreService implements ClusterStateApplier {
             }
 
             updater.accept(currentState, mdBuilder);
-            return allocationService.reroute(
-                builder.metadata(mdBuilder).blocks(blocks).routingTable(rtBuilder.build()).build(),
-                "restored snapshot [" + snapshot + "]"
-            );
+            final ClusterState updatedClusterState = builder.metadata(mdBuilder).blocks(blocks).routingTable(rtBuilder.build()).build();
+            if (searchableSnapshotsIndices.isEmpty() == false) {
+                ensureSearchableSnapshotsRestorable(updatedClusterState, snapshotInfo, searchableSnapshotsIndices);
+            }
+            return allocationService.reroute(updatedClusterState, "restored snapshot [" + snapshot + "]");
         }
 
         private void applyDataStreamRestores(ClusterState currentState, Metadata.Builder mdBuilder) {
@@ -1494,4 +1528,89 @@ public class RestoreService implements ClusterStateApplier {
         createIndexService.validateDotIndex(renamedIndexName, isHidden);
         createIndexService.validateIndexSettings(renamedIndexName, snapshotIndexMetadata.getSettings(), false);
     }
+
+    private static void ensureSearchableSnapshotsRestorable(
+        final ClusterState currentState,
+        final SnapshotInfo snapshotInfo,
+        final Set<Index> indices
+    ) {
+        final Metadata metadata = currentState.metadata();
+        for (Index index : indices) {
+            final Settings indexSettings = metadata.getIndexSafe(index).getSettings();
+            assert "snapshot".equals(INDEX_STORE_TYPE_SETTING.get(indexSettings)) : "not a snapshot backed index: " + index;
+
+            final String repositoryUuid = indexSettings.get(SEARCHABLE_SNAPSHOTS_REPOSITORY_UUID_SETTING_KEY);
+            final String repositoryName = indexSettings.get(SEARCHABLE_SNAPSHOTS_REPOSITORY_NAME_SETTING_KEY);
+            final String snapshotUuid = indexSettings.get(SEARCHABLE_SNAPSHOTS_SNAPSHOT_UUID_SETTING_KEY);
+
+            final boolean deleteSnapshot = indexSettings.getAsBoolean(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION, false);
+            if (deleteSnapshot && snapshotInfo.indices().size() != 1 && Objects.equals(snapshotUuid, snapshotInfo.snapshotId().getUUID())) {
+                throw new SnapshotRestoreException(
+                    repositoryName,
+                    snapshotInfo.snapshotId().getName(),
+                    String.format(
+                        Locale.ROOT,
+                        "cannot mount snapshot [%s/%s:%s] as index [%s] with the deletion of snapshot on index removal enabled "
+                            + "[index.store.snapshot.delete_searchable_snapshot: true]; snapshot contains [%d] indices instead of 1.",
+                        repositoryName,
+                        repositoryUuid,
+                        snapshotInfo.snapshotId().getName(),
+                        index.getName(),
+                        snapshotInfo.indices().size()
+                    )
+                );
+            }
+
+            for (IndexMetadata other : metadata) {
+                if (other.getIndex().equals(index)) {
+                    continue; // do not check the searchable snapshot index against itself
+                }
+                final Settings otherSettings = other.getSettings();
+                if ("snapshot".equals(INDEX_STORE_TYPE_SETTING.get(otherSettings)) == false) {
+                    continue; // other index is not a searchable snapshot index, skip
+                }
+                final String otherSnapshotUuid = otherSettings.get(SEARCHABLE_SNAPSHOTS_SNAPSHOT_UUID_SETTING_KEY);
+                if (Objects.equals(snapshotUuid, otherSnapshotUuid) == false) {
+                    continue; // other index is backed by a different snapshot, skip
+                }
+                final String otherRepositoryUuid = otherSettings.get(SEARCHABLE_SNAPSHOTS_REPOSITORY_UUID_SETTING_KEY);
+                final String otherRepositoryName = otherSettings.get(SEARCHABLE_SNAPSHOTS_REPOSITORY_NAME_SETTING_KEY);
+                if (matchRepository(repositoryUuid, repositoryName, otherRepositoryUuid, otherRepositoryName) == false) {
+                    continue; // other index is backed by a snapshot from a different repository, skip
+                }
+                final boolean otherDeleteSnap = otherSettings.getAsBoolean(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION, false);
+                if (deleteSnapshot != otherDeleteSnap) {
+                    throw new SnapshotRestoreException(
+                        repositoryName,
+                        snapshotInfo.snapshotId().getName(),
+                        String.format(
+                            Locale.ROOT,
+                            "cannot mount snapshot [%s/%s:%s] as index [%s] with [index.store.snapshot.delete_searchable_snapshot: %b]; "
+                                + "another index %s is mounted with [index.store.snapshot.delete_searchable_snapshot: %b].",
+                            repositoryName,
+                            repositoryUuid,
+                            snapshotInfo.snapshotId().getName(),
+                            index.getName(),
+                            deleteSnapshot,
+                            other.getIndex(),
+                            otherDeleteSnap
+                        )
+                    );
+                }
+            }
+        }
+    }
+
+    private static boolean matchRepository(
+        String repositoryUuid,
+        String repositoryName,
+        String otherRepositoryUuid,
+        String otherRepositoryName
+    ) {
+        if (Strings.hasLength(repositoryUuid) && Strings.hasLength(otherRepositoryUuid)) {
+            return Objects.equals(repositoryUuid, otherRepositoryUuid);
+        } else {
+            return Objects.equals(repositoryName, otherRepositoryName);
+        }
+    }
 }

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

@@ -18,6 +18,9 @@ public final class SearchableSnapshotsSettings {
     public static final String SEARCHABLE_SNAPSHOT_PARTIAL_SETTING_KEY = "index.store.snapshot.partial";
     public static final String SEARCHABLE_SNAPSHOTS_REPOSITORY_NAME_SETTING_KEY = "index.store.snapshot.repository_name";
     public static final String SEARCHABLE_SNAPSHOTS_REPOSITORY_UUID_SETTING_KEY = "index.store.snapshot.repository_uuid";
+    public static final String SEARCHABLE_SNAPSHOTS_SNAPSHOT_NAME_SETTING_KEY = "index.store.snapshot.snapshot_name";
+    public static final String SEARCHABLE_SNAPSHOTS_SNAPSHOT_UUID_SETTING_KEY = "index.store.snapshot.snapshot_uuid";
+    public static final String SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION = "index.store.snapshot.delete_searchable_snapshot";
 
     private SearchableSnapshotsSettings() {}
 

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

@@ -9,20 +9,31 @@ package org.elasticsearch.xpack.searchablesnapshots;
 
 import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
+import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.repositories.fs.FsRepository;
+import org.elasticsearch.snapshots.SnapshotRestoreException;
 
 import java.util.Arrays;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
+import java.util.Set;
 
 import static org.elasticsearch.cluster.metadata.IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING;
 import static org.elasticsearch.index.IndexSettings.INDEX_SOFT_DELETES_SETTING;
+import static org.elasticsearch.snapshots.SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest.Storage;
+import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.nullValue;
 
 public class SearchableSnapshotsRepositoryIntegTests extends BaseFrozenSearchableSnapshotsIntegTestCase {
 
@@ -110,4 +121,303 @@ public class SearchableSnapshotsRepositoryIntegTests extends BaseFrozenSearchabl
 
         assertAcked(clusterAdmin().prepareDeleteRepository(updatedRepositoryName));
     }
+
+    public void testMountIndexWithDeletionOfSnapshotFailsIfNotSingleIndexSnapshot() throws Exception {
+        final String repository = "repository-" + getTestName().toLowerCase(Locale.ROOT);
+        createRepository(repository, FsRepository.TYPE, randomRepositorySettings());
+
+        final int nbIndices = randomIntBetween(2, 5);
+        for (int i = 0; i < nbIndices; i++) {
+            createAndPopulateIndex(
+                "index-" + i,
+                Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0).put(INDEX_SOFT_DELETES_SETTING.getKey(), true)
+            );
+        }
+
+        final String snapshot = "snapshot";
+        createFullSnapshot(repository, snapshot);
+        assertAcked(client().admin().indices().prepareDelete("index-*"));
+
+        final String index = "index-" + randomInt(nbIndices - 1);
+        final String mountedIndex = "mounted-" + index;
+
+        final SnapshotRestoreException exception = expectThrows(
+            SnapshotRestoreException.class,
+            () -> mountSnapshot(repository, snapshot, index, mountedIndex, deleteSnapshotIndexSettings(true), randomFrom(Storage.values()))
+        );
+        assertThat(
+            exception.getMessage(),
+            allOf(
+                containsString("cannot mount snapshot [" + repository + '/'),
+                containsString(snapshot + "] as index [" + mountedIndex + "] with the deletion of snapshot on index removal enabled"),
+                containsString("[index.store.snapshot.delete_searchable_snapshot: true]; "),
+                containsString("snapshot contains [" + nbIndices + "] indices instead of 1.")
+            )
+        );
+    }
+
+    public void testMountIndexWithDifferentDeletionOfSnapshot() throws Exception {
+        final String repository = "repository-" + getTestName().toLowerCase(Locale.ROOT);
+        createRepository(repository, FsRepository.TYPE, randomRepositorySettings());
+
+        final String index = "index";
+        createAndPopulateIndex(index, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true));
+
+        final TotalHits totalHits = internalCluster().client().prepareSearch(index).setTrackTotalHits(true).get().getHits().getTotalHits();
+
+        final String snapshot = "snapshot";
+        createSnapshot(repository, snapshot, List.of(index));
+        assertAcked(client().admin().indices().prepareDelete(index));
+
+        final boolean deleteSnapshot = randomBoolean();
+        final String mounted = "mounted-with-setting-" + deleteSnapshot;
+        final Settings indexSettings = deleteSnapshotIndexSettingsOrNull(deleteSnapshot);
+
+        logger.info("--> mounting index [{}] with index settings [{}]", mounted, indexSettings);
+        mountSnapshot(repository, snapshot, index, mounted, indexSettings, randomFrom(Storage.values()));
+        assertThat(
+            getDeleteSnapshotIndexSetting(mounted),
+            indexSettings.hasValue(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION)
+                ? equalTo(Boolean.toString(deleteSnapshot))
+                : nullValue()
+        );
+        assertHitCount(client().prepareSearch(mounted).setTrackTotalHits(true).get(), totalHits.value);
+
+        final String mountedAgain = randomValueOtherThan(mounted, () -> randomAlphaOfLength(10).toLowerCase(Locale.ROOT));
+        final SnapshotRestoreException exception = expectThrows(
+            SnapshotRestoreException.class,
+            () -> mountSnapshot(repository, snapshot, index, mountedAgain, deleteSnapshotIndexSettings(deleteSnapshot == false))
+        );
+        assertThat(
+            exception.getMessage(),
+            allOf(
+                containsString("cannot mount snapshot [" + repository + '/'),
+                containsString(':' + snapshot + "] as index [" + mountedAgain + "] with "),
+                containsString("[index.store.snapshot.delete_searchable_snapshot: " + (deleteSnapshot == false) + "]; another "),
+                containsString("index [" + mounted + '/'),
+                containsString("is mounted with [index.store.snapshot.delete_searchable_snapshot: " + deleteSnapshot + "].")
+            )
+        );
+
+        mountSnapshot(repository, snapshot, index, mountedAgain, deleteSnapshotIndexSettings(deleteSnapshot));
+        assertThat(
+            getDeleteSnapshotIndexSetting(mounted),
+            indexSettings.hasValue(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION)
+                ? equalTo(Boolean.toString(deleteSnapshot))
+                : nullValue()
+        );
+        assertHitCount(client().prepareSearch(mountedAgain).setTrackTotalHits(true).get(), totalHits.value);
+
+        assertAcked(client().admin().indices().prepareDelete(mountedAgain));
+        assertAcked(client().admin().indices().prepareDelete(mounted));
+    }
+
+    public void testDeletionOfSnapshotSettingCannotBeUpdated() throws Exception {
+        final String repository = "repository-" + getTestName().toLowerCase(Locale.ROOT);
+        createRepository(repository, FsRepository.TYPE, randomRepositorySettings());
+
+        final String index = "index";
+        createAndPopulateIndex(index, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true));
+
+        final TotalHits totalHits = internalCluster().client().prepareSearch(index).setTrackTotalHits(true).get().getHits().getTotalHits();
+
+        final String snapshot = "snapshot";
+        createSnapshot(repository, snapshot, List.of(index));
+        assertAcked(client().admin().indices().prepareDelete(index));
+
+        final String mounted = "mounted-" + index;
+        final boolean deleteSnapshot = randomBoolean();
+        final Settings indexSettings = deleteSnapshotIndexSettingsOrNull(deleteSnapshot);
+
+        mountSnapshot(repository, snapshot, index, mounted, indexSettings, randomFrom(Storage.values()));
+        assertThat(
+            getDeleteSnapshotIndexSetting(mounted),
+            indexSettings.hasValue(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION)
+                ? equalTo(Boolean.toString(deleteSnapshot))
+                : nullValue()
+        );
+        assertHitCount(client().prepareSearch(mounted).setTrackTotalHits(true).get(), totalHits.value);
+
+        if (randomBoolean()) {
+            assertAcked(client().admin().indices().prepareClose(mounted));
+        }
+
+        final IllegalArgumentException exception = expectThrows(
+            IllegalArgumentException.class,
+            () -> client().admin()
+                .indices()
+                .prepareUpdateSettings(mounted)
+                .setSettings(deleteSnapshotIndexSettings(deleteSnapshot == false))
+                .get()
+        );
+        assertThat(
+            exception.getMessage(),
+            containsString("can not update private setting [index.store.snapshot.delete_searchable_snapshot]; ")
+        );
+
+        assertAcked(client().admin().indices().prepareDelete(mounted));
+    }
+
+    public void testRestoreSearchableSnapshotIndexConflicts() throws Exception {
+        final String repository = "repository-" + getTestName().toLowerCase(Locale.ROOT);
+        createRepository(repository, FsRepository.TYPE, randomRepositorySettings());
+
+        final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+        createAndPopulateIndex(indexName, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true));
+
+        final String snapshotOfIndex = "snapshot-of-index";
+        createSnapshot(repository, snapshotOfIndex, List.of(indexName));
+        assertAcked(client().admin().indices().prepareDelete(indexName));
+
+        final String mountedIndex = "mounted-index";
+        final boolean deleteSnapshot = randomBoolean();
+        final Settings indexSettings = deleteSnapshotIndexSettingsOrNull(deleteSnapshot);
+        logger.info("--> mounting snapshot of index [{}] as [{}] with index settings [{}]", indexName, mountedIndex, indexSettings);
+        mountSnapshot(repository, snapshotOfIndex, indexName, mountedIndex, indexSettings, randomFrom(Storage.values()));
+        assertThat(
+            getDeleteSnapshotIndexSetting(mountedIndex),
+            indexSettings.hasValue(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION)
+                ? equalTo(Boolean.toString(deleteSnapshot))
+                : nullValue()
+        );
+
+        final String snapshotOfMountedIndex = "snapshot-of-mounted-index";
+        createSnapshot(repository, snapshotOfMountedIndex, List.of(mountedIndex));
+        assertAcked(client().admin().indices().prepareDelete(mountedIndex));
+
+        final String mountedIndexAgain = "mounted-index-again";
+        final boolean deleteSnapshotAgain = deleteSnapshot == false;
+        final Settings indexSettingsAgain = deleteSnapshotIndexSettings(deleteSnapshotAgain);
+        logger.info("--> mounting snapshot of index [{}] again as [{}] with index settings [{}]", indexName, mountedIndex, indexSettings);
+        mountSnapshot(repository, snapshotOfIndex, indexName, mountedIndexAgain, indexSettingsAgain, randomFrom(Storage.values()));
+        assertThat(
+            getDeleteSnapshotIndexSetting(mountedIndexAgain),
+            indexSettingsAgain.hasValue(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION)
+                ? equalTo(Boolean.toString(deleteSnapshotAgain))
+                : nullValue()
+        );
+
+        logger.info("--> restoring snapshot of searchable snapshot index [{}] should be conflicting", mountedIndex);
+        final SnapshotRestoreException exception = expectThrows(
+            SnapshotRestoreException.class,
+            () -> client().admin()
+                .cluster()
+                .prepareRestoreSnapshot(repository, snapshotOfMountedIndex)
+                .setIndices(mountedIndex)
+                .setWaitForCompletion(true)
+                .get()
+        );
+        assertThat(
+            exception.getMessage(),
+            allOf(
+                containsString("cannot mount snapshot [" + repository + '/'),
+                containsString(':' + snapshotOfMountedIndex + "] as index [" + mountedIndex + "] with "),
+                containsString("[index.store.snapshot.delete_searchable_snapshot: " + deleteSnapshot + "]; another "),
+                containsString("index [" + mountedIndexAgain + '/'),
+                containsString("is mounted with [index.store.snapshot.delete_searchable_snapshot: " + deleteSnapshotAgain + "].")
+            )
+        );
+        assertAcked(client().admin().indices().prepareDelete("mounted-*"));
+    }
+
+    public void testRestoreSearchableSnapshotIndexWithDifferentSettingsConflicts() throws Exception {
+        final String repository = "repository-" + getTestName().toLowerCase(Locale.ROOT);
+        createRepository(repository, FsRepository.TYPE, randomRepositorySettings());
+
+        final int nbIndices = randomIntBetween(1, 3);
+        for (int i = 0; i < nbIndices; i++) {
+            createAndPopulateIndex("index-" + i, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true));
+        }
+
+        final String snapshotOfIndices = "snapshot-of-indices";
+        createFullSnapshot(repository, snapshotOfIndices);
+
+        final int nbMountedIndices = randomIntBetween(1, 3);
+        final Set<String> mountedIndices = new HashSet<>(nbMountedIndices);
+
+        final boolean deleteSnapshot = nbIndices == 1 && randomBoolean();
+        final Settings indexSettings = deleteSnapshotIndexSettingsOrNull(deleteSnapshot);
+
+        for (int i = 0; i < nbMountedIndices; i++) {
+            final String index = "index-" + randomInt(nbIndices - 1);
+            final String mountedIndex = "mounted-" + i;
+            logger.info("--> mounting snapshot of index [{}] as [{}] with index settings [{}]", index, mountedIndex, indexSettings);
+            mountSnapshot(repository, snapshotOfIndices, index, mountedIndex, indexSettings, randomFrom(Storage.values()));
+            assertThat(
+                getDeleteSnapshotIndexSetting(mountedIndex),
+                indexSettings.hasValue(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION)
+                    ? equalTo(Boolean.toString(deleteSnapshot))
+                    : nullValue()
+            );
+            if (randomBoolean()) {
+                assertAcked(client().admin().indices().prepareClose(mountedIndex));
+            }
+            mountedIndices.add(mountedIndex);
+        }
+
+        final String snapshotOfMountedIndices = "snapshot-of-mounted-indices";
+        createSnapshot(repository, snapshotOfMountedIndices, List.of("mounted-*"));
+
+        List<String> restorables = randomBoolean()
+            ? List.of("mounted-*")
+            : randomSubsetOf(randomIntBetween(1, nbMountedIndices), mountedIndices);
+        final SnapshotRestoreException exception = expectThrows(
+            SnapshotRestoreException.class,
+            () -> client().admin()
+                .cluster()
+                .prepareRestoreSnapshot(repository, snapshotOfMountedIndices)
+                .setIndices(restorables.toArray(String[]::new))
+                .setIndexSettings(deleteSnapshotIndexSettings(deleteSnapshot == false))
+                .setRenameReplacement("restored-with-different-setting-$1")
+                .setRenamePattern("(.+)")
+                .setWaitForCompletion(true)
+                .get()
+        );
+
+        assertThat(
+            exception.getMessage(),
+            containsString(
+                "cannot change value of [index.store.snapshot.delete_searchable_snapshot] when restoring searchable snapshot ["
+                    + repository
+                    + ':'
+                    + snapshotOfMountedIndices
+                    + "] as index [mounted-"
+            )
+        );
+
+        final RestoreSnapshotResponse restoreResponse = client().admin()
+            .cluster()
+            .prepareRestoreSnapshot(repository, snapshotOfMountedIndices)
+            .setIndices(restorables.toArray(String[]::new))
+            .setIndexSettings(indexSettings)
+            .setRenameReplacement("restored-with-same-setting-$1")
+            .setRenamePattern("(.+)")
+            .setWaitForCompletion(true)
+            .get();
+        assertThat(restoreResponse.getRestoreInfo().totalShards(), greaterThan(0));
+        assertThat(restoreResponse.getRestoreInfo().failedShards(), equalTo(0));
+
+        assertAcked(client().admin().indices().prepareDelete("mounted-*"));
+        assertAcked(client().admin().indices().prepareDelete("restored-with-same-setting-*"));
+    }
+
+    private static Settings deleteSnapshotIndexSettings(boolean value) {
+        return Settings.builder().put(SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION, value).build();
+    }
+
+    private static Settings deleteSnapshotIndexSettingsOrNull(boolean value) {
+        if (value) {
+            return deleteSnapshotIndexSettings(true);
+        } else if (randomBoolean()) {
+            return deleteSnapshotIndexSettings(false);
+        } else {
+            return Settings.EMPTY;
+        }
+    }
+
+    @Nullable
+    private static String getDeleteSnapshotIndexSetting(String indexName) {
+        final GetSettingsResponse getSettingsResponse = client().admin().indices().prepareGetSettings(indexName).get();
+        return getSettingsResponse.getSetting(indexName, SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION);
+    }
 }

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

@@ -154,13 +154,13 @@ public class SearchableSnapshots extends Plugin implements IndexStorePlugin, Eng
         Setting.Property.NotCopyableOnResize
     );
     public static final Setting<String> SNAPSHOT_SNAPSHOT_NAME_SETTING = Setting.simpleString(
-        "index.store.snapshot.snapshot_name",
+        SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOTS_SNAPSHOT_NAME_SETTING_KEY,
         Setting.Property.IndexScope,
         Setting.Property.PrivateIndex,
         Setting.Property.NotCopyableOnResize
     );
     public static final Setting<String> SNAPSHOT_SNAPSHOT_ID_SETTING = Setting.simpleString(
-        "index.store.snapshot.snapshot_uuid",
+        SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOTS_SNAPSHOT_UUID_SETTING_KEY,
         Setting.Property.IndexScope,
         Setting.Property.PrivateIndex,
         Setting.Property.NotCopyableOnResize
@@ -233,6 +233,19 @@ public class SearchableSnapshots extends Plugin implements IndexStorePlugin, Eng
         Setting.Property.NotCopyableOnResize
     );
 
+    /**
+     * Index setting used to indicate if the snapshot that is mounted as an index should be deleted when the index is deleted. This setting
+     * is only set for indices mounted in clusters on or after 8.0.0. Once set this setting cannot be updated.
+     */
+    public static final Setting<Boolean> DELETE_SEARCHABLE_SNAPSHOT_ON_INDEX_DELETION = Setting.boolSetting(
+        SearchableSnapshotsSettings.SEARCHABLE_SNAPSHOTS_DELETE_SNAPSHOT_ON_INDEX_DELETION,
+        false,
+        Setting.Property.Final,
+        Setting.Property.IndexScope,
+        Setting.Property.PrivateIndex,
+        Setting.Property.NotCopyableOnResize
+    );
+
     /**
      * Prefer to allocate to the data content tier and then the hot tier.
      * This affects the system searchable snapshot cache index (not the searchable snapshot index itself)
@@ -282,6 +295,7 @@ public class SearchableSnapshots extends Plugin implements IndexStorePlugin, Eng
             SNAPSHOT_CACHE_PREWARM_ENABLED_SETTING,
             SNAPSHOT_CACHE_EXCLUDED_FILE_TYPES_SETTING,
             SNAPSHOT_UNCACHED_CHUNK_SIZE_SETTING,
+            DELETE_SEARCHABLE_SNAPSHOT_ON_INDEX_DELETION,
             SearchableSnapshotsConstants.SNAPSHOT_PARTIAL_SETTING,
             SNAPSHOT_BLOB_CACHE_METADATA_FILES_MAX_LENGTH_SETTING,
             CacheService.SNAPSHOT_CACHE_RANGE_SIZE_SETTING,