Re #9: Added separate log level label and its configuration. Refactored serialization code (that writes into log buffer).

loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java

@@ -8,6 +8,7 @@ import pl.tkowalcz.tjahzi.TjahziLogger;
 
 import java.io.Serializable;
 import java.nio.ByteBuffer;
+import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
@@ -17,18 +18,23 @@ public class AppenderLogic implements BiConsumer<LogEvent, ByteBuffer> {
     protected static final org.apache.logging.log4j.Logger LOGGER = StatusLogger.getLogger();
 
     private final LoggingSystem loggingSystem;
-    private final TjahziLogger logger;
 
+    private final String logLevelLabel;
     private final Map<String, String> lokiLabels;
 
+    private final TjahziLogger logger;
+
     public AppenderLogic(
             LoggingSystem loggingSystem,
+            String logLevelLabel,
             Map<String, String> lokiLabels
     ) {
         this.loggingSystem = loggingSystem;
-        logger = loggingSystem.createLogger();
 
-        this.lokiLabels = lokiLabels;
+        this.lokiLabels = Collections.unmodifiableMap(lokiLabels);
+        this.logLevelLabel = logLevelLabel;
+
+        logger = loggingSystem.createLogger();
     }
 
     public void append(
@@ -51,9 +57,13 @@ public class AppenderLogic implements BiConsumer<LogEvent, ByteBuffer> {
 
     @Override
     public void accept(LogEvent event, ByteBuffer byteBuffer) {
+        String logLevel = event.getLevel().toString();
+
         logger.log(
                 event.getTimeMillis(),
                 lokiLabels,
+                logLevelLabel,
+                logLevel,
                 byteBuffer
         );
     }

loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java

@@ -68,6 +68,9 @@ public class LokiAppender extends AbstractAppender {
         @PluginBuilderAttribute
         private boolean useOffHeapBuffer = true;
 
+        @PluginBuilderAttribute
+        private String logLevelLabel;
+
         @PluginElement("Headers")
         private Header[] headers;
 
@@ -118,6 +121,7 @@ public class LokiAppender extends AbstractAppender {
                     getFilter(),
                     isIgnoreExceptions(),
                     getPropertyArray(),
+                    logLevelLabel,
                     lokiLabels,
                     loggingSystem
             );
@@ -186,6 +190,14 @@ public class LokiAppender extends AbstractAppender {
             this.useOffHeapBuffer = useOffHeapBuffer;
         }
 
+        public String getLogLevelLabel() {
+            return logLevelLabel;
+        }
+
+        public void setLogLevelLabel(String logLevelLabel) {
+            this.logLevelLabel = logLevelLabel;
+        }
+
         public void setHeaders(Header[] headers) {
             this.headers = headers;
         }
@@ -216,6 +228,7 @@ public class LokiAppender extends AbstractAppender {
             Filter filter,
             boolean ignoreExceptions,
             Property[] properties,
+            String logLevelLabel,
             Map<String, String> lokiLabels,
             LoggingSystem loggingSystem) {
         super(
@@ -230,6 +243,7 @@ public class LokiAppender extends AbstractAppender {
         this.loggingSystem = loggingSystem;
         this.appenderLogic = new AppenderLogic(
                 loggingSystem,
+                logLevelLabel,
                 lokiLabels
         );
     }

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java

@@ -0,0 +1,104 @@
+package pl.tkowalcz.tjahzi;
+
+import org.agrona.concurrent.AtomicBuffer;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+public class LogBufferSerializer {
+
+    private final AtomicBuffer buffer;
+
+    public LogBufferSerializer(AtomicBuffer buffer) {
+        this.buffer = buffer;
+    }
+
+    public void writeTo(
+            int cursor,
+            long timestamp,
+            Map<String, String> labels,
+            String logLevelLabel,
+            String logLevel,
+            ByteBuffer line
+    ) {
+        cursor = writeHeader(cursor, timestamp, labels);
+        cursor = writeLabels(cursor, labels);
+        cursor = writeLabel(
+                cursor,
+                logLevelLabel,
+                logLevel
+        );
+        writeLogLine(cursor, line);
+    }
+
+    private void writeLogLine(int cursor, ByteBuffer line) {
+        buffer.putInt(cursor, line.remaining());
+        cursor += Integer.BYTES;
+
+        buffer.putBytes(
+                cursor,
+                line,
+                line.remaining()
+        );
+    }
+
+    private int writeLabels(int cursor, Map<String, String> labels) {
+        for (Map.Entry<String, String> entry : labels.entrySet()) {
+            cursor = writeLabel(
+                    cursor,
+                    entry.getKey(),
+                    entry.getValue()
+            );
+        }
+        return cursor;
+    }
+
+    private int writeHeader(int cursor, long timestamp, Map<String, String> labels) {
+        buffer.putLong(cursor, timestamp);
+        cursor += Long.BYTES;
+
+        buffer.putInt(cursor, labels.size() + 1);
+        cursor += Integer.BYTES;
+        return cursor;
+    }
+
+    public int writeLabel(
+            int cursor,
+            String key,
+            String value
+    ) {
+        cursor += buffer.putStringAscii(cursor, key);
+        cursor += buffer.putStringAscii(cursor, value);
+
+        return cursor;
+    }
+
+    private int calculateRequiredSizeAscii(Map<String, String> labels) {
+        int timestampSize = Long.BYTES;
+        int labelsCountSize = Integer.BYTES;
+        int sum = timestampSize + labelsCountSize;
+
+        for (Map.Entry<String, String> entry : labels.entrySet()) {
+            sum += calculateLabeSizeAscii(entry.getKey(), entry.getValue());
+        }
+
+        return sum;
+    }
+
+    private int calculateLabeSizeAscii(String key, String value) {
+        return Integer.BYTES + key.length()
+                + Integer.BYTES + value.length();
+    }
+
+    public int calculateRequiredSizeAscii(
+            Map<String, String> labels,
+            String logLevelLabel,
+            String logLevel,
+            ByteBuffer line
+    ) {
+        int lineSize = Integer.BYTES + line.remaining();
+        int logLevelSize = calculateLabeSizeAscii(logLevelLabel, logLevel);
+
+        return calculateRequiredSizeAscii(labels) + lineSize + logLevelSize;
+    }
+}

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java

@@ -1,6 +1,5 @@
 package pl.tkowalcz.tjahzi;
 
-import org.agrona.concurrent.AtomicBuffer;
 import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
 
 import java.nio.ByteBuffer;
@@ -11,111 +10,42 @@ public class TjahziLogger {
     public static final int LOG_MESSAGE_TYPE_ID = 5;
 
     private final ManyToOneRingBuffer logBuffer;
+    private final LogBufferSerializer serializer;
 
     public TjahziLogger(ManyToOneRingBuffer logBuffer) {
         this.logBuffer = logBuffer;
+        this.serializer = new LogBufferSerializer(logBuffer.buffer());
     }
 
     public TjahziLogger log(long timestamp,
                             Map<String, String> labels,
-                            String line) {
-        int requiredSize = calculateRequiredSizeAscii(labels, line);
-        int claim = logBuffer.tryClaim(LOG_MESSAGE_TYPE_ID, requiredSize);
-
-        int cursor = claim;
-        if (requiredSize > 0) {
-            try {
-                AtomicBuffer buffer = this.logBuffer.buffer();
-
-                buffer.putLong(cursor, timestamp);
-                cursor += Long.BYTES;
-
-                buffer.putInt(cursor, labels.size());
-                cursor += Integer.BYTES;
-
-                for (Map.Entry<String, String> entry : labels.entrySet()) {
-                    String key = entry.getKey();
-                    String value = entry.getValue();
-
-                    cursor += buffer.putStringAscii(cursor, key);
-                    cursor += buffer.putStringAscii(cursor, value);
-                }
-
-                buffer.putStringAscii(cursor, line);
-                logBuffer.commit(claim);
-            } catch (Throwable t) {
-                logBuffer.abort(claim);
-            }
-        }
-
-        return this;
-    }
-
-    public TjahziLogger log(long timestamp,
-                            Map<String, String> labels,
+                            String logLevelLabel,
+                            String logLevel,
                             ByteBuffer line) {
-        int requiredSize = calculateRequiredSizeAscii(labels, line);
+        int requiredSize = serializer.calculateRequiredSizeAscii(
+                labels,
+                logLevelLabel,
+                logLevel,
+                line
+        );
+
         int claim = logBuffer.tryClaim(LOG_MESSAGE_TYPE_ID, requiredSize);
+        try {
+            serializer
+                    .writeTo(
+                            claim,
+                            timestamp,
+                            labels,
+                            logLevelLabel,
+                            logLevel,
+                            line
+                    );
 
-        int cursor = claim;
-        if (requiredSize > 0) {
-            try {
-                AtomicBuffer buffer = this.logBuffer.buffer();
-
-                buffer.putLong(cursor, timestamp);
-                cursor += Long.BYTES;
-
-                buffer.putInt(cursor, labels.size());
-                cursor += Integer.BYTES;
-
-                for (Map.Entry<String, String> entry : labels.entrySet()) {
-                    String key = entry.getKey();
-                    String value = entry.getValue();
-
-                    cursor += buffer.putStringAscii(cursor, key);
-                    cursor += buffer.putStringAscii(cursor, value);
-                }
-
-                buffer.putInt(cursor, line.remaining());
-                cursor += Integer.BYTES;
-                buffer.putBytes(
-                        cursor,
-                        line,
-                        line.remaining()
-                );
-                logBuffer.commit(claim);
-            } catch (Throwable t) {
-                logBuffer.abort(claim);
-            }
+            logBuffer.commit(claim);
+        } catch (Throwable t) {
+            logBuffer.abort(claim);
         }
 
         return this;
     }
-
-    private int calculateRequiredSizeAscii(Map<String, String> labels) {
-        return Long.BYTES +
-                Integer.BYTES +
-                labels
-                        .entrySet()
-                        .stream()
-                        .mapToInt(entry -> entry.getKey().length() + entry.getValue().length())
-                        .map(size -> size + 2 * Integer.BYTES)
-                        .sum();
-    }
-
-    private int calculateRequiredSizeAscii(
-            Map<String, String> labels,
-            String line
-    ) {
-        return calculateRequiredSizeAscii(labels) +
-                line.length() + Integer.BYTES;
-    }
-
-    private int calculateRequiredSizeAscii(
-            Map<String, String> labels,
-            ByteBuffer line
-    ) {
-        return calculateRequiredSizeAscii(labels) +
-                line.remaining() + Integer.BYTES;
-    }
 }

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java

@@ -9,6 +9,7 @@ import pl.tkowalcz.tjahzi.http.ClientConfiguration;
 import pl.tkowalcz.tjahzi.http.HttpClientFactory;
 import pl.tkowalcz.tjahzi.http.NettyHttpClient;
 
+import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
@@ -84,7 +85,9 @@ class HeadersTest {
         logger.log(
                 System.currentTimeMillis(),
                 Map.of(),
-                "Test"
+                "level",
+                "warn",
+                ByteBuffer.wrap("Test".getBytes())
         );
 
         // Then
@@ -129,7 +132,9 @@ class HeadersTest {
         logger.log(
                 System.currentTimeMillis(),
                 Map.of(),
-                "Test"
+                "level",
+                "warn",
+                ByteBuffer.wrap("Test".getBytes())
         );
 
         // Then
@@ -145,7 +150,7 @@ class HeadersTest {
     void shouldNotOverrideCrucialHeaders() {
         // Given
         wireMockServer.stubFor(
-                get(urlEqualTo("/loki/api/v1/push"))
+                post(urlEqualTo("/loki/api/v1/push"))
                         .willReturn(
                                 aResponse().withStatus(200)
                         ));
@@ -176,7 +181,9 @@ class HeadersTest {
         logger.log(
                 System.currentTimeMillis(),
                 Map.of(),
-                "Test"
+                "level",
+                "warn",
+                ByteBuffer.wrap("Test".getBytes())
         );
 
         // Then
@@ -186,7 +193,7 @@ class HeadersTest {
                         wireMockServer.verify(
                                 postRequestedFor(urlMatching("/loki/api/v1/push"))
                                         .withHeader("content-type", matching("application/x-protobuf"))
-                                        .withHeader("content-length", matching("29|30"))
+                                        .withHeader("content-length", matching("42|43"))
                                         .withHeader("host", matching("localhost"))
                         ));
     }

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java

@@ -17,6 +17,7 @@ import pl.tkowalcz.tjahzi.http.ClientConfiguration;
 import pl.tkowalcz.tjahzi.http.HttpClientFactory;
 import pl.tkowalcz.tjahzi.http.NettyHttpClient;
 
+import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
@@ -51,10 +52,7 @@ class LoggingSystemSanityCheckTest {
                 .build();
 
         NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
-                .getHttpClient(
-                        clientConfiguration,
-                        new String[0]
-                );
+                .getHttpClient(clientConfiguration);
 
         TjahziInitializer initializer = new TjahziInitializer();
         loggingSystem = initializer.createLoggingSystem(
@@ -84,7 +82,9 @@ class LoggingSystemSanityCheckTest {
         logger.log(
                 timestamp,
                 Map.of("version", "0.43", "server", "127.0.0.1"),
-                "Test"
+                "level",
+                "warn",
+                ByteBuffer.wrap("Test".getBytes())
         );
 
         // Then
@@ -110,6 +110,7 @@ class LoggingSystemSanityCheckTest {
                             .body("data.result.size()", equalTo(1))
                             .body("data.result[0].stream.server", equalTo("127.0.0.1"))
                             .body("data.result[0].stream.version", equalTo("0.43"))
+                            .body("data.result[0].stream.level", equalTo("warn"))
                             .body("data.result[0].values[0]", hasItems("" + (timestamp * 1000_000), "Test"));
                 });
     }

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java

@@ -19,6 +19,7 @@ import pl.tkowalcz.tjahzi.http.ClientConfiguration;
 import pl.tkowalcz.tjahzi.http.HttpClientFactory;
 import pl.tkowalcz.tjahzi.http.NettyHttpClient;
 
+import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -87,7 +88,9 @@ class LoggingSystemTest {
             logger.log(
                     timestamp + i,
                     Map.of("version", "0.43", "server", "127.0.0.1"),
-                    "Test" + i
+                    "level",
+                    "warn",
+                    ByteBuffer.wrap(("Test" + i).getBytes())
             );
         }
 

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java

@@ -9,6 +9,7 @@ import pl.tkowalcz.tjahzi.http.ClientConfiguration;
 import pl.tkowalcz.tjahzi.http.HttpClientFactory;
 import pl.tkowalcz.tjahzi.http.NettyHttpClient;
 
+import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
@@ -77,7 +78,9 @@ public class ReconnectTest {
         logger.log(
                 System.currentTimeMillis(),
                 Map.of(),
-                "Test"
+                "level",
+                "warn",
+                ByteBuffer.wrap("Test".getBytes())
         );
 
         wireMockServer.start();
@@ -117,7 +120,9 @@ public class ReconnectTest {
         logger.log(
                 System.currentTimeMillis(),
                 Map.of(),
-                "Test"
+                "level",
+                "warn",
+                ByteBuffer.wrap("Test".getBytes())
         );
 
         await()
@@ -136,7 +141,9 @@ public class ReconnectTest {
         logger.log(
                 System.currentTimeMillis(),
                 Map.of(),
-                "Test"
+                "level",
+                "warn",
+                ByteBuffer.wrap("Test".getBytes())
         );
 
         wireMockServer.start();