Add support for steps to change the target index name for later steps

modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java

@@ -64,6 +64,7 @@
 import org.elasticsearch.datastreams.lifecycle.rest.RestGetDataStreamLifecycleAction;
 import org.elasticsearch.datastreams.lifecycle.rest.RestPutDataStreamLifecycleAction;
 import org.elasticsearch.datastreams.lifecycle.transitions.DlmAction;
+import org.elasticsearch.datastreams.lifecycle.transitions.DlmStep;
 import org.elasticsearch.datastreams.lifecycle.transitions.steps.ForceMergeStep;
 import org.elasticsearch.datastreams.lifecycle.transitions.steps.MarkIndexForDLMForceMergeAction;
 import org.elasticsearch.datastreams.lifecycle.transitions.steps.TransportMarkIndexForDLMForceMergeAction;
@@ -218,6 +219,8 @@ public Collection<?> createComponents(PluginServices services) {
         // Register DLM actions here. Order matters - they will be executed in the order they are listed for a given index.
         List<DlmAction> dlmActions = List.of();
 
+        verifyActions(dlmActions);
+
         dataLifecycleInitialisationService.set(
             new DataStreamLifecycleService(
                 settings,
@@ -242,6 +245,26 @@ public Collection<?> createComponents(PluginServices services) {
         return components;
     }
 
+    // visible for testing
+    static void verifyActions(List<DlmAction> dlmActions) {
+        for (DlmAction action : dlmActions) {
+            if (action.steps().isEmpty()) {
+                throw new IllegalStateException("DLM action [" + action.name() + "] must have at least one step");
+            }
+            for (DlmStep step : action.steps()) {
+                if (step.possibleOutputIndexNamePatterns("dummy-index").isEmpty()) {
+                    throw new IllegalStateException(
+                        "DLM step ["
+                            + step.stepName()
+                            + "] in action ["
+                            + action.name()
+                            + "] must have at least one possible output index name pattern"
+                    );
+                }
+            }
+        }
+    }
+
     @Override
     public List<ActionHandler> getActions() {
         List<ActionHandler> actions = new ArrayList<>();

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java

@@ -577,7 +577,7 @@ Set<Index> maybeProcessDlmActions(ProjectState projectState, DataStream dataStre
 
             for (Index index : indicesEligibleForAction) {
                 long findStepStartTime = nowSupplier.getAsLong();
-                DlmStep stepToExecute = findFirstIncompleteStep(projectState, dataStream, action, index);
+                int stepToExecuteIndex = findFirstIncompleteStepIndex(projectState, dataStream, action, index);
                 if (logger.isTraceEnabled()) {
                     long findStepDuration = nowSupplier.getAsLong() - findStepStartTime;
                     logger.trace(
@@ -589,7 +589,8 @@ Set<Index> maybeProcessDlmActions(ProjectState projectState, DataStream dataStre
                     );
                 }
 
-                if (stepToExecute != null) {
+                if (stepToExecuteIndex >= 0) {
+                    DlmStep stepToExecute = action.steps().get(stepToExecuteIndex);
                     try {
                         logger.trace(
                             "Executing step [{}] for action [{}] on datastream [{}] index [{}]",
@@ -599,7 +600,8 @@ Set<Index> maybeProcessDlmActions(ProjectState projectState, DataStream dataStre
                             index.getName()
                         );
                         long stepStartTime = nowSupplier.getAsLong();
-                        DlmStepContext dlmStepContext = actionContext.stepContextFor(index);
+                        Index indexForExecution = resolveIndexOutputFromPreviousStep(stepToExecuteIndex, index, action, projectState);
+                        DlmStepContext dlmStepContext = actionContext.stepContextFor(indexForExecution);
                         stepToExecute.execute(dlmStepContext);
                         if (logger.isTraceEnabled()) {
                             long stepDuration = nowSupplier.getAsLong() - stepStartTime;
@@ -642,20 +644,23 @@ Set<Index> maybeProcessDlmActions(ProjectState projectState, DataStream dataStre
         return indicesProcessed;
     }
 
-    private DlmStep findFirstIncompleteStep(ProjectState projectState, DataStream dataStream, DlmAction action, Index index) {
-        DlmStep stepToExecute = null;
-        for (DlmStep step : action.steps().reversed()) {
+    private int findFirstIncompleteStepIndex(ProjectState projectState, DataStream dataStream, DlmAction action, Index index) {
+        assert action.steps().size() >= 1 : "an action must have at least one step";
+        int stepToExecute = -1;
+        for (int i = action.steps().size() - 1; i >= 0; i--) {
+            DlmStep step = action.steps().get(i);
             try {
                 long checkStartTime = nowSupplier.getAsLong();
-                if (step.stepCompleted(index, projectState) == false) {
-                    stepToExecute = step;
+                Index indexInUse = resolveIndexOutputFromPreviousStep(i, index, action, projectState);
+                if (step.stepCompleted(indexInUse, projectState) == false) {
+                    stepToExecute = i;
                     if (logger.isTraceEnabled()) {
                         logger.trace(
                             "Step [{}] for action [{}] on datastream [{}] index [{}] is not complete, checked in [{}]",
                             step.stepName(),
                             action.name(),
                             dataStream.getName(),
-                            index.getName(),
+                            indexInUse.getName(),
                             formatExecutionTime(nowSupplier.getAsLong() - checkStartTime)
                         );
                     }
@@ -666,7 +671,7 @@ private DlmStep findFirstIncompleteStep(ProjectState projectState, DataStream da
                             step.stepName(),
                             action.name(),
                             dataStream.getName(),
-                            index.getName(),
+                            indexInUse.getName(),
                             formatExecutionTime(nowSupplier.getAsLong() - checkStartTime)
                         );
                     }
@@ -689,6 +694,33 @@ private DlmStep findFirstIncompleteStep(ProjectState projectState, DataStream da
         return stepToExecute;
     }
 
+    // visible for testing
+    Index resolveIndexOutputFromPreviousStep(int stepToExecuteIndex, Index index, DlmAction action, ProjectState projectState) {
+        if (stepToExecuteIndex < 1) {
+            return index;
+        }
+
+        DlmStep previousStep = action.steps().get(stepToExecuteIndex - 1);
+        List<String> possibleOutputIndexNamePatterns = previousStep.possibleOutputIndexNamePatterns(index.getName());
+
+        for (String candidateIndexName : possibleOutputIndexNamePatterns) {
+            if (projectState.metadata().hasIndex(candidateIndexName)) {
+                return projectState.metadata().index(candidateIndexName).getIndex();
+            }
+        }
+
+        assert false
+            : "Unable to resolve index name for executing step ["
+                + action.steps().get(stepToExecuteIndex).stepName()
+                + "] for action ["
+                + action.name()
+                + "] with index ["
+                + index.getName()
+                + "]";
+
+        return index;
+    }
+
     // visible for testing
     static Set<Index> timeSeriesIndicesStillWithinTimeBounds(ProjectMetadata project, List<Index> targetIndices, LongSupplier nowSupplier) {
         Set<Index> tsIndicesWithinBounds = new HashSet<>();

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java

@@ -12,6 +12,8 @@
 import org.elasticsearch.cluster.ProjectState;
 import org.elasticsearch.index.Index;
 
+import java.util.List;
+
 /**
  * A step within a Data Lifecycle Management action. Each step is responsible for determining if it has been completed for a given index
  * and executing the necessary operations to complete the step.
@@ -42,4 +44,18 @@ public interface DlmStep {
      */
     String stepName();
 
+    /**
+     * A list of functions that can be used to generate possible output index name patterns for this step based on the input index name.
+     * This is used when a step might change the index that is targeted for later steps in the action such as cloning.
+     * <br>
+     * The list is ordered and the first pattern that generates an index name that exists in the cluster will be used by later steps.
+     * <br>
+     * If not overridden, this method simply returns a list with the input index name, meaning that by default steps will target
+     * the same index as the input index.
+     * @param indexName The input index name to generate patterns for
+     * @return A list of possible output index name patterns for the given input index name.
+     */
+    default List<String> possibleOutputIndexNamePatterns(String indexName) {
+        return List.of(indexName);
+    }
 }

modules/data-streams/src/test/java/org/elasticsearch/datastreams/DataStreamsPluginTests.java

@@ -0,0 +1,163 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+package org.elasticsearch.datastreams;
+
+import org.elasticsearch.cluster.ProjectState;
+import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.datastreams.lifecycle.transitions.DlmAction;
+import org.elasticsearch.datastreams.lifecycle.transitions.DlmActionContext;
+import org.elasticsearch.datastreams.lifecycle.transitions.DlmStep;
+import org.elasticsearch.datastreams.lifecycle.transitions.DlmStepContext;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.List;
+import java.util.function.Function;
+
+import static org.hamcrest.Matchers.containsString;
+
+public class DataStreamsPluginTests extends ESTestCase {
+
+    public void testVerifyActionsWithEmptyList() {
+        DataStreamsPlugin.verifyActions(List.of());
+    }
+
+    public void testVerifyActionsWithDefaultOutputPatterns() {
+        TestDlmStep step = new TestDlmStep("step-1");
+        DlmAction action = new TestDlmAction("valid-action", List.of(step));
+        DataStreamsPlugin.verifyActions(List.of(action));
+    }
+
+    public void testVerifyActionsWithSingleCustomOutputPattern() {
+        TestDlmStep step = new TestDlmStep("step-1");
+        step.outputIndexNamePatternFunctions = List.of(name -> "cloned-" + name);
+        DlmAction action = new TestDlmAction("valid-action", List.of(step));
+        DataStreamsPlugin.verifyActions(List.of(action));
+    }
+
+    public void testVerifyActionsWithMultipleCustomOutputPatterns() {
+        TestDlmStep step = new TestDlmStep("step-1");
+        step.outputIndexNamePatternFunctions = List.of(name -> "partial-" + name, name -> "full-" + name);
+        DlmAction action = new TestDlmAction("valid-action", List.of(step));
+        DataStreamsPlugin.verifyActions(List.of(action));
+    }
+
+    public void testVerifyActionsWithMixOfDefaultAndCustomOutputPatterns() {
+        TestDlmStep defaultStep = new TestDlmStep("default-step");
+        TestDlmStep customStep = new TestDlmStep("custom-step");
+        customStep.outputIndexNamePatternFunctions = List.of(name -> "cloned-" + name);
+        DlmAction action = new TestDlmAction("valid-action", List.of(defaultStep, customStep));
+        DataStreamsPlugin.verifyActions(List.of(action));
+    }
+
+    public void testVerifyActionsWithMultipleValidActions() {
+        TestDlmStep step1 = new TestDlmStep("step-1");
+        step1.outputIndexNamePatternFunctions = List.of(name -> "cloned-" + name);
+        DlmAction action1 = new TestDlmAction("action-1", List.of(step1, new TestDlmStep("step-2")));
+        DlmAction action2 = new TestDlmAction("action-2", List.of(new TestDlmStep("step-3")));
+        DataStreamsPlugin.verifyActions(List.of(action1, action2));
+    }
+
+    public void testVerifyActionsThrowsOnActionWithNoSteps() {
+        DlmAction action = new TestDlmAction("empty-action", List.of());
+        IllegalStateException e = expectThrows(IllegalStateException.class, () -> DataStreamsPlugin.verifyActions(List.of(action)));
+        assertThat(e.getMessage(), containsString("DLM action [empty-action] must have at least one step"));
+    }
+
+    public void testVerifyActionsThrowsOnStepWithEmptyOutputPatterns() {
+        TestDlmStep step = new TestDlmStep("bad-step");
+        step.outputIndexNamePatternFunctions = List.of();
+        DlmAction action = new TestDlmAction("my-action", List.of(step));
+
+        IllegalStateException e = expectThrows(IllegalStateException.class, () -> DataStreamsPlugin.verifyActions(List.of(action)));
+        assertThat(e.getMessage(), containsString("DLM step [bad-step] in action [my-action]"));
+        assertThat(e.getMessage(), containsString("must have at least one possible output index name pattern"));
+    }
+
+    public void testVerifyActionsThrowsOnSecondActionWithNoSteps() {
+        DlmAction validAction = new TestDlmAction("valid-action", List.of(new TestDlmStep("step-1")));
+        DlmAction emptyAction = new TestDlmAction("broken-action", List.of());
+        IllegalStateException e = expectThrows(
+            IllegalStateException.class,
+            () -> DataStreamsPlugin.verifyActions(List.of(validAction, emptyAction))
+        );
+        assertThat(e.getMessage(), containsString("DLM action [broken-action] must have at least one step"));
+    }
+
+    public void testVerifyActionsThrowsOnSecondStepWithEmptyOutputPatterns() {
+        TestDlmStep goodStep = new TestDlmStep("good-step");
+        TestDlmStep badStep = new TestDlmStep("bad-step");
+        badStep.outputIndexNamePatternFunctions = List.of();
+        DlmAction action = new TestDlmAction("my-action", List.of(goodStep, badStep));
+
+        IllegalStateException e = expectThrows(IllegalStateException.class, () -> DataStreamsPlugin.verifyActions(List.of(action)));
+        assertThat(e.getMessage(), containsString("DLM step [bad-step] in action [my-action]"));
+    }
+
+    private static class TestDlmStep implements DlmStep {
+        private final String name;
+        List<Function<String, String>> outputIndexNamePatternFunctions = null;
+
+        TestDlmStep(String name) {
+            this.name = name;
+        }
+
+        @Override
+        public boolean stepCompleted(Index index, ProjectState projectState) {
+            return false;
+        }
+
+        @Override
+        public void execute(DlmStepContext dlmStepContext) {}
+
+        @Override
+        public String stepName() {
+            return name;
+        }
+
+        @Override
+        public List<String> possibleOutputIndexNamePatterns(String indexName) {
+            if (outputIndexNamePatternFunctions != null) {
+                return outputIndexNamePatternFunctions.stream().map(func -> func.apply(indexName)).toList();
+            }
+            return DlmStep.super.possibleOutputIndexNamePatterns(indexName);
+        }
+    }
+
+    private static class TestDlmAction implements DlmAction {
+        private final String name;
+        private final List<DlmStep> steps;
+
+        TestDlmAction(String name, List<DlmStep> steps) {
+            this.name = name;
+            this.steps = steps;
+        }
+
+        @Override
+        public String name() {
+            return name;
+        }
+
+        @Override
+        public Function<DataStreamLifecycle, TimeValue> applyAfterTime() {
+            return dsl -> null;
+        }
+
+        @Override
+        public List<DlmStep> steps() {
+            return steps;
+        }
+
+        @Override
+        public boolean canRunOnProject(DlmActionContext dlmActionContext) {
+            return true;
+        }
+    }
+}