Skip to content

Commit

Permalink
Go through NodeManager to get current host name and address for prest…
Browse files Browse the repository at this point in the history
…osql
  • Loading branch information
James Taylor committed Nov 18, 2020
1 parent 68441a2 commit cb817d7
Show file tree
Hide file tree
Showing 6 changed files with 65 additions and 15 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,18 @@

import io.prestosql.spi.Node;
import io.prestosql.spi.NodeManager;
import java.net.UnknownHostException;
import java.util.Set;
import java.util.stream.Collectors;

import static java.util.Objects.requireNonNull;

/*
* Class that encapsulates logic using NodeManager to satisfy API
* of ClusterManager. This logic was put in a separate class so
* that both PrestoClusterManager and SyncPrestoClusterManager
* can share it.
*/
public class ClusterManagerNodeGetter {
public static Set<String> getNodesInternal(NodeManager nodeManager)
{
Expand All @@ -27,4 +34,15 @@ public static Set<String> getNodesInternal(NodeManager nodeManager)
.map(Node::getHost)
.collect(Collectors.toSet());
}

public static String getCurrentNodeHostname(NodeManager nodeManager) {
requireNonNull(nodeManager, "nodeManager is null");
return nodeManager.getCurrentNode().getHost();
}

public static String getCurrentNodeHostAddress(NodeManager nodeManager)
throws UnknownHostException {
requireNonNull(nodeManager, "nodeManager is null");
return nodeManager.getCurrentNode().getHostAndPort().toInetAddress().getHostAddress();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
*/
package com.qubole.rubix.prestosql;

import static java.util.Objects.requireNonNull;

import com.qubole.rubix.spi.AsyncClusterManager;
import com.qubole.rubix.spi.ClusterType;
import io.prestosql.spi.Node;
Expand Down Expand Up @@ -49,11 +51,27 @@ public void initialize(Configuration conf) throws UnknownHostException
}

@Override
public Set<String> getNodesInternal()
{
public Set<String> getNodesInternal() {
return ClusterManagerNodeGetter.getNodesInternal(nodeManager);
}

@Override
protected String getCurrentNodeHostname() {
return ClusterManagerNodeGetter.getCurrentNodeHostname(nodeManager);
}

@Override
protected String getCurrentNodeHostAddress() {
try {
return ClusterManagerNodeGetter.getCurrentNodeHostAddress(nodeManager);
}
catch (UnknownHostException e) {
log.warn("Could not get HostAddress from NodeManager", e);
// fallback
}
return super.getCurrentNodeHostAddress();
}

@Override
public ClusterType getClusterType()
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,17 +52,11 @@ public class StandaloneNodeManager
private final Node currentNode;
private final int serverPort;

public StandaloneNodeManager(Configuration conf) {
public StandaloneNodeManager(Configuration conf)
throws UnknownHostException {
this.serverPort = conf.getInt(SERVER_PORT_CONF_KEY, DEFAULT_SERVER_PORT);
this.serverAddress = ClusterUtil.getMasterHostname(conf);
Node currentNode = null;
try {
currentNode = new StandaloneNode(URI.create("http://" + InetAddress.getLocalHost().getHostAddress()));
}
catch (UnknownHostException e) {
LOG.warn("Unable to set current node", e);
}
this.currentNode = currentNode;
this.currentNode = new StandaloneNode(URI.create("http://" + InetAddress.getLocalHost().getHostAddress()));
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import com.qubole.rubix.spi.SyncClusterManager;
import io.prestosql.spi.Node;
import io.prestosql.spi.NodeManager;
import java.net.UnknownHostException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

Expand All @@ -38,11 +39,27 @@ protected boolean hasStateChanged() {
}

@Override
public Set<String> getNodesInternal()
{
public Set<String> getNodesInternal() {
return ClusterManagerNodeGetter.getNodesInternal(PrestoClusterManager.NODE_MANAGER);
}

@Override
protected String getCurrentNodeHostname() {
return ClusterManagerNodeGetter.getCurrentNodeHostname(PrestoClusterManager.NODE_MANAGER);
}

@Override
protected String getCurrentNodeHostAddress() {
try {
return ClusterManagerNodeGetter.getCurrentNodeHostAddress(PrestoClusterManager.NODE_MANAGER);
}
catch (UnknownHostException e) {
log.warn("Could not get HostAddress from NodeManager", e);
// fallback
}
return super.getCurrentNodeHostAddress();
}

@Override
public ClusterType getClusterType()
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,8 @@ private HttpServer createServer(String endpoint1, HttpHandler handler1, String e
return server;
}

abstract protected ClusterManager newPrestoClusterManager(Configuration conf);
abstract protected ClusterManager newPrestoClusterManager(Configuration conf)
throws UnknownHostException;

private ClusterManager getPrestoClusterManager()
throws UnknownHostException
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,13 @@
package com.qubole.rubix.prestosql;

import com.qubole.rubix.spi.ClusterManager;
import java.net.UnknownHostException;
import org.apache.hadoop.conf.Configuration;

public class TestSyncClusterManager extends TestClusterManager {
@Override
protected ClusterManager newPrestoClusterManager(Configuration conf) {
protected ClusterManager newPrestoClusterManager(Configuration conf)
throws UnknownHostException {
PrestoClusterManager.setNodeManager(new StandaloneNodeManager(conf));
return new SyncPrestoClusterManager();
}
Expand Down

0 comments on commit cb817d7

Please sign in to comment.