elastic · andreidan · Oct 3, 2023 · Oct 3, 2023
diff --git a/docs/changelog/100179.yaml b/docs/changelog/100179.yaml
@@ -0,0 +1,6 @@
+pr: 100179
+summary: ILM introduce the `check-ts-end-time-passed` step
+area: ILM+SLM
+type: bug
+issues:
+ - 99696
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleAction.java
@@ -27,6 +27,7 @@
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
@@ -133,6 +134,7 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
         StepKey timeSeriesIndexCheckBranchKey = new StepKey(phase, NAME, CONDITIONAL_TIME_SERIES_CHECK_KEY);
         StepKey checkNotWriteIndex = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME);
         StepKey waitForNoFollowerStepKey = new StepKey(phase, NAME, WaitForNoFollowersStep.NAME);
+        StepKey waitTimeSeriesEndTimePassesKey = new StepKey(phase, NAME, WaitUntilTimeSeriesEndTimePassesStep.NAME);
         StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyStep.NAME);
         StepKey cleanupDownsampleIndexKey = new StepKey(phase, NAME, CleanupTargetIndexStep.NAME);
         StepKey generateDownsampleIndexNameKey = new StepKey(phase, NAME, DownsamplePrepareLifeCycleStateStep.NAME);
@@ -161,8 +163,18 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
             checkNotWriteIndex,
             waitForNoFollowerStepKey
         );
-        WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(waitForNoFollowerStepKey, readOnlyKey, client);
+        WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(
+            waitForNoFollowerStepKey,
+            waitTimeSeriesEndTimePassesKey,
+            client
+        );
 
+        WaitUntilTimeSeriesEndTimePassesStep waitUntilTimeSeriesEndTimeStep = new WaitUntilTimeSeriesEndTimePassesStep(
+            waitTimeSeriesEndTimePassesKey,
+            readOnlyKey,
+            Instant::now,
+            client
+        );
         // Mark source index as read-only
         ReadOnlyStep readOnlyStep = new ReadOnlyStep(readOnlyKey, generateDownsampleIndexNameKey, client);
 
@@ -245,6 +257,7 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
             isTimeSeriesIndexBranchingStep,
             checkNotWriteIndexStep,
             waitForNoFollowersStep,
+            waitUntilTimeSeriesEndTimeStep,
             readOnlyStep,
             cleanupDownsampleIndexStep,
             generateDownsampleIndexNameStep,

diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/ForceMergeAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/ForceMergeAction.java
@@ -25,6 +25,7 @@
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
@@ -119,6 +120,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
 
         StepKey preForceMergeBranchingKey = new StepKey(phase, NAME, CONDITIONAL_SKIP_FORCE_MERGE_STEP);
         StepKey checkNotWriteIndexKey = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME);
+        StepKey waitTimeSeriesEndTimePassesKey = new StepKey(phase, NAME, WaitUntilTimeSeriesEndTimePassesStep.NAME);
         StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyAction.NAME);
 
         StepKey closeKey = new StepKey(phase, NAME, CloseIndexStep.NAME);
@@ -154,7 +156,14 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
         // Indices in this step key can skip the no-op step and jump directly to the step with closeKey/forcemergeKey key
         CheckNotDataStreamWriteIndexStep checkNotWriteIndexStep = new CheckNotDataStreamWriteIndexStep(
             checkNotWriteIndexKey,
-            codecChange ? closeKey : forceMergeKey
+            waitTimeSeriesEndTimePassesKey
+        );
+
+        WaitUntilTimeSeriesEndTimePassesStep waitUntilTimeSeriesEndTimeStep = new WaitUntilTimeSeriesEndTimePassesStep(
+            waitTimeSeriesEndTimePassesKey,
+            codecChange ? closeKey : forceMergeKey,
+            Instant::now,
+            client
         );
 
         // Indices already in this step key when upgrading need to know how to move forward but stop making the index
@@ -182,6 +191,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
         List<Step> mergeSteps = new ArrayList<>();
         mergeSteps.add(conditionalSkipShrinkStep);
         mergeSteps.add(checkNotWriteIndexStep);
+        mergeSteps.add(waitUntilTimeSeriesEndTimeStep);
         mergeSteps.add(noopStep);
 
         if (codecChange) {

diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/ReadOnlyAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/ReadOnlyAction.java
@@ -16,6 +16,7 @@
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.Arrays;
 import java.util.List;
 
@@ -58,10 +59,20 @@ public boolean isSafeAction() {
     @Override
     public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
         StepKey checkNotWriteIndex = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME);
+        StepKey waitTimeSeriesEndTimePassesKey = new StepKey(phase, NAME, WaitUntilTimeSeriesEndTimePassesStep.NAME);
         StepKey readOnlyKey = new StepKey(phase, NAME, NAME);
-        CheckNotDataStreamWriteIndexStep checkNotWriteIndexStep = new CheckNotDataStreamWriteIndexStep(checkNotWriteIndex, readOnlyKey);
+        CheckNotDataStreamWriteIndexStep checkNotWriteIndexStep = new CheckNotDataStreamWriteIndexStep(
+            checkNotWriteIndex,
+            waitTimeSeriesEndTimePassesKey
+        );
+        WaitUntilTimeSeriesEndTimePassesStep waitUntilTimeSeriesEndTimeStep = new WaitUntilTimeSeriesEndTimePassesStep(
+            waitTimeSeriesEndTimePassesKey,
+            readOnlyKey,
+            Instant::now,
+            client
+        );
         ReadOnlyStep readOnlyStep = new ReadOnlyStep(readOnlyKey, nextStepKey, client);
-        return Arrays.asList(checkNotWriteIndexStep, readOnlyStep);
+        return Arrays.asList(checkNotWriteIndexStep, waitUntilTimeSeriesEndTimeStep, readOnlyStep);
     }
 
     @Override

diff --git a/.../plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/SearchableSnapshotAction.java b/.../plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/SearchableSnapshotAction.java
@@ -26,6 +26,7 @@
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
@@ -106,6 +107,7 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey, XPac
         StepKey preActionBranchingKey = new StepKey(phase, NAME, CONDITIONAL_SKIP_ACTION_STEP);
         StepKey checkNoWriteIndex = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME);
         StepKey waitForNoFollowerStepKey = new StepKey(phase, NAME, WaitForNoFollowersStep.NAME);
+        StepKey waitTimeSeriesEndTimePassesKey = new StepKey(phase, NAME, WaitUntilTimeSeriesEndTimePassesStep.NAME);
         StepKey forceMergeStepKey = new StepKey(phase, NAME, ForceMergeStep.NAME);
         StepKey waitForSegmentCountKey = new StepKey(phase, NAME, SegmentCountStep.NAME);
         StepKey skipGeneratingSnapshotKey = new StepKey(phase, NAME, CONDITIONAL_SKIP_GENERATE_AND_CLEAN);
@@ -200,7 +202,13 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey, XPac
         );
         WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(
             waitForNoFollowerStepKey,
+            waitTimeSeriesEndTimePassesKey,
+            client
+        );
+        WaitUntilTimeSeriesEndTimePassesStep waitUntilTimeSeriesEndTimeStep = new WaitUntilTimeSeriesEndTimePassesStep(
+            waitTimeSeriesEndTimePassesKey,
             skipGeneratingSnapshotKey,
+            Instant::now,
             client
         );
 
@@ -321,6 +329,7 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey, XPac
         steps.add(conditionalSkipActionStep);
         steps.add(checkNoWriteIndexStep);
         steps.add(waitForNoFollowersStep);
+        steps.add(waitUntilTimeSeriesEndTimeStep);
         steps.add(skipGeneratingSnapshotStep);
         if (forceMergeIndex) {
             steps.add(forceMergeStep);

diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/ShrinkAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/ShrinkAction.java
@@ -25,6 +25,7 @@
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
@@ -142,6 +143,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
         StepKey preShrinkBranchingKey = new StepKey(phase, NAME, CONDITIONAL_SKIP_SHRINK_STEP);
         StepKey checkNotWriteIndex = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME);
         StepKey waitForNoFollowerStepKey = new StepKey(phase, NAME, WaitForNoFollowersStep.NAME);
+        StepKey waitTimeSeriesEndTimePassesKey = new StepKey(phase, NAME, WaitUntilTimeSeriesEndTimePassesStep.NAME);
         StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyAction.NAME);
         StepKey checkTargetShardsCountKey = new StepKey(phase, NAME, CheckTargetShardsCountStep.NAME);
         StepKey cleanupShrinkIndexKey = new StepKey(phase, NAME, CleanupShrinkIndexStep.NAME);
@@ -197,7 +199,17 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
             checkNotWriteIndex,
             waitForNoFollowerStepKey
         );
-        WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(waitForNoFollowerStepKey, readOnlyKey, client);
+        WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(
+            waitForNoFollowerStepKey,
+            waitTimeSeriesEndTimePassesKey,
+            client
+        );
+        WaitUntilTimeSeriesEndTimePassesStep waitUntilTimeSeriesEndTimeStep = new WaitUntilTimeSeriesEndTimePassesStep(
+            waitTimeSeriesEndTimePassesKey,
+            readOnlyKey,
+            Instant::now,
+            client
+        );
         ReadOnlyStep readOnlyStep = new ReadOnlyStep(readOnlyKey, checkTargetShardsCountKey, client);
         CheckTargetShardsCountStep checkTargetShardsCountStep = new CheckTargetShardsCountStep(
             checkTargetShardsCountKey,
@@ -271,6 +283,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
             conditionalSkipShrinkStep,
             checkNotWriteIndexStep,
             waitForNoFollowersStep,
+            waitUntilTimeSeriesEndTimeStep,
             readOnlyStep,
             checkTargetShardsCountStep,
             cleanupShrinkIndexStep,

diff --git a/.../plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/WaitForRolloverReadyStep.java b/.../plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/WaitForRolloverReadyStep.java
@@ -23,8 +23,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
-import org.elasticsearch.xcontent.ToXContentObject;
-import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.core.ilm.step.info.EmptyInfo;
 
 import java.util.HashMap;
 import java.util.Locale;
@@ -272,17 +271,4 @@ public boolean equals(Object obj) {
         WaitForRolloverReadyStep other = (WaitForRolloverReadyStep) obj;
         return super.equals(obj) && Objects.equals(conditions, other.conditions);
     }
-
-    // We currently have no information to provide for this AsyncWaitStep, so this is an empty object
-    private static final class EmptyInfo implements ToXContentObject {
-
-        static final EmptyInfo INSTANCE = new EmptyInfo();
-
-        private EmptyInfo() {}
-
-        @Override
-        public XContentBuilder toXContent(XContentBuilder builder, Params params) {
-            return builder;
-        }
-    }
 }
diff --git a/.../src/main/java/org/elasticsearch/xpack/core/ilm/WaitUntilTimeSeriesEndTimePassesStep.java b/.../src/main/java/org/elasticsearch/xpack/core/ilm/WaitUntilTimeSeriesEndTimePassesStep.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+package org.elasticsearch.xpack.core.ilm;
+
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.xpack.core.ilm.step.info.EmptyInfo;
+import org.elasticsearch.xpack.core.ilm.step.info.SingleMessageFieldInfo;
+
+import java.time.Instant;
+import java.util.Locale;
+import java.util.function.Supplier;
+
+/**
+ * This {@link Step} waits until the {@link org.elasticsearch.index.IndexSettings#TIME_SERIES_END_TIME} passes for time series indices.
+ * For regular indices this step doesn't wait at all and the condition is evaluated to true immediately.
+ *
+ * Note that this step doens't execute an async/transport action and is able to evaluate its condition based on the local information
+ * available however, we want this step to be executed periodically using the `AsyncWaitStep` infrastructure.
+ * The condition will be evaluated every {@link LifecycleSettings#LIFECYCLE_POLL_INTERVAL}.
+ */
+public class WaitUntilTimeSeriesEndTimePassesStep extends AsyncWaitStep {
+
+    public static final String NAME = "check-ts-end-time-passed";
+    private final Supplier<Instant> nowSupplier;
+
+    public WaitUntilTimeSeriesEndTimePassesStep(StepKey key, StepKey nextStepKey, Supplier<Instant> nowSupplier, Client client) {
+        super(key, nextStepKey, client);
+        this.nowSupplier = nowSupplier;
+    }
+
+    @Override
+    public boolean isRetryable() {
+        return true;
+    }
+
+    @Override
+    public void evaluateCondition(Metadata metadata, Index index, Listener listener, TimeValue masterTimeout) {
+        IndexMetadata indexMetadata = metadata.index(index);
+        assert indexMetadata != null
+            : "the index metadata for index [" + index.getName() + "] must exist in the cluster state for step " + "[" + NAME + "]";
+
+        if (IndexSettings.MODE.get(indexMetadata.getSettings()) != IndexMode.TIME_SERIES) {
+            // this index is not a time series index so no need to wait
+            listener.onResponse(true, EmptyInfo.INSTANCE);
+            return;
+        }
+        Instant configuredEndTime = IndexSettings.TIME_SERIES_END_TIME.get(indexMetadata.getSettings());
+        assert configuredEndTime != null : "a time series index must have an end time configured but [" + index.getName() + "] does not";
+        if (nowSupplier.get().isBefore(configuredEndTime)) {
+            listener.onResponse(
+                false,
+                new SingleMessageFieldInfo(
+                    String.format(
+                        Locale.ROOT,
+                        "The [%s] setting for index [%s] is [%s]. Waiting until the index's time series end time lapses before"
+                            + " proceeding with action [%s] as the index can still accept writes.",
+                        IndexSettings.TIME_SERIES_END_TIME.getKey(),
+                        index.getName(),
+                        configuredEndTime.toEpochMilli(),
+                        getKey().action()
+                    )
+                )
+            );
+            return;
+        }
+
+        listener.onResponse(true, EmptyInfo.INSTANCE);
+    }
+}
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/step/info/EmptyInfo.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/step/info/EmptyInfo.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.ilm.step.info;
+
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+/**
+ * An empty XContent object to indicate an ILM step is not providing any information.
+ */
+public final class EmptyInfo implements ToXContentObject {
+
+    public static final EmptyInfo INSTANCE = new EmptyInfo();
+
+    private EmptyInfo() {}
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) {
+        return builder;
+    }
+}