Skip to content

Commit 8f7c573

Browse files
ywelschywelsch
committed
Use default discovery implementation for single-node discovery (#40036)
Switches "discovery.type: single-node" from using a separate implementation for single-node discovery to using the existing standard discovery implementation, with two small adaptions: - auto-bootstrapping, but requiring initial_master_nodes not to be set. - not actively pinging other nodes using the Peerfinder - not allowing other nodes to join its single-node cluster (if they have e.g. been set up using regular discovery and connect to the single-disco node).
1 parent 3860ddd commit 8f7c573

File tree

10 files changed

+258
-328
lines changed

10 files changed

+258
-328
lines changed

‎server/src/main/java/org/elasticsearch/cluster/coordination/ClusterBootstrapService.java‎

Lines changed: 23 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -29,10 +29,13 @@
2929
import org.elasticsearch.common.settings.Setting.Property;
3030
import org.elasticsearch.common.settings.Settings;
3131
import org.elasticsearch.common.unit.TimeValue;
32+
import org.elasticsearch.discovery.DiscoveryModule;
33+
import org.elasticsearch.node.Node;
3234
import org.elasticsearch.threadpool.ThreadPool.Names;
3335
import org.elasticsearch.transport.TransportService;
3436

3537
import java.util.ArrayList;
38+
import java.util.Collections;
3639
import java.util.HashSet;
3740
import java.util.LinkedHashSet;
3841
import java.util.List;
@@ -77,15 +80,28 @@ public class ClusterBootstrapService {
7780
public ClusterBootstrapService(Settings settings, TransportService transportService,
7881
Supplier<Iterable<DiscoveryNode>> discoveredNodesSupplier, BooleanSupplier isBootstrappedSupplier,
7982
Consumer<VotingConfiguration> votingConfigurationConsumer) {
80-
81-
final List<String> initialMasterNodes = INITIAL_MASTER_NODES_SETTING.get(settings);
82-
bootstrapRequirements = unmodifiableSet(new LinkedHashSet<>(initialMasterNodes));
83-
if (bootstrapRequirements.size() != initialMasterNodes.size()) {
84-
throw new IllegalArgumentException(
85-
"setting [" + INITIAL_MASTER_NODES_SETTING.getKey() + "] contains duplicates: " + initialMasterNodes);
83+
if (DiscoveryModule.SINGLE_NODE_DISCOVERY_TYPE.equals(DiscoveryModule.DISCOVERY_TYPE_SETTING.get(settings))) {
84+
if (INITIAL_MASTER_NODES_SETTING.exists(settings)) {
85+
throw new IllegalArgumentException("setting [" + INITIAL_MASTER_NODES_SETTING.getKey() +
86+
"] is not allowed when [" + DiscoveryModule.DISCOVERY_TYPE_SETTING.getKey() + "] is set to [" +
87+
DiscoveryModule.SINGLE_NODE_DISCOVERY_TYPE + "]");
88+
}
89+
if (DiscoveryNode.isMasterNode(settings) == false) {
90+
throw new IllegalArgumentException("node with [" + DiscoveryModule.DISCOVERY_TYPE_SETTING.getKey() + "] set to [" +
91+
DiscoveryModule.SINGLE_NODE_DISCOVERY_TYPE + "] must be master-eligible");
92+
}
93+
bootstrapRequirements = Collections.singleton(Node.NODE_NAME_SETTING.get(settings));
94+
unconfiguredBootstrapTimeout = null;
95+
} else {
96+
final List<String> initialMasterNodes = INITIAL_MASTER_NODES_SETTING.get(settings);
97+
bootstrapRequirements = unmodifiableSet(new LinkedHashSet<>(initialMasterNodes));
98+
if (bootstrapRequirements.size() != initialMasterNodes.size()) {
99+
throw new IllegalArgumentException(
100+
"setting [" + INITIAL_MASTER_NODES_SETTING.getKey() + "] contains duplicates: " + initialMasterNodes);
101+
}
102+
unconfiguredBootstrapTimeout = discoveryIsConfigured(settings) ? null : UNCONFIGURED_BOOTSTRAP_TIMEOUT_SETTING.get(settings);
86103
}
87104

88-
unconfiguredBootstrapTimeout = discoveryIsConfigured(settings) ? null : UNCONFIGURED_BOOTSTRAP_TIMEOUT_SETTING.get(settings);
89105
this.transportService = transportService;
90106
this.discoveredNodesSupplier = discoveredNodesSupplier;
91107
this.isBootstrappedSupplier = isBootstrappedSupplier;

‎server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,12 +53,14 @@
5353
import org.elasticsearch.common.settings.ClusterSettings;
5454
import org.elasticsearch.common.settings.Setting;
5555
import org.elasticsearch.common.settings.Settings;
56+
import org.elasticsearch.common.transport.TransportAddress;
5657
import org.elasticsearch.common.unit.TimeValue;
5758
import org.elasticsearch.common.util.concurrent.EsExecutors;
5859
import org.elasticsearch.common.util.concurrent.ListenableFuture;
5960
import org.elasticsearch.common.xcontent.XContentHelper;
6061
import org.elasticsearch.common.xcontent.json.JsonXContent;
6162
import org.elasticsearch.discovery.Discovery;
63+
import org.elasticsearch.discovery.DiscoveryModule;
6264
import org.elasticsearch.discovery.DiscoveryStats;
6365
import org.elasticsearch.discovery.HandshakingTransportAddressConnector;
6466
import org.elasticsearch.discovery.PeerFinder;
@@ -72,6 +74,7 @@
7274

7375
import java.util.ArrayList;
7476
import java.util.Collection;
77+
import java.util.Collections;
7578
import java.util.HashSet;
7679
import java.util.List;
7780
import java.util.Optional;
@@ -99,6 +102,7 @@ public class Coordinator extends AbstractLifecycleComponent implements Discovery
99102
TimeValue.timeValueMillis(30000), TimeValue.timeValueMillis(1), Setting.Property.NodeScope);
100103

101104
private final Settings settings;
105+
private final boolean singleNodeDiscovery;
102106
private final TransportService transportService;
103107
private final MasterService masterService;
104108
private final AllocationService allocationService;
@@ -149,6 +153,7 @@ public Coordinator(String nodeName, Settings settings, ClusterSettings clusterSe
149153
this.masterService = masterService;
150154
this.allocationService = allocationService;
151155
this.onJoinValidators = JoinTaskExecutor.addBuiltInJoinValidators(onJoinValidators);
156+
this.singleNodeDiscovery = DiscoveryModule.SINGLE_NODE_DISCOVERY_TYPE.equals(DiscoveryModule.DISCOVERY_TYPE_SETTING.get(settings));
152157
this.joinHelper = new JoinHelper(settings, allocationService, masterService, transportService,
153158
this::getCurrentTerm, this::getStateForMasterService, this::handleJoinRequest, this::joinLeaderInTerm, this.onJoinValidators);
154159
this.persistedStateSupplier = persistedStateSupplier;
@@ -448,6 +453,13 @@ private void handleJoinRequest(JoinRequest joinRequest, JoinHelper.JoinCallback
448453
assert Thread.holdsLock(mutex) == false;
449454
assert getLocalNode().isMasterNode() : getLocalNode() + " received a join but is not master-eligible";
450455
logger.trace("handleJoinRequest: as {}, handling {}", mode, joinRequest);
456+
457+
if (singleNodeDiscovery && joinRequest.getSourceNode().equals(getLocalNode()) == false) {
458+
joinCallback.onFailure(new IllegalStateException("cannot join node with [" + DiscoveryModule.DISCOVERY_TYPE_SETTING.getKey() +
459+
"] set to [" + DiscoveryModule.SINGLE_NODE_DISCOVERY_TYPE + "] discovery"));
460+
return;
461+
}
462+
451463
transportService.connectToNode(joinRequest.getSourceNode());
452464

453465
final ClusterState stateForJoinValidation = getStateForMasterService();
@@ -666,6 +678,14 @@ protected void doStart() {
666678
coordinationState.set(new CoordinationState(settings, getLocalNode(), persistedState));
667679
peerFinder.setCurrentTerm(getCurrentTerm());
668680
configuredHostsResolver.start();
681+
VotingConfiguration votingConfiguration = coordinationState.get().getLastAcceptedState().getLastCommittedConfiguration();
682+
if (singleNodeDiscovery &&
683+
votingConfiguration.isEmpty() == false &&
684+
votingConfiguration.hasQuorum(Collections.singleton(getLocalNode().getId())) == false) {
685+
throw new IllegalStateException("cannot start with [" + DiscoveryModule.DISCOVERY_TYPE_SETTING.getKey() + "] set to [" +
686+
DiscoveryModule.SINGLE_NODE_DISCOVERY_TYPE + "] when local node " + getLocalNode() +
687+
" does not have quorum in voting configuration " + votingConfiguration);
688+
}
669689
ClusterState initialState = ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.get(settings))
670690
.blocks(ClusterBlocks.builder()
671691
.addGlobalBlock(STATE_NOT_RECOVERED_BLOCK)
@@ -1079,7 +1099,8 @@ private class CoordinatorPeerFinder extends PeerFinder {
10791099

10801100
CoordinatorPeerFinder(Settings settings, TransportService transportService, TransportAddressConnector transportAddressConnector,
10811101
ConfiguredHostsResolver configuredHostsResolver) {
1082-
super(settings, transportService, transportAddressConnector, configuredHostsResolver);
1102+
super(settings, transportService, transportAddressConnector,
1103+
singleNodeDiscovery ? hostsResolver -> Collections.emptyList() : configuredHostsResolver);
10831104
}
10841105

10851106
@Override
@@ -1090,6 +1111,13 @@ protected void onActiveMasterFound(DiscoveryNode masterNode, long term) {
10901111
}
10911112
}
10921113

1114+
@Override
1115+
protected void startProbe(TransportAddress transportAddress) {
1116+
if (singleNodeDiscovery == false) {
1117+
super.startProbe(transportAddress);
1118+
}
1119+
}
1120+
10931121
@Override
10941122
protected void onFoundPeersUpdated() {
10951123
synchronized (mutex) {

‎server/src/main/java/org/elasticsearch/discovery/DiscoveryModule.java‎

Lines changed: 15 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@
3636
import org.elasticsearch.common.settings.Setting.Property;
3737
import org.elasticsearch.common.settings.Settings;
3838
import org.elasticsearch.common.transport.TransportAddress;
39-
import org.elasticsearch.discovery.single.SingleNodeDiscovery;
4039
import org.elasticsearch.discovery.zen.ZenDiscovery;
4140
import org.elasticsearch.gateway.GatewayMetaState;
4241
import org.elasticsearch.plugins.DiscoveryPlugin;
@@ -51,7 +50,6 @@
5150
import java.util.HashSet;
5251
import java.util.List;
5352
import java.util.Map;
54-
import java.util.Objects;
5553
import java.util.Random;
5654
import java.util.Set;
5755
import java.util.function.BiConsumer;
@@ -70,6 +68,8 @@ public class DiscoveryModule {
7068
public static final String ZEN_DISCOVERY_TYPE = "legacy-zen";
7169
public static final String ZEN2_DISCOVERY_TYPE = "zen";
7270

71+
public static final String SINGLE_NODE_DISCOVERY_TYPE = "single-node";
72+
7373
public static final Setting<String> DISCOVERY_TYPE_SETTING =
7474
new Setting<>("discovery.type", ZEN2_DISCOVERY_TYPE, Function.identity(), Property.NodeScope);
7575
public static final Setting<List<String>> LEGACY_DISCOVERY_HOSTS_PROVIDER_SETTING =
@@ -119,6 +119,8 @@ public DiscoveryModule(Settings settings, ThreadPool threadPool, TransportServic
119119
List<SeedHostsProvider> filteredSeedProviders = seedProviderNames.stream()
120120
.map(hostProviders::get).map(Supplier::get).collect(Collectors.toList());
121121

122+
String discoveryType = DISCOVERY_TYPE_SETTING.get(settings);
123+
122124
final SeedHostsProvider seedHostsProvider = hostsResolver -> {
123125
final List<TransportAddress> addresses = new ArrayList<>();
124126
for (SeedHostsProvider provider : filteredSeedProviders) {
@@ -127,23 +129,20 @@ public DiscoveryModule(Settings settings, ThreadPool threadPool, TransportServic
127129
return Collections.unmodifiableList(addresses);
128130
};
129131

130-
Map<String, Supplier<Discovery>> discoveryTypes = new HashMap<>();
131-
discoveryTypes.put(ZEN_DISCOVERY_TYPE,
132-
() -> new ZenDiscovery(settings, threadPool, transportService, namedWriteableRegistry, masterService, clusterApplier,
133-
clusterSettings, seedHostsProvider, allocationService, joinValidators, gatewayMetaState));
134-
discoveryTypes.put(ZEN2_DISCOVERY_TYPE, () -> new Coordinator(NODE_NAME_SETTING.get(settings), settings, clusterSettings,
135-
transportService, namedWriteableRegistry, allocationService, masterService,
136-
() -> gatewayMetaState.getPersistedState(settings, (ClusterApplierService) clusterApplier), seedHostsProvider, clusterApplier,
137-
joinValidators, new Random(Randomness.get().nextLong())));
138-
discoveryTypes.put("single-node", () -> new SingleNodeDiscovery(settings, transportService, masterService, clusterApplier,
139-
gatewayMetaState));
140-
String discoveryType = DISCOVERY_TYPE_SETTING.get(settings);
141-
Supplier<Discovery> discoverySupplier = discoveryTypes.get(discoveryType);
142-
if (discoverySupplier == null) {
132+
if (ZEN2_DISCOVERY_TYPE.equals(discoveryType) || SINGLE_NODE_DISCOVERY_TYPE.equals(discoveryType)) {
133+
discovery = new Coordinator(NODE_NAME_SETTING.get(settings),
134+
settings, clusterSettings,
135+
transportService, namedWriteableRegistry, allocationService, masterService,
136+
() -> gatewayMetaState.getPersistedState(settings, (ClusterApplierService) clusterApplier), seedHostsProvider,
137+
clusterApplier, joinValidators, new Random(Randomness.get().nextLong()));
138+
} else if (ZEN_DISCOVERY_TYPE.equals(discoveryType)) {
139+
discovery = new ZenDiscovery(settings, threadPool, transportService, namedWriteableRegistry, masterService, clusterApplier,
140+
clusterSettings, seedHostsProvider, allocationService, joinValidators, gatewayMetaState);
141+
} else {
143142
throw new IllegalArgumentException("Unknown discovery type [" + discoveryType + "]");
144143
}
144+
145145
logger.info("using discovery type [{}] and seed hosts providers {}", discoveryType, seedProviderNames);
146-
discovery = Objects.requireNonNull(discoverySupplier.get());
147146
}
148147

149148
private List<String> getSeedProviderNames(Settings settings) {

‎server/src/main/java/org/elasticsearch/discovery/PeerFinder.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -308,7 +308,7 @@ public String toString() {
308308
return peersRemoved;
309309
}
310310

311-
private void startProbe(TransportAddress transportAddress) {
311+
protected void startProbe(TransportAddress transportAddress) {
312312
assert holdsLock() : "PeerFinder mutex not held";
313313
if (active == false) {
314314
logger.trace("startProbe({}) not running", transportAddress);

‎server/src/main/java/org/elasticsearch/discovery/single/SingleNodeDiscovery.java‎

Lines changed: 0 additions & 139 deletions
This file was deleted.

0 commit comments

Comments
 (0)