Skip to content

System Index Migration Failure Results in a Non-Recoverable State #122326

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Feb 13, 2025
Merged

System Index Migration Failure Results in a Non-Recoverable State #122326

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6fa041c
Retry on createIndex failure
JVerwolf Feb 11, 2025
abf0f97
Remove comment
JVerwolf Feb 11, 2025
aa2b112
remove dead code
JVerwolf Feb 12, 2025
98cabbd
Add additional failure branch
JVerwolf Feb 12, 2025
ed44d0d
PR feedback: Fix log line
JVerwolf Feb 12, 2025
b3faf74
Merge branch 'main' of github.com:elastic/elasticsearch into bugfix/f…
JVerwolf Feb 13, 2025
b1d8538
Fix logger usage
JVerwolf Feb 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions docs/changelog/122326.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
pr: 122326
summary: System Index Migration Failure Results in a Non-Recoverable State
area: Infra/Core
type: bug
issues: []
1 change: 0 additions & 1 deletion .../reindex/src/internalClusterTest/java/org/elasticsearch/migration/FeatureMigrationIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -277,7 +277,6 @@ public void testMigrateIndexWithWriteBlock() throws Exception {
});
}

@AwaitsFix(bugUrl = "ES-10666") // This test uncovered an existing issue
public void testIndexBlockIsRemovedWhenAliasRequestFails() throws Exception {
createSystemIndexForDescriptor(INTERNAL_UNMANAGED);
ensureGreen();
Expand Down
85 changes: 35 additions & 50 deletions server/src/main/java/org/elasticsearch/upgrades/SystemIndexMigrator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -207,49 +207,14 @@ public void run(SystemIndexMigrationTaskState taskState) {
}

// Kick off our callback "loop" - finishIndexAndLoop calls back into prepareNextIndex
cleanUpPreviousMigration(
taskState,
clusterState,
state -> prepareNextIndex(state, state2 -> migrateSingleIndex(state2, this::finishIndexAndLoop), stateFeatureName)
);
}

private void cleanUpPreviousMigration(
SystemIndexMigrationTaskState taskState,
ClusterState currentState,
Consumer<ClusterState> listener
) {
logger.debug("cleaning up previous migration, task state: [{}]", taskState == null ? "null" : Strings.toString(taskState));
if (taskState != null && taskState.getCurrentIndex() != null) {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The taskState is always null, so the cleanup logic is never invoked here.

SystemIndexMigrationInfo migrationInfo;
try {
migrationInfo = SystemIndexMigrationInfo.fromTaskState(
taskState,
systemIndices,
currentState.metadata(),
indexScopedSettings
);
} catch (Exception e) {
markAsFailed(e);
return;
}
final String newIndexName = migrationInfo.getNextIndexName();
logger.info("removing index [{}] from previous incomplete migration", newIndexName);

migrationInfo.createClient(baseClient)
.admin()
.indices()
.prepareDelete(newIndexName)
.execute(ActionListener.wrap(ackedResponse -> {
if (ackedResponse.isAcknowledged()) {
logger.debug("successfully removed index [{}]", newIndexName);
clearResults(clusterService, ActionListener.wrap(listener::accept, this::markAsFailed));
}
}, this::markAsFailed));
} else {
logger.debug("no incomplete index to remove");
clearResults(clusterService, ActionListener.wrap(listener::accept, this::markAsFailed));
}
clearResults(
Copy link
Contributor Author

@JVerwolf JVerwolf Feb 12, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The purpose ofclearResults is to remove the Persistent Task State leftover from prior runs so that we begin the migration with fresh start. I'm not sure if we ever even store the persistent task state in practice? The call to super.markAsFailed(e); in markAsFailed ultimately removes the task state, though perhaps this wasn't the original intention.

It may be possible that there are other failure scenarios that leave the teask sate (e.g. a node getting abruptly killed). So I left this herere for now untill I can reasont through this with @gwbrown when she's back.

clusterService,
ActionListener.wrap(
state -> prepareNextIndex(state2 -> migrateSingleIndex(state2, this::finishIndexAndLoop), stateFeatureName),
this::markAsFailed
)
);
}

private void finishIndexAndLoop(BulkByScrollResponse bulkResponse) {
Expand Down Expand Up @@ -289,11 +254,7 @@ private void finishIndexAndLoop(BulkByScrollResponse bulkResponse) {
}, this::markAsFailed)
);
} else {
prepareNextIndex(
clusterService.state(),
state2 -> migrateSingleIndex(state2, this::finishIndexAndLoop),
lastMigrationInfo.getFeatureName()
);
prepareNextIndex(state2 -> migrateSingleIndex(state2, this::finishIndexAndLoop), lastMigrationInfo.getFeatureName());
}
}

Expand All @@ -303,7 +264,6 @@ private void recordIndexMigrationSuccess(SystemIndexMigrationInfo lastMigrationI
SingleFeatureMigrationResult.success(),
ActionListener.wrap(state -> {
prepareNextIndex(
state,
clusterState -> migrateSingleIndex(clusterState, this::finishIndexAndLoop),
lastMigrationInfo.getFeatureName()
);
Expand All @@ -312,7 +272,7 @@ private void recordIndexMigrationSuccess(SystemIndexMigrationInfo lastMigrationI
updateTask.submit(clusterService);
}

private void prepareNextIndex(ClusterState clusterState, Consumer<ClusterState> listener, String lastFeatureName) {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The clusterState was never used, so I removed this extraneous param.

private void prepareNextIndex(Consumer<ClusterState> listener, String lastFeatureName) {
synchronized (migrationQueue) {
assert migrationQueue != null;
if (migrationQueue.isEmpty()) {
Expand Down Expand Up @@ -424,7 +384,7 @@ private void migrateSingleIndex(ClusterState clusterState, Consumer<BulkByScroll
logger.info("migrating index [{}] from feature [{}] to new index [{}]", oldIndexName, migrationInfo.getFeatureName(), newIndexName);
ActionListener<BulkByScrollResponse> innerListener = ActionListener.wrap(listener::accept, this::markAsFailed);
try {
createIndex(migrationInfo, innerListener.delegateFailureAndWrap((delegate, shardsAcknowledgedResponse) -> {
createIndexRetryOnFailure(migrationInfo, innerListener.delegateFailureAndWrap((delegate, shardsAcknowledgedResponse) -> {
logger.debug(
"while migrating [{}] , got create index response: [{}]",
oldIndexName,
Expand Down Expand Up @@ -509,6 +469,8 @@ private void migrateSingleIndex(ClusterState clusterState, Consumer<BulkByScroll
}

private void createIndex(SystemIndexMigrationInfo migrationInfo, ActionListener<ShardsAcknowledgedResponse> listener) {
logger.info("creating new system index [{}] from feature [{}]", migrationInfo.getNextIndexName(), migrationInfo.getFeatureName());

final CreateIndexClusterStateUpdateRequest createRequest = new CreateIndexClusterStateUpdateRequest(
"migrate-system-index",
migrationInfo.getNextIndexName(),
Expand All @@ -534,6 +496,29 @@ private void createIndex(SystemIndexMigrationInfo migrationInfo, ActionListener<
);
}

private void createIndexRetryOnFailure(SystemIndexMigrationInfo migrationInfo, ActionListener<ShardsAcknowledgedResponse> listener) {
createIndex(migrationInfo, ActionListener.wrap(listener::onResponse, e -> {
logger.warn("createIndex failed, retrying after removing index [{}] from previous attempt", migrationInfo.getNextIndexName());
deleteIndex(migrationInfo, ActionListener.wrap(cleanupResponse -> createIndex(migrationInfo, listener), e2 -> {
logger.warn("createIndex failed after retrying, aborting", e2);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I lost in callbacks, but isn't this error consumer for deleteIndex?

Copy link
Contributor Author

@JVerwolf JVerwolf Feb 12, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're right, I've fixed this now. Nice catch! Also, welcome to callback hell.

listener.onFailure(e2);
}));
}));
}

private <T> void deleteIndex(SystemIndexMigrationInfo migrationInfo, ActionListener<AcknowledgedResponse> listener) {
logger.info("removing index [{}] from feature [{}]", migrationInfo.getNextIndexName(), migrationInfo.getFeatureName());
String newIndexName = migrationInfo.getNextIndexName();
baseClient.admin().indices().prepareDelete(newIndexName).execute(ActionListener.wrap(ackedResponse -> {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This code is identical to the prior cleanUpPreviousMigration code except for that I'm using a baseClient here since I don't have access to the migrationInfo needed for migrationInfo.createClient(baseClient). This previous call produced an Origin Setting Client. I don't think we need that here, though I'm not 100% on this.

if (ackedResponse.isAcknowledged()) {
logger.info("successfully removed index [{}]", newIndexName);
listener.onResponse(ackedResponse);
Copy link
Contributor Author

@JVerwolf JVerwolf Feb 12, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This code is identical to the prior cleanUpPreviousMigration code except that I don't call clearResults(clusterService, ActionListener.wrap(listener::accept, this::markAsFailed)); here, as I've already cleaned up at the beginning of the migration, and I don't want to remove in-progress state since this now happens once the migration is already in progress.

} else {
listener.onFailure(new ElasticsearchException("Failed to acknowledge index deletion for [" + newIndexName + "]"));
}
}, listener::onFailure));
}

private void setAliasAndRemoveOldIndex(SystemIndexMigrationInfo migrationInfo, ActionListener<IndicesAliasesResponse> listener) {
final IndicesAliasesRequestBuilder aliasesRequest = migrationInfo.createClient(baseClient).admin().indices().prepareAliases();
aliasesRequest.removeIndex(migrationInfo.getCurrentIndexName());
Expand Down