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 20e72961f66d8..e5014b81d59e0 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.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 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 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 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 5edf99afd83ad..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 @@ -577,7 +577,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( @@ -589,7 +589,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 [{}]", @@ -599,7 +600,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; @@ -642,20 +644,23 @@ 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) { + 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,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 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 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..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 @@ -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,17 @@ public interface DlmStep { */ String stepName(); + /** + * 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 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); + } } 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..c173d5ee6c4ba --- /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.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> 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 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 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..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,6 +1841,7 @@ private static class TestDlmStep implements DlmStep { int completedCheckCount = 0; int executeCount = 0; final Set executedIndices = new HashSet<>(); + List> outputIndexNamePatternFunctions = 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(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 { @@ -2319,6 +2328,144 @@ 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.outputIndexNamePatternFunctions = 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 testResolveIndexForStepExecutionNoMatchThrowsAssertionError() { + TestDlmStep step1 = new TestDlmStep(); + step1.outputIndexNamePatternFunctions = 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(); + 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.outputIndexNamePatternFunctions = 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.outputIndexNamePatternFunctions = 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(); + String testName = randomAlphaOfLength(10); + List patterns = step.possibleOutputIndexNamePatterns(testName); + assertThat(patterns, hasSize(1)); + assertThat(patterns.getFirst(), is(testName)); + } + public void testFormatExecutionTimeMilliseconds() { assertThat(DataStreamLifecycleService.formatExecutionTime(500), equalTo("500ms/500ms")); }