Merge pull request #16 from tkowalcz/9-log-level-label

9 log level label

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

loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/BufferSizeConfigurationTest.java

@@ -3,14 +3,12 @@ package pl.tkowalcz.tjahzi.log4j2;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.core.LoggerContext;
 import org.junit.jupiter.api.Test;
-import org.testcontainers.junit.jupiter.Testcontainers;
 
 import java.net.URI;
 import java.net.URISyntaxException;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-@Testcontainers
 class BufferSizeConfigurationTest {
 
     @Test

loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LogLevelLabelConfigurationTest.java

@@ -0,0 +1,130 @@
+package pl.tkowalcz.tjahzi.log4j2;
+
+import io.restassured.RestAssured;
+import io.restassured.http.ContentType;
+import io.restassured.parsing.Parser;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.awaitility.Awaitility;
+import org.awaitility.Durations;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.BindMode;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+import static io.restassured.RestAssured.given;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.nullValue;
+
+@Testcontainers
+class LogLevelLabelConfigurationTest {
+
+    @Container
+    public GenericContainer loki = new GenericContainer("grafana/loki:latest")
+            .withCommand("-config.file=/etc/loki-config.yaml")
+            .withClasspathResourceMapping("loki-config.yaml",
+                    "/etc/loki-config.yaml",
+                    BindMode.READ_ONLY)
+            .waitingFor(
+                    Wait.forHttp("/ready")
+                            .forPort(3100)
+            )
+            .withExposedPorts(3100);
+
+    @Test
+    void shouldWorkWIthNoLogLevelConfigured() throws URISyntaxException {
+        // Given
+        System.setProperty("loki.host", loki.getHost());
+        System.setProperty("loki.port", loki.getFirstMappedPort().toString());
+
+        URI uri = getClass()
+                .getClassLoader()
+                .getResource("appender-test-with-log-label-unset.xml")
+                .toURI();
+
+        ((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false))
+                .setConfigLocation(uri);
+
+        // When
+        Logger logger = LogManager.getLogger(LokiAppenderTest.class);
+        logger.info("Test");
+
+        // Then
+        RestAssured.port = loki.getFirstMappedPort();
+        RestAssured.baseURI = "http://" + loki.getHost();
+        RestAssured.registerParser("text/plain", Parser.JSON);
+
+        Awaitility
+                .await()
+                .atMost(Durations.TEN_SECONDS)
+                .pollInterval(Durations.ONE_SECOND)
+                .ignoreExceptions()
+                .untilAsserted(() -> {
+                    given()
+                            .contentType(ContentType.URLENC)
+                            .urlEncodingEnabled(false)
+                            .formParam("query=%7Bserver%3D%22127.0.0.1%22%7D")
+                            .when()
+                            .get("/loki/api/v1/query_range")
+                            .then()
+                            .log()
+                            .all()
+                            .statusCode(200)
+                            .body("status", equalTo("success"))
+                            .body("data.result.size()", equalTo(1))
+                            .body("data.result[0].stream.server", equalTo("127.0.0.1"))
+                            .body("data.result[0].stream.log_level", nullValue());
+                });
+    }
+
+    @Test
+    void shouldSendLogLevelAsConfigured() throws URISyntaxException {
+        // Given
+        System.setProperty("loki.host", loki.getHost());
+        System.setProperty("loki.port", loki.getFirstMappedPort().toString());
+
+        URI uri = getClass()
+                .getClassLoader()
+                .getResource("appender-test-with-log-label-set.xml")
+                .toURI();
+
+        ((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false))
+                .setConfigLocation(uri);
+
+        // When
+        Logger logger = LogManager.getLogger(LokiAppenderTest.class);
+        logger.info("Test");
+
+        // Then
+        RestAssured.port = loki.getFirstMappedPort();
+        RestAssured.baseURI = "http://" + loki.getHost();
+        RestAssured.registerParser("text/plain", Parser.JSON);
+
+        Awaitility
+                .await()
+                .atMost(Durations.TEN_SECONDS)
+                .pollInterval(Durations.ONE_SECOND)
+                .ignoreExceptions()
+                .untilAsserted(() -> {
+                    given()
+                            .contentType(ContentType.URLENC)
+                            .urlEncodingEnabled(false)
+                            .formParam("query=%7Bserver%3D%22127.0.0.1%22%7D")
+                            .when()
+                            .get("/loki/api/v1/query_range")
+                            .then()
+                            .log()
+                            .all()
+                            .statusCode(200)
+                            .body("status", equalTo("success"))
+                            .body("data.result.size()", equalTo(1))
+                            .body("data.result[0].stream.server", equalTo("127.0.0.1"))
+                            .body("data.result[0].stream.log_level", equalTo("INFO"));
+                });
+    }
+}

loki-log4j2-appender/src/test/resources/appender-test-with-log-label-set.xml

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration status="OFF" shutdownHook="disable" packages="pl.tkowalcz.tjahzi.log4j2">
+    <Loggers>
+        <Root level="INFO">
+            <AppenderRef ref="Loki"/>
+        </Root>
+    </Loggers>
+
+    <appenders>
+        <Loki name="Loki">
+            <host>${sys:loki.host}</host>
+            <port>${sys:loki.port}</port>
+
+            <logLevelLabel>log_level</logLevelLabel>
+
+            <ThresholdFilter level="ALL"/>
+            <PatternLayout>
+                <Pattern>%X{tid} [%t] %d{MM-dd HH:mm:ss.SSS} %5p %c{1} - %m%n%exception{full}</Pattern>
+            </PatternLayout>
+
+            <Header name="server" value="127.0.0.1"/>
+            <Label name="server" value="127.0.0.1"/>
+        </Loki>
+    </appenders>
+</configuration>

loki-log4j2-appender/src/test/resources/appender-test-with-log-label-unset.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration status="OFF" shutdownHook="disable" packages="pl.tkowalcz.tjahzi.log4j2">
+    <Loggers>
+        <Root level="INFO">
+            <AppenderRef ref="Loki"/>
+        </Root>
+    </Loggers>
+
+    <appenders>
+        <Loki name="Loki">
+            <host>${sys:loki.host}</host>
+            <port>${sys:loki.port}</port>
+
+            <ThresholdFilter level="ALL"/>
+            <PatternLayout>
+                <Pattern>%X{tid} [%t] %d{MM-dd HH:mm:ss.SSS} %5p %c{1} - %m%n%exception{full}</Pattern>
+            </PatternLayout>
+
+            <Header name="server" value="127.0.0.1"/>
+            <Label name="server" value="127.0.0.1"/>
+        </Loki>
+    </appenders>
+</configuration>

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

@@ -0,0 +1,145 @@
+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 int calculateRequiredSizeAscii(
+            Map<String, String> labels,
+            String logLevelLabel,
+            String logLevel,
+            ByteBuffer line
+    ) {
+        int logLineSize = Integer.BYTES + line.remaining();
+        int logLevelSize = calculateOptionalLabelSizeAscii(logLevelLabel, logLevel);
+        int headerAndLabelsSize = calculateRequiredSizeAscii(labels);
+
+        return headerAndLabelsSize + logLineSize + logLevelSize;
+    }
+
+    public void writeTo(
+            int cursor,
+            long timestamp,
+            Map<String, String> labels,
+            String logLevelLabel,
+            String logLevel,
+            ByteBuffer line
+    ) {
+        cursor = writeHeader(
+                cursor,
+                timestamp,
+                labels,
+                logLevelLabel != null
+        );
+
+        cursor = writeLabels(cursor, labels);
+        cursor = writeOptionalLabel(
+                cursor,
+                logLevelLabel,
+                logLevel
+        );
+
+        writeLogLine(cursor, line);
+    }
+
+    private int writeHeader(
+            int cursor,
+            long timestamp,
+            Map<String, String> labels,
+            boolean hasLogLevelLabel
+    ) {
+        buffer.putLong(cursor, timestamp);
+        cursor += Long.BYTES;
+
+        int labelsCount = labels.size();
+        if (hasLogLevelLabel) {
+            labelsCount++;
+        }
+
+        buffer.putInt(cursor, labelsCount);
+        cursor += Integer.BYTES;
+        return cursor;
+    }
+
+    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 void writeLogLine(int cursor, ByteBuffer line) {
+        buffer.putInt(cursor, line.remaining());
+        cursor += Integer.BYTES;
+
+        buffer.putBytes(
+                cursor,
+                line,
+                line.remaining()
+        );
+    }
+
+    private int writeLabel(
+            int cursor,
+            String key,
+            String value
+    ) {
+        cursor += buffer.putStringAscii(cursor, key);
+        cursor += buffer.putStringAscii(cursor, value);
+
+        return cursor;
+    }
+
+    private int writeOptionalLabel(
+            int cursor,
+            String key,
+            String value
+    ) {
+        if (key != null) {
+            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 += calculateLabelSizeAscii(entry.getKey(), entry.getValue());
+        }
+
+        return sum;
+    }
+
+    private int calculateOptionalLabelSizeAscii(String key, String value) {
+        if (key == null) {
+            return 0;
+        }
+
+        return Integer.BYTES + key.length()
+                + Integer.BYTES + value.length();
+    }
+
+    private int calculateLabelSizeAscii(String key, String value) {
+        return Integer.BYTES + key.length()
+                + Integer.BYTES + value.length();
+    }
+}

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,64 @@ 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);
-
-        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);
-            }
+        if (claim > 0) {
+            putMessageOnRing(
+                    timestamp,
+                    labels,
+                    logLevelLabel,
+                    logLevel,
+                    line,
+                    claim
+            );
+        } else {
+            // TODO: increment dropped lines metric
         }
 
         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(
+    private void putMessageOnRing(
+            long timestamp,
             Map<String, String> labels,
-            String line
+            String logLevelLabel,
+            String logLevel,
+            ByteBuffer line,
+            int claim
     ) {
-        return calculateRequiredSizeAscii(labels) +
-                line.length() + Integer.BYTES;
-    }
+        try {
+            serializer
+                    .writeTo(
+                            claim,
+                            timestamp,
+                            labels,
+                            logLevelLabel,
+                            logLevel,
+                            line
+                    );
 
-    private int calculateRequiredSizeAscii(
-            Map<String, String> labels,
-            ByteBuffer line
-    ) {
-        return calculateRequiredSizeAscii(labels) +
-                line.remaining() + Integer.BYTES;
+            logBuffer.commit(claim);
+        } catch (Throwable t) {
+            logBuffer.abort(claim);
+        }
     }
 }

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

@@ -14,16 +14,15 @@ class ResponseHandler extends SimpleChannelInboundHandler<FullHttpResponse> {
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) {
-        System.out.println("msg = " + msg);
     }
 
     @Override
-    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         System.out.println("cause = " + cause);
     }
 
     @Override
-    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+    public void channelInactive(ChannelHandlerContext ctx) {
         System.out.println("ResponseHandler.channelInactive");
     }
 }

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/LogBufferSerializerTest.java

@@ -0,0 +1,115 @@
+package pl.tkowalcz.tjahzi;
+
+import org.agrona.concurrent.UnsafeBuffer;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class LogBufferSerializerTest {
+
+    private static Stream<Arguments> variousMessageConfigurations() {
+        return Stream.of(
+                Arguments.of(
+                        "Message with all fields filled",
+                        Map.of("ip", "10.0.0.42",
+                                "hostname", "razor-crest",
+                                "region", "outer-rim"
+                        ),
+                        "log_level",
+                        "WARN",
+                        ByteBuffer.wrap("[Mando] You have something I want.".getBytes()),
+                        38 // log line
+                                + 21 // log level
+                                + 45 // labels string sizes
+                                + 6 * 4 // labels string length indicators
+                                + 8 // timestamp
+                                + 4 // labels count
+                ),
+                Arguments.of(
+                        "Message with empty labels",
+                        Map.of(),
+                        "log_level",
+                        "WARN",
+                        ByteBuffer.wrap("[Mando] You have something I want.".getBytes()),
+                        38 // log line
+                                + 21 // log level
+                                + 8 // timestamp
+                                + 4 // labels count
+                ),
+                Arguments.of(
+                        "Message with null log level",
+                        Map.of("ip", "10.0.0.42",
+                                "hostname", "razor-crest",
+                                "region", "outer-rim"
+                        ),
+                        null,
+                        null,
+                        ByteBuffer.wrap("[Mando] You have something I want.".getBytes()),
+                        38 // log line
+                                + 45 // labels string sizes
+                                + 6 * 4 // labels string length indicators
+                                + 8 // timestamp
+                                + 4 // labels count
+                )
+        );
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("variousMessageConfigurations")
+    void shouldCalculateSizeOfMessage(
+            String testName,
+            Map<String, String> labels,
+            String logLevelLabel,
+            String logLevel,
+            ByteBuffer logLine,
+            int expectedSize) {
+        // Given
+        UnsafeBuffer buffer = new UnsafeBuffer(new byte[expectedSize]);
+        LogBufferSerializer serializer = new LogBufferSerializer(buffer);
+
+        // When
+        int actual = serializer.calculateRequiredSizeAscii(
+                labels,
+                logLevelLabel,
+                logLevel,
+                logLine
+        );
+
+        // Then
+        assertThat(actual).isEqualTo(expectedSize);
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("variousMessageConfigurations")
+    void shouldSerializeMessage(
+            String testName,
+            Map<String, String> labels,
+            String logLevelLabel,
+            String logLevel,
+            ByteBuffer logLine,
+            int expectedSize) {
+        // Given
+        UnsafeBuffer buffer = new UnsafeBuffer(new byte[expectedSize]);
+        LogBufferSerializer serializer = new LogBufferSerializer(buffer);
+
+        // When
+        serializer.writeTo(
+                0,
+                42L,
+                labels,
+                logLevelLabel,
+                logLevel,
+                logLine
+        );
+
+        // Then
+        // There was no buffer overflow so seems legit.
+        // TODO: verify contents after extracting deserialization code to separate class.
+    }
+}

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