Skip to content

Commit 1d6630c

Browse files
DaveCTurnerDaveCTurner
committed
Introduce zen2 discovery type
With this change it is now possible to start a node running Zen2.
1 parent ee05ef1 commit 1d6630c

File tree

6 files changed

+48
-32
lines changed

6 files changed

+48
-32
lines changed

‎test/framework/src/main/java/org/elasticsearch/cluster/coordination/InMemoryPersistedState.java renamed to ‎server/src/main/java/org/elasticsearch/cluster/coordination/InMemoryPersistedState.java

File renamed without changes.

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

+24-13
Original file line numberDiff line numberDiff line change
@@ -22,10 +22,13 @@
2222
import org.apache.logging.log4j.Logger;
2323
import org.apache.logging.log4j.LogManager;
2424
import org.elasticsearch.cluster.ClusterState;
25+
import org.elasticsearch.cluster.coordination.Coordinator;
2526
import org.elasticsearch.cluster.node.DiscoveryNode;
2627
import org.elasticsearch.cluster.routing.allocation.AllocationService;
2728
import org.elasticsearch.cluster.service.ClusterApplier;
29+
import org.elasticsearch.cluster.service.ClusterApplierService;
2830
import org.elasticsearch.cluster.service.MasterService;
31+
import org.elasticsearch.common.Randomness;
2932
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
3033
import org.elasticsearch.common.network.NetworkService;
3134
import org.elasticsearch.common.settings.ClusterSettings;
@@ -58,14 +61,19 @@
5861
import java.util.function.Supplier;
5962
import java.util.stream.Collectors;
6063

64+
import static org.elasticsearch.node.Node.NODE_NAME_SETTING;
65+
6166
/**
6267
* A module for loading classes for node discovery.
6368
*/
6469
public class DiscoveryModule {
6570
private static final Logger logger = LogManager.getLogger(DiscoveryModule.class);
6671

72+
public static final String ZEN_DISCOVERY_TYPE = "zen";
73+
public static final String ZEN2_DISCOVERY_TYPE = "zen2";
74+
6775
public static final Setting<String> DISCOVERY_TYPE_SETTING =
68-
new Setting<>("discovery.type", "zen", Function.identity(), Property.NodeScope);
76+
new Setting<>("discovery.type", ZEN_DISCOVERY_TYPE, Function.identity(), Property.NodeScope);
6977
public static final Setting<List<String>> DISCOVERY_HOSTS_PROVIDER_SETTING =
7078
Setting.listSetting("discovery.zen.hosts_provider", Collections.emptyList(), Function.identity(), Property.NodeScope);
7179

@@ -75,14 +83,14 @@ public DiscoveryModule(Settings settings, ThreadPool threadPool, TransportServic
7583
NamedWriteableRegistry namedWriteableRegistry, NetworkService networkService, MasterService masterService,
7684
ClusterApplier clusterApplier, ClusterSettings clusterSettings, List<DiscoveryPlugin> plugins,
7785
AllocationService allocationService, Path configFile, GatewayMetaState gatewayMetaState) {
78-
final Collection<BiConsumer<DiscoveryNode,ClusterState>> joinValidators = new ArrayList<>();
86+
final Collection<BiConsumer<DiscoveryNode, ClusterState>> joinValidators = new ArrayList<>();
7987
final Map<String, Supplier<UnicastHostsProvider>> hostProviders = new HashMap<>();
8088
hostProviders.put("settings", () -> new SettingsBasedHostsProvider(settings, transportService));
8189
hostProviders.put("file", () -> new FileBasedUnicastHostsProvider(configFile));
8290
for (DiscoveryPlugin plugin : plugins) {
83-
plugin.getZenHostsProviders(transportService, networkService).entrySet().forEach(entry -> {
84-
if (hostProviders.put(entry.getKey(), entry.getValue()) != null) {
85-
throw new IllegalArgumentException("Cannot register zen hosts provider [" + entry.getKey() + "] twice");
91+
plugin.getZenHostsProviders(transportService, networkService).forEach((key, value) -> {
92+
if (hostProviders.put(key, value) != null) {
93+
throw new IllegalArgumentException("Cannot register zen hosts provider [" + key + "] twice");
8694
}
8795
});
8896
BiConsumer<DiscoveryNode, ClusterState> joinValidator = plugin.getJoinValidator();
@@ -117,18 +125,21 @@ public DiscoveryModule(Settings settings, ThreadPool threadPool, TransportServic
117125
};
118126

119127
Map<String, Supplier<Discovery>> discoveryTypes = new HashMap<>();
120-
discoveryTypes.put("zen",
128+
discoveryTypes.put(ZEN_DISCOVERY_TYPE,
121129
() -> new ZenDiscovery(settings, threadPool, transportService, namedWriteableRegistry, masterService, clusterApplier,
122130
clusterSettings, hostsProvider, allocationService, Collections.unmodifiableCollection(joinValidators), gatewayMetaState));
131+
discoveryTypes.put(ZEN2_DISCOVERY_TYPE, () -> new Coordinator(NODE_NAME_SETTING.get(settings), settings, clusterSettings,
132+
transportService, namedWriteableRegistry, allocationService, masterService,
133+
() -> gatewayMetaState.getPersistedState(settings, (ClusterApplierService) clusterApplier), hostsProvider, clusterApplier,
134+
Randomness.get()));
123135
discoveryTypes.put("single-node", () -> new SingleNodeDiscovery(settings, transportService, masterService, clusterApplier));
124136
for (DiscoveryPlugin plugin : plugins) {
125-
plugin.getDiscoveryTypes(threadPool, transportService, namedWriteableRegistry,
126-
masterService, clusterApplier, clusterSettings, hostsProvider, allocationService, gatewayMetaState).entrySet()
127-
.forEach(entry -> {
128-
if (discoveryTypes.put(entry.getKey(), entry.getValue()) != null) {
129-
throw new IllegalArgumentException("Cannot register discovery type [" + entry.getKey() + "] twice");
130-
}
131-
});
137+
plugin.getDiscoveryTypes(threadPool, transportService, namedWriteableRegistry, masterService, clusterApplier, clusterSettings,
138+
hostsProvider, allocationService, gatewayMetaState).forEach((key, value) -> {
139+
if (discoveryTypes.put(key, value) != null) {
140+
throw new IllegalArgumentException("Cannot register discovery type [" + key + "] twice");
141+
}
142+
});
132143
}
133144
String discoveryType = DISCOVERY_TYPE_SETTING.get(settings);
134145
Supplier<Discovery> discoverySupplier = discoveryTypes.get(discoveryType);

‎server/src/main/java/org/elasticsearch/gateway/GatewayMetaState.java

+14
Original file line numberDiff line numberDiff line change
@@ -29,13 +29,16 @@
2929
import org.elasticsearch.cluster.ClusterStateApplier;
3030
import org.elasticsearch.cluster.block.ClusterBlocks;
3131
import org.elasticsearch.cluster.coordination.CoordinationState;
32+
import org.elasticsearch.cluster.coordination.CoordinationState.PersistedState;
33+
import org.elasticsearch.cluster.coordination.InMemoryPersistedState;
3234
import org.elasticsearch.cluster.metadata.IndexMetaData;
3335
import org.elasticsearch.cluster.metadata.Manifest;
3436
import org.elasticsearch.cluster.metadata.MetaData;
3537
import org.elasticsearch.cluster.metadata.MetaDataIndexUpgradeService;
3638
import org.elasticsearch.cluster.node.DiscoveryNode;
3739
import org.elasticsearch.cluster.routing.RoutingNode;
3840
import org.elasticsearch.cluster.routing.ShardRouting;
41+
import org.elasticsearch.cluster.service.ClusterApplierService;
3942
import org.elasticsearch.cluster.service.ClusterService;
4043
import org.elasticsearch.common.collect.ImmutableOpenMap;
4144
import org.elasticsearch.common.collect.Tuple;
@@ -108,6 +111,17 @@ public GatewayMetaState(Settings settings, NodeEnvironment nodeEnv, MetaStateSer
108111
incrementalWrite = false;
109112
}
110113

114+
public PersistedState getPersistedState(Settings settings, ClusterApplierService clusterApplier) {
115+
applyClusterStateUpdaters();
116+
if (DiscoveryNode.isMasterNode(settings) == false) {
117+
// use Zen1 way of writing cluster state for non-master-eligible nodes
118+
// this avoids concurrent manipulating of IndexMetadata with IndicesStore
119+
clusterApplier.addLowPriorityApplier(this);
120+
return new InMemoryPersistedState(getCurrentTerm(), getLastAcceptedState());
121+
}
122+
return this;
123+
}
124+
111125
private void initializeClusterState(ClusterName clusterName) throws IOException {
112126
long startNS = System.nanoTime();
113127
Tuple<Manifest, MetaData> manifestAndMetaData = metaStateService.loadFullState();

‎server/src/test/java/org/elasticsearch/bootstrap/BootstrapChecksTests.java

+4-2
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@
4040
import java.util.concurrent.atomic.AtomicReference;
4141
import java.util.function.Consumer;
4242

43+
import static org.elasticsearch.discovery.DiscoveryModule.ZEN2_DISCOVERY_TYPE;
44+
import static org.elasticsearch.discovery.DiscoveryModule.ZEN_DISCOVERY_TYPE;
4345
import static org.hamcrest.CoreMatchers.allOf;
4446
import static org.hamcrest.CoreMatchers.containsString;
4547
import static org.hamcrest.CoreMatchers.equalTo;
@@ -101,7 +103,7 @@ public void testEnforceLimitsWhenBoundToNonLocalAddress() {
101103
when(boundTransportAddress.boundAddresses()).thenReturn(transportAddresses.toArray(new TransportAddress[0]));
102104
when(boundTransportAddress.publishAddress()).thenReturn(publishAddress);
103105

104-
final String discoveryType = randomFrom("zen", "single-node");
106+
final String discoveryType = randomFrom(ZEN_DISCOVERY_TYPE, ZEN2_DISCOVERY_TYPE, "single-node");
105107

106108
assertEquals(BootstrapChecks.enforceLimits(boundTransportAddress, discoveryType), !"single-node".equals(discoveryType));
107109
}
@@ -119,7 +121,7 @@ public void testEnforceLimitsWhenPublishingToNonLocalAddress() {
119121
when(boundTransportAddress.boundAddresses()).thenReturn(transportAddresses.toArray(new TransportAddress[0]));
120122
when(boundTransportAddress.publishAddress()).thenReturn(publishAddress);
121123

122-
final String discoveryType = randomFrom("zen", "single-node");
124+
final String discoveryType = randomFrom(ZEN_DISCOVERY_TYPE, ZEN2_DISCOVERY_TYPE, "single-node");
123125

124126
assertEquals(BootstrapChecks.enforceLimits(boundTransportAddress, discoveryType), !"single-node".equals(discoveryType));
125127
}

‎test/framework/src/main/java/org/elasticsearch/test/discovery/TestZenDiscovery.java

+3-16
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,7 @@
1919

2020
package org.elasticsearch.test.discovery;
2121

22-
import org.elasticsearch.cluster.coordination.CoordinationState;
2322
import org.elasticsearch.cluster.coordination.Coordinator;
24-
import org.elasticsearch.cluster.coordination.InMemoryPersistedState;
25-
import org.elasticsearch.cluster.node.DiscoveryNode;
2623
import org.elasticsearch.cluster.routing.allocation.AllocationService;
2724
import org.elasticsearch.cluster.service.ClusterApplier;
2825
import org.elasticsearch.cluster.service.ClusterApplierService;
@@ -81,20 +78,10 @@ public Map<String, Supplier<Discovery>> getDiscoveryTypes(ThreadPool threadPool,
8178
Settings fixedSettings = Settings.builder().put(settings).putList(DISCOVERY_ZEN_PING_UNICAST_HOSTS_SETTING.getKey()).build();
8279
return Collections.singletonMap("test-zen", () -> {
8380
if (USE_ZEN2.get(settings)) {
84-
Supplier<CoordinationState.PersistedState> persistedStateSupplier = () -> {
85-
gatewayMetaState.applyClusterStateUpdaters();
86-
if (DiscoveryNode.isMasterNode(settings) == false) {
87-
// use Zen1 way of writing cluster state for non-master-eligible nodes
88-
// this avoids concurrent manipulating of IndexMetadata with IndicesStore
89-
((ClusterApplierService) clusterApplier).addLowPriorityApplier(gatewayMetaState);
90-
return new InMemoryPersistedState(gatewayMetaState.getCurrentTerm(), gatewayMetaState.getLastAcceptedState());
91-
}
92-
return gatewayMetaState;
93-
};
94-
9581
return new Coordinator("test_node", fixedSettings, clusterSettings, transportService, namedWriteableRegistry,
96-
allocationService, masterService, persistedStateSupplier, hostsProvider, clusterApplier,
97-
new Random(Randomness.get().nextLong()));
82+
allocationService, masterService,
83+
() -> gatewayMetaState.getPersistedState(settings, (ClusterApplierService) clusterApplier), hostsProvider,
84+
clusterApplier, new Random(Randomness.get().nextLong()));
9885
} else {
9986
return new TestZenDiscovery(fixedSettings, threadPool, transportService, namedWriteableRegistry, masterService,
10087
clusterApplier, clusterSettings, hostsProvider, allocationService, gatewayMetaState);

‎x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/SecurityTests.java

+3-1
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,8 @@
6464
import java.util.stream.Collectors;
6565

6666
import static org.elasticsearch.cluster.metadata.IndexMetaData.INDEX_FORMAT_SETTING;
67+
import static org.elasticsearch.discovery.DiscoveryModule.ZEN2_DISCOVERY_TYPE;
68+
import static org.elasticsearch.discovery.DiscoveryModule.ZEN_DISCOVERY_TYPE;
6769
import static org.elasticsearch.xpack.security.support.SecurityIndexManager.SECURITY_INDEX_NAME;
6870
import static org.elasticsearch.xpack.security.support.SecurityIndexManager.INTERNAL_INDEX_FORMAT;
6971
import static org.hamcrest.Matchers.containsString;
@@ -281,7 +283,7 @@ public void testTLSJoinValidator() throws Exception {
281283
int numIters = randomIntBetween(1, 10);
282284
for (int i = 0; i < numIters; i++) {
283285
boolean tlsOn = randomBoolean();
284-
String discoveryType = randomFrom("single-node", "zen", randomAlphaOfLength(4));
286+
String discoveryType = randomFrom("single-node", ZEN_DISCOVERY_TYPE, ZEN2_DISCOVERY_TYPE, randomAlphaOfLength(4));
285287
Security.ValidateTLSOnJoin validator = new Security.ValidateTLSOnJoin(tlsOn, discoveryType);
286288
MetaData.Builder builder = MetaData.builder();
287289
License license = TestUtils.generateSignedLicense(TimeValue.timeValueHours(24));

0 commit comments

Comments
 (0)