Skip to content

Commit

Permalink
Merge pull request #256 from gregschohn/ProxyConnectionCache
Browse files Browse the repository at this point in the history
Add optional connection pooling to the destination server for the Capture Proxy
  • Loading branch information
gregschohn authored Aug 14, 2023
2 parents 701c38b + f4392ce commit edfb16a
Show file tree
Hide file tree
Showing 13 changed files with 666 additions and 120 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.opensearch.migrations.trafficcapture.IConnectionCaptureFactory;
import org.opensearch.migrations.trafficcapture.StreamChannelConnectionCaptureSerializer;
import org.opensearch.migrations.trafficcapture.kafkaoffloader.KafkaCaptureFactory;
import org.opensearch.migrations.trafficcapture.proxyserver.netty.BacksideConnectionPool;
import org.opensearch.migrations.trafficcapture.proxyserver.netty.NettyScanningHttpProxy;
import org.opensearch.security.ssl.DefaultSecurityKeyStore;
import org.opensearch.security.ssl.util.SSLConfigConstants;
Expand All @@ -31,6 +32,7 @@
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.Optional;
import java.util.Properties;
import java.util.UUID;
Expand Down Expand Up @@ -100,6 +102,25 @@ static class Parameters {
arity = 1,
description = "Exposed port for clients to connect to this proxy.")
int frontsidePort = 0;
@Parameter(required = false,
names = {"--numThreads"},
arity = 1,
description = "How many threads netty should create in its event loop group")
int numThreads = 1;
@Parameter(required = false,
names = {"--destinationConnectionPoolSize"},
arity = 1,
description = "Number of socket connections that should be maintained to the destination server " +
"to reduce the perceived latency to clients. Each thread will have its own cache, so the " +
"total number of outstanding warm connections will be multiplied by numThreads.")
int destinationConnectionPoolSize = 0;
@Parameter(required = false,
names = {"--destinationConnectionPoolTimeout"},
arity = 1,
description = "Of the socket connections maintained by the destination connection pool, " +
"how long after connection should the be recycled " +
"(closed with a new connection taking its place)")
String destinationConnectionPoolTimeout = "PT30S";
}

public static Parameters parseArgs(String[] args) {
Expand Down Expand Up @@ -243,10 +264,14 @@ public static void main(String[] args) throws InterruptedException, IOException

sksOp.ifPresent(x->x.initHttpSSLConfig());
var proxy = new NettyScanningHttpProxy(params.frontsidePort);

try {
proxy.start(backsideUri, loadBacksideSslContext(backsideUri, params.allowInsecureConnectionsToBackside),
sksOp.map(sks-> (Supplier<SSLEngine>) () -> {
var pooledConnectionTimeout = params.destinationConnectionPoolSize == 0 ? Duration.ZERO :
Duration.parse(params.destinationConnectionPoolTimeout);
var backsideConnectionPool = new BacksideConnectionPool(backsideUri,
loadBacksideSslContext(backsideUri, params.allowInsecureConnectionsToBackside),
params.destinationConnectionPoolSize, pooledConnectionTimeout);
proxy.start(backsideConnectionPool, params.numThreads,
sksOp.map(sks -> (Supplier<SSLEngine>) () -> {
try {
var sslEngine = sks.createHTTPSSLEngine();
return sslEngine;
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
package org.opensearch.migrations.trafficcapture.proxyserver.netty;

import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import org.slf4j.event.Level;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.DefaultChannelPromise;
import io.netty.channel.EventLoop;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.concurrent.FastThreadLocal;
import lombok.extern.slf4j.Slf4j;

import javax.net.ssl.SSLEngine;
import java.net.URI;
import java.time.Duration;
import java.util.HashMap;
import java.util.StringJoiner;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
public class BacksideConnectionPool {
private final URI backsideUri;
private final SslContext backsideSslContext;
private final FastThreadLocal channelClassToConnectionCacheForEachThread;
private final Duration inactivityTimeout;
private final int poolSize;

public BacksideConnectionPool(URI backsideUri, SslContext backsideSslContext,
int poolSize, Duration inactivityTimeout) {
this.backsideUri = backsideUri;
this.backsideSslContext = backsideSslContext;
this.channelClassToConnectionCacheForEachThread = new FastThreadLocal();
this.inactivityTimeout = inactivityTimeout;
this.poolSize = poolSize;
}

public ChannelFuture getOutboundConnectionFuture(EventLoop eventLoop) {
if (poolSize == 0) {
return buildConnectionFuture(eventLoop);
}
return getExpiringWarmChannelPool(eventLoop).getAvailableOrNewItem();
}

private ExpiringSubstitutableItemPool<ChannelFuture, Void>
getExpiringWarmChannelPool(EventLoop eventLoop) {
var thisContextsConnectionCache = (ExpiringSubstitutableItemPool<ChannelFuture, Void>)
channelClassToConnectionCacheForEachThread.get();
if (thisContextsConnectionCache == null) {
thisContextsConnectionCache =
new ExpiringSubstitutableItemPool<ChannelFuture, Void>(inactivityTimeout,
eventLoop,
() -> buildConnectionFuture(eventLoop),
x->x.channel().close(), poolSize, Duration.ZERO);
if (log.isInfoEnabled()) {
final var finalChannelClassToChannelPoolMap = thisContextsConnectionCache;
logProgressAtInterval(Level.INFO, eventLoop,
thisContextsConnectionCache, Duration.ofSeconds(30));
}
channelClassToConnectionCacheForEachThread.set(thisContextsConnectionCache);
}

return thisContextsConnectionCache;
}

private void logProgressAtInterval(Level logLevel, EventLoop eventLoop,
ExpiringSubstitutableItemPool<ChannelFuture, Void> channelPoolMap,
Duration frequency) {
eventLoop.schedule(() -> {
log.atLevel(logLevel).log(channelPoolMap.getStats().toString());
logProgressAtInterval(logLevel, eventLoop, channelPoolMap, frequency);
}, frequency.toMillis(), TimeUnit.MILLISECONDS);
}

private ChannelFuture buildConnectionFuture(EventLoop eventLoop) {
// Start the connection attempt.
Bootstrap b = new Bootstrap();
b.group(eventLoop)
.channel(NioSocketChannel.class)
.handler(new ChannelDuplexHandler())
.option(ChannelOption.AUTO_READ, false);
var f = b.connect(backsideUri.getHost(), backsideUri.getPort());
var rval = new DefaultChannelPromise(f.channel());
f.addListener((ChannelFutureListener) connectFuture -> {
if (connectFuture.isSuccess()) {
// connection complete start to read first data
log.debug("Done setting up backend channel & it was successful (" + connectFuture.channel() + ")");
if (backsideSslContext != null) {
var pipeline = connectFuture.channel().pipeline();
SSLEngine sslEngine = backsideSslContext.newEngine(connectFuture.channel().alloc());
sslEngine.setUseClientMode(true);
var sslHandler = new SslHandler(sslEngine);
pipeline.addFirst("ssl", sslHandler);
sslHandler.handshakeFuture().addListener(handshakeFuture -> {
if (handshakeFuture.isSuccess()) {
rval.setSuccess();
} else {
rval.setFailure(handshakeFuture.cause());
}
});
} else {
rval.setSuccess();
}
} else {
rval.setFailure(connectFuture.cause());
}
});
return rval;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

@Override
public void channelInactive(ChannelHandlerContext ctx) {
log.debug("inactive channel - closing");
log.debug("inactive channel - closing (" + ctx.channel() + ")");
FrontsideHandler.closeAndFlush(writeBackChannel);
}

Expand Down
Loading

0 comments on commit edfb16a

Please sign in to comment.