Enhance multicloud support

Author: Vincent Royer

core/src/main/java/org/apache/cassandra/service/ElassandraDaemon.java

@@ -439,14 +439,14 @@ public Settings nodeSettings(Settings settings) {
                 .put("network.bind_host", DatabaseDescriptor.getRpcAddress().getHostAddress())
                 .put("network.publish_host", FBUtilities.getBroadcastRpcAddress().getHostAddress())
                 .put("transport.bind_host", FBUtilities.getLocalAddress().getHostAddress())
-                .put("transport.publish_host", FBUtilities.getBroadcastAddress().getHostAddress())
+                .put("transport.publish_host", Boolean.getBoolean("es.use_internal_address") ? FBUtilities.getLocalAddress().getHostAddress() : FBUtilities.getBroadcastAddress().getHostAddress())
                 .put("path.data",getElasticsearchDataDir())
                 .put(settings)
                  // not overloadable settings.
-                .put("discovery.type","cassandra")
-                .put("node.data",true)
-                .put("node.master",true)
-                .put("node.name", CassandraDiscovery.buildNodeName(DatabaseDescriptor.getListenAddress()))
+                .put("discovery.type", "cassandra")
+                .put("node.data", true)
+                .put("node.master", true)
+                .put("node.name", CassandraDiscovery.buildNodeName(DatabaseDescriptor.getBroadcastAddress()))
                 .put("node.attr.dc", DatabaseDescriptor.getLocalDataCenter())
                 .put("node.attr.rack", DatabaseDescriptor.getEndpointSnitch().getRack(FBUtilities.getBroadcastAddress()))
                 .put("cluster.name", ClusterService.getElasticsearchClusterName(env.settings()))

core/src/main/java/org/elassandra/cluster/routing/AbstractSearchStrategy.java

@@ -106,8 +106,14 @@ public Router(final Index index, final String ksName, BiFunction<Index, UUID, Sh
                 this.strategy = Keyspace.open(ksName).getReplicationStrategy();
                 this.metadata = StorageService.instance.getTokenMetadata().cloneOnlyTokenMap();
                 for(DiscoveryNode node : clusterState.nodes()) {
-                    for(Token token : this.metadata.getTokens(node.getInetAddress())) 
-                        this.tokenToNodes.put(token, node);
+                    InetAddress endpoint = node.getNameAsInetAddress();
+                    if (endpoint == null) {
+                        endpoint = this.metadata.getEndpointForHostId(node.uuid());
+                    }
+                    if (endpoint != null && this.metadata.isMember(endpoint)) {
+                        for(Token token : this.metadata.getTokens(endpoint)) 
+                            this.tokenToNodes.put(token, node);
+                    }
                 }
             } else {
                 this.strategy = null;
@@ -128,7 +134,7 @@ public Router(final Index index, final String ksName, BiFunction<Index, UUID, Sh
 
                 // greenshard = available node -> token range bitset, 
                 boolean orphanRange = true;
-                for(InetAddress endpoint :  (this.metadata == null) ? Collections.singletonList(localNode.getInetAddress()) : this.strategy.calculateNaturalEndpoints(token, this.metadata)) {
+                for(InetAddress endpoint :  (this.metadata == null) ? Collections.singletonList(localNode.getNameAsInetAddress()) : this.strategy.calculateNaturalEndpoints(token, this.metadata)) {
                     UUID uuid = StorageService.instance.getHostId(endpoint);
                     DiscoveryNode node =  (uuid == null) ? clusterState.nodes().findByInetAddress(endpoint) : clusterState.nodes().get(uuid.toString());
                     if (node != null && node.status() == DiscoveryNode.DiscoveryNodeStatus.ALIVE) {

core/src/main/java/org/elassandra/discovery/CassandraDiscovery.java

@@ -151,8 +151,9 @@ protected void doStart()  {
 
             // initialize cluster from cassandra system.peers 
             // WARNING: system.peers may be incomplete because commitlogs not yet applied
-            for (UntypedResultSet.Row row : executeInternal("SELECT peer, data_center, rack, rpc_address, host_id from system." + SystemKeyspace.PEERS)) {
+            for (UntypedResultSet.Row row : executeInternal("SELECT peer, data_center, rack, preferred_ip, rpc_address, host_id from system." + SystemKeyspace.PEERS)) {
                 InetAddress peer = row.getInetAddress("peer");
+                InetAddress preferred_ip = row.has("preferred_ip") ? row.getInetAddress("preferred_ip") : null;
                 InetAddress rpc_address = row.has("rpc_address") ? row.getInetAddress("rpc_address") : null;
                 String datacenter = row.has("data_center") ? row.getString("data_center") : null;
                 String host_id = row.has("host_id") ? row.getUUID("host_id").toString() : null;
@@ -162,7 +163,11 @@ protected void doStart()  {
                     if (row.has("rack"))
                         attrs.put("rack", row.getString("rack"));
 
-                    DiscoveryNode dn = new DiscoveryNode(buildNodeName(peer), host_id, new InetSocketTransportAddress(rpc_address, publishPort()), attrs, CASSANDRA_ROLES, Version.CURRENT);
+                    DiscoveryNode dn = new DiscoveryNode(buildNodeName(peer), host_id, 
+                            new InetSocketTransportAddress(Boolean.getBoolean("es.use_internal_address") && preferred_ip != null ? preferred_ip : rpc_address, publishPort()), 
+                            attrs, 
+                            CASSANDRA_ROLES, 
+                            Version.CURRENT);
                     EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(peer);
                     if (state == null) {
                         dn.status( DiscoveryNodeStatus.UNKNOWN );
@@ -295,8 +300,9 @@ public void updateClusterGroupsFromGossiper() {
 
                         InetAddress internal_address = com.google.common.net.InetAddresses.forString(state.getApplicationState(ApplicationState.INTERNAL_IP).value);
                         InetAddress rpc_address = com.google.common.net.InetAddresses.forString(state.getApplicationState(ApplicationState.RPC_ADDRESS).value);
+                        
                         dn = new DiscoveryNode(buildNodeName(internal_address), hostId.toString(), 
-                                new InetSocketTransportAddress(rpc_address, publishPort()), attrs, CASSANDRA_ROLES, Version.CURRENT);
+                                new InetSocketTransportAddress(Boolean.getBoolean("es.use_internal_address") ? internal_address : rpc_address, publishPort()), attrs, CASSANDRA_ROLES, Version.CURRENT);
                         dn.status(status);
 
                         if (!localAddress.equals(endpoint)) {
@@ -518,7 +524,8 @@ private boolean isMember(InetAddress endpoint) {
      * ES RPC adress can be different from the cassandra broadcast address.
      */
     private boolean isNormal(DiscoveryNode node) {
-        EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(node.getInetAddress());
+        // endpoint address = C* broadcast address = Elasticsearch node name (transport may be bound to C* internal or C* RPC broadcast)
+        EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(InetAddresses.forString(node.getName()));
         if (state == null) {
             logger.warn("Node endpoint address=[{}] name=[{}] state not found", node.getInetAddress(), node.getName());
             return false;

core/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.cluster.node;
 
 import org.elasticsearch.Version;
-import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.InetSocketTransportAddress;
 import org.elasticsearch.common.transport.LocalTransportAddress;
@@ -80,6 +80,7 @@ public static boolean isIngestNode(Settings settings) {
     private final String hostName;
     private final String hostAddress;
     private final TransportAddress address;
+    private transient final InetAddress nodeNameAddress;
     private final Map<String, String> attributes;
     private final Version version;
     private final Set<Role> roles;
@@ -139,6 +140,10 @@ public UUID uuid() {
         return this.nodeUuid;
     }
 
+    public InetAddress getNameAsInetAddress() {
+        return this.nodeNameAddress;
+    }
+    
     /**
      * The inet listen address of the node.
      */
@@ -229,11 +234,17 @@ public DiscoveryNode(String nodeName, String nodeId, TransportAddress address,
      */
     public DiscoveryNode(String nodeName, String nodeId, String ephemeralId, String hostName, String hostAddress,
                          TransportAddress address, Map<String, String> attributes, Set<Role> roles, Version version) {
+        InetAddress nodeAddr = null;
         if (nodeName != null) {
             this.nodeName = nodeName.intern();
+            try {
+                nodeAddr = InetAddresses.forString(nodeName);
+            } catch (Exception e) {
+            }
         } else {
             this.nodeName = "";
         }
+        this.nodeNameAddress = nodeAddr;
         this.nodeId = nodeId.intern();
         try {
             this.nodeUuid = UUID.fromString(nodeId);
@@ -293,6 +304,12 @@ public static Set<Role> getRolesFromSettings(Settings settings) {
      */
     public DiscoveryNode(StreamInput in) throws IOException {
         this.nodeName = in.readString().intern();
+        InetAddress nodeNameAddress = null;
+        try {
+            nodeNameAddress = InetAddresses.forString(this.nodeName);
+        } catch (Exception e) {
+        }
+        this.nodeNameAddress = nodeNameAddress;
         this.nodeId = in.readString().intern();
         this.ephemeralId = in.readString().intern();
         this.hostName = in.readString().intern();

core/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodes.java

@@ -269,7 +269,7 @@ public DiscoveryNode findByAddress(TransportAddress address) {
     public DiscoveryNode findByInetAddress(InetAddress address) {
         for (ObjectCursor<DiscoveryNode> cursor : nodes.values()) {
             DiscoveryNode node = cursor.value;
-            if (node.getInetAddress().equals(address)) {
+            if (node.getInetAddress().equals(address) || node.getName().equals(address.getHostAddress())) {
                 return node;
             }
         }

core/src/main/java/org/elasticsearch/cluster/routing/IndexRoutingTable.java

@@ -377,7 +377,7 @@ public Builder(Index index, ClusterService clusterService, ClusterState targetSt
             } catch (NullPointerException | java.lang.AssertionError e) {
                 // thrown by cassandra when the keyspace is not yet create locally. 
                 // We must wait for a gossip schema change to update the routing Table.
-                Loggers.getLogger(getClass().getName()).warn("Keyspace not available: {}", e, this.index);
+                Loggers.getLogger(getClass().getName()).warn("Keyspace not available for index ["+this.index+"]", e);
             }
         }
 
@@ -398,9 +398,9 @@ public Builder(Index index, ClusterService clusterService, ClusterState targetSt
             } catch (NullPointerException | java.lang.AssertionError e) {
                 // thrown by cassandra when the keyspace is not yet create locally. 
                 // We must wait for a gossip schema change to update the routing Table.
-                Loggers.getLogger(getClass()).warn("Keyspace {} not  available", e, this.index);
+                Loggers.getLogger(getClass()).warn("Keyspace not available for index ["+this.index+"]", e);
             } catch (Exception e1) {
-                Loggers.getLogger(getClass()).warn("Failde to compute route for {}", e1, this.index);
+                Loggers.getLogger(getClass()).warn("Failed to compute route for index ["+this.index+"]", e1);
             }
         }
 

docs/elassandra/source/configuration.rst

@@ -46,12 +46,15 @@ Elasticsearch configuration rely on cassandra configuration file **conf/cassandr
 
 Node role (master, primary, data) is automatically set by elassandra, standard configuration should only set **cluster_name**, **rpc_address** in the ``conf/cassandra.yaml``.
 
-By default, Elasticsearch HTTP is bound to Cassandra RPC addresses, while Elasticsearch transport protocol is bound to Cassandra internal addresses. 
-You can overload these default settings by defining Elasticsearch network settings in conf/elasticsearch.yaml (in order to bind Elasticsearch transport on 
-a public interface if you want to use the Elasticsearch transport client from your application).
+By default, Elasticsearch HTTP is bound to Cassandra RPC address ``rpc_address``, while Elasticsearch transport protocol is bound to Cassandra internal address ``listen_address``.
+You can overload these default settings by defining Elasticsearch network settings in conf/elasticsearch.yaml (in order to bind Elasticsearch transport on another interface).
+
+By default, Elasticsearch transport publish adress is the Cassandra broadcast adress. However, in some network configurations (including multi-cloud deployment), the Cassandra broadcast adress is a public address managed by a firewall, and
+it would invlove a network overhead for elasticsearch inter-node communication. In such case, you can set the system property ``es.use_internal_address=true`` to use the Cassandra  ``listen_address`` as the elasticsearch transport published address.
+
 
 .. CAUTION::
-   If you use the `GossipPropertyFile <https://docs.datastax.com/en/cassandra/2.0/cassandra/architecture/architectureSnitchGossipPF_c.html>`_ Snitch to configure your cassandra datacenter and rack properties in **conf/cassandra-rackdc.properties**, keep
+   If you use the `GossipingPropertyFile <https://docs.datastax.com/en/cassandra/2.0/cassandra/architecture/architectureSnitchGossipPF_c.html>`_ Snitch to configure your cassandra datacenter and rack properties in **conf/cassandra-rackdc.properties**, keep
    in mind this snitch falls back to the PropertyFileSnitch when gossip is not enabled. So, when re-starting the first node, dead nodes can appear in the default DC and rack configured in **conf/cassandra-topology.properties**. This also
    breaks the replica placement strategy and the computation of the Elasticsearch routing tables. So it is strongly recommended to set the same default rack and datacenter in both the **conf/cassandra-topology.properties** and **conf/cassandra-rackdc.properties**.