Allow forcemerge in the hot phase for ILM policies (#52073)

* Allow forcemerge in the hot phase for ILM policies

This commit changes the `forcemerge` action to also be allowed in the `hot` phase for policies. The
forcemerge will occur after a rollover, and allows users to take advantage of higher disk speeds for
performing the force merge (on a separate node type, for example).

On caveat with this is that a `forcemerge` in the `hot` phase *MUST* be accompanied by a `rollover`
action. ILM validates policies to ensure this is the case.

Resolves #43165

* Use anyMatch instead of findAny in validation

* Make randomTimeseriesLifecyclePolicy single-pass

Author: dakrone

docs/reference/ilm/policy-definitions.asciidoc

@@ -292,11 +292,15 @@ PUT _ilm/policy/my_policy
 [[ilm-forcemerge-action]]
 ==== Force Merge
 
-Phases allowed: warm.
+Phases allowed: hot, warm.
 
 NOTE: Index will be be made read-only when this action is run
 (see: <<dynamic-index-settings,index.blocks.write>>)
 
+NOTE: If the `forcemerge` action is used in the `hot` phase, the `rollover` action *must* be preset.
+ILM validates this predicate and will refuse a policy with a forcemerge in the hot phase without a
+rollover action.
+
 The Force Merge Action <<indices-forcemerge,force merges>> the index into at
 most a specific number of <<indices-segments,segments>>.
 

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

@@ -31,8 +31,13 @@ public class TimeseriesLifecycleType implements LifecycleType {
     public static final TimeseriesLifecycleType INSTANCE = new TimeseriesLifecycleType();
 
     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 = Arrays.asList(SetPriorityAction.NAME, UnfollowAction.NAME, RolloverAction.NAME);
+    static final String HOT_PHASE = "hot";
+    static final String WARM_PHASE = "warm";
+    static final String COLD_PHASE = "cold";
+    static final String DELETE_PHASE = "delete";
+    static final List<String> VALID_PHASES = Arrays.asList(HOT_PHASE, WARM_PHASE, COLD_PHASE, DELETE_PHASE);
+    static final List<String> ORDERED_VALID_HOT_ACTIONS = Arrays.asList(SetPriorityAction.NAME, UnfollowAction.NAME, RolloverAction.NAME,
+        ForceMergeAction.NAME);
     static final List<String> ORDERED_VALID_WARM_ACTIONS = Arrays.asList(SetPriorityAction.NAME, UnfollowAction.NAME, ReadOnlyAction.NAME,
         AllocateAction.NAME, ShrinkAction.NAME, ForceMergeAction.NAME);
     static final List<String> ORDERED_VALID_COLD_ACTIONS = Arrays.asList(SetPriorityAction.NAME, UnfollowAction.NAME, AllocateAction.NAME,
@@ -45,10 +50,10 @@ public class TimeseriesLifecycleType implements LifecycleType {
     private static Map<String, Set<String>> ALLOWED_ACTIONS = new HashMap<>();
 
     static {
-        ALLOWED_ACTIONS.put("hot", VALID_HOT_ACTIONS);
-        ALLOWED_ACTIONS.put("warm", VALID_WARM_ACTIONS);
-        ALLOWED_ACTIONS.put("cold", VALID_COLD_ACTIONS);
-        ALLOWED_ACTIONS.put("delete", VALID_DELETE_ACTIONS);
+        ALLOWED_ACTIONS.put(HOT_PHASE, VALID_HOT_ACTIONS);
+        ALLOWED_ACTIONS.put(WARM_PHASE, VALID_WARM_ACTIONS);
+        ALLOWED_ACTIONS.put(COLD_PHASE, VALID_COLD_ACTIONS);
+        ALLOWED_ACTIONS.put(DELETE_PHASE, VALID_DELETE_ACTIONS);
     }
 
     private TimeseriesLifecycleType() {
@@ -126,16 +131,16 @@ public String getPreviousPhaseName(String currentPhaseName, Map<String, Phase> p
     public List<LifecycleAction> getOrderedActions(Phase phase) {
         Map<String, LifecycleAction> actions = phase.getActions();
         switch (phase.getName()) {
-            case "hot":
+            case HOT_PHASE:
                 return ORDERED_VALID_HOT_ACTIONS.stream().map(a -> actions.getOrDefault(a, null))
                     .filter(Objects::nonNull).collect(Collectors.toList());
-            case "warm":
+            case WARM_PHASE:
                 return ORDERED_VALID_WARM_ACTIONS.stream() .map(a -> actions.getOrDefault(a, null))
                     .filter(Objects::nonNull).collect(Collectors.toList());
-            case "cold":
+            case COLD_PHASE:
                 return ORDERED_VALID_COLD_ACTIONS.stream().map(a -> actions.getOrDefault(a, null))
                     .filter(Objects::nonNull).collect(Collectors.toList());
-            case "delete":
+            case DELETE_PHASE:
                 return ORDERED_VALID_DELETE_ACTIONS.stream().map(a -> actions.getOrDefault(a, null))
                     .filter(Objects::nonNull).collect(Collectors.toList());
             default:
@@ -147,20 +152,20 @@ public List<LifecycleAction> getOrderedActions(Phase phase) {
     public String getNextActionName(String currentActionName, Phase phase) {
         List<String> orderedActionNames;
         switch (phase.getName()) {
-        case "hot":
+        case HOT_PHASE:
             orderedActionNames = ORDERED_VALID_HOT_ACTIONS;
             break;
-        case "warm":
+        case WARM_PHASE:
             orderedActionNames = ORDERED_VALID_WARM_ACTIONS;
             break;
-        case "cold":
+        case COLD_PHASE:
             orderedActionNames = ORDERED_VALID_COLD_ACTIONS;
             break;
-        case "delete":
+        case DELETE_PHASE:
             orderedActionNames = ORDERED_VALID_DELETE_ACTIONS;
             break;
         default:
-            throw new IllegalArgumentException("lifecycle type[" + TYPE + "] does not support phase[" + phase.getName() + "]");
+            throw new IllegalArgumentException("lifecycle type [" + TYPE + "] does not support phase [" + phase.getName() + "]");
         }
 
         int index = orderedActionNames.indexOf(currentActionName);
@@ -195,5 +200,19 @@ public void validate(Collection<Phase> phases) {
                 }
             });
         });
+
+        // Check for forcemerge in 'hot' without a rollover action
+        if (phases.stream()
+            // Is there a hot phase
+            .filter(phase -> HOT_PHASE.equals(phase.getName()))
+            // That contains the 'forcemerge' action
+            .filter(phase -> phase.getActions().containsKey(ForceMergeAction.NAME))
+            // But does *not* contain the 'rollover' action?
+            .anyMatch(phase -> phase.getActions().containsKey(RolloverAction.NAME) == false)) {
+            // If there is, throw an exception
+            throw new IllegalArgumentException("the [" + ForceMergeAction.NAME +
+                "] action may not be used in the [" + HOT_PHASE +
+                "] phase without an accompanying [" + RolloverAction.NAME + "] action");
+        }
     }
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/LifecyclePolicyTests.java

@@ -190,6 +190,12 @@ public static LifecyclePolicy randomTimeseriesLifecyclePolicy(@Nullable String l
             TimeValue after = TimeValue.parseTimeValue(randomTimeValue(0, 1000000000, "s", "m", "h", "d"), "test_after");
             Map<String, LifecycleAction> actions = new HashMap<>();
             List<String> actionNames = randomSubsetOf(validActions.apply(phase));
+
+            // If the hot phase contains a forcemerge, also make sure to add a rollover, or else the policy will not validate
+            if (phase.equals(TimeseriesLifecycleType.HOT_PHASE) && actionNames.contains(ForceMergeAction.NAME)) {
+                actionNames.add(RolloverAction.NAME);
+            }
+
             for (String action : actionNames) {
                 actions.put(action, randomAction.apply(action));
             }

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

@@ -10,11 +10,13 @@
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -27,6 +29,7 @@
 import static org.elasticsearch.xpack.core.ilm.TimeseriesLifecycleType.VALID_HOT_ACTIONS;
 import static org.elasticsearch.xpack.core.ilm.TimeseriesLifecycleType.VALID_PHASES;
 import static org.elasticsearch.xpack.core.ilm.TimeseriesLifecycleType.VALID_WARM_ACTIONS;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
 public class TimeseriesLifecycleTypeTests extends ESTestCase {
@@ -64,7 +67,7 @@ public void testValidateHotPhase() {
         Map<String, LifecycleAction> actions = VALID_HOT_ACTIONS
             .stream().map(this::getTestAction).collect(Collectors.toMap(LifecycleAction::getWriteableName, Function.identity()));
         if (randomBoolean()) {
-            invalidAction = getTestAction(randomFrom("allocate", "forcemerge", "delete", "shrink", "freeze"));
+            invalidAction = getTestAction(randomFrom("allocate", "delete", "shrink", "freeze"));
             actions.put(invalidAction.getWriteableName(), invalidAction);
         }
         Map<String, Phase> hotPhase = Collections.singletonMap("hot",
@@ -78,6 +81,22 @@ public void testValidateHotPhase() {
         } else {
             TimeseriesLifecycleType.INSTANCE.validate(hotPhase.values());
         }
+
+        {
+            final Consumer<Collection<String>> validateHotActions = hotActions -> {
+                final Map<String, LifecycleAction> hotActionMap = hotActions.stream()
+                    .map(this::getTestAction)
+                    .collect(Collectors.toMap(LifecycleAction::getWriteableName, Function.identity()));
+                TimeseriesLifecycleType.INSTANCE.validate(Collections.singleton(new Phase("hot", TimeValue.ZERO, hotActionMap)));
+            };
+
+            validateHotActions.accept(Arrays.asList(RolloverAction.NAME));
+            validateHotActions.accept(Arrays.asList(RolloverAction.NAME, ForceMergeAction.NAME));
+            IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
+                () -> validateHotActions.accept(Arrays.asList(ForceMergeAction.NAME)));
+            assertThat(e.getMessage(),
+                containsString("the [forcemerge] action may not be used in the [hot] phase without an accompanying [rollover] action"));
+        }
     }
 
     public void testValidateWarmPhase() {
@@ -340,11 +359,12 @@ public void testGetNextActionName() {
 
         assertNextActionName("hot", RolloverAction.NAME, null, new String[] {});
         assertNextActionName("hot", RolloverAction.NAME, null, new String[] { RolloverAction.NAME });
+        assertNextActionName("hot", RolloverAction.NAME, ForceMergeAction.NAME, ForceMergeAction.NAME, RolloverAction.NAME);
+        assertNextActionName("hot", ForceMergeAction.NAME, null, RolloverAction.NAME, ForceMergeAction.NAME);
 
         assertInvalidAction("hot", "foo", new String[] { RolloverAction.NAME });
         assertInvalidAction("hot", AllocateAction.NAME, new String[] { RolloverAction.NAME });
         assertInvalidAction("hot", DeleteAction.NAME, new String[] { RolloverAction.NAME });
-        assertInvalidAction("hot", ForceMergeAction.NAME, new String[] { RolloverAction.NAME });
         assertInvalidAction("hot", ReadOnlyAction.NAME, new String[] { RolloverAction.NAME });
         assertInvalidAction("hot", ShrinkAction.NAME, new String[] { RolloverAction.NAME });
 
@@ -465,7 +485,7 @@ public void testGetNextActionName() {
         Phase phase = new Phase("foo", TimeValue.ZERO, Collections.emptyMap());
         IllegalArgumentException exception = expectThrows(IllegalArgumentException.class,
                 () -> TimeseriesLifecycleType.INSTANCE.getNextActionName(ShrinkAction.NAME, phase));
-        assertEquals("lifecycle type[" + TimeseriesLifecycleType.TYPE + "] does not support phase[" + phase.getName() + "]",
+        assertEquals("lifecycle type [" + TimeseriesLifecycleType.TYPE + "] does not support phase [" + phase.getName() + "]",
                 exception.getMessage());
     }
 

x-pack/plugin/ilm/qa/rest/src/test/resources/rest-api-spec/test/ilm/10_basic.yml

@@ -216,3 +216,25 @@ setup:
       catch: missing
       ilm.get_lifecycle:
         policy: "my_timeseries_lifecycle"
+
+---
+"Test Invalid Policy":
+  - do:
+      catch: bad_request
+      ilm.put_lifecycle:
+        policy: "my_invalid_lifecycle"
+        body: |
+           {
+             "policy": {
+               "phases": {
+                 "hot": {
+                   "min_age": "0s",
+                   "actions": {
+                     "forcemerge": {
+                       "max_num_segments": 1
+                     }
+                   }
+                 }
+               }
+             }
+           }