Skip to content

Commit 6b13189

Browse files
bizybotbizybot
authored
Add pre-upgrade check to test cluster routing allocation is enabled (elastic#39340) (elastic#39816)
When following the steps mentioned in upgrade guide https://www.elastic.co/guide/en/elastic-stack/6.6/upgrading-elastic-stack.html if we disable the cluster shard allocation but fail to enable it after upgrading the nodes and plugins, the next step of upgrading internal indices fails. As we did not check the bulk request response for reindexing, we delete the old index assuming it has been created. This is fatal as we cannot recover from this state. This commit adds a pre-upgrade check to test the cluster shard allocation setting and fail upgrade if it is disabled. In case there are search or bulk failures then we remove the read-only block and fail the upgrade index request. Closes elastic#39339
1 parent 9d35df8 commit 6b13189

File tree

5 files changed

+122
-46
lines changed

5 files changed

+122
-46
lines changed

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/upgrade/IndexUpgradeCheckVersion.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
package org.elasticsearch.xpack.core.upgrade;
77

88
public final class IndexUpgradeCheckVersion {
9-
public static final int UPRADE_VERSION = 6;
9+
public static final int UPGRADE_VERSION = 6;
1010

1111
private IndexUpgradeCheckVersion() {}
1212

‎x-pack/plugin/upgrade/src/main/java/org/elasticsearch/xpack/upgrade/IndexUpgradeCheck.java

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,13 @@
55
*/
66
package org.elasticsearch.xpack.upgrade;
77

8+
import org.elasticsearch.ElasticsearchException;
89
import org.elasticsearch.action.ActionListener;
910
import org.elasticsearch.client.Client;
1011
import org.elasticsearch.cluster.ClusterState;
1112
import org.elasticsearch.cluster.metadata.IndexMetaData;
13+
import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
14+
import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider.Allocation;
1215
import org.elasticsearch.cluster.service.ClusterService;
1316
import org.elasticsearch.index.reindex.BulkByScrollResponse;
1417
import org.elasticsearch.protocol.xpack.migration.UpgradeActionRequired;
@@ -18,7 +21,6 @@
1821
import org.elasticsearch.xpack.core.upgrade.IndexUpgradeCheckVersion;
1922

2023
import java.util.function.BiConsumer;
21-
import java.util.function.Consumer;
2224
import java.util.function.Function;
2325

2426
/**
@@ -51,7 +53,17 @@ public IndexUpgradeCheck(String name,
5153
Function<IndexMetaData, UpgradeActionRequired> actionRequired,
5254
Client client, ClusterService clusterService, String[] types, Script updateScript) {
5355
this(name, actionRequired, client, clusterService, types, updateScript,
54-
listener -> listener.onResponse(null), (t, listener) -> listener.onResponse(TransportResponse.Empty.INSTANCE));
56+
(cs, listener) -> {
57+
Allocation clusterRoutingAllocation = EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING
58+
.get(cs.getMetaData().settings());
59+
if (Allocation.NONE == clusterRoutingAllocation) {
60+
listener.onFailure(new ElasticsearchException(
61+
"pre-upgrade check failed, please enable cluster routing allocation using setting [{}]",
62+
EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING.getKey()));
63+
} else {
64+
listener.onResponse(null);
65+
}
66+
}, (t, listener) -> listener.onResponse(TransportResponse.Empty.INSTANCE));
5567
}
5668

5769
/**
@@ -69,11 +81,11 @@ public IndexUpgradeCheck(String name,
6981
public IndexUpgradeCheck(String name,
7082
Function<IndexMetaData, UpgradeActionRequired> actionRequired,
7183
Client client, ClusterService clusterService, String[] types, Script updateScript,
72-
Consumer<ActionListener<T>> preUpgrade,
84+
BiConsumer<ClusterState, ActionListener<T>> preUpgrade,
7385
BiConsumer<T, ActionListener<TransportResponse.Empty>> postUpgrade) {
7486
this.name = name;
7587
this.actionRequired = actionRequired;
76-
this.reindexer = new InternalIndexReindexer<>(client, clusterService, IndexUpgradeCheckVersion.UPRADE_VERSION, updateScript,
88+
this.reindexer = new InternalIndexReindexer<>(client, clusterService, IndexUpgradeCheckVersion.UPGRADE_VERSION, updateScript,
7789
types, preUpgrade, postUpgrade);
7890
}
7991

@@ -106,4 +118,9 @@ public void upgrade(TaskId task, IndexMetaData indexMetaData, ClusterState state
106118
ActionListener<BulkByScrollResponse> listener) {
107119
reindexer.upgrade(task, indexMetaData.getIndex().getName(), state, listener);
108120
}
121+
122+
// pkg scope for testing
123+
InternalIndexReindexer getInternalIndexReindexer() {
124+
return reindexer;
125+
}
109126
}

‎x-pack/plugin/upgrade/src/main/java/org/elasticsearch/xpack/upgrade/InternalIndexReindexer.java

Lines changed: 56 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,9 @@
55
*/
66
package org.elasticsearch.xpack.upgrade;
77

8+
import org.apache.logging.log4j.LogManager;
9+
import org.apache.logging.log4j.Logger;
10+
import org.elasticsearch.ElasticsearchException;
811
import org.elasticsearch.action.ActionListener;
912
import org.elasticsearch.action.support.master.AcknowledgedResponse;
1013
import org.elasticsearch.client.Client;
@@ -15,6 +18,7 @@
1518
import org.elasticsearch.cluster.metadata.IndexMetaData;
1619
import org.elasticsearch.cluster.metadata.MetaData;
1720
import org.elasticsearch.cluster.service.ClusterService;
21+
import org.elasticsearch.common.Strings;
1822
import org.elasticsearch.common.settings.Settings;
1923
import org.elasticsearch.index.IndexNotFoundException;
2024
import org.elasticsearch.index.reindex.BulkByScrollResponse;
@@ -25,7 +29,6 @@
2529
import org.elasticsearch.transport.TransportResponse;
2630

2731
import java.util.function.BiConsumer;
28-
import java.util.function.Consumer;
2932

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

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

4347
private final Client client;
4448
private final ClusterService clusterService;
4549
private final Script transformScript;
4650
private final String[] types;
4751
private final int version;
48-
private final Consumer<ActionListener<T>> preUpgrade;
52+
private final BiConsumer<ClusterState, ActionListener<T>> preUpgrade;
4953
private final BiConsumer<T, ActionListener<TransportResponse.Empty>> postUpgrade;
5054

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

6367
public void upgrade(TaskId task, String index, ClusterState clusterState, ActionListener<BulkByScrollResponse> listener) {
6468
ParentTaskAssigningClient parentAwareClient = new ParentTaskAssigningClient(client, task);
65-
preUpgrade.accept(ActionListener.wrap(
69+
preUpgrade.accept(clusterState, ActionListener.wrap(
6670
t -> innerUpgrade(parentAwareClient, index, clusterState, ActionListener.wrap(
6771
response -> postUpgrade.accept(t, ActionListener.wrap(
6872
empty -> listener.onResponse(response),
@@ -76,32 +80,61 @@ public void upgrade(TaskId task, String index, ClusterState clusterState, Action
7680
private void innerUpgrade(ParentTaskAssigningClient parentAwareClient, String index, ClusterState clusterState,
7781
ActionListener<BulkByScrollResponse> listener) {
7882
String newIndex = index + "-" + version;
83+
logger.trace("upgrading index {} to new index {}", index, newIndex);
7984
try {
8085
checkMasterAndDataNodeVersion(clusterState);
81-
parentAwareClient.admin().indices().prepareCreate(newIndex).execute(ActionListener.wrap(createIndexResponse ->
82-
setReadOnlyBlock(index, ActionListener.wrap(setReadOnlyResponse ->
83-
reindex(parentAwareClient, index, newIndex, ActionListener.wrap(
84-
bulkByScrollResponse -> // Successful completion of reindexing - delete old index
85-
removeReadOnlyBlock(parentAwareClient, index, ActionListener.wrap(unsetReadOnlyResponse ->
86-
parentAwareClient.admin().indices().prepareAliases().removeIndex(index)
87-
.addAlias(newIndex, index).execute(ActionListener.wrap(deleteIndexResponse ->
88-
listener.onResponse(bulkByScrollResponse), listener::onFailure
89-
)), listener::onFailure
90-
)),
91-
e -> // Something went wrong during reindexing - remove readonly flag and report the error
92-
removeReadOnlyBlock(parentAwareClient, index, ActionListener.wrap(unsetReadOnlyResponse -> {
93-
listener.onFailure(e);
94-
}, e1 -> {
95-
listener.onFailure(e);
96-
}))
97-
)), listener::onFailure
98-
)), listener::onFailure
99-
));
86+
parentAwareClient.admin().indices().prepareCreate(newIndex).execute(ActionListener.wrap(createIndexResponse -> {
87+
setReadOnlyBlock(index, ActionListener.wrap(
88+
setReadOnlyResponse -> reindex(parentAwareClient, index, newIndex, ActionListener.wrap(bulkByScrollResponse -> {
89+
if ((bulkByScrollResponse.getBulkFailures() != null
90+
&& bulkByScrollResponse.getBulkFailures().isEmpty() == false)
91+
|| (bulkByScrollResponse.getSearchFailures() != null
92+
&& bulkByScrollResponse.getSearchFailures().isEmpty() == false)) {
93+
ElasticsearchException ex = logAndThrowExceptionForFailures(bulkByScrollResponse);
94+
removeReadOnlyBlockOnReindexFailure(parentAwareClient, index, listener, ex);
95+
} else {
96+
// Successful completion of reindexing - remove read only and delete old index
97+
removeReadOnlyBlock(parentAwareClient, index,
98+
ActionListener.wrap(unsetReadOnlyResponse -> parentAwareClient.admin().indices().prepareAliases()
99+
.removeIndex(index).addAlias(newIndex, index)
100+
.execute(ActionListener.wrap(
101+
deleteIndexResponse -> listener.onResponse(bulkByScrollResponse),
102+
listener::onFailure)),
103+
listener::onFailure));
104+
}
105+
}, e -> {
106+
logger.error("error occurred while reindexing", e);
107+
removeReadOnlyBlockOnReindexFailure(parentAwareClient, index, listener, e);
108+
})), listener::onFailure));
109+
}, listener::onFailure));
100110
} catch (Exception ex) {
111+
logger.error("error occurred while upgrading index", ex);
112+
removeReadOnlyBlockOnReindexFailure(parentAwareClient, index, listener, ex);
101113
listener.onFailure(ex);
102114
}
103115
}
104116

117+
private void removeReadOnlyBlockOnReindexFailure(ParentTaskAssigningClient parentAwareClient, String index,
118+
ActionListener<BulkByScrollResponse> listener, Exception ex) {
119+
removeReadOnlyBlock(parentAwareClient, index, ActionListener.wrap(unsetReadOnlyResponse -> {
120+
listener.onFailure(ex);
121+
}, e1 -> {
122+
listener.onFailure(ex);
123+
}));
124+
}
125+
126+
private ElasticsearchException logAndThrowExceptionForFailures(BulkByScrollResponse bulkByScrollResponse) {
127+
String bulkFailures = (bulkByScrollResponse.getBulkFailures() != null)
128+
? Strings.collectionToCommaDelimitedString(bulkByScrollResponse.getBulkFailures())
129+
: "";
130+
String searchFailures = (bulkByScrollResponse.getSearchFailures() != null)
131+
? Strings.collectionToCommaDelimitedString(bulkByScrollResponse.getSearchFailures())
132+
: "";
133+
logger.error("error occurred while reindexing, bulk failures [{}], search failures [{}]", bulkFailures, searchFailures);
134+
return new ElasticsearchException("error occurred while reindexing, bulk failures [{}], search failures [{}]", bulkFailures,
135+
searchFailures);
136+
}
137+
105138
private void checkMasterAndDataNodeVersion(ClusterState clusterState) {
106139
if (clusterState.nodes().getMinNodeVersion().before(Upgrade.UPGRADE_INTRODUCED)) {
107140
throw new IllegalStateException("All nodes should have at least version [" + Upgrade.UPGRADE_INTRODUCED + "] to upgrade");

‎x-pack/plugin/upgrade/src/test/java/org/elasticsearch/xpack/upgrade/IndexUpgradeIT.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ public void testInternalUpgradePrePostChecks() throws Exception {
9696
}
9797
},
9898
client(), internalCluster().clusterService(internalCluster().getMasterName()), Strings.EMPTY_ARRAY, null,
99-
listener -> {
99+
(cs, listener) -> {
100100
assertFalse(preUpgradeIsCalled.getAndSet(true));
101101
assertFalse(postUpgradeIsCalled.get());
102102
listener.onResponse(val);

0 commit comments

Comments
 (0)