Add a _freeze / _unfreeze API

docs/reference/frozen-indices.asciidoc

@@ -0,0 +1,79 @@
+[role="xpack"]
+[testenv="basic"]
+[[frozen-indices]]
+= Frozen Indices
+
+Elasticsearch indices can require a significant amount of memory available in order to be open and searchable. Yet, not all indices need
+to be writable at the same time and have different access patterns over time. For example indices in the time-series or logging use-case
+are unlikely to be queried once they age out but still need to be kept around for retention policy purposes.
+
+In order to keep indices available and queryable for a longer period but at the same time reduce their hardware requirements they can be transitioned
+into a frozen state. Once an index is frozen, all of it's transient shard memory (aside from mappings and analyzers)
+is moved to persistent storage. This allows for a much higher disk to heap storage ratio on individual nodes. Once an index is
+frozen, it is made read-only and drops its transient data structures from memory.  These will need to be reloaded on demand (and subsequently dropped) for each search request that targets the frozen index.  A search request that hits
+one or more frozen shards will be executed on a throttled threadpool that ensures that we never search more than
+`N` (`1` by default) searches concurrently (see <<search-throttled>>). This protects nodes from exceeding the available memory due to incoming search requests.
+
+In contrast to ordinary open indices, frozen indices are expected to execute slowly and are not designed for high query load. Parallelism is
+gained only on a per-node level and loading data-structures on demand is expected to be one or more orders of a magnitude slower than query
+execution on a per shard level. Depending on the data in an index, a frozen index may execute searches in the seconds to minutes range, when the same index in an unfrozen state may execute the same search request in milliseconds.
+
+== Best Practices
+
+Since frozen indices provide a much higher disk to heap ratio at the expense of search latency,  it is advisable to allocate frozen indices to
+dedicated nodes to prevent searches on frozen indices influencing traffic on low latency nodes. There is significant overhead in loading
+data-structures on demand which can cause page-faults and garbage collections, which further slow down query execution.
+
+Since indices that are eligible for freezing are unlikely to change in the future, disk space can be optimized as described in <<tune-for-disk-usage>>.
+
+== Freezing and unfreezing an index
+
+The freeze and unfreeze index APIs are used to freeze and unfreeze an index.
+A frozen index has almost no overhead on the cluster (except
+for maintaining its metadata in memory), and is blocked for write operations.
+When a frozen index is unfrozen, then index will go through the normal recovery process and become writeable again.
+
+The REST endpoint is `/{index}/_freeze` and `/{index}/_unfreeze`. For
+example:
+
+[source,js]
+--------------------------------------------------
+POST /my_index/_freeze
+
+POST /my_index/_unfreeze
+--------------------------------------------------
+// CONSOLE
+// TEST[s/^/PUT my_index\n/]
+
+
+[IMPORTANT]
+================================
+ Freezing an index will close the index and reopen it within the same API call. This causes primaries to not be allocated for a short
+ amount of time and causes the cluster to go red until the primaries are allocated again. This limitation might be removed in the future.
+================================
+
+== Searching a frozen index
+
+Frozen indices are throttled in order limit memory consumptions per node. The number of concurrently loaded frozen indices per node is
+limited by the number of threads in the <<search-throttled>> threadpool,  which is `1` by default. 
+Search requests will not be executed against frozen indices by default, even if a frozen index is named explicitly. Th is is ```
+to prevent accidental slowdowns by targeting a frozen index by mistake.. To include frozen indices a search request must be executed with
+the query parameter `ignore_throttled=false`.
+
+[source,js]
+--------------------------------------------------
+GET /twitter/_search?q=user:kimchy&ignore_throttled=false
+--------------------------------------------------
+// CONSOLE
+// TEST[setup:twitter]
+
+[IMPORTANT]
+================================
+While frozen indices are slow to search, they can be pre-filtered efficiently. The request parameter `pre_filter_shard_size` specifies
+a threshold that, when exceeded, will enforce a round-trip to pre-filter search shards that cannot possibly match.
+This filter phase can limit the number of shards significantly. For instance, if a date range filter is applied, then all indices (frozen or unfrozen) that do not contain documents within the date range can be skipped efficiently.
+The default value for `pre_filter_shard_size` is `128` but it's recommended to set it to `1` when searching frozen indices. There is no
+significant overhead associated with this pre-filter phase.
+================================
+
+

docs/reference/modules/threadpool.asciidoc

@@ -19,6 +19,10 @@ There are several thread pools, but the important ones include:
     `int((# of available_processors * 3) / 2) + 1`, and initial queue_size of
     `1000`.
 
+[[search-throttled]]`search_throttled`::
+    For count/search/suggest/get operations on `search_throttled indices`. Thread pool type is
+    `fixed_auto_queue_size` with a size of `1`, and initial queue_size of `100`.
+
 `get`::
     For get operations. Thread pool type is `fixed`
     with a size of `# of available processors`,

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);
     }
 

server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateService.java

@@ -100,46 +100,50 @@ protected ClusterStateUpdateResponse newResponse(boolean acknowledged) {
 
             @Override
             public ClusterState execute(ClusterState currentState) {
-                Set<IndexMetaData> indicesToClose = new HashSet<>();
-                for (Index index : request.indices()) {
-                    final IndexMetaData indexMetaData = currentState.metaData().getIndexSafe(index);
-                    if (indexMetaData.getState() != IndexMetaData.State.CLOSE) {
-                        indicesToClose.add(indexMetaData);
-                    }
-                }
+                return closeIndex(currentState, request.indices(), indicesAsString);
+            }
+        });
+    }
 
-                if (indicesToClose.isEmpty()) {
-                    return currentState;
-                }
+    public ClusterState closeIndex(ClusterState currentState, final Index[] indices, String indicesAsString) {
+        Set<IndexMetaData> indicesToClose = new HashSet<>();
+        for (Index index : indices) {
+            final IndexMetaData indexMetaData = currentState.metaData().getIndexSafe(index);
+            if (indexMetaData.getState() != IndexMetaData.State.CLOSE) {
+                indicesToClose.add(indexMetaData);
+            }
+        }
 
-                // Check if index closing conflicts with any running restores
-                RestoreService.checkIndexClosing(currentState, indicesToClose);
-                // Check if index closing conflicts with any running snapshots
-                SnapshotsService.checkIndexClosing(currentState, indicesToClose);
-                logger.info("closing indices [{}]", indicesAsString);
+        if (indicesToClose.isEmpty()) {
+            return currentState;
+        }
 
-                MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
-                ClusterBlocks.Builder blocksBuilder = ClusterBlocks.builder()
-                        .blocks(currentState.blocks());
-                for (IndexMetaData openIndexMetadata : indicesToClose) {
-                    final String indexName = openIndexMetadata.getIndex().getName();
-                    mdBuilder.put(IndexMetaData.builder(openIndexMetadata).state(IndexMetaData.State.CLOSE));
-                    blocksBuilder.addIndexBlock(indexName, INDEX_CLOSED_BLOCK);
-                }
+        // Check if index closing conflicts with any running restores
+        RestoreService.checkIndexClosing(currentState, indicesToClose);
+        // Check if index closing conflicts with any running snapshots
+        SnapshotsService.checkIndexClosing(currentState, indicesToClose);
+        logger.info("closing indices [{}]", indicesAsString);
+
+        MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
+        ClusterBlocks.Builder blocksBuilder = ClusterBlocks.builder()
+            .blocks(currentState.blocks());
+        for (IndexMetaData openIndexMetadata : indicesToClose) {
+            final String indexName = openIndexMetadata.getIndex().getName();
+            mdBuilder.put(IndexMetaData.builder(openIndexMetadata).state(IndexMetaData.State.CLOSE));
+            blocksBuilder.addIndexBlock(indexName, INDEX_CLOSED_BLOCK);
+        }
 
-                ClusterState updatedState = ClusterState.builder(currentState).metaData(mdBuilder).blocks(blocksBuilder).build();
+        ClusterState updatedState = ClusterState.builder(currentState).metaData(mdBuilder).blocks(blocksBuilder).build();
 
-                RoutingTable.Builder rtBuilder = RoutingTable.builder(currentState.routingTable());
-                for (IndexMetaData index : indicesToClose) {
-                    rtBuilder.remove(index.getIndex().getName());
-                }
+        RoutingTable.Builder rtBuilder = RoutingTable.builder(currentState.routingTable());
+        for (IndexMetaData index : indicesToClose) {
+            rtBuilder.remove(index.getIndex().getName());
+        }
 
-                //no explicit wait for other nodes needed as we use AckedClusterStateUpdateTask
-                return  allocationService.reroute(
-                        ClusterState.builder(updatedState).routingTable(rtBuilder.build()).build(),
-                        "indices closed [" + indicesAsString + "]");
-            }
-        });
+        //no explicit wait for other nodes needed as we use AckedClusterStateUpdateTask
+        return  allocationService.reroute(
+            ClusterState.builder(updatedState).routingTable(rtBuilder.build()).build(),
+            "indices closed [" + indicesAsString + "]");
     }
 
     public void openIndex(final OpenIndexClusterStateUpdateRequest request,

server/src/main/java/org/elasticsearch/index/engine/ReadOnlyEngine.java

@@ -20,6 +20,7 @@
 
 import org.apache.lucene.index.DirectoryReader;
 import org.apache.lucene.index.IndexCommit;
+import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.IndexWriter;
 import org.apache.lucene.index.SegmentCommitInfo;
 import org.apache.lucene.index.SegmentInfos;
@@ -66,7 +67,7 @@ public class ReadOnlyEngine extends Engine {
     private final IndexCommit indexCommit;
     private final Lock indexWriterLock;
     private final DocsStats docsStats;
-    protected final RamAccountingSearcherFactory searcherFactory;
+    private final RamAccountingSearcherFactory searcherFactory;
 
     /**
      * Creates a new ReadOnlyEngine. This ctor can also be used to open a read-only engine on top of an already opened
@@ -414,4 +415,8 @@ public void updateMaxUnsafeAutoIdTimestamp(long newTimestamp) {
     public void initializeMaxSeqNoOfUpdatesOrDeletes() {
         advanceMaxSeqNoOfUpdatesOrDeletes(seqNoStats.getMaxSeqNo());
     }
+
+    protected void processReaders(IndexReader reader, IndexReader previousReader) {
+        searcherFactory.processReaders(reader, previousReader);
+    }
 }

test/framework/src/main/java/org/elasticsearch/index/shard/IndexShardTestCase.java

@@ -389,14 +389,26 @@ protected IndexShard reinitShard(IndexShard current, IndexingOperationListener..
      * @param listeners new listerns to use for the newly created shard
      */
     protected IndexShard reinitShard(IndexShard current, ShardRouting routing, IndexingOperationListener... listeners) throws IOException {
+        return reinitShard(current, routing, current.engineFactory, listeners);
+    }
+
+    /**
+     * Takes an existing shard, closes it and starts a new initialing shard at the same location
+     *
+     * @param routing   the shard routing to use for the newly created shard.
+     * @param listeners new listerns to use for the newly created shard
+     * @param engineFactory the engine factory for the new shard
+     */
+    protected IndexShard reinitShard(IndexShard current, ShardRouting routing, EngineFactory engineFactory,
+                                     IndexingOperationListener... listeners) throws IOException {
         closeShards(current);
         return newShard(
                 routing,
                 current.shardPath(),
                 current.indexSettings().getIndexMetaData(),
                 null,
                 null,
-                current.engineFactory,
+                engineFactory,
                 current.getGlobalCheckpointSyncer(),
             EMPTY_EVENT_LISTENER, listeners);
     }

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

@@ -158,7 +158,7 @@ private synchronized DirectoryReader getOrOpenReader() throws IOException {
                     listeners.beforeRefresh();
                 }
                 reader = DirectoryReader.open(engineConfig.getStore().directory());
-                searcherFactory.processReaders(reader, null);
+                processReaders(reader, null);
                 reader = lastOpenedReader = wrapReader(reader, Function.identity());
                 reader.getReaderCacheHelper().addClosedListener(this::onReaderClosed);
                 for (ReferenceManager.RefreshListener listeners : config ().getInternalRefreshListener()) {

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

@@ -6,13 +6,15 @@
 package org.elasticsearch.xpack.core;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.action.support.PlainActionFuture;
 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.TransportFreezeIndexAction;
+import org.elasticsearch.xpack.core.action.TransportFreezeIndexAction.FreezeIndexAction;
+import org.elasticsearch.xpack.core.action.TransportFreezeIndexAction.FreezeRequest;
+import org.elasticsearch.xpack.core.action.TransportFreezeIndexAction.FreezeResponse;
 import org.elasticsearch.xpack.core.action.XPackInfoAction;
 import org.elasticsearch.xpack.core.action.XPackInfoRequestBuilder;
 import org.elasticsearch.xpack.core.ccr.client.CcrClient;
@@ -25,6 +27,7 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ExecutionException;
 
 import static org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken.BASIC_AUTH_HEADER;
 import static org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
@@ -106,7 +109,20 @@ public void info(XPackInfoRequest request, ActionListener<XPackInfoResponse> lis
         client.execute(XPackInfoAction.INSTANCE, request, listener);
     }
 
-    public void freeze(TransportFreezeIndexAction.FreezeRequest request, ActionListener<AcknowledgedResponse> listener) {
-        client.execute(TransportFreezeIndexAction.FreezeIndexAction.INSTANCE, request, listener);
+    /**
+     * Freezes or unfreeze one or more indices
+     */
+    public void freeze(FreezeRequest request, ActionListener<FreezeResponse> listener) {
+        client.execute(FreezeIndexAction.INSTANCE, request, listener);
+    }
+
+    /**
+     * Freeze or unfreeze one or more indices
+     */
+    public FreezeResponse freeze(FreezeRequest request)
+        throws ExecutionException, InterruptedException {
+        PlainActionFuture<FreezeResponse> future = new PlainActionFuture<>();
+        freeze(request, future);
+        return future.get();
     }
 }

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

@@ -37,6 +37,7 @@
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.Transport;
+import org.elasticsearch.xpack.core.action.TransportFreezeIndexAction;
 import org.elasticsearch.xpack.core.action.XPackInfoAction;
 import org.elasticsearch.xpack.core.action.XPackUsageAction;
 import org.elasticsearch.xpack.core.beats.BeatsFeatureSetUsage;
@@ -344,7 +345,9 @@ public List<Action<? extends ActionResponse>> getClientActions() {
                 ExplainLifecycleAction.INSTANCE,
                 RemoveIndexLifecyclePolicyAction.INSTANCE,
                 MoveToStepAction.INSTANCE,
-                RetryAction.INSTANCE
+                RetryAction.INSTANCE,
+                // Frozen indices
+                TransportFreezeIndexAction.FreezeIndexAction.INSTANCE
         );
     }
 

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

@@ -63,6 +63,7 @@
 import org.elasticsearch.xpack.core.action.XPackInfoAction;
 import org.elasticsearch.xpack.core.action.XPackUsageAction;
 import org.elasticsearch.xpack.core.ml.MlMetadata;
+import org.elasticsearch.xpack.core.rest.action.RestFreezeIndexAction;
 import org.elasticsearch.xpack.core.rest.action.RestXPackInfoAction;
 import org.elasticsearch.xpack.core.rest.action.RestXPackUsageAction;
 import org.elasticsearch.xpack.core.security.authc.TokenMetaData;
@@ -297,6 +298,7 @@ public List<RestHandler> getRestHandlers(Settings settings, RestController restC
         List<RestHandler> handlers = new ArrayList<>();
         handlers.add(new RestXPackInfoAction(settings, restController));
         handlers.add(new RestXPackUsageAction(settings, restController));
+        handlers.add(new RestFreezeIndexAction(settings, restController));
         handlers.addAll(licensing.getRestHandlers(settings, restController, clusterSettings, indexScopedSettings, settingsFilter,
                 indexNameExpressionResolver, nodesInCluster));
         return handlers;