Add Clear Cache API for Searchable Snapshots (#53009)

This commit adds an API to clear the cache used for searchable snapshots:

POST /_searchable_snapshots/cache/clear
POST <index>/_searchable_snapshots/cache/clear

This API is useful in tests or to simply free disk space if needed. 
Note that this API does not clear cache stats.

Relates #50999

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

@@ -0,0 +1,145 @@
+---
+setup:
+
+  - do:
+      indices.create:
+        index: docs
+        body:
+          settings:
+            number_of_shards:   1
+            number_of_replicas: 0
+
+  - do:
+      bulk:
+        body:
+          - index:
+              _index: docs
+              _id:    1
+          - field: doc
+          - index:
+              _index: docs
+              _id:    2
+          - field: doc
+          - index:
+              _index: docs
+              _id:    3
+          - field: other
+
+  - do:
+      snapshot.create_repository:
+        repository: repository-fs
+        body:
+          type: fs
+          settings:
+            location: "repository-fs"
+
+  # Remove the snapshot if a previous test failed to delete it.
+  # Useful for third party tests that runs the test against a real external service.
+  - do:
+      snapshot.delete:
+        repository: repository-fs
+        snapshot: snapshot
+        ignore: 404
+
+  - do:
+      snapshot.create:
+        repository: repository-fs
+        snapshot: snapshot
+        wait_for_completion: true
+
+  - do:
+      indices.delete:
+        index: docs
+
+  - do:
+      snapshot.create_repository:
+        repository: repository-searchable-snapshots
+        body:
+          type: searchable
+          settings:
+            delegate_type: fs
+            location: "repository-fs"
+
+---
+teardown:
+
+  - do:
+      snapshot.delete:
+        repository: repository-fs
+        snapshot: snapshot
+        ignore: 404
+
+  - do:
+      snapshot.delete_repository:
+        repository: repository-fs
+
+  - do:
+      snapshot.delete_repository:
+        repository: repository-searchable-snapshots
+
+---
+"Clear searchable snapshots cache":
+  - skip:
+      version: " - 7.99.99"
+      reason:  searchable snapshots introduced in 8.0
+
+  - do:
+      catch: missing
+      searchable_snapshots.clear_cache: {}
+
+  - match: { error.root_cause.0.type: "resource_not_found_exception" }
+  - match: { error.root_cause.0.reason: "No searchable snapshots indices found" }
+
+  - do:
+      catch: missing
+      searchable_snapshots.clear_cache:
+        index: _all
+
+  - match: { error.root_cause.0.type: "resource_not_found_exception" }
+  - match: { error.root_cause.0.reason: "No searchable snapshots indices found" }
+
+  - do:
+      catch:   missing
+      searchable_snapshots.clear_cache:
+        index: "unknown"
+
+  - do:
+      indices.create:
+        index: non_searchable_snapshot_index
+
+  - do:
+      catch: missing
+      searchable_snapshots.clear_cache:
+        index: non_*
+
+  - match: { error.root_cause.0.type: "resource_not_found_exception" }
+  - match: { error.root_cause.0.reason: "No searchable snapshots indices found" }
+
+  - do:
+      snapshot.restore:
+        repository: repository-searchable-snapshots
+        snapshot: snapshot
+        wait_for_completion: true
+
+  - match: { snapshot.snapshot: snapshot }
+  - match: { snapshot.shards.failed: 0 }
+  - match: { snapshot.shards.successful: 1 }
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: docs
+        body:
+          query:
+            match:
+              field: "doc"
+
+  - match: { hits.total: 2 }
+
+  - do:
+      searchable_snapshots.clear_cache:
+        index: "docs"
+
+  - match:  { _shards.total: 1 }
+  - match:  { _shards.failed: 0 }
+

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

@@ -33,6 +33,14 @@ setup:
           settings:
             location: "repository-fs"
 
+  # Remove the snapshot if a previous test failed to delete it.
+  # Useful for third party tests that runs the test against a real external service.
+  - do:
+      snapshot.delete:
+        repository: repository-fs
+        snapshot: snapshot
+        ignore: 404
+
   - do:
       snapshot.create:
         repository: repository-fs
@@ -133,6 +141,9 @@ teardown:
       searchable_snapshots.stats:
         index: "d*"
 
+  - match:  { _shards.total: 1 }
+  - match:  { _shards.failed: 0 }
+
   - length:  { indices: 1 }
   - length:  { indices.docs.shards: 1 }
   - length:  { indices.docs.shards.0: 1 }

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

@@ -38,9 +38,12 @@
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.watcher.ResourceWatcherService;
+import org.elasticsearch.xpack.searchablesnapshots.action.ClearSearchableSnapshotsCacheAction;
 import org.elasticsearch.xpack.searchablesnapshots.action.SearchableSnapshotsStatsAction;
+import org.elasticsearch.xpack.searchablesnapshots.action.TransportClearSearchableSnapshotsCacheAction;
 import org.elasticsearch.xpack.searchablesnapshots.action.TransportSearchableSnapshotsStatsAction;
 import org.elasticsearch.xpack.searchablesnapshots.cache.CacheService;
+import org.elasticsearch.xpack.searchablesnapshots.rest.RestClearSearchableSnapshotsCacheAction;
 import org.elasticsearch.xpack.searchablesnapshots.rest.RestSearchableSnapshotsStatsAction;
 
 import java.util.Collection;
@@ -126,14 +129,20 @@ public Map<String, Repository.Factory> getRepositories(Environment env, NamedXCo
 
     @Override
     public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
-        return List.of(new ActionHandler<>(SearchableSnapshotsStatsAction.INSTANCE, TransportSearchableSnapshotsStatsAction.class));
+        return List.of(
+            new ActionHandler<>(SearchableSnapshotsStatsAction.INSTANCE, TransportSearchableSnapshotsStatsAction.class),
+            new ActionHandler<>(ClearSearchableSnapshotsCacheAction.INSTANCE, TransportClearSearchableSnapshotsCacheAction.class)
+        );
     }
 
     public List<RestHandler> getRestHandlers(Settings settings, RestController restController, ClusterSettings clusterSettings,
                                              IndexScopedSettings indexScopedSettings, SettingsFilter settingsFilter,
                                              IndexNameExpressionResolver indexNameExpressionResolver,
                                              Supplier<DiscoveryNodes> nodesInCluster) {
-        return List.of(new RestSearchableSnapshotsStatsAction());
+        return List.of(
+            new RestSearchableSnapshotsStatsAction(),
+            new RestClearSearchableSnapshotsCacheAction()
+        );
     }
 }
 

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

@@ -0,0 +1,118 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * 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.action;
+
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.FilterDirectory;
+import org.elasticsearch.ResourceNotFoundException;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.broadcast.BroadcastRequest;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
+import org.elasticsearch.action.support.broadcast.node.TransportBroadcastByNodeAction;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.cluster.routing.ShardsIterator;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.searchablesnapshots.InMemoryNoOpCommitDirectory;
+import org.elasticsearch.xpack.searchablesnapshots.cache.CacheDirectory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.elasticsearch.index.IndexModule.INDEX_STORE_TYPE_SETTING;
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotRepository.SNAPSHOT_CACHE_ENABLED_SETTING;
+import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotRepository.SNAPSHOT_DIRECTORY_FACTORY_KEY;
+
+public abstract class AbstractTransportSearchableSnapshotsAction
+    <Request extends BroadcastRequest<Request>, Response extends BroadcastResponse, ShardOperationResult extends Writeable>
+    extends TransportBroadcastByNodeAction<Request, Response, ShardOperationResult> {
+
+    private final IndicesService indicesService;
+
+    AbstractTransportSearchableSnapshotsAction(String actionName, ClusterService clusterService, TransportService transportService,
+                                               ActionFilters actionFilters, IndexNameExpressionResolver resolver,
+                                               Writeable.Reader<Request> request, String executor, IndicesService indicesService) {
+        super(actionName, clusterService, transportService, actionFilters, resolver, request, executor);
+        this.indicesService = indicesService;
+    }
+
+    AbstractTransportSearchableSnapshotsAction(String actionName, ClusterService clusterService, TransportService transportService,
+                                               ActionFilters actionFilters, IndexNameExpressionResolver resolver,
+                                               Writeable.Reader<Request> request, String executor, IndicesService indicesService,
+                                               boolean canTripCircuitBreaker) {
+        super(actionName, clusterService, transportService, actionFilters, resolver, request, executor, canTripCircuitBreaker);
+        this.indicesService = indicesService;
+    }
+
+    @Override
+    protected ClusterBlockException checkGlobalBlock(ClusterState state, Request request) {
+        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
+    }
+
+    @Override
+    protected ClusterBlockException checkRequestBlock(ClusterState state, Request request, String[] indices) {
+        return state.blocks().indicesBlockedException(ClusterBlockLevel.METADATA_READ, indices);
+    }
+
+    @Override
+    protected ShardsIterator shards(ClusterState state, Request request, String[] concreteIndices) {
+        final List<String> searchableSnapshotIndices = new ArrayList<>();
+        for (String concreteIndex : concreteIndices) {
+            IndexMetaData indexMetaData = state.metaData().index(concreteIndex);
+            if (indexMetaData != null) {
+                Settings indexSettings = indexMetaData.getSettings();
+                if (INDEX_STORE_TYPE_SETTING.get(indexSettings).equals(SNAPSHOT_DIRECTORY_FACTORY_KEY)) {
+                    if (SNAPSHOT_CACHE_ENABLED_SETTING.get(indexSettings)) {
+                        searchableSnapshotIndices.add(concreteIndex);
+                    }
+                }
+            }
+        }
+        if (searchableSnapshotIndices.isEmpty()) {
+            throw new ResourceNotFoundException("No searchable snapshots indices found");
+        }
+        return state.routingTable().allShards(searchableSnapshotIndices.toArray(new String[0]));
+    }
+
+    @Override
+    protected ShardOperationResult shardOperation(Request request, ShardRouting shardRouting) throws IOException {
+        final IndexShard indexShard = indicesService.indexServiceSafe(shardRouting.index()).getShard(shardRouting.id());
+        final CacheDirectory cacheDirectory = unwrapCacheDirectory(indexShard.store().directory());
+        assert cacheDirectory != null;
+        assert cacheDirectory.getShardId().equals(shardRouting.shardId());
+        return executeShardOperation(request, shardRouting, cacheDirectory);
+    }
+
+    protected abstract ShardOperationResult executeShardOperation(Request request, ShardRouting shardRouting,
+                                                                  CacheDirectory cacheDirectory) throws IOException;
+
+    @Nullable
+    private static CacheDirectory unwrapCacheDirectory(Directory dir) {
+        while (dir != null) {
+            if (dir instanceof CacheDirectory) {
+                return (CacheDirectory) dir;
+            } else if (dir instanceof InMemoryNoOpCommitDirectory) {
+                dir = ((InMemoryNoOpCommitDirectory) dir).getRealDirectory();
+            } else if (dir instanceof FilterDirectory) {
+                dir = ((FilterDirectory) dir).getDelegate();
+            } else {
+                dir = null;
+            }
+        }
+        return null;
+    }
+}

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

@@ -0,0 +1,18 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * 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.action;
+
+import org.elasticsearch.action.ActionType;
+
+public class ClearSearchableSnapshotsCacheAction extends ActionType<ClearSearchableSnapshotsCacheResponse> {
+
+    public static final ClearSearchableSnapshotsCacheAction INSTANCE = new ClearSearchableSnapshotsCacheAction();
+    static final String NAME = "cluster:admin/xpack/searchable_snapshots/cache/clear";
+
+    private ClearSearchableSnapshotsCacheAction() {
+        super(NAME, ClearSearchableSnapshotsCacheResponse::new);
+    }
+}

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

@@ -0,0 +1,27 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * 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.action;
+
+import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.action.support.broadcast.BroadcastRequest;
+import org.elasticsearch.common.io.stream.StreamInput;
+
+import java.io.IOException;
+
+public class ClearSearchableSnapshotsCacheRequest extends BroadcastRequest<ClearSearchableSnapshotsCacheRequest> {
+
+    public ClearSearchableSnapshotsCacheRequest(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    public ClearSearchableSnapshotsCacheRequest(String... indices) {
+        super(indices);
+    }
+
+    protected ClearSearchableSnapshotsCacheRequest(String[] indices, IndicesOptions indicesOptions) {
+        super(indices, indicesOptions);
+    }
+}

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

@@ -0,0 +1,25 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * 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.action;
+
+import org.elasticsearch.action.support.DefaultShardOperationFailedException;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
+import org.elasticsearch.common.io.stream.StreamInput;
+
+import java.io.IOException;
+import java.util.List;
+
+public class ClearSearchableSnapshotsCacheResponse extends BroadcastResponse {
+
+    ClearSearchableSnapshotsCacheResponse(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    ClearSearchableSnapshotsCacheResponse(int totalShards, int successfulShards, int failedShards,
+                                          List<DefaultShardOperationFailedException> shardFailures) {
+        super(totalShards, successfulShards, failedShards, shardFailures);
+    }
+}