From 431bdeb1bb17090206ed2e26a486b0ddd00f08bb Mon Sep 17 00:00:00 2001 From: Andrew Ross Date: Wed, 12 Oct 2022 21:03:49 -0700 Subject: [PATCH] Introduce experimental searchable snapshot API (#4680) This commit adds a new parameter to the snapshot restore API to restore to a new type of "remote snapshot" index where, unlike traditional snapshot restores, the index data is not all downloaded to disk and instead is read on-demand at search time. The feature is functional with this commit, and includes a simple end-to-end integration test, but is far from complete. See tracking issue #2919 for the rest of the work planned/underway. All new capabilities are gated behind a new searchable snapshot feature flag. Signed-off-by: Andrew Ross Signed-off-by: Andrew Ross --- CHANGELOG.md | 2 + .../snapshots/SearchableSnapshotIT.java | 135 ++++++++++++++ .../restore/RestoreSnapshotRequest.java | 78 +++++++- .../RestoreSnapshotRequestBuilder.java | 8 + .../cluster/routing/RecoverySource.java | 33 +++- .../common/settings/IndexScopedSettings.java | 11 +- .../opensearch/common/util/FeatureFlags.java | 8 + ...ransportNodesListGatewayStartedShards.java | 3 +- .../org/opensearch/index/IndexModule.java | 54 ++++-- .../org/opensearch/index/IndexSettings.java | 24 +++ .../opensearch/index/shard/IndexShard.java | 35 ++-- .../RemoveCorruptedShardDataCommand.java | 3 +- .../index/shard/ShardStateMetadata.java | 47 ++++- .../opensearch/index/shard/StoreRecovery.java | 3 - .../InMemoryRemoteSnapshotDirectory.java | 169 ++++++++++++++++++ .../store/RemoteSnapshotDirectoryFactory.java | 49 +++++ .../opensearch/indices/IndicesService.java | 12 +- .../main/java/org/opensearch/node/Node.java | 20 ++- .../opensearch/snapshots/RestoreService.java | 42 ++++- .../index/shard/IndexShardTests.java | 10 +- .../index/shard/ShardPathTests.java | 15 +- .../index/store/FsDirectoryFactoryTests.java | 4 +- .../test/store/MockFSDirectoryFactory.java | 7 +- 23 files changed, 705 insertions(+), 67 deletions(-) create mode 100644 server/src/internalClusterTest/java/org/opensearch/snapshots/SearchableSnapshotIT.java create mode 100644 server/src/main/java/org/opensearch/index/store/InMemoryRemoteSnapshotDirectory.java create mode 100644 server/src/main/java/org/opensearch/index/store/RemoteSnapshotDirectoryFactory.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 09d7cb8ff79df..4310103803fbe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,8 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) - Recommission API changes for service layer ([#4320](https://github.com/opensearch-project/OpenSearch/pull/4320)) - Update GeoGrid base class access modifier to support extensibility ([#4572](https://github.com/opensearch-project/OpenSearch/pull/4572)) - Add a new node role 'search' which is dedicated to provide search capability ([#4689](https://github.com/opensearch-project/OpenSearch/pull/4689)) +- Introduce experimental searchable snapshot API ([#4680](https://github.com/opensearch-project/OpenSearch/pull/4680)) + ### Dependencies - Bumps `log4j-core` from 2.18.0 to 2.19.0 - Bumps `reactor-netty-http` from 1.0.18 to 1.0.23 diff --git a/server/src/internalClusterTest/java/org/opensearch/snapshots/SearchableSnapshotIT.java b/server/src/internalClusterTest/java/org/opensearch/snapshots/SearchableSnapshotIT.java new file mode 100644 index 0000000000000..96fcf0053c9ab --- /dev/null +++ b/server/src/internalClusterTest/java/org/opensearch/snapshots/SearchableSnapshotIT.java @@ -0,0 +1,135 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package org.opensearch.snapshots; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; + +import org.hamcrest.MatcherAssert; +import org.junit.BeforeClass; +import org.opensearch.action.admin.cluster.node.stats.NodesStatsResponse; +import org.opensearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse; +import org.opensearch.action.admin.cluster.snapshots.restore.RestoreSnapshotRequest; +import org.opensearch.client.Client; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.cluster.routing.GroupShardsIterator; +import org.opensearch.cluster.routing.ShardIterator; +import org.opensearch.cluster.routing.ShardRouting; +import org.opensearch.common.io.PathUtils; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.util.FeatureFlags; +import org.opensearch.index.Index; +import org.opensearch.monitor.fs.FsInfo; + +import com.carrotsearch.randomizedtesting.generators.RandomPicks; + +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.is; +import static org.opensearch.action.admin.cluster.node.stats.NodesStatsRequest.Metric.FS; +import static org.opensearch.common.util.CollectionUtils.iterableAsArrayList; + +public final class SearchableSnapshotIT extends AbstractSnapshotIntegTestCase { + + @BeforeClass + public static void assumeFeatureFlag() { + assumeTrue( + "Searchable snapshot feature flag is enabled", + Boolean.parseBoolean(System.getProperty(FeatureFlags.SEARCHABLE_SNAPSHOT)) + ); + } + + @Override + protected boolean addMockInternalEngine() { + return false; + } + + public void testCreateSearchableSnapshot() throws Exception { + final Client client = client(); + createRepository("test-repo", "fs"); + createIndex( + "test-idx-1", + Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, "0").put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, "1").build() + ); + createIndex( + "test-idx-2", + Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, "0").put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, "1").build() + ); + ensureGreen(); + indexRandomDocs("test-idx-1", 100); + indexRandomDocs("test-idx-2", 100); + + logger.info("--> snapshot"); + final CreateSnapshotResponse createSnapshotResponse = client.admin() + .cluster() + .prepareCreateSnapshot("test-repo", "test-snap") + .setWaitForCompletion(true) + .setIndices("test-idx-1", "test-idx-2") + .get(); + MatcherAssert.assertThat(createSnapshotResponse.getSnapshotInfo().successfulShards(), greaterThan(0)); + MatcherAssert.assertThat( + createSnapshotResponse.getSnapshotInfo().successfulShards(), + equalTo(createSnapshotResponse.getSnapshotInfo().totalShards()) + ); + + assertTrue(client.admin().indices().prepareDelete("test-idx-1", "test-idx-2").get().isAcknowledged()); + + logger.info("--> restore indices as 'remote_snapshot'"); + client.admin() + .cluster() + .prepareRestoreSnapshot("test-repo", "test-snap") + .setRenamePattern("(.+)") + .setRenameReplacement("$1-copy") + .setStorageType(RestoreSnapshotRequest.StorageType.REMOTE_SNAPSHOT) + .setWaitForCompletion(true) + .execute() + .actionGet(); + ensureGreen(); + + assertDocCount("test-idx-1-copy", 100L); + assertDocCount("test-idx-2-copy", 100L); + assertIndexDirectoryDoesNotExist("test-idx-1-copy", "test-idx-2-copy"); + } + + /** + * Picks a shard out of the cluster state for each given index and asserts + * that the 'index' directory does not exist in the node's file system. + * This assertion is digging a bit into the implementation details to + * verify that the Lucene segment files are not copied from the snapshot + * repository to the node's local disk for a remote snapshot index. + */ + private void assertIndexDirectoryDoesNotExist(String... indexNames) { + final ClusterState state = client().admin().cluster().prepareState().get().getState(); + for (String indexName : indexNames) { + final Index index = state.metadata().index(indexName).getIndex(); + // Get the primary shards for the given index + final GroupShardsIterator shardIterators = state.getRoutingTable() + .activePrimaryShardsGrouped(new String[] { indexName }, false); + // Randomly pick one of the shards + final List iterators = iterableAsArrayList(shardIterators); + final ShardIterator shardIterator = RandomPicks.randomFrom(random(), iterators); + final ShardRouting shardRouting = shardIterator.nextOrNull(); + assertNotNull(shardRouting); + assertTrue(shardRouting.primary()); + assertTrue(shardRouting.assignedToNode()); + // Get the file system stats for the assigned node + final String nodeId = shardRouting.currentNodeId(); + final NodesStatsResponse nodeStats = client().admin().cluster().prepareNodesStats(nodeId).addMetric(FS.metricName()).get(); + for (FsInfo.Path info : nodeStats.getNodes().get(0).getFs()) { + // Build the expected path for the index data for a "normal" + // index and assert it does not exist + final String path = info.getPath(); + final Path file = PathUtils.get(path) + .resolve("indices") + .resolve(index.getUUID()) + .resolve(Integer.toString(shardRouting.getId())) + .resolve("index"); + MatcherAssert.assertThat("Expect file not to exist: " + file, Files.exists(file), is(false)); + } + } + } +} diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java index 1b673217a248b..3ecf5ab19c0e4 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequest.java @@ -33,6 +33,7 @@ package org.opensearch.action.admin.cluster.snapshots.restore; import org.opensearch.LegacyESVersion; +import org.opensearch.Version; import org.opensearch.action.ActionRequestValidationException; import org.opensearch.action.support.IndicesOptions; import org.opensearch.action.support.clustermanager.ClusterManagerNodeRequest; @@ -42,6 +43,7 @@ import org.opensearch.common.io.stream.StreamOutput; import org.opensearch.common.logging.DeprecationLogger; import org.opensearch.common.settings.Settings; +import org.opensearch.common.util.FeatureFlags; import org.opensearch.common.xcontent.ToXContentObject; import org.opensearch.common.xcontent.XContentBuilder; import org.opensearch.common.xcontent.XContentType; @@ -68,6 +70,38 @@ public class RestoreSnapshotRequest extends ClusterManagerNodeRequest source) { } else { throw new IllegalArgumentException("malformed ignore_index_settings section, should be an array of strings"); } + } else if (name.equals("storage_type")) { + if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT)) { + if (entry.getValue() instanceof String) { + storageType(StorageType.fromString((String) entry.getValue())); + } else { + throw new IllegalArgumentException("malformed storage_type"); + } + } else { + throw new IllegalArgumentException( + "Unsupported parameter " + name + ". Feature flag is not enabled for this experimental feature" + ); + } } else { if (IndicesOptions.isIndicesOptions(name) == false) { throw new IllegalArgumentException("Unknown parameter " + name); @@ -579,6 +648,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws builder.value(ignoreIndexSetting); } builder.endArray(); + if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT) && storageType != null) { + storageType.toXContent(builder); + } builder.endObject(); return builder; } @@ -605,7 +677,8 @@ public boolean equals(Object o) { && Objects.equals(renameReplacement, that.renameReplacement) && Objects.equals(indexSettings, that.indexSettings) && Arrays.equals(ignoreIndexSettings, that.ignoreIndexSettings) - && Objects.equals(snapshotUuid, that.snapshotUuid); + && Objects.equals(snapshotUuid, that.snapshotUuid) + && Objects.equals(storageType, that.storageType); } @Override @@ -621,7 +694,8 @@ public int hashCode() { partial, includeAliases, indexSettings, - snapshotUuid + snapshotUuid, + storageType ); result = 31 * result + Arrays.hashCode(indices); result = 31 * result + Arrays.hashCode(ignoreIndexSettings); diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java index 68397851699fb..0104637a00035 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/snapshots/restore/RestoreSnapshotRequestBuilder.java @@ -248,4 +248,12 @@ public RestoreSnapshotRequestBuilder setIgnoreIndexSettings(List ignoreI request.ignoreIndexSettings(ignoreIndexSettings); return this; } + + /** + * Sets the storage type + */ + public RestoreSnapshotRequestBuilder setStorageType(RestoreSnapshotRequest.StorageType storageType) { + request.storageType(storageType); + return this; + } } diff --git a/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java b/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java index 539773296ed74..728bf9d1ae90e 100644 --- a/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java +++ b/server/src/main/java/org/opensearch/cluster/routing/RecoverySource.java @@ -38,6 +38,7 @@ import org.opensearch.common.io.stream.StreamInput; import org.opensearch.common.io.stream.StreamOutput; import org.opensearch.common.io.stream.Writeable; +import org.opensearch.common.util.FeatureFlags; import org.opensearch.common.xcontent.ToXContent; import org.opensearch.common.xcontent.ToXContentObject; import org.opensearch.common.xcontent.XContentBuilder; @@ -257,12 +258,24 @@ public static class SnapshotRecoverySource extends RecoverySource { private final Snapshot snapshot; private final IndexId index; private final Version version; + private final boolean isSearchableSnapshot; public SnapshotRecoverySource(String restoreUUID, Snapshot snapshot, Version version, IndexId indexId) { + this(restoreUUID, snapshot, version, indexId, false); + } + + public SnapshotRecoverySource( + String restoreUUID, + Snapshot snapshot, + Version version, + IndexId indexId, + boolean isSearchableSnapshot + ) { this.restoreUUID = restoreUUID; this.snapshot = Objects.requireNonNull(snapshot); this.version = Objects.requireNonNull(version); this.index = Objects.requireNonNull(indexId); + this.isSearchableSnapshot = isSearchableSnapshot; } SnapshotRecoverySource(StreamInput in) throws IOException { @@ -274,6 +287,11 @@ public SnapshotRecoverySource(String restoreUUID, Snapshot snapshot, Version ver } else { index = new IndexId(in.readString(), IndexMetadata.INDEX_UUID_NA_VALUE); } + if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT) && in.getVersion().onOrAfter(Version.V_3_0_0)) { + isSearchableSnapshot = in.readBoolean(); + } else { + isSearchableSnapshot = false; + } } public String restoreUUID() { @@ -298,6 +316,10 @@ public Version version() { return version; } + public boolean isSearchableSnapshot() { + return isSearchableSnapshot; + } + @Override protected void writeAdditionalFields(StreamOutput out) throws IOException { out.writeString(restoreUUID); @@ -308,6 +330,9 @@ protected void writeAdditionalFields(StreamOutput out) throws IOException { } else { out.writeString(index.getName()); } + if (FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT) && out.getVersion().onOrAfter(Version.V_3_0_0)) { + out.writeBoolean(isSearchableSnapshot); + } } @Override @@ -321,7 +346,8 @@ public void addAdditionalFields(XContentBuilder builder, ToXContent.Params param .field("snapshot", snapshot.getSnapshotId().getName()) .field("version", version.toString()) .field("index", index.getName()) - .field("restoreUUID", restoreUUID); + .field("restoreUUID", restoreUUID) + .field("isSearchableSnapshot", isSearchableSnapshot); } @Override @@ -342,12 +368,13 @@ public boolean equals(Object o) { return restoreUUID.equals(that.restoreUUID) && snapshot.equals(that.snapshot) && index.equals(that.index) - && version.equals(that.version); + && version.equals(that.version) + && isSearchableSnapshot == that.isSearchableSnapshot; } @Override public int hashCode() { - return Objects.hash(restoreUUID, snapshot, index, version); + return Objects.hash(restoreUUID, snapshot, index, version, isSearchableSnapshot); } } diff --git a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java index 7be9adc786f24..079fc38415328 100644 --- a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java @@ -221,12 +221,19 @@ public final class IndexScopedSettings extends AbstractScopedSettings { */ public static final Map> FEATURE_FLAGGED_INDEX_SETTINGS = Map.of( FeatureFlags.REPLICATION_TYPE, - Collections.singletonList(IndexMetadata.INDEX_REPLICATION_TYPE_SETTING), + List.of(IndexMetadata.INDEX_REPLICATION_TYPE_SETTING), FeatureFlags.REMOTE_STORE, - Arrays.asList( + List.of( IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING, IndexMetadata.INDEX_REMOTE_TRANSLOG_STORE_ENABLED_SETTING, IndexMetadata.INDEX_REMOTE_STORE_REPOSITORY_SETTING + ), + FeatureFlags.SEARCHABLE_SNAPSHOT, + List.of( + IndexSettings.SEARCHABLE_SNAPSHOT_REPOSITORY, + IndexSettings.SEARCHABLE_SNAPSHOT_INDEX_ID, + IndexSettings.SEARCHABLE_SNAPSHOT_ID_NAME, + IndexSettings.SEARCHABLE_SNAPSHOT_ID_UUID ) ); diff --git a/server/src/main/java/org/opensearch/common/util/FeatureFlags.java b/server/src/main/java/org/opensearch/common/util/FeatureFlags.java index fa39dc9ac5aa0..7297479776da9 100644 --- a/server/src/main/java/org/opensearch/common/util/FeatureFlags.java +++ b/server/src/main/java/org/opensearch/common/util/FeatureFlags.java @@ -29,6 +29,14 @@ public class FeatureFlags { */ public static final String REMOTE_STORE = "opensearch.experimental.feature.remote_store.enabled"; + /** + * Gates the functionality of a new parameter to the snapshot restore API + * that allows for creation of a new index type that searches a snapshot + * directly in a remote repository without restoring all index data to disk + * ahead of time. + */ + public static final String SEARCHABLE_SNAPSHOT = "opensearch.experimental.feature.searchable_snapshot.enabled"; + /** * Used to test feature flags whose values are expected to be booleans. * This method returns true if the value is "true" (case-insensitive), diff --git a/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java b/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java index 5b79ca5970e63..fb114bff9aa12 100644 --- a/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java +++ b/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java @@ -159,7 +159,8 @@ protected NodeGatewayStartedShards nodeOperation(NodeRequest request) { nodeEnv.availableShardPaths(request.shardId) ); if (shardStateMetadata != null) { - if (indicesService.getShardOrNull(shardId) == null) { + if (indicesService.getShardOrNull(shardId) == null + && shardStateMetadata.indexDataLocation == ShardStateMetadata.IndexDataLocation.LOCAL) { final String customDataPath; if (request.getCustomDataPath() != null) { customDataPath = request.getCustomDataPath(); diff --git a/server/src/main/java/org/opensearch/index/IndexModule.java b/server/src/main/java/org/opensearch/index/IndexModule.java index e52a2ba39ed52..2a462f6165184 100644 --- a/server/src/main/java/org/opensearch/index/IndexModule.java +++ b/server/src/main/java/org/opensearch/index/IndexModule.java @@ -70,12 +70,14 @@ import org.opensearch.index.shard.SearchOperationListener; import org.opensearch.index.similarity.SimilarityService; import org.opensearch.index.store.FsDirectoryFactory; +import org.opensearch.index.store.RemoteSnapshotDirectoryFactory; import org.opensearch.indices.IndicesQueryCache; import org.opensearch.indices.breaker.CircuitBreakerService; import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache; import org.opensearch.indices.mapper.MapperRegistry; import org.opensearch.indices.recovery.RecoveryState; import org.opensearch.plugins.IndexStorePlugin; +import org.opensearch.repositories.RepositoriesService; import org.opensearch.script.ScriptService; import org.opensearch.search.aggregations.support.ValuesSourceRegistry; import org.opensearch.threadpool.ThreadPool; @@ -94,6 +96,7 @@ import java.util.function.BooleanSupplier; import java.util.function.Consumer; import java.util.function.Function; +import java.util.function.Supplier; /** * IndexModule represents the central extension point for index level custom implementations like: @@ -390,15 +393,6 @@ IndexEventListener freeze() { // pkg private for testing } } - public static boolean isBuiltinType(String storeType) { - for (Type type : Type.values()) { - if (type.match(storeType)) { - return true; - } - } - return false; - } - /** * Type of file system * @@ -409,7 +403,8 @@ public enum Type { NIOFS("niofs"), MMAPFS("mmapfs"), SIMPLEFS("simplefs"), - FS("fs"); + FS("fs"), + REMOTE_SNAPSHOT("remote_snapshot"); private final String settingsKey; private final boolean deprecated; @@ -426,7 +421,7 @@ public enum Type { private static final Map TYPES; static { - final Map types = new HashMap<>(4); + final Map types = new HashMap<>(values().length); for (final Type type : values()) { types.put(type.settingsKey, type); } @@ -441,6 +436,10 @@ public boolean isDeprecated() { return deprecated; } + static boolean containsSettingsKey(String key) { + return TYPES.containsKey(key); + } + public static Type fromSettingsKey(final String key) { final Type type = TYPES.get(key); if (type == null) { @@ -459,6 +458,13 @@ public boolean match(String setting) { return getSettingsKey().equals(setting); } + /** + * Convenience method to check whether the given IndexSettings contains + * an {@link #INDEX_STORE_TYPE_SETTING} set to the value of this type. + */ + public boolean match(IndexSettings settings) { + return match(INDEX_STORE_TYPE_SETTING.get(settings.getSettings())); + } } public static Type defaultStoreType(final boolean allowMmap) { @@ -562,7 +568,7 @@ private static IndexStorePlugin.DirectoryFactory getDirectoryFactory( if (storeType.isEmpty() || Type.FS.getSettingsKey().equals(storeType)) { type = defaultStoreType(allowMmap); } else { - if (isBuiltinType(storeType)) { + if (Type.containsSettingsKey(storeType)) { type = Type.fromSettingsKey(storeType); } else { type = null; @@ -572,7 +578,7 @@ private static IndexStorePlugin.DirectoryFactory getDirectoryFactory( throw new IllegalArgumentException("store type [" + storeType + "] is not allowed because mmap is disabled"); } final IndexStorePlugin.DirectoryFactory factory; - if (storeType.isEmpty() || isBuiltinType(storeType)) { + if (storeType.isEmpty()) { factory = DEFAULT_DIRECTORY_FACTORY; } else { factory = indexStoreFactories.get(storeType); @@ -641,4 +647,26 @@ private void ensureNotFrozen() { } } + public static Map createBuiltInDirectoryFactories( + Supplier repositoriesService + ) { + final Map factories = new HashMap<>(); + for (Type type : Type.values()) { + switch (type) { + case HYBRIDFS: + case NIOFS: + case FS: + case MMAPFS: + case SIMPLEFS: + factories.put(type.getSettingsKey(), DEFAULT_DIRECTORY_FACTORY); + break; + case REMOTE_SNAPSHOT: + factories.put(type.getSettingsKey(), new RemoteSnapshotDirectoryFactory(repositoriesService)); + break; + default: + throw new IllegalStateException("No directory factory mapping for built-in type " + type); + } + } + return factories; + } } diff --git a/server/src/main/java/org/opensearch/index/IndexSettings.java b/server/src/main/java/org/opensearch/index/IndexSettings.java index 00daea147f16f..7648f0a192ce7 100644 --- a/server/src/main/java/org/opensearch/index/IndexSettings.java +++ b/server/src/main/java/org/opensearch/index/IndexSettings.java @@ -550,6 +550,30 @@ public final class IndexSettings { Property.Dynamic ); + public static final Setting SEARCHABLE_SNAPSHOT_REPOSITORY = Setting.simpleString( + "index.searchable_snapshot.repository", + Property.IndexScope, + Property.InternalIndex + ); + + public static final Setting SEARCHABLE_SNAPSHOT_ID_UUID = Setting.simpleString( + "index.searchable_snapshot.snapshot_id.uuid", + Property.IndexScope, + Property.InternalIndex + ); + + public static final Setting SEARCHABLE_SNAPSHOT_ID_NAME = Setting.simpleString( + "index.searchable_snapshot.snapshot_id.name", + Property.IndexScope, + Property.InternalIndex + ); + + public static final Setting SEARCHABLE_SNAPSHOT_INDEX_ID = Setting.simpleString( + "index.searchable_snapshot.index.id", + Property.IndexScope, + Property.InternalIndex + ); + private final Index index; private final Version version; private final Logger logger; diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java index 52ecc5bc66607..3a3c4b19a02f6 100644 --- a/server/src/main/java/org/opensearch/index/shard/IndexShard.java +++ b/server/src/main/java/org/opensearch/index/shard/IndexShard.java @@ -678,7 +678,7 @@ public void onFailure(Exception e) { this.shardRouting = newRouting; assert this.shardRouting.primary() == false || this.shardRouting.started() == false || // note that we use started and not - // active to avoid relocating shards + // active to avoid relocating shards this.indexShardOperationPermits.isBlocked() || // if permits are blocked, we are still transitioning this.replicationTracker.isPrimaryMode() : "a started primary with non-pending operation term must be in primary mode " + this.shardRouting; @@ -1990,7 +1990,12 @@ public void openEngineAndRecoverFromTranslog() throws IOException { translogRecoveryStats::incrementRecoveredOperations ); }; - loadGlobalCheckpointToReplicationTracker(); + + // Do not load the global checkpoint if this is a remote snapshot index + if (IndexModule.Type.REMOTE_SNAPSHOT.match(indexSettings) == false) { + loadGlobalCheckpointToReplicationTracker(); + } + innerOpenEngineAndTranslog(replicationTracker); getEngine().translogManager() .recoverFromTranslog(translogRecoveryRunner, getEngine().getProcessedLocalCheckpoint(), Long.MAX_VALUE); @@ -3089,13 +3094,18 @@ public void startRecovery( } break; case SNAPSHOT: - final String repo = ((SnapshotRecoverySource) recoveryState.getRecoverySource()).snapshot().getRepository(); - executeRecovery( - "from snapshot", - recoveryState, - recoveryListener, - l -> restoreFromRepository(repositoriesService.repository(repo), l) - ); + final SnapshotRecoverySource recoverySource = (SnapshotRecoverySource) recoveryState.getRecoverySource(); + if (recoverySource.isSearchableSnapshot()) { + executeRecovery("from snapshot (remote)", recoveryState, recoveryListener, this::recoverFromStore); + } else { + final String repo = recoverySource.snapshot().getRepository(); + executeRecovery( + "from snapshot", + recoveryState, + recoveryListener, + l -> restoreFromRepository(repositoriesService.repository(repo), l) + ); + } break; case LOCAL_SHARDS: final IndexMetadata indexMetadata = indexSettings().getIndexMetadata(); @@ -3256,10 +3266,15 @@ private static void persistMetadata( writeReason = "routing changed from " + currentRouting + " to " + newRouting; } logger.trace("{} writing shard state, reason [{}]", shardId, writeReason); + + final ShardStateMetadata.IndexDataLocation indexDataLocation = IndexSettings.SEARCHABLE_SNAPSHOT_REPOSITORY.exists( + indexSettings.getSettings() + ) ? ShardStateMetadata.IndexDataLocation.REMOTE : ShardStateMetadata.IndexDataLocation.LOCAL; final ShardStateMetadata newShardStateMetadata = new ShardStateMetadata( newRouting.primary(), indexSettings.getUUID(), - newRouting.allocationId() + newRouting.allocationId(), + indexDataLocation ); ShardStateMetadata.FORMAT.writeAndCleanup(newShardStateMetadata, shardPath.getShardStatePath()); } else { diff --git a/server/src/main/java/org/opensearch/index/shard/RemoveCorruptedShardDataCommand.java b/server/src/main/java/org/opensearch/index/shard/RemoveCorruptedShardDataCommand.java index ccc620fc8cf64..c7e380f842fa0 100644 --- a/server/src/main/java/org/opensearch/index/shard/RemoveCorruptedShardDataCommand.java +++ b/server/src/main/java/org/opensearch/index/shard/RemoveCorruptedShardDataCommand.java @@ -484,7 +484,8 @@ private void newAllocationId(ShardPath shardPath, Terminal terminal) throws IOEx final ShardStateMetadata newShardStateMetadata = new ShardStateMetadata( shardStateMetadata.primary, shardStateMetadata.indexUUID, - newAllocationId + newAllocationId, + ShardStateMetadata.IndexDataLocation.LOCAL ); ShardStateMetadata.FORMAT.writeAndCleanup(newShardStateMetadata, shardStatePath); diff --git a/server/src/main/java/org/opensearch/index/shard/ShardStateMetadata.java b/server/src/main/java/org/opensearch/index/shard/ShardStateMetadata.java index 9cd9149cda913..9e334bc6ffd54 100644 --- a/server/src/main/java/org/opensearch/index/shard/ShardStateMetadata.java +++ b/server/src/main/java/org/opensearch/index/shard/ShardStateMetadata.java @@ -56,17 +56,39 @@ public final class ShardStateMetadata { private static final String PRIMARY_KEY = "primary"; private static final String INDEX_UUID_KEY = "index_uuid"; private static final String ALLOCATION_ID_KEY = "allocation_id"; + private static final String INDEX_DATA_LOCATION_KEY = "index_data_location"; + + /** + * Enumeration of types of data locations for an index + */ + public enum IndexDataLocation { + /** + * Indicates index data is on the local disk + */ + LOCAL, + /** + * Indicates index data is remote, such as for a searchable snapshot + * index + */ + REMOTE + } public final String indexUUID; public final boolean primary; @Nullable public final AllocationId allocationId; // can be null if we read from legacy format (see fromXContent and MultiDataPathUpgrader) + public final IndexDataLocation indexDataLocation; public ShardStateMetadata(boolean primary, String indexUUID, AllocationId allocationId) { + this(primary, indexUUID, allocationId, IndexDataLocation.LOCAL); + } + + public ShardStateMetadata(boolean primary, String indexUUID, AllocationId allocationId, IndexDataLocation indexDataLocation) { assert indexUUID != null; this.primary = primary; this.indexUUID = indexUUID; this.allocationId = allocationId; + this.indexDataLocation = Objects.requireNonNull(indexDataLocation); } @Override @@ -89,6 +111,9 @@ public boolean equals(Object o) { if (Objects.equals(allocationId, that.allocationId) == false) { return false; } + if (Objects.equals(indexDataLocation, that.indexDataLocation) == false) { + return false; + } return true; } @@ -98,17 +123,16 @@ public int hashCode() { int result = indexUUID.hashCode(); result = 31 * result + (allocationId != null ? allocationId.hashCode() : 0); result = 31 * result + (primary ? 1 : 0); + result = 31 * result + indexDataLocation.hashCode(); return result; } @Override public String toString() { - return "primary [" + primary + "], allocation [" + allocationId + "]"; + return "primary [" + primary + "], allocation [" + allocationId + "], index data location [" + indexDataLocation + "]"; } - public static final MetadataStateFormat FORMAT = new MetadataStateFormat( - SHARD_STATE_FILE_PREFIX - ) { + public static final MetadataStateFormat FORMAT = new MetadataStateFormat<>(SHARD_STATE_FILE_PREFIX) { @Override protected XContentBuilder newXContentBuilder(XContentType type, OutputStream stream) throws IOException { @@ -124,6 +148,11 @@ public void toXContent(XContentBuilder builder, ShardStateMetadata shardStateMet if (shardStateMetadata.allocationId != null) { builder.field(ALLOCATION_ID_KEY, shardStateMetadata.allocationId); } + // Omit the index data location field if it is LOCAL (the implicit default) + // to maintain compatibility for local indices + if (shardStateMetadata.indexDataLocation != IndexDataLocation.LOCAL) { + builder.field(INDEX_DATA_LOCATION_KEY, shardStateMetadata.indexDataLocation); + } } @Override @@ -136,6 +165,7 @@ public ShardStateMetadata fromXContent(XContentParser parser) throws IOException String currentFieldName = null; String indexUUID = IndexMetadata.INDEX_UUID_NA_VALUE; AllocationId allocationId = null; + IndexDataLocation indexDataLocation = IndexDataLocation.LOCAL; while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) { if (token == XContentParser.Token.FIELD_NAME) { currentFieldName = parser.currentName(); @@ -144,6 +174,13 @@ public ShardStateMetadata fromXContent(XContentParser parser) throws IOException primary = parser.booleanValue(); } else if (INDEX_UUID_KEY.equals(currentFieldName)) { indexUUID = parser.text(); + } else if (INDEX_DATA_LOCATION_KEY.equals(currentFieldName)) { + final String stringValue = parser.text(); + try { + indexDataLocation = IndexDataLocation.valueOf(stringValue); + } catch (IllegalArgumentException e) { + throw new CorruptStateException("unexpected value for data location [" + stringValue + "]"); + } } else { throw new CorruptStateException("unexpected field in shard state [" + currentFieldName + "]"); } @@ -160,7 +197,7 @@ public ShardStateMetadata fromXContent(XContentParser parser) throws IOException if (primary == null) { throw new CorruptStateException("missing value for [primary] in shard state"); } - return new ShardStateMetadata(primary, indexUUID, allocationId); + return new ShardStateMetadata(primary, indexUUID, allocationId, indexDataLocation); } }; } diff --git a/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java b/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java index 06916c4cc87fe..6ca5036808818 100644 --- a/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java +++ b/server/src/main/java/org/opensearch/index/shard/StoreRecovery.java @@ -104,9 +104,6 @@ final class StoreRecovery { */ void recoverFromStore(final IndexShard indexShard, ActionListener listener) { if (canRecover(indexShard)) { - RecoverySource.Type recoveryType = indexShard.recoveryState().getRecoverySource().getType(); - assert recoveryType == RecoverySource.Type.EMPTY_STORE || recoveryType == RecoverySource.Type.EXISTING_STORE - : "expected store recovery type but was: " + recoveryType; ActionListener.completeWith(recoveryListener(indexShard, listener), () -> { logger.debug("starting recovery from store ..."); internalRecoverFromStore(indexShard); diff --git a/server/src/main/java/org/opensearch/index/store/InMemoryRemoteSnapshotDirectory.java b/server/src/main/java/org/opensearch/index/store/InMemoryRemoteSnapshotDirectory.java new file mode 100644 index 0000000000000..0757d88a4099a --- /dev/null +++ b/server/src/main/java/org/opensearch/index/store/InMemoryRemoteSnapshotDirectory.java @@ -0,0 +1,169 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.index.store; + +import java.io.IOException; +import java.io.InputStream; +import java.util.Collection; +import java.util.Collections; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.IOContext; +import org.apache.lucene.store.IndexInput; +import org.apache.lucene.store.IndexOutput; +import org.apache.lucene.store.Lock; +import org.apache.lucene.store.NoLockFactory; +import org.apache.lucene.util.SetOnce; +import org.opensearch.common.blobstore.BlobContainer; +import org.opensearch.common.blobstore.BlobPath; +import org.opensearch.common.lucene.store.ByteArrayIndexInput; +import org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot; +import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.opensearch.snapshots.SnapshotId; + +/** + * Trivial in-memory implementation of a Directory that reads from a snapshot + * in a repository. This is functional but is only temporary to demonstrate + * functional searchable snapshot functionality. The proper implementation will + * be implemented per https://github.com/opensearch-project/OpenSearch/issues/3114. + * + * @opensearch.internal + */ +public final class InMemoryRemoteSnapshotDirectory extends Directory { + + private final BlobStoreRepository blobStoreRepository; + private final SnapshotId snapshotId; + private final BlobPath blobPath; + private final SetOnce blobContainer = new SetOnce<>(); + private final SetOnce> fileInfoMap = new SetOnce<>(); + + public InMemoryRemoteSnapshotDirectory(BlobStoreRepository blobStoreRepository, BlobPath blobPath, SnapshotId snapshotId) { + this.blobStoreRepository = blobStoreRepository; + this.snapshotId = snapshotId; + this.blobPath = blobPath; + } + + @Override + public String[] listAll() throws IOException { + return fileInfoMap().keySet().toArray(new String[0]); + } + + @Override + public void deleteFile(String name) throws IOException {} + + @Override + public IndexOutput createOutput(String name, IOContext context) { + return NoopIndexOutput.INSTANCE; + } + + @Override + public IndexInput openInput(String name, IOContext context) throws IOException { + final BlobStoreIndexShardSnapshot.FileInfo fileInfo = fileInfoMap().get(name); + + // Virtual files are contained entirely in the metadata hash field + if (fileInfo.name().startsWith("v__")) { + return new ByteArrayIndexInput(name, fileInfo.metadata().hash().bytes); + } + + try (InputStream is = blobContainer().readBlob(fileInfo.name())) { + return new ByteArrayIndexInput(name, is.readAllBytes()); + } + } + + @Override + public void close() throws IOException {} + + @Override + public long fileLength(String name) throws IOException { + initializeIfNecessary(); + return fileInfoMap.get().get(name).length(); + } + + @Override + public Set getPendingDeletions() throws IOException { + return Collections.emptySet(); + } + + @Override + public IndexOutput createTempOutput(String prefix, String suffix, IOContext context) { + throw new UnsupportedOperationException(); + } + + @Override + public void sync(Collection names) throws IOException {} + + @Override + public void syncMetaData() {} + + @Override + public void rename(String source, String dest) throws IOException {} + + @Override + public Lock obtainLock(String name) throws IOException { + return NoLockFactory.INSTANCE.obtainLock(null, null); + } + + static class NoopIndexOutput extends IndexOutput { + + final static NoopIndexOutput INSTANCE = new NoopIndexOutput(); + + NoopIndexOutput() { + super("noop", "noop"); + } + + @Override + public void close() throws IOException { + + } + + @Override + public long getFilePointer() { + return 0; + } + + @Override + public long getChecksum() throws IOException { + return 0; + } + + @Override + public void writeByte(byte b) throws IOException { + + } + + @Override + public void writeBytes(byte[] b, int offset, int length) throws IOException { + + } + } + + private BlobContainer blobContainer() { + initializeIfNecessary(); + return blobContainer.get(); + } + + private Map fileInfoMap() { + initializeIfNecessary(); + return fileInfoMap.get(); + } + + /** + * Bit of a hack to lazily initialize the blob store to avoid running afoul + * of the assertion in {@code BlobStoreRepository#assertSnapshotOrGenericThread}. + */ + private void initializeIfNecessary() { + if (blobContainer.get() == null || fileInfoMap.get() == null) { + blobContainer.set(blobStoreRepository.blobStore().blobContainer(blobPath)); + final BlobStoreIndexShardSnapshot snapshot = blobStoreRepository.loadShardSnapshot(blobContainer.get(), snapshotId); + fileInfoMap.set( + snapshot.indexFiles().stream().collect(Collectors.toMap(BlobStoreIndexShardSnapshot.FileInfo::physicalName, f -> f)) + ); + } + } +} diff --git a/server/src/main/java/org/opensearch/index/store/RemoteSnapshotDirectoryFactory.java b/server/src/main/java/org/opensearch/index/store/RemoteSnapshotDirectoryFactory.java new file mode 100644 index 0000000000000..bf7806b836b65 --- /dev/null +++ b/server/src/main/java/org/opensearch/index/store/RemoteSnapshotDirectoryFactory.java @@ -0,0 +1,49 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.index.store; + +import java.io.IOException; +import java.util.function.Supplier; + +import org.apache.lucene.store.Directory; +import org.opensearch.common.blobstore.BlobPath; +import org.opensearch.index.IndexSettings; +import org.opensearch.index.shard.ShardPath; +import org.opensearch.plugins.IndexStorePlugin; +import org.opensearch.repositories.RepositoriesService; +import org.opensearch.repositories.Repository; +import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.opensearch.snapshots.SnapshotId; + +/** + * Factory for a Directory implementation that can read directly from index + * data stored remotely in a repository. + * + * @opensearch.internal + */ +public final class RemoteSnapshotDirectoryFactory implements IndexStorePlugin.DirectoryFactory { + private final Supplier repositoriesService; + + public RemoteSnapshotDirectoryFactory(Supplier repositoriesService) { + this.repositoriesService = repositoriesService; + } + + @Override + public Directory newDirectory(IndexSettings indexSettings, ShardPath shardPath) throws IOException { + final String repositoryName = IndexSettings.SEARCHABLE_SNAPSHOT_REPOSITORY.get(indexSettings.getSettings()); + final Repository repository = repositoriesService.get().repository(repositoryName); + assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository"; + final BlobStoreRepository blobStoreRepository = (BlobStoreRepository) repository; + final BlobPath blobPath = new BlobPath().add("indices") + .add(IndexSettings.SEARCHABLE_SNAPSHOT_INDEX_ID.get(indexSettings.getSettings())) + .add(Integer.toString(shardPath.getShardId().getId())); + final SnapshotId snapshotId = new SnapshotId( + IndexSettings.SEARCHABLE_SNAPSHOT_ID_NAME.get(indexSettings.getSettings()), + IndexSettings.SEARCHABLE_SNAPSHOT_ID_UUID.get(indexSettings.getSettings()) + ); + return new InMemoryRemoteSnapshotDirectory(blobStoreRepository, blobPath, snapshotId); + } +} diff --git a/server/src/main/java/org/opensearch/indices/IndicesService.java b/server/src/main/java/org/opensearch/indices/IndicesService.java index f2961c0f3b13d..b2f48ccdd389c 100644 --- a/server/src/main/java/org/opensearch/indices/IndicesService.java +++ b/server/src/main/java/org/opensearch/indices/IndicesService.java @@ -109,6 +109,7 @@ import org.opensearch.index.engine.InternalEngineFactory; import org.opensearch.index.engine.NRTReplicationEngineFactory; import org.opensearch.index.engine.NoOpEngine; +import org.opensearch.index.engine.ReadOnlyEngine; import org.opensearch.index.fielddata.IndexFieldDataCache; import org.opensearch.index.flush.FlushStats; import org.opensearch.index.get.GetStats; @@ -130,6 +131,7 @@ import org.opensearch.index.shard.IndexingOperationListener; import org.opensearch.index.shard.IndexingStats; import org.opensearch.index.shard.ShardId; +import org.opensearch.index.translog.TranslogStats; import org.opensearch.indices.breaker.CircuitBreakerService; import org.opensearch.indices.cluster.IndicesClusterStateService; import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache; @@ -336,13 +338,6 @@ public void onRemoval(ShardId shardId, String fieldName, boolean wasEvicted, lon this.metaStateService = metaStateService; this.engineFactoryProviders = engineFactoryProviders; - // do not allow any plugin-provided index store type to conflict with a built-in type - for (final String indexStoreType : directoryFactories.keySet()) { - if (IndexModule.isBuiltinType(indexStoreType)) { - throw new IllegalStateException("registered index store type [" + indexStoreType + "] conflicts with a built-in type"); - } - } - this.directoryFactories = directoryFactories; this.recoveryStateFactories = recoveryStateFactories; // doClose() is called when shutting down a node, yet there might still be ongoing requests @@ -769,6 +764,9 @@ private EngineFactory getEngineFactory(final IndexSettings idxSettings) { if (idxSettings.isSegRepEnabled()) { return new NRTReplicationEngineFactory(); } + if (IndexModule.Type.REMOTE_SNAPSHOT.match(idxSettings)) { + return config -> new ReadOnlyEngine(config, new SeqNoStats(0, 0, 0), new TranslogStats(), true, Function.identity(), false); + } return new InternalEngineFactory(); } else if (engineFactories.size() == 1) { assert engineFactories.get(0).isPresent(); diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java index 24300c884d194..504550378e14f 100644 --- a/server/src/main/java/org/opensearch/node/Node.java +++ b/server/src/main/java/org/opensearch/node/Node.java @@ -38,6 +38,7 @@ import org.apache.lucene.util.SetOnce; import org.opensearch.common.util.FeatureFlags; import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance; +import org.opensearch.index.IndexModule; import org.opensearch.index.IndexingPressureService; import org.opensearch.indices.replication.SegmentReplicationSourceFactory; import org.opensearch.indices.replication.SegmentReplicationTargetService; @@ -213,6 +214,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; @@ -604,11 +606,23 @@ protected Node( .map(plugin -> (Function>) plugin::getEngineFactory) .collect(Collectors.toList()); - final Map indexStoreFactories = pluginsService.filterPlugins(IndexStorePlugin.class) + final Map builtInDirectoryFactories = IndexModule.createBuiltInDirectoryFactories( + repositoriesServiceReference::get + ); + final Map directoryFactories = new HashMap<>(); + pluginsService.filterPlugins(IndexStorePlugin.class) .stream() .map(IndexStorePlugin::getDirectoryFactories) .flatMap(m -> m.entrySet().stream()) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)) + .forEach((k, v) -> { + // do not allow any plugin-provided index store type to conflict with a built-in type + if (builtInDirectoryFactories.containsKey(k)) { + throw new IllegalStateException("registered index store type [" + k + "] conflicts with a built-in type"); + } + directoryFactories.put(k, v); + }); + directoryFactories.putAll(builtInDirectoryFactories); final Map recoveryStateFactories = pluginsService.filterPlugins( IndexStorePlugin.class @@ -653,7 +667,7 @@ protected Node( client, metaStateService, engineFactoryProviders, - indexStoreFactories, + Map.copyOf(directoryFactories), searchModule.getValuesSourceRegistry(), recoveryStateFactories, remoteDirectoryFactory diff --git a/server/src/main/java/org/opensearch/snapshots/RestoreService.java b/server/src/main/java/org/opensearch/snapshots/RestoreService.java index ca5078c4d1c56..60c01d0b04639 100644 --- a/server/src/main/java/org/opensearch/snapshots/RestoreService.java +++ b/server/src/main/java/org/opensearch/snapshots/RestoreService.java @@ -84,7 +84,9 @@ import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; +import org.opensearch.common.util.FeatureFlags; import org.opensearch.index.Index; +import org.opensearch.index.IndexModule; import org.opensearch.index.IndexSettings; import org.opensearch.index.shard.IndexShard; import org.opensearch.index.shard.ShardId; @@ -421,20 +423,30 @@ public ClusterState execute(ClusterState currentState) { .getMaxNodeVersion() .minimumIndexCompatibilityVersion(); for (Map.Entry indexEntry : indices.entrySet()) { + String renamedIndexName = indexEntry.getKey(); String index = indexEntry.getValue(); boolean partial = checkPartial(index); - SnapshotRecoverySource recoverySource = new SnapshotRecoverySource( + + IndexMetadata snapshotIndexMetadata = updateIndexSettings( + metadata.index(index), + request.indexSettings(), + request.ignoreIndexSettings() + ); + final boolean isSearchableSnapshot = FeatureFlags.isEnabled(FeatureFlags.SEARCHABLE_SNAPSHOT) + && IndexModule.Type.REMOTE_SNAPSHOT.getSettingsKey().equals(request.storageType().toString()); + if (isSearchableSnapshot) { + snapshotIndexMetadata = addSnapshotToIndexSettings( + snapshotIndexMetadata, + snapshot, + repositoryData.resolveIndexId(index) + ); + } + final SnapshotRecoverySource recoverySource = new SnapshotRecoverySource( restoreUUID, snapshot, snapshotInfo.version(), - repositoryData.resolveIndexId(index) - ); - String renamedIndexName = indexEntry.getKey(); - IndexMetadata snapshotIndexMetadata = metadata.index(index); - snapshotIndexMetadata = updateIndexSettings( - snapshotIndexMetadata, - request.indexSettings(), - request.ignoreIndexSettings() + repositoryData.resolveIndexId(index), + isSearchableSnapshot ); try { snapshotIndexMetadata = metadataIndexUpgradeService.upgradeIndexMetadata( @@ -1207,4 +1219,16 @@ public void applyClusterState(ClusterChangedEvent event) { logger.warn("Failed to update restore state ", t); } } + + private static IndexMetadata addSnapshotToIndexSettings(IndexMetadata metadata, Snapshot snapshot, IndexId indexId) { + final Settings newSettings = Settings.builder() + .put(IndexModule.INDEX_STORE_TYPE_SETTING.getKey(), IndexModule.Type.REMOTE_SNAPSHOT.getSettingsKey()) + .put(IndexSettings.SEARCHABLE_SNAPSHOT_REPOSITORY.getKey(), snapshot.getRepository()) + .put(IndexSettings.SEARCHABLE_SNAPSHOT_ID_UUID.getKey(), snapshot.getSnapshotId().getUUID()) + .put(IndexSettings.SEARCHABLE_SNAPSHOT_ID_NAME.getKey(), snapshot.getSnapshotId().getName()) + .put(IndexSettings.SEARCHABLE_SNAPSHOT_INDEX_ID.getKey(), indexId.getId()) + .put(metadata.getSettings()) + .build(); + return IndexMetadata.builder(metadata).settings(newSettings).build(); + } } diff --git a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java index 27c0437236f63..470eeea771f2b 100644 --- a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java +++ b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java @@ -327,7 +327,8 @@ public void testShardStateMetaHashCodeEquals() { ShardStateMetadata meta = new ShardStateMetadata( randomBoolean(), randomRealisticUnicodeOfCodepointLengthBetween(1, 10), - allocationId + allocationId, + randomFrom(ShardStateMetadata.IndexDataLocation.values()) ); assertEquals(meta, new ShardStateMetadata(meta.primary, meta.indexUUID, meta.allocationId)); @@ -339,7 +340,12 @@ public void testShardStateMetaHashCodeEquals() { Set hashCodes = new HashSet<>(); for (int i = 0; i < 30; i++) { // just a sanity check that we impl hashcode allocationId = randomBoolean() ? null : randomAllocationId(); - meta = new ShardStateMetadata(randomBoolean(), randomRealisticUnicodeOfCodepointLengthBetween(1, 10), allocationId); + meta = new ShardStateMetadata( + randomBoolean(), + randomRealisticUnicodeOfCodepointLengthBetween(1, 10), + allocationId, + randomFrom(ShardStateMetadata.IndexDataLocation.values()) + ); hashCodes.add(meta.hashCode()); } assertTrue("more than one unique hashcode expected but got: " + hashCodes.size(), hashCodes.size() > 1); diff --git a/server/src/test/java/org/opensearch/index/shard/ShardPathTests.java b/server/src/test/java/org/opensearch/index/shard/ShardPathTests.java index beda468b45fb0..25ec7c7987855 100644 --- a/server/src/test/java/org/opensearch/index/shard/ShardPathTests.java +++ b/server/src/test/java/org/opensearch/index/shard/ShardPathTests.java @@ -35,6 +35,7 @@ import org.opensearch.common.settings.Settings; import org.opensearch.env.Environment; import org.opensearch.env.NodeEnvironment; +import org.opensearch.gateway.WriteStateException; import org.opensearch.index.Index; import org.opensearch.test.OpenSearchTestCase; @@ -50,7 +51,7 @@ public void testLoadShardPath() throws IOException { ShardId shardId = new ShardId("foo", "0xDEADBEEF", 0); Path[] paths = env.availableShardPaths(shardId); Path path = randomFrom(paths); - ShardStateMetadata.FORMAT.writeAndCleanup(new ShardStateMetadata(true, "0xDEADBEEF", AllocationId.newInitializing()), path); + writeShardStateMetadata("0xDEADBEEF", path); ShardPath shardPath = ShardPath.loadShardPath(logger, env, shardId, ""); assertEquals(path, shardPath.getDataPath()); assertEquals("0xDEADBEEF", shardPath.getShardId().getIndex().getUUID()); @@ -66,7 +67,7 @@ public void testFailLoadShardPathOnMultiState() throws IOException { ShardId shardId = new ShardId("foo", indexUUID, 0); Path[] paths = env.availableShardPaths(shardId); assumeTrue("This test tests multi data.path but we only got one", paths.length > 1); - ShardStateMetadata.FORMAT.writeAndCleanup(new ShardStateMetadata(true, indexUUID, AllocationId.newInitializing()), paths); + writeShardStateMetadata(indexUUID, paths); Exception e = expectThrows(IllegalStateException.class, () -> ShardPath.loadShardPath(logger, env, shardId, "")); assertThat(e.getMessage(), containsString("more than one shard state found")); } @@ -77,7 +78,7 @@ public void testFailLoadShardPathIndexUUIDMissmatch() throws IOException { ShardId shardId = new ShardId("foo", "foobar", 0); Path[] paths = env.availableShardPaths(shardId); Path path = randomFrom(paths); - ShardStateMetadata.FORMAT.writeAndCleanup(new ShardStateMetadata(true, "0xDEADBEEF", AllocationId.newInitializing()), path); + writeShardStateMetadata("0xDEADBEEF", path); Exception e = expectThrows(IllegalStateException.class, () -> ShardPath.loadShardPath(logger, env, shardId, "")); assertThat(e.getMessage(), containsString("expected: foobar on shard path")); } @@ -121,7 +122,7 @@ public void testGetRootPaths() throws IOException { ShardId shardId = new ShardId("foo", indexUUID, 0); Path[] paths = env.availableShardPaths(shardId); Path path = randomFrom(paths); - ShardStateMetadata.FORMAT.writeAndCleanup(new ShardStateMetadata(true, indexUUID, AllocationId.newInitializing()), path); + writeShardStateMetadata(indexUUID, path); ShardPath shardPath = ShardPath.loadShardPath(logger, env, shardId, customDataPath); boolean found = false; for (Path p : env.nodeDataPaths()) { @@ -148,4 +149,10 @@ public void testGetRootPaths() throws IOException { } } + private static void writeShardStateMetadata(String indexUUID, Path... paths) throws WriteStateException { + ShardStateMetadata.FORMAT.writeAndCleanup( + new ShardStateMetadata(true, indexUUID, AllocationId.newInitializing(), ShardStateMetadata.IndexDataLocation.LOCAL), + paths + ); + } } diff --git a/server/src/test/java/org/opensearch/index/store/FsDirectoryFactoryTests.java b/server/src/test/java/org/opensearch/index/store/FsDirectoryFactoryTests.java index cf8d6677b4227..ce40de0e9aa71 100644 --- a/server/src/test/java/org/opensearch/index/store/FsDirectoryFactoryTests.java +++ b/server/src/test/java/org/opensearch/index/store/FsDirectoryFactoryTests.java @@ -57,6 +57,8 @@ import java.util.Arrays; import java.util.Locale; +import static org.opensearch.test.store.MockFSDirectoryFactory.FILE_SYSTEM_BASED_STORE_TYPES; + public class FsDirectoryFactoryTests extends OpenSearchTestCase { public void testPreload() throws IOException { @@ -170,7 +172,7 @@ public void testStoreDirectory() throws IOException { // default doTestStoreDirectory(tempDir, null, IndexModule.Type.FS); // explicit directory impls - for (IndexModule.Type type : IndexModule.Type.values()) { + for (IndexModule.Type type : FILE_SYSTEM_BASED_STORE_TYPES) { doTestStoreDirectory(tempDir, type.name().toLowerCase(Locale.ROOT), type); } } diff --git a/test/framework/src/main/java/org/opensearch/test/store/MockFSDirectoryFactory.java b/test/framework/src/main/java/org/opensearch/test/store/MockFSDirectoryFactory.java index 47952af1cd06c..e38b62c419334 100644 --- a/test/framework/src/main/java/org/opensearch/test/store/MockFSDirectoryFactory.java +++ b/test/framework/src/main/java/org/opensearch/test/store/MockFSDirectoryFactory.java @@ -63,10 +63,15 @@ import java.io.PrintStream; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.List; import java.util.Random; import java.util.Set; +import java.util.stream.Collectors; public class MockFSDirectoryFactory implements IndexStorePlugin.DirectoryFactory { + public static final List FILE_SYSTEM_BASED_STORE_TYPES = Arrays.stream(IndexModule.Type.values()) + .filter(t -> (t == IndexModule.Type.REMOTE_SNAPSHOT) == false) + .collect(Collectors.toUnmodifiableList()); public static final Setting RANDOM_IO_EXCEPTION_RATE_ON_OPEN_SETTING = Setting.doubleSetting( "index.store.mock.random.io_exception_rate_on_open", @@ -168,7 +173,7 @@ private Directory randomDirectoryService(Random random, IndexSettings indexSetti .put(indexSettings.getIndexMetadata().getSettings()) .put( IndexModule.INDEX_STORE_TYPE_SETTING.getKey(), - RandomPicks.randomFrom(random, IndexModule.Type.values()).getSettingsKey() + RandomPicks.randomFrom(random, FILE_SYSTEM_BASED_STORE_TYPES).getSettingsKey() ) ) .build();