Skip to content

Commit

Permalink
Do not resolve addresses in remote connection info (#36671)
Browse files Browse the repository at this point in the history 
The remote connection info API leads to resolving addresses of seed
nodes when invoked. This is problematic because if a hostname fails to
resolve, we would not display any remote connection info. Yet, a
hostname not resolving can happen across remote clusters, especially in
the modern world of cloud services with dynamically chaning
IPs. Instead, the remote connection info API should be providing the
configured seed nodes. This commit changes the remote connection info to
display the configured seed nodes, avoiding a hostname resolution. Note
that care was taken to preserve backwards compatibility with previous
versions that expect the remote connection info to serialize a transport
address instead of a string representing the hostname.
  • Loading branch information
@jasontedor @davidkyle
jasontedor authored and davidkyle committed Dec 18, 2018
1 parent e9bd724 commit 86e6d18
Show file tree
Hide file tree
Showing 6 changed files with 188 additions and 124 deletions.
29 changes: 19 additions & 10 deletions server/src/main/java/org/elasticsearch/transport/RemoteClusterAware.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -183,10 +183,11 @@ protected RemoteClusterAware(Settings settings) {
* (ProxyAddresss, [SeedNodeSuppliers]). If a cluster is configured with a proxy address all seed nodes will point to
* {@link TransportAddress#META_ADDRESS} and their configured address will be used as the hostname for the generated discovery node.
*/
protected static Map<String, Tuple<String, List<Supplier<DiscoveryNode>>>> buildRemoteClustersDynamicConfig(Settings settings) {
final Map<String, Tuple<String, List<Supplier<DiscoveryNode>>>> remoteSeeds =
protected static Map<String, Tuple<String, List<Tuple<String, Supplier<DiscoveryNode>>>>> buildRemoteClustersDynamicConfig(
final Settings settings) {
final Map<String, Tuple<String, List<Tuple<String, Supplier<DiscoveryNode>>>>> remoteSeeds =
buildRemoteClustersDynamicConfig(settings, REMOTE_CLUSTERS_SEEDS);
final Map<String, Tuple<String, List<Supplier<DiscoveryNode>>>> searchRemoteSeeds =
final Map<String, Tuple<String, List<Tuple<String, Supplier<DiscoveryNode>>>>> searchRemoteSeeds =
buildRemoteClustersDynamicConfig(settings, SEARCH_REMOTE_CLUSTERS_SEEDS);
// sort the intersection for predictable output order
final NavigableSet<String> intersection =
Expand All @@ -205,7 +206,7 @@ protected static Map<String, Tuple<String, List<Supplier<DiscoveryNode>>>> build
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}

private static Map<String, Tuple<String, List<Supplier<DiscoveryNode>>>> buildRemoteClustersDynamicConfig(
private static Map<String, Tuple<String, List<Tuple<String, Supplier<DiscoveryNode>>>>> buildRemoteClustersDynamicConfig(
final Settings settings, final Setting.AffixSetting<List<String>> seedsSetting) {
final Stream<Setting<List<String>>> allConcreteSettings = seedsSetting.getAllConcreteSettings(settings);
return allConcreteSettings.collect(
Expand All @@ -214,9 +215,9 @@ private static Map<String, Tuple<String, List<Supplier<DiscoveryNode>>>> buildRe
List<String> addresses = concreteSetting.get(settings);
final boolean proxyMode =
REMOTE_CLUSTERS_PROXY.getConcreteSettingForNamespace(clusterName).existsOrFallbackExists(settings);
List<Supplier<DiscoveryNode>> nodes = new ArrayList<>(addresses.size());
List<Tuple<String, Supplier<DiscoveryNode>>> nodes = new ArrayList<>(addresses.size());
for (String address : addresses) {
nodes.add(() -> buildSeedNode(clusterName, address, proxyMode));
nodes.add(Tuple.tuple(address, () -> buildSeedNode(clusterName, address, proxyMode)));
}
return new Tuple<>(REMOTE_CLUSTERS_PROXY.getConcreteSettingForNamespace(clusterName).get(settings), nodes);
}));
Expand Down Expand Up @@ -304,16 +305,24 @@ public void listenForUpdates(ClusterSettings clusterSettings) {
(namespace, value) -> {});
}


protected static InetSocketAddress parseSeedAddress(String remoteHost) {
String host = remoteHost.substring(0, indexOfPortSeparator(remoteHost));
static InetSocketAddress parseSeedAddress(String remoteHost) {
final Tuple<String, Integer> hostPort = parseHostPort(remoteHost);
final String host = hostPort.v1();
assert hostPort.v2() != null : remoteHost;
final int port = hostPort.v2();
InetAddress hostAddress;
try {
hostAddress = InetAddress.getByName(host);
} catch (UnknownHostException e) {
throw new IllegalArgumentException("unknown host [" + host + "]", e);
}
return new InetSocketAddress(hostAddress, parsePort(remoteHost));
return new InetSocketAddress(hostAddress, port);
}

public static Tuple<String, Integer> parseHostPort(final String remoteHost) {
final String host = remoteHost.substring(0, indexOfPortSeparator(remoteHost));
final int port = parsePort(remoteHost);
return Tuple.tuple(host, port);
}

private static int parsePort(String remoteHost) {
Expand Down
25 changes: 16 additions & 9 deletions server/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
Expand Down Expand Up @@ -95,7 +96,7 @@ final class RemoteClusterConnection implements TransportConnectionListener, Clos
private final Predicate<DiscoveryNode> nodePredicate;
private final ThreadPool threadPool;
private volatile String proxyAddress;
private volatile List<Supplier<DiscoveryNode>> seedNodes;
private volatile List<Tuple<String, Supplier<DiscoveryNode>>> seedNodes;
private volatile boolean skipUnavailable;
private final ConnectHandler connectHandler;
private final TimeValue initialConnectionTimeout;
Expand All @@ -111,15 +112,15 @@ final class RemoteClusterConnection implements TransportConnectionListener, Clos
* @param nodePredicate a predicate to filter eligible remote nodes to connect to
* @param proxyAddress the proxy address
*/
RemoteClusterConnection(Settings settings, String clusterAlias, List<Supplier<DiscoveryNode>> seedNodes,
RemoteClusterConnection(Settings settings, String clusterAlias, List<Tuple<String, Supplier<DiscoveryNode>>> seedNodes,
TransportService transportService, int maxNumRemoteConnections, Predicate<DiscoveryNode> nodePredicate,
String proxyAddress) {
this(settings, clusterAlias, seedNodes, transportService, maxNumRemoteConnections, nodePredicate, proxyAddress,
createConnectionManager(settings, clusterAlias, transportService));
}

// Public for tests to pass a StubbableConnectionManager
RemoteClusterConnection(Settings settings, String clusterAlias, List<Supplier<DiscoveryNode>> seedNodes,
RemoteClusterConnection(Settings settings, String clusterAlias, List<Tuple<String, Supplier<DiscoveryNode>>> seedNodes,
TransportService transportService, int maxNumRemoteConnections, Predicate<DiscoveryNode> nodePredicate,
String proxyAddress, ConnectionManager connectionManager) {
this.transportService = transportService;
Expand Down Expand Up @@ -155,7 +156,10 @@ private static DiscoveryNode maybeAddProxyAddress(String proxyAddress, Discovery
/**
* Updates the list of seed nodes for this cluster connection
*/
synchronized void updateSeedNodes(String proxyAddress, List<Supplier<DiscoveryNode>> seedNodes, ActionListener<Void> connectListener) {
synchronized void updateSeedNodes(
final String proxyAddress,
final List<Tuple<String, Supplier<DiscoveryNode>>> seedNodes,
final ActionListener<Void> connectListener) {
this.seedNodes = Collections.unmodifiableList(new ArrayList<>(seedNodes));
this.proxyAddress = proxyAddress;
connectHandler.connect(connectListener);
Expand Down Expand Up @@ -465,7 +469,7 @@ protected void doRun() {
maybeConnect();
}
});
collectRemoteNodes(seedNodes.iterator(), transportService, connectionManager, listener);
collectRemoteNodes(seedNodes.stream().map(Tuple::v2).iterator(), transportService, connectionManager, listener);
}
});
}
Expand Down Expand Up @@ -672,10 +676,13 @@ void addConnectedNode(DiscoveryNode node) {
* Get the information about remote nodes to be rendered on {@code _remote/info} requests.
*/
public RemoteConnectionInfo getConnectionInfo() {
List<TransportAddress> seedNodeAddresses = seedNodes.stream().map(node -> node.get().getAddress()).collect
(Collectors.toList());
return new RemoteConnectionInfo(clusterAlias, seedNodeAddresses, maxNumRemoteConnections, connectedNodes.size(),
initialConnectionTimeout, skipUnavailable);
return new RemoteConnectionInfo(
clusterAlias,
seedNodes.stream().map(Tuple::v1).collect(Collectors.toList()),
maxNumRemoteConnections,
connectedNodes.size(),
initialConnectionTimeout,
skipUnavailable);
}

int getNumNodesConnected() {
Expand Down
16 changes: 9 additions & 7 deletions server/src/main/java/org/elasticsearch/transport/RemoteClusterService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,7 @@ public String getKey(final String key) {
* @param seeds a cluster alias to discovery node mapping representing the remote clusters seeds nodes
* @param connectionListener a listener invoked once every configured cluster has been connected to
*/
private synchronized void updateRemoteClusters(Map<String, Tuple<String, List<Supplier<DiscoveryNode>>>> seeds,
private synchronized void updateRemoteClusters(Map<String, Tuple<String, List<Tuple<String, Supplier<DiscoveryNode>>>>> seeds,
ActionListener<Void> connectionListener) {
if (seeds.containsKey(LOCAL_CLUSTER_GROUP_KEY)) {
throw new IllegalArgumentException("remote clusters must not have the empty string as its key");
Expand All @@ -212,8 +212,8 @@ private synchronized void updateRemoteClusters(Map<String, Tuple<String, List<Su
} else {
CountDown countDown = new CountDown(seeds.size());
remoteClusters.putAll(this.remoteClusters);
for (Map.Entry<String, Tuple<String, List<Supplier<DiscoveryNode>>>> entry : seeds.entrySet()) {
List<Supplier<DiscoveryNode>> seedList = entry.getValue().v2();
for (Map.Entry<String, Tuple<String, List<Tuple<String, Supplier<DiscoveryNode>>>>> entry : seeds.entrySet()) {
List<Tuple<String, Supplier<DiscoveryNode>>> seedList = entry.getValue().v2();
String proxyAddress = entry.getValue().v1();

RemoteClusterConnection remote = this.remoteClusters.get(entry.getKey());
Expand Down Expand Up @@ -408,9 +408,10 @@ void updateRemoteCluster(
final List<String> addresses,
final String proxyAddress,
final ActionListener<Void> connectionListener) {
final List<Supplier<DiscoveryNode>> nodes = addresses.stream().<Supplier<DiscoveryNode>>map(address -> () ->
buildSeedNode(clusterAlias, address, Strings.hasLength(proxyAddress))
).collect(Collectors.toList());
final List<Tuple<String, Supplier<DiscoveryNode>>> nodes =
addresses.stream().<Tuple<String, Supplier<DiscoveryNode>>>map(address -> Tuple.tuple(address, () ->
buildSeedNode(clusterAlias, address, Strings.hasLength(proxyAddress)))
).collect(Collectors.toList());
updateRemoteClusters(Collections.singletonMap(clusterAlias, new Tuple<>(proxyAddress, nodes)), connectionListener);
}

Expand All @@ -421,7 +422,8 @@ void updateRemoteCluster(
void initializeRemoteClusters() {
final TimeValue timeValue = REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING.get(settings);
final PlainActionFuture<Void> future = new PlainActionFuture<>();
Map<String, Tuple<String, List<Supplier<DiscoveryNode>>>> seeds = RemoteClusterAware.buildRemoteClustersDynamicConfig(settings);
Map<String, Tuple<String, List<Tuple<String, Supplier<DiscoveryNode>>>>> seeds =
RemoteClusterAware.buildRemoteClustersDynamicConfig(settings);
updateRemoteClusters(seeds, future);
try {
future.get(timeValue.millis(), TimeUnit.MILLISECONDS);
Expand Down
52 changes: 44 additions & 8 deletions server/src/main/java/org/elasticsearch/transport/RemoteConnectionInfo.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,11 @@
* specific language governing permissions and limitations
* under the License.
*/

package org.elasticsearch.transport;

import org.elasticsearch.Version;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.io.stream.Writeable;
Expand All @@ -27,25 +29,29 @@
import org.elasticsearch.common.xcontent.ToXContentFragment;
import org.elasticsearch.common.xcontent.XContentBuilder;

import static java.util.Collections.emptyList;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

import static java.util.Collections.emptyList;

/**
* This class encapsulates all remote cluster information to be rendered on
* {@code _remote/info} requests.
*/
public final class RemoteConnectionInfo implements ToXContentFragment, Writeable {
final List<TransportAddress> seedNodes;
final List<String> seedNodes;
final int connectionsPerCluster;
final TimeValue initialConnectionTimeout;
final int numNodesConnected;
final String clusterAlias;
final boolean skipUnavailable;

RemoteConnectionInfo(String clusterAlias, List<TransportAddress> seedNodes,
RemoteConnectionInfo(String clusterAlias, List<String> seedNodes,
int connectionsPerCluster, int numNodesConnected,
TimeValue initialConnectionTimeout, boolean skipUnavailable) {
this.clusterAlias = clusterAlias;
Expand All @@ -57,7 +63,17 @@ public final class RemoteConnectionInfo implements ToXContentFragment, Writeable
}

public RemoteConnectionInfo(StreamInput input) throws IOException {
seedNodes = input.readList(TransportAddress::new);
if (input.getVersion().onOrAfter(Version.V_7_0_0)) {
seedNodes = Arrays.asList(input.readStringArray());
} else {
// versions prior to 7.0.0 sent the resolved transport address of the seed nodes
final List<TransportAddress> transportAddresses = input.readList(TransportAddress::new);
seedNodes =
transportAddresses
.stream()
.map(a -> a.address().getHostString() + ":" + a.address().getPort())
.collect(Collectors.toList());
}
if (input.getVersion().before(Version.V_7_0_0)) {
/*
* Versions before 7.0 sent the HTTP addresses of all nodes in the
Expand All @@ -78,7 +94,26 @@ public RemoteConnectionInfo(StreamInput input) throws IOException {

@Override
public void writeTo(StreamOutput out) throws IOException {
out.writeList(seedNodes);
if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
out.writeStringArray(seedNodes.toArray(new String[0]));
} else {
// versions prior to 7.0.0 received the resolved transport address of the seed nodes
out.writeList(seedNodes
.stream()
.map(
s -> {
final Tuple<String, Integer> hostPort = RemoteClusterAware.parseHostPort(s);
assert hostPort.v2() != null : s;
try {
return new TransportAddress(
InetAddress.getByAddress(hostPort.v1(), TransportAddress.META_ADDRESS.getAddress()),
hostPort.v2());
} catch (final UnknownHostException e) {
throw new AssertionError(e);
}
})
.collect(Collectors.toList()));
}
if (out.getVersion().before(Version.V_7_0_0)) {
/*
* Versions before 7.0 sent the HTTP addresses of all nodes in the
Expand All @@ -104,8 +139,8 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
builder.startObject(clusterAlias);
{
builder.startArray("seeds");
for (TransportAddress addr : seedNodes) {
builder.value(addr.toString());
for (String addr : seedNodes) {
builder.value(addr);
}
builder.endArray();
builder.field("connected", numNodesConnected > 0);
Expand Down Expand Up @@ -136,4 +171,5 @@ public int hashCode() {
return Objects.hash(seedNodes, connectionsPerCluster, initialConnectionTimeout,
numNodesConnected, clusterAlias, skipUnavailable);
}

}
Loading

0 comments on commit 86e6d18

Please sign in to comment.