Skip to content
This repository was archived by the owner on Aug 13, 2026. It is now read-only.

Fix for #572. Refresh connection IP address info with (possibly) new DNS info when creating a new connection. #573

Merged
alexmoore merged 10 commits into from
Mar 2, 2016
Merged

Fix for #572. Refresh connection IP address info with (possibly) new DNS info when creating a new connection. #573

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
fbf10c2
Small code cleanups, adding more tests around node channel management…
alexmoore Nov 10, 2015
4f9a000
Refresh network connection addresses every time we open a connection
alexmoore Nov 11, 2015
87cd5c7
add warning message for DNSTTL=4eva
alexmoore Nov 11, 2015
44af471
Merge branch 'develop' into fixes/am/572
alexmoore Feb 12, 2016
02a57a5
Fix merge issues
alexmoore Feb 12, 2016
b12ad35
Change TTL check to cover default (null) and manually set -1 cases
alexmoore Feb 12, 2016
07d22c1
Fancier warning message and deduction on whether to log warning
alexmoore Feb 16, 2016
43d04a4
Merge branch 'develop' into fixes/am/572
alexmoore Mar 1, 2016
bc2d380
Merge branch 'develop' into fixes/am/572
alexmoore Mar 2, 2016
f31e557
Add some comments to explain test
alexmoore Mar 2, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
207 changes: 131 additions & 76 deletions src/main/java/com/basho/riak/client/core/RiakNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.security.KeyStore;
import java.security.Security;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
Expand Down Expand Up @@ -202,6 +203,7 @@ private RiakNode(Builder builder)
permits = new Sync(builder.maxConnections);
}

checkNetworkAddressCacheSettings();

this.state = State.CREATED;
}
Expand Down Expand Up @@ -246,14 +248,9 @@ public synchronized RiakNode start() throws UnknownHostException
ownsBootstrap = true;
}

InetSocketAddress socketAddress = new InetSocketAddress(remoteAddress, port);

if (socketAddress.isUnresolved())
{
throw new UnknownHostException("RiakNode:start - Failed resolving host " + remoteAddress);
}
bootstrap.handler(new RiakChannelInitializer(this));

bootstrap.handler(new RiakChannelInitializer(this)).remoteAddress(socketAddress);
refreshBootstrapRemoteAddress();

if (connectionTimeout > 0)
{
Expand All @@ -268,7 +265,7 @@ public synchronized RiakNode start() throws UnknownHostException
Channel channel;
try
{
channel = doGetConnection();
channel = doGetConnection(false);
minChannels.add(channel);
}
catch (ConnectionFailedException ex)
Expand All @@ -293,6 +290,19 @@ public synchronized RiakNode start() throws UnknownHostException
return this;
}

private void refreshBootstrapRemoteAddress() throws UnknownHostException
{
// Refresh the address, hope their DNS TTL settings allow this.
InetSocketAddress socketAddress = new InetSocketAddress(remoteAddress, port);

if (socketAddress.isUnresolved())
{
throw new UnknownHostException("RiakNode:start - Failed resolving host " + remoteAddress);
}

bootstrap.remoteAddress(socketAddress);
}

public synchronized Future<Boolean> shutdown()
{
stateCheck(State.RUNNING, State.HEALTH_CHECKING);
Expand Down Expand Up @@ -640,18 +650,23 @@ private Channel getConnection()
{
try
{
channel = doGetConnection();
channel = doGetConnection(true);

@lukebakken lukebakken Feb 12, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

It seems weird that we get a connection during shutdown.

@alexmoore alexmoore Feb 16, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Where do we get a connection on shutdown?

@lukebakken lukebakken Feb 16, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Is that what is happening here? shutdown() method calls doGetConnection ?

@alexmoore alexmoore Feb 16, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Not that I can see, the only places we call doGetConnection/1 are in start(), the general getConnection/1 method, and the health checker.

@lukebakken lukebakken Feb 16, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Argh stupid GH code folding. Even when I expanded the diff it still wasn't clear. Sorry!

channel.closeFuture().removeListener(inAvailableCloseListener);
}
catch (ConnectionFailedException ex)
{
permits.release();
}
catch (UnknownHostException ex)
{
permits.release();
logger.error("Unknown host encountered while trying to open connection; {}", ex);
}
}
return channel;
}

private Channel doGetConnection() throws ConnectionFailedException
private Channel doGetConnection(boolean forceAddressRefresh) throws ConnectionFailedException, UnknownHostException
{
ChannelWithIdleTime cwi;
while ((cwi = available.poll()) != null)
Expand All @@ -667,6 +682,11 @@ private Channel doGetConnection() throws ConnectionFailedException
}
}

if(forceAddressRefresh)
{
refreshBootstrapRemoteAddress();
}

ChannelFuture f = bootstrap.connect();

try
Expand Down Expand Up @@ -694,83 +714,87 @@ private Channel doGetConnection() throws ConnectionFailedException

if (trustStore != null)
{
SSLContext context;
try
{
context = SSLContext.getInstance("TLS");
TrustManagerFactory tmf =
TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(trustStore);
if (keyStore!=null)
{
KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
kmf.init(keyStore, keyPassword==null?"".toCharArray():keyPassword.toCharArray());
context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
}
else
{
context.init(null, tmf.getTrustManagers(), null);
}

}
catch (Exception ex)
{
c.close();
logger.error("Failure configuring SSL; {}:{} {}", remoteAddress, port, ex);
throw new ConnectionFailedException(ex);
}
setupTLSAndAuthenticate(c);
}

SSLEngine engine = context.createSSLEngine();
return c;

Set<String> protocols = new HashSet<String>(Arrays.asList(engine.getSupportedProtocols()));
}

if (protocols.contains("TLSv1.2"))
private void setupTLSAndAuthenticate(Channel c) throws ConnectionFailedException
{
SSLContext context;
try
{
context = SSLContext.getInstance("TLS");
TrustManagerFactory tmf =
TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(trustStore);
if(keyStore!=null)
{
engine.setEnabledProtocols(new String[] {"TLSv1.2"});
logger.debug("Using TLSv1.2");
KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
kmf.init(keyStore, keyPassword==null?"".toCharArray():keyPassword.toCharArray());
context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
}
else if (protocols.contains("TLSv1.1"))
else
{
engine.setEnabledProtocols(new String[] {"TLSv1.1"});
logger.debug("Using TLSv1.1");
context.init(null, tmf.getTrustManagers(), null);
}

engine.setUseClientMode(true);
RiakSecurityDecoder decoder = new RiakSecurityDecoder(engine, username, password);
c.pipeline().addFirst(decoder);
}
catch (Exception ex)
{
c.close();
logger.error("Failure configuring SSL; {}:{} {}", remoteAddress, port, ex);
throw new ConnectionFailedException(ex);
}

try
{
DefaultPromise<Void> promise = decoder.getPromise();
logger.debug("Waiting on SSL Promise");
promise.await();
SSLEngine engine = context.createSSLEngine();

if (promise.isSuccess())
{
logger.debug("Auth succeeded; {}:{}", remoteAddress, port);
}
else
{
c.close();
logger.error("Failure during Auth; {}:{} {}",remoteAddress, port, promise.cause());
throw new ConnectionFailedException(promise.cause());
}
Set<String> protocols = new HashSet<String>(Arrays.asList(engine.getSupportedProtocols()));

if (protocols.contains("TLSv1.2"))

@lukebakken lukebakken Nov 14, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I'm curious why not just pass the supported protocols as the enabled ones? TLS should negotiate the highest supported version.

@alexmoore alexmoore Feb 12, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

True. I've been meaning to revamp some of the TLS code + testing sometime soon so I'll roll this into that issue.

{
engine.setEnabledProtocols(new String[] {"TLSv1.2"});
logger.debug("Using TLSv1.2");
}
else if (protocols.contains("TLSv1.1"))
{
engine.setEnabledProtocols(new String[] {"TLSv1.1"});
logger.debug("Using TLSv1.1");
}

engine.setUseClientMode(true);
RiakSecurityDecoder decoder = new RiakSecurityDecoder(engine, username, password);
c.pipeline().addFirst(decoder);

try
{
DefaultPromise<Void> promise = decoder.getPromise();
logger.debug("Waiting on SSL Promise");
promise.await();

if (promise.isSuccess())
{
logger.debug("Auth succeeded; {}:{}", remoteAddress, port);
}
catch (InterruptedException e)
else
{
c.close();
logger.error("Thread interrupted during Auth; {}:{}",
remoteAddress, port);
Thread.currentThread().interrupt();
throw new ConnectionFailedException(e);
logger.error("Failure during Auth; {}:{} {}",remoteAddress, port, promise.cause());
throw new ConnectionFailedException(promise.cause());
}

}

return c;

}
catch (InterruptedException e)
{
c.close();
logger.error("Thread interrupted during Auth; {}:{}",
remoteAddress, port);
Thread.currentThread().interrupt();
throw new ConnectionFailedException(e);
}
}

/**
Expand Down Expand Up @@ -878,6 +902,34 @@ public void onException(Channel channel, final Throwable t)
}
}

private void checkNetworkAddressCacheSettings()
{
final String property = Security.getProperty("networkaddress.cache.ttl");

final boolean usingSecurityMgr = System.getSecurityManager() != null;
final boolean propertyUndefined = property == null;
boolean logWarning = false;

if(propertyUndefined && usingSecurityMgr)
{
logWarning = true;
}
else if(!propertyUndefined)
{
final int cacheTTL = Integer.parseInt(property);
logWarning = (cacheTTL == -1);
}

if (logWarning)
{
logger.warn(
"The Java Security \"networkaddress.cache.ttl\" property may be set to cache DNS lookups forever. " +
"Using domain names for Riak nodes or an intermediate load balancer could result in stale IP " +
"addresses being used for new connections, causing connection errors. " +
"If you use domain names for Riak nodes, please set this property to a value greater than zero.");
}
}

/**
* Returns the {@code remoteAddress} for this RiakNode
*
Expand Down Expand Up @@ -1063,7 +1115,7 @@ public void run()
}
}

private void checkHealth()
private void checkHealth()
{
try
{
Expand All @@ -1073,7 +1125,7 @@ private void checkHealth()
// connections from the available queue and either
// return/create a new one (meaning the node is up) or throw
// an exception if a connection can't be made.
Channel c = doGetConnection();
Channel c = doGetConnection(true);
logger.debug("Healthcheck channel: {} isOpen: {} handlers:{}", c.hashCode(), c.isOpen(), c.pipeline().names());


Expand Down Expand Up @@ -1123,18 +1175,21 @@ private void checkHealth()
{
healthCheckFailed(ex);
}
catch (IllegalStateException e)
catch (UnknownHostException ex)
{
healthCheckFailed(ex);
}
catch (IllegalStateException ex)
{
// no-op; there's a race condition where the bootstrap is shutting down
// right when a healthcheck occurs and netty will throw this
logger.debug("Illegal state exception during healthcheck.");
logger.debug("Stack: {}", e);
logger.debug("Stack: {}", ex);
}
catch (RuntimeException e)
catch (RuntimeException ex)
{
logger.error("Runtime exception during healthcheck: {}",e);
logger.error("Runtime exception during healthcheck: {}", ex);
}

}

private void healthCheckFailed(Throwable cause)
Expand Down
Loading