Replace PhaseAfterStep with PhaseCompleteStep (#33398)
This removes `PhaseAfterStep` in favor of a new `PhaseCompleteStep`. This step in only a marker that the `LifecyclePolicyRunner` needs to halt until the time indicated for entering the next phase. This also fixes a bug where phase times were encapsulated into the policy instead of dynamically adjusting to policy changes. Supersedes #33140, which it replaces Relates to #29823
This commit is contained in:
Lee Hinman
GitHub
parent
8b8ff2bc6e
commit
96d515e3f5
|
|
@ -145,35 +145,39 @@ public class IndexLifecycleIT extends ESRestHighLevelClientTestCase {
|
|||
highLevelClient().indexLifecycle()::putLifecyclePolicyAsync);
|
||||
assertTrue(putResponse.isAcknowledged());
|
||||
|
||||
createIndex("foo", Settings.builder().put("index.lifecycle.name", policy.getName()).build());
|
||||
createIndex("baz", Settings.builder().put("index.lifecycle.name", policy.getName()).build());
|
||||
createIndex("foo-01", Settings.builder().put("index.lifecycle.name", policy.getName())
|
||||
.put("index.lifecycle.rollover_alias", "foo-alias").build(), "", "\"foo-alias\" : {}");
|
||||
|
||||
createIndex("baz-01", Settings.builder().put("index.lifecycle.name", policy.getName())
|
||||
.put("index.lifecycle.rollover_alias", "baz-alias").build(), "", "\"baz-alias\" : {}");
|
||||
|
||||
createIndex("squash", Settings.EMPTY);
|
||||
assertBusy(() -> {
|
||||
GetSettingsRequest getSettingsRequest = new GetSettingsRequest().indices("foo", "baz");
|
||||
GetSettingsRequest getSettingsRequest = new GetSettingsRequest().indices("foo-01", "baz-01");
|
||||
GetSettingsResponse settingsResponse = highLevelClient().indices().getSettings(getSettingsRequest, RequestOptions.DEFAULT);
|
||||
assertThat(settingsResponse.getSetting("foo", "index.lifecycle.name"), equalTo(policy.getName()));
|
||||
assertThat(settingsResponse.getSetting("baz", "index.lifecycle.name"), equalTo(policy.getName()));
|
||||
assertThat(settingsResponse.getSetting("foo", "index.lifecycle.phase"), equalTo("hot"));
|
||||
assertThat(settingsResponse.getSetting("baz", "index.lifecycle.phase"), equalTo("hot"));
|
||||
assertThat(settingsResponse.getSetting("foo-01", "index.lifecycle.name"), equalTo(policy.getName()));
|
||||
assertThat(settingsResponse.getSetting("baz-01", "index.lifecycle.name"), equalTo(policy.getName()));
|
||||
assertThat(settingsResponse.getSetting("foo-01", "index.lifecycle.phase"), equalTo("hot"));
|
||||
assertThat(settingsResponse.getSetting("baz-01", "index.lifecycle.phase"), equalTo("hot"));
|
||||
});
|
||||
|
||||
ExplainLifecycleRequest req = new ExplainLifecycleRequest();
|
||||
req.indices("foo", "baz", "squash");
|
||||
req.indices("foo-01", "baz-01", "squash");
|
||||
ExplainLifecycleResponse response = execute(req, highLevelClient().indexLifecycle()::explainLifecycle,
|
||||
highLevelClient().indexLifecycle()::explainLifecycleAsync);
|
||||
Map<String, IndexLifecycleExplainResponse> indexResponses = response.getIndexResponses();
|
||||
assertEquals(3, indexResponses.size());
|
||||
IndexLifecycleExplainResponse fooResponse = indexResponses.get("foo");
|
||||
IndexLifecycleExplainResponse fooResponse = indexResponses.get("foo-01");
|
||||
assertNotNull(fooResponse);
|
||||
assertTrue(fooResponse.managedByILM());
|
||||
assertEquals("foo", fooResponse.getIndex());
|
||||
assertEquals("foo-01", fooResponse.getIndex());
|
||||
assertEquals("hot", fooResponse.getPhase());
|
||||
assertEquals("rollover", fooResponse.getAction());
|
||||
assertEquals("attempt_rollover", fooResponse.getStep());
|
||||
IndexLifecycleExplainResponse bazResponse = indexResponses.get("baz");
|
||||
IndexLifecycleExplainResponse bazResponse = indexResponses.get("baz-01");
|
||||
assertNotNull(bazResponse);
|
||||
assertTrue(bazResponse.managedByILM());
|
||||
assertEquals("baz", bazResponse.getIndex());
|
||||
assertEquals("baz-01", bazResponse.getIndex());
|
||||
assertEquals("hot", bazResponse.getPhase());
|
||||
assertEquals("rollover", bazResponse.getAction());
|
||||
assertEquals("attempt_rollover", bazResponse.getStep());
|
||||
|
|
|
|||
|
|
@ -543,6 +543,14 @@ public abstract class ESRestTestCase extends ESTestCase {
|
|||
client().performRequest(request);
|
||||
}
|
||||
|
||||
protected static void createIndex(String name, Settings settings, String mapping, String aliases) throws IOException {
|
||||
Request request = new Request("PUT", "/" + name);
|
||||
request.setJsonEntity("{\n \"settings\": " + Strings.toString(settings)
|
||||
+ ", \"mappings\" : {" + mapping + "}"
|
||||
+ ", \"aliases\": {" + aliases + "} }");
|
||||
client().performRequest(request);
|
||||
}
|
||||
|
||||
protected static void deleteIndex(String name) throws IOException {
|
||||
Request request = new Request("DELETE", "/" + name);
|
||||
client().performRequest(request);
|
||||
|
|
|
|||
|
|
@ -28,7 +28,6 @@ import java.util.ListIterator;
|
|||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.LongSupplier;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
|
|
@ -165,10 +164,9 @@ public class LifecyclePolicy extends AbstractDiffable<LifecyclePolicy>
|
|||
*
|
||||
* @param client The Elasticsearch Client to use during execution of {@link AsyncActionStep}
|
||||
* and {@link AsyncWaitStep} steps.
|
||||
* @param nowSupplier The supplier of the current time for {@link PhaseAfterStep} steps.
|
||||
* @return The list of {@link Step} objects in order of their execution.
|
||||
*/
|
||||
public List<Step> toSteps(Client client, LongSupplier nowSupplier) {
|
||||
public List<Step> toSteps(Client client) {
|
||||
List<Step> steps = new ArrayList<>();
|
||||
List<Phase> orderedPhases = type.getOrderedPhases(phases);
|
||||
ListIterator<Phase> phaseIterator = orderedPhases.listIterator(orderedPhases.size());
|
||||
|
|
@ -187,8 +185,8 @@ public class LifecyclePolicy extends AbstractDiffable<LifecyclePolicy>
|
|||
if (phase != null) {
|
||||
// after step should have the name of the previous phase since the index is still in the
|
||||
// previous phase until the after condition is reached
|
||||
Step.StepKey afterStepKey = new Step.StepKey(previousPhase.getName(), PhaseAfterStep.NAME, PhaseAfterStep.NAME);
|
||||
Step phaseAfterStep = new PhaseAfterStep(nowSupplier, phase.getAfter(), afterStepKey, lastStepKey);
|
||||
Step.StepKey afterStepKey = new Step.StepKey(previousPhase.getName(), PhaseCompleteStep.NAME, PhaseCompleteStep.NAME);
|
||||
Step phaseAfterStep = new PhaseCompleteStep(afterStepKey, lastStepKey);
|
||||
steps.add(phaseAfterStep);
|
||||
lastStepKey = phaseAfterStep.getKey();
|
||||
}
|
||||
|
|
@ -211,8 +209,8 @@ public class LifecyclePolicy extends AbstractDiffable<LifecyclePolicy>
|
|||
|
||||
if (phase != null) {
|
||||
// The very first after step is in a phase before the hot phase so call this "new"
|
||||
Step.StepKey afterStepKey = new Step.StepKey("new", PhaseAfterStep.NAME, PhaseAfterStep.NAME);
|
||||
Step phaseAfterStep = new PhaseAfterStep(nowSupplier, phase.getAfter(), afterStepKey, lastStepKey);
|
||||
Step.StepKey afterStepKey = new Step.StepKey("new", PhaseCompleteStep.NAME, PhaseCompleteStep.NAME);
|
||||
Step phaseAfterStep = new PhaseCompleteStep(afterStepKey, lastStepKey);
|
||||
steps.add(phaseAfterStep);
|
||||
lastStepKey = phaseAfterStep.getKey();
|
||||
}
|
||||
|
|
@ -289,7 +287,7 @@ public class LifecyclePolicy extends AbstractDiffable<LifecyclePolicy>
|
|||
// there are no more steps we should execute
|
||||
return TerminalPolicyStep.KEY;
|
||||
} else {
|
||||
return new StepKey(currentPhaseName, PhaseAfterStep.NAME, PhaseAfterStep.NAME);
|
||||
return new StepKey(currentPhaseName, PhaseCompleteStep.NAME, PhaseCompleteStep.NAME);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -306,7 +304,7 @@ public class LifecyclePolicy extends AbstractDiffable<LifecyclePolicy>
|
|||
// InitializePolicyContextStep
|
||||
return InitializePolicyContextStep.KEY;
|
||||
}
|
||||
return new StepKey(prevPhaseName, PhaseAfterStep.NAME, PhaseAfterStep.NAME);
|
||||
return new StepKey(prevPhaseName, PhaseCompleteStep.NAME, PhaseCompleteStep.NAME);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,60 +0,0 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.core.indexlifecycle;
|
||||
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.metadata.IndexMetaData;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.index.Index;
|
||||
|
||||
import java.util.Objects;
|
||||
import java.util.function.LongSupplier;
|
||||
|
||||
public class PhaseAfterStep extends ClusterStateWaitStep {
|
||||
public static final String NAME = "after";
|
||||
private final TimeValue after;
|
||||
private final LongSupplier nowSupplier;
|
||||
|
||||
PhaseAfterStep(LongSupplier nowSupplier, TimeValue after, StepKey key, StepKey nextStepKey) {
|
||||
super(key, nextStepKey);
|
||||
this.nowSupplier = nowSupplier;
|
||||
this.after = after;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Result isConditionMet(Index index, ClusterState clusterState) {
|
||||
IndexMetaData indexMetaData = clusterState.metaData().index(index);
|
||||
long lifecycleDate = indexMetaData.getSettings()
|
||||
.getAsLong(LifecycleSettings.LIFECYCLE_INDEX_CREATION_DATE, -1L);
|
||||
return new Result(nowSupplier.getAsLong() >= lifecycleDate + after.getMillis(), null);
|
||||
}
|
||||
|
||||
TimeValue getAfter() {
|
||||
return after;
|
||||
}
|
||||
|
||||
LongSupplier getNowSupplier() {
|
||||
return nowSupplier;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(super.hashCode(), after);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (obj == null) {
|
||||
return false;
|
||||
}
|
||||
if (getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
PhaseAfterStep other = (PhaseAfterStep) obj;
|
||||
return super.equals(obj) &&
|
||||
Objects.equals(after, other.after);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.core.indexlifecycle;
|
||||
|
||||
/**
|
||||
* This is essentially a marker that a phase has ended, and we need to check
|
||||
* the age of an index before proceeding to the next phase.
|
||||
*/
|
||||
public class PhaseCompleteStep extends Step {
|
||||
public static final String NAME = "complete";
|
||||
|
||||
public PhaseCompleteStep(StepKey key, StepKey nextStepKey) {
|
||||
super(key, nextStepKey);
|
||||
}
|
||||
}
|
||||
|
|
@ -30,7 +30,6 @@ import java.util.Map;
|
|||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.LongSupplier;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
|
|
@ -170,11 +169,10 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
|
|||
|
||||
public void testFirstAndLastSteps() {
|
||||
Client client = mock(Client.class);
|
||||
LongSupplier nowSupplier = () -> 0L;
|
||||
lifecycleName = randomAlphaOfLengthBetween(1, 20);
|
||||
Map<String, Phase> phases = new LinkedHashMap<>();
|
||||
LifecyclePolicy policy = new LifecyclePolicy(TestLifecycleType.INSTANCE, lifecycleName, phases);
|
||||
List<Step> steps = policy.toSteps(client, nowSupplier);
|
||||
List<Step> steps = policy.toSteps(client);
|
||||
assertThat(steps.size(), equalTo(2));
|
||||
assertThat(steps.get(0), instanceOf(InitializePolicyContextStep.class));
|
||||
assertThat(steps.get(0).getKey(), equalTo(new StepKey("new", "init", "init")));
|
||||
|
|
@ -184,7 +182,6 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
|
|||
|
||||
public void testToStepsWithOneStep() {
|
||||
Client client = mock(Client.class);
|
||||
LongSupplier nowSupplier = () -> 0L;
|
||||
MockStep mockStep = new MockStep(
|
||||
new Step.StepKey("test", "test", "test"), TerminalPolicyStep.KEY);
|
||||
|
||||
|
|
@ -196,8 +193,8 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
|
|||
phases.put(firstPhase.getName(), firstPhase);
|
||||
LifecyclePolicy policy = new LifecyclePolicy(TestLifecycleType.INSTANCE, lifecycleName, phases);
|
||||
StepKey firstStepKey = InitializePolicyContextStep.KEY;
|
||||
StepKey secondStepKey = new StepKey("new", PhaseAfterStep.NAME, PhaseAfterStep.NAME);
|
||||
List<Step> steps = policy.toSteps(client, nowSupplier);
|
||||
StepKey secondStepKey = new StepKey("new", PhaseCompleteStep.NAME, PhaseCompleteStep.NAME);
|
||||
List<Step> steps = policy.toSteps(client);
|
||||
assertThat(steps.size(), equalTo(4));
|
||||
assertSame(steps.get(0).getKey(), firstStepKey);
|
||||
assertThat(steps.get(0).getNextStepKey(), equalTo(secondStepKey));
|
||||
|
|
@ -210,13 +207,12 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
|
|||
|
||||
public void testToStepsWithTwoPhases() {
|
||||
Client client = mock(Client.class);
|
||||
LongSupplier nowSupplier = () -> 0L;
|
||||
MockStep secondActionStep = new MockStep(new StepKey("second_phase", "test2", "test"), TerminalPolicyStep.KEY);
|
||||
MockStep secondAfter = new MockStep(new StepKey("first_phase", PhaseAfterStep.NAME, PhaseAfterStep.NAME),
|
||||
MockStep secondAfter = new MockStep(new StepKey("first_phase", PhaseCompleteStep.NAME, PhaseCompleteStep.NAME),
|
||||
secondActionStep.getKey());
|
||||
MockStep firstActionAnotherStep = new MockStep(new StepKey("first_phase", "test", "bar"), secondAfter.getKey());
|
||||
MockStep firstActionStep = new MockStep(new StepKey("first_phase", "test", "foo"), firstActionAnotherStep.getKey());
|
||||
MockStep firstAfter = new MockStep(new StepKey("new", PhaseAfterStep.NAME, PhaseAfterStep.NAME), firstActionStep.getKey());
|
||||
MockStep firstAfter = new MockStep(new StepKey("new", PhaseCompleteStep.NAME, PhaseCompleteStep.NAME), firstActionStep.getKey());
|
||||
MockStep init = new MockStep(InitializePolicyContextStep.KEY, firstAfter.getKey());
|
||||
|
||||
lifecycleName = randomAlphaOfLengthBetween(1, 20);
|
||||
|
|
@ -231,17 +227,17 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
|
|||
phases.put(secondPhase.getName(), secondPhase);
|
||||
LifecyclePolicy policy = new LifecyclePolicy(TestLifecycleType.INSTANCE, lifecycleName, phases);
|
||||
|
||||
List<Step> steps = policy.toSteps(client, nowSupplier);
|
||||
List<Step> steps = policy.toSteps(client);
|
||||
assertThat(steps.size(), equalTo(7));
|
||||
assertThat(steps.get(0).getClass(), equalTo(InitializePolicyContextStep.class));
|
||||
assertThat(steps.get(0).getKey(), equalTo(init.getKey()));
|
||||
assertThat(steps.get(0).getNextStepKey(), equalTo(init.getNextStepKey()));
|
||||
assertThat(steps.get(1).getClass(), equalTo(PhaseAfterStep.class));
|
||||
assertThat(steps.get(1).getClass(), equalTo(PhaseCompleteStep.class));
|
||||
assertThat(steps.get(1).getKey(), equalTo(firstAfter.getKey()));
|
||||
assertThat(steps.get(1).getNextStepKey(), equalTo(firstAfter.getNextStepKey()));
|
||||
assertThat(steps.get(2), equalTo(firstActionStep));
|
||||
assertThat(steps.get(3), equalTo(firstActionAnotherStep));
|
||||
assertThat(steps.get(4).getClass(), equalTo(PhaseAfterStep.class));
|
||||
assertThat(steps.get(4).getClass(), equalTo(PhaseCompleteStep.class));
|
||||
assertThat(steps.get(4).getKey(), equalTo(secondAfter.getKey()));
|
||||
assertThat(steps.get(4).getNextStepKey(), equalTo(secondAfter.getNextStepKey()));
|
||||
assertThat(steps.get(5), equalTo(secondActionStep));
|
||||
|
|
@ -337,7 +333,7 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
|
|||
currentStep = new StepKey("phase_1", "action_3", "step_missing");
|
||||
nextStep = policy.getNextValidStep(currentStep);
|
||||
assertNotNull(nextStep);
|
||||
assertEquals(new StepKey("phase_1", PhaseAfterStep.NAME, PhaseAfterStep.NAME), nextStep);
|
||||
assertEquals(new StepKey("phase_1", PhaseCompleteStep.NAME, PhaseCompleteStep.NAME), nextStep);
|
||||
|
||||
// current action exists but step does not and action is last in the
|
||||
// last phase
|
||||
|
|
@ -356,7 +352,7 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
|
|||
currentStep = new StepKey("phase_1", "action_4", "step_2");
|
||||
nextStep = policy.getNextValidStep(currentStep);
|
||||
assertNotNull(nextStep);
|
||||
assertEquals(new StepKey("phase_1", PhaseAfterStep.NAME, PhaseAfterStep.NAME), nextStep);
|
||||
assertEquals(new StepKey("phase_1", PhaseCompleteStep.NAME, PhaseCompleteStep.NAME), nextStep);
|
||||
|
||||
// current action no longer exists and action was last in the last phase
|
||||
currentStep = new StepKey("phase_4", "action_4", "step_2");
|
||||
|
|
@ -368,7 +364,7 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
|
|||
currentStep = new StepKey("phase_3", "action_2", "step_2");
|
||||
nextStep = policy.getNextValidStep(currentStep);
|
||||
assertNotNull(nextStep);
|
||||
assertEquals(new StepKey("phase_2", PhaseAfterStep.NAME, PhaseAfterStep.NAME), nextStep);
|
||||
assertEquals(new StepKey("phase_2", PhaseCompleteStep.NAME, PhaseCompleteStep.NAME), nextStep);
|
||||
|
||||
// current phase no longer exists and was last phase
|
||||
currentStep = new StepKey("phase_5", "action_2", "step_2");
|
||||
|
|
|
|||
|
|
@ -1,103 +0,0 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.core.indexlifecycle;
|
||||
|
||||
|
||||
import org.elasticsearch.Version;
|
||||
import org.elasticsearch.cluster.ClusterName;
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.metadata.IndexMetaData;
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.index.Index;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.ClusterStateWaitStep.Result;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step.StepKey;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
public class PhaseAfterStepTests extends AbstractStepTestCase<PhaseAfterStep> {
|
||||
|
||||
@Override
|
||||
public PhaseAfterStep createRandomInstance() {
|
||||
StepKey stepKey = randomStepKey();
|
||||
StepKey nextStepKey = randomStepKey();
|
||||
TimeValue after = createRandomTimeValue();
|
||||
return new PhaseAfterStep(null, after, stepKey, nextStepKey);
|
||||
}
|
||||
|
||||
private TimeValue createRandomTimeValue() {
|
||||
return new TimeValue(randomLongBetween(1, 10000), randomFrom(TimeUnit.SECONDS, TimeUnit.MINUTES, TimeUnit.HOURS, TimeUnit.DAYS));
|
||||
}
|
||||
|
||||
@Override
|
||||
public PhaseAfterStep mutateInstance(PhaseAfterStep instance) {
|
||||
StepKey key = instance.getKey();
|
||||
StepKey nextKey = instance.getNextStepKey();
|
||||
TimeValue after = instance.getAfter();
|
||||
|
||||
switch (between(0, 2)) {
|
||||
case 0:
|
||||
key = new StepKey(key.getPhase(), key.getAction(), key.getName() + randomAlphaOfLength(5));
|
||||
break;
|
||||
case 1:
|
||||
nextKey = new StepKey(key.getPhase(), key.getAction(), key.getName() + randomAlphaOfLength(5));
|
||||
break;
|
||||
case 2:
|
||||
after = randomValueOtherThan(after, this::createRandomTimeValue);
|
||||
break;
|
||||
default:
|
||||
throw new AssertionError("Illegal randomisation branch");
|
||||
}
|
||||
|
||||
return new PhaseAfterStep(instance.getNowSupplier(), after, key, nextKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
public PhaseAfterStep copyInstance(PhaseAfterStep instance) {
|
||||
return new PhaseAfterStep(instance.getNowSupplier(), instance.getAfter(),
|
||||
instance.getKey(), instance.getNextStepKey());
|
||||
}
|
||||
|
||||
public void testConditionMet() {
|
||||
long creationDate = randomNonNegativeLong();
|
||||
IndexMetaData indexMetadata = IndexMetaData.builder(randomAlphaOfLength(5))
|
||||
.settings(settings(Version.CURRENT).put(LifecycleSettings.LIFECYCLE_INDEX_CREATION_DATE, creationDate))
|
||||
.creationDate(creationDate)
|
||||
.numberOfShards(1).numberOfReplicas(0).build();
|
||||
MetaData metaData = MetaData.builder()
|
||||
.persistentSettings(settings(Version.CURRENT).build())
|
||||
.put(IndexMetaData.builder(indexMetadata))
|
||||
.build();
|
||||
Index index = indexMetadata.getIndex();
|
||||
ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT).metaData(metaData).build();
|
||||
long after = randomNonNegativeLong();
|
||||
long now = creationDate + after + randomIntBetween(0, 2);
|
||||
PhaseAfterStep step = new PhaseAfterStep(() -> now, TimeValue.timeValueMillis(after), null, null);
|
||||
Result result = step.isConditionMet(index, clusterState);
|
||||
assertTrue(result.isComplete());
|
||||
assertNull(result.getInfomationContext());
|
||||
}
|
||||
|
||||
public void testConditionNotMet() {
|
||||
long creationDate = randomNonNegativeLong();
|
||||
IndexMetaData indexMetadata = IndexMetaData.builder(randomAlphaOfLength(5))
|
||||
.settings(settings(Version.CURRENT).put(LifecycleSettings.LIFECYCLE_INDEX_CREATION_DATE, creationDate))
|
||||
.creationDate(creationDate)
|
||||
.numberOfShards(1).numberOfReplicas(0).build();
|
||||
MetaData metaData = MetaData.builder()
|
||||
.persistentSettings(settings(Version.CURRENT).build())
|
||||
.put(IndexMetaData.builder(indexMetadata))
|
||||
.build();
|
||||
Index index = indexMetadata.getIndex();
|
||||
ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT).metaData(metaData).build();
|
||||
long after = randomNonNegativeLong();
|
||||
long now = creationDate + after - randomIntBetween(1, 1000);
|
||||
PhaseAfterStep step = new PhaseAfterStep(() -> now, TimeValue.timeValueMillis(after), null, null);
|
||||
Result result = step.isConditionMet(index, clusterState);
|
||||
assertFalse(result.isComplete());
|
||||
assertNull(result.getInfomationContext());
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.core.indexlifecycle;
|
||||
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step.StepKey;
|
||||
|
||||
public class PhaseCompleteStepTests extends AbstractStepTestCase<PhaseCompleteStep> {
|
||||
|
||||
@Override
|
||||
public PhaseCompleteStep createRandomInstance() {
|
||||
StepKey stepKey = randomStepKey();
|
||||
StepKey nextStepKey = randomStepKey();
|
||||
return new PhaseCompleteStep(stepKey, nextStepKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
public PhaseCompleteStep mutateInstance(PhaseCompleteStep instance) {
|
||||
StepKey key = instance.getKey();
|
||||
StepKey nextKey = instance.getNextStepKey();
|
||||
|
||||
if (randomBoolean()) {
|
||||
key = new StepKey(key.getPhase(), key.getAction(), key.getName() + randomAlphaOfLength(5));
|
||||
} else {
|
||||
nextKey = new StepKey(key.getPhase(), key.getAction(), key.getName() + randomAlphaOfLength(5));
|
||||
}
|
||||
|
||||
return new PhaseCompleteStep(key, nextKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
public PhaseCompleteStep copyInstance(PhaseCompleteStep instance) {
|
||||
return new PhaseCompleteStep(instance.getKey(), instance.getNextStepKey());
|
||||
}
|
||||
}
|
||||
|
|
@ -16,6 +16,7 @@ import org.elasticsearch.cluster.service.ClusterService;
|
|||
import org.elasticsearch.common.Strings;
|
||||
import org.elasticsearch.common.bytes.BytesReference;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.ToXContentObject;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
|
|
@ -30,6 +31,7 @@ import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
|||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleSettings;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Phase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.PhaseCompleteStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.RolloverAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step.StepKey;
|
||||
|
|
@ -51,6 +53,31 @@ public class IndexLifecycleRunner {
|
|||
this.nowSupplier = nowSupplier;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return true or false depending on whether the index is ready to be in {@code phase}
|
||||
*/
|
||||
boolean isReadyToTransitionToThisPhase(final String policy, final IndexMetaData indexMetaData, final String phase) {
|
||||
final Settings indexSettings = indexMetaData.getSettings();
|
||||
if (indexSettings.hasValue(LifecycleSettings.LIFECYCLE_INDEX_CREATION_DATE) == false) {
|
||||
logger.trace("no index creation date has been set yet");
|
||||
return true;
|
||||
}
|
||||
final long lifecycleDate = indexSettings.getAsLong(LifecycleSettings.LIFECYCLE_INDEX_CREATION_DATE, -1L);
|
||||
assert lifecycleDate >= 0 : "expected index to have a lifecycle date but it did not";
|
||||
final TimeValue after = stepRegistry.getIndexAgeForPhase(policy, phase);
|
||||
final long now = nowSupplier.getAsLong();
|
||||
final TimeValue age = new TimeValue(now - lifecycleDate);
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.trace("[{}] checking for index age to be at least [{}] before performing actions in " +
|
||||
"the \"{}\" phase. Now: {}, lifecycle date: {}, age: [{}/{}s]",
|
||||
indexMetaData.getIndex().getName(), after, phase,
|
||||
new TimeValue(now).seconds(),
|
||||
new TimeValue(lifecycleDate).seconds(),
|
||||
age, age.seconds());
|
||||
}
|
||||
return now >= lifecycleDate + after.getMillis();
|
||||
}
|
||||
|
||||
public void runPolicy(String policy, IndexMetaData indexMetaData, ClusterState currentState,
|
||||
boolean fromClusterStateChange) {
|
||||
Settings indexSettings = indexMetaData.getSettings();
|
||||
|
|
@ -67,13 +94,23 @@ public class IndexLifecycleRunner {
|
|||
+ "] with policy [" + policy + "] is not recognized");
|
||||
return;
|
||||
}
|
||||
logger.debug("running policy with current-step[" + currentStep.getKey() + "]");
|
||||
logger.debug("running policy with current-step [" + currentStep.getKey() + "]");
|
||||
if (currentStep instanceof TerminalPolicyStep) {
|
||||
logger.debug("policy [" + policy + "] for index [" + indexMetaData.getIndex().getName() + "] complete, skipping execution");
|
||||
return;
|
||||
} else if (currentStep instanceof ErrorStep) {
|
||||
logger.debug(
|
||||
"policy [" + policy + "] for index [" + indexMetaData.getIndex().getName() + "] on an error step, skipping execution");
|
||||
} else if (currentStep instanceof ClusterStateActionStep || currentStep instanceof ClusterStateWaitStep) {
|
||||
"policy [" + policy + "] for index [" + indexMetaData.getIndex().getName() + "] on an error step, skipping execution");
|
||||
return;
|
||||
} else if (currentStep instanceof PhaseCompleteStep) {
|
||||
// Only proceed to the next step if enough time has elapsed to go into the next phase
|
||||
if (isReadyToTransitionToThisPhase(policy, indexMetaData, currentStep.getNextStepKey().getPhase())) {
|
||||
moveToStep(indexMetaData.getIndex(), policy, currentStep.getKey(), currentStep.getNextStepKey());
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (currentStep instanceof ClusterStateActionStep || currentStep instanceof ClusterStateWaitStep) {
|
||||
executeClusterStateSteps(indexMetaData.getIndex(), policy, currentStep);
|
||||
} else if (currentStep instanceof AsyncWaitStep) {
|
||||
if (fromClusterStateChange == false) {
|
||||
|
|
|
|||
|
|
@ -146,7 +146,7 @@ public class IndexLifecycleService extends AbstractComponent
|
|||
policyRegistry.removeIndices(event.indicesDeleted());
|
||||
}
|
||||
if (event.state().metaData().custom(IndexLifecycleMetadata.TYPE) != null) {
|
||||
policyRegistry.update(event.state(), client, nowSupplier);
|
||||
policyRegistry.update(event.state(), client);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ package org.elasticsearch.xpack.indexlifecycle;
|
|||
import org.elasticsearch.ElasticsearchException;
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.ClusterStateUpdateTask;
|
||||
import org.elasticsearch.cluster.metadata.IndexMetaData;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.index.Index;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleSettings;
|
||||
|
|
@ -51,7 +52,12 @@ public class MoveToNextStepUpdateTask extends ClusterStateUpdateTask {
|
|||
|
||||
@Override
|
||||
public ClusterState execute(ClusterState currentState) {
|
||||
Settings indexSettings = currentState.getMetaData().index(index).getSettings();
|
||||
IndexMetaData indexMetaData = currentState.getMetaData().index(index);
|
||||
if (indexMetaData == null) {
|
||||
// Index must have been since deleted, ignore it
|
||||
return currentState;
|
||||
}
|
||||
Settings indexSettings = indexMetaData.getSettings();
|
||||
if (policy.equals(LifecycleSettings.LIFECYCLE_NAME_SETTING.get(indexSettings))
|
||||
&& currentStepKey.equals(IndexLifecycleRunner.getCurrentStepKey(indexSettings))) {
|
||||
return IndexLifecycleRunner.moveClusterStateToNextStep(index, currentState, currentStepKey, nextStepKey, nowSupplier);
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ import org.elasticsearch.cluster.metadata.IndexMetaData;
|
|||
import org.elasticsearch.common.Nullable;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.common.xcontent.DeprecationHandler;
|
||||
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
|
|
@ -39,7 +40,6 @@ import java.util.List;
|
|||
import java.util.Map;
|
||||
import java.util.SortedMap;
|
||||
import java.util.TreeMap;
|
||||
import java.util.function.LongSupplier;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
public class PolicyStepsRegistry {
|
||||
|
|
@ -97,7 +97,7 @@ public class PolicyStepsRegistry {
|
|||
}
|
||||
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
public void update(ClusterState clusterState, Client client, LongSupplier nowSupplier) {
|
||||
public void update(ClusterState clusterState, Client client) {
|
||||
final IndexLifecycleMetadata meta = clusterState.metaData().custom(IndexLifecycleMetadata.TYPE);
|
||||
|
||||
assert meta != null : "IndexLifecycleMetadata cannot be null when updating the policy steps registry";
|
||||
|
|
@ -134,7 +134,7 @@ public class PolicyStepsRegistry {
|
|||
LifecyclePolicySecurityClient policyClient = new LifecyclePolicySecurityClient(client, ClientHelper.INDEX_LIFECYCLE_ORIGIN,
|
||||
policyMetadata.getHeaders());
|
||||
lifecyclePolicyMap.put(policyMetadata.getName(), policyMetadata);
|
||||
List<Step> policyAsSteps = policyMetadata.getPolicy().toSteps(policyClient, nowSupplier);
|
||||
List<Step> policyAsSteps = policyMetadata.getPolicy().toSteps(policyClient);
|
||||
if (policyAsSteps.isEmpty() == false) {
|
||||
firstStepMap.put(policyMetadata.getName(), policyAsSteps.get(0));
|
||||
final Map<Step.StepKey, Step> stepMapForPolicy = new HashMap<>();
|
||||
|
|
@ -192,7 +192,7 @@ public class PolicyStepsRegistry {
|
|||
}
|
||||
LifecyclePolicySecurityClient policyClient = new LifecyclePolicySecurityClient(client,
|
||||
ClientHelper.INDEX_LIFECYCLE_ORIGIN, lifecyclePolicyMap.get(policy).getHeaders());
|
||||
final List<Step> steps = policyToExecute.toSteps(policyClient, nowSupplier);
|
||||
final List<Step> steps = policyToExecute.toSteps(policyClient);
|
||||
// Build a list of steps that correspond with the phase the index is currently in
|
||||
final List<Step> phaseSteps;
|
||||
if (steps == null) {
|
||||
|
|
@ -257,4 +257,22 @@ public class PolicyStepsRegistry {
|
|||
return firstStepMap.get(policy);
|
||||
}
|
||||
|
||||
public TimeValue getIndexAgeForPhase(final String policy, final String phase) {
|
||||
// These built in phases should never wait
|
||||
if (InitializePolicyContextStep.INITIALIZATION_PHASE.equals(phase) || TerminalPolicyStep.COMPLETED_PHASE.equals(phase)) {
|
||||
return TimeValue.ZERO;
|
||||
}
|
||||
final LifecyclePolicyMetadata meta = lifecyclePolicyMap.get(policy);
|
||||
if (meta == null) {
|
||||
throw new IllegalArgumentException("no policy found with name \"" + policy + "\"");
|
||||
} else {
|
||||
final Phase retrievedPhase = meta.getPolicy().getPhases().get(phase);
|
||||
if (retrievedPhase == null) {
|
||||
// We don't have that phase registered, proceed right through it
|
||||
return TimeValue.ZERO;
|
||||
} else {
|
||||
return retrievedPhase.getAfter();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,15 +15,12 @@ import org.elasticsearch.cluster.metadata.IndexMetaData;
|
|||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNode;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNodes;
|
||||
import org.elasticsearch.common.ParseField;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.transport.TransportAddress;
|
||||
import org.elasticsearch.common.unit.TimeValue;
|
||||
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
|
||||
import org.elasticsearch.index.Index;
|
||||
import org.elasticsearch.node.Node;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.OperationMode;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
|
|
@ -31,6 +28,7 @@ import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyMetadata;
|
|||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleSettings;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.MockAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.MockStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.OperationMode;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Phase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step.StepKey;
|
||||
|
|
@ -129,44 +127,7 @@ public class ExecuteStepsUpdateTaskTests extends ESTestCase {
|
|||
.metaData(metaData)
|
||||
.nodes(DiscoveryNodes.builder().localNodeId(nodeId).masterNodeId(nodeId).add(masterNode).build())
|
||||
.build();
|
||||
policyStepsRegistry.update(clusterState, client, () -> 0L);
|
||||
}
|
||||
|
||||
public void testExecuteAllUntilEndOfPhase() throws IOException {
|
||||
NamedXContentRegistry registry = new NamedXContentRegistry(
|
||||
Collections.singletonList(new NamedXContentRegistry.Entry(LifecycleAction.class, new ParseField(MockAction.NAME),
|
||||
(p) -> {
|
||||
MockAction.parse(p);
|
||||
return new MockAction(Arrays.asList(firstStep, allClusterSecondStep));
|
||||
})));
|
||||
policyStepsRegistry = new PolicyStepsRegistry(registry);
|
||||
setupIndexPolicy(allClusterPolicyName);
|
||||
|
||||
Step startStep = policyStepsRegistry.getFirstStep(allClusterPolicyName);
|
||||
long now = randomNonNegativeLong();
|
||||
// test execute start till end of phase `new`
|
||||
ExecuteStepsUpdateTask task = new ExecuteStepsUpdateTask(allClusterPolicyName, index, startStep, policyStepsRegistry, () -> now);
|
||||
ClusterState newState = task.execute(clusterState);
|
||||
|
||||
// Update the registry so the next phase's steps are loaded
|
||||
policyStepsRegistry.update(newState, client, () -> now);
|
||||
|
||||
// verify that both the `new` phase was executed and the next phase is to begin
|
||||
StepKey currentStepKey = IndexLifecycleRunner.getCurrentStepKey(newState.metaData().index(index).getSettings());
|
||||
assertThat(currentStepKey, equalTo(firstStep.getKey()));
|
||||
// test execute all actions in same phase
|
||||
task = new ExecuteStepsUpdateTask(allClusterPolicyName, index, firstStep, policyStepsRegistry, () -> now);
|
||||
newState = task.execute(newState);
|
||||
policyStepsRegistry.update(newState, client, () -> now);
|
||||
|
||||
assertThat(firstStep.getExecuteCount(), equalTo(1L));
|
||||
assertThat(allClusterSecondStep.getExecuteCount(), equalTo(1L));
|
||||
currentStepKey = IndexLifecycleRunner.getCurrentStepKey(newState.metaData().index(index).getSettings());
|
||||
assertThat(currentStepKey, equalTo(TerminalPolicyStep.KEY));
|
||||
assertNull(policyStepsRegistry.getStep(index, currentStepKey).getNextStepKey());
|
||||
assertThat(LifecycleSettings.LIFECYCLE_PHASE_TIME_SETTING.get(newState.metaData().index(index).getSettings()), equalTo(now));
|
||||
assertThat(LifecycleSettings.LIFECYCLE_ACTION_TIME_SETTING.get(newState.metaData().index(index).getSettings()), equalTo(now));
|
||||
assertThat(LifecycleSettings.LIFECYCLE_STEP_INFO_SETTING.get(newState.metaData().index(index).getSettings()), equalTo(""));
|
||||
policyStepsRegistry.update(clusterState, client);
|
||||
}
|
||||
|
||||
public void testNeverExecuteNonClusterStateStep() throws IOException {
|
||||
|
|
@ -201,7 +162,7 @@ public class ExecuteStepsUpdateTaskTests extends ESTestCase {
|
|||
.put(LifecycleSettings.LIFECYCLE_PHASE, (String) null)
|
||||
.put(LifecycleSettings.LIFECYCLE_ACTION, (String) null)
|
||||
.put(LifecycleSettings.LIFECYCLE_STEP, (String) null).build()))).build();
|
||||
policyStepsRegistry.update(clusterState, client, () -> 0);
|
||||
policyStepsRegistry.update(clusterState, client);
|
||||
|
||||
Step invalidStep = new MockClusterStateActionStep(firstStepKey, secondStepKey);
|
||||
long now = randomNonNegativeLong();
|
||||
|
|
@ -266,6 +227,6 @@ public class ExecuteStepsUpdateTaskTests extends ESTestCase {
|
|||
.put(LifecycleSettings.LIFECYCLE_PHASE, stepKey.getPhase())
|
||||
.put(LifecycleSettings.LIFECYCLE_ACTION, stepKey.getAction())
|
||||
.put(LifecycleSettings.LIFECYCLE_STEP, stepKey.getName()).build()))).build();
|
||||
policyStepsRegistry.update(clusterState, client, () -> 0);
|
||||
policyStepsRegistry.update(clusterState, client);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,8 +22,6 @@ import org.elasticsearch.common.xcontent.ToXContentObject;
|
|||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.json.JsonXContent;
|
||||
import org.elasticsearch.index.Index;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyTests;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.OperationMode;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.AbstractStepTestCase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.AsyncActionStep;
|
||||
|
|
@ -35,9 +33,11 @@ import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
|
|||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyMetadata;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicyTests;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleSettings;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.MockAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.MockStep;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.OperationMode;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Phase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.RolloverAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.Step;
|
||||
|
|
@ -54,6 +54,8 @@ import java.util.List;
|
|||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.SortedMap;
|
||||
import java.util.TreeMap;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
|
|
@ -1058,6 +1060,47 @@ public class IndexLifecycleRunnerTests extends ESTestCase {
|
|||
assertIndexNotManagedByILM(newClusterState, index);
|
||||
}
|
||||
|
||||
public void testIsReadyToTransition() {
|
||||
String policyName = "async_action_policy";
|
||||
StepKey stepKey = new StepKey("phase", MockAction.NAME, MockAction.NAME);
|
||||
MockAsyncActionStep step = new MockAsyncActionStep(stepKey, null);
|
||||
step.setWillComplete(true);
|
||||
SortedMap<String, LifecyclePolicyMetadata> lifecyclePolicyMap = new TreeMap<>(Collections.singletonMap(policyName,
|
||||
new LifecyclePolicyMetadata(createPolicy(policyName, null, step.getKey()), new HashMap<>())));
|
||||
Index index = new Index("my_index", "uuid");
|
||||
Map<String, Step> firstStepMap = Collections.singletonMap(policyName, step);
|
||||
Map<StepKey, Step> policySteps = Collections.singletonMap(step.getKey(), step);
|
||||
Map<String, Map<StepKey, Step>> stepMap = Collections.singletonMap(policyName, policySteps);
|
||||
Map<Index, List<Step>> indexSteps = Collections.singletonMap(index, Collections.singletonList(step));
|
||||
PolicyStepsRegistry policyStepsRegistry = new PolicyStepsRegistry(lifecyclePolicyMap, firstStepMap,
|
||||
stepMap, indexSteps, NamedXContentRegistry.EMPTY);
|
||||
ClusterService clusterService = mock(ClusterService.class);
|
||||
final AtomicLong now = new AtomicLong(5);
|
||||
IndexLifecycleRunner runner = new IndexLifecycleRunner(policyStepsRegistry, clusterService, now::get);
|
||||
IndexMetaData indexMetaData = IndexMetaData.builder("my_index").settings(settings(Version.CURRENT))
|
||||
.numberOfShards(randomIntBetween(1, 5))
|
||||
.numberOfReplicas(randomIntBetween(0, 5))
|
||||
.build();
|
||||
// With no time, always transition
|
||||
assertTrue("index should be able to transition with no creation date",
|
||||
runner.isReadyToTransitionToThisPhase(policyName, indexMetaData, "phase"));
|
||||
indexMetaData = IndexMetaData.builder(indexMetaData)
|
||||
.settings(Settings.builder()
|
||||
.put(indexMetaData.getSettings())
|
||||
.put(LifecycleSettings.LIFECYCLE_INDEX_CREATION_DATE, 10L)
|
||||
.build())
|
||||
.build();
|
||||
// Index is not old enough to transition
|
||||
assertFalse("index is not able to transition if it isn't old enough",
|
||||
runner.isReadyToTransitionToThisPhase(policyName, indexMetaData, "phase"));
|
||||
|
||||
// Set to the fuuuuuttuuuuuuurre
|
||||
now.set(Long.MAX_VALUE);
|
||||
assertTrue("index should be able to transition past phase's age",
|
||||
runner.isReadyToTransitionToThisPhase(policyName, indexMetaData, "phase"));
|
||||
}
|
||||
|
||||
|
||||
public static void assertIndexNotManagedByILM(ClusterState clusterState, Index index) {
|
||||
MetaData metadata = clusterState.metaData();
|
||||
assertNotNull(metadata);
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ public class MoveToNextStepUpdateTaskTests extends ESTestCase {
|
|||
|
||||
public void testExecuteSuccessfullyMoved() {
|
||||
long now = randomNonNegativeLong();
|
||||
List<Step> steps = lifecyclePolicy.toSteps(null, () -> now);
|
||||
List<Step> steps = lifecyclePolicy.toSteps(null);
|
||||
StepKey currentStepKey = steps.get(0).getKey();
|
||||
StepKey nextStepKey = steps.get(0).getNextStepKey();
|
||||
|
||||
|
|
@ -103,7 +103,7 @@ public class MoveToNextStepUpdateTaskTests extends ESTestCase {
|
|||
|
||||
public void testExecuteSuccessfulMoveWithInvalidNextStep() {
|
||||
long now = randomNonNegativeLong();
|
||||
List<Step> steps = lifecyclePolicy.toSteps(null, () -> now);
|
||||
List<Step> steps = lifecyclePolicy.toSteps(null);
|
||||
StepKey currentStepKey = steps.get(0).getKey();
|
||||
StepKey invalidNextStep = new StepKey("next-invalid", "next-invalid", "next-invalid");
|
||||
|
||||
|
|
|
|||
|
|
@ -109,7 +109,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
String policyName = randomAlphaOfLength(5);
|
||||
LifecyclePolicy newPolicy = LifecyclePolicyTests.randomTestLifecyclePolicy(policyName);
|
||||
logger.info("--> policy: {}", newPolicy);
|
||||
List<Step> policySteps = newPolicy.toSteps(client, () -> 0L);
|
||||
List<Step> policySteps = newPolicy.toSteps(client);
|
||||
Map<String, String> headers = new HashMap<>();
|
||||
if (randomBoolean()) {
|
||||
headers.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
|
||||
|
|
@ -149,7 +149,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
PolicyStepsRegistry registry = new PolicyStepsRegistry(NamedXContentRegistry.EMPTY);
|
||||
|
||||
// add new policy
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
registry.update(currentState, client);
|
||||
|
||||
assertThat(registry.getFirstStep(newPolicy.getName()), equalTo(policySteps.get(0)));
|
||||
assertThat(registry.getLifecyclePolicyMap().size(), equalTo(1));
|
||||
|
|
@ -167,7 +167,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
.put(LifecycleSettings.LIFECYCLE_PHASE, step.getKey().getPhase()))))
|
||||
.nodes(DiscoveryNodes.builder().localNodeId(nodeId).masterNodeId(nodeId).add(masterNode).build())
|
||||
.build();
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
registry.update(currentState, client);
|
||||
assertThat(registeredStepsForPolicy.get(step.getKey()), equalTo(step));
|
||||
assertThat(registry.getStep(index, step.getKey()), equalTo(step));
|
||||
}
|
||||
|
|
@ -175,7 +175,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
Map<String, LifecyclePolicyMetadata> registryPolicyMap = registry.getLifecyclePolicyMap();
|
||||
Map<String, Step> registryFirstStepMap = registry.getFirstStepMap();
|
||||
Map<String, Map<Step.StepKey, Step>> registryStepMap = registry.getStepMap();
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
registry.update(currentState, client);
|
||||
assertThat(registry.getLifecyclePolicyMap(), equalTo(registryPolicyMap));
|
||||
assertThat(registry.getFirstStepMap(), equalTo(registryFirstStepMap));
|
||||
assertThat(registry.getStepMap(), equalTo(registryStepMap));
|
||||
|
|
@ -186,7 +186,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
.metaData(
|
||||
MetaData.builder(metaData)
|
||||
.putCustom(IndexLifecycleMetadata.TYPE, lifecycleMetadata)).build();
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
registry.update(currentState, client);
|
||||
assertTrue(registry.getLifecyclePolicyMap().isEmpty());
|
||||
assertTrue(registry.getFirstStepMap().isEmpty());
|
||||
assertTrue(registry.getStepMap().isEmpty());
|
||||
|
|
@ -219,7 +219,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
.build();
|
||||
PolicyStepsRegistry registry = new PolicyStepsRegistry(NamedXContentRegistry.EMPTY);
|
||||
// add new policy
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
registry.update(currentState, client);
|
||||
|
||||
// swap out policy
|
||||
newPolicy = LifecyclePolicyTests.randomTestLifecyclePolicy(policyName);
|
||||
|
|
@ -227,7 +227,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
new LifecyclePolicyMetadata(newPolicy, Collections.emptyMap())), OperationMode.RUNNING);
|
||||
currentState = ClusterState.builder(currentState)
|
||||
.metaData(MetaData.builder(metaData).putCustom(IndexLifecycleMetadata.TYPE, lifecycleMetadata)).build();
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
registry.update(currentState, client);
|
||||
// TODO(talevy): assert changes... right now we do not support updates to policies. will require internal cleanup
|
||||
}
|
||||
|
||||
|
|
@ -249,7 +249,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
LifecyclePolicy updatedPolicy = new LifecyclePolicy(policyName, phases);
|
||||
logger.info("--> policy: {}", newPolicy);
|
||||
logger.info("--> updated policy: {}", updatedPolicy);
|
||||
List<Step> policySteps = newPolicy.toSteps(client, () -> 0L);
|
||||
List<Step> policySteps = newPolicy.toSteps(client);
|
||||
Map<String, String> headers = new HashMap<>();
|
||||
if (randomBoolean()) {
|
||||
headers.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
|
||||
|
|
@ -289,7 +289,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
PolicyStepsRegistry registry = new PolicyStepsRegistry(NamedXContentRegistry.EMPTY);
|
||||
|
||||
// add new policy
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
registry.update(currentState, client);
|
||||
|
||||
Map<Step.StepKey, Step> registeredStepsForPolicy = registry.getStepMap().get(newPolicy.getName());
|
||||
Step shrinkStep = registeredStepsForPolicy.entrySet().stream()
|
||||
|
|
@ -314,7 +314,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
|
|||
currentState = ClusterState.builder(ClusterName.DEFAULT).metaData(metaData).build();
|
||||
|
||||
// Update the policies
|
||||
registry.update(currentState, client, () -> 0L);
|
||||
registry.update(currentState, client);
|
||||
|
||||
registeredStepsForPolicy = registry.getStepMap().get(newPolicy.getName());
|
||||
shrinkStep = registeredStepsForPolicy.entrySet().stream()
|
||||
|
|
|
|||
|
|
@ -76,8 +76,8 @@ teardown:
|
|||
body:
|
||||
current_step:
|
||||
phase: "new"
|
||||
action: "after"
|
||||
name: "after"
|
||||
action: "complete"
|
||||
name: "complete"
|
||||
next_step:
|
||||
phase: "warm"
|
||||
action: "forcemerge"
|
||||
|
|
@ -116,8 +116,8 @@ teardown:
|
|||
indices.get:
|
||||
index: my_index
|
||||
- match: { my_index.settings.index.lifecycle.name: "my_moveable_timeseries_lifecycle" }
|
||||
- match: { my_index.settings.index.lifecycle.step: "after" }
|
||||
- match: { my_index.settings.index.lifecycle.action: "after" }
|
||||
- match: { my_index.settings.index.lifecycle.step: "complete" }
|
||||
- match: { my_index.settings.index.lifecycle.action: "complete" }
|
||||
- match: { my_index.settings.index.lifecycle.phase: "new" }
|
||||
|
||||
---
|
||||
|
|
@ -130,8 +130,8 @@ teardown:
|
|||
body:
|
||||
current_step:
|
||||
phase: "new"
|
||||
action: "after"
|
||||
name: "after"
|
||||
action: "complete"
|
||||
name: "complete"
|
||||
next_step:
|
||||
phase: "invalid"
|
||||
action: "invalid"
|
||||
|
|
@ -144,8 +144,8 @@ teardown:
|
|||
indices.get:
|
||||
index: my_index
|
||||
- match: { my_index.settings.index.lifecycle.name: "my_moveable_timeseries_lifecycle" }
|
||||
- match: { my_index.settings.index.lifecycle.step: "after" }
|
||||
- match: { my_index.settings.index.lifecycle.action: "after" }
|
||||
- match: { my_index.settings.index.lifecycle.step: "complete" }
|
||||
- match: { my_index.settings.index.lifecycle.action: "complete" }
|
||||
- match: { my_index.settings.index.lifecycle.phase: "new" }
|
||||
|
||||
---
|
||||
|
|
|
|||
|
|
@ -73,8 +73,8 @@ teardown:
|
|||
indices.get:
|
||||
index: my_index
|
||||
- match: { my_index.settings.index.lifecycle.name: "my_lifecycle" }
|
||||
- match: { my_index.settings.index.lifecycle.step: "after" }
|
||||
- match: { my_index.settings.index.lifecycle.action: "after" }
|
||||
- match: { my_index.settings.index.lifecycle.step: "complete" }
|
||||
- match: { my_index.settings.index.lifecycle.action: "complete" }
|
||||
- match: { my_index.settings.index.lifecycle.phase: "new" }
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -110,8 +110,8 @@ teardown:
|
|||
- match: { indices.my_index.index: "my_index" }
|
||||
- match: { indices.my_index.policy: "my_moveable_timeseries_lifecycle" }
|
||||
- match: { indices.my_index.phase: "new" }
|
||||
- match: { indices.my_index.action: "after" }
|
||||
- match: { indices.my_index.step: "after" }
|
||||
- match: { indices.my_index.action: "complete" }
|
||||
- match: { indices.my_index.step: "complete" }
|
||||
- is_false: indices.my_index.failed_step
|
||||
- is_false: indices.my_index.step_info
|
||||
|
||||
|
|
@ -131,8 +131,8 @@ teardown:
|
|||
- match: { indices.my_index.index: "my_index" }
|
||||
- match: { indices.my_index.policy: "my_moveable_timeseries_lifecycle" }
|
||||
- match: { indices.my_index.phase: "new" }
|
||||
- match: { indices.my_index.action: "after" }
|
||||
- match: { indices.my_index.step: "after" }
|
||||
- match: { indices.my_index.action: "complete" }
|
||||
- match: { indices.my_index.step: "complete" }
|
||||
- is_false: indices.my_index.failed_step
|
||||
- is_false: indices.my_index.step_info
|
||||
|
||||
|
|
@ -140,8 +140,8 @@ teardown:
|
|||
- match: { indices.my_index2.index: "my_index2" }
|
||||
- match: { indices.my_index2.policy: "my_moveable_timeseries_lifecycle" }
|
||||
- match: { indices.my_index2.phase: "new" }
|
||||
- match: { indices.my_index2.action: "after" }
|
||||
- match: { indices.my_index2.step: "after" }
|
||||
- match: { indices.my_index2.action: "complete" }
|
||||
- match: { indices.my_index2.step: "complete" }
|
||||
- is_false: indices.my_index2.failed_step
|
||||
- is_false: indices.my_index2.step_info
|
||||
|
||||
|
|
@ -161,8 +161,8 @@ teardown:
|
|||
- match: { indices.my_index.index: "my_index" }
|
||||
- match: { indices.my_index.policy: "my_moveable_timeseries_lifecycle" }
|
||||
- match: { indices.my_index.phase: "new" }
|
||||
- match: { indices.my_index.action: "after" }
|
||||
- match: { indices.my_index.step: "after" }
|
||||
- match: { indices.my_index.action: "complete" }
|
||||
- match: { indices.my_index.step: "complete" }
|
||||
- is_false: indices.my_index.failed_step
|
||||
- is_false: indices.my_index.step_info
|
||||
|
||||
|
|
@ -170,8 +170,8 @@ teardown:
|
|||
- match: { indices.my_index2.index: "my_index2" }
|
||||
- match: { indices.my_index2.policy: "my_moveable_timeseries_lifecycle" }
|
||||
- match: { indices.my_index2.phase: "new" }
|
||||
- match: { indices.my_index2.action: "after" }
|
||||
- match: { indices.my_index2.step: "after" }
|
||||
- match: { indices.my_index2.action: "complete" }
|
||||
- match: { indices.my_index2.step: "complete" }
|
||||
- is_false: indices.my_index2.failed_step
|
||||
- is_false: indices.my_index2.step_info
|
||||
|
||||
|
|
@ -179,8 +179,8 @@ teardown:
|
|||
- match: { indices.another_index.index: "another_index" }
|
||||
- match: { indices.another_index.policy: "my_moveable_timeseries_lifecycle" }
|
||||
- match: { indices.another_index.phase: "new" }
|
||||
- match: { indices.another_index.action: "after" }
|
||||
- match: { indices.another_index.step: "after" }
|
||||
- match: { indices.another_index.action: "complete" }
|
||||
- match: { indices.another_index.step: "complete" }
|
||||
- is_false: indices.another_index.failed_step
|
||||
- is_false: indices.another_index.step_info
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue