Adding best_compression

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

@@ -0,0 +1,44 @@
+/*
+ * 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.core.ilm;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+
+/**
+ * Invokes a close step on a single index.
+ */
+
+public class CloseIndexStep extends AsyncActionStep {
+    public static final String NAME = "close-index";
+
+    CloseIndexStep(StepKey key, StepKey nextStepKey, Client client) {
+        super(key, nextStepKey, client);
+    }
+
+    @Override
+    public void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState,
+                              ClusterStateObserver observer, Listener listener) {
+        if (indexMetaData.getState() == IndexMetaData.State.OPEN) {
+            CloseIndexRequest request = new CloseIndexRequest(indexMetaData.getIndex().getName());
+            getClient().admin().indices()
+                .close(request, ActionListener.wrap(closeIndexResponse -> {
+                    assert closeIndexResponse.isAcknowledged() : "close index response is not acknowledged";
+                    listener.onResponse(true);
+                }, listener::onFailure));
+        }
+        else {
+            listener.onResponse(true);
+        }
+    }
+
+
+}

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

@@ -5,7 +5,10 @@
  */
 package org.elasticsearch.xpack.core.ilm;
 
+import org.apache.lucene.codecs.Codec;
+import org.elasticsearch.Version;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.Strings;
@@ -15,6 +18,8 @@
 import org.elasticsearch.common.xcontent.ConstructingObjectParser;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.index.codec.CodecService;
+import org.elasticsearch.index.engine.EngineConfig;
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 
 import java.io.IOException;
@@ -28,42 +33,61 @@
 public class ForceMergeAction implements LifecycleAction {
     public static final String NAME = "forcemerge";
     public static final ParseField MAX_NUM_SEGMENTS_FIELD = new ParseField("max_num_segments");
+    public static final ParseField CODEC = new ParseField("index_codec");
 
     private static final ConstructingObjectParser<ForceMergeAction, Void> PARSER = new ConstructingObjectParser<>(NAME,
         false, a -> {
         int maxNumSegments = (int) a[0];
-        return new ForceMergeAction(maxNumSegments);
+        Codec codec = a[1] != null ? Codec.forName((String) a[1]) : Codec.getDefault();
+        return new ForceMergeAction(maxNumSegments, codec);
     });
 
     static {
         PARSER.declareInt(ConstructingObjectParser.constructorArg(), MAX_NUM_SEGMENTS_FIELD);
+        PARSER.declareInt(ConstructingObjectParser.constructorArg(), CODEC);
     }
 
     private final int maxNumSegments;
+    private final Codec codec;
 
     public static ForceMergeAction parse(XContentParser parser) {
         return PARSER.apply(parser, null);
     }
 
-    public ForceMergeAction(int maxNumSegments) {
+    public ForceMergeAction(int maxNumSegments, Codec codec) {
         if (maxNumSegments <= 0) {
             throw new IllegalArgumentException("[" + MAX_NUM_SEGMENTS_FIELD.getPreferredName()
                 + "] must be a positive integer");
         }
         this.maxNumSegments = maxNumSegments;
+        this.codec = codec;
     }
 
     public ForceMergeAction(StreamInput in) throws IOException {
         this.maxNumSegments = in.readVInt();
+        if (in.getVersion().onOrAfter(Version.V_8_0_0)) {
+            this.codec = Codec.forName(in.readString());
+        } else {
+            this.codec = Codec.getDefault();
+        }
     }
 
     public int getMaxNumSegments() {
         return maxNumSegments;
     }
 
+    public Codec getCodec() {
+        return this.codec;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeVInt(maxNumSegments);
+        if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+            out.writeString(codec.getName());
+        } else {
+            out.writeString(Codec.getDefault().getName());
+        }
     }
 
     @Override
@@ -80,18 +104,39 @@ public boolean isSafeAction() {
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         builder.field(MAX_NUM_SEGMENTS_FIELD.getPreferredName(), maxNumSegments);
+        builder.field(CODEC.getPreferredName(), codec);
         builder.endObject();
         return builder;
     }
 
     @Override
     public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey) {
         Settings readOnlySettings = Settings.builder().put(IndexMetaData.SETTING_BLOCKS_WRITE, true).build();
+        Settings bestCompressionSettings = Settings.builder()
+            .put(EngineConfig.INDEX_CODEC_SETTING.getKey(), CodecService.BEST_COMPRESSION_CODEC).build();
 
         StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyAction.NAME);
         StepKey forceMergeKey = new StepKey(phase, NAME, ForceMergeStep.NAME);
         StepKey countKey = new StepKey(phase, NAME, SegmentCountStep.NAME);
 
+
+        if (codec.getName().equals(CodecService.BEST_COMPRESSION_CODEC)) {
+            StepKey closeKey = new StepKey(phase, NAME, CloseIndexStep.NAME);
+            StepKey openKey = new StepKey(phase, NAME, OpenIndexStep.NAME);
+            StepKey waitForGreenIndexKey = new StepKey(phase, NAME, WaitForIndexColorStep.NAME);
+            StepKey updateCompressionKey = new StepKey(phase, NAME, UpdateSettingsStep.NAME);
+
+            CloseIndexStep closeIndexStep = new CloseIndexStep(closeKey, updateCompressionKey, client);
+            UpdateSettingsStep updateBestCompressionSettings = new UpdateSettingsStep(updateCompressionKey,
+                openKey, client, bestCompressionSettings);
+            OpenIndexStep openIndexStep = new OpenIndexStep(openKey, waitForGreenIndexKey, client);
+            WaitForIndexColorStep waitForIndexGreenStep = new WaitForIndexColorStep(waitForGreenIndexKey,
+                forceMergeKey, ClusterHealthStatus.GREEN);
+            ForceMergeStep forceMergeStep = new ForceMergeStep(forceMergeKey, nextStepKey, client, maxNumSegments);
+            return Arrays.asList(closeIndexStep, updateBestCompressionSettings,
+                openIndexStep, waitForIndexGreenStep, forceMergeStep);
+        }
+
         UpdateSettingsStep readOnlyStep = new UpdateSettingsStep(readOnlyKey, forceMergeKey, client, readOnlySettings);
         ForceMergeStep forceMergeStep = new ForceMergeStep(forceMergeKey, countKey, client, maxNumSegments);
         SegmentCountStep segmentCountStep = new SegmentCountStep(countKey, nextStepKey, client, maxNumSegments);
@@ -100,7 +145,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
 
     @Override
     public int hashCode() {
-        return Objects.hash(maxNumSegments);
+        return Objects.hash(maxNumSegments, codec);
     }
 
     @Override
@@ -112,7 +157,8 @@ public boolean equals(Object obj) {
             return false;
         }
         ForceMergeAction other = (ForceMergeAction) obj;
-        return Objects.equals(maxNumSegments, other.maxNumSegments);
+        return Objects.equals(maxNumSegments, other.maxNumSegments)
+            && Objects.equals(codec, other.codec);
     }
 
     @Override

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

@@ -0,0 +1,44 @@
+/*
+ * 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.core.ilm;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+
+/**
+ * Invokes a open step on a single index.
+ */
+
+final class OpenIndexStep extends AsyncActionStep {
+
+    static final String NAME = "open-index";
+
+    OpenIndexStep(StepKey key, StepKey nextStepKey, Client client) {
+        super(key, nextStepKey, client);
+    }
+
+    @Override
+    public void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState,
+                              ClusterStateObserver observer, Listener listener) {
+        if (indexMetaData.getState() == IndexMetaData.State.CLOSE) {
+            OpenIndexRequest request = new OpenIndexRequest(indexMetaData.getIndex().getName());
+            getClient().admin().indices()
+                .open(request,
+                    ActionListener.wrap(openIndexResponse-> {
+                        assert openIndexResponse.isAcknowledged() : "open Index response is not acknowledged";
+                        listener.onResponse(true);
+                    }, listener::onFailure));
+
+        } else {
+            listener.onResponse(true);
+        }
+    }
+}

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

@@ -0,0 +1,158 @@
+/*
+ * 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.core.ilm;
+
+import com.carrotsearch.hppc.cursors.ObjectCursor;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.cluster.routing.IndexRoutingTable;
+import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
+import org.elasticsearch.cluster.routing.RoutingTable;
+import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.index.Index;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * Wait Step for index based on color
+ * */
+
+class WaitForIndexColorStep extends ClusterStateWaitStep {
+
+    static final String NAME = "wait-for-index-color-step";
+
+    private final ClusterHealthStatus color;
+
+    WaitForIndexColorStep(StepKey key, StepKey nextStepKey, ClusterHealthStatus color) {
+        super(key, nextStepKey);
+        this.color = color;
+    }
+
+    public ClusterHealthStatus getColor() {
+        return this.color;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), this.color);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null) return false;
+        if (!(obj instanceof WaitForIndexColorStep)) return false;
+        WaitForIndexColorStep other = (WaitForIndexColorStep) obj;
+        return super.equals(obj) && Objects.equals(this.color, other.color);
+    }
+
+    @Override
+    public Result isConditionMet(Index index, ClusterState clusterState) {
+        RoutingTable routingTable = clusterState.routingTable();
+        IndexRoutingTable indexRoutingTable = routingTable.index(index);
+        Result result;
+        switch (this.color) {
+            case GREEN:
+                result = waitForGreen(indexRoutingTable);
+                break;
+            case YELLOW:
+                result = waitForYellow(indexRoutingTable);
+                break;
+            case RED:
+                result = waitForRed(indexRoutingTable);
+                break;
+            default:
+                result = new Result(false, new Info("No index color match"));
+                break;
+        }
+        return result;
+    }
+
+    private Result waitForRed(IndexRoutingTable indexRoutingTable) {
+        if (indexRoutingTable == null) {
+            return new Result(true, new Info("Index is red"));
+        }
+        return new Result(false, new Info("Index is not red"));
+    }
+
+    private Result waitForYellow(IndexRoutingTable indexRoutingTable) {
+        if (indexRoutingTable == null) {
+            return new Result(false, new Info("index is red; no IndexRoutingTable"));
+        }
+
+        boolean indexIsAtLeastYellow = indexRoutingTable.allPrimaryShardsActive();
+        if (indexIsAtLeastYellow) {
+            return new Result(true, null);
+        } else {
+            return new Result(false, new Info("index is red; not all primary shards are active"));
+        }
+    }
+
+    private Result waitForGreen(IndexRoutingTable indexRoutingTable) {
+        if (indexRoutingTable == null) {
+            return new Result(false, new Info("index is red; no IndexRoutingTable"));
+        }
+
+        boolean indexIsGreen = false;
+        if (indexRoutingTable.allPrimaryShardsActive()) {
+            boolean replicaIndexIsGreen = false;
+            for (ObjectCursor<IndexShardRoutingTable> shardRouting : indexRoutingTable.getShards().values()) {
+                replicaIndexIsGreen = shardRouting.value.replicaShards().stream().allMatch(ShardRouting::active);
+                if (!replicaIndexIsGreen) {
+                    return new Result(false, new Info("index is yellow; not all replica shards are active"));
+                }
+            }
+            indexIsGreen = replicaIndexIsGreen;
+        }
+
+
+        if (indexIsGreen) {
+            return new Result(true, null);
+        } else {
+            return new Result(false, new Info("index is not green; not all shards are active"));
+        }
+    }
+
+    static final class Info implements ToXContentObject {
+
+        static final ParseField MESSAGE_FIELD = new ParseField("message");
+
+        private final String message;
+
+        Info(String message) {
+            this.message = message;
+        }
+
+        String getMessage() {
+            return message;
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            builder.field(MESSAGE_FIELD.getPreferredName(), message);
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Info info = (Info) o;
+            return Objects.equals(getMessage(), info.getMessage());
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(getMessage());
+        }
+    }
+}