Add API to mount a snapshot as a searchable index

x-pack/plugin/searchable-snapshots/qa/rest/src/test/resources/rest-api-spec/test/clear_cache.yml

@@ -51,15 +51,6 @@ setup:
       indices.delete:
         index: docs
 
-  - do:
-      snapshot.create_repository:
-        repository: repository-searchable-snapshots
-        body:
-          type: searchable
-          settings:
-            delegate_type: fs
-            location: "repository-fs"
-
 ---
 teardown:
 
@@ -73,10 +64,6 @@ teardown:
       snapshot.delete_repository:
         repository: repository-fs
 
-  - do:
-      snapshot.delete_repository:
-        repository: repository-searchable-snapshots
-
 ---
 "Clear searchable snapshots cache":
   - skip:
@@ -116,10 +103,12 @@ teardown:
   - match: { error.root_cause.0.reason: "No searchable snapshots indices found" }
 
   - do:
-      snapshot.restore:
-        repository: repository-searchable-snapshots
+      searchable_snapshots.mount:
+        repository: repository-fs
         snapshot: snapshot
         wait_for_completion: true
+        body:
+          index: docs
 
   - match: { snapshot.snapshot: snapshot }
   - match: { snapshot.shards.failed: 0 }

x-pack/plugin/searchable-snapshots/qa/rest/src/test/resources/rest-api-spec/test/stats.yml

@@ -102,21 +102,12 @@ teardown:
   - match: { error.root_cause.0.reason: "No searchable snapshots indices found" }
 
   - do:
-      snapshot.create_repository:
-        repository: repository-searchable-snapshots
-        body:
-          type: searchable
-          settings:
-            delegate_type: fs
-            location: "repository-fs"
-
-  - match: { acknowledged: true }
-
-  - do:
-      snapshot.restore:
-        repository: repository-searchable-snapshots
+      searchable_snapshots.mount:
+        repository: repository-fs
         snapshot: snapshot
         wait_for_completion: true
+        body:
+          index: docs
 
   - match: { snapshot.snapshot: snapshot }
   - match: { snapshot.shards.failed: 0 }

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/InMemoryNoOpCommitDirectory.java → x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/InMemoryNoOpCommitDirectory.java

@@ -3,7 +3,7 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-package org.elasticsearch.xpack.searchablesnapshots;
+package org.elasticsearch.index.store;
 
 import org.apache.lucene.store.ByteBuffersDirectory;
 import org.apache.lucene.store.Directory;

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/SearchableSnapshotDirectory.java

@@ -5,6 +5,9 @@
  */
 package org.elasticsearch.index.store;
 
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.index.NoMergePolicy;
 import org.apache.lucene.store.BaseDirectory;
 import org.apache.lucene.store.BufferedIndexInput;
 import org.apache.lucene.store.Directory;
@@ -13,14 +16,36 @@
 import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.store.SingleInstanceLockFactory;
 import org.elasticsearch.common.blobstore.BlobContainer;
+import org.elasticsearch.common.lucene.Lucene;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.seqno.SequenceNumbers;
+import org.elasticsearch.index.shard.ShardPath;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
+import org.elasticsearch.index.translog.Translog;
+import org.elasticsearch.repositories.IndexId;
+import org.elasticsearch.repositories.RepositoriesService;
+import org.elasticsearch.repositories.Repository;
+import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
+import org.elasticsearch.snapshots.SnapshotId;
+import org.elasticsearch.xpack.searchablesnapshots.cache.CacheDirectory;
+import org.elasticsearch.xpack.searchablesnapshots.cache.CacheService;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.nio.file.Path;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.LongSupplier;
+
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots.SNAPSHOT_CACHE_ENABLED_SETTING;
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots.SNAPSHOT_INDEX_ID_SETTING;
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots.SNAPSHOT_REPOSITORY_SETTING;
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots.SNAPSHOT_SNAPSHOT_ID_SETTING;
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots.SNAPSHOT_SNAPSHOT_NAME_SETTING;
 
 /**
  * Implementation of {@link Directory} that exposes files from a snapshot as a Lucene directory. Because snapshot are immutable this
@@ -38,7 +63,7 @@ public class SearchableSnapshotDirectory extends BaseDirectory {
     private final BlobStoreIndexShardSnapshot snapshot;
     private final BlobContainer blobContainer;
 
-    public SearchableSnapshotDirectory(final BlobStoreIndexShardSnapshot snapshot, final BlobContainer blobContainer) {
+    SearchableSnapshotDirectory(final BlobStoreIndexShardSnapshot snapshot, final BlobContainer blobContainer) {
         super(new SingleInstanceLockFactory());
         this.snapshot = Objects.requireNonNull(snapshot);
         this.blobContainer = Objects.requireNonNull(blobContainer);
@@ -121,4 +146,52 @@ public void rename(String source, String dest) {
     private static UnsupportedOperationException unsupportedException() {
         return new UnsupportedOperationException("Searchable snapshot directory does not support this operation");
     }
+
+    public static Directory create(RepositoriesService repositories,
+                                   CacheService cache,
+                                   IndexSettings indexSettings,
+                                   ShardPath shardPath,
+                                   LongSupplier currentTimeNanosSupplier) throws IOException {
+
+        final Repository repository = repositories.repository(SNAPSHOT_REPOSITORY_SETTING.get(indexSettings.getSettings()));
+        if (repository instanceof BlobStoreRepository == false) {
+            throw new IllegalArgumentException("Repository [" + repository + "] is not searchable");
+        }
+        final BlobStoreRepository blobStoreRepository = (BlobStoreRepository) repository;
+
+        final IndexId indexId = new IndexId(indexSettings.getIndex().getName(), SNAPSHOT_INDEX_ID_SETTING.get(indexSettings.getSettings()));
+        final BlobContainer blobContainer = blobStoreRepository.shardContainer(indexId, shardPath.getShardId().id());
+
+        final SnapshotId snapshotId = new SnapshotId(SNAPSHOT_SNAPSHOT_NAME_SETTING.get(indexSettings.getSettings()),
+            SNAPSHOT_SNAPSHOT_ID_SETTING.get(indexSettings.getSettings()));
+        final BlobStoreIndexShardSnapshot snapshot = blobStoreRepository.loadShardSnapshot(blobContainer, snapshotId);
+
+        Directory directory = new SearchableSnapshotDirectory(snapshot, blobContainer);
+        if (SNAPSHOT_CACHE_ENABLED_SETTING.get(indexSettings.getSettings())) {
+            final Path cacheDir = shardPath.getDataPath().resolve("snapshots").resolve(snapshotId.getUUID());
+            directory = new CacheDirectory(directory, cache, cacheDir, snapshotId, indexId, shardPath.getShardId(),
+                currentTimeNanosSupplier);
+        }
+        directory = new InMemoryNoOpCommitDirectory(directory);
+
+        final IndexWriterConfig indexWriterConfig = new IndexWriterConfig(null)
+            .setSoftDeletesField(Lucene.SOFT_DELETES_FIELD)
+            .setMergePolicy(NoMergePolicy.INSTANCE);
+
+        try (IndexWriter indexWriter = new IndexWriter(directory, indexWriterConfig)) {
+            final Map<String, String> userData = new HashMap<>();
+            indexWriter.getLiveCommitData().forEach(e -> userData.put(e.getKey(), e.getValue()));
+
+            final String translogUUID = Translog.createEmptyTranslog(shardPath.resolveTranslog(),
+                Long.parseLong(userData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)),
+                shardPath.getShardId(), 0L);
+
+            userData.put(Translog.TRANSLOG_UUID_KEY, translogUUID);
+            indexWriter.setLiveCommitData(userData.entrySet());
+            indexWriter.commit();
+        }
+
+        return directory;
+    }
+
 }