Allow searchable snapshot cache service to periodically fsync cache files

server/src/main/java/org/elasticsearch/common/cache/Cache.java

@@ -650,6 +650,36 @@ public void remove() {
         };
     }
 
+    /**
+     * An LRU sequencing of the entries in the cache. This sequence is not protected from mutations
+     * to the cache (except for {@link Iterator#remove()}. The result of iteration under any other mutation is
+     * undefined.
+     *
+     * @return an LRU-ordered {@link Iterable} over the entries in the cache
+     */
+    public Iterable<Tuple<K,V>> entries() {
+        return () -> new Iterator<>() {
+
+            private final CacheIterator iterator = new CacheIterator(head);
+
+            @Override
+            public boolean hasNext() {
+                return iterator.hasNext();
+            }
+
+            @Override
+            public Tuple<K,V> next() {
+                final Cache.Entry<K, V> next = iterator.next();
+                return Tuple.tuple(next.key, next.value);
+            }
+
+            @Override
+            public void remove() {
+                iterator.remove();
+            }
+        };
+    }
+
     private class CacheIterator implements Iterator<Entry<K, V>> {
         private Entry<K, V> current;
         private Entry<K, V> next;

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

@@ -77,6 +77,13 @@ protected void closeInternal() {
      **/
     private final AtomicBoolean needsFsync = new AtomicBoolean();
 
+    /**
+     * A consumer that accepts the current {@link CacheFile} instance every time the {@link #needsFsync} flag is toggled to {@code true},
+     * which indicates that the cache file has been updated.
+     * See {@link #markAsNeedsFSync()} method.
+     */
+    private final Consumer<CacheFile> needsFsyncListener;
+
     /**
      * A reference counted holder for the current channel to the physical file backing this cache file instance.
      * By guarding access to the file channel by ref-counting and giving the channel its own life-cycle we remove all need for
@@ -117,10 +124,11 @@ protected void closeInternal() {
     @Nullable
     private volatile FileChannelReference channelRef;
 
-    public CacheFile(String description, long length, Path file) {
+    public CacheFile(String description, long length, Path file, @Nullable Consumer<CacheFile> fsyncListener) {
         this.tracker = new SparseFileTracker(file.toString(), length);
         this.description = Objects.requireNonNull(description);
         this.file = Objects.requireNonNull(file);
+        this.needsFsyncListener = fsyncListener != null ? fsyncListener : cacheFile -> {};
         assert invariant();
     }
 
@@ -320,7 +328,7 @@ protected void doRun() throws Exception {
                             reference.decRef();
                         }
                         gap.onCompletion();
-                        needsFsync.set(true);
+                        markAsNeedsFSync();
                     }
 
                     @Override
@@ -420,6 +428,15 @@ boolean needsFsync() {
         return needsFsync.get();
     }
 
+    /**
+     * Marks the current cache file as "fsync needed" and notifies the corresponding listener.
+     */
+    private void markAsNeedsFSync() {
+        if (needsFsync.getAndSet(true) == false) {
+            needsFsyncListener.accept(this);
+        }
+    }
+
     /**
      * Ensure that all ranges of data written to the cache file are written to the storage device that contains it. This method performs
      * synchronization only if data has been written to the cache since the last time the method was called. If calling this method
@@ -444,12 +461,12 @@ public SortedSet<Tuple<Long, Long>> fsync() throws IOException {
                         assert completedRanges != null;
                         assert completedRanges.isEmpty() == false;
 
-                        IOUtils.fsync(file, false, false); // TODO don't forget to fsync parent directory
+                        IOUtils.fsync(file, false, false);
                         success = true;
                         return completedRanges;
                     } finally {
                         if (success == false) {
-                            needsFsync.set(true);
+                            markAsNeedsFSync();
                         }
                     }
                 }

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

@@ -25,19 +25,19 @@ public CacheKey(SnapshotId snapshotId, IndexId indexId, ShardId shardId, String
         this.fileName = Objects.requireNonNull(fileName);
     }
 
-    SnapshotId getSnapshotId() {
+    public SnapshotId getSnapshotId() {
         return snapshotId;
     }
 
-    IndexId getIndexId() {
+    public IndexId getIndexId() {
         return indexId;
     }
 
-    ShardId getShardId() {
+    public ShardId getShardId() {
         return shardId;
     }
 
-    String getFileName() {
+    public String getFileName() {
         return fileName;
     }
 

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

@@ -11,6 +11,7 @@
 import org.elasticsearch.blobstore.cache.BlobStoreCacheService;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.decider.AllocationDecider;
@@ -67,6 +68,7 @@
 import org.elasticsearch.xpack.searchablesnapshots.rest.RestMountSearchableSnapshotAction;
 import org.elasticsearch.xpack.searchablesnapshots.rest.RestSearchableSnapshotsStatsAction;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -182,7 +184,9 @@ public List<Setting<?>> getSettings() {
             SNAPSHOT_CACHE_EXCLUDED_FILE_TYPES_SETTING,
             SNAPSHOT_UNCACHED_CHUNK_SIZE_SETTING,
             CacheService.SNAPSHOT_CACHE_SIZE_SETTING,
-            CacheService.SNAPSHOT_CACHE_RANGE_SIZE_SETTING
+            CacheService.SNAPSHOT_CACHE_RANGE_SIZE_SETTING,
+            CacheService.SNAPSHOT_CACHE_SYNC_INTERVAL_SETTING,
+            CacheService.SNAPSHOT_CACHE_MAX_FILES_TO_SYNC_AT_ONCE_SETTING
         );
     }
 
@@ -200,19 +204,29 @@ public Collection<Object> createComponents(
         final IndexNameExpressionResolver resolver,
         final Supplier<RepositoriesService> repositoriesServiceSupplier
     ) {
-        final CacheService cacheService = new CacheService(new NodeEnvironmentCacheCleaner(nodeEnvironment), settings);
-        this.cacheService.set(cacheService);
+        final List<Object> components = new ArrayList<>();
         this.repositoriesServiceSupplier = repositoriesServiceSupplier;
         this.threadPool.set(threadPool);
-        final BlobStoreCacheService blobStoreCacheService = new BlobStoreCacheService(
-            clusterService,
-            threadPool,
-            client,
-            SNAPSHOT_BLOB_CACHE_INDEX
-        );
-        this.blobStoreCacheService.set(blobStoreCacheService);
         this.failShardsListener.set(new FailShardsOnInvalidLicenseClusterListener(getLicenseState(), clusterService.getRerouteService()));
-        return List.of(cacheService, blobStoreCacheService);
+        if (DiscoveryNode.isDataNode(settings)) {
+            final CacheService cacheService = new CacheService(
+                settings,
+                clusterService,
+                threadPool,
+                new NodeEnvironmentCacheCleaner(nodeEnvironment)
+            );
+            this.cacheService.set(cacheService);
+            components.add(cacheService);
+            final BlobStoreCacheService blobStoreCacheService = new BlobStoreCacheService(
+                clusterService,
+                threadPool,
+                client,
+                SNAPSHOT_BLOB_CACHE_INDEX
+            );
+            this.blobStoreCacheService.set(blobStoreCacheService);
+            components.add(blobStoreCacheService);
+        }
+        return Collections.unmodifiableList(components);
     }
 
     @Override