From 86d2dc4745112c578915c99bf68039416574563e Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sat, 9 Jan 2021 00:18:18 +0100 Subject: [PATCH 1/8] Re #7: Cleaned up import --- .../src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java index 00ca13f..66aa827 100644 --- a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java +++ b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java @@ -1,5 +1,6 @@ package pl.tkowalcz.tjahzi.log4j2; +import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.core.Layout; import org.apache.logging.log4j.core.LogEvent; import org.apache.logging.log4j.status.StatusLogger; @@ -14,7 +15,7 @@ import java.util.function.BiConsumer; public class AppenderLogic implements BiConsumer { - protected static final org.apache.logging.log4j.Logger LOGGER = StatusLogger.getLogger(); + protected static final Logger LOGGER = StatusLogger.getLogger(); private final LoggingSystem loggingSystem; From 64626a39fae8c24a5dc3889ba50fbec81f1008c8 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sat, 9 Jan 2021 10:34:52 +0100 Subject: [PATCH 2/8] Re #7: Added component named MonitoringModule with empty implementation and one that uses AtomicLongs for counting events and Agrona's DistinctErrorLog for storing exceptions. While various counters are being updated by other components and response times are logged we do not provide way of plugging in ones own implementation or reading the counters back (yet). --- .../tjahzi/log4j2/LokiAppenderBuilder.java | 5 + .../pl/tkowalcz/tjahzi/LoggingSystem.java | 8 +- .../pl/tkowalcz/tjahzi/TjahziInitializer.java | 7 +- .../java/pl/tkowalcz/tjahzi/TjahziLogger.java | 10 +- .../tkowalcz/tjahzi/http/BlockingRetry.java | 20 ++- .../tkowalcz/tjahzi/http/BootstrapUtil.java | 10 +- .../tjahzi/http/ClientConfiguration.java | 7 + .../http/ClientConfigurationBuilder.java | 9 ++ .../tjahzi/http/EventLoopGroupRetry.java | 20 ++- .../tjahzi/http/HttpClientFactory.java | 4 + .../tjahzi/http/HttpClientInitializer.java | 34 ++++- .../tkowalcz/tjahzi/http/HttpConnection.java | 22 ++- .../tkowalcz/tjahzi/http/NettyHttpClient.java | 4 +- .../tkowalcz/tjahzi/http/ResponseHandler.java | 15 +- .../tjahzi/stats/EmptyMonitoringModule.java | 62 ++++++++ .../tjahzi/stats/MonitoringModule.java | 51 +++++++ .../stats/PipelinedHttpRequestTimer.java | 31 ++++ .../stats/StandardMonitoringModule.java | 138 ++++++++++++++++++ .../tjahzi/stats/TimingRingBuffer.java | 31 ++++ .../java/pl/tkowalcz/tjahzi/HeadersTest.java | 12 +- .../tjahzi/LoggingSystemSanityCheckTest.java | 9 +- .../pl/tkowalcz/tjahzi/LoggingSystemTest.java | 7 +- .../pl/tkowalcz/tjahzi/ReconnectTest.java | 16 +- .../tjahzi/ResourcesCleanupOnCloseTest.java | 6 + .../tjahzi/http/BlockingRetryTest.java | 10 +- .../tjahzi/http/EventLoopGroupRetryTest.java | 10 +- 26 files changed, 518 insertions(+), 40 deletions(-) create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/PipelinedHttpRequestTimer.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/TimingRingBuffer.java diff --git a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java index 641e0e4..c969a47 100644 --- a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java +++ b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java @@ -12,6 +12,7 @@ import pl.tkowalcz.tjahzi.github.GitHubDocs; import pl.tkowalcz.tjahzi.http.ClientConfiguration; import pl.tkowalcz.tjahzi.http.HttpClientFactory; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.util.HashMap; import java.util.stream.Stream; @@ -76,10 +77,13 @@ public class LokiAppenderBuilder> extends Abstr .flatMap(header -> Stream.of(header.getName(), header.getValue())) .toArray(String[]::new); + StandardMonitoringModule monitoringModule = new StandardMonitoringModule(); + NettyHttpClient httpClient = HttpClientFactory .defaultFactory() .getHttpClient( configurationBuilder, + monitoringModule, additionalHeaders ); @@ -97,6 +101,7 @@ public class LokiAppenderBuilder> extends Abstr LoggingSystem loggingSystem = new TjahziInitializer().createLoggingSystem( httpClient, + monitoringModule, lokiLabels, bufferSizeBytes, isUseOffHeapBuffer() diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LoggingSystem.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LoggingSystem.java index 91c7c54..01c349c 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LoggingSystem.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LoggingSystem.java @@ -2,6 +2,7 @@ package pl.tkowalcz.tjahzi; import org.agrona.concurrent.AgentRunner; import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; import java.io.Closeable; import java.util.function.Consumer; @@ -10,20 +11,25 @@ public class LoggingSystem { private final ManyToOneRingBuffer logBuffer; private final AgentRunner runner; + + private final MonitoringModule monitoringModule; private final Closeable[] resourcesToCleanup; public LoggingSystem( ManyToOneRingBuffer logBuffer, AgentRunner runner, + MonitoringModule monitoringModule, Closeable... resourcesToCleanup ) { this.logBuffer = logBuffer; this.runner = runner; + + this.monitoringModule = monitoringModule; this.resourcesToCleanup = resourcesToCleanup; } public TjahziLogger createLogger() { - return new TjahziLogger(logBuffer); + return new TjahziLogger(logBuffer, monitoringModule); } public void start() { diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java index a859797..a0c7a08 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java @@ -6,6 +6,7 @@ import org.agrona.concurrent.UnsafeBuffer; import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer; import org.agrona.concurrent.ringbuffer.RingBufferDescriptor; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; import java.nio.ByteBuffer; import java.util.Map; @@ -16,6 +17,7 @@ public class TjahziInitializer { public LoggingSystem createLoggingSystem( NettyHttpClient httpClient, + MonitoringModule monitoringModule, Map staticLabels, int bufferSizeBytes, boolean offHeap) { @@ -34,7 +36,7 @@ public class TjahziInitializer { AgentRunner runner = new AgentRunner( new SleepingMillisIdleStrategy(), - Throwable::printStackTrace, + monitoringModule::addAgentError, null, agent ); @@ -42,6 +44,7 @@ public class TjahziInitializer { return new LoggingSystem( logBuffer, runner, + monitoringModule, httpClient ); } @@ -58,7 +61,7 @@ public class TjahziInitializer { } long candidatePowerOfTwo = Long.highestOneBit(bufferSize) << 1; - if (candidatePowerOfTwo + RingBufferDescriptor.TRAILER_LENGTH > Integer.MAX_VALUE) { + if (candidatePowerOfTwo + RingBufferDescriptor.TRAILER_LENGTH >= Integer.MAX_VALUE) { return (int) (candidatePowerOfTwo >> 1); } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java index f7e7832..046c4d5 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java @@ -1,6 +1,7 @@ package pl.tkowalcz.tjahzi; import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; import java.nio.ByteBuffer; import java.util.Map; @@ -10,10 +11,14 @@ public class TjahziLogger { public static final int LOG_MESSAGE_TYPE_ID = 5; private final ManyToOneRingBuffer logBuffer; + private final MonitoringModule monitoringModule; + private final LogBufferSerializer serializer; - public TjahziLogger(ManyToOneRingBuffer logBuffer) { + public TjahziLogger(ManyToOneRingBuffer logBuffer, MonitoringModule monitoringModule) { this.logBuffer = logBuffer; + this.monitoringModule = monitoringModule; + this.serializer = new LogBufferSerializer(logBuffer.buffer()); } @@ -40,7 +45,7 @@ public class TjahziLogger { claim ); } else { - // TODO: increment dropped lines metric + monitoringModule.incrementDroppedPuts(); } return this; @@ -68,6 +73,7 @@ public class TjahziLogger { logBuffer.commit(claim); } catch (Throwable t) { logBuffer.abort(claim); + monitoringModule.incrementDroppedPuts(t); } } } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java index 72f2ef8..5d20bce 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java @@ -1,11 +1,14 @@ package pl.tkowalcz.tjahzi.http; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; + import java.util.function.Consumer; class BlockingRetry implements Retry { private final ExponentialBackoffStrategy strategy; private final Consumer operation; + private final MonitoringModule monitoringModule; private int retries; @@ -13,17 +16,25 @@ class BlockingRetry implements Retry { BlockingRetry( Consumer operation, ExponentialBackoffStrategy strategy, - int retries) { + int retries, + MonitoringModule monitoringModule + ) { this.operation = operation; this.strategy = strategy; + this.retries = retries; + this.monitoringModule = monitoringModule; } - BlockingRetry(Consumer operation, int retries) { + BlockingRetry( + Consumer operation, + int retries, + MonitoringModule monitoringModule) { this( operation, ExponentialBackoffStrategy.withDefault(), - retries + retries, + monitoringModule ); } @@ -31,6 +42,7 @@ class BlockingRetry implements Retry { public void retry() { if (retries > 0) { retries--; + monitoringModule.incrementRetriedHttpRequests(); try { Thread.sleep(strategy.getAsLong(), 0); @@ -39,6 +51,8 @@ class BlockingRetry implements Retry { } operation.accept(this); + } else { + monitoringModule.incrementFailedHttpRequests(); } } } 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 index c750967..dde23b7 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java @@ -6,15 +6,16 @@ import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; public class BootstrapUtil { public static ChannelFuture initConnection( EventLoopGroup group, - ClientConfiguration clientConfiguration + ClientConfiguration clientConfiguration, + MonitoringModule monitoringModule ) { Bootstrap bootstrap = new Bootstrap(); - ResponseHandler responseHandler = new ResponseHandler(); return bootstrap.group(group) .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) @@ -24,8 +25,9 @@ public class BootstrapUtil { .channel(NioSocketChannel.class) .handler( new HttpClientInitializer( - responseHandler, - clientConfiguration.getRequestTimeoutMillis() + monitoringModule, + clientConfiguration.getRequestTimeoutMillis(), + clientConfiguration.getMaxInFlightRequests() ) ) .remoteAddress( diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfiguration.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfiguration.java index ed6030e..4352a0c 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfiguration.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfiguration.java @@ -8,6 +8,7 @@ public class ClientConfiguration { private final int connectionTimeoutMillis; private final int requestTimeoutMillis; + private final int maxInFlightRequests; private final int maxRetries; @@ -17,12 +18,14 @@ public class ClientConfiguration { int port, int connectionTimeoutMillis, int requestTimeoutMillis, + int maxInFlightRequests, int maxRetries) { this.logEndpoint = logEndpoint; this.host = host; this.port = port; this.connectionTimeoutMillis = connectionTimeoutMillis; this.requestTimeoutMillis = requestTimeoutMillis; + this.maxInFlightRequests = maxInFlightRequests; this.maxRetries = maxRetries; } @@ -47,6 +50,10 @@ public class ClientConfiguration { return requestTimeoutMillis; } + public int getMaxInFlightRequests() { + return maxInFlightRequests; + } + public int getMaxRetries() { return maxRetries; } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java index b07cd52..aa5aa2c 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java @@ -7,6 +7,8 @@ public class ClientConfigurationBuilder { public static final int DEFAULT_CONNECT_TIMEOUT_MILLIS = 5_000; public static final int DEFAULT_REQUEST_TIMEOUT_MILLIS = 60_000; + public static final int DEFAULT_MAX_IN_FLIGHT_REQUESTS = 10; + public static final int DEFAULT_MAX_RETRIES = 0; private String logEndpoint = DEFAULT_LOG_ENDPOINT; @@ -15,6 +17,7 @@ public class ClientConfigurationBuilder { private int connectionTimeoutMillis = DEFAULT_CONNECT_TIMEOUT_MILLIS; private int requestTimeoutMillis = DEFAULT_REQUEST_TIMEOUT_MILLIS; + private int maxInFlightRequests = DEFAULT_MAX_IN_FLIGHT_REQUESTS; private int maxRetries = DEFAULT_MAX_RETRIES; @@ -43,6 +46,11 @@ public class ClientConfigurationBuilder { return this; } + public ClientConfigurationBuilder withMaxInFlightRequests(int maxInFlightRequests) { + this.maxInFlightRequests = maxInFlightRequests; + return this; + } + public ClientConfigurationBuilder withMaxRetries(int maxRetries) { this.maxRetries = maxRetries; return this; @@ -55,6 +63,7 @@ public class ClientConfigurationBuilder { port, connectionTimeoutMillis, requestTimeoutMillis, + maxInFlightRequests, maxRetries ); } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java index 37707d9..4a36597 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java @@ -1,5 +1,7 @@ package pl.tkowalcz.tjahzi.http; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; + import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; @@ -10,26 +12,38 @@ class EventLoopGroupRetry implements Retry, Runnable { private final Consumer operation; private final ExponentialBackoffStrategy strategy; + private final MonitoringModule monitoringModule; + // VisibleForTesting public EventLoopGroupRetry( ScheduledExecutorService group, Consumer operation, - ExponentialBackoffStrategy strategy) { + ExponentialBackoffStrategy strategy, + MonitoringModule monitoringModule + ) { this.group = group; this.operation = operation; this.strategy = strategy; + + this.monitoringModule = monitoringModule; } - public EventLoopGroupRetry(ScheduledExecutorService group, Consumer operation) { + public EventLoopGroupRetry( + ScheduledExecutorService group, + Consumer operation, + MonitoringModule monitoringModule + ) { this( group, operation, - ExponentialBackoffStrategy.withDefault() + ExponentialBackoffStrategy.withDefault(), + monitoringModule ); } @Override public void run() { + monitoringModule.incrementHttpConnectAttempts(); operation.accept(this); } 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 062639f..25d1dd7 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 @@ -1,14 +1,18 @@ package pl.tkowalcz.tjahzi.http; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; + public class HttpClientFactory { private static final HttpClientFactory INSTANCE = new HttpClientFactory(); public NettyHttpClient getHttpClient( ClientConfiguration clientConfiguration, + MonitoringModule monitoringModule, String... additionalHeaders) { return new NettyHttpClient( clientConfiguration, + monitoringModule, 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 c8d447a..f04b78e 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 @@ -7,26 +7,50 @@ import io.netty.handler.codec.http.HttpClientCodec; import io.netty.handler.codec.http.HttpContentDecompressor; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.timeout.IdleStateHandler; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; +import pl.tkowalcz.tjahzi.stats.PipelinedHttpRequestTimer; class HttpClientInitializer extends ChannelInitializer { private static final int BYTES_IN_MEGABYTE = 1024 * 1024; + private static final int MAX_CONTENT_LENGTH = 10 * BYTES_IN_MEGABYTE; + private final MonitoringModule monitoringModule; private final ResponseHandler responseHandler; - private final int requestTimeoutMillis; - HttpClientInitializer(ResponseHandler responseHandler, int requestTimeoutMillis) { - this.responseHandler = responseHandler; + private final int requestTimeoutMillis; + private final int maxNumberOfRequests; + + HttpClientInitializer( + MonitoringModule monitoringModule, + int requestTimeoutMillis, + int maxNumberOfRequests + ) { + this.monitoringModule = monitoringModule; + this.responseHandler = new ResponseHandler(monitoringModule); + this.requestTimeoutMillis = requestTimeoutMillis; + this.maxNumberOfRequests = maxNumberOfRequests; } @Override protected void initChannel(SocketChannel ch) { ChannelPipeline p = ch.pipeline(); - p.addLast(new IdleStateHandler(requestTimeoutMillis, 0, 60)); + p.addLast(new IdleStateHandler( + requestTimeoutMillis, + 0, + 60 + ) + ); p.addLast(new HttpClientCodec()); - p.addLast(new HttpObjectAggregator(BYTES_IN_MEGABYTE)); + p.addLast(new HttpObjectAggregator(MAX_CONTENT_LENGTH)); p.addLast(new HttpContentDecompressor()); p.addLast(responseHandler); + p.addLast( + new PipelinedHttpRequestTimer( + monitoringModule, + maxNumberOfRequests + ) + ); } } 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 index e6d3b2a..041154f 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java @@ -3,6 +3,7 @@ package pl.tkowalcz.tjahzi.http; import io.netty.channel.ChannelFuture; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.handler.codec.http.FullHttpRequest; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; import java.io.Closeable; import java.util.concurrent.ThreadFactory; @@ -10,26 +11,33 @@ import java.util.concurrent.ThreadFactory; public class HttpConnection implements Closeable { private final ClientConfiguration clientConfiguration; - private final NioEventLoopGroup group; + private final MonitoringModule monitoringModule; + private final NioEventLoopGroup group; private volatile ChannelFuture lokiConnection; - public HttpConnection(ClientConfiguration clientConfiguration) { + public HttpConnection(ClientConfiguration clientConfiguration, MonitoringModule monitoringModule) { this.clientConfiguration = clientConfiguration; + this.monitoringModule = monitoringModule; ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client"); ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker"); this.group = new NioEventLoopGroup(1, threadFactory); - EventLoopGroupRetry retry = new EventLoopGroupRetry(group, this::recreateConnection); + EventLoopGroupRetry retry = new EventLoopGroupRetry( + group, + this::recreateConnection, + monitoringModule + ); recreateConnection(retry); } private void recreateConnection(Retry retry) { lokiConnection = BootstrapUtil.initConnection( group, - clientConfiguration + clientConfiguration, + monitoringModule ); lokiConnection.addListener((ChannelFuture future) -> { @@ -47,7 +55,8 @@ public class HttpConnection implements Closeable { public void execute(FullHttpRequest request) { BlockingRetry blockingRetry = new BlockingRetry( retry -> execute(request, retry), - clientConfiguration.getMaxRetries() + clientConfiguration.getMaxRetries(), + monitoringModule ); execute(request, blockingRetry); @@ -59,10 +68,9 @@ public class HttpConnection implements Closeable { stableReference.awaitUninterruptibly(); if (stableReference.isSuccess() && stableReference.channel().isActive()) { stableReference.channel().writeAndFlush(request); + monitoringModule.incrementSentHttpRequests(); } else { - // TODO: increase error counters retry.retry(); - // Dropping message on the floor } } 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 27282cc..9d46870 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 @@ -4,6 +4,7 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; import io.netty.handler.codec.compression.Snappy; import io.netty.handler.codec.http.*; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; import java.io.Closeable; @@ -19,6 +20,7 @@ public class NettyHttpClient implements Closeable { public NettyHttpClient( ClientConfiguration clientConfiguration, + MonitoringModule monitoringModule, String[] additionalHeaders) { this.clientConfiguration = clientConfiguration; this.headers = new ReadOnlyHttpHeaders( @@ -26,7 +28,7 @@ public class NettyHttpClient implements Closeable { additionalHeaders ); - lokiConnection = new HttpConnection(clientConfiguration); + lokiConnection = new HttpConnection(clientConfiguration, monitoringModule); } public void log(ByteBuf dataBuffer) { 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 index 8ce408a..f87e569 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java @@ -4,25 +4,34 @@ import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.FullHttpResponse; +import io.netty.handler.codec.http.HttpResponseStatus; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; @ChannelHandler.Sharable class ResponseHandler extends SimpleChannelInboundHandler { - ResponseHandler() { + private final MonitoringModule monitoringModule; + + ResponseHandler(MonitoringModule monitoringModule) { super(false); + this.monitoringModule = monitoringModule; } @Override protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) { + monitoringModule.incrementHttpResponses(); + if (msg.status() != HttpResponseStatus.OK) { + monitoringModule.incrementHttpErrors(msg.status(), msg.content()); + } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { - System.out.println("cause = " + cause); + monitoringModule.addPipelineError(cause); } @Override public void channelInactive(ChannelHandlerContext ctx) { - System.out.println("ResponseHandler.channelInactive"); + monitoringModule.incrementChannelInactive(); } } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java new file mode 100644 index 0000000..28558e4 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java @@ -0,0 +1,62 @@ +package pl.tkowalcz.tjahzi.stats; + +import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.http.HttpResponseStatus; + +import java.time.Clock; + +public class EmptyMonitoringModule implements MonitoringModule { + + @Override + public Clock getClock() { + return Clock.systemUTC(); + } + + @Override + public void incrementDroppedPuts() { + } + + @Override + public void incrementDroppedPuts(Throwable throwable) { + } + + @Override + public void incrementSentHttpRequests() { + } + + @Override + public void incrementFailedHttpRequests() { + } + + @Override + public void incrementRetriedHttpRequests() { + } + + @Override + public void addAgentError(Throwable throwable) { + } + + @Override + public void incrementHttpConnectAttempts() { + } + + @Override + public void addPipelineError(Throwable cause) { + } + + @Override + public void incrementChannelInactive() { + } + + @Override + public void incrementHttpResponses() { + } + + @Override + public void incrementHttpErrors(HttpResponseStatus status, ByteBuf responseContent) { + } + + @Override + public void recordResponseTime(long time) { + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java new file mode 100644 index 0000000..e000d18 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java @@ -0,0 +1,51 @@ +package pl.tkowalcz.tjahzi.stats; + +import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.http.HttpResponseStatus; + +import java.time.Clock; + +public interface MonitoringModule { + + Clock getClock(); + + void incrementDroppedPuts(); + + void incrementDroppedPuts(Throwable throwable); + +// long getDroppedPuts(); + + void incrementSentHttpRequests(); + +// long getSentHttpRequests(); + + void incrementFailedHttpRequests(); + +// long getFailedHttpRequests(); + + void incrementRetriedHttpRequests(); + +// long getRetriedHttpRequests(); + + void addAgentError(Throwable throwable); + +// long getAgentErrors(); + + void incrementHttpConnectAttempts(); + +// long getHttpConnectAttempts(); + + void addPipelineError(Throwable cause); + + void incrementChannelInactive(); + +// long getChannelInactive(); + + void incrementHttpResponses(); + +// long getHttpResponses(); + + void incrementHttpErrors(HttpResponseStatus status, ByteBuf content); + + void recordResponseTime(long time); +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/PipelinedHttpRequestTimer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/PipelinedHttpRequestTimer.java new file mode 100644 index 0000000..d6660e1 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/PipelinedHttpRequestTimer.java @@ -0,0 +1,31 @@ +package pl.tkowalcz.tjahzi.stats; + +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; + +public class PipelinedHttpRequestTimer extends ChannelDuplexHandler { + + private final MonitoringModule monitoringModule; + private final TimingRingBuffer timer; + + public PipelinedHttpRequestTimer( + MonitoringModule monitoringModule, + int maxNumberOfRequests + ) { + this.monitoringModule = monitoringModule; + this.timer = new TimingRingBuffer(monitoringModule.getClock(), maxNumberOfRequests); + } + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + super.write(ctx, msg, promise); + timer.record(); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + super.channelRead(ctx, msg); + monitoringModule.recordResponseTime(timer.measure()); + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java new file mode 100644 index 0000000..2f8c482 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java @@ -0,0 +1,138 @@ +package pl.tkowalcz.tjahzi.stats; + +import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.http.HttpResponseStatus; +import org.agrona.concurrent.SystemEpochClock; +import org.agrona.concurrent.UnsafeBuffer; +import org.agrona.concurrent.errors.DistinctErrorLog; + +import java.nio.charset.Charset; +import java.time.Clock; +import java.util.concurrent.atomic.AtomicLong; + +public class StandardMonitoringModule implements MonitoringModule { + + private AtomicLong droppedPuts = new AtomicLong(); + private AtomicLong httpConnectAttempts = new AtomicLong(); + + private AtomicLong sentHttpRequests = new AtomicLong(); + private AtomicLong failedHttpRequests = new AtomicLong(); + private AtomicLong retriedHttpRequests = new AtomicLong(); + private AtomicLong httpResponses = new AtomicLong(); + private AtomicLong channelInactive = new AtomicLong(); + private AtomicLong agentErrors = new AtomicLong(); + + private final DistinctErrorLog distinctErrorLog; + + public StandardMonitoringModule() { + distinctErrorLog = new DistinctErrorLog( + new UnsafeBuffer(new byte[1024]), + new SystemEpochClock() + ); + + distinctErrorLog.record(new NullPointerException()); + } + + @Override + public Clock getClock() { + return Clock.systemUTC(); + } + + @Override + public void incrementDroppedPuts() { + droppedPuts.incrementAndGet(); + } + + public long getDroppedPuts() { + return droppedPuts.get(); + } + + @Override + public void incrementDroppedPuts(Throwable throwable) { + incrementDroppedPuts(); + distinctErrorLog.record(throwable); + } + + @Override + public void incrementSentHttpRequests() { + sentHttpRequests.incrementAndGet(); + } + + public long getSentHttpRequests() { + return sentHttpRequests.get(); + } + + @Override + public void incrementFailedHttpRequests() { + failedHttpRequests.incrementAndGet(); + } + + public long getFailedHttpRequests() { + return failedHttpRequests.get(); + } + + @Override + public void incrementRetriedHttpRequests() { + retriedHttpRequests.incrementAndGet(); + } + + public long getRetriedHttpRequests() { + return retriedHttpRequests.get(); + } + + @Override + public void addAgentError(Throwable throwable) { + agentErrors.incrementAndGet(); + distinctErrorLog.record(throwable); + } + + public long getAgentErrors() { + return agentErrors.get(); + } + + @Override + public void incrementHttpConnectAttempts() { + httpConnectAttempts.incrementAndGet(); + } + + public long getHttpConnectAttempts() { + return httpConnectAttempts.get(); + } + + @Override + public void incrementChannelInactive() { + channelInactive.incrementAndGet(); + } + + public long getChannelInactive() { + return channelInactive.get(); + } + + @Override + public void incrementHttpResponses() { + httpResponses.incrementAndGet(); + } + + public long getHttpResponses() { + return httpResponses.get(); + } + + @Override + public void addPipelineError(Throwable cause) { + distinctErrorLog.record(cause); + } + + @Override + public void incrementHttpErrors(HttpResponseStatus status, ByteBuf content) { + distinctErrorLog.record( + new RuntimeException( + content.toString(Charset.defaultCharset()) + ) + ); + } + + @Override + public void recordResponseTime(long time) { + + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/TimingRingBuffer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/TimingRingBuffer.java new file mode 100644 index 0000000..1de4dbc --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/TimingRingBuffer.java @@ -0,0 +1,31 @@ +package pl.tkowalcz.tjahzi.stats; + +import java.time.Clock; + +public class TimingRingBuffer { + + private final Clock clock; + private final long[] times; + + private int readPosition; + private int writePosition; + + public TimingRingBuffer(Clock clock, int maxNumberOfRequests) { + this.clock = clock; + times = new long[maxNumberOfRequests]; + } + + public void record() { + int index = Math.abs(writePosition % times.length); + writePosition++; + + times[index] = clock.millis(); + } + + public long measure() { + int index = Math.abs(readPosition % times.length); + readPosition++; + + return clock.millis() - times[index]; + } +} diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java index 64b877a..5b512c0 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java @@ -8,6 +8,7 @@ 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 pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.nio.ByteBuffer; import java.util.Map; @@ -16,6 +17,7 @@ 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; +import static org.mockito.Mockito.mock; class HeadersTest { @@ -23,6 +25,7 @@ class HeadersTest { private TjahziInitializer initializer; private LoggingSystem loggingSystem; + private StandardMonitoringModule monitoringModule; @BeforeEach void setUp() { @@ -35,6 +38,7 @@ class HeadersTest { wireMockServer.start(); initializer = new TjahziInitializer(); + monitoringModule = mock(StandardMonitoringModule.class); } @AfterEach @@ -70,12 +74,14 @@ class HeadersTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( clientConfiguration, + monitoringModule, "X-Scope-OrgID", "Circus", "C", "Control" ); loggingSystem = initializer.createLoggingSystem( httpClient, + monitoringModule, Map.of(), 1024 * 1024, false @@ -120,11 +126,13 @@ class HeadersTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( - clientConfiguration + clientConfiguration, + monitoringModule ); loggingSystem = initializer.createLoggingSystem( httpClient, + monitoringModule, Map.of(), 1024 * 1024, false @@ -169,6 +177,7 @@ class HeadersTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( clientConfiguration, + monitoringModule, "content-type", "text/plain", "content-length", "5232423423", "host", "remote" @@ -176,6 +185,7 @@ class HeadersTest { loggingSystem = initializer.createLoggingSystem( httpClient, + monitoringModule, Map.of(), 1024 * 1024, false diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java index 087c072..2d20601 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java @@ -16,6 +16,7 @@ import org.testcontainers.junit.jupiter.Testcontainers; import pl.tkowalcz.tjahzi.http.ClientConfiguration; import pl.tkowalcz.tjahzi.http.HttpClientFactory; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.nio.ByteBuffer; import java.util.Map; @@ -24,6 +25,7 @@ import java.util.concurrent.TimeUnit; import static io.restassured.RestAssured.given; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.CoreMatchers.hasItems; +import static org.mockito.Mockito.mock; @Testcontainers class LoggingSystemSanityCheckTest { @@ -41,6 +43,7 @@ class LoggingSystemSanityCheckTest { .withExposedPorts(3100); private LoggingSystem loggingSystem; + private final StandardMonitoringModule monitoringModule = mock(StandardMonitoringModule.class); @BeforeEach void setUp() { @@ -52,11 +55,15 @@ class LoggingSystemSanityCheckTest { .build(); NettyHttpClient httpClient = HttpClientFactory.defaultFactory() - .getHttpClient(clientConfiguration); + .getHttpClient( + clientConfiguration, + monitoringModule + ); TjahziInitializer initializer = new TjahziInitializer(); loggingSystem = initializer.createLoggingSystem( httpClient, + monitoringModule, Map.of("version", "0.43", "server", "127.0.0.1"), 1024 * 1024, false diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java index efd0e61..fa5ccfa 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java @@ -18,6 +18,7 @@ import org.testcontainers.junit.jupiter.Testcontainers; import pl.tkowalcz.tjahzi.http.ClientConfiguration; import pl.tkowalcz.tjahzi.http.HttpClientFactory; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.nio.ByteBuffer; import java.util.List; @@ -27,6 +28,7 @@ import java.util.concurrent.TimeUnit; import static io.restassured.RestAssured.given; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.CoreMatchers.hasItems; +import static org.mockito.Mockito.mock; @Testcontainers class LoggingSystemTest { @@ -43,6 +45,7 @@ class LoggingSystemTest { ) .withExposedPorts(3100); + private final StandardMonitoringModule monitoringModule = mock(StandardMonitoringModule.class); private LoggingSystem loggingSystem; @BeforeEach @@ -56,12 +59,14 @@ class LoggingSystemTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( - clientConfiguration + clientConfiguration, + monitoringModule ); TjahziInitializer initializer = new TjahziInitializer(); loggingSystem = initializer.createLoggingSystem( httpClient, + monitoringModule, Map.of("version", "0.43", "server", "127.0.0.1"), 1024 * 1024, false diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java index 40d15e2..ef1aec0 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java @@ -8,6 +8,7 @@ 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 pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.nio.ByteBuffer; import java.util.Map; @@ -16,6 +17,7 @@ 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; +import static org.mockito.Mockito.mock; public class ReconnectTest { @@ -23,6 +25,7 @@ public class ReconnectTest { private TjahziInitializer initializer; private LoggingSystem loggingSystem; + private StandardMonitoringModule monitoringModule; @BeforeEach void setUp() { @@ -38,6 +41,7 @@ public class ReconnectTest { )); initializer = new TjahziInitializer(); + monitoringModule = mock(StandardMonitoringModule.class); } @AfterEach @@ -65,10 +69,14 @@ public class ReconnectTest { .build(); NettyHttpClient httpClient = HttpClientFactory.defaultFactory() - .getHttpClient(clientConfiguration); + .getHttpClient( + clientConfiguration, + monitoringModule + ); loggingSystem = initializer.createLoggingSystem( httpClient, + monitoringModule, Map.of(), 1024 * 1024, false @@ -108,12 +116,16 @@ public class ReconnectTest { .build(); NettyHttpClient httpClient = HttpClientFactory.defaultFactory() - .getHttpClient(clientConfiguration); + .getHttpClient( + clientConfiguration, + monitoringModule + ); wireMockServer.start(); loggingSystem = initializer.createLoggingSystem( httpClient, + monitoringModule, Map.of(), 1024 * 1024, false diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java index a14608c..1a2afb2 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java @@ -9,6 +9,7 @@ import org.testcontainers.junit.jupiter.Testcontainers; import pl.tkowalcz.tjahzi.http.ClientConfiguration; import pl.tkowalcz.tjahzi.http.HttpClientFactory; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.lang.management.ManagementFactory; import java.lang.management.ThreadInfo; @@ -18,12 +19,14 @@ import java.util.concurrent.TimeUnit; import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; @Testcontainers class ResourcesCleanupOnCloseTest { private WireMockServer wireMockServer; private TjahziInitializer initializer; + private StandardMonitoringModule monitoringModule; @BeforeEach void setUp() { @@ -35,6 +38,7 @@ class ResourcesCleanupOnCloseTest { wireMockServer.start(); initializer = new TjahziInitializer(); + monitoringModule = mock(StandardMonitoringModule.class); } @AfterEach @@ -55,12 +59,14 @@ class ResourcesCleanupOnCloseTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( clientConfiguration, + monitoringModule, "X-Scope-OrgID", "Circus", "C", "Control" ); LoggingSystem loggingSystem = initializer.createLoggingSystem( httpClient, + monitoringModule, Map.of(), 1024 * 1024, false diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java index 7d497e9..0347d80 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java @@ -1,13 +1,17 @@ package pl.tkowalcz.tjahzi.http; import org.junit.jupiter.api.Test; +import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; class BlockingRetryTest { + private StandardMonitoringModule monitoringModule = mock(StandardMonitoringModule.class); + @Test void shouldRetryAndInvokeOperation() { // Given @@ -19,7 +23,8 @@ class BlockingRetryTest { __.retry(); }, new ExponentialBackoffStrategy(1, 1, 1), - 5 + 5, + monitoringModule ); // When @@ -39,7 +44,8 @@ class BlockingRetryTest { retryCounter.incrementAndGet(); }, new ExponentialBackoffStrategy(1, 1, 1), - 5 + 5, + monitoringModule ); // When diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java index 1f72760..68b5b66 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java @@ -5,15 +5,19 @@ 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.stats.StandardMonitoringModule; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; +import static org.mockito.Mockito.mock; class EventLoopGroupRetryTest { + private final StandardMonitoringModule monitoringModule = mock(StandardMonitoringModule.class); + private NioEventLoopGroup eventLoopGroup; @BeforeEach @@ -37,7 +41,8 @@ class EventLoopGroupRetryTest { retryCounter.incrementAndGet(); __.retry(); }, - new ExponentialBackoffStrategy(250, 1000, 2) + new ExponentialBackoffStrategy(250, 1000, 2), + monitoringModule ); // When @@ -59,7 +64,8 @@ class EventLoopGroupRetryTest { __ -> { retryCounter.incrementAndGet(); }, - new ExponentialBackoffStrategy(1, 1, 1) + new ExponentialBackoffStrategy(1, 1, 1), + monitoringModule ); // When From a6e63cb6b9f164c5c3c85753c1e6933beaaae568 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sat, 16 Jan 2021 19:16:56 +0100 Subject: [PATCH 3/8] Re #7: Added configuration option 'maxRequestsInFlight' that is used to size monitoring array. Added Netty pipeline stage for tracking number of requests in flight. Added monitoring of number of bytes of logs sent. Added thread that dumps stats to stdout. --- .../tjahzi/log4j2/LokiAppenderBuilder.java | 12 ++++ .../tkowalcz/tjahzi/http/BootstrapUtil.java | 2 +- .../tjahzi/http/ClientConfiguration.java | 10 ++-- .../http/ClientConfigurationBuilder.java | 10 ++-- .../tjahzi/http/HttpClientInitializer.java | 10 ++-- .../tkowalcz/tjahzi/http/HttpConnection.java | 2 +- .../tjahzi/http/InFlightRequestsTracker.java | 24 ++++++++ .../PipelinedHttpRequestTimer.java | 8 ++- .../tkowalcz/tjahzi/http/ResponseHandler.java | 4 +- .../tjahzi/stats/EmptyMonitoringModule.java | 2 +- .../tjahzi/stats/MonitoringModule.java | 18 +----- .../stats/StandardMonitoringModule.java | 56 +++++++++++++++---- .../tjahzi/stats/StatsDumpingThread.java | 37 ++++++++++++ .../tjahzi/stats/TimingRingBuffer.java | 4 +- 14 files changed, 146 insertions(+), 53 deletions(-) create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/InFlightRequestsTracker.java rename tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/{stats => http}/PipelinedHttpRequestTimer.java (82%) create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StatsDumpingThread.java diff --git a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java index c969a47..51f0876 100644 --- a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java +++ b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java @@ -57,6 +57,9 @@ public class LokiAppenderBuilder> extends Abstr @PluginBuilderAttribute private String logLevelLabel; + @PluginBuilderAttribute + private int maxRequestsInFlight; + @PluginElement("Headers") private Header[] headers; @@ -71,6 +74,7 @@ public class LokiAppenderBuilder> extends Abstr .withPort(port) .withMaxRetries(maxRetries) .withRequestTimeoutMillis(readTimeoutMillis) + .withMaxRequestsInFlight(maxRequestsInFlight) .build(); String[] additionalHeaders = stream(headers) @@ -189,6 +193,14 @@ public class LokiAppenderBuilder> extends Abstr this.logLevelLabel = logLevelLabel; } + public void setMaxRequestsInFlight(int maxRequestsInFlight) { + this.maxRequestsInFlight = maxRequestsInFlight; + } + + public int getMaxRequestsInFlight() { + return maxRequestsInFlight; + } + public void setHeaders(Header[] headers) { this.headers = headers; } 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 index dde23b7..369257e 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java @@ -27,7 +27,7 @@ public class BootstrapUtil { new HttpClientInitializer( monitoringModule, clientConfiguration.getRequestTimeoutMillis(), - clientConfiguration.getMaxInFlightRequests() + clientConfiguration.getMaxRequestsInFlight() ) ) .remoteAddress( diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfiguration.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfiguration.java index 4352a0c..6243d72 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfiguration.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfiguration.java @@ -8,7 +8,7 @@ public class ClientConfiguration { private final int connectionTimeoutMillis; private final int requestTimeoutMillis; - private final int maxInFlightRequests; + private final int maxRequestsInFlight; private final int maxRetries; @@ -18,14 +18,14 @@ public class ClientConfiguration { int port, int connectionTimeoutMillis, int requestTimeoutMillis, - int maxInFlightRequests, + int maxRequestsInFlight, int maxRetries) { this.logEndpoint = logEndpoint; this.host = host; this.port = port; this.connectionTimeoutMillis = connectionTimeoutMillis; this.requestTimeoutMillis = requestTimeoutMillis; - this.maxInFlightRequests = maxInFlightRequests; + this.maxRequestsInFlight = maxRequestsInFlight; this.maxRetries = maxRetries; } @@ -50,8 +50,8 @@ public class ClientConfiguration { return requestTimeoutMillis; } - public int getMaxInFlightRequests() { - return maxInFlightRequests; + public int getMaxRequestsInFlight() { + return maxRequestsInFlight; } public int getMaxRetries() { diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java index aa5aa2c..38ac480 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java @@ -7,7 +7,7 @@ public class ClientConfigurationBuilder { public static final int DEFAULT_CONNECT_TIMEOUT_MILLIS = 5_000; public static final int DEFAULT_REQUEST_TIMEOUT_MILLIS = 60_000; - public static final int DEFAULT_MAX_IN_FLIGHT_REQUESTS = 10; + public static final int DEFAULT_MAX_REQUESTS_IN_FLIGHT = 10; public static final int DEFAULT_MAX_RETRIES = 0; @@ -17,7 +17,7 @@ public class ClientConfigurationBuilder { private int connectionTimeoutMillis = DEFAULT_CONNECT_TIMEOUT_MILLIS; private int requestTimeoutMillis = DEFAULT_REQUEST_TIMEOUT_MILLIS; - private int maxInFlightRequests = DEFAULT_MAX_IN_FLIGHT_REQUESTS; + private int maxRequestsInFlight = DEFAULT_MAX_REQUESTS_IN_FLIGHT; private int maxRetries = DEFAULT_MAX_RETRIES; @@ -46,8 +46,8 @@ public class ClientConfigurationBuilder { return this; } - public ClientConfigurationBuilder withMaxInFlightRequests(int maxInFlightRequests) { - this.maxInFlightRequests = maxInFlightRequests; + public ClientConfigurationBuilder withMaxRequestsInFlight(int maxRequestsInFlight) { + this.maxRequestsInFlight = maxRequestsInFlight; return this; } @@ -63,7 +63,7 @@ public class ClientConfigurationBuilder { port, connectionTimeoutMillis, requestTimeoutMillis, - maxInFlightRequests, + maxRequestsInFlight, maxRetries ); } 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 f04b78e..a3d87f4 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 @@ -8,7 +8,6 @@ import io.netty.handler.codec.http.HttpContentDecompressor; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.timeout.IdleStateHandler; import pl.tkowalcz.tjahzi.stats.MonitoringModule; -import pl.tkowalcz.tjahzi.stats.PipelinedHttpRequestTimer; class HttpClientInitializer extends ChannelInitializer { @@ -19,18 +18,18 @@ class HttpClientInitializer extends ChannelInitializer { private final ResponseHandler responseHandler; private final int requestTimeoutMillis; - private final int maxNumberOfRequests; + private final int maxRequestsInFlight; HttpClientInitializer( MonitoringModule monitoringModule, int requestTimeoutMillis, - int maxNumberOfRequests + int maxRequestsInFlight ) { this.monitoringModule = monitoringModule; this.responseHandler = new ResponseHandler(monitoringModule); this.requestTimeoutMillis = requestTimeoutMillis; - this.maxNumberOfRequests = maxNumberOfRequests; + this.maxRequestsInFlight = maxRequestsInFlight; } @Override @@ -49,8 +48,9 @@ class HttpClientInitializer extends ChannelInitializer { p.addLast( new PipelinedHttpRequestTimer( monitoringModule, - maxNumberOfRequests + maxRequestsInFlight ) ); + p.addLast(new InFlightRequestsTracker()); } } 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 index 041154f..76bfdd5 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java @@ -68,7 +68,7 @@ public class HttpConnection implements Closeable { stableReference.awaitUninterruptibly(); if (stableReference.isSuccess() && stableReference.channel().isActive()) { stableReference.channel().writeAndFlush(request); - monitoringModule.incrementSentHttpRequests(); + monitoringModule.incrementSentHttpRequests(request.content().readableBytes()); } else { retry.retry(); } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/InFlightRequestsTracker.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/InFlightRequestsTracker.java new file mode 100644 index 0000000..c75db9b --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/InFlightRequestsTracker.java @@ -0,0 +1,24 @@ +package pl.tkowalcz.tjahzi.http; + +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; + +import java.util.concurrent.atomic.AtomicInteger; + +public class InFlightRequestsTracker extends ChannelDuplexHandler { + + private final AtomicInteger requestsInFlight = new AtomicInteger(); + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + super.write(ctx, msg, promise); + requestsInFlight.incrementAndGet(); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + super.channelRead(ctx, msg); + requestsInFlight.decrementAndGet(); + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/PipelinedHttpRequestTimer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java similarity index 82% rename from tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/PipelinedHttpRequestTimer.java rename to tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java index d6660e1..6945c70 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/PipelinedHttpRequestTimer.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java @@ -1,8 +1,10 @@ -package pl.tkowalcz.tjahzi.stats; +package pl.tkowalcz.tjahzi.http; import io.netty.channel.ChannelDuplexHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; +import pl.tkowalcz.tjahzi.stats.TimingRingBuffer; public class PipelinedHttpRequestTimer extends ChannelDuplexHandler { @@ -11,10 +13,10 @@ public class PipelinedHttpRequestTimer extends ChannelDuplexHandler { public PipelinedHttpRequestTimer( MonitoringModule monitoringModule, - int maxNumberOfRequests + int maxRequestsInFlight ) { this.monitoringModule = monitoringModule; - this.timer = new TimingRingBuffer(monitoringModule.getClock(), maxNumberOfRequests); + this.timer = new TimingRingBuffer(monitoringModule.getClock(), maxRequestsInFlight); } @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 index f87e569..c6efc4c 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java @@ -4,7 +4,7 @@ import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.FullHttpResponse; -import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpStatusClass; import pl.tkowalcz.tjahzi.stats.MonitoringModule; @ChannelHandler.Sharable @@ -20,7 +20,7 @@ class ResponseHandler extends SimpleChannelInboundHandler { @Override protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) { monitoringModule.incrementHttpResponses(); - if (msg.status() != HttpResponseStatus.OK) { + if (msg.status().codeClass() != HttpStatusClass.SUCCESS) { monitoringModule.incrementHttpErrors(msg.status(), msg.content()); } } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java index 28558e4..ab2e027 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java @@ -21,7 +21,7 @@ public class EmptyMonitoringModule implements MonitoringModule { } @Override - public void incrementSentHttpRequests() { + public void incrementSentHttpRequests(int sizeBytes) { } @Override diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java index e000d18..103d6df 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java @@ -13,38 +13,22 @@ public interface MonitoringModule { void incrementDroppedPuts(Throwable throwable); -// long getDroppedPuts(); - - void incrementSentHttpRequests(); - -// long getSentHttpRequests(); + void incrementSentHttpRequests(int sizeBytes); void incrementFailedHttpRequests(); -// long getFailedHttpRequests(); - void incrementRetriedHttpRequests(); -// long getRetriedHttpRequests(); - void addAgentError(Throwable throwable); -// long getAgentErrors(); - void incrementHttpConnectAttempts(); -// long getHttpConnectAttempts(); - void addPipelineError(Throwable cause); void incrementChannelInactive(); -// long getChannelInactive(); - void incrementHttpResponses(); -// long getHttpResponses(); - void incrementHttpErrors(HttpResponseStatus status, ByteBuf content); void recordResponseTime(long time); diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java index 2f8c482..dc91883 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java @@ -1,5 +1,6 @@ package pl.tkowalcz.tjahzi.stats; +import com.google.common.util.concurrent.Uninterruptibles; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.http.HttpResponseStatus; import org.agrona.concurrent.SystemEpochClock; @@ -8,25 +9,35 @@ import org.agrona.concurrent.errors.DistinctErrorLog; import java.nio.charset.Charset; import java.time.Clock; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; -public class StandardMonitoringModule implements MonitoringModule { +public class StandardMonitoringModule implements MonitoringModule, Runnable { - private AtomicLong droppedPuts = new AtomicLong(); - private AtomicLong httpConnectAttempts = new AtomicLong(); + private static final int ERROR_LOG_CAPACITY = 1024; - private AtomicLong sentHttpRequests = new AtomicLong(); - private AtomicLong failedHttpRequests = new AtomicLong(); - private AtomicLong retriedHttpRequests = new AtomicLong(); - private AtomicLong httpResponses = new AtomicLong(); - private AtomicLong channelInactive = new AtomicLong(); - private AtomicLong agentErrors = new AtomicLong(); + private final AtomicLong droppedPuts = new AtomicLong(); + private final AtomicLong httpConnectAttempts = new AtomicLong(); + + private final AtomicLong sentHttpRequests = new AtomicLong(); + private final AtomicLong sentBytes = new AtomicLong(); + + private final AtomicLong failedHttpRequests = new AtomicLong(); + private final AtomicLong retriedHttpRequests = new AtomicLong(); + private final AtomicLong httpResponses = new AtomicLong(); + private final AtomicLong channelInactive = new AtomicLong(); + private final AtomicLong agentErrors = new AtomicLong(); private final DistinctErrorLog distinctErrorLog; public StandardMonitoringModule() { + StatsDumpingThread thread = new StatsDumpingThread(this); + if (thread.isEnabled()) { + thread.start(); + } + distinctErrorLog = new DistinctErrorLog( - new UnsafeBuffer(new byte[1024]), + new UnsafeBuffer(new byte[ERROR_LOG_CAPACITY]), new SystemEpochClock() ); @@ -54,8 +65,9 @@ public class StandardMonitoringModule implements MonitoringModule { } @Override - public void incrementSentHttpRequests() { + public void incrementSentHttpRequests(int sizeBytes) { sentHttpRequests.incrementAndGet(); + sentBytes.addAndGet(sizeBytes); } public long getSentHttpRequests() { @@ -133,6 +145,28 @@ public class StandardMonitoringModule implements MonitoringModule { @Override public void recordResponseTime(long time) { + } + @Override + public void run() { + while (true) { + System.out.println(toString()); + Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS); + } + } + + @Override + public String toString() { + return "StandardMonitoringModule{" + + "droppedPuts=" + droppedPuts + + ", httpConnectAttempts=" + httpConnectAttempts + + ", sentHttpRequests=" + sentHttpRequests + + ", sentKilobytes=" + (sentBytes.longValue() / 1024) + + ", failedHttpRequests=" + failedHttpRequests + + ", retriedHttpRequests=" + retriedHttpRequests + + ", httpResponses=" + httpResponses + + ", channelInactive=" + channelInactive + + ", agentErrors=" + agentErrors + + '}'; } } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StatsDumpingThread.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StatsDumpingThread.java new file mode 100644 index 0000000..f4b9275 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StatsDumpingThread.java @@ -0,0 +1,37 @@ +package pl.tkowalcz.tjahzi.stats; + +import com.google.common.util.concurrent.Uninterruptibles; + +import java.util.concurrent.TimeUnit; + +public class StatsDumpingThread extends Thread { + + private static final boolean LAUNCH_STATS_DUMP_THREAD = Boolean.getBoolean("tjahzi.monitoring.dumpStats.enabled"); + + private static final int DEFAULT_LAUNCH_DUMP_INTERVAL_SECONDS = 60; + private static final int LAUNCH_DUMP_INTERVAL_SECONDS = Integer.getInteger( + "tjahzi.monitoring.dumpStats.intervalSeconds", + DEFAULT_LAUNCH_DUMP_INTERVAL_SECONDS + ); + + private final MonitoringModule monitoringModule; + + public StatsDumpingThread(MonitoringModule monitoringModule) { + super("tjahzi.monitoring.dumpStats.thread"); + setDaemon(true); + + this.monitoringModule = monitoringModule; + } + + public boolean isEnabled() { + return LAUNCH_STATS_DUMP_THREAD; + } + + @Override + public void run() { + while (isEnabled()) { + System.out.println(monitoringModule.toString()); + Uninterruptibles.sleepUninterruptibly(LAUNCH_DUMP_INTERVAL_SECONDS, TimeUnit.SECONDS); + } + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/TimingRingBuffer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/TimingRingBuffer.java index 1de4dbc..0ddc919 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/TimingRingBuffer.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/TimingRingBuffer.java @@ -10,9 +10,9 @@ public class TimingRingBuffer { private int readPosition; private int writePosition; - public TimingRingBuffer(Clock clock, int maxNumberOfRequests) { + public TimingRingBuffer(Clock clock, int maxRequestsInFlight) { this.clock = clock; - times = new long[maxNumberOfRequests]; + times = new long[maxRequestsInFlight]; } public void record() { From 6b0ec584971aad7dfee167018b0624eb2aca8516 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sat, 16 Jan 2021 20:47:18 +0100 Subject: [PATCH 4/8] Re #7: Added default value to maxRequestsInFlight, and check in the builder if it is > 0 (or else exception is thrown). --- .../java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java | 2 +- .../pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java index 51f0876..f58ff39 100644 --- a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java +++ b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java @@ -58,7 +58,7 @@ public class LokiAppenderBuilder> extends Abstr private String logLevelLabel; @PluginBuilderAttribute - private int maxRequestsInFlight; + private int maxRequestsInFlight = 100; @PluginElement("Headers") private Header[] headers; diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java index 38ac480..d0ebc19 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ClientConfigurationBuilder.java @@ -7,7 +7,7 @@ public class ClientConfigurationBuilder { public static final int DEFAULT_CONNECT_TIMEOUT_MILLIS = 5_000; public static final int DEFAULT_REQUEST_TIMEOUT_MILLIS = 60_000; - public static final int DEFAULT_MAX_REQUESTS_IN_FLIGHT = 10; + public static final int DEFAULT_MAX_REQUESTS_IN_FLIGHT = 100; public static final int DEFAULT_MAX_RETRIES = 0; @@ -57,6 +57,10 @@ public class ClientConfigurationBuilder { } public ClientConfiguration build() { + if (maxRequestsInFlight <= 0) { + throw new IllegalArgumentException("Property maxRequestsInFlight must be greater than 0"); + } + return new ClientConfiguration( logEndpoint, host, From 90fcdfd5729b86180639e08bd29e85006cad83ea Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sat, 16 Jan 2021 20:47:44 +0100 Subject: [PATCH 5/8] Re #7: Cleaned up after moving debug thread logic to separate class. --- .../stats/StandardMonitoringModule.java | 44 +------------------ 1 file changed, 1 insertion(+), 43 deletions(-) diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java index dc91883..8839af6 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java @@ -1,6 +1,5 @@ package pl.tkowalcz.tjahzi.stats; -import com.google.common.util.concurrent.Uninterruptibles; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.http.HttpResponseStatus; import org.agrona.concurrent.SystemEpochClock; @@ -9,10 +8,9 @@ import org.agrona.concurrent.errors.DistinctErrorLog; import java.nio.charset.Charset; import java.time.Clock; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; -public class StandardMonitoringModule implements MonitoringModule, Runnable { +public class StandardMonitoringModule implements MonitoringModule { private static final int ERROR_LOG_CAPACITY = 1024; @@ -54,10 +52,6 @@ public class StandardMonitoringModule implements MonitoringModule, Runnable { droppedPuts.incrementAndGet(); } - public long getDroppedPuts() { - return droppedPuts.get(); - } - @Override public void incrementDroppedPuts(Throwable throwable) { incrementDroppedPuts(); @@ -70,65 +64,37 @@ public class StandardMonitoringModule implements MonitoringModule, Runnable { sentBytes.addAndGet(sizeBytes); } - public long getSentHttpRequests() { - return sentHttpRequests.get(); - } - @Override public void incrementFailedHttpRequests() { failedHttpRequests.incrementAndGet(); } - public long getFailedHttpRequests() { - return failedHttpRequests.get(); - } - @Override public void incrementRetriedHttpRequests() { retriedHttpRequests.incrementAndGet(); } - public long getRetriedHttpRequests() { - return retriedHttpRequests.get(); - } - @Override public void addAgentError(Throwable throwable) { agentErrors.incrementAndGet(); distinctErrorLog.record(throwable); } - public long getAgentErrors() { - return agentErrors.get(); - } - @Override public void incrementHttpConnectAttempts() { httpConnectAttempts.incrementAndGet(); } - public long getHttpConnectAttempts() { - return httpConnectAttempts.get(); - } - @Override public void incrementChannelInactive() { channelInactive.incrementAndGet(); } - public long getChannelInactive() { - return channelInactive.get(); - } - @Override public void incrementHttpResponses() { httpResponses.incrementAndGet(); } - public long getHttpResponses() { - return httpResponses.get(); - } - @Override public void addPipelineError(Throwable cause) { distinctErrorLog.record(cause); @@ -147,14 +113,6 @@ public class StandardMonitoringModule implements MonitoringModule, Runnable { public void recordResponseTime(long time) { } - @Override - public void run() { - while (true) { - System.out.println(toString()); - Uninterruptibles.sleepUninterruptibly(10, TimeUnit.SECONDS); - } - } - @Override public String toString() { return "StandardMonitoringModule{" + From 8f1c9791d1a0cd99c7d4b62ca136786c8c2de56b Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Wed, 20 Jan 2021 22:55:53 +0100 Subject: [PATCH 6/8] Moved metric calculating pipeline stages before request handler otherwise they would not run. --- .../java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 a3d87f4..6d33a99 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 @@ -44,7 +44,6 @@ class HttpClientInitializer extends ChannelInitializer { p.addLast(new HttpClientCodec()); p.addLast(new HttpObjectAggregator(MAX_CONTENT_LENGTH)); p.addLast(new HttpContentDecompressor()); - p.addLast(responseHandler); p.addLast( new PipelinedHttpRequestTimer( monitoringModule, @@ -52,5 +51,6 @@ class HttpClientInitializer extends ChannelInitializer { ) ); p.addLast(new InFlightRequestsTracker()); + p.addLast(responseHandler); } } From 6d679896a11d6892258f5ae4059488690f3c1b9c Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Wed, 20 Jan 2021 22:58:04 +0100 Subject: [PATCH 7/8] Added way to inject custom monitoring modules and provided default implementations as well as Dropwizard based implementation. Dropwizard dependency is marked as "provided" so it does not impact resulting artifact and can only be used if target application also includes dropwizard library. --- .../tkowalcz/tjahzi/log4j2/LokiAppender.java | 14 +- .../tjahzi/log4j2/LokiAppenderBuilder.java | 11 +- tjahzi-core/pom.xml | 6 + .../stats/DropwizardMonitoringModule.java | 127 ++++++++++++++++++ .../tjahzi/stats/EmptyMonitoringModule.java | 7 - .../tjahzi/stats/MonitoringModule.java | 4 +- .../stats/MutableMonitoringModuleWrapper.java | 73 ++++++++++ .../stats/StandardMonitoringModule.java | 6 - 8 files changed, 229 insertions(+), 19 deletions(-) create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/DropwizardMonitoringModule.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MutableMonitoringModuleWrapper.java diff --git a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java index c8058e6..625e136 100644 --- a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java +++ b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java @@ -10,6 +10,8 @@ import org.apache.logging.log4j.core.config.Property; import org.apache.logging.log4j.core.config.plugins.Plugin; import org.apache.logging.log4j.core.config.plugins.PluginBuilderFactory; import pl.tkowalcz.tjahzi.LoggingSystem; +import pl.tkowalcz.tjahzi.stats.MonitoringModule; +import pl.tkowalcz.tjahzi.stats.MutableMonitoringModuleWrapper; import java.io.Serializable; import java.util.Objects; @@ -29,7 +31,10 @@ public class LokiAppender extends AbstractAppender { return new LokiAppenderBuilder().asBuilder(); } + private final LoggingSystem loggingSystem; + private final MutableMonitoringModuleWrapper monitoringModuleWrapper; + private final AppenderLogic appenderLogic; LokiAppender( @@ -39,7 +44,9 @@ public class LokiAppender extends AbstractAppender { boolean ignoreExceptions, Property[] properties, String logLevelLabel, - LoggingSystem loggingSystem) { + LoggingSystem loggingSystem, + MutableMonitoringModuleWrapper monitoringModuleWrapper + ) { super( name, filter, @@ -47,6 +54,7 @@ public class LokiAppender extends AbstractAppender { ignoreExceptions, properties ); + this.monitoringModuleWrapper = monitoringModuleWrapper; Objects.requireNonNull(layout, "layout"); this.loggingSystem = loggingSystem; @@ -60,6 +68,10 @@ public class LokiAppender extends AbstractAppender { return loggingSystem; } + public void setMonitoringModule(MonitoringModule monitoringModule) { + monitoringModuleWrapper.setMonitoringModule(monitoringModule); + } + @Override public void start() { loggingSystem.start(); diff --git a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java index f58ff39..7b3d38c 100644 --- a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java +++ b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java @@ -12,6 +12,7 @@ import pl.tkowalcz.tjahzi.github.GitHubDocs; import pl.tkowalcz.tjahzi.http.ClientConfiguration; import pl.tkowalcz.tjahzi.http.HttpClientFactory; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import pl.tkowalcz.tjahzi.stats.MutableMonitoringModuleWrapper; import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.util.HashMap; @@ -81,13 +82,14 @@ public class LokiAppenderBuilder> extends Abstr .flatMap(header -> Stream.of(header.getName(), header.getValue())) .toArray(String[]::new); - StandardMonitoringModule monitoringModule = new StandardMonitoringModule(); + MutableMonitoringModuleWrapper monitoringModuleWrapper = new MutableMonitoringModuleWrapper(); + monitoringModuleWrapper.setMonitoringModule(new StandardMonitoringModule()); NettyHttpClient httpClient = HttpClientFactory .defaultFactory() .getHttpClient( configurationBuilder, - monitoringModule, + monitoringModuleWrapper, additionalHeaders ); @@ -105,7 +107,7 @@ public class LokiAppenderBuilder> extends Abstr LoggingSystem loggingSystem = new TjahziInitializer().createLoggingSystem( httpClient, - monitoringModule, + monitoringModuleWrapper, lokiLabels, bufferSizeBytes, isUseOffHeapBuffer() @@ -118,7 +120,8 @@ public class LokiAppenderBuilder> extends Abstr isIgnoreExceptions(), getPropertyArray(), logLevelLabel, - loggingSystem + loggingSystem, + monitoringModuleWrapper ); } diff --git a/tjahzi-core/pom.xml b/tjahzi-core/pom.xml index cb8bc4f..9da063c 100644 --- a/tjahzi-core/pom.xml +++ b/tjahzi-core/pom.xml @@ -75,6 +75,12 @@ test + + io.dropwizard + dropwizard-core + 2.0.18 + provided + diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/DropwizardMonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/DropwizardMonitoringModule.java new file mode 100644 index 0000000..00fdfd3 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/DropwizardMonitoringModule.java @@ -0,0 +1,127 @@ +package pl.tkowalcz.tjahzi.stats; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Meter; +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.Timer; +import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.http.HttpResponseStatus; + +import java.time.Clock; +import java.util.concurrent.TimeUnit; + +public class DropwizardMonitoringModule implements MonitoringModule { + + private final Counter droppedPuts; + private final Counter httpConnectAttempts; + + private final Counter sentHttpRequests; + private final Meter sentBytes; + + private final Counter failedHttpRequests; + private final Counter retriedHttpRequests; + private final Counter httpResponses; + private final Timer responseTimes; + + private final Counter channelInactive; + private final Counter agentErrors; + private final Counter responseErrors; + + public DropwizardMonitoringModule(MetricRegistry metricRegistry, String prefix) { + droppedPuts = metricRegistry.counter(prefix + ".droppedPuts"); + httpConnectAttempts = metricRegistry.counter(prefix + ".httpConnectAttempts"); + + sentHttpRequests = metricRegistry.counter(prefix + ".sentHttpRequests"); + sentBytes = metricRegistry.meter(prefix + ".sentBytes"); + + failedHttpRequests = metricRegistry.counter(prefix + ".failedHttpRequests"); + retriedHttpRequests = metricRegistry.counter(prefix + ".retriedHttpRequests"); + httpResponses = metricRegistry.counter(prefix + ".httpResponses"); + responseTimes = metricRegistry.timer(prefix + ".responseTimes"); + + channelInactive = metricRegistry.counter(prefix + ".channelInactive"); + agentErrors = metricRegistry.counter(prefix + ".agentErrors"); + responseErrors = metricRegistry.counter(prefix + ".responseErrors"); + } + + @Override + public Clock getClock() { + return Clock.systemUTC(); + } + + @Override + public void incrementDroppedPuts() { + droppedPuts.inc(); + } + + @Override + public void incrementDroppedPuts(Throwable throwable) { + incrementDroppedPuts(); + } + + @Override + public void incrementSentHttpRequests(int sizeBytes) { + sentHttpRequests.inc(); + sentBytes.mark(sizeBytes); + } + + @Override + public void incrementFailedHttpRequests() { + failedHttpRequests.inc(); + } + + @Override + public void incrementRetriedHttpRequests() { + retriedHttpRequests.inc(); + } + + @Override + public void addAgentError(Throwable throwable) { + agentErrors.inc(); + } + + @Override + public void incrementHttpConnectAttempts() { + httpConnectAttempts.inc(); + } + + @Override + public void incrementChannelInactive() { + channelInactive.inc(); + } + + @Override + public void incrementHttpResponses() { + httpResponses.inc(); + } + + @Override + public void addPipelineError(Throwable cause) { + responseErrors.inc(); + } + + @Override + public void incrementHttpErrors(HttpResponseStatus status, ByteBuf content) { + responseErrors.inc(); + } + + @Override + public void recordResponseTime(long time) { + responseTimes.update(time, TimeUnit.MILLISECONDS); + } + + @Override + public String toString() { + return "StandardMonitoringModule{" + + "droppedPuts=" + droppedPuts.getCount() + + ", httpConnectAttempts=" + httpConnectAttempts + + ", sentHttpRequests=" + sentHttpRequests + + ", sentKilobytes=" + (sentBytes.getCount() / 1024) + + ", failedHttpRequests=" + failedHttpRequests + + ", retriedHttpRequests=" + retriedHttpRequests + + ", httpResponses=" + httpResponses + + ", channelInactive=" + channelInactive + + ", agentErrors=" + agentErrors + + '}'; + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java index ab2e027..2fca4c8 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java @@ -3,15 +3,8 @@ package pl.tkowalcz.tjahzi.stats; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.http.HttpResponseStatus; -import java.time.Clock; - public class EmptyMonitoringModule implements MonitoringModule { - @Override - public Clock getClock() { - return Clock.systemUTC(); - } - @Override public void incrementDroppedPuts() { } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java index 103d6df..3c1e972 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java @@ -7,7 +7,9 @@ import java.time.Clock; public interface MonitoringModule { - Clock getClock(); + default Clock getClock() { + return Clock.systemUTC(); + } void incrementDroppedPuts(); diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MutableMonitoringModuleWrapper.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MutableMonitoringModuleWrapper.java new file mode 100644 index 0000000..00ba2e9 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MutableMonitoringModuleWrapper.java @@ -0,0 +1,73 @@ +package pl.tkowalcz.tjahzi.stats; + +import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.http.HttpResponseStatus; + +public class MutableMonitoringModuleWrapper implements MonitoringModule { + + private MonitoringModule monitoringModule; + + public void setMonitoringModule(MonitoringModule monitoringModule) { + this.monitoringModule = monitoringModule; + } + + @Override + public void incrementDroppedPuts() { + monitoringModule.incrementDroppedPuts(); + } + + @Override + public void incrementDroppedPuts(Throwable throwable) { + monitoringModule.incrementDroppedPuts(throwable); + } + + @Override + public void incrementSentHttpRequests(int sizeBytes) { + monitoringModule.incrementSentHttpRequests(sizeBytes); + } + + @Override + public void incrementFailedHttpRequests() { + monitoringModule.incrementFailedHttpRequests(); + } + + @Override + public void incrementRetriedHttpRequests() { + monitoringModule.incrementRetriedHttpRequests(); + } + + @Override + public void addAgentError(Throwable throwable) { + monitoringModule.addAgentError(throwable); + } + + @Override + public void incrementHttpConnectAttempts() { + monitoringModule.incrementHttpConnectAttempts(); + } + + @Override + public void addPipelineError(Throwable cause) { + monitoringModule.addPipelineError(cause); + } + + @Override + public void incrementChannelInactive() { + monitoringModule.incrementChannelInactive(); + } + + @Override + public void incrementHttpResponses() { + monitoringModule.incrementHttpResponses(); + } + + @Override + public void incrementHttpErrors(HttpResponseStatus status, ByteBuf content) { + monitoringModule.incrementHttpErrors(status, content); + } + + @Override + public void recordResponseTime(long time) { + monitoringModule.recordResponseTime(time); + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java index 8839af6..3fbff74 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java @@ -7,7 +7,6 @@ import org.agrona.concurrent.UnsafeBuffer; import org.agrona.concurrent.errors.DistinctErrorLog; import java.nio.charset.Charset; -import java.time.Clock; import java.util.concurrent.atomic.AtomicLong; public class StandardMonitoringModule implements MonitoringModule { @@ -42,11 +41,6 @@ public class StandardMonitoringModule implements MonitoringModule { distinctErrorLog.record(new NullPointerException()); } - @Override - public Clock getClock() { - return Clock.systemUTC(); - } - @Override public void incrementDroppedPuts() { droppedPuts.incrementAndGet(); From 67a0bfbcc682caef829964b04b628a057ff47270 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Fri, 22 Jan 2021 19:54:21 +0100 Subject: [PATCH 8/8] Re #7: Added more tests and some small changes of monitoring and Netty code. Added closing connection if keepalive is not in force. --- loki-log4j2-appender/pom.xml | 7 + .../tkowalcz/tjahzi/log4j2/LokiAppender.java | 4 + .../log4j2/LokiAppenderMonitoringTest.java | 58 +++++ tjahzi-core/pom.xml | 6 +- .../tjahzi/http/HttpClientInitializer.java | 11 +- .../tkowalcz/tjahzi/http/HttpConnection.java | 1 - .../tjahzi/http/InFlightRequestsTracker.java | 24 -- .../http/PipelinedHttpRequestTimer.java | 1 + ...er.java => RequestAndResponseHandler.java} | 28 ++- .../stats/StandardMonitoringModule.java | 25 ++ .../tjahzi/http/BlockingRetryTest.java | 9 +- .../tjahzi/http/EventLoopGroupRetryTest.java | 10 +- .../http/PipelinedHttpRequestTimerTest.java | 228 ++++++++++++++++++ .../tkowalcz/tjahzi/stats/SettableClock.java | 44 ++++ .../tjahzi/stats/TimingRingBufferTest.java | 120 +++++++++ 15 files changed, 531 insertions(+), 45 deletions(-) create mode 100644 loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderMonitoringTest.java delete mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/InFlightRequestsTracker.java rename tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/{ResponseHandler.java => RequestAndResponseHandler.java} (50%) create mode 100644 tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimerTest.java create mode 100644 tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/SettableClock.java create mode 100644 tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/TimingRingBufferTest.java diff --git a/loki-log4j2-appender/pom.xml b/loki-log4j2-appender/pom.xml index 47d2614..92a91ad 100644 --- a/loki-log4j2-appender/pom.xml +++ b/loki-log4j2-appender/pom.xml @@ -55,6 +55,13 @@ test + + io.dropwizard.metrics + metrics-core + 4.1.17 + provided + + org.junit.jupiter diff --git a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java index 625e136..991a274 100644 --- a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java +++ b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java @@ -68,6 +68,10 @@ public class LokiAppender extends AbstractAppender { return loggingSystem; } + /** + * This is an entry point to set monitoring (statistics) hooks for this appender. This + * API is in beta and is subject to change (and probably will). + */ public void setMonitoringModule(MonitoringModule monitoringModule) { monitoringModuleWrapper.setMonitoringModule(monitoringModule); } diff --git a/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderMonitoringTest.java b/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderMonitoringTest.java new file mode 100644 index 0000000..abd007d --- /dev/null +++ b/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderMonitoringTest.java @@ -0,0 +1,58 @@ +package pl.tkowalcz.tjahzi.log4j2; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.MetricRegistry; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.core.LoggerContext; +import org.junit.jupiter.api.Test; +import org.testcontainers.junit.jupiter.Testcontainers; +import pl.tkowalcz.tjahzi.stats.DropwizardMonitoringModule; + +import java.net.URI; +import java.net.URISyntaxException; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +@Testcontainers +class LokiAppenderMonitoringTest { + + @Test + void shouldInjectMonitoringAndUseIt() throws URISyntaxException { + // Given + URI uri = getClass() + .getClassLoader() + .getResource("basic-appender-test-log4j2-configuration.xml") + .toURI(); + + LoggerContext context = (LoggerContext) LogManager.getContext(false); + context.setConfigLocation(uri); + + LokiAppender loki = context.getConfiguration().getAppender("Loki"); + MetricRegistry metricRegistry = new MetricRegistry(); + loki.setMonitoringModule( + new DropwizardMonitoringModule( + metricRegistry, + "appender.loki" + ) + ); + + // When + Logger logger = LogManager.getLogger(LokiAppenderMonitoringTest.class); + logger.info("Test test test"); + + // Then + assertThat(metricRegistry.getMetrics()).isNotEmpty(); + + Counter connectAttemptsCounter = metricRegistry + .getCounters() + .get("appender.loki.httpConnectAttempts"); + assertThat(connectAttemptsCounter).isNotNull(); + + await().untilAsserted(() -> { + assertThat(connectAttemptsCounter.getCount()).isEqualTo(1); + } + ); + } +} diff --git a/tjahzi-core/pom.xml b/tjahzi-core/pom.xml index 9da063c..879a39e 100644 --- a/tjahzi-core/pom.xml +++ b/tjahzi-core/pom.xml @@ -76,9 +76,9 @@ - io.dropwizard - dropwizard-core - 2.0.18 + io.dropwizard.metrics + metrics-core + 4.1.17 provided 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 6d33a99..3cb928e 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 @@ -1,21 +1,21 @@ package pl.tkowalcz.tjahzi.http; +import io.netty.channel.Channel; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; -import io.netty.channel.socket.SocketChannel; import io.netty.handler.codec.http.HttpClientCodec; import io.netty.handler.codec.http.HttpContentDecompressor; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.timeout.IdleStateHandler; import pl.tkowalcz.tjahzi.stats.MonitoringModule; -class HttpClientInitializer extends ChannelInitializer { +class HttpClientInitializer extends ChannelInitializer { private static final int BYTES_IN_MEGABYTE = 1024 * 1024; private static final int MAX_CONTENT_LENGTH = 10 * BYTES_IN_MEGABYTE; private final MonitoringModule monitoringModule; - private final ResponseHandler responseHandler; + private final RequestAndResponseHandler responseHandler; private final int requestTimeoutMillis; private final int maxRequestsInFlight; @@ -26,14 +26,14 @@ class HttpClientInitializer extends ChannelInitializer { int maxRequestsInFlight ) { this.monitoringModule = monitoringModule; - this.responseHandler = new ResponseHandler(monitoringModule); + this.responseHandler = new RequestAndResponseHandler(monitoringModule); this.requestTimeoutMillis = requestTimeoutMillis; this.maxRequestsInFlight = maxRequestsInFlight; } @Override - protected void initChannel(SocketChannel ch) { + protected void initChannel(Channel ch) { ChannelPipeline p = ch.pipeline(); p.addLast(new IdleStateHandler( requestTimeoutMillis, @@ -50,7 +50,6 @@ class HttpClientInitializer extends ChannelInitializer { maxRequestsInFlight ) ); - p.addLast(new InFlightRequestsTracker()); 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 index 76bfdd5..ff4ef80 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java @@ -68,7 +68,6 @@ public class HttpConnection implements Closeable { stableReference.awaitUninterruptibly(); if (stableReference.isSuccess() && stableReference.channel().isActive()) { stableReference.channel().writeAndFlush(request); - monitoringModule.incrementSentHttpRequests(request.content().readableBytes()); } else { retry.retry(); } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/InFlightRequestsTracker.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/InFlightRequestsTracker.java deleted file mode 100644 index c75db9b..0000000 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/InFlightRequestsTracker.java +++ /dev/null @@ -1,24 +0,0 @@ -package pl.tkowalcz.tjahzi.http; - -import io.netty.channel.ChannelDuplexHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelPromise; - -import java.util.concurrent.atomic.AtomicInteger; - -public class InFlightRequestsTracker extends ChannelDuplexHandler { - - private final AtomicInteger requestsInFlight = new AtomicInteger(); - - @Override - public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { - super.write(ctx, msg, promise); - requestsInFlight.incrementAndGet(); - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - super.channelRead(ctx, msg); - requestsInFlight.decrementAndGet(); - } -} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java index 6945c70..d294cd7 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java @@ -31,3 +31,4 @@ public class PipelinedHttpRequestTimer extends ChannelDuplexHandler { monitoringModule.recordResponseTime(timer.measure()); } } + diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/RequestAndResponseHandler.java similarity index 50% rename from tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java rename to tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/RequestAndResponseHandler.java index c6efc4c..ab1e7f1 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/RequestAndResponseHandler.java @@ -1,28 +1,46 @@ package pl.tkowalcz.tjahzi.http; +import io.netty.channel.ChannelDuplexHandler; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.ChannelPromise; +import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.FullHttpResponse; import io.netty.handler.codec.http.HttpStatusClass; +import io.netty.handler.codec.http.HttpUtil; import pl.tkowalcz.tjahzi.stats.MonitoringModule; @ChannelHandler.Sharable -class ResponseHandler extends SimpleChannelInboundHandler { +class RequestAndResponseHandler extends ChannelDuplexHandler { private final MonitoringModule monitoringModule; - ResponseHandler(MonitoringModule monitoringModule) { - super(false); + RequestAndResponseHandler(MonitoringModule monitoringModule) { this.monitoringModule = monitoringModule; } @Override - protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) { + public void write(ChannelHandlerContext ctx, Object object, ChannelPromise promise) throws Exception { + if (object instanceof FullHttpRequest) { + int payloadSize = ((FullHttpRequest) object).content().readableBytes(); + monitoringModule.incrementSentHttpRequests(payloadSize); + } + + super.write(ctx, object, promise); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object object) { + FullHttpResponse msg = (FullHttpResponse) object; + monitoringModule.incrementHttpResponses(); if (msg.status().codeClass() != HttpStatusClass.SUCCESS) { monitoringModule.incrementHttpErrors(msg.status(), msg.content()); } + + if (!HttpUtil.isKeepAlive(msg)) { + ctx.close(); + } } @Override diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java index 3fbff74..efaacca 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java @@ -58,16 +58,32 @@ public class StandardMonitoringModule implements MonitoringModule { sentBytes.addAndGet(sizeBytes); } + public long getSentHttpRequests() { + return sentHttpRequests.get(); + } + + public long getSentBytes() { + return sentBytes.get(); + } + @Override public void incrementFailedHttpRequests() { failedHttpRequests.incrementAndGet(); } + public long getFailedHttpRequests() { + return failedHttpRequests.get(); + } + @Override public void incrementRetriedHttpRequests() { retriedHttpRequests.incrementAndGet(); } + public long getRetriedHttpRequests() { + return retriedHttpRequests.get(); + } + @Override public void addAgentError(Throwable throwable) { agentErrors.incrementAndGet(); @@ -79,6 +95,10 @@ public class StandardMonitoringModule implements MonitoringModule { httpConnectAttempts.incrementAndGet(); } + public long getHttpConnectAttempts() { + return httpConnectAttempts.get(); + } + @Override public void incrementChannelInactive() { channelInactive.incrementAndGet(); @@ -89,6 +109,10 @@ public class StandardMonitoringModule implements MonitoringModule { httpResponses.incrementAndGet(); } + public long getHttpResponses() { + return httpResponses.get(); + } + @Override public void addPipelineError(Throwable cause) { distinctErrorLog.record(cause); @@ -96,6 +120,7 @@ public class StandardMonitoringModule implements MonitoringModule { @Override public void incrementHttpErrors(HttpResponseStatus status, ByteBuf content) { + failedHttpRequests.incrementAndGet(); distinctErrorLog.record( new RuntimeException( content.toString(Charset.defaultCharset()) diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java index 0347d80..1c4d672 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java @@ -6,16 +6,14 @@ import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.mock; class BlockingRetryTest { - private StandardMonitoringModule monitoringModule = mock(StandardMonitoringModule.class); - @Test void shouldRetryAndInvokeOperation() { // Given AtomicInteger retryCounter = new AtomicInteger(); + StandardMonitoringModule monitoringModule = new StandardMonitoringModule(); BlockingRetry retry = new BlockingRetry( __ -> { @@ -32,12 +30,15 @@ class BlockingRetryTest { // Then assertThat(retryCounter).hasValue(5); + assertThat(monitoringModule.getRetriedHttpRequests()).isEqualTo(5); + assertThat(monitoringModule.getFailedHttpRequests()).isEqualTo(1); } @Test void shouldNotRetryOnItsOwn() { // Given AtomicInteger retryCounter = new AtomicInteger(); + StandardMonitoringModule monitoringModule = new StandardMonitoringModule(); BlockingRetry retry = new BlockingRetry( __ -> { @@ -53,5 +54,7 @@ class BlockingRetryTest { // Then assertThat(retryCounter).hasValue(1); + assertThat(monitoringModule.getRetriedHttpRequests()).isEqualTo(1); + assertThat(monitoringModule.getFailedHttpRequests()).isEqualTo(0); } } diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java index 68b5b66..49e1e5c 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java @@ -12,17 +12,16 @@ import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; -import static org.mockito.Mockito.mock; class EventLoopGroupRetryTest { - private final StandardMonitoringModule monitoringModule = mock(StandardMonitoringModule.class); - private NioEventLoopGroup eventLoopGroup; + private StandardMonitoringModule monitoringModule; @BeforeEach void setUp() { eventLoopGroup = new NioEventLoopGroup(Executors.defaultThreadFactory()); + monitoringModule = new StandardMonitoringModule(); } @AfterEach @@ -34,6 +33,7 @@ class EventLoopGroupRetryTest { void shouldRetryAndInvokeOperation() { // Given AtomicInteger retryCounter = new AtomicInteger(); + StandardMonitoringModule monitoringModule = new StandardMonitoringModule(); EventLoopGroupRetry retry = new EventLoopGroupRetry( eventLoopGroup, @@ -52,6 +52,8 @@ class EventLoopGroupRetryTest { await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> { assertThat(retryCounter).hasValueGreaterThan(3); }); + + assertThat(monitoringModule.getHttpConnectAttempts()).isGreaterThanOrEqualTo(3); } @Test @@ -75,5 +77,7 @@ class EventLoopGroupRetryTest { await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> { assertThat(retryCounter).hasValue(1); }); + + assertThat(monitoringModule.getHttpConnectAttempts()).isEqualTo(1); } } diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimerTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimerTest.java new file mode 100644 index 0000000..ad1d5d8 --- /dev/null +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimerTest.java @@ -0,0 +1,228 @@ +package pl.tkowalcz.tjahzi.http; + +import io.netty.buffer.Unpooled; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.*; +import org.junit.jupiter.api.Test; +import org.testcontainers.shaded.org.apache.commons.lang.mutable.MutableLong; +import pl.tkowalcz.tjahzi.stats.SettableClock; +import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; + +import java.time.Clock; + +import static org.assertj.core.api.Assertions.assertThat; + +class PipelinedHttpRequestTimerTest { + + @Test + void shouldCloseChannelIfNoKeepalive() { + // Given + SettableClock clock = new SettableClock(); + + StandardMonitoringModule monitoringModule = new StandardMonitoringModule() { + @Override + public Clock getClock() { + return clock; + } + }; + + HttpClientInitializer initializer = new HttpClientInitializer( + monitoringModule, + 10_000, + 42 + ); + + FullHttpRequest request = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, + HttpMethod.POST, + ClientConfigurationBuilder.DEFAULT_LOG_ENDPOINT, + Unpooled.wrappedBuffer("Boo hoo".getBytes()) + ); + + HttpHeaders headers = new DefaultHttpHeaders(); + headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE); + + DefaultFullHttpResponse response = new DefaultFullHttpResponse( + HttpVersion.HTTP_1_1, + HttpResponseStatus.NO_CONTENT, + Unpooled.wrappedBuffer("OK".getBytes()), + headers, + EmptyHttpHeaders.INSTANCE + ); + + // When + EmbeddedChannel channel = new EmbeddedChannel(initializer); + channel.writeOneOutbound(request); + channel.flush(); + + channel.writeOneInbound(response); + channel.flush(); + + // Then + assertThat(channel.isOpen()).isFalse(); + } + + @Test + void shouldNotCloseChannelOnKeepalive() { + // Given + SettableClock clock = new SettableClock(); + + StandardMonitoringModule monitoringModule = new StandardMonitoringModule() { + @Override + public Clock getClock() { + return clock; + } + }; + + HttpClientInitializer initializer = new HttpClientInitializer( + monitoringModule, + 10_000, + 42 + ); + + FullHttpRequest request = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, + HttpMethod.POST, + ClientConfigurationBuilder.DEFAULT_LOG_ENDPOINT, + Unpooled.wrappedBuffer("Boo hoo".getBytes()) + ); + + HttpHeaders headers = new DefaultHttpHeaders(); + headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE); + + DefaultFullHttpResponse response = new DefaultFullHttpResponse( + HttpVersion.HTTP_1_1, + HttpResponseStatus.NO_CONTENT, + Unpooled.wrappedBuffer("OK".getBytes()), + headers, + EmptyHttpHeaders.INSTANCE + ); + + // When + EmbeddedChannel channel = new EmbeddedChannel(initializer); + channel.writeOneOutbound(request); + channel.flush(); + + channel.writeOneInbound(response); + channel.flush(); + + // Then + assertThat(channel.isOpen()).isTrue(); + } + + @Test + void shouldReportRequestMetrics() { + // Given + SettableClock clock = new SettableClock(); + MutableLong requestRTT = new MutableLong(); + + StandardMonitoringModule monitoringModule = new StandardMonitoringModule() { + @Override + public Clock getClock() { + return clock; + } + + @Override + public void recordResponseTime(long time) { + requestRTT.setValue(time); + } + }; + + HttpClientInitializer initializer = new HttpClientInitializer( + monitoringModule, + 10_000, + 42 + ); + + FullHttpRequest request = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, + HttpMethod.POST, + ClientConfigurationBuilder.DEFAULT_LOG_ENDPOINT, + Unpooled.wrappedBuffer("Boo hoo".getBytes()) + ); + + HttpHeaders headers = new DefaultHttpHeaders(); + headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE); + + DefaultFullHttpResponse response = new DefaultFullHttpResponse( + HttpVersion.HTTP_1_1, + HttpResponseStatus.NO_CONTENT + ); + + // When + EmbeddedChannel channel = new EmbeddedChannel(initializer); + channel.writeOneOutbound(request); + channel.flush(); + + clock.tick(); + + channel.writeOneInbound(response); + channel.flush(); + + // Then + assertThat(monitoringModule.getFailedHttpRequests()).isZero(); + assertThat(monitoringModule.getSentHttpRequests()).isEqualTo(1); + assertThat(monitoringModule.getHttpResponses()).isEqualTo(1); + assertThat(monitoringModule.getSentBytes()).isEqualTo(7); + + assertThat(requestRTT.intValue()).isEqualTo(SettableClock.TICK_AMOUNT); + } + + @Test + void shouldReportRequestMetricsOnError() { + // Given + SettableClock clock = new SettableClock(); + MutableLong requestRTT = new MutableLong(); + + StandardMonitoringModule monitoringModule = new StandardMonitoringModule() { + @Override + public Clock getClock() { + return clock; + } + + @Override + public void recordResponseTime(long time) { + requestRTT.setValue(time); + } + }; + + HttpClientInitializer initializer = new HttpClientInitializer( + monitoringModule, + 10_000, + 42 + ); + + FullHttpRequest request = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, + HttpMethod.POST, + ClientConfigurationBuilder.DEFAULT_LOG_ENDPOINT, + Unpooled.wrappedBuffer("Boo hoo".getBytes()) + ); + + HttpHeaders headers = new DefaultHttpHeaders(); + headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE); + + DefaultFullHttpResponse response = new DefaultFullHttpResponse( + HttpVersion.HTTP_1_1, + HttpResponseStatus.INTERNAL_SERVER_ERROR + ); + + // When + EmbeddedChannel channel = new EmbeddedChannel(initializer); + channel.writeOneOutbound(request); + channel.flush(); + + clock.tick(); + + channel.writeOneInbound(response); + channel.flush(); + + // Then + assertThat(monitoringModule.getFailedHttpRequests()).isEqualTo(1); + assertThat(monitoringModule.getSentHttpRequests()).isEqualTo(1); + assertThat(monitoringModule.getHttpResponses()).isEqualTo(1); + assertThat(monitoringModule.getSentBytes()).isEqualTo(7); + + assertThat(requestRTT.intValue()).isEqualTo(SettableClock.TICK_AMOUNT); + } +} diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/SettableClock.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/SettableClock.java new file mode 100644 index 0000000..c5d7c22 --- /dev/null +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/SettableClock.java @@ -0,0 +1,44 @@ +package pl.tkowalcz.tjahzi.stats; + +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneId; + +public class SettableClock extends Clock { + + public static final long TICK_AMOUNT = 11L; + + private long millis; + + public void setMillis(long millis) { + this.millis = millis; + } + + public void tick() { + millis += TICK_AMOUNT; + } + + public void tick(long amount) { + millis += amount; + } + + @Override + public ZoneId getZone() { + return ZoneId.systemDefault(); + } + + @Override + public Clock withZone(ZoneId zone) { + throw new UnsupportedOperationException(); + } + + @Override + public Instant instant() { + throw new UnsupportedOperationException(); + } + + @Override + public long millis() { + return millis; + } +} diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/TimingRingBufferTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/TimingRingBufferTest.java new file mode 100644 index 0000000..6fb36c9 --- /dev/null +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/stats/TimingRingBufferTest.java @@ -0,0 +1,120 @@ +package pl.tkowalcz.tjahzi.stats; + +import org.junit.jupiter.api.Test; + +import java.time.Clock; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class TimingRingBufferTest { + + @Test + void shouldActLikeARingBufferOfTimings() { + // Given + Clock clock = mock(Clock.class); + + long firstRecording = 10L; + long secondRecording = 20L; + long thirdRecording = 33L; + long firstResponse = 38L; + long secondResponse = 41L; + long thirdResponse = 56L; + + when(clock.millis()) + .thenReturn(firstRecording) + .thenReturn(secondRecording) + .thenReturn(thirdRecording) + .thenReturn(firstResponse) + .thenReturn(secondResponse) + .thenReturn(thirdResponse); + + TimingRingBuffer buffer = new TimingRingBuffer( + clock, + 10 + ); + + // When + buffer.record(); + buffer.record(); + buffer.record(); + + long firstRTT = buffer.measure(); + long secondRTT = buffer.measure(); + long thirdRTT = buffer.measure(); + + // Then + assertThat(firstRTT).isEqualTo(firstResponse - firstRecording); + assertThat(secondRTT).isEqualTo(secondResponse - secondRecording); + assertThat(thirdRTT).isEqualTo(thirdResponse - thirdRecording); + } + + @Test + void shouldHandleIntermixedResponsesAndNewRexuests() { + // Given + Clock clock = mock(Clock.class); + + long firstRecording = 10L; + long secondRecording = 20L; + long firstResponse = 22L; + long thirdRecording = 33L; + long secondResponse = 41L; + long fourthRecording = 48L; + long thirdResponse = 56L; + long fourthResponse = 58L; + + when(clock.millis()) + .thenReturn(firstRecording) + .thenReturn(secondRecording) + .thenReturn(firstResponse) + .thenReturn(thirdRecording) + .thenReturn(secondResponse) + .thenReturn(fourthRecording) + .thenReturn(thirdResponse) + .thenReturn(fourthResponse); + + TimingRingBuffer buffer = new TimingRingBuffer( + clock, + 10 + ); + + // When + buffer.record(); + buffer.record(); + long firstRTT = buffer.measure(); + buffer.record(); + long secondRTT = buffer.measure(); + buffer.record(); + long thirdRTT = buffer.measure(); + long fourthRTT = buffer.measure(); + + // Then + assertThat(firstRTT).isEqualTo(firstResponse - firstRecording); + assertThat(secondRTT).isEqualTo(secondResponse - secondRecording); + assertThat(thirdRTT).isEqualTo(thirdResponse - thirdRecording); + assertThat(fourthRTT).isEqualTo(fourthResponse - fourthRecording); + } + + @Test + void shouldHandleWrappingAround() { + // Given + SettableClock clock = new SettableClock(); + + TimingRingBuffer buffer = new TimingRingBuffer( + clock, + 10 + ); + + // When & Then + for (int i = 0; i < 23; i++) { + long tick = 40L; + + buffer.record(); + clock.tick(tick); + long actual = buffer.measure(); + + assertThat(actual).isEqualTo(tick); + } + } +}