Skip to content
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

Add pre-upgrade check to test cluster routing allocation is enabled #39340

Merged
merged 4 commits into from
Mar 8, 2019
Merged

Add pre-upgrade check to test cluster routing allocation is enabled #39340

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b9d5b7d
Add pre-upgrade check to test cluster routing allocation is enabled
Feb 25, 2019
b373c17
Merge branch 'master' into fix-internal-index-reindexer
Feb 27, 2019
c17c8a8
revert unwanted change
Feb 27, 2019
9f49851
revert unwanted change
Feb 27, 2019
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
2 changes: 1 addition & 1 deletion ...gin/core/src/main/java/org/elasticsearch/xpack/core/upgrade/IndexUpgradeCheckVersion.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
package org.elasticsearch.xpack.core.upgrade;

public final class IndexUpgradeCheckVersion {
public static final int UPRADE_VERSION = 6;
public static final int UPGRADE_VERSION = 6;

private IndexUpgradeCheckVersion() {}

Expand Down
25 changes: 21 additions & 4 deletions x-pack/plugin/upgrade/src/main/java/org/elasticsearch/xpack/upgrade/IndexUpgradeCheck.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,13 @@
*/
package org.elasticsearch.xpack.upgrade;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider.Allocation;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.protocol.xpack.migration.UpgradeActionRequired;
Expand All @@ -18,7 +21,6 @@
import org.elasticsearch.xpack.core.upgrade.IndexUpgradeCheckVersion;

import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;

/**
Expand Down Expand Up @@ -51,7 +53,17 @@ public IndexUpgradeCheck(String name,
Function<IndexMetaData, UpgradeActionRequired> actionRequired,
Client client, ClusterService clusterService, String[] types, Script updateScript) {
this(name, actionRequired, client, clusterService, types, updateScript,
listener -> listener.onResponse(null), (t, listener) -> listener.onResponse(TransportResponse.Empty.INSTANCE));
(cs, listener) -> {
Allocation clusterRoutingAllocation = EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING
.get(cs.getMetaData().settings());
if (Allocation.NONE == clusterRoutingAllocation) {
listener.onFailure(new ElasticsearchException(
"pre-upgrade check failed, please enable cluster routing allocation using setting [{}]",
EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING.getKey()));
} else {
listener.onResponse(null);
}
}, (t, listener) -> listener.onResponse(TransportResponse.Empty.INSTANCE));
}

/**
Expand All @@ -69,11 +81,11 @@ public IndexUpgradeCheck(String name,
public IndexUpgradeCheck(String name,
Function<IndexMetaData, UpgradeActionRequired> actionRequired,
Client client, ClusterService clusterService, String[] types, Script updateScript,
Consumer<ActionListener<T>> preUpgrade,
BiConsumer<ClusterState, ActionListener<T>> preUpgrade,
BiConsumer<T, ActionListener<TransportResponse.Empty>> postUpgrade) {
this.name = name;
this.actionRequired = actionRequired;
this.reindexer = new InternalIndexReindexer<>(client, clusterService, IndexUpgradeCheckVersion.UPRADE_VERSION, updateScript,
this.reindexer = new InternalIndexReindexer<>(client, clusterService, IndexUpgradeCheckVersion.UPGRADE_VERSION, updateScript,
types, preUpgrade, postUpgrade);
}

Expand Down Expand Up @@ -106,4 +118,9 @@ public void upgrade(TaskId task, IndexMetaData indexMetaData, ClusterState state
ActionListener<BulkByScrollResponse> listener) {
reindexer.upgrade(task, indexMetaData.getIndex().getName(), state, listener);
}

// pkg scope for testing
InternalIndexReindexer getInternalIndexReindexer() {
Copy link
Contributor

Choose a reason for hiding this comment

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

That concerns me a bit. Could you add a comment why this is necessary?

return reindexer;
}
}
79 changes: 56 additions & 23 deletions .../plugin/upgrade/src/main/java/org/elasticsearch/xpack/upgrade/InternalIndexReindexer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,9 @@
*/
package org.elasticsearch.xpack.upgrade;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.Client;
Expand All @@ -15,6 +18,7 @@
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
Expand All @@ -25,7 +29,6 @@
import org.elasticsearch.transport.TransportResponse;

import java.util.function.BiConsumer;
import java.util.function.Consumer;

import static org.elasticsearch.index.IndexSettings.same;

Expand All @@ -39,17 +42,18 @@
* - Delete index .{name} and add alias .{name} to .{name}-6
*/
public class InternalIndexReindexer<T> {
private static final Logger logger = LogManager.getLogger(InternalIndexReindexer.class);

private final Client client;
private final ClusterService clusterService;
private final Script transformScript;
private final String[] types;
private final int version;
private final Consumer<ActionListener<T>> preUpgrade;
private final BiConsumer<ClusterState, ActionListener<T>> preUpgrade;
private final BiConsumer<T, ActionListener<TransportResponse.Empty>> postUpgrade;

public InternalIndexReindexer(Client client, ClusterService clusterService, int version, Script transformScript, String[] types,
Consumer<ActionListener<T>> preUpgrade,
BiConsumer<ClusterState,ActionListener<T>> preUpgrade,
BiConsumer<T, ActionListener<TransportResponse.Empty>> postUpgrade) {
this.client = client;
this.clusterService = clusterService;
Expand All @@ -62,7 +66,7 @@ public InternalIndexReindexer(Client client, ClusterService clusterService, int

public void upgrade(TaskId task, String index, ClusterState clusterState, ActionListener<BulkByScrollResponse> listener) {
ParentTaskAssigningClient parentAwareClient = new ParentTaskAssigningClient(client, task);
preUpgrade.accept(ActionListener.wrap(
preUpgrade.accept(clusterState, ActionListener.wrap(
t -> innerUpgrade(parentAwareClient, index, clusterState, ActionListener.wrap(
response -> postUpgrade.accept(t, ActionListener.wrap(
empty -> listener.onResponse(response),
Expand All @@ -76,32 +80,61 @@ public void upgrade(TaskId task, String index, ClusterState clusterState, Action
private void innerUpgrade(ParentTaskAssigningClient parentAwareClient, String index, ClusterState clusterState,
ActionListener<BulkByScrollResponse> listener) {
String newIndex = index + "-" + version;
logger.trace("upgrading index {} to new index {}", index, newIndex);
try {
checkMasterAndDataNodeVersion(clusterState);
Copy link
Contributor

Choose a reason for hiding this comment

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

This is where I would've put the allocation check, but no need to amend the PR for this. The reason is to keep all the checks in a single place.

parentAwareClient.admin().indices().prepareCreate(newIndex).execute(ActionListener.wrap(createIndexResponse ->
setReadOnlyBlock(index, ActionListener.wrap(setReadOnlyResponse ->
reindex(parentAwareClient, index, newIndex, ActionListener.wrap(
bulkByScrollResponse -> // Successful completion of reindexing - delete old index
removeReadOnlyBlock(parentAwareClient, index, ActionListener.wrap(unsetReadOnlyResponse ->
parentAwareClient.admin().indices().prepareAliases().removeIndex(index)
.addAlias(newIndex, index).execute(ActionListener.wrap(deleteIndexResponse ->
listener.onResponse(bulkByScrollResponse), listener::onFailure
)), listener::onFailure
)),
e -> // Something went wrong during reindexing - remove readonly flag and report the error
removeReadOnlyBlock(parentAwareClient, index, ActionListener.wrap(unsetReadOnlyResponse -> {
listener.onFailure(e);
}, e1 -> {
listener.onFailure(e);
}))
)), listener::onFailure
)), listener::onFailure
));
parentAwareClient.admin().indices().prepareCreate(newIndex).execute(ActionListener.wrap(createIndexResponse -> {
setReadOnlyBlock(index, ActionListener.wrap(
setReadOnlyResponse -> reindex(parentAwareClient, index, newIndex, ActionListener.wrap(bulkByScrollResponse -> {
if ((bulkByScrollResponse.getBulkFailures() != null
&& bulkByScrollResponse.getBulkFailures().isEmpty() == false)
|| (bulkByScrollResponse.getSearchFailures() != null
&& bulkByScrollResponse.getSearchFailures().isEmpty() == false)) {
ElasticsearchException ex = logAndThrowExceptionForFailures(bulkByScrollResponse);
removeReadOnlyBlockOnReindexFailure(parentAwareClient, index, listener, ex);
} else {
// Successful completion of reindexing - remove read only and delete old index
removeReadOnlyBlock(parentAwareClient, index,
ActionListener.wrap(unsetReadOnlyResponse -> parentAwareClient.admin().indices().prepareAliases()
.removeIndex(index).addAlias(newIndex, index)
.execute(ActionListener.wrap(
deleteIndexResponse -> listener.onResponse(bulkByScrollResponse),
listener::onFailure)),
listener::onFailure));
}
}, e -> {
logger.error("error occurred while reindexing", e);
removeReadOnlyBlockOnReindexFailure(parentAwareClient, index, listener, e);
})), listener::onFailure));
}, listener::onFailure));
} catch (Exception ex) {
logger.error("error occurred while upgrading index", ex);
removeReadOnlyBlockOnReindexFailure(parentAwareClient, index, listener, ex);
listener.onFailure(ex);
}
}

private void removeReadOnlyBlockOnReindexFailure(ParentTaskAssigningClient parentAwareClient, String index,
ActionListener<BulkByScrollResponse> listener, Exception ex) {
removeReadOnlyBlock(parentAwareClient, index, ActionListener.wrap(unsetReadOnlyResponse -> {
listener.onFailure(ex);
}, e1 -> {
listener.onFailure(ex);
}));
}

private ElasticsearchException logAndThrowExceptionForFailures(BulkByScrollResponse bulkByScrollResponse) {
String bulkFailures = (bulkByScrollResponse.getBulkFailures() != null)
? Strings.collectionToCommaDelimitedString(bulkByScrollResponse.getBulkFailures())
: "";
String searchFailures = (bulkByScrollResponse.getSearchFailures() != null)
? Strings.collectionToCommaDelimitedString(bulkByScrollResponse.getSearchFailures())
: "";
logger.error("error occurred while reindexing, bulk failures [{}], search failures [{}]", bulkFailures, searchFailures);
return new ElasticsearchException("error occurred while reindexing, bulk failures [{}], search failures [{}]", bulkFailures,
searchFailures);
}

private void checkMasterAndDataNodeVersion(ClusterState clusterState) {
if (clusterState.nodes().getMinNodeVersion().before(Upgrade.UPGRADE_INTRODUCED)) {
throw new IllegalStateException("All nodes should have at least version [" + Upgrade.UPGRADE_INTRODUCED + "] to upgrade");
Expand Down
2 changes: 1 addition & 1 deletion x-pack/plugin/upgrade/src/test/java/org/elasticsearch/xpack/upgrade/IndexUpgradeIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ public void testInternalUpgradePrePostChecks() throws Exception {
}
},
client(), internalCluster().clusterService(internalCluster().getMasterName()), Strings.EMPTY_ARRAY, null,
listener -> {
(cs, listener) -> {
assertFalse(preUpgradeIsCalled.getAndSet(true));
assertFalse(postUpgradeIsCalled.get());
listener.onResponse(val);
Expand Down
Loading