From 47c6c8d2fb7b354e95f67ce03af37c561b61c23f Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Tue, 1 Dec 2020 22:03:06 +0100 Subject: [PATCH] Re #8: Working on Netty http pipeline. --- .../tkowalcz/tjahzi/http/BootstrapUtil.java | 38 ++++++ .../tjahzi/http/HttpClientFactory.java | 2 +- .../tjahzi/http/HttpClientInitializer.java | 9 +- .../tkowalcz/tjahzi/http/HttpConnection.java | 5 + .../tkowalcz/tjahzi/http/NettyHttpClient.java | 111 ++++++++++-------- .../tkowalcz/tjahzi/http/ResponseHandler.java | 29 +++++ .../pl/tkowalcz/tjahzi/ReconnectTest.java | 95 +++++++++++++++ 7 files changed, 239 insertions(+), 50 deletions(-) create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java create mode 100644 tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java new file mode 100644 index 0000000..98c83c8 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java @@ -0,0 +1,38 @@ +package pl.tkowalcz.tjahzi.http; + +import io.netty.bootstrap.Bootstrap; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.socket.nio.NioSocketChannel; + +public class BootstrapUtil { + + public static ChannelFuture initConnection( + EventLoopGroup group, + ClientConfiguration clientConfiguration + ) { + Bootstrap bootstrap = new Bootstrap(); + ResponseHandler responseHandler = new ResponseHandler(); + + ChannelFuture channelFuture = bootstrap.group(group) + .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) + .option(ChannelOption.SO_KEEPALIVE, true) + .option(ChannelOption.TCP_NODELAY, false) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfiguration.getConnectionTimeoutMillis()) + .channel(NioSocketChannel.class) + .handler( + new HttpClientInitializer( + responseHandler, + clientConfiguration.getRequestTimeoutMillis() + ) + ) + .remoteAddress( + clientConfiguration.getHost(), + clientConfiguration.getPort() + ).connect(); + + return channelFuture; + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java index bb4b57a..062639f 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java @@ -6,7 +6,7 @@ public class HttpClientFactory { public NettyHttpClient getHttpClient( ClientConfiguration clientConfiguration, - String[] additionalHeaders) { + String... additionalHeaders) { return new NettyHttpClient( clientConfiguration, additionalHeaders diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java index 4a2d2ee..c8d447a 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java @@ -10,9 +10,13 @@ import io.netty.handler.timeout.IdleStateHandler; class HttpClientInitializer extends ChannelInitializer { + private static final int BYTES_IN_MEGABYTE = 1024 * 1024; + + private final ResponseHandler responseHandler; private final int requestTimeoutMillis; - HttpClientInitializer(int requestTimeoutMillis) { + HttpClientInitializer(ResponseHandler responseHandler, int requestTimeoutMillis) { + this.responseHandler = responseHandler; this.requestTimeoutMillis = requestTimeoutMillis; } @@ -21,7 +25,8 @@ class HttpClientInitializer extends ChannelInitializer { ChannelPipeline p = ch.pipeline(); p.addLast(new IdleStateHandler(requestTimeoutMillis, 0, 60)); p.addLast(new HttpClientCodec()); - p.addLast(new HttpObjectAggregator(1048576)); + p.addLast(new HttpObjectAggregator(BYTES_IN_MEGABYTE)); p.addLast(new HttpContentDecompressor()); + p.addLast(responseHandler); } } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java new file mode 100644 index 0000000..d3ad5d2 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java @@ -0,0 +1,5 @@ +package pl.tkowalcz.tjahzi.http; + +public class HttpConnection { + +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java index 39c4ee5..da63c89 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java @@ -1,82 +1,99 @@ package pl.tkowalcz.tjahzi.http; -import io.netty.bootstrap.Bootstrap; import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; -import io.netty.channel.Channel; -import io.netty.channel.ChannelOption; +import io.netty.channel.ChannelFuture; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.compression.Snappy; import io.netty.handler.codec.http.*; import java.io.Closeable; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; public class NettyHttpClient implements Closeable { public static final String PROTOBUF_MIME_TYPE = "application/x-protobuf"; - private final EventLoopGroup group; - private final String logEndpoint; - - private final Channel channel; - private final String host; + private final ClientConfiguration clientConfiguration; private final HttpHeaders headers; private final Snappy snappy = new Snappy(); + private final EventLoopGroup group; + private volatile ChannelFuture lokiConnection; +// private HttpConnection lokiConnection; + public NettyHttpClient( ClientConfiguration clientConfiguration, String[] additionalHeaders) { - host = clientConfiguration.getHost(); - logEndpoint = clientConfiguration.getLogEndpoint(); - - ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client"); - ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker"); - group = new NioEventLoopGroup(1, threadFactory); - - Bootstrap bootstrap = new Bootstrap(); - channel = bootstrap.group(group) - .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) - .option(ChannelOption.SO_KEEPALIVE, true) - .option(ChannelOption.TCP_NODELAY, false) - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfiguration.getConnectionTimeoutMillis()) - .channel(NioSocketChannel.class) - .handler(new HttpClientInitializer(clientConfiguration.getRequestTimeoutMillis())) - .remoteAddress( - clientConfiguration.getHost(), - clientConfiguration.getPort() - ) - .connect() - .syncUninterruptibly() - .channel(); - - headers = new ReadOnlyHttpHeaders( + this.clientConfiguration = clientConfiguration; + this.headers = new ReadOnlyHttpHeaders( true, additionalHeaders ); + + ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client"); + ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker"); + NioEventLoopGroup group = new NioEventLoopGroup(1, threadFactory); + this.group = group; + + foobar(clientConfiguration); + } + + private void foobar(ClientConfiguration clientConfiguration) { + lokiConnection = BootstrapUtil.initConnection( + group, + clientConfiguration + ); + + lokiConnection.addListener(future -> { + System.out.println("NettyHttpClient.foobar"); + if (!future.isSuccess()) { + group.schedule(() -> { + foobar(clientConfiguration); + }, + 1000, TimeUnit.MILLISECONDS); + // TODO: increase error counters + + } + }); } public void log(ByteBuf dataBuffer) { - ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer(); - snappy.encode(dataBuffer, output, dataBuffer.readableBytes()); + log(dataBuffer, clientConfiguration.getMaxRetries()); + } - FullHttpRequest request = new DefaultFullHttpRequest( - HttpVersion.HTTP_1_1, - HttpMethod.POST, - logEndpoint, - output - ); + public void log(ByteBuf dataBuffer, int retries) { + lokiConnection.awaitUninterruptibly(); + if (lokiConnection.isSuccess()) { + ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer(); + snappy.encode(dataBuffer, output, dataBuffer.readableBytes()); - request.headers() - .add(headers) - .set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE) - .set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes()) - .set(HttpHeaderNames.HOST, host); + FullHttpRequest request = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, + HttpMethod.POST, + clientConfiguration.getLogEndpoint(), + output + ); - channel.writeAndFlush(request); + request.headers() + .add(headers) + .set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE) + .set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes()) + .set(HttpHeaderNames.HOST, clientConfiguration.getHost()); + + lokiConnection.channel().writeAndFlush(request); + } else { + // TODO: increase error counters + + if (retries > 0) { + log(dataBuffer, retries - 1); + } + + // Dropping message on the floor + } } @Override diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java new file mode 100644 index 0000000..5a2e355 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java @@ -0,0 +1,29 @@ +package pl.tkowalcz.tjahzi.http; + +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.http.FullHttpResponse; + +@ChannelHandler.Sharable +class ResponseHandler extends SimpleChannelInboundHandler { + + ResponseHandler() { + super(false); + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) { + System.out.println("msg = " + msg); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + System.out.println("cause = " + cause); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + System.out.println("ResponseHandler.channelInactive"); + } +} diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java new file mode 100644 index 0000000..b470934 --- /dev/null +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java @@ -0,0 +1,95 @@ +package pl.tkowalcz.tjahzi; + +import com.github.tomakehurst.wiremock.WireMockServer; +import org.awaitility.Durations; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import pl.tkowalcz.tjahzi.http.ClientConfiguration; +import pl.tkowalcz.tjahzi.http.HttpClientFactory; +import pl.tkowalcz.tjahzi.http.NettyHttpClient; + +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import static com.github.tomakehurst.wiremock.client.WireMock.*; +import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig; +import static org.awaitility.Awaitility.await; + +public class ReconnectTest { + + private WireMockServer wireMockServer; + + private TjahziInitializer initializer; + private LoggingSystem loggingSystem; + + @BeforeEach + void setUp() { + wireMockServer = new WireMockServer( + wireMockConfig() + .dynamicPort() + .dynamicHttpsPort() + ); + +// wireMockServer.start(); + + initializer = new TjahziInitializer(); + } + + @AfterEach + void tearDown() { + wireMockServer.stop(); + + if (loggingSystem != null) { + loggingSystem.close( + (int) TimeUnit.SECONDS.toMillis(10), + System.out::println + ); + + loggingSystem = null; + } + } + + @Test + void shouldIncludeAdditionalHeaders() { + // Given +// wireMockServer.stubFor( +// get(urlEqualTo("/loki/api/v1/push")) +// .willReturn( +// aResponse().withStatus(200) +// )); + + ClientConfiguration clientConfiguration = ClientConfiguration.builder() + .withConnectionTimeoutMillis(10_000) + .withHost("localhost") + .withPort(12322) + .withMaxRetries(1) + .build(); + + NettyHttpClient httpClient = HttpClientFactory.defaultFactory() + .getHttpClient(clientConfiguration); + + loggingSystem = initializer.createLoggingSystem( + httpClient, + 1024 * 1024, + false + ); + + // When + TjahziLogger logger = loggingSystem.createLogger(); + logger.log( + System.currentTimeMillis(), + Map.of(), + "Test" + ); + + // Then + await() + .atMost(Durations.TEN_MINUTES) + .untilAsserted(() -> + wireMockServer.verify( + postRequestedFor(urlMatching("/loki/api/v1/push")) + .withHeader("X-Scope-OrgID", equalTo("Circus")) + )); + } +}