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

[BACKPORT] Add pre-upgrade check to test cluster routing allocation is enabled #39817

Merged
merged 1 commit into from
Mar 12, 2019
Merged

[BACKPORT] Add pre-upgrade check to test cluster routing allocation is enabled #39817

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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.common.component.AbstractComponent;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
Expand All @@ -19,7 +22,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 @@ -52,7 +54,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 @@ -70,11 +82,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 @@ -107,4 +119,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() {
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);
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
41 changes: 36 additions & 5 deletions x-pack/plugin/upgrade/src/main/java/org/elasticsearch/xpack/upgrade/Upgrade.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,12 @@
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.node.DiscoveryNodes;
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.common.io.stream.NamedWriteableRegistry;
import org.elasticsearch.common.settings.ClusterSettings;
Expand Down Expand Up @@ -167,7 +170,15 @@ static BiFunction<Client, ClusterService, IndexUpgradeCheck> getSecurityUpgradeC
" ctx._type = \"doc\";" +
"}\n",
new HashMap<>()),
listener -> listener.onResponse(null),
(cs, listener) -> {
if (isClusterRoutingAllocationEnabled(cs) == false) {
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);
}
},
(success, listener) -> postSecurityUpgrade(clientWithOrigin, listener));
};
}
Expand Down Expand Up @@ -261,7 +272,7 @@ static BiFunction<Client, ClusterService, IndexUpgradeCheck> getWatchesIndexUpgr
" ctx._source.status = ctx._source.remove(\"_status\");\n" +
"}",
new HashMap<>()),
booleanActionListener -> preWatchesIndexUpgrade(clientWithOrigin, booleanActionListener),
(cs, booleanActionListener) -> preWatchesIndexUpgrade(clientWithOrigin, cs, booleanActionListener),
(shouldStartWatcher, listener) -> postWatchesIndexUpgrade(clientWithOrigin, shouldStartWatcher, listener)
);
};
Expand All @@ -285,7 +296,7 @@ static BiFunction<Client, ClusterService, IndexUpgradeCheck> getTriggeredWatches
clusterService,
new String[]{"triggered-watch"},
new Script(ScriptType.INLINE, "painless", "ctx._type = \"doc\";\n", new HashMap<>()),
booleanActionListener -> preTriggeredWatchesIndexUpgrade(clientWithOrigin, booleanActionListener),
(cs, booleanActionListener) -> preTriggeredWatchesIndexUpgrade(clientWithOrigin, cs, booleanActionListener),
(shouldStartWatcher, listener) -> postWatchesIndexUpgrade(clientWithOrigin, shouldStartWatcher, listener)
);
};
Expand All @@ -295,7 +306,12 @@ private static boolean indexOrAliasExists(IndexMetaData indexMetaData, String na
return name.equals(indexMetaData.getIndex().getName()) || indexMetaData.getAliases().containsKey(name);
}

static void preTriggeredWatchesIndexUpgrade(Client client, ActionListener<Boolean> listener) {
static void preTriggeredWatchesIndexUpgrade(Client client, ClusterState cs, ActionListener<Boolean> listener) {
if (isClusterRoutingAllocationEnabled(cs) == false) {
listener.onFailure(new ElasticsearchException(
"pre-upgrade check failed, please enable cluster routing allocation using setting [{}]",
EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING.getKey()));
}
new WatcherClient(client).prepareWatcherStats().execute(ActionListener.wrap(
stats -> {
if (stats.watcherMetaData().manuallyStopped()) {
Expand All @@ -316,6 +332,15 @@ static void preTriggeredWatchesIndexUpgrade(Client client, ActionListener<Boolea
listener::onFailure));
}

static boolean isClusterRoutingAllocationEnabled(ClusterState cs) {
Allocation clusterRoutingAllocation = EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING
.get(cs.getMetaData().settings());
if (Allocation.NONE == clusterRoutingAllocation) {
return false;
}
return true;
}

private static void preTriggeredWatchesIndexUpgrade(final Client client, final ActionListener<Boolean> listener,
final boolean restart) {
final String legacyTriggeredWatchesTemplateName = "triggered_watches";
Expand All @@ -338,7 +363,13 @@ private static void preTriggeredWatchesIndexUpgrade(final Client client, final A
.setSource(triggeredWatchesTemplate, XContentType.JSON).execute(putTriggeredWatchesListener);
}

static void preWatchesIndexUpgrade(Client client, ActionListener<Boolean> listener) {
static void preWatchesIndexUpgrade(Client client, ClusterState cs, ActionListener<Boolean> listener) {
if (isClusterRoutingAllocationEnabled(cs) == false) {
listener.onFailure(new ElasticsearchException(
"pre-upgrade check failed, please enable cluster routing allocation using setting [{}]",
EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING.getKey()));
}

new WatcherClient(client).prepareWatcherStats().execute(ActionListener.wrap(
stats -> {
if (stats.watcherMetaData().manuallyStopped()) {
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 @@ -97,7 +97,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