Skip to content

Commit 3c971f2

Browse files
andreidanandreidan
authored
ILM retryable async action steps (#50522) (#50591)
This adds support for retrying AsyncActionSteps by triggering the async step after ILM was moved back on the failed step (the async step we'll be attempting to run after the cluster state reflects ILM being moved back on the failed step). This also marks the RolloverStep as retryable and adds an integration test where the RolloverStep is failing to execute as the rolled over index already exists to test that the async action RolloverStep is retried until the rolled over index is deleted. (cherry picked from commit 8bee5f4) Signed-off-by: Andrei Dan <[email protected]>
1 parent 0bcbdde commit 3c971f2

File tree

3 files changed

+155
-63
lines changed

3 files changed

+155
-63
lines changed

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/RolloverStep.java

+5
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,11 @@ public RolloverStep(StepKey key, StepKey nextStepKey, Client client) {
3030
super(key, nextStepKey, client);
3131
}
3232

33+
@Override
34+
public boolean isRetryable() {
35+
return true;
36+
}
37+
3338
@Override
3439
public void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState,
3540
ClusterStateObserver observer, Listener listener) {

‎x-pack/plugin/ilm/qa/multi-node/src/test/java/org/elasticsearch/xpack/ilm/TimeSeriesLifecycleActionsIT.java

+136-63
Original file line numberDiff line numberDiff line change
@@ -949,69 +949,142 @@ public void testILMRolloverRetriesOnReadOnlyBlock() throws Exception {
949949
assertBusy(() -> assertThat(getStepKeyForIndex(firstIndex), equalTo(TerminalPolicyStep.KEY)));
950950
}
951951

952-
public void testILMRolloverOnManuallyRolledIndex() throws Exception {
953-
String originalIndex = index + "-000001";
954-
String secondIndex = index + "-000002";
955-
String thirdIndex = index + "-000003";
956-
957-
// Set up a policy with rollover
958-
createNewSingletonPolicy("hot", new RolloverAction(null, null, 2L));
959-
Request createIndexTemplate = new Request("PUT", "_template/rolling_indexes");
960-
createIndexTemplate.setJsonEntity("{" +
961-
"\"index_patterns\": [\""+ index + "-*\"], \n" +
962-
" \"settings\": {\n" +
963-
" \"number_of_shards\": 1,\n" +
964-
" \"number_of_replicas\": 0,\n" +
965-
" \"index.lifecycle.name\": \"" + policy+ "\", \n" +
966-
" \"index.lifecycle.rollover_alias\": \"alias\"\n" +
967-
" }\n" +
968-
"}");
969-
client().performRequest(createIndexTemplate);
970-
971-
createIndexWithSettings(
972-
originalIndex,
973-
Settings.builder().put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
974-
.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0),
975-
true
976-
);
977-
978-
// Index a document
979-
index(client(), originalIndex, "1", "foo", "bar");
980-
Request refreshOriginalIndex = new Request("POST", "/" + originalIndex + "/_refresh");
981-
client().performRequest(refreshOriginalIndex);
982-
983-
// Manual rollover
984-
Request rolloverRequest = new Request("POST", "/alias/_rollover");
985-
rolloverRequest.setJsonEntity("{\n" +
986-
" \"conditions\": {\n" +
987-
" \"max_docs\": \"1\"\n" +
988-
" }\n" +
989-
"}"
990-
);
991-
client().performRequest(rolloverRequest);
992-
assertBusy(() -> assertTrue(indexExists(secondIndex)));
993-
994-
// Index another document into the original index so the ILM rollover policy condition is met
995-
index(client(), originalIndex, "2", "foo", "bar");
996-
client().performRequest(refreshOriginalIndex);
997-
998-
// Wait for the rollover policy to execute
999-
assertBusy(() -> assertThat(getStepKeyForIndex(originalIndex), equalTo(TerminalPolicyStep.KEY)));
1000-
1001-
// ILM should manage the second index after attempting (and skipping) rolling the original index
1002-
assertBusy(() -> assertTrue((boolean) explainIndex(secondIndex).getOrDefault("managed", true)));
1003-
1004-
// index some documents to trigger an ILM rollover
1005-
index(client(), "alias", "1", "foo", "bar");
1006-
index(client(), "alias", "2", "foo", "bar");
1007-
index(client(), "alias", "3", "foo", "bar");
1008-
Request refreshSecondIndex = new Request("POST", "/" + secondIndex + "/_refresh");
1009-
client().performRequest(refreshSecondIndex).getStatusLine();
1010-
1011-
// ILM should rollover the second index even though it skipped the first one
1012-
assertBusy(() -> assertThat(getStepKeyForIndex(secondIndex), equalTo(TerminalPolicyStep.KEY)));
1013-
assertBusy(() -> assertTrue(indexExists(thirdIndex)));
1014-
}
952+
public void testILMRolloverOnManuallyRolledIndex() throws Exception {
953+
String originalIndex = index + "-000001";
954+
String secondIndex = index + "-000002";
955+
String thirdIndex = index + "-000003";
956+
957+
// Set up a policy with rollover
958+
createNewSingletonPolicy("hot", new RolloverAction(null, null, 2L));
959+
Request createIndexTemplate = new Request("PUT", "_template/rolling_indexes");
960+
createIndexTemplate.setJsonEntity("{" +
961+
"\"index_patterns\": [\"" + index + "-*\"], \n" +
962+
" \"settings\": {\n" +
963+
" \"number_of_shards\": 1,\n" +
964+
" \"number_of_replicas\": 0,\n" +
965+
" \"index.lifecycle.name\": \"" + policy + "\", \n" +
966+
" \"index.lifecycle.rollover_alias\": \"alias\"\n" +
967+
" }\n" +
968+
"}");
969+
client().performRequest(createIndexTemplate);
970+
971+
createIndexWithSettings(
972+
originalIndex,
973+
Settings.builder().put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
974+
.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0),
975+
true
976+
);
977+
978+
// Index a document
979+
index(client(), originalIndex, "1", "foo", "bar");
980+
Request refreshOriginalIndex = new Request("POST", "/" + originalIndex + "/_refresh");
981+
client().performRequest(refreshOriginalIndex);
982+
983+
// Manual rollover
984+
Request rolloverRequest = new Request("POST", "/alias/_rollover");
985+
rolloverRequest.setJsonEntity("{\n" +
986+
" \"conditions\": {\n" +
987+
" \"max_docs\": \"1\"\n" +
988+
" }\n" +
989+
"}"
990+
);
991+
client().performRequest(rolloverRequest);
992+
assertBusy(() -> assertTrue(indexExists(secondIndex)));
993+
994+
// Index another document into the original index so the ILM rollover policy condition is met
995+
index(client(), originalIndex, "2", "foo", "bar");
996+
client().performRequest(refreshOriginalIndex);
997+
998+
// Wait for the rollover policy to execute
999+
assertBusy(() -> assertThat(getStepKeyForIndex(originalIndex), equalTo(TerminalPolicyStep.KEY)));
1000+
1001+
// ILM should manage the second index after attempting (and skipping) rolling the original index
1002+
assertBusy(() -> assertTrue((boolean) explainIndex(secondIndex).getOrDefault("managed", true)));
1003+
1004+
// index some documents to trigger an ILM rollover
1005+
index(client(), "alias", "1", "foo", "bar");
1006+
index(client(), "alias", "2", "foo", "bar");
1007+
index(client(), "alias", "3", "foo", "bar");
1008+
Request refreshSecondIndex = new Request("POST", "/" + secondIndex + "/_refresh");
1009+
client().performRequest(refreshSecondIndex).getStatusLine();
1010+
1011+
// ILM should rollover the second index even though it skipped the first one
1012+
assertBusy(() -> assertThat(getStepKeyForIndex(secondIndex), equalTo(TerminalPolicyStep.KEY)));
1013+
assertBusy(() -> assertTrue(indexExists(thirdIndex)));
1014+
}
1015+
1016+
public void testRolloverStepRetriesUntilRolledOverIndexIsDeleted() throws Exception {
1017+
String index = this.index + "-000001";
1018+
String rolledIndex = this.index + "-000002";
1019+
1020+
createNewSingletonPolicy("hot", new RolloverAction(null, TimeValue.timeValueSeconds(1), null));
1021+
1022+
// create the rolled index so the rollover of the first index fails
1023+
createIndexWithSettings(
1024+
rolledIndex,
1025+
Settings.builder().put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
1026+
.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
1027+
.put(RolloverAction.LIFECYCLE_ROLLOVER_ALIAS, "alias"),
1028+
false
1029+
);
1030+
1031+
createIndexWithSettings(
1032+
index,
1033+
Settings.builder().put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
1034+
.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
1035+
.put(LifecycleSettings.LIFECYCLE_NAME, policy)
1036+
.put(RolloverAction.LIFECYCLE_ROLLOVER_ALIAS, "alias"),
1037+
true
1038+
);
1039+
1040+
assertBusy(() -> assertThat((Integer) explainIndex(index).get(FAILED_STEP_RETRY_COUNT_FIELD), greaterThanOrEqualTo(1)), 30,
1041+
TimeUnit.SECONDS);
1042+
1043+
Request moveToStepRequest = new Request("POST", "_ilm/move/" + index);
1044+
moveToStepRequest.setJsonEntity("{\n" +
1045+
" \"current_step\": {\n" +
1046+
" \"phase\": \"hot\",\n" +
1047+
" \"action\": \"rollover\",\n" +
1048+
" \"name\": \"check-rollover-ready\"\n" +
1049+
" },\n" +
1050+
" \"next_step\": {\n" +
1051+
" \"phase\": \"hot\",\n" +
1052+
" \"action\": \"rollover\",\n" +
1053+
" \"name\": \"attempt-rollover\"\n" +
1054+
" }\n" +
1055+
"}");
1056+
1057+
// Using {@link #waitUntil} here as ILM moves back and forth between the {@link WaitForRolloverReadyStep} step and
1058+
// {@link org.elasticsearch.xpack.core.ilm.ErrorStep} in order to retry the failing step. As {@link #assertBusy}
1059+
// increases the wait time between calls exponentially, we might miss the window where the policy is on
1060+
// {@link WaitForRolloverReadyStep} and the move to `attempt-rollover` request will not be successful.
1061+
waitUntil(() -> {
1062+
try {
1063+
return client().performRequest(moveToStepRequest).getStatusLine().getStatusCode() == 200;
1064+
} catch (IOException e) {
1065+
return false;
1066+
}
1067+
}, 30, TimeUnit.SECONDS);
1068+
1069+
// Similar to above, using {@link #waitUntil} as we want to make sure the `attempt-rollover` step started failing and is being
1070+
// retried (which means ILM moves back and forth between the `attempt-rollover` step and the `error` step)
1071+
waitUntil(() -> {
1072+
try {
1073+
Map<String, Object> explainIndexResponse = explainIndex(index);
1074+
String step = (String) explainIndexResponse.get("step");
1075+
Integer retryCount = (Integer) explainIndexResponse.get(FAILED_STEP_RETRY_COUNT_FIELD);
1076+
return step != null && step.equals("attempt-rollover") && retryCount != null && retryCount >= 1;
1077+
} catch (IOException e) {
1078+
return false;
1079+
}
1080+
}, 30, TimeUnit.SECONDS);
1081+
1082+
deleteIndex(rolledIndex);
1083+
1084+
// the rollover step should eventually succeed
1085+
assertBusy(() -> assertThat(indexExists(rolledIndex), is(true)));
1086+
assertBusy(() -> assertThat(getStepKeyForIndex(index), equalTo(TerminalPolicyStep.KEY)));
1087+
}
10151088

10161089
public void testHistoryIsWrittenWithSuccess() throws Exception {
10171090
String index = "index";

‎x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycleRunner.java

+14
Original file line numberDiff line numberDiff line change
@@ -203,6 +203,20 @@ public void onFailure(String source, Exception e) {
203203
logger.error(new ParameterizedMessage("retry execution of step [{}] for index [{}] failed",
204204
failedStep.getKey().getName(), index), e);
205205
}
206+
207+
@Override
208+
public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
209+
if (oldState.equals(newState) == false) {
210+
IndexMetaData newIndexMeta = newState.metaData().index(index);
211+
Step indexMetaCurrentStep = getCurrentStep(stepRegistry, policy, newIndexMeta);
212+
StepKey stepKey = indexMetaCurrentStep.getKey();
213+
if (stepKey != null && stepKey != TerminalPolicyStep.KEY && newIndexMeta != null) {
214+
logger.trace("policy [{}] for index [{}] was moved back on the failed step for as part of an automatic " +
215+
"retry. Attempting to execute the failed step [{}] if it's an async action", policy, index, stepKey);
216+
maybeRunAsyncAction(newState, newIndexMeta, policy, stepKey);
217+
}
218+
}
219+
}
206220
});
207221
} else {
208222
logger.debug("policy [{}] for index [{}] on an error step after a terminal error, skipping execution", policy, index);

0 commit comments

Comments
 (0)