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/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; 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..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 @@ -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,14 @@ 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); + } + @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 641e0e4..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,8 @@ 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; import java.util.stream.Stream; @@ -56,6 +58,9 @@ public class LokiAppenderBuilder> extends Abstr @PluginBuilderAttribute private String logLevelLabel; + @PluginBuilderAttribute + private int maxRequestsInFlight = 100; + @PluginElement("Headers") private Header[] headers; @@ -70,16 +75,21 @@ public class LokiAppenderBuilder> extends Abstr .withPort(port) .withMaxRetries(maxRetries) .withRequestTimeoutMillis(readTimeoutMillis) + .withMaxRequestsInFlight(maxRequestsInFlight) .build(); String[] additionalHeaders = stream(headers) .flatMap(header -> Stream.of(header.getName(), header.getValue())) .toArray(String[]::new); + MutableMonitoringModuleWrapper monitoringModuleWrapper = new MutableMonitoringModuleWrapper(); + monitoringModuleWrapper.setMonitoringModule(new StandardMonitoringModule()); + NettyHttpClient httpClient = HttpClientFactory .defaultFactory() .getHttpClient( configurationBuilder, + monitoringModuleWrapper, additionalHeaders ); @@ -97,6 +107,7 @@ public class LokiAppenderBuilder> extends Abstr LoggingSystem loggingSystem = new TjahziInitializer().createLoggingSystem( httpClient, + monitoringModuleWrapper, lokiLabels, bufferSizeBytes, isUseOffHeapBuffer() @@ -109,7 +120,8 @@ public class LokiAppenderBuilder> extends Abstr isIgnoreExceptions(), getPropertyArray(), logLevelLabel, - loggingSystem + loggingSystem, + monitoringModuleWrapper ); } @@ -184,6 +196,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/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 cb8bc4f..879a39e 100644 --- a/tjahzi-core/pom.xml +++ b/tjahzi-core/pom.xml @@ -75,6 +75,12 @@ test + + io.dropwizard.metrics + metrics-core + 4.1.17 + provided + 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..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 @@ -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.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 ed6030e..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,6 +8,7 @@ public class ClientConfiguration { private final int connectionTimeoutMillis; private final int requestTimeoutMillis; + private final int maxRequestsInFlight; private final int maxRetries; @@ -17,12 +18,14 @@ public class ClientConfiguration { int port, int connectionTimeoutMillis, int requestTimeoutMillis, + int maxRequestsInFlight, int maxRetries) { this.logEndpoint = logEndpoint; this.host = host; this.port = port; this.connectionTimeoutMillis = connectionTimeoutMillis; this.requestTimeoutMillis = requestTimeoutMillis; + this.maxRequestsInFlight = maxRequestsInFlight; this.maxRetries = maxRetries; } @@ -47,6 +50,10 @@ public class ClientConfiguration { return requestTimeoutMillis; } + public int getMaxRequestsInFlight() { + return maxRequestsInFlight; + } + 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..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,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_REQUESTS_IN_FLIGHT = 100; + 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 maxRequestsInFlight = DEFAULT_MAX_REQUESTS_IN_FLIGHT; private int maxRetries = DEFAULT_MAX_RETRIES; @@ -43,18 +46,28 @@ public class ClientConfigurationBuilder { return this; } + public ClientConfigurationBuilder withMaxRequestsInFlight(int maxRequestsInFlight) { + this.maxRequestsInFlight = maxRequestsInFlight; + return this; + } + public ClientConfigurationBuilder withMaxRetries(int maxRetries) { this.maxRetries = maxRetries; return this; } public ClientConfiguration build() { + if (maxRequestsInFlight <= 0) { + throw new IllegalArgumentException("Property maxRequestsInFlight must be greater than 0"); + } + return new ClientConfiguration( logEndpoint, host, port, connectionTimeoutMillis, requestTimeoutMillis, + maxRequestsInFlight, 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..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,32 +1,55 @@ 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 RequestAndResponseHandler responseHandler; - private final ResponseHandler responseHandler; private final int requestTimeoutMillis; + private final int maxRequestsInFlight; + + HttpClientInitializer( + MonitoringModule monitoringModule, + int requestTimeoutMillis, + int maxRequestsInFlight + ) { + this.monitoringModule = monitoringModule; + this.responseHandler = new RequestAndResponseHandler(monitoringModule); - HttpClientInitializer(ResponseHandler responseHandler, int requestTimeoutMillis) { - this.responseHandler = responseHandler; 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, 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( + new PipelinedHttpRequestTimer( + monitoringModule, + maxRequestsInFlight + ) + ); 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 e6d3b2a..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 @@ -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); @@ -60,9 +69,7 @@ public class HttpConnection implements Closeable { if (stableReference.isSuccess() && stableReference.channel().isActive()) { stableReference.channel().writeAndFlush(request); } 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/PipelinedHttpRequestTimer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java new file mode 100644 index 0000000..d294cd7 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/PipelinedHttpRequestTimer.java @@ -0,0 +1,34 @@ +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 { + + private final MonitoringModule monitoringModule; + private final TimingRingBuffer timer; + + public PipelinedHttpRequestTimer( + MonitoringModule monitoringModule, + int maxRequestsInFlight + ) { + this.monitoringModule = monitoringModule; + this.timer = new TimingRingBuffer(monitoringModule.getClock(), maxRequestsInFlight); + } + + @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/http/RequestAndResponseHandler.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/RequestAndResponseHandler.java new file mode 100644 index 0000000..ab1e7f1 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/RequestAndResponseHandler.java @@ -0,0 +1,55 @@ +package pl.tkowalcz.tjahzi.http; + +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +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 RequestAndResponseHandler extends ChannelDuplexHandler { + + private final MonitoringModule monitoringModule; + + RequestAndResponseHandler(MonitoringModule monitoringModule) { + this.monitoringModule = monitoringModule; + } + + @Override + 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 + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + monitoringModule.addPipelineError(cause); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) { + monitoringModule.incrementChannelInactive(); + } +} 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 deleted file mode 100644 index 8ce408a..0000000 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java +++ /dev/null @@ -1,28 +0,0 @@ -package pl.tkowalcz.tjahzi.http; - -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.SimpleChannelInboundHandler; -import io.netty.handler.codec.http.FullHttpResponse; - -@ChannelHandler.Sharable -class ResponseHandler extends SimpleChannelInboundHandler { - - ResponseHandler() { - super(false); - } - - @Override - protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) { - } - - @Override - public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { - System.out.println("cause = " + cause); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) { - System.out.println("ResponseHandler.channelInactive"); - } -} 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 new file mode 100644 index 0000000..2fca4c8 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/EmptyMonitoringModule.java @@ -0,0 +1,55 @@ +package pl.tkowalcz.tjahzi.stats; + +import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.http.HttpResponseStatus; + +public class EmptyMonitoringModule implements MonitoringModule { + + @Override + public void incrementDroppedPuts() { + } + + @Override + public void incrementDroppedPuts(Throwable throwable) { + } + + @Override + public void incrementSentHttpRequests(int sizeBytes) { + } + + @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..3c1e972 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/MonitoringModule.java @@ -0,0 +1,37 @@ +package pl.tkowalcz.tjahzi.stats; + +import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.http.HttpResponseStatus; + +import java.time.Clock; + +public interface MonitoringModule { + + default Clock getClock() { + return Clock.systemUTC(); + } + + void incrementDroppedPuts(); + + void incrementDroppedPuts(Throwable throwable); + + void incrementSentHttpRequests(int sizeBytes); + + void incrementFailedHttpRequests(); + + void incrementRetriedHttpRequests(); + + void addAgentError(Throwable throwable); + + void incrementHttpConnectAttempts(); + + void addPipelineError(Throwable cause); + + void incrementChannelInactive(); + + void incrementHttpResponses(); + + void incrementHttpErrors(HttpResponseStatus status, ByteBuf content); + + void recordResponseTime(long time); +} 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 new file mode 100644 index 0000000..efaacca --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/stats/StandardMonitoringModule.java @@ -0,0 +1,149 @@ +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.util.concurrent.atomic.AtomicLong; + +public class StandardMonitoringModule implements MonitoringModule { + + private static final int ERROR_LOG_CAPACITY = 1024; + + 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[ERROR_LOG_CAPACITY]), + new SystemEpochClock() + ); + + distinctErrorLog.record(new NullPointerException()); + } + + @Override + public void incrementDroppedPuts() { + droppedPuts.incrementAndGet(); + } + + @Override + public void incrementDroppedPuts(Throwable throwable) { + incrementDroppedPuts(); + distinctErrorLog.record(throwable); + } + + @Override + public void incrementSentHttpRequests(int sizeBytes) { + sentHttpRequests.incrementAndGet(); + 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(); + distinctErrorLog.record(throwable); + } + + @Override + public void incrementHttpConnectAttempts() { + httpConnectAttempts.incrementAndGet(); + } + + public long getHttpConnectAttempts() { + return httpConnectAttempts.get(); + } + + @Override + public void incrementChannelInactive() { + channelInactive.incrementAndGet(); + } + + @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) { + failedHttpRequests.incrementAndGet(); + distinctErrorLog.record( + new RuntimeException( + content.toString(Charset.defaultCharset()) + ) + ); + } + + @Override + public void recordResponseTime(long time) { + } + + @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 new file mode 100644 index 0000000..0ddc919 --- /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 maxRequestsInFlight) { + this.clock = clock; + times = new long[maxRequestsInFlight]; + } + + 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..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 @@ -1,6 +1,7 @@ package pl.tkowalcz.tjahzi.http; import org.junit.jupiter.api.Test; +import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule; import java.util.concurrent.atomic.AtomicInteger; @@ -12,6 +13,7 @@ class BlockingRetryTest { void shouldRetryAndInvokeOperation() { // Given AtomicInteger retryCounter = new AtomicInteger(); + StandardMonitoringModule monitoringModule = new StandardMonitoringModule(); BlockingRetry retry = new BlockingRetry( __ -> { @@ -19,7 +21,8 @@ class BlockingRetryTest { __.retry(); }, new ExponentialBackoffStrategy(1, 1, 1), - 5 + 5, + monitoringModule ); // When @@ -27,19 +30,23 @@ 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( __ -> { retryCounter.incrementAndGet(); }, new ExponentialBackoffStrategy(1, 1, 1), - 5 + 5, + monitoringModule ); // When @@ -47,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 1f72760..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 @@ -5,6 +5,7 @@ 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; @@ -15,10 +16,12 @@ import static org.awaitility.Awaitility.await; class EventLoopGroupRetryTest { private NioEventLoopGroup eventLoopGroup; + private StandardMonitoringModule monitoringModule; @BeforeEach void setUp() { eventLoopGroup = new NioEventLoopGroup(Executors.defaultThreadFactory()); + monitoringModule = new StandardMonitoringModule(); } @AfterEach @@ -30,6 +33,7 @@ class EventLoopGroupRetryTest { void shouldRetryAndInvokeOperation() { // Given AtomicInteger retryCounter = new AtomicInteger(); + StandardMonitoringModule monitoringModule = new StandardMonitoringModule(); EventLoopGroupRetry retry = new EventLoopGroupRetry( eventLoopGroup, @@ -37,7 +41,8 @@ class EventLoopGroupRetryTest { retryCounter.incrementAndGet(); __.retry(); }, - new ExponentialBackoffStrategy(250, 1000, 2) + new ExponentialBackoffStrategy(250, 1000, 2), + monitoringModule ); // When @@ -47,6 +52,8 @@ class EventLoopGroupRetryTest { await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> { assertThat(retryCounter).hasValueGreaterThan(3); }); + + assertThat(monitoringModule.getHttpConnectAttempts()).isGreaterThanOrEqualTo(3); } @Test @@ -59,7 +66,8 @@ class EventLoopGroupRetryTest { __ -> { retryCounter.incrementAndGet(); }, - new ExponentialBackoffStrategy(1, 1, 1) + new ExponentialBackoffStrategy(1, 1, 1), + monitoringModule ); // When @@ -69,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); + } + } +}