Inject Unfollow before Rollover and Shrink

client/rest-high-level/build.gradle

@@ -107,6 +107,7 @@ integTestCluster {
   // Truststore settings are not used since TLS is not enabled. Included for testing the get certificates API
   setting 'xpack.security.http.ssl.certificate_authorities', 'testnode.crt'
   setting 'xpack.security.transport.ssl.truststore.path', 'testnode.jks'
+  setting 'indices.lifecycle.poll_interval', '1000ms'
   keystoreSetting 'xpack.security.transport.ssl.truststore.secure_password', 'testnode'
   setupCommand 'setupDummyUser',
           'bin/elasticsearch-users',

client/rest-high-level/src/test/java/org/elasticsearch/client/IndexLifecycleIT.java

@@ -184,31 +184,37 @@ public void testExplainLifecycle() throws Exception {
 
         createIndex("squash", Settings.EMPTY);
 
-        ExplainLifecycleRequest req = new ExplainLifecycleRequest("foo-01", "baz-01", "squash");
-        ExplainLifecycleResponse response = execute(req, highLevelClient().indexLifecycle()::explainLifecycle,
+        // The injected Unfollow step will run pretty rapidly here, so we need
+        // to wait for it to settle into the "stable" step of waiting to be
+        // ready to roll over
+        assertBusy(() -> {
+            ExplainLifecycleRequest req = new ExplainLifecycleRequest("foo-01", "baz-01", "squash");
+            ExplainLifecycleResponse response = execute(req, highLevelClient().indexLifecycle()::explainLifecycle,
                 highLevelClient().indexLifecycle()::explainLifecycleAsync);
-        Map<String, IndexLifecycleExplainResponse> indexResponses = response.getIndexResponses();
-        assertEquals(3, indexResponses.size());
-        IndexLifecycleExplainResponse fooResponse = indexResponses.get("foo-01");
-        assertNotNull(fooResponse);
-        assertTrue(fooResponse.managedByILM());
-        assertEquals("foo-01", fooResponse.getIndex());
-        assertEquals("hot", fooResponse.getPhase());
-        assertEquals("rollover", fooResponse.getAction());
-        assertEquals("check-rollover-ready", fooResponse.getStep());
-        assertEquals(new PhaseExecutionInfo(policy.getName(), new Phase("", hotPhase.getMinimumAge(), hotPhase.getActions()),
+            Map<String, IndexLifecycleExplainResponse> indexResponses = response.getIndexResponses();
+            assertEquals(3, indexResponses.size());
+            IndexLifecycleExplainResponse fooResponse = indexResponses.get("foo-01");
+            assertNotNull(fooResponse);
+            assertTrue(fooResponse.managedByILM());
+            assertEquals("foo-01", fooResponse.getIndex());
+            assertEquals("hot", fooResponse.getPhase());
+            assertEquals("rollover", fooResponse.getAction());
+            assertEquals("check-rollover-ready", fooResponse.getStep());
+            assertEquals(new PhaseExecutionInfo(policy.getName(), new Phase("", hotPhase.getMinimumAge(), hotPhase.getActions()),
                 1L, expectedPolicyModifiedDate), fooResponse.getPhaseExecutionInfo());
-        IndexLifecycleExplainResponse bazResponse = indexResponses.get("baz-01");
-        assertNotNull(bazResponse);
-        assertTrue(bazResponse.managedByILM());
-        assertEquals("baz-01", bazResponse.getIndex());
-        assertEquals("hot", bazResponse.getPhase());
-        assertEquals("rollover", bazResponse.getAction());
-        assertEquals("check-rollover-ready", bazResponse.getStep());
-        IndexLifecycleExplainResponse squashResponse = indexResponses.get("squash");
-        assertNotNull(squashResponse);
-        assertFalse(squashResponse.managedByILM());
-        assertEquals("squash", squashResponse.getIndex());
+            IndexLifecycleExplainResponse bazResponse = indexResponses.get("baz-01");
+            assertNotNull(bazResponse);
+            assertTrue(bazResponse.managedByILM());
+            assertEquals("baz-01", bazResponse.getIndex());
+            assertEquals("hot", bazResponse.getPhase());
+            assertEquals("rollover", bazResponse.getAction());
+            assertEquals("check-rollover-ready", bazResponse.getStep());
+            IndexLifecycleExplainResponse squashResponse = indexResponses.get("squash");
+            assertNotNull(squashResponse);
+            assertFalse(squashResponse.managedByILM());
+            assertEquals("squash", squashResponse.getIndex());
+
+        });
     }
 
     public void testDeleteLifecycle() throws IOException {

client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/ILMDocumentationIT.java

@@ -22,6 +22,7 @@
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.LatchedActionListener;
+import org.elasticsearch.action.admin.indices.alias.Alias;
 import org.elasticsearch.client.ESRestHighLevelClientTestCase;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
@@ -38,17 +39,17 @@
 import org.elasticsearch.client.indexlifecycle.LifecycleManagementStatusRequest;
 import org.elasticsearch.client.indexlifecycle.LifecycleManagementStatusResponse;
 import org.elasticsearch.client.indexlifecycle.LifecyclePolicy;
-import org.elasticsearch.client.indexlifecycle.OperationMode;
 import org.elasticsearch.client.indexlifecycle.LifecyclePolicyMetadata;
+import org.elasticsearch.client.indexlifecycle.OperationMode;
 import org.elasticsearch.client.indexlifecycle.Phase;
 import org.elasticsearch.client.indexlifecycle.PutLifecyclePolicyRequest;
 import org.elasticsearch.client.indexlifecycle.RemoveIndexLifecyclePolicyRequest;
 import org.elasticsearch.client.indexlifecycle.RemoveIndexLifecyclePolicyResponse;
 import org.elasticsearch.client.indexlifecycle.RetryLifecyclePolicyRequest;
 import org.elasticsearch.client.indexlifecycle.RolloverAction;
+import org.elasticsearch.client.indexlifecycle.ShrinkAction;
 import org.elasticsearch.client.indexlifecycle.StartILMRequest;
 import org.elasticsearch.client.indexlifecycle.StopILMRequest;
-import org.elasticsearch.client.indexlifecycle.ShrinkAction;
 import org.elasticsearch.client.indices.CreateIndexRequest;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Strings;
@@ -337,11 +338,13 @@ public void testExplainLifecycle() throws Exception {
                 new PutLifecyclePolicyRequest(policy);
             client.indexLifecycle().putLifecyclePolicy(putRequest, RequestOptions.DEFAULT);
 
-            CreateIndexRequest createIndexRequest = new CreateIndexRequest("my_index")
+            CreateIndexRequest createIndexRequest = new CreateIndexRequest("my_index-1")
                 .settings(Settings.builder()
                     .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
                     .put("index.lifecycle.name", "my_policy")
+                    .put("index.lifecycle.rollover_alias", "my_alias")
                     .build());
+            createIndexRequest.alias(new Alias("my_alias").writeIndex(true));
             client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
             CreateIndexRequest createOtherIndexRequest = new CreateIndexRequest("other_index")
                 .settings(Settings.builder()
@@ -352,58 +355,62 @@ public void testExplainLifecycle() throws Exception {
 
             // wait for the policy to become active
             assertBusy(() -> assertNotNull(client.indexLifecycle()
-                .explainLifecycle(new ExplainLifecycleRequest("my_index"), RequestOptions.DEFAULT)
-                .getIndexResponses().get("my_index").getAction()));
+                .explainLifecycle(new ExplainLifecycleRequest("my_index-1"), RequestOptions.DEFAULT)
+                .getIndexResponses().get("my_index-1").getAction()));
         }
 
         // tag::ilm-explain-lifecycle-request
         ExplainLifecycleRequest request =
-            new ExplainLifecycleRequest("my_index", "other_index"); // <1>
+            new ExplainLifecycleRequest("my_index-1", "other_index"); // <1>
         // end::ilm-explain-lifecycle-request
 
-        // tag::ilm-explain-lifecycle-execute
-        ExplainLifecycleResponse response = client.indexLifecycle()
-            .explainLifecycle(request, RequestOptions.DEFAULT);
-        // end::ilm-explain-lifecycle-execute
-        assertNotNull(response);
-
-        // tag::ilm-explain-lifecycle-response
-        Map<String, IndexLifecycleExplainResponse> indices =
-            response.getIndexResponses();
-        IndexLifecycleExplainResponse myIndex = indices.get("my_index");
-        String policyName = myIndex.getPolicyName(); // <1>
-        boolean isManaged = myIndex.managedByILM(); // <2>
-
-        String phase = myIndex.getPhase(); // <3>
-        long phaseTime = myIndex.getPhaseTime(); // <4>
-        String action = myIndex.getAction(); // <5>
-        long actionTime = myIndex.getActionTime();
-        String step = myIndex.getStep(); // <6>
-        long stepTime = myIndex.getStepTime();
-
-        String failedStep = myIndex.getFailedStep(); // <7>
-        // end::ilm-explain-lifecycle-response
-        assertEquals("my_policy", policyName);
-        assertTrue(isManaged);
-
-        assertEquals("hot", phase);
-        assertNotEquals(0, phaseTime);
-        assertEquals("rollover", action);
-        assertNotEquals(0, actionTime);
-        assertEquals("check-rollover-ready", step);
-        assertNotEquals(0, stepTime);
-
-        assertNull(failedStep);
-
-        IndexLifecycleExplainResponse otherIndex = indices.get("other_index");
-        assertFalse(otherIndex.managedByILM());
-        assertNull(otherIndex.getPolicyName());
-        assertNull(otherIndex.getPhase());
-        assertNull(otherIndex.getAction());
-        assertNull(otherIndex.getStep());
-        assertNull(otherIndex.getFailedStep());
-        assertNull(otherIndex.getPhaseExecutionInfo());
-        assertNull(otherIndex.getStepInfo());
+
+        assertBusy(() -> {
+            // tag::ilm-explain-lifecycle-execute
+            ExplainLifecycleResponse response = client.indexLifecycle()
+                .explainLifecycle(request, RequestOptions.DEFAULT);
+            // end::ilm-explain-lifecycle-execute
+            assertNotNull(response);
+
+            // tag::ilm-explain-lifecycle-response
+            Map<String, IndexLifecycleExplainResponse> indices =
+                response.getIndexResponses();
+            IndexLifecycleExplainResponse myIndex = indices.get("my_index-1");
+            String policyName = myIndex.getPolicyName(); // <1>
+            boolean isManaged = myIndex.managedByILM(); // <2>
+
+            String phase = myIndex.getPhase(); // <3>
+            long phaseTime = myIndex.getPhaseTime(); // <4>
+            String action = myIndex.getAction(); // <5>
+            long actionTime = myIndex.getActionTime();
+            String step = myIndex.getStep(); // <6>
+            long stepTime = myIndex.getStepTime();
+
+            String failedStep = myIndex.getFailedStep(); // <7>
+            // end::ilm-explain-lifecycle-response
+
+            assertEquals("my_policy", policyName);
+            assertTrue(isManaged);
+
+            assertEquals("hot", phase);
+            assertNotEquals(0, phaseTime);
+            assertEquals("rollover", action);
+            assertNotEquals(0, actionTime);
+            assertEquals("check-rollover-ready", step);
+            assertNotEquals(0, stepTime);
+
+            assertNull(failedStep);
+
+            IndexLifecycleExplainResponse otherIndex = indices.get("other_index");
+            assertFalse(otherIndex.managedByILM());
+            assertNull(otherIndex.getPolicyName());
+            assertNull(otherIndex.getPhase());
+            assertNull(otherIndex.getAction());
+            assertNull(otherIndex.getStep());
+            assertNull(otherIndex.getFailedStep());
+            assertNull(otherIndex.getPhaseExecutionInfo());
+            assertNull(otherIndex.getStepInfo());
+        });
 
         // tag::ilm-explain-lifecycle-execute-listener
         ActionListener<ExplainLifecycleResponse> listener =

docs/reference/ilm/policy-definitions.asciidoc

@@ -353,6 +353,13 @@ index format must match pattern '^.*-\\d+$', for example (`logs-000001`).
 The managed index must set `index.lifecycle.rollover_alias` as the
 alias to rollover. The index must also be the write index for the alias.
 
+[IMPORTANT]
+If a policy using the Rollover action is used on a <<ccr-put-follow,follower
+index>>, policy execution will wait until the leader index rolls over (or has
+<<skipping-rollover, otherwise been marked as complete>>), then convert the
+follower index into a regular index as if <<ilm-unfollow-action,the Unfollow
+action>> had been used instead of rolling over.
+
 For example, if an index to be managed has an alias `my_data`. The managed
 index "my_index" must be the write index for the alias. For more information, read
 <<indices-rollover-is-write-index,Write Index Alias Behavior>>.
@@ -578,6 +585,13 @@ PUT _ilm/policy/my_policy
 
 NOTE: Index will be be made read-only when this action is run
 (see: <<dynamic-index-settings,index.blocks.write>>)
+[IMPORTANT]
+If a policy using the Shrink action is used on a <<ccr-put-follow,follower
+index>>, policy execution will wait until the leader index rolls over (or has
+<<skipping-rollover, otherwise been marked as complete>>), then convert the
+follower index into a regular index as if <<ilm-unfollow-action,the Unfollow
+action>> had been used before shrink is applied, as shrink cannot be safely
+applied to follower indices.
 
 This action shrinks an existing index into a new index with fewer primary
 shards. It calls the <<indices-shrink-index,Shrink API>> to shrink the index.
@@ -622,11 +636,27 @@ PUT _ilm/policy/my_policy
 [[ilm-unfollow-action]]
 ==== Unfollow
 
+[IMPORTANT]
+This action may be used explicitly, as shown below, but this action is also run
+before <<ilm-rollover-action,the Rollover action>> and <<ilm-shrink-action,the
+Shrink action>> as described in the documentation for those actions.
+
 This action turns a {ref}/ccr-apis.html[ccr] follower index
 into a regular index. This can be desired when moving follower
 indices into the next phase. Also certain actions like shrink
 and rollover can then be performed safely on follower indices.
 
+This action will wait until is it safe to convert a follower index into a
+regular index. In particular, the following conditions must be met:
+
+* The leader index must have `index.lifecycle.indexing_complete` set to `true`.
+This happens automatically if the leader index is rolled over using
+<<ilm-rollover-action,the Rollover action>>, or may be set manually using
+the <<indices-update-settings,Index Settings API>>.
+* All operations performed on the leader index must have been replicated to the
+follower index. This ensures that no operations will be lost when the index is
+converted into a regular index.
+
 If the unfollow action encounters a follower index then
 the following operations will be performed on it:
 

docs/reference/ilm/using-policies-rollover.asciidoc

@@ -123,6 +123,7 @@ When the rollover is performed, the newly-created index is set as the write
 index for the rolled over alias. Documents sent to the alias are indexed into
 the new index, enabling indexing to continue uninterrupted.
 
+[[skipping-rollover]]
 === Skipping Rollover
 
 The `index.lifecycle.indexing_complete` setting indicates to {ilm} whether this

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

@@ -25,7 +25,6 @@
  */
 public class ReadOnlyAction implements LifecycleAction {
     public static final String NAME = "readonly";
-    public static final ReadOnlyAction INSTANCE = new ReadOnlyAction();
 
     private static final ObjectParser<ReadOnlyAction, Void> PARSER = new ObjectParser<>(NAME, false, ReadOnlyAction::new);
 

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

@@ -6,14 +6,12 @@
 package org.elasticsearch.xpack.core.indexlifecycle;
 
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.set.Sets;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -44,8 +42,6 @@ public class TimeseriesLifecycleType implements LifecycleType {
     static final Set<String> VALID_WARM_ACTIONS = Sets.newHashSet(ORDERED_VALID_WARM_ACTIONS);
     static final Set<String> VALID_COLD_ACTIONS = Sets.newHashSet(ORDERED_VALID_COLD_ACTIONS);
     static final Set<String> VALID_DELETE_ACTIONS = Sets.newHashSet(ORDERED_VALID_DELETE_ACTIONS);
-    private static final Phase EMPTY_WARM_PHASE = new Phase("warm", TimeValue.ZERO,
-        Collections.singletonMap("readonly", ReadOnlyAction.INSTANCE));
     private static Map<String, Set<String>> ALLOWED_ACTIONS = new HashMap<>();
 
     static {
@@ -72,6 +68,13 @@ public List<Phase> getOrderedPhases(Map<String, Phase> phases) {
         for (String phaseName : VALID_PHASES) {
             Phase phase = phases.get(phaseName);
             if (phase != null) {
+                Map<String, LifecycleAction> actions = phase.getActions();
+                if (actions.containsKey(UnfollowAction.NAME) == false
+                    && (actions.containsKey(RolloverAction.NAME) || actions.containsKey(ShrinkAction.NAME))) {
+                    Map<String, LifecycleAction> actionMap = new HashMap<>(phase.getActions());
+                    actionMap.put(UnfollowAction.NAME, new UnfollowAction());
+                    phase = new Phase(phase.getName(), phase.getMinimumAge(), actionMap);
+                }
                 orderedPhases.add(phase);
             }
         }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/indexlifecycle/TimeseriesLifecycleTypeTests.java

@@ -152,6 +152,32 @@ public void testGetOrderedPhases() {
         assertTrue(isSorted(TimeseriesLifecycleType.INSTANCE.getOrderedPhases(phaseMap), Phase::getName, VALID_PHASES));
     }
 
+    public void testUnfollowInjections() {
+        assertTrue(isUnfollowInjected("hot", RolloverAction.NAME));
+        assertTrue(isUnfollowInjected("warm", ShrinkAction.NAME));
+
+        assertFalse(isUnfollowInjected("hot", SetPriorityAction.NAME));
+        assertFalse(isUnfollowInjected("warm", SetPriorityAction.NAME));
+        assertFalse(isUnfollowInjected("warm", AllocateAction.NAME));
+        assertFalse(isUnfollowInjected("warm", ReadOnlyAction.NAME));
+        assertFalse(isUnfollowInjected("warm", ForceMergeAction.NAME));
+        assertFalse(isUnfollowInjected("cold", SetPriorityAction.NAME));
+        assertFalse(isUnfollowInjected("cold", AllocateAction.NAME));
+        assertFalse(isUnfollowInjected("cold", FreezeAction.NAME));
+        assertFalse(isUnfollowInjected("delete", DeleteAction.NAME));
+
+    }
+
+    private boolean isUnfollowInjected(String phaseName, String actionName) {
+        Map<String, Phase> phaseMap = new HashMap<>();
+        Map<String, LifecycleAction> actionsMap = new HashMap<>();
+        actionsMap.put(actionName, getTestAction(actionName));
+        Phase warmPhase = new Phase(phaseName, TimeValue.ZERO, actionsMap);
+        phaseMap.put(phaseName, warmPhase);
+        List<Phase> phases = TimeseriesLifecycleType.INSTANCE.getOrderedPhases(phaseMap);
+        Phase processedWarmPhase = phases.stream().filter(phase -> phase.getName().equals(phaseName)).findFirst().get();
+        return processedWarmPhase.getActions().containsKey("unfollow");
+    }
 
     public void testGetOrderedActionsInvalidPhase() {
         IllegalArgumentException exception = expectThrows(IllegalArgumentException.class, () -> TimeseriesLifecycleType.INSTANCE