[ILM] Add unfollow action

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

@@ -56,6 +56,7 @@
 import org.elasticsearch.xpack.core.indexlifecycle.RolloverAction;
 import org.elasticsearch.xpack.core.indexlifecycle.ShrinkAction;
 import org.elasticsearch.xpack.core.indexlifecycle.TimeseriesLifecycleType;
+import org.elasticsearch.xpack.core.indexlifecycle.UnfollowAction;
 import org.elasticsearch.xpack.core.indexlifecycle.action.DeleteLifecycleAction;
 import org.elasticsearch.xpack.core.indexlifecycle.action.ExplainLifecycleAction;
 import org.elasticsearch.xpack.core.indexlifecycle.action.GetLifecycleAction;
@@ -425,7 +426,8 @@ public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, RolloverAction.NAME, RolloverAction::new),
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, ShrinkAction.NAME, ShrinkAction::new),
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, DeleteAction.NAME, DeleteAction::new),
-                new NamedWriteableRegistry.Entry(LifecycleAction.class, FreezeAction.NAME, FreezeAction::new)
+                new NamedWriteableRegistry.Entry(LifecycleAction.class, FreezeAction.NAME, FreezeAction::new),
+                new NamedWriteableRegistry.Entry(LifecycleAction.class, UnfollowAction.NAME, UnfollowAction::new)
         );
     }
 

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

@@ -34,7 +34,7 @@ public class TimeseriesLifecycleType implements LifecycleType {
 
     public static final String TYPE = "timeseries";
     static final List<String> VALID_PHASES = Arrays.asList("hot", "warm", "cold", "delete");
-    static final List<String> ORDERED_VALID_HOT_ACTIONS = Collections.singletonList(RolloverAction.NAME);
+    static final List<String> ORDERED_VALID_HOT_ACTIONS = Arrays.asList(UnfollowAction.NAME, RolloverAction.NAME);
     static final List<String> ORDERED_VALID_WARM_ACTIONS = Arrays.asList(ReadOnlyAction.NAME, AllocateAction.NAME,
         ShrinkAction.NAME, ForceMergeAction.NAME);
     static final List<String> ORDERED_VALID_COLD_ACTIONS = Arrays.asList(AllocateAction.NAME, FreezeAction.NAME);

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

@@ -0,0 +1,97 @@
+/*
+ * 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.indexlifecycle;
+
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ObjectParser;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.xpack.core.indexlifecycle.Step.StepKey;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+public final class UnfollowAction implements LifecycleAction {
+
+    public static final String NAME = "unfollow";
+    public static final String CCR_METADATA_KEY = "ccr";
+
+    public UnfollowAction() {}
+
+    @Override
+    public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
+        StepKey indexingComplete = new StepKey(phase, NAME, WaitForIndexingCompleteStep.NAME);
+        StepKey waitForFollowShardTasks = new StepKey(phase, NAME, WaitForFollowShardTasksStep.NAME);
+        StepKey unfollowIndex = new StepKey(phase, NAME, UnfollowFollowIndexStep.NAME);
+
+        WaitForIndexingCompleteStep step1 = new WaitForIndexingCompleteStep(indexingComplete, waitForFollowShardTasks);
+        WaitForFollowShardTasksStep step2 = new WaitForFollowShardTasksStep(waitForFollowShardTasks, unfollowIndex, client);
+        UnfollowFollowIndexStep  step3 = new UnfollowFollowIndexStep(unfollowIndex, nextStepKey, client);
+        return Arrays.asList(step1, step2, step3);
+    }
+
+    @Override
+    public List<StepKey> toStepKeys(String phase) {
+        StepKey indexingCompleteStep = new StepKey(phase, NAME, WaitForIndexingCompleteStep.NAME);
+        StepKey waitForFollowShardTasksStep = new StepKey(phase, NAME, WaitForFollowShardTasksStep.NAME);
+        StepKey unfollowIndexStep = new StepKey(phase, NAME, UnfollowFollowIndexStep.NAME);
+        return Arrays.asList(indexingCompleteStep, waitForFollowShardTasksStep, unfollowIndexStep);
+    }
+
+    @Override
+    public boolean isSafeAction() {
+        // There are no settings to change, so therefor this action should be safe:
+        return true;
+    }
+
+    @Override
+    public String getWriteableName() {
+        return NAME;
+    }
+
+    public UnfollowAction(StreamInput in) throws IOException {}
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {}
+
+    private static final ObjectParser<UnfollowAction, Void> PARSER = new ObjectParser<>(NAME, UnfollowAction::new);
+
+    public static UnfollowAction parse(XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public int hashCode() {
+        return 1;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null) {
+            return false;
+        }
+        if (obj.getClass() != getClass()) {
+            return false;
+        }
+        return true;
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
+}

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

@@ -0,0 +1,85 @@
+/*
+ * 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.indexlifecycle;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
+import org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.xpack.core.ccr.action.PauseFollowAction;
+import org.elasticsearch.xpack.core.ccr.action.UnfollowAction;
+
+import java.util.Map;
+
+import static org.elasticsearch.xpack.core.indexlifecycle.UnfollowAction.CCR_METADATA_KEY;
+
+final class UnfollowFollowIndexStep extends AsyncActionStep {
+
+    static final String NAME = "unfollow-index";
+
+    UnfollowFollowIndexStep(StepKey key, StepKey nextStepKey, Client client) {
+        super(key, nextStepKey, client);
+    }
+
+    @Override
+    public void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState, Listener listener) {
+        String followerIndex = indexMetaData.getIndex().getName();
+        Map<String, String> customIndexMetadata = indexMetaData.getCustomData(CCR_METADATA_KEY);
+        if (customIndexMetadata == null) {
+            listener.onResponse(true);
+            return;
+        }
+
+        pauseFollowerIndex(followerIndex, listener);
+    }
+
+    void pauseFollowerIndex(final String followerIndex, final Listener listener) {
+        PauseFollowAction.Request request = new PauseFollowAction.Request(followerIndex);
+        getClient().execute(PauseFollowAction.INSTANCE, request, ActionListener.wrap(
+            r -> {
+                assert r.isAcknowledged();
+                closeFollowerIndex(followerIndex, listener);
+            },
+            listener::onFailure
+        ));
+    }
+
+    void closeFollowerIndex(final String followerIndex, final Listener listener) {
+        CloseIndexRequest closeIndexRequest = new CloseIndexRequest(followerIndex);
+        getClient().admin().indices().close(closeIndexRequest, ActionListener.wrap(
+            acknowledgedResponse -> {
+                assert acknowledgedResponse.isAcknowledged();
+                unfollow(followerIndex, listener);
+            },
+            listener::onFailure)
+        );
+    }
+
+    void unfollow(final String followerIndex, final Listener listener) {
+        UnfollowAction.Request request = new UnfollowAction.Request(followerIndex);
+        getClient().execute(UnfollowAction.INSTANCE, request, ActionListener.wrap(
+            r -> {
+                assert r.isAcknowledged();
+                openIndex(followerIndex, listener);
+            },
+            listener::onFailure
+        ));
+    }
+
+    void openIndex(final String index, final Listener listener) {
+        OpenIndexRequest request = new OpenIndexRequest(index);
+        getClient().admin().indices().open(request, ActionListener.wrap(
+            openIndexResponse -> {
+                assert openIndexResponse.isAcknowledged();
+                listener.onResponse(true);
+            },
+            listener::onFailure
+        ));
+    }
+
+}

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

@@ -0,0 +1,176 @@
+/*
+ * 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.indexlifecycle;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
+import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import static org.elasticsearch.xpack.core.indexlifecycle.UnfollowAction.CCR_METADATA_KEY;
+
+final class WaitForFollowShardTasksStep extends AsyncWaitStep {
+
+    static final String NAME = "wait-for-follow-shard-tasks";
+
+    WaitForFollowShardTasksStep(StepKey key, StepKey nextStepKey, Client client) {
+        super(key, nextStepKey, client);
+    }
+
+    @Override
+    public void evaluateCondition(IndexMetaData indexMetaData, Listener listener) {
+        Map<String, String> customIndexMetadata = indexMetaData.getCustomData(CCR_METADATA_KEY);
+        if (customIndexMetadata == null) {
+            listener.onResponse(true, null);
+            return;
+        }
+
+        FollowStatsAction.StatsRequest request = new FollowStatsAction.StatsRequest();
+        request.setIndices(new String[]{indexMetaData.getIndex().getName()});
+        getClient().execute(FollowStatsAction.INSTANCE, request,
+            ActionListener.wrap(r -> handleResponse(r, listener), listener::onFailure));
+    }
+
+    void handleResponse(FollowStatsAction.StatsResponses responses, Listener listener) {
+        List<ShardFollowNodeTaskStatus> unSyncedShardFollowStatuses = responses.getStatsResponses()
+            .stream()
+            .map(FollowStatsAction.StatsResponse::status)
+            .filter(shardFollowStatus -> shardFollowStatus.leaderGlobalCheckpoint() != shardFollowStatus.followerGlobalCheckpoint())
+            .collect(Collectors.toList());
+
+        // Follow stats api needs to return stats for follower index and all shard follow tasks should be synced:
+        boolean conditionMet = responses.getStatsResponses().size() > 0 && unSyncedShardFollowStatuses.isEmpty();
+        if (conditionMet) {
+            listener.onResponse(true, null);
+        } else {
+            List<Info.ShardFollowTaskInfo> shardFollowTaskInfos = unSyncedShardFollowStatuses
+                .stream()
+                .map(status -> new Info.ShardFollowTaskInfo(status.followerIndex(), status.getShardId(),
+                    status.leaderGlobalCheckpoint(), status.followerGlobalCheckpoint()))
+                .collect(Collectors.toList());
+            listener.onResponse(false, new Info(shardFollowTaskInfos));
+        }
+    }
+
+    static final class Info implements ToXContentObject {
+
+        static final ParseField SHARD_FOLLOW_TASKS = new ParseField("shard_follow_tasks");
+        static final ParseField MESSAGE = new ParseField("message");
+
+        private final List<ShardFollowTaskInfo> shardFollowTaskInfos;
+
+        Info(List<ShardFollowTaskInfo> shardFollowTaskInfos) {
+            this.shardFollowTaskInfos = shardFollowTaskInfos;
+        }
+
+        List<ShardFollowTaskInfo> getShardFollowTaskInfos() {
+            return shardFollowTaskInfos;
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            builder.field(SHARD_FOLLOW_TASKS.getPreferredName(), shardFollowTaskInfos);
+            String message = "Waiting for [" + shardFollowTaskInfos.size() + "] shard follow tasks to be in sync";
+            builder.field(MESSAGE.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(shardFollowTaskInfos, info.shardFollowTaskInfos);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(shardFollowTaskInfos);
+        }
+
+        @Override
+        public String toString() {
+            return Strings.toString(this);
+        }
+
+        static final class ShardFollowTaskInfo implements ToXContentObject {
+
+            static final ParseField FOLLOWER_INDEX_FIELD = new ParseField("follower_index");
+            static final ParseField SHARD_ID_FIELD = new ParseField("shard_id");
+            static final ParseField LEADER_GLOBAL_CHECKPOINT_FIELD = new ParseField("leader_global_checkpoint");
+            static final ParseField FOLLOWER_GLOBAL_CHECKPOINT_FIELD = new ParseField("follower_global_checkpoint");
+
+            private final String followerIndex;
+            private final int shardId;
+            private final long leaderGlobalCheckpoint;
+            private final long followerGlobalCheckpoint;
+
+            ShardFollowTaskInfo(String followerIndex, int shardId, long leaderGlobalCheckpoint, long followerGlobalCheckpoint) {
+                this.followerIndex = followerIndex;
+                this.shardId = shardId;
+                this.leaderGlobalCheckpoint = leaderGlobalCheckpoint;
+                this.followerGlobalCheckpoint = followerGlobalCheckpoint;
+            }
+
+            String getFollowerIndex() {
+                return followerIndex;
+            }
+
+
+            int getShardId() {
+                return shardId;
+            }
+
+            long getLeaderGlobalCheckpoint() {
+                return leaderGlobalCheckpoint;
+            }
+
+            long getFollowerGlobalCheckpoint() {
+                return followerGlobalCheckpoint;
+            }
+
+            @Override
+            public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+                builder.startObject();
+                builder.field(FOLLOWER_INDEX_FIELD.getPreferredName(), followerIndex);
+                builder.field(SHARD_ID_FIELD.getPreferredName(), shardId);
+                builder.field(LEADER_GLOBAL_CHECKPOINT_FIELD.getPreferredName(), leaderGlobalCheckpoint);
+                builder.field(FOLLOWER_GLOBAL_CHECKPOINT_FIELD.getPreferredName(), followerGlobalCheckpoint);
+                builder.endObject();
+                return builder;
+            }
+
+            @Override
+            public boolean equals(Object o) {
+                if (this == o) return true;
+                if (o == null || getClass() != o.getClass()) return false;
+                ShardFollowTaskInfo that = (ShardFollowTaskInfo) o;
+                return shardId == that.shardId &&
+                    leaderGlobalCheckpoint == that.leaderGlobalCheckpoint &&
+                    followerGlobalCheckpoint == that.followerGlobalCheckpoint &&
+                    Objects.equals(followerIndex, that.followerIndex);
+            }
+
+            @Override
+            public int hashCode() {
+                return Objects.hash(followerIndex, shardId, leaderGlobalCheckpoint, followerGlobalCheckpoint);
+            }
+        }
+    }
+}