WIP leapfrog ILM deletes

dakrone · Nov 22, 2024 · 9eddc49 · 9eddc49
1 parent 7e801e0
commit 9eddc49
Show file tree

Hide file tree

Showing 2 changed files with 64 additions and 5 deletions.
diff --git a/...k/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/TimeseriesLifecycleType.java b/...k/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/TimeseriesLifecycleType.java
@@ -45,11 +45,11 @@ public class TimeseriesLifecycleType implements LifecycleType {
 
     public static final String TYPE = "timeseries";
 
-    static final String HOT_PHASE = "hot";
-    static final String WARM_PHASE = "warm";
-    static final String COLD_PHASE = "cold";
-    static final String FROZEN_PHASE = "frozen";
-    static final String DELETE_PHASE = "delete";
+    public static final String HOT_PHASE = "hot";
+    public static final String WARM_PHASE = "warm";
+    public static final String COLD_PHASE = "cold";
+    public static final String FROZEN_PHASE = "frozen";
+    public static final String DELETE_PHASE = "delete";
     public static final List<String> ORDERED_VALID_PHASES = List.of(HOT_PHASE, WARM_PHASE, COLD_PHASE, FROZEN_PHASE, DELETE_PHASE);
 
     public static final List<String> ORDERED_VALID_HOT_ACTIONS = Stream.of(

diff --git a/x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycleRunner.java b/x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycleRunner.java
@@ -31,9 +31,11 @@
 import org.elasticsearch.xpack.core.ilm.ClusterStateWaitStep;
 import org.elasticsearch.xpack.core.ilm.ErrorStep;
 import org.elasticsearch.xpack.core.ilm.PhaseCompleteStep;
+import org.elasticsearch.xpack.core.ilm.RolloverAction;
 import org.elasticsearch.xpack.core.ilm.Step;
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 import org.elasticsearch.xpack.core.ilm.TerminalPolicyStep;
+import org.elasticsearch.xpack.core.ilm.TimeseriesLifecycleType;
 import org.elasticsearch.xpack.ilm.history.ILMHistoryItem;
 import org.elasticsearch.xpack.ilm.history.ILMHistoryStore;
 
@@ -170,6 +172,53 @@ boolean isReadyToTransitionToThisPhase(final String policy, final IndexMetadata
         return now >= lifecycleDate + after.getMillis();
     }
 
+    private static final Set<String> nonLeapFrogSteps = new HashSet<>(
+        TimeseriesLifecycleType.ORDERED_VALID_HOT_ACTIONS.subList(
+            0,
+            TimeseriesLifecycleType.ORDERED_VALID_HOT_ACTIONS.indexOf(RolloverAction.NAME) + 1
+        )
+    );
+
+    /**
+     * Return true or false depending on whether the index is ready to be in {@code phase}
+     */
+    boolean couldBeMovedToDeletePhase(final String policy, final IndexMetadata indexMetadata) {
+        final Long lifecycleDate = calculateOriginationMillis(indexMetadata);
+        if (lifecycleDate == null) {
+            return false;
+        }
+        final LifecycleExecutionState executionState = indexMetadata.getLifecycleExecutionState();
+        if (executionState.phase().equals(TimeseriesLifecycleType.HOT_PHASE) && nonLeapFrogSteps.contains(executionState.action()) == true) {
+            // Don't leap-frog these steps, because the index hasn't rolled over yet.
+            return false;
+        }
+        final TimeValue after = stepRegistry.getIndexAgeForPhase(policy, TimeseriesLifecycleType.DELETE_PHASE);
+        final long now = nowSupplier.getAsLong();
+        if (logger.isTraceEnabled()) {
+            final long ageMillis = now - lifecycleDate;
+            final TimeValue age;
+            if (ageMillis >= 0) {
+                age = new TimeValue(ageMillis);
+            } else if (ageMillis == Long.MIN_VALUE) {
+                age = new TimeValue(Long.MAX_VALUE);
+            } else {
+                age = new TimeValue(-ageMillis);
+            }
+            logger.trace(
+                "[{}] checking for index age to be at least [{}] before performing jumping to "
+                    + "the delete phase. Now: {}, lifecycle date: {}, age: [{}{}/{}s]",
+                indexMetadata.getIndex().getName(),
+                after,
+                new TimeValue(now).seconds(),
+                new TimeValue(lifecycleDate).seconds(),
+                ageMillis < 0 ? "-" : "",
+                age,
+                age.seconds()
+            );
+        }
+        return now >= lifecycleDate + after.getMillis();
+    }
+
     /**
      * Run the current step, only if it is an asynchronous wait step. These
      * wait criteria are checked periodically from the ILM scheduler
@@ -414,6 +463,16 @@ void runPolicyAfterStateChange(String policy, IndexMetadata indexMetadata) {
             }
         }
 
+        if (TimeseriesLifecycleType.DELETE_PHASE.equals(currentStep.getKey().phase()) == false && couldBeMovedToDeletePhase(policy, indexMetadata)) {
+            logger.info("--> {} could be deleted! moving to delete...", indexMetadata.getIndex().getName());
+            moveToStep(
+                indexMetadata.getIndex(),
+                policy,
+                currentStep.getKey(),
+                stepRegistry.getFirstStepForPhase(clusterService.state(), indexMetadata.getIndex(), TimeseriesLifecycleType.DELETE_PHASE)
+            );
+        }
+
         if (currentStep instanceof TerminalPolicyStep) {
             logger.debug("policy [{}] for index [{}] complete, skipping execution", policy, index);
             return;