Remove DiscoveryNode#shouldConnectTo method
This method was originally introduced to prevent client nodes from connecting to other client nodes directly in the cluster. That said, it worked only if node.client was set to true and not when node.master and node.data were both set to false. It looks safe to remove, which allows to solve all kinds of problems around monitoring that happen wherever there are 2 or more clients nodes in the cluster, and a REST call hits one of them (node counts are off, clients nodes are missing).
This commit is contained in:
javanna
Luca Cavanna
parent
1e70e16c89
commit
04947a5fef
|
|
@ -110,7 +110,7 @@ public abstract class TransportNodesAction<NodesRequest extends BaseNodesRequest
|
|||
}
|
||||
|
||||
|
||||
private class AsyncAction {
|
||||
class AsyncAction {
|
||||
|
||||
private final NodesRequest request;
|
||||
private final String[] nodesIds;
|
||||
|
|
@ -120,7 +120,7 @@ public abstract class TransportNodesAction<NodesRequest extends BaseNodesRequest
|
|||
private final AtomicInteger counter = new AtomicInteger();
|
||||
private final Task task;
|
||||
|
||||
private AsyncAction(Task task, NodesRequest request, ActionListener<NodesResponse> listener) {
|
||||
AsyncAction(Task task, NodesRequest request, ActionListener<NodesResponse> listener) {
|
||||
this.task = task;
|
||||
this.request = request;
|
||||
this.listener = listener;
|
||||
|
|
@ -135,7 +135,7 @@ public abstract class TransportNodesAction<NodesRequest extends BaseNodesRequest
|
|||
this.responses = new AtomicReferenceArray<>(this.nodesIds.length);
|
||||
}
|
||||
|
||||
private void start() {
|
||||
void start() {
|
||||
if (nodesIds.length == 0) {
|
||||
// nothing to notify
|
||||
threadPool.generic().execute(new Runnable() {
|
||||
|
|
@ -158,11 +158,6 @@ public abstract class TransportNodesAction<NodesRequest extends BaseNodesRequest
|
|||
try {
|
||||
if (node == null) {
|
||||
onFailure(idx, nodeId, new NoSuchNodeException(nodeId));
|
||||
} else if (!clusterService.localNode().shouldConnectTo(node) && !clusterService.localNode().equals(node)) {
|
||||
// the check "!clusterService.localNode().equals(node)" is to maintain backward comp. where before
|
||||
// we allowed to connect from "local" client node to itself, certain tests rely on it, if we remove it, we need to fix
|
||||
// those (and they randomize the client node usage, so tricky to find when)
|
||||
onFailure(idx, nodeId, new NodeShouldNotConnectException(clusterService.localNode(), node));
|
||||
} else {
|
||||
ChildTaskRequest nodeRequest = newNodeRequest(nodeId, request);
|
||||
if (task != null) {
|
||||
|
|
|
|||
|
|
@ -235,12 +235,6 @@ public abstract class TransportTasksAction<
|
|||
try {
|
||||
if (node == null) {
|
||||
onFailure(idx, nodeId, new NoSuchNodeException(nodeId));
|
||||
} else if (!clusterService.localNode().shouldConnectTo(node) && !clusterService.localNode().equals(node)) {
|
||||
// the check "!clusterService.localNode().equals(node)" is to maintain backward comp. where before
|
||||
// we allowed to connect from "local" client node to itself, certain tests rely on it, if we remove it, we
|
||||
// need to fix
|
||||
// those (and they randomize the client node usage, so tricky to find when)
|
||||
onFailure(idx, nodeId, new NodeShouldNotConnectException(clusterService.localNode(), node));
|
||||
} else {
|
||||
NodeTaskRequest nodeRequest = new NodeTaskRequest(request);
|
||||
nodeRequest.setParentTask(clusterService.localNode().id(), task.getId());
|
||||
|
|
|
|||
|
|
@ -209,16 +209,6 @@ public class DiscoveryNode implements Streamable, ToXContent {
|
|||
this.version = version;
|
||||
}
|
||||
|
||||
/**
|
||||
* Should this node form a connection to the provided node.
|
||||
*/
|
||||
public boolean shouldConnectTo(DiscoveryNode otherNode) {
|
||||
if (clientNode() && otherNode.clientNode()) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* The address that the node can be communicated with.
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -570,9 +570,6 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe
|
|||
|
||||
// TODO, do this in parallel (and wait)
|
||||
for (DiscoveryNode node : nodesDelta.addedNodes()) {
|
||||
if (!nodeRequiresConnection(node)) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
transportService.connectToNode(node);
|
||||
} catch (Throwable e) {
|
||||
|
|
@ -824,9 +821,6 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe
|
|||
if (lifecycle.stoppedOrClosed()) {
|
||||
return;
|
||||
}
|
||||
if (!nodeRequiresConnection(node)) {
|
||||
continue;
|
||||
}
|
||||
if (clusterState.nodes().nodeExists(node.id())) { // we double check existence of node since connectToNode might take time...
|
||||
if (!transportService.nodeConnected(node)) {
|
||||
try {
|
||||
|
|
@ -873,10 +867,6 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe
|
|||
return Strings.randomBase64UUID(random);
|
||||
}
|
||||
|
||||
private boolean nodeRequiresConnection(DiscoveryNode node) {
|
||||
return localNode().shouldConnectTo(node);
|
||||
}
|
||||
|
||||
private static class LocalNodeMasterListeners implements ClusterStateListener {
|
||||
|
||||
private final List<LocalNodeMasterListener> listeners = new CopyOnWriteArrayList<>();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,226 @@
|
|||
/*
|
||||
* Licensed to Elasticsearch under one or more contributor
|
||||
* license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright
|
||||
* ownership. Elasticsearch licenses this file to you under
|
||||
* the Apache License, Version 2.0 (the "License"); you may
|
||||
* not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.action.support.nodes;
|
||||
|
||||
import org.elasticsearch.Version;
|
||||
import org.elasticsearch.action.support.ActionFilters;
|
||||
import org.elasticsearch.action.support.PlainActionFuture;
|
||||
import org.elasticsearch.action.support.broadcast.node.TransportBroadcastByNodeActionTests;
|
||||
import org.elasticsearch.cluster.ClusterName;
|
||||
import org.elasticsearch.cluster.ClusterService;
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNode;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNodes;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.transport.DummyTransportAddress;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
import org.elasticsearch.test.cluster.TestClusterService;
|
||||
import org.elasticsearch.test.transport.CapturingTransport;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicReferenceArray;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
public class TransportNodesActionTests extends ESTestCase {
|
||||
|
||||
private static ThreadPool THREAD_POOL;
|
||||
private static ClusterName CLUSTER_NAME = new ClusterName("test-cluster");
|
||||
|
||||
private TestClusterService clusterService;
|
||||
private CapturingTransport transport;
|
||||
private TestTransportNodesAction action;
|
||||
|
||||
public void testRequestIsSentToEachNode() throws Exception {
|
||||
TestNodesRequest request = new TestNodesRequest();
|
||||
PlainActionFuture<TestNodesResponse> listener = new PlainActionFuture<>();
|
||||
action.new AsyncAction(null, request, listener).start();
|
||||
Map<String, List<CapturingTransport.CapturedRequest>> capturedRequests = transport.getCapturedRequestsByTargetNodeAndClear();
|
||||
int numNodes = clusterService.state().getNodes().size();
|
||||
// check a request was sent to the right number of nodes
|
||||
assertEquals(numNodes, capturedRequests.size());
|
||||
}
|
||||
|
||||
public void testNodesSelectors() {
|
||||
int numSelectors = randomIntBetween(1, 5);
|
||||
Set<String> nodeSelectors = new HashSet<>();
|
||||
for (int i = 0; i < numSelectors; i++) {
|
||||
nodeSelectors.add(randomFrom(NodeSelector.values()).selector);
|
||||
}
|
||||
int numNodeIds = randomIntBetween(0, 3);
|
||||
String[] nodeIds = clusterService.state().nodes().nodes().keys().toArray(String.class);
|
||||
for (int i = 0; i < numNodeIds; i++) {
|
||||
String nodeId = randomFrom(nodeIds);
|
||||
nodeSelectors.add(nodeId);
|
||||
}
|
||||
String[] finalNodesIds = nodeSelectors.toArray(new String[nodeSelectors.size()]);
|
||||
TestNodesRequest request = new TestNodesRequest(finalNodesIds);
|
||||
action.new AsyncAction(null, request, new PlainActionFuture<>()).start();
|
||||
Map<String, List<CapturingTransport.CapturedRequest>> capturedRequests = transport.getCapturedRequestsByTargetNodeAndClear();
|
||||
assertEquals(clusterService.state().nodes().resolveNodesIds(finalNodesIds).length, capturedRequests.size());
|
||||
}
|
||||
|
||||
private enum NodeSelector {
|
||||
LOCAL("_local"), ELECTED_MASTER("_master"), MASTER_ELIGIBLE("master:true"), DATA("data:true"), CUSTOM_ATTRIBUTE("attr:value");
|
||||
|
||||
private final String selector;
|
||||
|
||||
NodeSelector(String selector) {
|
||||
this.selector = selector;
|
||||
}
|
||||
}
|
||||
|
||||
@BeforeClass
|
||||
public static void startThreadPool() {
|
||||
THREAD_POOL = new ThreadPool(TransportBroadcastByNodeActionTests.class.getSimpleName());
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void destroyThreadPool() {
|
||||
ThreadPool.terminate(THREAD_POOL, 30, TimeUnit.SECONDS);
|
||||
// since static must set to null to be eligible for collection
|
||||
THREAD_POOL = null;
|
||||
}
|
||||
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
super.setUp();
|
||||
transport = new CapturingTransport();
|
||||
clusterService = new TestClusterService(THREAD_POOL);
|
||||
final TransportService transportService = new TransportService(transport, THREAD_POOL);
|
||||
transportService.start();
|
||||
transportService.acceptIncomingRequests();
|
||||
int numNodes = randomIntBetween(3, 10);
|
||||
DiscoveryNodes.Builder discoBuilder = DiscoveryNodes.builder();
|
||||
List<DiscoveryNode> discoveryNodes = new ArrayList<>();
|
||||
for (int i = 0; i < numNodes; i++) {
|
||||
Map<String, String> attributes = new HashMap<>();
|
||||
if (randomBoolean()) {
|
||||
attributes.put("master", Boolean.toString(randomBoolean()));
|
||||
attributes.put("data", Boolean.toString(randomBoolean()));
|
||||
attributes.put("ingest", Boolean.toString(randomBoolean()));
|
||||
} else {
|
||||
attributes.put("client", "true");
|
||||
}
|
||||
if (frequently()) {
|
||||
attributes.put("custom", randomBoolean() ? "match" : randomAsciiOfLengthBetween(3, 5));
|
||||
}
|
||||
final DiscoveryNode node = newNode(i, attributes);
|
||||
discoBuilder = discoBuilder.put(node);
|
||||
discoveryNodes.add(node);
|
||||
}
|
||||
discoBuilder.localNodeId(randomFrom(discoveryNodes).id());
|
||||
discoBuilder.masterNodeId(randomFrom(discoveryNodes).id());
|
||||
ClusterState.Builder stateBuilder = ClusterState.builder(CLUSTER_NAME);
|
||||
stateBuilder.nodes(discoBuilder);
|
||||
ClusterState clusterState = stateBuilder.build();
|
||||
clusterService.setState(clusterState);
|
||||
action = new TestTransportNodesAction(
|
||||
Settings.EMPTY,
|
||||
THREAD_POOL,
|
||||
clusterService,
|
||||
transportService,
|
||||
new ActionFilters(Collections.emptySet()),
|
||||
TestNodesRequest::new,
|
||||
TestNodeRequest::new,
|
||||
ThreadPool.Names.SAME
|
||||
);
|
||||
}
|
||||
|
||||
private static DiscoveryNode newNode(int nodeId, Map<String, String> attributes) {
|
||||
String node = "node_" + nodeId;
|
||||
return new DiscoveryNode(node, node, DummyTransportAddress.INSTANCE, attributes, Version.CURRENT);
|
||||
}
|
||||
|
||||
private static class TestTransportNodesAction extends TransportNodesAction<TestNodesRequest, TestNodesResponse, TestNodeRequest,
|
||||
TestNodeResponse> {
|
||||
|
||||
TestTransportNodesAction(Settings settings, ThreadPool threadPool, ClusterService clusterService, TransportService
|
||||
transportService, ActionFilters actionFilters, Supplier<TestNodesRequest> request,
|
||||
Supplier<TestNodeRequest> nodeRequest, String nodeExecutor) {
|
||||
super(settings, "indices:admin/test", CLUSTER_NAME, threadPool, clusterService, transportService, actionFilters,
|
||||
null, request, nodeRequest, nodeExecutor);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected TestNodesResponse newResponse(TestNodesRequest request, AtomicReferenceArray nodesResponses) {
|
||||
final List<TestNodeResponse> nodeResponses = new ArrayList<>();
|
||||
for (int i = 0; i < nodesResponses.length(); i++) {
|
||||
Object resp = nodesResponses.get(i);
|
||||
if (resp instanceof TestNodeResponse) {
|
||||
nodeResponses.add((TestNodeResponse) resp);
|
||||
}
|
||||
}
|
||||
return new TestNodesResponse(nodeResponses);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected TestNodeRequest newNodeRequest(String nodeId, TestNodesRequest request) {
|
||||
return new TestNodeRequest();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected TestNodeResponse newNodeResponse() {
|
||||
return new TestNodeResponse();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected TestNodeResponse nodeOperation(TestNodeRequest request) {
|
||||
return new TestNodeResponse();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean accumulateExceptions() {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private static class TestNodesRequest extends BaseNodesRequest<TestNodesRequest> {
|
||||
TestNodesRequest(String... nodesIds) {
|
||||
super(nodesIds);
|
||||
}
|
||||
}
|
||||
|
||||
private static class TestNodesResponse extends BaseNodesResponse<TestNodeResponse> {
|
||||
|
||||
private final List<TestNodeResponse> nodeResponses;
|
||||
|
||||
TestNodesResponse(List<TestNodeResponse> nodeResponses) {
|
||||
this.nodeResponses = nodeResponses;
|
||||
}
|
||||
}
|
||||
|
||||
private static class TestNodeRequest extends BaseNodeRequest {
|
||||
}
|
||||
|
||||
private static class TestNodeResponse extends BaseNodeResponse {
|
||||
}
|
||||
}
|
||||
|
|
@ -38,13 +38,8 @@ import static org.hamcrest.Matchers.notNullValue;
|
|||
/**
|
||||
*
|
||||
*/
|
||||
@ClusterScope(scope= Scope.TEST, numDataNodes =0)
|
||||
@ClusterScope(scope= Scope.TEST, numDataNodes = 0)
|
||||
public class SimpleNodesInfoIT extends ESIntegTestCase {
|
||||
static final class Fields {
|
||||
static final String SITE_PLUGIN = "dummy";
|
||||
static final String SITE_PLUGIN_DESCRIPTION = "This is a description for a dummy test site plugin.";
|
||||
static final String SITE_PLUGIN_VERSION = "0.0.7-BOND-SITE";
|
||||
}
|
||||
|
||||
public void testNodesInfos() throws Exception {
|
||||
List<String> nodesIds = internalCluster().startNodesAsync(2).get();
|
||||
|
|
|
|||
Loading…
Reference in New Issue