Skip to content

Commit ecbc360

Browse files
nielsbaumannielsbauman
authored
Run TransportClusterStateAction on local node (elastic#129872)
This action solely needs the cluster state, it can run on any node. Since this action is invoked across clusters, we need to be able to (de)serialize requests and responses. We introduce a new `RemoteClusterStateRequest` that wraps the existing `ClusterStateRequest` and implements (de)serialization.
1 parent ac9bef6 commit ecbc360

File tree

53 files changed

+478
-459
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

53 files changed

+478
-459
lines changed

‎docs/changelog/129872.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 129872
2+
summary: Run `TransportClusterStateAction` on local node
3+
area: Distributed
4+
type: enhancement
5+
issues: []

‎plugins/discovery-gce/src/internalClusterTest/java/org/elasticsearch/discovery/gce/GceDiscoverTests.java‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -67,8 +67,7 @@ public void testJoin() {
6767

6868
ClusterStateResponse clusterStateResponse = client(masterNode).admin()
6969
.cluster()
70-
.prepareState(TEST_REQUEST_TIMEOUT)
71-
.setMasterNodeTimeout(TimeValue.timeValueSeconds(1))
70+
.prepareState(TimeValue.timeValueSeconds(1))
7271
.clear()
7372
.setNodes(true)
7473
.get();
@@ -79,11 +78,9 @@ public void testJoin() {
7978
registerGceNode(secondNode);
8079
clusterStateResponse = client(secondNode).admin()
8180
.cluster()
82-
.prepareState(TEST_REQUEST_TIMEOUT)
83-
.setMasterNodeTimeout(TimeValue.timeValueSeconds(1))
81+
.prepareState(TimeValue.timeValueSeconds(1))
8482
.clear()
8583
.setNodes(true)
86-
.setLocal(true)
8784
.get();
8885
assertNotNull(clusterStateResponse.getState().nodes().getMasterNodeId());
8986

‎qa/ccs-unavailable-clusters/src/javaRestTest/java/org/elasticsearch/search/CrossClusterSearchUnavailableClusterIT.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,8 @@
1414
import org.apache.http.nio.entity.NStringEntity;
1515
import org.elasticsearch.TransportVersion;
1616
import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
17-
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
1817
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
18+
import org.elasticsearch.action.admin.cluster.state.RemoteClusterStateRequest;
1919
import org.elasticsearch.action.search.SearchRequest;
2020
import org.elasticsearch.action.search.SearchShardsRequest;
2121
import org.elasticsearch.action.search.SearchShardsResponse;
@@ -117,7 +117,7 @@ private static MockTransportService startTransport(
117117
newService.registerRequestHandler(
118118
ClusterStateAction.NAME,
119119
EsExecutors.DIRECT_EXECUTOR_SERVICE,
120-
ClusterStateRequest::new,
120+
RemoteClusterStateRequest::new,
121121
(request, channel, task) -> {
122122
DiscoveryNodes.Builder builder = DiscoveryNodes.builder();
123123
for (DiscoveryNode node : knownNodes) {

‎rest-api-spec/src/main/resources/rest-api-spec/api/cluster.state.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,12 +69,13 @@
6969
},
7070
"params":{
7171
"local":{
72+
"deprecated":true,
7273
"type":"boolean",
7374
"description":"Return local information, do not retrieve the state from master node (default: false)"
7475
},
7576
"master_timeout":{
7677
"type":"time",
77-
"description":"Specify timeout for connection to master"
78+
"description":"Timeout for waiting for new cluster state in case it is blocked"
7879
},
7980
"flat_settings":{
8081
"type":"boolean",

‎server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/state/TransportClusterStateActionDisruptionIT.java‎

Lines changed: 2 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@
99
package org.elasticsearch.action.admin.cluster.state;
1010

1111
import org.elasticsearch.action.support.SubscribableListener;
12-
import org.elasticsearch.cluster.ClusterState;
1312
import org.elasticsearch.cluster.block.ClusterBlockException;
1413
import org.elasticsearch.cluster.coordination.ClusterBootstrapService;
1514
import org.elasticsearch.cluster.metadata.Metadata;
@@ -18,7 +17,6 @@
1817
import org.elasticsearch.cluster.service.ClusterService;
1918
import org.elasticsearch.common.settings.Settings;
2019
import org.elasticsearch.core.TimeValue;
21-
import org.elasticsearch.discovery.MasterNotDiscoveredException;
2220
import org.elasticsearch.gateway.GatewayService;
2321
import org.elasticsearch.plugins.Plugin;
2422
import org.elasticsearch.test.ESIntegTestCase;
@@ -47,34 +45,15 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
4745
return Collections.singletonList(MockTransportService.TestPlugin.class);
4846
}
4947

50-
public void testNonLocalRequestAlwaysFindsMaster() throws Exception {
51-
runRepeatedlyWhileChangingMaster(() -> {
52-
final ClusterStateRequestBuilder clusterStateRequestBuilder = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT)
53-
.clear()
54-
.setNodes(true)
55-
.setBlocks(true)
56-
.setMasterNodeTimeout(TimeValue.timeValueMillis(100));
57-
final ClusterStateResponse clusterStateResponse;
58-
try {
59-
clusterStateResponse = clusterStateRequestBuilder.get();
60-
} catch (MasterNotDiscoveredException e) {
61-
return; // ok, we hit the disconnected node
62-
}
63-
assertNotNull("should always contain a master node", clusterStateResponse.getState().nodes().getMasterNodeId());
64-
});
65-
}
66-
6748
public void testLocalRequestAlwaysSucceeds() throws Exception {
6849
runRepeatedlyWhileChangingMaster(() -> {
6950
final String node = randomFrom(internalCluster().getNodeNames());
7051
final DiscoveryNodes discoveryNodes = client(node).admin()
7152
.cluster()
72-
.prepareState(TEST_REQUEST_TIMEOUT)
53+
.prepareState(TimeValue.timeValueMillis(100))
7354
.clear()
74-
.setLocal(true)
7555
.setNodes(true)
7656
.setBlocks(true)
77-
.setMasterNodeTimeout(TimeValue.timeValueMillis(100))
7857
.get()
7958
.getState()
8059
.nodes();
@@ -87,39 +66,6 @@ public void testLocalRequestAlwaysSucceeds() throws Exception {
8766
});
8867
}
8968

90-
public void testNonLocalRequestAlwaysFindsMasterAndWaitsForMetadata() throws Exception {
91-
runRepeatedlyWhileChangingMaster(() -> {
92-
final String node = randomFrom(internalCluster().getNodeNames());
93-
final long metadataVersion = internalCluster().getInstance(ClusterService.class, node)
94-
.getClusterApplierService()
95-
.state()
96-
.metadata()
97-
.version();
98-
final long waitForMetadataVersion = randomLongBetween(Math.max(1, metadataVersion - 3), metadataVersion + 5);
99-
final ClusterStateRequestBuilder clusterStateRequestBuilder = client(node).admin()
100-
.cluster()
101-
.prepareState(TEST_REQUEST_TIMEOUT)
102-
.clear()
103-
.setNodes(true)
104-
.setMetadata(true)
105-
.setBlocks(true)
106-
.setMasterNodeTimeout(TimeValue.timeValueMillis(100))
107-
.setWaitForTimeOut(TimeValue.timeValueMillis(100))
108-
.setWaitForMetadataVersion(waitForMetadataVersion);
109-
final ClusterStateResponse clusterStateResponse;
110-
try {
111-
clusterStateResponse = clusterStateRequestBuilder.get();
112-
} catch (MasterNotDiscoveredException e) {
113-
return; // ok, we hit the disconnected node
114-
}
115-
if (clusterStateResponse.isWaitForTimedOut() == false) {
116-
final ClusterState state = clusterStateResponse.getState();
117-
assertNotNull("should always contain a master node", state.nodes().getMasterNodeId());
118-
assertThat("waited for metadata version", state.metadata().version(), greaterThanOrEqualTo(waitForMetadataVersion));
119-
}
120-
});
121-
}
122-
12369
public void testLocalRequestWaitsForMetadata() throws Exception {
12470
runRepeatedlyWhileChangingMaster(() -> {
12571
final String node = randomFrom(internalCluster().getNodeNames());
@@ -131,13 +77,11 @@ public void testLocalRequestWaitsForMetadata() throws Exception {
13177
final long waitForMetadataVersion = randomLongBetween(Math.max(1, metadataVersion - 3), metadataVersion + 5);
13278
final ClusterStateResponse clusterStateResponse = client(node).admin()
13379
.cluster()
134-
.prepareState(TEST_REQUEST_TIMEOUT)
80+
.prepareState(TimeValue.timeValueMillis(100))
13581
.clear()
136-
.setLocal(true)
13782
.setMetadata(true)
13883
.setBlocks(true)
13984
.setWaitForMetadataVersion(waitForMetadataVersion)
140-
.setMasterNodeTimeout(TimeValue.timeValueMillis(100))
14185
.setWaitForTimeOut(TimeValue.timeValueMillis(100))
14286
.get();
14387
if (clusterStateResponse.isWaitForTimedOut() == false) {

‎server/src/internalClusterTest/java/org/elasticsearch/action/support/ActiveShardsObserverIT.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -135,9 +135,7 @@ public void testCreateIndexStopsWaitingWhenIndexDeleted() throws Exception {
135135
.execute();
136136

137137
logger.info("--> wait until the cluster state contains the new index");
138-
assertBusy(
139-
() -> assertTrue(clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState().metadata().getProject().hasIndex(indexName))
140-
);
138+
awaitClusterState(state -> state.metadata().getProject().hasIndex(indexName));
141139

142140
logger.info("--> delete the index");
143141
assertAcked(indicesAdmin().prepareDelete(indexName));

‎server/src/internalClusterTest/java/org/elasticsearch/cluster/MinimumMasterNodesIT.java‎

Lines changed: 13 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414
import org.elasticsearch.action.admin.cluster.configuration.TransportAddVotingConfigExclusionsAction;
1515
import org.elasticsearch.action.admin.cluster.configuration.TransportClearVotingConfigExclusionsAction;
1616
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
17-
import org.elasticsearch.client.internal.Client;
1817
import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
1918
import org.elasticsearch.cluster.coordination.NoMasterBlockService;
2019
import org.elasticsearch.cluster.metadata.Metadata;
@@ -68,7 +67,7 @@ public void testTwoNodesNoMasterBlock() throws Exception {
6867
String node1Name = internalCluster().startNode(settings);
6968

7069
logger.info("--> should be blocked, no master...");
71-
ClusterState state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
70+
ClusterState state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
7271
assertThat(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(true));
7372
assertThat(state.nodes().getSize(), equalTo(1)); // verify that we still see the local node in the cluster state
7473

@@ -81,9 +80,7 @@ public void testTwoNodesNoMasterBlock() throws Exception {
8180
.get();
8281
assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
8382

84-
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
85-
assertThat(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(false));
86-
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
83+
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
8784
assertThat(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(false));
8885

8986
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
@@ -123,12 +120,9 @@ public void testTwoNodesNoMasterBlock() throws Exception {
123120
Settings masterDataPathSettings = internalCluster().dataPathSettings(masterNode);
124121
internalCluster().stopNode(masterNode);
125122

126-
assertBusy(() -> {
127-
ClusterState clusterState = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
128-
assertTrue(clusterState.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
129-
});
123+
awaitClusterState(otherNode, clusterState -> clusterState.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
130124

131-
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
125+
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
132126
assertThat(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(true));
133127
// verify that both nodes are still in the cluster state but there is no master
134128
assertThat(state.nodes().getSize(), equalTo(2));
@@ -144,9 +138,7 @@ public void testTwoNodesNoMasterBlock() throws Exception {
144138
.get();
145139
assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
146140

147-
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
148-
assertThat(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(false));
149-
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
141+
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
150142
assertThat(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(false));
151143

152144
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
@@ -176,10 +168,7 @@ public void testTwoNodesNoMasterBlock() throws Exception {
176168
Settings otherNodeDataPathSettings = internalCluster().dataPathSettings(otherNode);
177169
internalCluster().stopNode(otherNode);
178170

179-
assertBusy(() -> {
180-
ClusterState state1 = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
181-
assertThat(state1.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(true));
182-
});
171+
awaitClusterState(masterNode, clusterState -> clusterState.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
183172

184173
logger.info("--> starting the previous master node again...");
185174
internalCluster().startNode(Settings.builder().put(settings).put(otherNodeDataPathSettings).build());
@@ -192,9 +181,7 @@ public void testTwoNodesNoMasterBlock() throws Exception {
192181
.get();
193182
assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
194183

195-
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
196-
assertThat(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(false));
197-
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
184+
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
198185
assertThat(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(false));
199186

200187
state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
@@ -220,12 +207,9 @@ public void testThreeNodesNoMasterBlock() throws Exception {
220207

221208
ClusterState state;
222209

223-
assertBusy(() -> {
224-
for (Client client : clients()) {
225-
ClusterState state1 = client.admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
226-
assertThat(state1.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(true));
227-
}
228-
});
210+
for (var node : internalCluster().getNodeNames()) {
211+
awaitClusterState(node, clusterState -> clusterState.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
212+
}
229213

230214
logger.info("--> start one more node");
231215
internalCluster().startNode(settings);
@@ -261,8 +245,9 @@ public void testThreeNodesNoMasterBlock() throws Exception {
261245
assertHitCount(prepareSearch().setSize(0).setQuery(QueryBuilders.matchAllQuery()), 100);
262246
}
263247

248+
final var masterName = internalCluster().getMasterName();
264249
List<String> nonMasterNodes = new ArrayList<>(
265-
Sets.difference(Sets.newHashSet(internalCluster().getNodeNames()), Collections.singleton(internalCluster().getMasterName()))
250+
Sets.difference(Sets.newHashSet(internalCluster().getNodeNames()), Collections.singleton(masterName))
266251
);
267252
Settings nonMasterDataPathSettings1 = internalCluster().dataPathSettings(nonMasterNodes.get(0));
268253
Settings nonMasterDataPathSettings2 = internalCluster().dataPathSettings(nonMasterNodes.get(1));
@@ -271,10 +256,7 @@ public void testThreeNodesNoMasterBlock() throws Exception {
271256

272257
logger.info("--> verify that there is no master anymore on remaining node");
273258
// spin here to wait till the state is set
274-
assertBusy(() -> {
275-
ClusterState st = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
276-
assertThat(st.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID), equalTo(true));
277-
});
259+
awaitClusterState(masterName, clusterState -> clusterState.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
278260

279261
logger.info("--> start back the 2 nodes ");
280262
internalCluster().startNodes(nonMasterDataPathSettings1, nonMasterDataPathSettings2);

‎server/src/internalClusterTest/java/org/elasticsearch/cluster/NoMasterNodeIT.java‎

Lines changed: 10 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -82,17 +82,10 @@ public void testNoMasterActions() throws Exception {
8282
internalCluster().setDisruptionScheme(disruptionScheme);
8383
disruptionScheme.startDisrupting();
8484

85-
final Client clientToMasterlessNode = client();
86-
87-
assertBusy(() -> {
88-
ClusterState state = clientToMasterlessNode.admin()
89-
.cluster()
90-
.prepareState(TEST_REQUEST_TIMEOUT)
91-
.setLocal(true)
92-
.get()
93-
.getState();
94-
assertTrue(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
95-
});
85+
final String masterlessNode = internalCluster().getRandomNodeName();
86+
final Client clientToMasterlessNode = client(masterlessNode);
87+
88+
awaitClusterState(masterlessNode, state -> state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
9689

9790
assertRequestBuilderThrows(
9891
clientToMasterlessNode.prepareGet("test", "1"),
@@ -246,17 +239,10 @@ public void testNoMasterActionsWriteMasterBlock() throws Exception {
246239
internalCluster().setDisruptionScheme(disruptionScheme);
247240
disruptionScheme.startDisrupting();
248241

249-
final Client clientToMasterlessNode = client();
242+
final String masterlessNode = internalCluster().getRandomNodeName();
243+
final Client clientToMasterlessNode = client(masterlessNode);
250244

251-
assertBusy(() -> {
252-
ClusterState state = clientToMasterlessNode.admin()
253-
.cluster()
254-
.prepareState(TEST_REQUEST_TIMEOUT)
255-
.setLocal(true)
256-
.get()
257-
.getState();
258-
assertTrue(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
259-
});
245+
awaitClusterState(masterlessNode, state -> state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
260246

261247
GetResponse getResponse = clientToMasterlessNode.prepareGet("test1", "1").get();
262248
assertExists(getResponse);
@@ -346,12 +332,9 @@ public void testNoMasterActionsMetadataWriteMasterBlock() throws Exception {
346332
internalCluster().setDisruptionScheme(disruptionScheme);
347333
disruptionScheme.startDisrupting();
348334

349-
assertBusy(() -> {
350-
for (String node : nodesWithShards) {
351-
ClusterState state = client(node).admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
352-
assertTrue(state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
353-
}
354-
});
335+
for (String node : nodesWithShards) {
336+
awaitClusterState(node, state -> state.blocks().hasGlobalBlockWithId(NoMasterBlockService.NO_MASTER_BLOCK_ID));
337+
}
355338

356339
GetResponse getResponse = client(randomFrom(nodesWithShards)).prepareGet("test1", "1").get();
357340
assertExists(getResponse);

‎server/src/internalClusterTest/java/org/elasticsearch/cluster/coordination/InitialClusterStateIT.java‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -34,13 +34,7 @@ protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
3434

3535
private static void assertClusterUuid(boolean expectCommitted, String expectedValue) {
3636
for (String nodeName : internalCluster().getNodeNames()) {
37-
final Metadata metadata = client(nodeName).admin()
38-
.cluster()
39-
.prepareState(TEST_REQUEST_TIMEOUT)
40-
.setLocal(true)
41-
.get()
42-
.getState()
43-
.metadata();
37+
final Metadata metadata = client(nodeName).admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).get().getState().metadata();
4438
assertEquals(expectCommitted, metadata.clusterUUIDCommitted());
4539
assertEquals(expectedValue, metadata.clusterUUID());
4640

0 commit comments

Comments
 (0)