Allow searchable snapshot cache service to periodically fsync cache files

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

@@ -77,6 +77,12 @@ protected void closeInternal() {
  **/
  private final AtomicBoolean needsFsync = new AtomicBoolean();
 
+ /**
+ * A runnable that is executed 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 Runnable needsFsyncRunnable;
+
  /**
  * 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 +123,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, Runnable onNeedFSync) {
  this.tracker = new SparseFileTracker(file.toString(), length);
  this.description = Objects.requireNonNull(description);
  this.file = Objects.requireNonNull(file);
+ this.needsFsyncRunnable = Objects.requireNonNull(onNeedFSync);
  assert invariant();
  }
 
@@ -320,7 +327,7 @@ protected void doRun() throws Exception {
  reference.decRef();
  }
  gap.onCompletion();
- needsFsync.set(true);
+ markAsNeedsFSync();
  }
 
  @Override
@@ -420,6 +427,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) {
+ needsFsyncRunnable.run();
+ }
+ }
+
  /**
  * 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 +460,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,10 @@ 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,
+ CacheService.SNAPSHOT_CACHE_SYNC_SHUTDOWN_TIMEOUT
  );
  }
 
@@ -200,19 +205,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