Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Do not resolve addresses in remote connection info #36671

Merged
merged 3 commits into from
Dec 17, 2018
Merged

Do not resolve addresses in remote connection info #36671

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2bdeca9
Do not resolve addresses in remote connection info
jasontedor Dec 15, 2018
33c9806
Add port assertion
jasontedor Dec 15, 2018
b5f9477
Reuse
jasontedor Dec 15, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
29 changes: 19 additions & 10 deletions server/src/main/java/org/elasticsearch/transport/RemoteClusterAware.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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)) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is the bwc aspect of serialization sufficiently tested? I do see RemoteClusterConnectionTests#testRemoteConnectionInfoBwComp, just wondering whether that is good enough.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that's good enough. It shows that if a < 7.0.0 node sends us how it serializes messages today that we can de-serialize it and get what we expect, and shows that if we are sending to a < 7.0.0 that we serialize as that node expects. Also, those versions will change to 6.6.0 when this is backported.

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);
}

}