Merge pull request #20 from tkowalcz/7-monitoring

7 monitoring

loki-log4j2-appender/pom.xml

@@ -55,6 +55,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+            <version>4.1.17</version>
+            <scope>provided</scope>
+        </dependency>
+
         <!-- Testcontainers -->
         <dependency>
             <groupId>org.junit.jupiter</groupId>

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<LogEvent, ByteBuffer> {
 
-    protected static final org.apache.logging.log4j.Logger LOGGER = StatusLogger.getLogger();
+    protected static final Logger LOGGER = StatusLogger.getLogger();
 
     private final LoggingSystem loggingSystem;
 

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<B>().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();

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<B extends LokiAppenderBuilder<B>> extends Abstr
     @PluginBuilderAttribute
     private String logLevelLabel;
 
+    @PluginBuilderAttribute
+    private int maxRequestsInFlight = 100;
+
     @PluginElement("Headers")
     private Header[] headers;
 
@@ -70,16 +75,21 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> 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<B extends LokiAppenderBuilder<B>> extends Abstr
 
         LoggingSystem loggingSystem = new TjahziInitializer().createLoggingSystem(
                 httpClient,
+                monitoringModuleWrapper,
                 lokiLabels,
                 bufferSizeBytes,
                 isUseOffHeapBuffer()
@@ -109,7 +120,8 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
                 isIgnoreExceptions(),
                 getPropertyArray(),
                 logLevelLabel,
-                loggingSystem
+                loggingSystem,
+                monitoringModuleWrapper
         );
     }
 
@@ -184,6 +196,14 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> 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;
     }

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);
+                }
+        );
+    }
+}

tjahzi-core/pom.xml

@@ -75,6 +75,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+            <version>4.1.17</version>
+            <scope>provided</scope>
+        </dependency>
 
         <!-- Testcontainers -->
         <dependency>

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() {

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<String, String> 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);
             }
 

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);
         }
     }
 }

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<BlockingRetry> operation;
+    private final MonitoringModule monitoringModule;
 
     private int retries;
 
@@ -13,17 +16,25 @@ class BlockingRetry implements Retry {
     BlockingRetry(
             Consumer<BlockingRetry> operation,
             ExponentialBackoffStrategy strategy,
-            int retries) {
+            int retries,
+            MonitoringModule monitoringModule
+    ) {
         this.operation = operation;
         this.strategy = strategy;
+
         this.retries = retries;
+        this.monitoringModule = monitoringModule;
     }
 
-    BlockingRetry(Consumer<BlockingRetry> operation, int retries) {
+    BlockingRetry(
+            Consumer<BlockingRetry> 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();
         }
     }
 }

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(

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;
     }

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
         );
     }

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<EventLoopGroupRetry> operation;
     private final ExponentialBackoffStrategy strategy;
 
+    private final MonitoringModule monitoringModule;
+
     // VisibleForTesting
     public EventLoopGroupRetry(
             ScheduledExecutorService group,
             Consumer<EventLoopGroupRetry> operation,
-            ExponentialBackoffStrategy strategy) {
+            ExponentialBackoffStrategy strategy,
+            MonitoringModule monitoringModule
+    ) {
         this.group = group;
         this.operation = operation;
         this.strategy = strategy;
+
+        this.monitoringModule = monitoringModule;
     }
 
-    public EventLoopGroupRetry(ScheduledExecutorService group, Consumer<EventLoopGroupRetry> operation) {
+    public EventLoopGroupRetry(
+            ScheduledExecutorService group,
+            Consumer<EventLoopGroupRetry> operation,
+            MonitoringModule monitoringModule
+    ) {
         this(
                 group,
                 operation,
-                ExponentialBackoffStrategy.withDefault()
+                ExponentialBackoffStrategy.withDefault(),
+                monitoringModule
         );
     }
 
     @Override
     public void run() {
+        monitoringModule.incrementHttpConnectAttempts();
         operation.accept(this);
     }
 

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
         );
     }

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<SocketChannel> {
+class HttpClientInitializer extends ChannelInitializer<Channel> {
 
     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);
     }
 }

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
         }
     }
 

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) {

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());
+    }
+}
+ 

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();
+    }
+}

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java

@@ -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<FullHttpResponse> {
-
-    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");
-    }
-}

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 +
+                '}';
+    }
+}

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) {
+    }
+}

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);
+}

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);
+    }
+}

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 +
+                '}';
+    }
+}

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);
+        }
+    }
+}

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];
+    }
+}

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

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

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

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

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

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);
     }
 }

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);
     }
 }

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);
+    }
+}

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;
+    }
+}

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);
+        }
+    }
+}