Skip to content

Commit

Permalink
Don't halt policy execution on policy trigger exception (elastic#49128)
Browse files Browse the repository at this point in the history 
When triggered either by becoming master, a new cluster state, or a
periodic schedule, an ILM policy execution through
`maybeRunAsyncAction`, `runPolicyAfterStateChange`, or
`runPeriodicStep` throwing an exception will cause the loop the
terminate. This means that any indices that would have been processed
after the index where the exception was thrown will not be processed by
ILM.

For most execution this is not a problem because the actual running of
steps is protected by a try/catch that moves the index to the ERROR step
in the event of a problem. If an exception occurs prior to step
execution (for example, in fetching and parsing the current
policy/step) however, it causes the loop termination previously
mentioned.

This commit wraps the invocation of the methods specified above in a
try/catch block that provides better logging and does not bubble the
exception up.
  • Loading branch information
@dakrone
dakrone committed Nov 15, 2019
1 parent 79f504e commit 6181a42
Show file tree
Hide file tree
Showing 3 changed files with 164 additions and 25 deletions.
78 changes: 55 additions & 23 deletions x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycleService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import com.carrotsearch.hppc.cursors.ObjectCursor;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.lucene.util.SetOnce;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterChangedEvent;
Expand Down Expand Up @@ -119,19 +120,35 @@ public void onMaster() {
final LifecycleExecutionState lifecycleState = LifecycleExecutionState.fromIndexMetadata(idxMeta);
StepKey stepKey = IndexLifecycleRunner.getCurrentStepKey(lifecycleState);

if (OperationMode.STOPPING == currentMode) {
if (stepKey != null && IGNORE_STEPS_MAINTENANCE_REQUESTED.contains(stepKey.getName())) {
logger.info("waiting to stop ILM because index [{}] with policy [{}] is currently in step [{}]",
idxMeta.getIndex().getName(), policyName, stepKey.getName());
try {
if (OperationMode.STOPPING == currentMode) {
if (stepKey != null && IGNORE_STEPS_MAINTENANCE_REQUESTED.contains(stepKey.getName())) {
logger.info("waiting to stop ILM because index [{}] with policy [{}] is currently in step [{}]",
idxMeta.getIndex().getName(), policyName, stepKey.getName());
lifecycleRunner.maybeRunAsyncAction(clusterState, idxMeta, policyName, stepKey);
// ILM is trying to stop, but this index is in a Shrink step (or other dangerous step) so we can't stop
safeToStop = false;
} else {
logger.info("skipping policy execution of step [{}] for index [{}] with policy [{}]" +
" because ILM is stopping",
stepKey == null ? "n/a" : stepKey.getName(), idxMeta.getIndex().getName(), policyName);
}
} else {
lifecycleRunner.maybeRunAsyncAction(clusterState, idxMeta, policyName, stepKey);
// ILM is trying to stop, but this index is in a Shrink step (or other dangerous step) so we can't stop
safeToStop = false;
}
} catch (Exception e) {
if (logger.isTraceEnabled()) {
logger.warn(new ParameterizedMessage("async action execution failed during master election trigger" +
" for index [{}] with policy [{}] in step [{}], lifecycle state: [{}]",
idxMeta.getIndex().getName(), policyName, stepKey, lifecycleState.asMap()), e);
} else {
logger.info("skipping policy execution of step [{}] for index [{}] with policy [{}] because ILM is stopping",
stepKey == null ? "n/a" : stepKey.getName(), idxMeta.getIndex().getName(), policyName);
logger.warn(new ParameterizedMessage("async action execution failed during master election trigger" +
" for index [{}] with policy [{}] in step [{}]",
idxMeta.getIndex().getName(), policyName, stepKey), e);

}
} else {
lifecycleRunner.maybeRunAsyncAction(clusterState, idxMeta, policyName, stepKey);
// Don't rethrow the exception, we don't want a failure for one index to be
// called to cause actions not to be triggered for further indices
}
}
}
Expand Down Expand Up @@ -264,27 +281,42 @@ void triggerPolicies(ClusterState clusterState, boolean fromClusterStateChange)
final LifecycleExecutionState lifecycleState = LifecycleExecutionState.fromIndexMetadata(idxMeta);
StepKey stepKey = IndexLifecycleRunner.getCurrentStepKey(lifecycleState);

if (OperationMode.STOPPING == currentMode) {
if (stepKey != null && IGNORE_STEPS_MAINTENANCE_REQUESTED.contains(stepKey.getName())) {
logger.info("waiting to stop ILM because index [{}] with policy [{}] is currently in step [{}]",
idxMeta.getIndex().getName(), policyName, stepKey.getName());
try {
if (OperationMode.STOPPING == currentMode) {
if (stepKey != null && IGNORE_STEPS_MAINTENANCE_REQUESTED.contains(stepKey.getName())) {
logger.info("waiting to stop ILM because index [{}] with policy [{}] is currently in step [{}]",
idxMeta.getIndex().getName(), policyName, stepKey.getName());
if (fromClusterStateChange) {
lifecycleRunner.runPolicyAfterStateChange(policyName, idxMeta);
} else {
lifecycleRunner.runPeriodicStep(policyName, idxMeta);
}
// ILM is trying to stop, but this index is in a Shrink step (or other dangerous step) so we can't stop
safeToStop = false;
} else {
logger.info("skipping policy execution of step [{}] for index [{}] with policy [{}] because ILM is stopping",
stepKey == null ? "n/a" : stepKey.getName(), idxMeta.getIndex().getName(), policyName);
}
} else {
if (fromClusterStateChange) {
lifecycleRunner.runPolicyAfterStateChange(policyName, idxMeta);
} else {
lifecycleRunner.runPeriodicStep(policyName, idxMeta);
}
// ILM is trying to stop, but this index is in a Shrink step (or other dangerous step) so we can't stop
safeToStop = false;
} else {
logger.info("skipping policy execution of step [{}] for index [{}] with policy [{}] because ILM is stopping",
stepKey == null ? "n/a" : stepKey.getName(), idxMeta.getIndex().getName(), policyName);
}
} else {
if (fromClusterStateChange) {
lifecycleRunner.runPolicyAfterStateChange(policyName, idxMeta);
} catch (Exception e) {
if (logger.isTraceEnabled()) {
logger.warn(new ParameterizedMessage("async action execution failed during policy trigger" +
" for index [{}] with policy [{}] in step [{}], lifecycle state: [{}]",
idxMeta.getIndex().getName(), policyName, stepKey, lifecycleState.asMap()), e);
} else {
lifecycleRunner.runPeriodicStep(policyName, idxMeta);
logger.warn(new ParameterizedMessage("async action execution failed during policy trigger" +
" for index [{}] with policy [{}] in step [{}]",
idxMeta.getIndex().getName(), policyName, stepKey), e);

}
// Don't rethrow the exception, we don't want a failure for one index to be
// called to cause actions not to be triggered for further indices
}
}
}
Expand Down
4 changes: 2 additions & 2 deletions x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/ilm/IndexLifecycleRunnerTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1350,7 +1350,7 @@ private void assertClusterStateStepInfo(ClusterState oldClusterState, Index inde
assertEquals(newLifecycleState.getStepTime(), newLifecycleState.getStepTime());
}

private static class MockAsyncActionStep extends AsyncActionStep {
static class MockAsyncActionStep extends AsyncActionStep {

private Exception exception;
private boolean willComplete;
Expand Down Expand Up @@ -1399,7 +1399,7 @@ public void performAction(IndexMetaData indexMetaData, ClusterState currentState

}

private static class MockAsyncWaitStep extends AsyncWaitStep {
static class MockAsyncWaitStep extends AsyncWaitStep {

private Exception exception;
private boolean willComplete;
Expand Down
107 changes: 107 additions & 0 deletions x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/ilm/IndexLifecycleServiceTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,9 @@
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import static org.elasticsearch.node.Node.NODE_MASTER_SETTING;
import static org.elasticsearch.xpack.core.ilm.AbstractStepTestCase.randomStepKey;
Expand Down Expand Up @@ -300,6 +302,111 @@ public void testRequestedStopOnSafeAction() {
assertTrue(moveToMaintenance.get());
}

public void testExceptionStillProcessesOtherIndices() {
doTestExceptionStillProcessesOtherIndices(false);
}

public void testExceptionStillProcessesOtherIndicesOnMaster() {
doTestExceptionStillProcessesOtherIndices(true);
}

@SuppressWarnings("unchecked")
public void doTestExceptionStillProcessesOtherIndices(boolean useOnMaster) {
String policy1 = randomAlphaOfLengthBetween(1, 20);
Step.StepKey i1currentStepKey = randomStepKey();
final Step i1mockStep;
if (useOnMaster) {
i1mockStep = new IndexLifecycleRunnerTests.MockAsyncActionStep(i1currentStepKey, randomStepKey());
} else {
i1mockStep = new IndexLifecycleRunnerTests.MockClusterStateActionStep(i1currentStepKey, randomStepKey());
}
MockAction i1mockAction = new MockAction(Collections.singletonList(i1mockStep));
Phase i1phase = new Phase("phase", TimeValue.ZERO, Collections.singletonMap("action", i1mockAction));
LifecyclePolicy i1policy = newTestLifecyclePolicy(policy1, Collections.singletonMap(i1phase.getName(), i1phase));
Index index1 = new Index(randomAlphaOfLengthBetween(1, 20), randomAlphaOfLengthBetween(1, 20));
LifecycleExecutionState.Builder i1lifecycleState = LifecycleExecutionState.builder();
i1lifecycleState.setPhase(i1currentStepKey.getPhase());
i1lifecycleState.setAction(i1currentStepKey.getAction());
i1lifecycleState.setStep(i1currentStepKey.getName());

String policy2 = randomValueOtherThan(policy1, () -> randomAlphaOfLengthBetween(1, 20));
Step.StepKey i2currentStepKey = randomStepKey();
final Step i2mockStep;
if (useOnMaster) {
i2mockStep = new IndexLifecycleRunnerTests.MockAsyncActionStep(i2currentStepKey, randomStepKey());
} else {
i2mockStep = new IndexLifecycleRunnerTests.MockClusterStateActionStep(i2currentStepKey, randomStepKey());
}
MockAction mockAction = new MockAction(Collections.singletonList(i2mockStep));
Phase i2phase = new Phase("phase", TimeValue.ZERO, Collections.singletonMap("action", mockAction));
LifecyclePolicy i2policy = newTestLifecyclePolicy(policy1, Collections.singletonMap(i2phase.getName(), i1phase));
Index index2 = new Index(randomAlphaOfLengthBetween(1, 20), randomAlphaOfLengthBetween(1, 20));
LifecycleExecutionState.Builder i2lifecycleState = LifecycleExecutionState.builder();
i2lifecycleState.setPhase(i2currentStepKey.getPhase());
i2lifecycleState.setAction(i2currentStepKey.getAction());
i2lifecycleState.setStep(i2currentStepKey.getName());

CountDownLatch stepLatch = new CountDownLatch(2);
boolean failStep1 = randomBoolean();
if (useOnMaster) {
((IndexLifecycleRunnerTests.MockAsyncActionStep) i1mockStep).setLatch(stepLatch);
((IndexLifecycleRunnerTests.MockAsyncActionStep) i1mockStep)
.setException(failStep1 ? new IllegalArgumentException("forcing a failure for index 1") : null);
((IndexLifecycleRunnerTests.MockAsyncActionStep) i2mockStep).setLatch(stepLatch);
((IndexLifecycleRunnerTests.MockAsyncActionStep) i2mockStep)
.setException(failStep1 ? null : new IllegalArgumentException("forcing a failure for index 2"));
} else {
((IndexLifecycleRunnerTests.MockClusterStateActionStep) i1mockStep).setLatch(stepLatch);
((IndexLifecycleRunnerTests.MockClusterStateActionStep) i1mockStep)
.setException(failStep1 ? new IllegalArgumentException("forcing a failure for index 1") : null);
((IndexLifecycleRunnerTests.MockClusterStateActionStep) i1mockStep).setLatch(stepLatch);
((IndexLifecycleRunnerTests.MockClusterStateActionStep) i1mockStep)
.setException(failStep1 ? null : new IllegalArgumentException("forcing a failure for index 2"));
}

SortedMap<String, LifecyclePolicyMetadata> policyMap = new TreeMap<>();
policyMap.put(policy1, new LifecyclePolicyMetadata(i1policy, Collections.emptyMap(),
randomNonNegativeLong(), randomNonNegativeLong()));
policyMap.put(policy2, new LifecyclePolicyMetadata(i2policy, Collections.emptyMap(),
randomNonNegativeLong(), randomNonNegativeLong()));

IndexMetaData i1indexMetadata = IndexMetaData.builder(index1.getName())
.settings(settings(Version.CURRENT).put(LifecycleSettings.LIFECYCLE_NAME_SETTING.getKey(), policy1))
.putCustom(ILM_CUSTOM_METADATA_KEY, i1lifecycleState.build().asMap())
.numberOfShards(randomIntBetween(1, 5)).numberOfReplicas(randomIntBetween(0, 5)).build();
IndexMetaData i2indexMetadata = IndexMetaData.builder(index2.getName())
.settings(settings(Version.CURRENT).put(LifecycleSettings.LIFECYCLE_NAME_SETTING.getKey(), policy1))
.putCustom(ILM_CUSTOM_METADATA_KEY, i2lifecycleState.build().asMap())
.numberOfShards(randomIntBetween(1, 5)).numberOfReplicas(randomIntBetween(0, 5)).build();
ImmutableOpenMap.Builder<String, IndexMetaData> indices = ImmutableOpenMap.<String, IndexMetaData> builder()
.fPut(index1.getName(), i1indexMetadata)
.fPut(index2.getName(), i2indexMetadata);

MetaData metaData = MetaData.builder()
.putCustom(IndexLifecycleMetadata.TYPE, new IndexLifecycleMetadata(policyMap, OperationMode.RUNNING))
.indices(indices.build())
.persistentSettings(settings(Version.CURRENT).build())
.build();

ClusterState currentState = ClusterState.builder(ClusterName.DEFAULT)
.metaData(metaData)
.nodes(DiscoveryNodes.builder().localNodeId(nodeId).masterNodeId(nodeId).add(masterNode).build())
.build();

if (useOnMaster) {
when(clusterService.state()).thenReturn(currentState);
indexLifecycleService.onMaster();
} else {
indexLifecycleService.triggerPolicies(currentState, randomBoolean());
}
try {
stepLatch.await(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
logger.error("failure while waiting for step execution", e);
fail("both steps should have been executed, even with an exception");
}
}

public void testTriggeredDifferentJob() {
Mockito.reset(clusterService);
SchedulerEngine.Event schedulerEvent = new SchedulerEngine.Event("foo", randomLong(), randomLong());
Expand Down

0 comments on commit 6181a42

Please sign in to comment.