From c7fec577b05b6cbd9cb811572cd989844b0a6d0f Mon Sep 17 00:00:00 2001 From: Luke Whiting Date: Tue, 24 Feb 2026 11:31:12 +0000 Subject: [PATCH 1/5] Add support for steps to change the target index name for later steps --- .../datastreams/DataStreamsPlugin.java | 23 +++ .../lifecycle/DataStreamLifecycleService.java | 44 ++++- .../lifecycle/transitions/DlmStep.java | 16 ++ .../datastreams/DataStreamsPluginTests.java | 163 ++++++++++++++++++ .../DataStreamLifecycleServiceTests.java | 151 ++++++++++++++++ 5 files changed, 390 insertions(+), 7 deletions(-) create mode 100644 modules/data-streams/src/test/java/org/elasticsearch/datastreams/DataStreamsPluginTests.java diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java index 7af612951a7f4..e41759ae1f733 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java +++ b/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.options.action.DeleteDataStreamOptionsAction; import org.elasticsearch.datastreams.options.action.GetDataStreamOptionsAction; import org.elasticsearch.datastreams.options.action.TransportDeleteDataStreamOptionsAction; @@ -214,6 +215,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 dlmActions = List.of(); + verifyActions(dlmActions); + dataLifecycleInitialisationService.set( new DataStreamLifecycleService( settings, @@ -238,6 +241,26 @@ public Collection createComponents(PluginServices services) { return components; } + // visible for testing + static void verifyActions(List 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().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 getActions() { List actions = new ArrayList<>(); diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java index 1ef274c72c4a2..137e8147631ed 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java @@ -568,7 +568,7 @@ Set 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( @@ -580,7 +580,8 @@ Set 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 [{}]", @@ -590,7 +591,8 @@ Set 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; @@ -633,13 +635,15 @@ Set 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) { + int stepToExecute = -1; + for (int i = action.steps().size() - 1; i >= 0; i--) { + DlmStep step = action.steps().get(i); try { long checkStartTime = nowSupplier.getAsLong(); + index = resolveIndexOutputFromPreviousStep(i, index, action, projectState); if (step.stepCompleted(index, projectState) == false) { - stepToExecute = step; + stepToExecute = i; if (logger.isTraceEnabled()) { logger.trace( "Step [{}] for action [{}] on datastream [{}] index [{}] is not complete, checked in [{}]", @@ -680,6 +684,32 @@ 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> possibleOutputIndexNamePatterns = previousStep.possibleOutputIndexNamePatterns(); + + for (Function possibleOutputIndexNamePattern : possibleOutputIndexNamePatterns) { + String candidateIndexName = possibleOutputIndexNamePattern.apply(index.getName()); + if (projectState.metadata().hasIndex(candidateIndexName)) { + return projectState.metadata().index(candidateIndexName).getIndex(); + } + } + + logger.warn( + "Unable to resolve index name for executing step [{}] for action [{}] with index [{}], defaulting to original index name", + action.steps().get(stepToExecuteIndex).stepName(), + action.name(), + index.getName() + ); + + return index; + } + // visible for testing static Set timeSeriesIndicesStillWithinTimeBounds(ProjectMetadata project, List targetIndices, LongSupplier nowSupplier) { Set tsIndicesWithinBounds = new HashSet<>(); diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java index 16d80e48ad788..fce36fdeb4766 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java @@ -12,6 +12,9 @@ import org.elasticsearch.cluster.ProjectState; import org.elasticsearch.index.Index; +import java.util.List; +import java.util.function.Function; + /** * 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 +45,17 @@ public interface DlmStep { */ String stepName(); + /** + * A list of functions that can be used to generate possible output index name patterns for this step based 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. + *
+ * The list is ordered and the first function that generates an index name that exists in the cluster will be used by later steps. + *
+ * If not overridden, this method simply returns a list with the identity function, meaning that by default steps will target + * the same index as the input index. + * @return A list of functions that take an index name and return a possible output index name pattern. + */ + default List> possibleOutputIndexNamePatterns() { + return List.of(Function.identity()); + } } diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/DataStreamsPluginTests.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/DataStreamsPluginTests.java new file mode 100644 index 0000000000000..768b0a056dd46 --- /dev/null +++ b/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.outputIndexNamePatterns = 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.outputIndexNamePatterns = 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.outputIndexNamePatterns = 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.outputIndexNamePatterns = 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.outputIndexNamePatterns = 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.outputIndexNamePatterns = 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> outputIndexNamePatterns = 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> possibleOutputIndexNamePatterns() { + if (outputIndexNamePatterns != null) { + return outputIndexNamePatterns; + } + return DlmStep.super.possibleOutputIndexNamePatterns(); + } + } + + private static class TestDlmAction implements DlmAction { + private final String name; + private final List steps; + + TestDlmAction(String name, List steps) { + this.name = name; + this.steps = steps; + } + + @Override + public String name() { + return name; + } + + @Override + public Function applyAfterTime() { + return dsl -> null; + } + + @Override + public List steps() { + return steps; + } + + @Override + public boolean canRunOnProject(DlmActionContext dlmActionContext) { + return true; + } + } +} diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java index 0fd7a9e931ae6..dc83e582cf185 100644 --- a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java +++ b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java @@ -1841,6 +1841,7 @@ private static class TestDlmStep implements DlmStep { int completedCheckCount = 0; int executeCount = 0; final Set executedIndices = new HashSet<>(); + List> outputIndexNamePatterns = null; @Override public boolean stepCompleted(Index index, ProjectState projectState) { @@ -1861,6 +1862,14 @@ public void execute(DlmStepContext dlmStepContext) { public String stepName() { return "Test Step"; } + + @Override + public List> possibleOutputIndexNamePatterns() { + if (outputIndexNamePatterns != null) { + return outputIndexNamePatterns; + } + return DlmStep.super.possibleOutputIndexNamePatterns(); + } } private static class TestDlmAction implements DlmAction { @@ -2319,6 +2328,148 @@ public void testMaybeProcessDlmActionsCanRunOnProjectReturnsFalse() { assertThat(processedIndices, empty()); } + public void testResolveIndexOutputFromPreviousStepExecutionFirstStep() { + TestDlmStep step1 = new TestDlmStep(); + TestDlmStep step2 = new TestDlmStep(); + TestDlmAction action = new TestDlmAction(TimeValue.timeValueMillis(1), step1, step2); + + ProjectMetadata.Builder builder = ProjectMetadata.builder(randomProjectIdOrDefault()); + IndexMetadata indexMetadata = IndexMetadata.builder("original-index") + .settings(settings(IndexVersion.current())) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + builder.put(indexMetadata, false); + ProjectState projectState = projectStateFromProject(builder); + + Index originalIndex = indexMetadata.getIndex(); + Index result = dataStreamLifecycleService.resolveIndexOutputFromPreviousStep(0, originalIndex, action, projectState); + assertThat(result, is(originalIndex)); + } + + public void testResolveIndexForStepExecutionPreviousStepOutputMatchesExistingIndex() { + TestDlmStep step1 = new TestDlmStep(); + step1.outputIndexNamePatterns = List.of(name -> "cloned-" + name); + TestDlmStep step2 = new TestDlmStep(); + TestDlmAction action = new TestDlmAction(TimeValue.timeValueMillis(1), step1, step2); + + ProjectMetadata.Builder builder = ProjectMetadata.builder(randomProjectIdOrDefault()); + IndexMetadata originalMeta = IndexMetadata.builder("original-index") + .settings(settings(IndexVersion.current())) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + IndexMetadata clonedMeta = IndexMetadata.builder("cloned-original-index") + .settings(settings(IndexVersion.current())) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + builder.put(originalMeta, false); + builder.put(clonedMeta, false); + ProjectState projectState = projectStateFromProject(builder); + + Index result = dataStreamLifecycleService.resolveIndexOutputFromPreviousStep(1, originalMeta.getIndex(), action, projectState); + assertThat(result, is(clonedMeta.getIndex())); + } + + public void testResolveIndexForStepExecutionNoMatchReturnsOriginalIndex() { + TestDlmStep step1 = new TestDlmStep(); + step1.outputIndexNamePatterns = List.of(name -> "nonexistent-" + name); + TestDlmStep step2 = new TestDlmStep(); + TestDlmAction action = new TestDlmAction(TimeValue.timeValueMillis(1), step1, step2); + + ProjectMetadata.Builder builder = ProjectMetadata.builder(randomProjectIdOrDefault()); + IndexMetadata originalMeta = IndexMetadata.builder("original-index") + .settings(settings(IndexVersion.current())) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + builder.put(originalMeta, false); + ProjectState projectState = projectStateFromProject(builder); + + Index originalIndex = originalMeta.getIndex(); + try (var mockLog = MockLog.capture(DataStreamLifecycleService.class)) { + mockLog.addExpectation( + new MockLog.SeenEventExpectation( + "resolve index warning", + DataStreamLifecycleService.class.getCanonicalName(), + Level.WARN, + "Unable to resolve index name for executing step [Test Step] for action [Test DLM Action] with index [" + + originalIndex.getName() + + "], defaulting to original index name" + ) + ); + + Index result = dataStreamLifecycleService.resolveIndexOutputFromPreviousStep(1, originalIndex, action, projectState); + assertThat(result, is(originalIndex)); + mockLog.assertAllExpectationsMatched(); + } + } + + public void testResolveIndexOutputFromPreviousStepMultiplePatternsFirstMatchWins() { + TestDlmStep step1 = new TestDlmStep(); + step1.outputIndexNamePatterns = List.of(name -> "first-" + name, name -> "second-" + name); + TestDlmStep step2 = new TestDlmStep(); + TestDlmAction action = new TestDlmAction(TimeValue.timeValueMillis(1), step1, step2); + + ProjectMetadata.Builder builder = ProjectMetadata.builder(randomProjectIdOrDefault()); + IndexMetadata originalMeta = IndexMetadata.builder("original-index") + .settings(settings(IndexVersion.current())) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + IndexMetadata firstMeta = IndexMetadata.builder("first-original-index") + .settings(settings(IndexVersion.current())) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + IndexMetadata secondMeta = IndexMetadata.builder("second-original-index") + .settings(settings(IndexVersion.current())) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + builder.put(originalMeta, false); + builder.put(firstMeta, false); + builder.put(secondMeta, false); + ProjectState projectState = projectStateFromProject(builder); + + Index result = dataStreamLifecycleService.resolveIndexOutputFromPreviousStep(1, originalMeta.getIndex(), action, projectState); + assertThat(result, is(firstMeta.getIndex())); + } + + public void testResolveIndexOutputFromPreviousStepSkipsNonMatchingPatternsUsesSecond() { + TestDlmStep step1 = new TestDlmStep(); + step1.outputIndexNamePatterns = List.of(name -> "nonexistent-" + name, name -> "second-" + name); + TestDlmStep step2 = new TestDlmStep(); + TestDlmAction action = new TestDlmAction(TimeValue.timeValueMillis(1), step1, step2); + + ProjectMetadata.Builder builder = ProjectMetadata.builder(randomProjectIdOrDefault()); + IndexMetadata originalMeta = IndexMetadata.builder("original-index") + .settings(settings(IndexVersion.current())) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + IndexMetadata secondMeta = IndexMetadata.builder("second-original-index") + .settings(settings(IndexVersion.current())) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + builder.put(originalMeta, false); + builder.put(secondMeta, false); + ProjectState projectState = projectStateFromProject(builder); + + Index result = dataStreamLifecycleService.resolveIndexOutputFromPreviousStep(1, originalMeta.getIndex(), action, projectState); + assertThat(result, is(secondMeta.getIndex())); + } + + public void testDlmStepDefaultPossibleOutputIndexNamePatterns() { + TestDlmStep step = new TestDlmStep(); + List> patterns = step.possibleOutputIndexNamePatterns(); + assertThat(patterns, hasSize(1)); + String testName = randomAlphaOfLength(10); + assertThat(patterns.get(0).apply(testName), is(testName)); + } + public void testFormatExecutionTimeMilliseconds() { assertThat(DataStreamLifecycleService.formatExecutionTime(500), equalTo("500ms/500ms")); } From 97950c34d33b2830a28224da877f1eff30d59edb Mon Sep 17 00:00:00 2001 From: Luke Whiting Date: Tue, 24 Feb 2026 14:45:38 +0000 Subject: [PATCH 2/5] Fix mutated index name when finding first incomplete step --- .../datastreams/lifecycle/DataStreamLifecycleService.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java index 137e8147631ed..dd338b12bdccf 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java @@ -641,8 +641,8 @@ private int findFirstIncompleteStepIndex(ProjectState projectState, DataStream d DlmStep step = action.steps().get(i); try { long checkStartTime = nowSupplier.getAsLong(); - index = resolveIndexOutputFromPreviousStep(i, index, action, projectState); - if (step.stepCompleted(index, projectState) == false) { + Index indexInUse = resolveIndexOutputFromPreviousStep(i, index, action, projectState); + if (step.stepCompleted(indexInUse, projectState) == false) { stepToExecute = i; if (logger.isTraceEnabled()) { logger.trace( @@ -650,7 +650,7 @@ private int findFirstIncompleteStepIndex(ProjectState projectState, DataStream d step.stepName(), action.name(), dataStream.getName(), - index.getName(), + indexInUse.getName(), formatExecutionTime(nowSupplier.getAsLong() - checkStartTime) ); } @@ -661,7 +661,7 @@ private int findFirstIncompleteStepIndex(ProjectState projectState, DataStream d step.stepName(), action.name(), dataStream.getName(), - index.getName(), + indexInUse.getName(), formatExecutionTime(nowSupplier.getAsLong() - checkStartTime) ); } From da9e6503d69b1c39ec1a05941872b4ca28701948 Mon Sep 17 00:00:00 2001 From: Luke Whiting Date: Tue, 24 Feb 2026 14:47:36 +0000 Subject: [PATCH 3/5] Update modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../datastreams/lifecycle/transitions/DlmStep.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java index fce36fdeb4766..7e165e75522b1 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java @@ -46,7 +46,7 @@ public interface DlmStep { String stepName(); /** - * A list of functions that can be used to generate possible output index name patterns for this step based input index name. + * 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. *
* The list is ordered and the first function that generates an index name that exists in the cluster will be used by later steps. From bb47a8bd54f96b31a85fcd2d3dc446b48c42198f Mon Sep 17 00:00:00 2001 From: Luke Whiting Date: Fri, 27 Feb 2026 11:50:12 +0000 Subject: [PATCH 4/5] PR Changes --- .../datastreams/DataStreamsPlugin.java | 2 +- .../lifecycle/DataStreamLifecycleService.java | 20 +++---- .../lifecycle/transitions/DlmStep.java | 12 ++--- .../datastreams/DataStreamsPluginTests.java | 22 ++++---- .../DataStreamLifecycleServiceTests.java | 54 +++++++++---------- 5 files changed, 54 insertions(+), 56 deletions(-) diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java index e41759ae1f733..28c58d1430e4e 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java @@ -248,7 +248,7 @@ static void verifyActions(List dlmActions) { throw new IllegalStateException("DLM action [" + action.name() + "] must have at least one step"); } for (DlmStep step : action.steps()) { - if (step.possibleOutputIndexNamePatterns().isEmpty()) { + if (step.possibleOutputIndexNamePatterns("dummy-index").isEmpty()) { throw new IllegalStateException( "DLM step [" + step.stepName() diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java index dd338b12bdccf..4a69d4af1bb8f 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java @@ -636,6 +636,7 @@ Set maybeProcessDlmActions(ProjectState projectState, DataStream dataStre } 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); @@ -691,21 +692,22 @@ Index resolveIndexOutputFromPreviousStep(int stepToExecuteIndex, Index index, Dl } DlmStep previousStep = action.steps().get(stepToExecuteIndex - 1); - List> possibleOutputIndexNamePatterns = previousStep.possibleOutputIndexNamePatterns(); + List possibleOutputIndexNamePatterns = previousStep.possibleOutputIndexNamePatterns(index.getName()); - for (Function possibleOutputIndexNamePattern : possibleOutputIndexNamePatterns) { - String candidateIndexName = possibleOutputIndexNamePattern.apply(index.getName()); + for (String candidateIndexName : possibleOutputIndexNamePatterns) { if (projectState.metadata().hasIndex(candidateIndexName)) { return projectState.metadata().index(candidateIndexName).getIndex(); } } - logger.warn( - "Unable to resolve index name for executing step [{}] for action [{}] with index [{}], defaulting to original index name", - action.steps().get(stepToExecuteIndex).stepName(), - action.name(), - index.getName() - ); + 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; } diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java index 7e165e75522b1..aab92c8de642a 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java @@ -13,7 +13,6 @@ import org.elasticsearch.index.Index; import java.util.List; -import java.util.function.Function; /** * A step within a Data Lifecycle Management action. Each step is responsible for determining if it has been completed for a given index @@ -49,13 +48,14 @@ public interface DlmStep { * 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. *
- * The list is ordered and the first function that generates an index name that exists in the cluster will be used by later steps. + * The list is ordered and the first pattern that generates an index name that exists in the cluster will be used by later steps. *
- * If not overridden, this method simply returns a list with the identity function, meaning that by default steps will target + * 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. - * @return A list of functions that take an index name and return a possible output index name pattern. + * @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> possibleOutputIndexNamePatterns() { - return List.of(Function.identity()); + default List possibleOutputIndexNamePatterns(String indexName) { + return List.of(indexName); } } diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/DataStreamsPluginTests.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/DataStreamsPluginTests.java index 768b0a056dd46..c173d5ee6c4ba 100644 --- a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/DataStreamsPluginTests.java +++ b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/DataStreamsPluginTests.java @@ -37,14 +37,14 @@ public void testVerifyActionsWithDefaultOutputPatterns() { public void testVerifyActionsWithSingleCustomOutputPattern() { TestDlmStep step = new TestDlmStep("step-1"); - step.outputIndexNamePatterns = List.of(name -> "cloned-" + name); + 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.outputIndexNamePatterns = List.of(name -> "partial-" + name, name -> "full-" + name); + step.outputIndexNamePatternFunctions = List.of(name -> "partial-" + name, name -> "full-" + name); DlmAction action = new TestDlmAction("valid-action", List.of(step)); DataStreamsPlugin.verifyActions(List.of(action)); } @@ -52,14 +52,14 @@ public void testVerifyActionsWithMultipleCustomOutputPatterns() { public void testVerifyActionsWithMixOfDefaultAndCustomOutputPatterns() { TestDlmStep defaultStep = new TestDlmStep("default-step"); TestDlmStep customStep = new TestDlmStep("custom-step"); - customStep.outputIndexNamePatterns = List.of(name -> "cloned-" + name); + 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.outputIndexNamePatterns = List.of(name -> "cloned-" + name); + 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)); @@ -73,7 +73,7 @@ public void testVerifyActionsThrowsOnActionWithNoSteps() { public void testVerifyActionsThrowsOnStepWithEmptyOutputPatterns() { TestDlmStep step = new TestDlmStep("bad-step"); - step.outputIndexNamePatterns = List.of(); + step.outputIndexNamePatternFunctions = List.of(); DlmAction action = new TestDlmAction("my-action", List.of(step)); IllegalStateException e = expectThrows(IllegalStateException.class, () -> DataStreamsPlugin.verifyActions(List.of(action))); @@ -94,7 +94,7 @@ public void testVerifyActionsThrowsOnSecondActionWithNoSteps() { public void testVerifyActionsThrowsOnSecondStepWithEmptyOutputPatterns() { TestDlmStep goodStep = new TestDlmStep("good-step"); TestDlmStep badStep = new TestDlmStep("bad-step"); - badStep.outputIndexNamePatterns = List.of(); + badStep.outputIndexNamePatternFunctions = List.of(); DlmAction action = new TestDlmAction("my-action", List.of(goodStep, badStep)); IllegalStateException e = expectThrows(IllegalStateException.class, () -> DataStreamsPlugin.verifyActions(List.of(action))); @@ -103,7 +103,7 @@ public void testVerifyActionsThrowsOnSecondStepWithEmptyOutputPatterns() { private static class TestDlmStep implements DlmStep { private final String name; - List> outputIndexNamePatterns = null; + List> outputIndexNamePatternFunctions = null; TestDlmStep(String name) { this.name = name; @@ -123,11 +123,11 @@ public String stepName() { } @Override - public List> possibleOutputIndexNamePatterns() { - if (outputIndexNamePatterns != null) { - return outputIndexNamePatterns; + public List possibleOutputIndexNamePatterns(String indexName) { + if (outputIndexNamePatternFunctions != null) { + return outputIndexNamePatternFunctions.stream().map(func -> func.apply(indexName)).toList(); } - return DlmStep.super.possibleOutputIndexNamePatterns(); + return DlmStep.super.possibleOutputIndexNamePatterns(indexName); } } diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java index dc83e582cf185..97bc96d2950ce 100644 --- a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java +++ b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java @@ -1841,7 +1841,7 @@ private static class TestDlmStep implements DlmStep { int completedCheckCount = 0; int executeCount = 0; final Set executedIndices = new HashSet<>(); - List> outputIndexNamePatterns = null; + List> outputIndexNamePatternFunctions = null; @Override public boolean stepCompleted(Index index, ProjectState projectState) { @@ -1864,11 +1864,11 @@ public String stepName() { } @Override - public List> possibleOutputIndexNamePatterns() { - if (outputIndexNamePatterns != null) { - return outputIndexNamePatterns; + public List possibleOutputIndexNamePatterns(String indexName) { + if (outputIndexNamePatternFunctions != null) { + return outputIndexNamePatternFunctions.stream().map(func -> func.apply(indexName)).toList(); } - return DlmStep.super.possibleOutputIndexNamePatterns(); + return DlmStep.super.possibleOutputIndexNamePatterns(indexName); } } @@ -2349,7 +2349,7 @@ public void testResolveIndexOutputFromPreviousStepExecutionFirstStep() { public void testResolveIndexForStepExecutionPreviousStepOutputMatchesExistingIndex() { TestDlmStep step1 = new TestDlmStep(); - step1.outputIndexNamePatterns = List.of(name -> "cloned-" + name); + step1.outputIndexNamePatternFunctions = List.of(name -> "cloned-" + name); TestDlmStep step2 = new TestDlmStep(); TestDlmAction action = new TestDlmAction(TimeValue.timeValueMillis(1), step1, step2); @@ -2372,9 +2372,9 @@ public void testResolveIndexForStepExecutionPreviousStepOutputMatchesExistingInd assertThat(result, is(clonedMeta.getIndex())); } - public void testResolveIndexForStepExecutionNoMatchReturnsOriginalIndex() { + public void testResolveIndexForStepExecutionNoMatchThrowsAssertionError() { TestDlmStep step1 = new TestDlmStep(); - step1.outputIndexNamePatterns = List.of(name -> "nonexistent-" + name); + step1.outputIndexNamePatternFunctions = List.of(name -> "nonexistent-" + name); TestDlmStep step2 = new TestDlmStep(); TestDlmAction action = new TestDlmAction(TimeValue.timeValueMillis(1), step1, step2); @@ -2388,27 +2388,23 @@ public void testResolveIndexForStepExecutionNoMatchReturnsOriginalIndex() { ProjectState projectState = projectStateFromProject(builder); Index originalIndex = originalMeta.getIndex(); - try (var mockLog = MockLog.capture(DataStreamLifecycleService.class)) { - mockLog.addExpectation( - new MockLog.SeenEventExpectation( - "resolve index warning", - DataStreamLifecycleService.class.getCanonicalName(), - Level.WARN, - "Unable to resolve index name for executing step [Test Step] for action [Test DLM Action] with index [" - + originalIndex.getName() - + "], defaulting to original index name" - ) - ); - - Index result = dataStreamLifecycleService.resolveIndexOutputFromPreviousStep(1, originalIndex, action, projectState); - assertThat(result, is(originalIndex)); - mockLog.assertAllExpectationsMatched(); - } + AssertionError e = expectThrows( + AssertionError.class, + () -> dataStreamLifecycleService.resolveIndexOutputFromPreviousStep(1, originalIndex, action, projectState) + ); + assertThat( + e.getMessage(), + is( + "Unable to resolve index name for executing step [Test Step] for action [Test DLM Action] with index [" + + originalIndex.getName() + + "]" + ) + ); } public void testResolveIndexOutputFromPreviousStepMultiplePatternsFirstMatchWins() { TestDlmStep step1 = new TestDlmStep(); - step1.outputIndexNamePatterns = List.of(name -> "first-" + name, name -> "second-" + name); + step1.outputIndexNamePatternFunctions = List.of(name -> "first-" + name, name -> "second-" + name); TestDlmStep step2 = new TestDlmStep(); TestDlmAction action = new TestDlmAction(TimeValue.timeValueMillis(1), step1, step2); @@ -2439,7 +2435,7 @@ public void testResolveIndexOutputFromPreviousStepMultiplePatternsFirstMatchWins public void testResolveIndexOutputFromPreviousStepSkipsNonMatchingPatternsUsesSecond() { TestDlmStep step1 = new TestDlmStep(); - step1.outputIndexNamePatterns = List.of(name -> "nonexistent-" + name, name -> "second-" + name); + step1.outputIndexNamePatternFunctions = List.of(name -> "nonexistent-" + name, name -> "second-" + name); TestDlmStep step2 = new TestDlmStep(); TestDlmAction action = new TestDlmAction(TimeValue.timeValueMillis(1), step1, step2); @@ -2464,10 +2460,10 @@ public void testResolveIndexOutputFromPreviousStepSkipsNonMatchingPatternsUsesSe public void testDlmStepDefaultPossibleOutputIndexNamePatterns() { TestDlmStep step = new TestDlmStep(); - List> patterns = step.possibleOutputIndexNamePatterns(); - assertThat(patterns, hasSize(1)); String testName = randomAlphaOfLength(10); - assertThat(patterns.get(0).apply(testName), is(testName)); + List patterns = step.possibleOutputIndexNamePatterns(testName); + assertThat(patterns, hasSize(1)); + assertThat(patterns.getFirst(), is(testName)); } public void testFormatExecutionTimeMilliseconds() { From 629c436c6394c0b4a3567d8888318724b7e05eb0 Mon Sep 17 00:00:00 2001 From: Luke Whiting Date: Mon, 2 Mar 2026 11:37:03 +0000 Subject: [PATCH 5/5] Additional PR changes --- .../lifecycle/DataStreamLifecycleService.java | 35 +++++++++---------- .../lifecycle/transitions/DlmStep.java | 15 ++++---- 2 files changed, 24 insertions(+), 26 deletions(-) diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java index 5096b41131eae..1e347f86cbb5c 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java @@ -701,24 +701,23 @@ Index resolveIndexOutputFromPreviousStep(int stepToExecuteIndex, Index index, Dl } DlmStep previousStep = action.steps().get(stepToExecuteIndex - 1); - List 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; + List possibleIndexNames = previousStep.possibleOutputIndexNamePatterns(index.getName()); + + return possibleIndexNames.stream() + .filter(projectState.metadata()::hasIndex) + .findFirst() + .map(possibleName -> projectState.metadata().index(possibleName).getIndex()) + .orElseGet(() -> { + 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 diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java index aab92c8de642a..3d41e74b565ae 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/transitions/DlmStep.java @@ -45,15 +45,14 @@ 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. + * Returns a list of possible index name patterns that this step may end up creating. This is then used by later steps to + * determine the name of the index they should work on after this step is run. The order is important as the first pattern that + * matches an existing index will be used by the later step. *
- * The list is ordered and the first pattern that generates an index name that exists in the cluster will be used by later steps. - *
- * 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. + * The default implementation returns in the steps input index name as the only possible output index name pattern, + * which is sufficient for steps that do not change the index name. + * @param indexName Index name this step ran on + * @return List of possible index name patterns that this step may end up creating */ default List possibleOutputIndexNamePatterns(String indexName) { return List.of(indexName);