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 @@
 
 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.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 static <B extends LokiAppenderBuilder<B>> B newBuilder() {
         return new LokiAppenderBuilder<B>().asBuilder();
     }
 
+
     private final LoggingSystem loggingSystem;
+    private final MutableMonitoringModuleWrapper monitoringModuleWrapper;
+
     private final AppenderLogic appenderLogic;
 
     LokiAppender(
@@ -39,14 +44,17 @@ public static <B extends LokiAppenderBuilder<B>> B newBuilder() {
             boolean ignoreExceptions,
             Property[] properties,
             String logLevelLabel,
-            LoggingSystem loggingSystem) {
+            LoggingSystem loggingSystem,
+            MutableMonitoringModuleWrapper monitoringModuleWrapper
+    ) {
         super(
                 name,
                 filter,
                 layout,
                 ignoreExceptions,
                 properties
         );
+        this.monitoringModuleWrapper = monitoringModuleWrapper;
         Objects.requireNonNull(layout, "layout");
 
         this.loggingSystem = loggingSystem;
@@ -60,6 +68,14 @@ public LoggingSystem getLoggingSystem() {
         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.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 LokiAppender build() {
                 .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 LokiAppender build() {
 
         LoggingSystem loggingSystem = new TjahziInitializer().createLoggingSystem(
                 httpClient,
+                monitoringModuleWrapper,
                 lokiLabels,
                 bufferSizeBytes,
                 isUseOffHeapBuffer()
@@ -109,7 +120,8 @@ public LokiAppender build() {
                 isIgnoreExceptions(),
                 getPropertyArray(),
                 logLevelLabel,
-                loggingSystem
+                loggingSystem,
+                monitoringModuleWrapper
         );
     }
 
@@ -184,6 +196,14 @@ public void setLogLevelLabel(String logLevelLabel) {
         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 @@
 
 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.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,14 +36,15 @@ public LoggingSystem createLoggingSystem(
 
         AgentRunner runner = new AgentRunner(
                 new SleepingMillisIdleStrategy(),
-                Throwable::printStackTrace,
+                monitoringModule::addAgentError,
                 null,
                 agent
         );
 
         return new LoggingSystem(
                 logBuffer,
                 runner,
+                monitoringModule,
                 httpClient
         );
     }
@@ -58,7 +61,7 @@ static int findNearestPowerOfTwo(int bufferSize) {
             }
 
             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 TjahziLogger log(long timestamp,
                     claim
             );
         } else {
-            // TODO: increment dropped lines metric
+            monitoringModule.incrementDroppedPuts();
         }
 
         return this;
@@ -68,6 +73,7 @@ private void putMessageOnRing(
             logBuffer.commit(claim);
         } catch (Throwable t) {
             logBuffer.abort(claim);
+            monitoringModule.incrementDroppedPuts(t);
         }
     }
 }