Add a java level freeze/unfreeze API

This change adds a high level freeze API that allows to open an
index frozen and vice versa. Indices must be closed in order to
become frozen and an open but frozen index must be closed to be
defrosted. This change also adds a `index.frozen` setting to
mark frozen indices and integrates the frozen engine with the
`SearchOperationListener` that resets and releases the directory
reader after and before search phases.

Relates to elastic#34352
Depends on elastic#34357

server/src/main/java/org/elasticsearch/action/admin/indices/open/OpenIndexClusterStateUpdateRequest.java

@@ -28,7 +28,7 @@ public class OpenIndexClusterStateUpdateRequest extends IndicesClusterStateUpdat
 
     private ActiveShardCount waitForActiveShards = ActiveShardCount.DEFAULT;
 
-    OpenIndexClusterStateUpdateRequest() {
+    public OpenIndexClusterStateUpdateRequest() {
 
     }
 

server/src/main/java/org/elasticsearch/action/admin/indices/open/OpenIndexResponse.java

@@ -40,10 +40,10 @@ public class OpenIndexResponse extends ShardsAcknowledgedResponse {
         declareAcknowledgedAndShardsAcknowledgedFields(PARSER);
     }
 
-    OpenIndexResponse() {
+    public OpenIndexResponse() {
     }
 
-    OpenIndexResponse(boolean acknowledged, boolean shardsAcknowledged) {
+    public OpenIndexResponse(boolean acknowledged, boolean shardsAcknowledged) {
         super(acknowledged, shardsAcknowledged);
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/index/engine/FrozenEngine.java

@@ -33,7 +33,11 @@
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.lucene.Lucene;
 import org.elasticsearch.common.metrics.CounterMetric;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.core.internal.io.IOUtils;
+import org.elasticsearch.index.shard.SearchOperationListener;
+import org.elasticsearch.search.internal.SearchContext;
+import org.elasticsearch.transport.TransportRequest;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -47,6 +51,8 @@
  * phase starts. This allows releasing references as soon as possible on the search layer.
  */
 public final class FrozenEngine extends ReadOnlyEngine {
+    public static final Setting<Boolean> INDEX_FROZEN = Setting.boolSetting("index.frozen", false, Setting.Property.IndexScope,
+        Setting.Property.PrivateIndex);
     private final CounterMetric openedReaders = new CounterMetric();
     private volatile DirectoryReader lastOpenedReader;
 
@@ -218,6 +224,43 @@ void resetReader() throws IOException {
         }
     }
 
+    /*
+     * We register this listener for a frozen index that will
+     *  1. reset the reader every time the search context is validated which happens when the context is looked up ie. on a fetch phase
+     *  etc.
+     *  2. register a releasable resource that is cleaned after each phase that releases the reader for this searcher
+     */
+    public static class ReacquireEngineSearcherListener implements SearchOperationListener {
+
+        @Override
+        public void validateSearchContext(SearchContext context, TransportRequest transportRequest) {
+            Searcher engineSearcher = context.searcher().getEngineSearcher();
+            if (engineSearcher instanceof FrozenEngineSearcher) {
+                try {
+                    ((FrozenEngineSearcher) engineSearcher).resetReader();
+                } catch (IOException e) {
+                    throw new UncheckedIOException(e);
+                }
+            }
+            // also register a release resource in this case if we have multiple roundtrips like in DFS
+            onNewContext(context);
+        }
+
+        @Override
+        public void onNewContext(SearchContext context) {
+            Searcher engineSearcher = context.searcher().getEngineSearcher();
+            context.addReleasable(() -> {
+                if (engineSearcher instanceof FrozenEngineSearcher) {
+                    try {
+                        ((FrozenEngineSearcher) engineSearcher).releaseReader();
+                    } catch (IOException e) {
+                        throw new UncheckedIOException(e);
+                    }
+                }
+            }, SearchContext.Lifetime.PHASE);
+        }
+    }
+
     /**
      * This class allows us to use the same high level reader across multiple search phases but replace the underpinnings
      * on/after each search phase. This is really important otherwise we would hold on to multiple readers across phases.

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClient.java

@@ -6,11 +6,13 @@
 package org.elasticsearch.xpack.core;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.open.OpenIndexResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.license.LicensingClient;
 import org.elasticsearch.protocol.xpack.XPackInfoRequest;
 import org.elasticsearch.protocol.xpack.XPackInfoResponse;
+import org.elasticsearch.xpack.core.action.TransportOpenIndexAndFreezeAction;
 import org.elasticsearch.xpack.core.action.XPackInfoAction;
 import org.elasticsearch.xpack.core.action.XPackInfoRequestBuilder;
 import org.elasticsearch.xpack.core.ccr.client.CcrClient;
@@ -96,4 +98,9 @@ public XPackInfoRequestBuilder prepareInfo() {
     public void info(XPackInfoRequest request, ActionListener<XPackInfoResponse> listener) {
         client.execute(XPackInfoAction.INSTANCE, request, listener);
     }
+
+    public void openAndFreeze(TransportOpenIndexAndFreezeAction.OpenIndexAndFreezeRequest request, ActionListener<OpenIndexResponse>
+        listener) {
+        client.execute(TransportOpenIndexAndFreezeAction.OpenIndexAndFreezeAction.INSTANCE, request, listener);
+    }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackPlugin.java

@@ -37,8 +37,10 @@
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.index.IndexModule;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.engine.EngineFactory;
+import org.elasticsearch.index.engine.FrozenEngine;
 import org.elasticsearch.license.LicenseService;
 import org.elasticsearch.license.LicensesMetaData;
 import org.elasticsearch.license.Licensing;
@@ -55,6 +57,7 @@
 import org.elasticsearch.snapshots.SourceOnlySnapshotRepository;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.watcher.ResourceWatcherService;
+import org.elasticsearch.xpack.core.action.TransportOpenIndexAndFreezeAction;
 import org.elasticsearch.xpack.core.action.TransportXPackInfoAction;
 import org.elasticsearch.xpack.core.action.TransportXPackUsageAction;
 import org.elasticsearch.xpack.core.action.XPackInfoAction;
@@ -266,6 +269,8 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
         List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> actions = new ArrayList<>();
         actions.add(new ActionHandler<>(XPackInfoAction.INSTANCE, TransportXPackInfoAction.class));
         actions.add(new ActionHandler<>(XPackUsageAction.INSTANCE, TransportXPackUsageAction.class));
+        actions.add(new ActionHandler<>(TransportOpenIndexAndFreezeAction.OpenIndexAndFreezeAction.INSTANCE,
+            TransportOpenIndexAndFreezeAction.class));
         actions.addAll(licensing.getActions());
         return actions;
     }
@@ -359,14 +364,26 @@ public Map<String, Repository.Factory> getRepositories(Environment env, NamedXCo
     public Optional<EngineFactory> getEngineFactory(IndexSettings indexSettings) {
         if (indexSettings.getValue(SourceOnlySnapshotRepository.SOURCE_ONLY)) {
             return Optional.of(SourceOnlySnapshotRepository.getEngineFactory());
+        } else if (indexSettings.getValue(FrozenEngine.INDEX_FROZEN)) {
+            return Optional.of(FrozenEngine::new);
         }
+
         return Optional.empty();
     }
 
     @Override
     public List<Setting<?>> getSettings() {
         List<Setting<?>> settings = super.getSettings();
         settings.add(SourceOnlySnapshotRepository.SOURCE_ONLY);
+        settings.add(FrozenEngine.INDEX_FROZEN);
         return settings;
     }
+
+    @Override
+    public void onIndexModule(IndexModule indexModule) {
+        if (FrozenEngine.INDEX_FROZEN.get(indexModule.getSettings())) {
+            indexModule.addSearchOperationListener(new FrozenEngine.ReacquireEngineSearcherListener());
+        }
+        super.onIndexModule(indexModule);
+    }
 }