Add searchable snapshots cache directory (#50693)

This pull request adds a new CacheDirectory implementation which allows 
to cache ranges of bytes of Lucene files. This directory implementation 
uses a CacheService to retrieve and to read ranges of bytes from cached 
files stored on disk. The cached files are created as sparse files on disk 
and the availability of bytes ranges are tracked through a SparseFileTracker 
instance.

Co-authored-by: David Turner <[email protected]>

Author: tlrx

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

@@ -27,8 +27,11 @@
 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.IOException;
+import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.function.Function;
@@ -53,6 +56,8 @@ public class SearchableSnapshotRepository extends FilterRepository {
         Setting.simpleString("index.store.snapshot.snapshot_uuid", Setting.Property.IndexScope, Setting.Property.PrivateIndex);
     public static final Setting<String> SNAPSHOT_INDEX_ID_SETTING =
         Setting.simpleString("index.store.snapshot.index_uuid", Setting.Property.IndexScope, Setting.Property.PrivateIndex);
+    public static final Setting<Boolean> SNAPSHOT_CACHE_ENABLED_SETTING =
+        Setting.boolSetting("index.store.snapshot.cache.enabled", true, Setting.Property.IndexScope);
 
     public static final String SNAPSHOT_DIRECTORY_FACTORY_KEY = "snapshot";
 
@@ -64,12 +69,12 @@ public class SearchableSnapshotRepository extends FilterRepository {
     public SearchableSnapshotRepository(Repository in) {
         super(in);
         if (in instanceof BlobStoreRepository == false) {
-            throw new IllegalArgumentException("Repository [" + in + "] does not support searchable snapshots" );
+            throw new IllegalArgumentException("Repository [" + in + "] does not support searchable snapshots");
         }
         blobStoreRepository = (BlobStoreRepository) in;
     }
 
-    private Directory makeDirectory(IndexSettings indexSettings, ShardPath shardPath) throws IOException {
+    private Directory makeDirectory(IndexSettings indexSettings, ShardPath shardPath, CacheService cacheService) throws IOException {
 
         IndexId indexId = new IndexId(indexSettings.getIndex().getName(), SNAPSHOT_INDEX_ID_SETTING.get(indexSettings.getSettings()));
         BlobContainer blobContainer = blobStoreRepository.shardContainer(indexId, shardPath.getShardId().id());
@@ -78,10 +83,14 @@ private Directory makeDirectory(IndexSettings indexSettings, ShardPath shardPath
             SNAPSHOT_SNAPSHOT_ID_SETTING.get(indexSettings.getSettings()));
         BlobStoreIndexShardSnapshot snapshot = blobStoreRepository.loadShardSnapshot(blobContainer, snapshotId);
 
-        final SearchableSnapshotDirectory searchableSnapshotDirectory = new SearchableSnapshotDirectory(snapshot, blobContainer);
-        final InMemoryNoOpCommitDirectory inMemoryNoOpCommitDirectory = new InMemoryNoOpCommitDirectory(searchableSnapshotDirectory);
+        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, cacheService, cacheDir);
+        }
+        directory = new InMemoryNoOpCommitDirectory(directory);
 
-        try (IndexWriter indexWriter = new IndexWriter(inMemoryNoOpCommitDirectory, new IndexWriterConfig())) {
+        try (IndexWriter indexWriter = new IndexWriter(directory, new IndexWriterConfig())) {
             final Map<String, String> userData = new HashMap<>();
             indexWriter.getLiveCommitData().forEach(e -> userData.put(e.getKey(), e.getValue()));
 
@@ -94,7 +103,7 @@ private Directory makeDirectory(IndexSettings indexSettings, ShardPath shardPath
             indexWriter.commit();
         }
 
-        return inMemoryNoOpCommitDirectory;
+        return directory;
     }
 
     @Override
@@ -136,17 +145,21 @@ public Repository create(RepositoryMetaData metaData, Function<String, Factory>
         };
     }
 
-    public static IndexStorePlugin.DirectoryFactory newDirectoryFactory(final Supplier<RepositoriesService> repositoriesService) {
+    public static IndexStorePlugin.DirectoryFactory newDirectoryFactory(final Supplier<RepositoriesService> repositoriesService,
+                                                                        final Supplier<CacheService> cacheService) {
         return (indexSettings, shardPath) -> {
             final RepositoriesService repositories = repositoriesService.get();
             assert repositories != null;
 
             final Repository repository = repositories.repository(SNAPSHOT_REPOSITORY_SETTING.get(indexSettings.getSettings()));
             if (repository instanceof SearchableSnapshotRepository == false) {
-                throw new IllegalArgumentException("Repository [" + repository + "] is not searchable" );
+                throw new IllegalArgumentException("Repository [" + repository + "] is not searchable");
             }
 
-            return ((SearchableSnapshotRepository)repository).makeDirectory(indexSettings, shardPath);
+            final CacheService cache = cacheService.get();
+            assert cache != null;
+
+            return ((SearchableSnapshotRepository) repository).makeDirectory(indexSettings, shardPath, cache);
         };
     }
 }

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

@@ -6,10 +6,14 @@
 package org.elasticsearch.xpack.searchablesnapshots;
 
 import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.engine.EngineFactory;
 import org.elasticsearch.index.engine.ReadOnlyEngine;
@@ -21,7 +25,12 @@
 import org.elasticsearch.repositories.RepositoriesModule;
 import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.repositories.Repository;
+import org.elasticsearch.script.ScriptService;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.watcher.ResourceWatcherService;
+import org.elasticsearch.xpack.searchablesnapshots.cache.CacheService;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -36,17 +45,41 @@
 public class SearchableSnapshots extends Plugin implements IndexStorePlugin, RepositoryPlugin, EnginePlugin {
 
     private final SetOnce<RepositoriesService> repositoriesService;
+    private final SetOnce<CacheService> cacheService;
+    private final Settings settings;
 
-    public SearchableSnapshots() {
+    public SearchableSnapshots(final Settings settings) {
         this.repositoriesService = new SetOnce<>();
+        this.cacheService = new SetOnce<>();
+        this.settings = settings;
     }
 
     @Override
     public List<Setting<?>> getSettings() {
         return List.of(SearchableSnapshotRepository.SNAPSHOT_REPOSITORY_SETTING,
             SearchableSnapshotRepository.SNAPSHOT_SNAPSHOT_NAME_SETTING,
             SearchableSnapshotRepository.SNAPSHOT_SNAPSHOT_ID_SETTING,
-            SearchableSnapshotRepository.SNAPSHOT_INDEX_ID_SETTING);
+            SearchableSnapshotRepository.SNAPSHOT_INDEX_ID_SETTING,
+            SearchableSnapshotRepository.SNAPSHOT_CACHE_ENABLED_SETTING,
+            CacheService.SNAPSHOT_CACHE_SIZE_SETTING
+        );
+    }
+
+    @Override
+    public Collection<Object> createComponents(
+            final Client client,
+            final ClusterService clusterService,
+            final ThreadPool threadPool,
+            final ResourceWatcherService resourceWatcherService,
+            final ScriptService scriptService,
+            final NamedXContentRegistry xContentRegistry,
+            final Environment environment,
+            final NodeEnvironment nodeEnvironment,
+            final NamedWriteableRegistry namedWriteableRegistry) {
+
+        final CacheService cacheService = new CacheService(settings);
+        this.cacheService.set(cacheService);
+        return List.of(cacheService);
     }
 
     @Override
@@ -57,7 +90,7 @@ public void onRepositoriesModule(RepositoriesModule repositoriesModule) {
     @Override
     public Map<String, DirectoryFactory> getDirectoryFactories() {
         return Map.of(SearchableSnapshotRepository.SNAPSHOT_DIRECTORY_FACTORY_KEY,
-            SearchableSnapshotRepository.newDirectoryFactory(repositoriesService::get));
+            SearchableSnapshotRepository.newDirectoryFactory(repositoriesService::get, cacheService::get));
     }
 
     @Override