honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

remove the jgroups plugin

This commit is contained in:
kimchy 2010-06-21 22:51:54 +03:00
parent 893de86e9e
commit 96d08cc641
16 changed files with 0 additions and 1077 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.idea/modules.xml
View File

@ -7,7 +7,6 @@
<module fileurl="file://$PROJECT_DIR$/.idea/modules/elasticsearch-root.iml" filepath="$PROJECT_DIR$/.idea/modules/elasticsearch-root.iml" />
<module fileurl="file://$PROJECT_DIR$/.idea/modules/plugin-analysis-icu.iml" filepath="$PROJECT_DIR$/.idea/modules/plugin-analysis-icu.iml" />
<module fileurl="file://$PROJECT_DIR$/.idea/modules//plugin-client-groovy.iml" filepath="$PROJECT_DIR$/.idea/modules//plugin-client-groovy.iml" />
<module fileurl="file://$PROJECT_DIR$/.idea/modules//plugin-discovery-jgroups.iml" filepath="$PROJECT_DIR$/.idea/modules//plugin-discovery-jgroups.iml" />
<module fileurl="file://$PROJECT_DIR$/.idea/modules//plugin-mapper-attachments.iml" filepath="$PROJECT_DIR$/.idea/modules//plugin-mapper-attachments.iml" />
<module fileurl="file://$PROJECT_DIR$/.idea/modules//plugin-transport-memcached.iml" filepath="$PROJECT_DIR$/.idea/modules//plugin-transport-memcached.iml" />
<module fileurl="file://$PROJECT_DIR$/.idea/modules//plugins-cloud.iml" filepath="$PROJECT_DIR$/.idea/modules//plugins-cloud.iml" />

1
.idea/modules/elasticsearch-root.iml
View File

@ -14,7 +14,6 @@
<orderEntry type="sourceFolder" forTests="false" />
<orderEntry type="module" module-name="elasticsearch" />
<orderEntry type="module" module-name="plugin-client-groovy" />
<orderEntry type="module" module-name="plugin-discovery-jgroups" />
<orderEntry type="module" module-name="plugin-mapper-attachments" />
<orderEntry type="module" module-name="plugin-transport-memcached" />
<orderEntry type="module" module-name="plugin-analysis-icu" />

29
.idea/modules/plugin-discovery-jgroups.iml
View File

@ -1,29 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="JAVA_MODULE" version="4">
<component name="NewModuleRootManager" inherit-compiler-output="false">
<output url="file://$MODULE_DIR$/../../plugins/discovery/jgroups/build/classes/main" />
<output-test url="file://$MODULE_DIR$/../../plugins/discovery/jgroups/build/classes/test" />
<exclude-output />
<content url="file://$MODULE_DIR$/../../plugins/discovery/jgroups">
<sourceFolder url="file://$MODULE_DIR$/../../plugins/discovery/jgroups/src/main/java" isTestSource="false" />
<sourceFolder url="file://$MODULE_DIR$/../../plugins/discovery/jgroups/src/test/java" isTestSource="true" />
<excludeFolder url="file://$MODULE_DIR$/../../plugins/discovery/jgroups/build" />
</content>
<orderEntry type="inheritedJdk" />
<orderEntry type="sourceFolder" forTests="false" />
<orderEntry type="module" module-name="elasticsearch" />
<orderEntry type="module-library">
<library name="jgroups">
<CLASSES>
<root url="jar://$GRADLE_REPOSITORY$/jgroups/jgroups/jars/jgroups-2.9.0.GA.jar!/" />
</CLASSES>
<JAVADOC />
<SOURCES />
</library>
</orderEntry>
<orderEntry type="module" module-name="test-testng" scope="TEST" />
<orderEntry type="library" scope="TEST" name="testng" level="project" />
<orderEntry type="library" scope="TEST" name="hamcrest" level="project" />
</component>
</module>

137
plugins/discovery/jgroups/build.gradle
View File

@ -1,137 +0,0 @@
dependsOn(':elasticsearch')
apply plugin: 'java'
apply plugin: 'maven'
archivesBaseName = "elasticsearch-discovery-jgroups"
explodedDistDir = new File(distsDir, 'exploded')
configurations.compile.transitive = true
configurations.testCompile.transitive = true
// no need to use the resource dir
sourceSets.main.resources.srcDirs 'src/main/java'
sourceSets.test.resources.srcDirs 'src/test/java'
jar {
// from sourceSets.main.allJava
manifest {
attributes("Implementation-Title": "ElasticSearch", "Implementation-Version": rootProject.version, "Implementation-Date": buildTimeStr)
}
}
configurations {
dists
distLib {
visible = false
transitive = false
}
}
dependencies {
compile project(':elasticsearch')
compile('jgroups:jgroups:2.9.0.GA') { transitive = false }
distLib('jgroups:jgroups:2.9.0.GA') { transitive = false }
testCompile project(':test-testng')
testCompile('org.testng:testng:5.10:jdk15') { transitive = false }
}
test {
useTestNG()
jmvArgs = ["-ea", "-Xmx1024m"]
suiteName = project.name
listeners = ["org.elasticsearch.util.testng.Listeners"]
systemProperties["es.test.log.conf"] = System.getProperty("es.test.log.conf", "log4j-gradle.properties")
}
task explodedDist(dependsOn: [jar], description: 'Builds the plugin zip file') << {
[explodedDistDir]*.mkdirs()
copy {
from configurations.distLib
into explodedDistDir
}
// remove elasticsearch files (compile above adds the elasticsearch one)
ant.delete { fileset(dir: explodedDistDir, includes: "elasticsearch-*.jar") }
copy {
from libsDir
into explodedDistDir
}
ant.delete { fileset(dir: explodedDistDir, includes: "elasticsearch-*-javadoc.jar") }
ant.delete { fileset(dir: explodedDistDir, includes: "elasticsearch-*-sources.jar") }
}
task zip(type: Zip, dependsOn: ['explodedDist']) {
from(explodedDistDir) {
}
}
task release(dependsOn: [zip]) << {
ant.delete(dir: explodedDistDir)
copy {
from distsDir
into(new File(rootProject.distsDir, "plugins"))
}
}
configurations {
deployerJars
}
dependencies {
deployerJars "org.apache.maven.wagon:wagon-http:1.0-beta-2"
}
task sourcesJar(type: Jar, dependsOn: classes) {
classifier = 'sources'
from sourceSets.main.allSource
}
task javadocJar(type: Jar, dependsOn: javadoc) {
classifier = 'javadoc'
from javadoc.destinationDir
}
artifacts {
archives sourcesJar
archives javadocJar
}
uploadArchives {
repositories.mavenDeployer {
configuration = configurations.deployerJars
repository(url: rootProject.mavenRepoUrl) {
authentication(userName: rootProject.mavenRepoUser, password: rootProject.mavenRepoPass)
}
snapshotRepository(url: rootProject.mavenSnapshotRepoUrl) {
authentication(userName: rootProject.mavenRepoUser, password: rootProject.mavenRepoPass)
}
pom.project {
inceptionYear '2009'
name 'elasticsearch-plugins-disocvery-jgroups'
description 'Memcacehd Plugin for ElasticSearch'
licenses {
license {
name 'The Apache Software License, Version 2.0'
url 'http://www.apache.org/licenses/LICENSE-2.0.txt'
distribution 'repo'
}
}
scm {
connection 'git://github.com/elasticsearch/elasticsearch.git'
developerConnection 'git@github.com:elasticsearch/elasticsearch.git'
url 'http://github.com/elasticsearch/elasticsearch'
}
}
pom.whenConfigured {pom ->
pom.dependencies = pom.dependencies.findAll {dep -> dep.scope != 'test' } // removes the test scoped ones
}
}
}

76
plugins/discovery/jgroups/src/main/java/config/jgroups/tcp-nio.xml
View File

@ -1,76 +0,0 @@
<!-- Changes from default jgroups tcp-nio.xml: -->
<!-- 1. Changed print_local_addr from true to false -->
<!-- 2. Disabled STREAMING_STATE_TRANSFER, we don't use it -->
<!-- 3. Added ${jgroups.bind_port:7800} to bind_port -->
<!--
TCP based stack, with flow control and message bundling. This is usually used when IP
multicasting cannot be used in a network, e.g. because it is disabled (routers discard multicast).
Note that TCP.bind_addr and TCPPING.initial_hosts should be set, possibly via system properties, e.g.
-Djgroups.bind_addr=192.168.5.2 and -Djgroups.tcpping.initial_hosts=192.168.5.2[7800]".
author: Bela Ban
version: $Id: tcp-nio.xml,v 1.19 2009/12/11 10:45:19 belaban Exp $
-->
<config xmlns="urn:org:jgroups"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/JGroups-2.8.xsd">
<TCP_NIO
bind_port="${jgroups.bind_port:7800}"
loopback="true"
recv_buf_size="20000000"
send_buf_size="640000"
discard_incompatible_packets="true"
max_bundle_size="64000"
max_bundle_timeout="30"
enable_bundling="true"
use_send_queues="true"
sock_conn_timeout="300"
thread_pool.enabled="true"
thread_pool.min_threads="1"
thread_pool.max_threads="25"
thread_pool.keep_alive_time="5000"
thread_pool.queue_enabled="false"
thread_pool.queue_max_size="100"
thread_pool.rejection_policy="Run"
oob_thread_pool.enabled="true"
oob_thread_pool.min_threads="1"
oob_thread_pool.max_threads="8"
oob_thread_pool.keep_alive_time="5000"
oob_thread_pool.queue_enabled="false"
oob_thread_pool.queue_max_size="100"
oob_thread_pool.rejection_policy="Run"
reader_threads="3"
writer_threads="3"
processor_threads="0"
processor_minThreads="0"
processor_maxThreads="0"
processor_queueSize="100"
processor_keepAliveTime="9223372036854775807"/>
<TCPPING timeout="3000"
initial_hosts="${jgroups.tcpping.initial_hosts:localhost[7800],localhost[7801]}"
port_range="1"
num_initial_members="3"/>
<MERGE2 max_interval="100000"
min_interval="20000"/>
<FD_SOCK/>
<FD timeout="10000" max_tries="5"/>
<VERIFY_SUSPECT timeout="1500"/>
<pbcast.NAKACK
use_mcast_xmit="false" gc_lag="0"
retransmit_timeout="300,600,1200,2400,4800"
discard_delivered_msgs="true"/>
<UNICAST timeout="300,600,1200"/>
<pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
max_bytes="400000"/>
<pbcast.GMS print_local_addr="false" join_timeout="3000"
view_bundling="true"/>
<FC max_credits="2000000"
min_threshold="0.10"/>
<FRAG2 frag_size="60000"/>
<!--<pbcast.STREAMING_STATE_TRANSFER/>-->
<!-- pbcast.STATE_TRANSFER/> -->
</config>

70
plugins/discovery/jgroups/src/main/java/config/jgroups/tcp.xml
View File

@ -1,70 +0,0 @@
<!-- Changes from default jgroups tcp.xml: -->
<!-- 1. Changed print_local_addr from true to false -->
<!-- 2. Disabled STREAMING_STATE_TRANSFER, we don't use it -->
<!-- 3. Added ${jgroups.bind_port:7800} to bind_port -->
<!--
TCP based stack, with flow control and message bundling. This is usually used when IP
multicasting cannot be used in a network, e.g. because it is disabled (routers discard multicast).
Note that TCP.bind_addr and TCPPING.initial_hosts should be set, possibly via system properties, e.g.
-Djgroups.bind_addr=192.168.5.2 and -Djgroups.tcpping.initial_hosts=192.168.5.2[7800]
author: Bela Ban
version: $Id: tcp.xml,v 1.40 2009/12/18 09:28:30 belaban Exp $
-->
<config xmlns="urn:org:jgroups"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/JGroups-2.8.xsd">
<TCP bind_port="${jgroups.bind_port:7800}"
loopback="true"
recv_buf_size="${tcp.recv_buf_size:20M}"
send_buf_size="${tcp.send_buf_size:640K}"
discard_incompatible_packets="true"
max_bundle_size="64K"
max_bundle_timeout="30"
enable_bundling="true"
use_send_queues="true"
sock_conn_timeout="300"
timer.num_threads="4"
thread_pool.enabled="true"
thread_pool.min_threads="1"
thread_pool.max_threads="10"
thread_pool.keep_alive_time="5000"
thread_pool.queue_enabled="false"
thread_pool.queue_max_size="100"
thread_pool.rejection_policy="discard"
oob_thread_pool.enabled="true"
oob_thread_pool.min_threads="1"
oob_thread_pool.max_threads="8"
oob_thread_pool.keep_alive_time="5000"
oob_thread_pool.queue_enabled="false"
oob_thread_pool.queue_max_size="100"
oob_thread_pool.rejection_policy="discard"/>
<TCPPING timeout="3000"
initial_hosts="${jgroups.tcpping.initial_hosts:localhost[7800],localhost[7801]}"
port_range="1"
num_initial_members="3"/>
<MERGE2 min_interval="10000"
max_interval="30000"/>
<FD_SOCK/>
<FD timeout="3000" max_tries="3"/>
<VERIFY_SUSPECT timeout="1500"/>
<BARRIER/>
<pbcast.NAKACK
use_mcast_xmit="false" gc_lag="0"
retransmit_timeout="300,600,1200,2400,4800"
discard_delivered_msgs="true"/>
<UNICAST timeout="300,600,1200"/>
<pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
max_bytes="400K"/>
<pbcast.GMS print_local_addr="false" join_timeout="3000"
view_bundling="true"/>
<FC max_credits="2M"
min_threshold="0.10"/>
<FRAG2 frag_size="60K"/>
<!--<pbcast.STREAMING_STATE_TRANSFER/>-->
<!-- <pbcast.STATE_TRANSFER/> -->
</config>

24
plugins/discovery/jgroups/src/main/java/config/jgroups/tcpgossip.xml
View File

@ -1,24 +0,0 @@
<!-- Changes from default jgroups tcp.xml: -->
<!-- 1. Changed print_local_addr from true to false -->
<!-- ************ JGroups Protocol Stack Configuration ************** -->
<!-- generated by XmlConfigurator on Mon Apr 26 11:26:18 PDT 2004 -->
<!-- input file: tcpgossip.old.xml -->
<config xmlns="urn:org:jgroups"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/JGroups-2.8.xsd">
<TCP bind_port="7800" bind_addr="localhost" loopback="true"/>
<TCPGOSSIP timeout="3000" initial_hosts="localhost[12001]" num_initial_members="3"/>
<MERGE2 max_interval="30000"
min_interval="10000"/>
<FD timeout="2000" max_tries="4"/>
<VERIFY_SUSPECT timeout="1500"/>
<pbcast.NAKACK gc_lag="100" retransmit_timeout="600,1200,2400,4800"/>
<UNICAST timeout="300,600,1200"/>
<pbcast.STABLE stability_delay="1000" desired_avg_gossip="20000" max_bytes="0"/>
<pbcast.GMS print_local_addr="false" join_timeout="5000"/>
<FC max_credits="2000000"
min_threshold="0.10"/>
<FRAG2 frag_size="60000"/>
</config>

73
plugins/discovery/jgroups/src/main/java/config/jgroups/udp.xml
View File

@ -1,73 +0,0 @@
<!-- Changes from default jgroups udp.xml: -->
<!-- 1. Changed print_local_addr from true to false -->
<!-- 2. Disabled STATE_TRANSFER, we don't use it -->
<!--
Default stack using IP multicasting. It is similar to the "udp"
stack in stacks.xml, but doesn't use streaming state transfer and flushing
author: Bela Ban
version: $Id: udp.xml,v 1.40 2010/02/08 07:11:15 belaban Exp $
-->
<config xmlns="urn:org:jgroups"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/JGroups-2.8.xsd">
<UDP
mcast_port="${jgroups.udp.mcast_port:45588}"
tos="8"
ucast_recv_buf_size="20M"
ucast_send_buf_size="640K"
mcast_recv_buf_size="25M"
mcast_send_buf_size="640K"
loopback="true"
discard_incompatible_packets="true"
max_bundle_size="64K"
max_bundle_timeout="30"
ip_ttl="${jgroups.udp.ip_ttl:2}"
enable_bundling="true"
enable_diagnostics="true"
thread_naming_pattern="cl"
timer.num_threads="4"
thread_pool.enabled="true"
thread_pool.min_threads="2"
thread_pool.max_threads="8"
thread_pool.keep_alive_time="5000"
thread_pool.queue_enabled="true"
thread_pool.queue_max_size="10000"
thread_pool.rejection_policy="discard"
oob_thread_pool.enabled="true"
oob_thread_pool.min_threads="1"
oob_thread_pool.max_threads="8"
oob_thread_pool.keep_alive_time="5000"
oob_thread_pool.queue_enabled="false"
oob_thread_pool.queue_max_size="100"
oob_thread_pool.rejection_policy="Run"/>
<PING timeout="2000"
num_initial_members="3"/>
<MERGE2 max_interval="30000"
min_interval="10000"/>
<FD_SOCK/>
<FD_ALL/>
<VERIFY_SUSPECT timeout="1500"/>
<BARRIER/>
<pbcast.NAKACK use_stats_for_retransmission="false"
exponential_backoff="0"
use_mcast_xmit="true" gc_lag="0"
retransmit_timeout="300,600,1200"
discard_delivered_msgs="true"/>
<UNICAST timeout="300,600,1200"/>
<pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
max_bytes="1M"/>
<pbcast.GMS print_local_addr="false" join_timeout="3000"
view_bundling="true"/>
<FC max_credits="500K"
min_threshold="0.20"/>
<FRAG2 frag_size="60K"/>
<!--pbcast.STREAMING_STATE_TRANSFER /-->
<!--<pbcast.STATE_TRANSFER />-->
<!-- pbcast.FLUSH /-->
</config>

1
plugins/discovery/jgroups/src/main/java/es-plugin.properties
View File

@ -1 +0,0 @@
plugin=org.elasticsearch.plugin.discovery.jgroups.JgroupsDiscoveryPlugin

136
plugins/discovery/jgroups/src/main/java/org/elasticsearch/discovery/jgroups/JgroupsCustomLogFactory.java
View File

@ -1,136 +0,0 @@
/*
* Licensed to Elastic Search and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. Elastic Search 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.discovery.jgroups;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;
import org.jgroups.logging.CustomLogFactory;
import org.jgroups.logging.Log;
/**
* @author kimchy (Shay Banon)
*/
public class JgroupsCustomLogFactory implements CustomLogFactory {
@Override public Log getLog(Class clazz) {
return getLog(clazz.getName());
}
@Override public Log getLog(String category) {
return new JgroupsESLog(ESLoggerFactory.getLogger(category.replace("org.jgroups.", "jgroups.").replace(".protocols.", ".")));
}
private static class JgroupsESLog implements Log {
private final ESLogger logger;
private JgroupsESLog(ESLogger logger) {
this.logger = logger;
}
@Override public boolean isFatalEnabled() {
return logger.isErrorEnabled();
}
@Override public boolean isErrorEnabled() {
return logger.isErrorEnabled();
}
@Override public boolean isWarnEnabled() {
return logger.isWarnEnabled();
}
@Override public boolean isInfoEnabled() {
return logger.isInfoEnabled();
}
@Override public boolean isDebugEnabled() {
return logger.isDebugEnabled();
}
@Override public boolean isTraceEnabled() {
return logger.isTraceEnabled();
}
@Override public void debug(String msg) {
logger.debug(msg);
}
@Override public void debug(String msg, Throwable throwable) {
logger.debug(msg, throwable);
}
@Override public void error(String msg) {
logger.error(msg);
}
@Override public void error(String msg, Throwable throwable) {
logger.error(msg, throwable);
}
@Override public void fatal(String msg) {
logger.error(msg);
}
@Override public void fatal(String msg, Throwable throwable) {
logger.error(msg, throwable);
}
@Override public void info(String msg) {
logger.info(msg);
}
@Override public void info(String msg, Throwable throwable) {
logger.info(msg, throwable);
}
@Override public void trace(Object msg) {
logger.trace(msg.toString());
}
@Override public void trace(Object msg, Throwable throwable) {
logger.trace(msg.toString(), throwable);
}
@Override public void trace(String msg) {
logger.trace(msg);
}
@Override public void trace(String msg, Throwable throwable) {
logger.trace(msg, throwable);
}
@Override public void warn(String msg) {
logger.warn(msg);
}
@Override public void warn(String msg, Throwable throwable) {
logger.warn(msg, throwable);
}
@Override public void setLevel(String level) {
// ignore
}
@Override public String getLevel() {
return null;
}
}
}

370
plugins/discovery/jgroups/src/main/java/org/elasticsearch/discovery/jgroups/JgroupsDiscovery.java
View File

@ -1,370 +0,0 @@
/*
* Licensed to Elastic Search and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. Elastic Search 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.discovery.jgroups;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.ElasticSearchIllegalStateException;
import org.elasticsearch.cluster.*;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.stream.BytesStreamInput;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.network.NetworkService;
import org.elasticsearch.common.network.NetworkUtils;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.discovery.Discovery;
import org.elasticsearch.discovery.DiscoveryException;
import org.elasticsearch.discovery.InitialStateDiscoveryListener;
import org.elasticsearch.env.Environment;
import org.elasticsearch.transport.TransportService;
import org.jgroups.*;
import java.io.IOException;
import java.net.Inet4Address;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.URL;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.elasticsearch.cluster.ClusterState.*;
import static org.elasticsearch.cluster.node.DiscoveryNode.*;
import static org.elasticsearch.common.collect.Maps.*;
import static org.elasticsearch.common.collect.Sets.*;
/**
* @author kimchy (Shay Banon)
*/
public class JgroupsDiscovery extends AbstractLifecycleComponent<Discovery> implements Discovery, Receiver {
static {
System.setProperty("jgroups.logging.log_factory_class", JgroupsCustomLogFactory.class.getName());
}
private final ClusterName clusterName;
private final TransportService transportService;
private final ClusterService clusterService;
private final NetworkService networkService;
private final Channel channel;
private volatile boolean addressSet = false;
private DiscoveryNode localNode;
private final AtomicBoolean initialStateSent = new AtomicBoolean();
private final CopyOnWriteArrayList<InitialStateDiscoveryListener> initialStateListeners = new CopyOnWriteArrayList<InitialStateDiscoveryListener>();
@Inject public JgroupsDiscovery(Settings settings, Environment environment, ClusterName clusterName,
TransportService transportService, ClusterService clusterService, NetworkService networkService) {
super(settings);
this.clusterName = clusterName;
this.transportService = transportService;
this.clusterService = clusterService;
this.networkService = networkService;
String config = componentSettings.get("config", "udp");
String actualConfig = config;
if (!config.endsWith(".xml")) {
actualConfig = "jgroups/" + config + ".xml";
}
URL configUrl = environment.resolveConfig(actualConfig);
logger.debug("Using configuration [{}]", configUrl);
Map<String, String> sysPropsSet = newHashMap();
try {
// prepare system properties to configure jgroups based on the settings
for (Map.Entry<String, String> entry : settings.getAsMap().entrySet()) {
if (entry.getKey().startsWith("discovery.jgroups")) {
String jgroupsKey = entry.getKey().substring("discovery.".length());
if (System.getProperty(jgroupsKey) == null) {
sysPropsSet.put(jgroupsKey, entry.getValue());
System.setProperty(jgroupsKey, entry.getValue());
}
}
}
if (System.getProperty("jgroups.bind_addr") == null) {
// automatically set the bind address based on ElasticSearch default bindings...
try {
InetAddress bindAddress = networkService.resolveBindHostAddress(null, NetworkService.LOCAL);
if ((bindAddress instanceof Inet4Address && NetworkUtils.isIPv4()) || (bindAddress instanceof Inet6Address && !NetworkUtils.isIPv4())) {
sysPropsSet.put("jgroups.bind_addr", bindAddress.getHostAddress());
System.setProperty("jgroups.bind_addr", bindAddress.getHostAddress());
}
} catch (IOException e) {
// ignore this
}
}
channel = new JChannel(configUrl);
} catch (ChannelException e) {
throw new DiscoveryException("Failed to create jgroups channel with config [" + configUrl + "]", e);
} finally {
for (String keyToRemove : sysPropsSet.keySet()) {
System.getProperties().remove(keyToRemove);
}
}
}
@Override public void addListener(InitialStateDiscoveryListener listener) {
initialStateListeners.add(listener);
}
@Override public void removeListener(InitialStateDiscoveryListener listener) {
initialStateListeners.remove(listener);
}
@Override protected void doStart() throws ElasticSearchException {
try {
channel.connect(clusterName.value());
channel.setReceiver(this);
logger.debug("Connected to cluster [{}], address [{}]", channel.getClusterName(), channel.getAddress());
this.localNode = new DiscoveryNode(settings.get("name"), channel.getAddress().toString(), transportService.boundAddress().publishAddress(), buildCommonNodesAttributes(settings));
if (isMaster()) {
clusterService.submitStateUpdateTask("jgroups-disco-initial_connect(master)", new ProcessedClusterStateUpdateTask() {
@Override public ClusterState execute(ClusterState currentState) {
DiscoveryNodes.Builder builder = new DiscoveryNodes.Builder()
.localNodeId(localNode.id())
.masterNodeId(localNode.id())
// put our local node
.put(localNode);
return newClusterStateBuilder().state(currentState).nodes(builder).build();
}
@Override public void clusterStateProcessed(ClusterState clusterState) {
sendInitialStateEventIfNeeded();
}
});
addressSet = true;
} else {
clusterService.submitStateUpdateTask("jgroups-disco-initialconnect", new ClusterStateUpdateTask() {
@Override public ClusterState execute(ClusterState currentState) {
DiscoveryNodes.Builder builder = new DiscoveryNodes.Builder()
.localNodeId(localNode.id())
.put(localNode);
return newClusterStateBuilder().state(currentState).nodes(builder).build();
}
});
try {
channel.send(new Message(channel.getView().getCreator(), channel.getAddress(), nodeMessagePayload()));
addressSet = true;
logger.debug("Sent (initial) node information to master [{}], node [{}]", channel.getView().getCreator(), localNode);
} catch (Exception e) {
logger.warn("Can't send address to master [" + channel.getView().getCreator() + "] will try again later...", e);
}
}
} catch (ChannelException e) {
throw new DiscoveryException("Can't connect to group [" + clusterName + "]", e);
}
}
@Override protected void doStop() throws ElasticSearchException {
initialStateSent.set(false);
if (channel.isConnected()) {
channel.disconnect();
}
}
@Override protected void doClose() throws ElasticSearchException {
if (channel.isOpen()) {
channel.close();
}
}
public String nodeDescription() {
return channel.getClusterName() + "/" + channel.getAddress();
}
@Override public void publish(ClusterState clusterState) {
if (!isMaster()) {
throw new ElasticSearchIllegalStateException("Shouldn't publish state when not master");
}
try {
channel.send(new Message(null, null, ClusterState.Builder.toBytes(clusterState)));
} catch (Exception e) {
logger.error("Failed to send cluster state to nodes", e);
}
}
@Override public void receive(Message msg) {
if (msg.getSrc().equals(channel.getAddress())) {
return; // my own message, ignore.
}
// message from the master, the cluster state has changed.
if (msg.getSrc().equals(channel.getView().getCreator())) {
try {
byte[] buffer = msg.getBuffer();
final ClusterState clusterState = ClusterState.Builder.fromBytes(buffer, settings, localNode);
// ignore cluster state messages that do not include "me", not in the game yet...
if (clusterState.nodes().localNode() != null) {
clusterService.submitStateUpdateTask("jgroups-disco-receive(from master)", new ProcessedClusterStateUpdateTask() {
@Override public ClusterState execute(ClusterState currentState) {
return clusterState;
}
@Override public void clusterStateProcessed(ClusterState clusterState) {
sendInitialStateEventIfNeeded();
}
});
}
} catch (Exception e) {
logger.error("Received corrupted cluster state.", e);
}
return;
}
// direct message from a member indicating it has joined the jgroups cluster and provides us its node information
if (isMaster()) {
try {
BytesStreamInput is = new BytesStreamInput(msg.getBuffer());
final DiscoveryNode newNode = DiscoveryNode.readNode(is);
is.close();
if (logger.isDebugEnabled()) {
logger.debug("Received node information from [{}], node [{}]", msg.getSrc(), newNode);
}
if (!transportService.addressSupported(newNode.address().getClass())) {
// TODO, what should we do now? Maybe inform that node that its crap?
logger.warn("Received a wrong address type from [" + msg.getSrc() + "], ignoring... (received_address[" + newNode.address() + ")");
} else {
clusterService.submitStateUpdateTask("jgroups-disco-receive(from node[" + newNode + "])", new ClusterStateUpdateTask() {
@Override public ClusterState execute(ClusterState currentState) {
if (currentState.nodes().nodeExists(newNode.id())) {
// no change, the node already exists in the cluster
logger.warn("Received an address [{}] for an existing node [{}]", newNode.address(), newNode);
return currentState;
}
return newClusterStateBuilder().state(currentState).nodes(currentState.nodes().newNode(newNode)).build();
}
});
}
} catch (Exception e) {
logger.warn("Can't read address from cluster member [" + msg.getSrc() + "] message [" + msg.getClass().getName() + "/" + msg + "]", e);
}
return;
}
logger.error("A message between two members that neither of them is the master is not allowed.");
}
private boolean isMaster() {
return channel.getAddress().equals(channel.getView().getCreator());
}
@Override public byte[] getState() {
return new byte[0];
}
@Override public void setState(byte[] state) {
}
@Override public void viewAccepted(final View newView) {
if (!addressSet) {
try {
channel.send(new Message(newView.getCreator(), channel.getAddress(), nodeMessagePayload()));
logger.debug("Sent (view) node information to master [{}], node [{}]", newView.getCreator(), localNode);
addressSet = true;
} catch (Exception e) {
logger.warn("Can't send address to master [" + newView.getCreator() + "] will try again later...", e);
}
}
// I am the master
if (channel.getAddress().equals(newView.getCreator())) {
final Set<String> newMembers = newHashSet();
for (Address address : newView.getMembers()) {
newMembers.add(address.toString());
}
clusterService.submitStateUpdateTask("jgroups-disco-view", new ClusterStateUpdateTask() {
@Override public ClusterState execute(ClusterState currentState) {
DiscoveryNodes newNodes = currentState.nodes().removeDeadMembers(newMembers, newView.getCreator().toString());
DiscoveryNodes.Delta delta = newNodes.delta(currentState.nodes());
if (delta.added()) {
logger.warn("No new nodes should be created when a new discovery view is accepted");
}
// we want to send a new cluster state any how on view change (that's why its commented)
// for cases where we have client node joining (and it needs the cluster state)
// if (!delta.removed()) {
// // no nodes were removed, return the current state
// return currentState;
// }
return newClusterStateBuilder().state(currentState).nodes(newNodes).build();
}
});
} else {
// check whether I have been removed due to temporary disconnect
final String me = channel.getAddress().toString();
boolean foundMe = false;
for (DiscoveryNode node : clusterService.state().nodes()) {
if (node.id().equals(me)) {
foundMe = true;
break;
}
}
if (!foundMe) {
logger.warn("Disconnected from cluster, resending to master [{}], node [{}]", newView.getCreator(), localNode);
try {
channel.send(new Message(newView.getCreator(), channel.getAddress(), nodeMessagePayload()));
addressSet = true;
} catch (Exception e) {
addressSet = false;
logger.warn("Can't send address to master [" + newView.getCreator() + "] will try again later...", e);
}
}
}
}
private byte[] nodeMessagePayload() throws IOException {
BytesStreamOutput os = BytesStreamOutput.Cached.cached();
localNode.writeTo(os);
return os.copiedByteArray();
}
private void sendInitialStateEventIfNeeded() {
if (initialStateSent.compareAndSet(false, true)) {
for (InitialStateDiscoveryListener listener : initialStateListeners) {
listener.initialStateProcessed();
}
}
}
@Override public void suspect(Address suspectedMember) {
}
@Override public void block() {
logger.warn("Blocked...");
}
}

40
plugins/discovery/jgroups/src/main/java/org/elasticsearch/discovery/jgroups/JgroupsDiscoveryModule.java
View File

@ -1,40 +0,0 @@
/*
* Licensed to Elastic Search and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. Elastic Search 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.discovery.jgroups;
import org.elasticsearch.common.inject.AbstractModule;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.discovery.Discovery;
/**
* @author kimchy (Shay Banon)
*/
public class JgroupsDiscoveryModule extends AbstractModule {
private final Settings settings;
public JgroupsDiscoveryModule(Settings settings) {
this.settings = settings;
}
@Override protected void configure() {
bind(Discovery.class).to(JgroupsDiscovery.class).asEagerSingleton();
}
}

36
plugins/discovery/jgroups/src/main/java/org/elasticsearch/plugin/discovery/jgroups/JgroupsDiscoveryPlugin.java
View File

@ -1,36 +0,0 @@
/*
* Licensed to Elastic Search and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. Elastic Search 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.plugin.discovery.jgroups;
import org.elasticsearch.plugins.AbstractPlugin;
/**
* @author kimchy (shay.banon)
*/
public class JgroupsDiscoveryPlugin extends AbstractPlugin {
@Override public String name() {
return "discovery-jgroups";
}
@Override public String description() {
return "Adds jgroups as a discovery type";
}
}

24
plugins/discovery/jgroups/src/test/java/org/elasticsearch/discovery/jgroups/SimpleJgroupsDiscoveryTests.java
View File

@ -1,24 +0,0 @@
/*
* Licensed to Elastic Search and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. Elastic Search 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.discovery.jgroups;
public class SimpleJgroupsDiscoveryTests {
}

58
plugins/discovery/jgroups/src/test/java/org/elasticsearch/discovery/jgroups/SimpleJgroupsTests.java
View File

@ -1,58 +0,0 @@
/*
* Licensed to Elastic Search and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. Elastic Search 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.discovery.jgroups;
import org.jgroups.ChannelException;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
/**
* @author kimchy (Shay Banon)
*/
public class SimpleJgroupsTests {
static {
System.setProperty("jgroups.logging.log_factory_class", JgroupsCustomLogFactory.class.getName());
}
private JChannel channel1;
private JChannel channel2;
@BeforeMethod public void setupChannels() throws ChannelException {
channel1 = new JChannel("udp.xml");
channel1.connect("test");
channel2 = new JChannel("udp.xml");
channel2.connect("test");
}
@AfterMethod public void closeChannels() {
channel1.close();
channel2.close();
}
@Test public void testUdpJgroups() throws Exception {
channel1.send(new Message(null, channel1.getAddress(), "test"));
}
}

1
settings.gradle
View File

@ -14,7 +14,6 @@ include 'plugins-analysis-icu'
include 'plugins-mapper-attachments'
include 'plugins-client-groovy'
include 'plugins-transport-memcached'
include 'plugins-discovery-jgroups'
rootProject.name = 'elasticsearch-root'
rootProject.children.each {project ->