Merge pull request #19 from tkowalcz/17-labels-skip-log-buffer

Re #17: Moved static set of labels to other side of log buffer - no n…

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

@@ -8,7 +8,6 @@ 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;
@@ -20,18 +19,13 @@ public class AppenderLogic implements BiConsumer<LogEvent, ByteBuffer> {
     private final LoggingSystem loggingSystem;
 
     private final String logLevelLabel;
-    private final Map<String, String> lokiLabels;
-
     private final TjahziLogger logger;
 
     public AppenderLogic(
             LoggingSystem loggingSystem,
-            String logLevelLabel,
-            Map<String, String> lokiLabels
+            String logLevelLabel
     ) {
         this.loggingSystem = loggingSystem;
-
-        this.lokiLabels = Collections.unmodifiableMap(lokiLabels);
         this.logLevelLabel = logLevelLabel;
 
         logger = loggingSystem.createLogger();
@@ -61,7 +55,7 @@ public class AppenderLogic implements BiConsumer<LogEvent, ByteBuffer> {
 
         logger.log(
                 event.getTimeMillis(),
-                lokiLabels,
+                Map.of(),
                 logLevelLabel,
                 logLevel,
                 byteBuffer

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

@@ -22,7 +22,6 @@ import pl.tkowalcz.tjahzi.http.NettyHttpClient;
 
 import java.io.Serializable;
 import java.util.HashMap;
-import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
@@ -106,15 +105,16 @@ public class LokiAppender extends AbstractAppender {
                 );
             }
 
+            HashMap<String, String> lokiLabels = Maps.newHashMap();
+            stream(labels).forEach(__ -> lokiLabels.put(__.getName(), __.getValue()));
+
             LoggingSystem loggingSystem = new TjahziInitializer().createLoggingSystem(
                     httpClient,
+                    lokiLabels,
                     bufferSizeBytes,
                     isUseOffHeapBuffer()
             );
 
-            HashMap<String, String> lokiLabels = Maps.newHashMap();
-            stream(labels).forEach(__ -> lokiLabels.put(__.getName(), __.getValue()));
-
             return new LokiAppender(
                     getName(),
                     getLayout(),
@@ -122,7 +122,6 @@ public class LokiAppender extends AbstractAppender {
                     isIgnoreExceptions(),
                     getPropertyArray(),
                     logLevelLabel,
-                    lokiLabels,
                     loggingSystem
             );
         }
@@ -229,7 +228,6 @@ public class LokiAppender extends AbstractAppender {
             boolean ignoreExceptions,
             Property[] properties,
             String logLevelLabel,
-            Map<String, String> lokiLabels,
             LoggingSystem loggingSystem) {
         super(
                 name,
@@ -243,8 +241,7 @@ public class LokiAppender extends AbstractAppender {
         this.loggingSystem = loggingSystem;
         this.appenderLogic = new AppenderLogic(
                 loggingSystem,
-                logLevelLabel,
-                lokiLabels
+                logLevelLabel
         );
     }
 
@@ -254,6 +251,7 @@ public class LokiAppender extends AbstractAppender {
 
     @Override
     public void start() {
+        loggingSystem.start();
         super.start();
     }
 

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

@@ -1,20 +1,16 @@
 package pl.tkowalcz.tjahzi;
 
-import com.google.protobuf.Timestamp;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufOutputStream;
 import io.netty.buffer.PooledByteBufAllocator;
-import javolution.text.TextBuilder;
 import logproto.Logproto;
 import org.agrona.MutableDirectBuffer;
 import org.agrona.concurrent.Agent;
 import org.agrona.concurrent.MessageHandler;
 import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
 import pl.tkowalcz.tjahzi.http.NettyHttpClient;
-import pl.tkowalcz.tjahzi.http.TextBuilders;
 
 import java.io.IOException;
-import java.util.HashMap;
 import java.util.Map;
 
 public class LogBufferAgent implements Agent, MessageHandler {
@@ -24,14 +20,20 @@ public class LogBufferAgent implements Agent, MessageHandler {
     private final ManyToOneRingBuffer logBuffer;
     private final NettyHttpClient httpClient;
 
+    private final Map<String, String> staticLabels;
+    private final LogBufferDeserializer logBufferDeserializer = new LogBufferDeserializer();
+
     private Logproto.PushRequest.Builder request = Logproto.PushRequest.newBuilder();
 
     public LogBufferAgent(
             ManyToOneRingBuffer logBuffer,
-            NettyHttpClient httpClient
+            NettyHttpClient httpClient,
+            Map<String, String> staticLabels
     ) {
         this.logBuffer = logBuffer;
         this.httpClient = httpClient;
+
+        this.staticLabels = staticLabels;
     }
 
     @Override
@@ -44,10 +46,6 @@ public class LogBufferAgent implements Agent, MessageHandler {
                     new ByteBufOutputStream(buffer)
             );
 
-//            StringBuilder dump = new StringBuilder();
-//            ByteBufUtil.appendPrettyHexDump(dump, buffer.duplicate());
-//            System.out.println(dump);
-
             httpClient.log(buffer);
             request = Logproto.PushRequest.newBuilder();
         }
@@ -56,7 +54,12 @@ public class LogBufferAgent implements Agent, MessageHandler {
     }
 
     @Override
-    public void onMessage(int msgTypeId, MutableDirectBuffer buffer, int index, int length) {
+    public void onMessage(
+            int msgTypeId,
+            MutableDirectBuffer buffer,
+            int index,
+            int length
+    ) {
         if (msgTypeId == TjahziLogger.LOG_MESSAGE_TYPE_ID) {
             processMessage(buffer, index);
         } else {
@@ -65,52 +68,15 @@ public class LogBufferAgent implements Agent, MessageHandler {
     }
 
     private void processMessage(MutableDirectBuffer buffer, int index) {
-        long timestamp = buffer.getLong(index);
-        index += Long.BYTES;
-
-        int labelsCount = buffer.getInt(index);
-        index += Integer.BYTES;
-
-        Map<String, String> labels = new HashMap<>();
-        for (int i = 0; i < labelsCount; i++) {
-            String key = buffer.getStringAscii(index);
-            index += key.length() + Integer.BYTES;
-
-            String value = buffer.getStringAscii(index);
-            index += value.length() + Integer.BYTES;
-
-            labels.put(key, value);
-        }
-
-        String line = buffer.getStringAscii(index);
-
-        Logproto.StreamAdapter stream = Logproto.StreamAdapter.newBuilder()
-                .setLabels(buildLabelsString(labels).toString())
-                .addEntries(Logproto.EntryAdapter.newBuilder()
-                        .setTimestamp(Timestamp.newBuilder()
-                                .setSeconds(timestamp / 1000)
-                                .setNanos((int) (timestamp % 1000) * 1000_000))
-                        .setLine(line)
-                )
-                .build();
+        Logproto.StreamAdapter stream = logBufferDeserializer.deserialize(
+                buffer,
+                index,
+                staticLabels
+        );
 
         request.addStreams(stream);
     }
 
-    private CharSequence buildLabelsString(Map<String, String> labels) {
-        TextBuilder textBuilder = TextBuilders.threadLocal();
-
-        textBuilder.append("{ ");
-        labels.forEach((key, value) -> textBuilder.append(key)
-                .append("=")
-                .append("\"")
-                .append(value)
-                .append("\","));
-
-        textBuilder.setCharAt(textBuilder.length() - 1, '}');
-        return textBuilder;
-    }
-
     @Override
     public String roleName() {
         return "ReadingLogBufferAndSendingHttp";

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

@@ -0,0 +1,107 @@
+package pl.tkowalcz.tjahzi;
+
+import com.google.protobuf.Timestamp;
+import javolution.text.TextBuilder;
+import logproto.Logproto;
+import org.agrona.DirectBuffer;
+import pl.tkowalcz.tjahzi.http.TextBuilders;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public class LogBufferDeserializer {
+
+    public Logproto.StreamAdapter deserialize(
+            DirectBuffer buffer,
+            int index,
+            Map<String, String> staticLabels
+    ) {
+        long timestamp = buffer.getLong(index);
+        index += Long.BYTES;
+
+        Map<String, String> labels = new LinkedHashMap<>();
+        index = readLabels(
+                buffer,
+                index,
+                labels
+        );
+
+        String line = buffer.getStringAscii(index);
+
+        return createGrpcStream(
+                staticLabels,
+                timestamp,
+                labels,
+                line
+        );
+    }
+
+    private int readLabels(
+            DirectBuffer buffer,
+            int index,
+            Map<String, String> labels
+    ) {
+        int labelsCount = buffer.getInt(index);
+        index += Integer.BYTES;
+
+        for (int i = 0; i < labelsCount; i++) {
+            String key = buffer.getStringAscii(index);
+            index += key.length() + Integer.BYTES;
+
+            String value = buffer.getStringAscii(index);
+            index += value.length() + Integer.BYTES;
+
+            labels.put(key, value);
+        }
+
+        return index;
+    }
+
+    private Logproto.StreamAdapter createGrpcStream(
+            Map<String, String> staticLabels,
+            long timestamp,
+            Map<String, String> labels,
+            String line
+    ) {
+        long timestampSeconds = timestamp / 1000;
+        int timestampNanos = (int) (timestamp % 1000) * 1000_000;
+
+        return Logproto.StreamAdapter.newBuilder()
+                .setLabels(
+                        buildLabelsStringIncludingStatic(labels, staticLabels)
+                )
+                .addEntries(
+                        Logproto.EntryAdapter.newBuilder()
+                                .setTimestamp(
+                                        Timestamp.newBuilder()
+                                                .setSeconds(timestampSeconds)
+                                                .setNanos(timestampNanos)
+                                )
+                                .setLine(line)
+                )
+                .build();
+    }
+
+    private static String buildLabelsStringIncludingStatic(
+            Map<String, String> staticLabels,
+            Map<String, String> labels) {
+        TextBuilder textBuilder = TextBuilders.threadLocal();
+
+        textBuilder.append("{ ");
+        staticLabels.forEach((key, value) -> textBuilder.append(key)
+                .append("=")
+                .append("\"")
+                .append(value)
+                .append("\","));
+
+        labels.forEach((key, value) -> textBuilder.append(key)
+                .append("=")
+                .append("\"")
+                .append(value)
+                .append("\","));
+
+
+        textBuilder.setCharAt(textBuilder.length() - 1, '}');
+        return textBuilder.toString();
+    }
+}

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

@@ -26,6 +26,10 @@ public class LoggingSystem {
         return new TjahziLogger(logBuffer);
     }
 
+    public void start() {
+        AgentRunner.startOnThread(runner);
+    }
+
     public void close(
             int retryCloseTimeoutMs,
             Consumer<Thread> closeFailAction) {

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

@@ -8,6 +8,7 @@ import org.agrona.concurrent.ringbuffer.RingBufferDescriptor;
 import pl.tkowalcz.tjahzi.http.NettyHttpClient;
 
 import java.nio.ByteBuffer;
+import java.util.Map;
 
 public class TjahziInitializer {
 
@@ -15,6 +16,7 @@ public class TjahziInitializer {
 
     public LoggingSystem createLoggingSystem(
             NettyHttpClient httpClient,
+            Map<String, String> staticLabels,
             int bufferSizeBytes,
             boolean offHeap) {
         bufferSizeBytes = findNearestPowerOfTwo(bufferSizeBytes);
@@ -26,7 +28,8 @@ public class TjahziInitializer {
 
         LogBufferAgent agent = new LogBufferAgent(
                 logBuffer,
-                httpClient
+                httpClient,
+                staticLabels
         );
 
         AgentRunner runner = new AgentRunner(
@@ -36,8 +39,6 @@ public class TjahziInitializer {
                 agent
         );
 
-        AgentRunner.startOnThread(runner);
-
         return new LoggingSystem(
                 logBuffer,
                 runner,

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

@@ -76,9 +76,11 @@ class HeadersTest {
 
         loggingSystem = initializer.createLoggingSystem(
                 httpClient,
+                Map.of(),
                 1024 * 1024,
                 false
         );
+        loggingSystem.start();
 
         // WHen
         TjahziLogger logger = loggingSystem.createLogger();
@@ -123,9 +125,11 @@ class HeadersTest {
 
         loggingSystem = initializer.createLoggingSystem(
                 httpClient,
+                Map.of(),
                 1024 * 1024,
                 false
         );
+        loggingSystem.start();
 
         // WHen
         TjahziLogger logger = loggingSystem.createLogger();
@@ -172,9 +176,11 @@ class HeadersTest {
 
         loggingSystem = initializer.createLoggingSystem(
                 httpClient,
+                Map.of(),
                 1024 * 1024,
                 false
         );
+        loggingSystem.start();
 
         // When
         TjahziLogger logger = loggingSystem.createLogger();

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

@@ -0,0 +1,78 @@
+package pl.tkowalcz.tjahzi;
+
+import com.google.common.collect.Streams;
+import com.google.protobuf.Timestamp;
+import logproto.Logproto;
+import org.agrona.concurrent.UnsafeBuffer;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static java.util.stream.Collectors.joining;
+import static org.assertj.core.api.Assertions.assertThat;
+import static pl.tkowalcz.tjahzi.LogBufferSerializerDeserializerTest.toPropertyStream;
+
+class LogBufferDeserializerTest {
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("pl.tkowalcz.tjahzi.LogBufferSerializerDeserializerTest#variousMessageConfigurations")
+    void shouldDeserializeMessageAndAddStaticLabels(
+            String testName,
+            Map<String, String> labels,
+            String logLevelLabel,
+            String logLevel,
+            String logLine,
+            int expectedSize) {
+        // Given
+
+        UnsafeBuffer buffer = new UnsafeBuffer(new byte[expectedSize]);
+        LogBufferSerializer serializer = new LogBufferSerializer(buffer);
+        serializer.writeTo(
+                0,
+                32042L,
+                labels,
+                logLevelLabel,
+                logLevel,
+                ByteBuffer.wrap(logLine.getBytes())
+        );
+
+        Map<String, String> staticLabels = Map.of(
+                "foo", "bar",
+                "bazz", "buzz"
+        );
+
+        // When
+        LogBufferDeserializer deserializer = new LogBufferDeserializer();
+        Logproto.StreamAdapter stream = deserializer.deserialize(
+                buffer,
+                0,
+                staticLabels
+        );
+
+        // Then
+        assertThat(stream.getEntriesList()).hasSize(1);
+        assertThat(stream.getEntriesList().get(0).getTimestamp()).isEqualTo(
+                Timestamp.newBuilder()
+                        .setSeconds(32)
+                        .setNanos(42_000_000)
+                        .build()
+        );
+        assertThat(stream.getEntriesList().get(0).getLine()).isEqualTo(logLine);
+
+        Stream<String> logLevelStream = logLevelLabel == null ? Stream.of() : toPropertyStream(Map.of(logLevelLabel, logLevel));
+        Stream<String> incomingLabelsStream = toPropertyStream(labels);
+        Stream<String> staticLabelsStream = toPropertyStream(staticLabels);
+
+        assertThat(stream.getLabels()).isEqualToIgnoringWhitespace(
+                Streams.concat(
+                        incomingLabelsStream,
+                        logLevelStream,
+                        staticLabelsStream
+                )
+                        .collect(joining(",", "{", "}"))
+        );
+    }
+}

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

@@ -0,0 +1,120 @@
+package pl.tkowalcz.tjahzi;
+
+import com.google.common.collect.Streams;
+import com.google.protobuf.Timestamp;
+import logproto.Logproto;
+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 java.util.stream.Collectors.joining;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class LogBufferSerializerDeserializerTest {
+
+    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",
+                        "[Mando] You have something I want.",
+                        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",
+                        "[Mando] You have something I want.",
+                        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,
+                        "[Mando] You have something I want.",
+                        38 // log line
+                                + 45 // labels string sizes
+                                + 6 * 4 // labels string length indicators
+                                + 8 // timestamp
+                                + 4 // labels count
+                )
+        );
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("variousMessageConfigurations")
+    void shouldSerializeMessage(
+            String testName,
+            Map<String, String> labels,
+            String logLevelLabel,
+            String logLevel,
+            String logLine,
+            int expectedSize) {
+        // Given
+        UnsafeBuffer buffer = new UnsafeBuffer(new byte[expectedSize]);
+        LogBufferSerializer serializer = new LogBufferSerializer(buffer);
+
+        // When
+        serializer.writeTo(
+                0,
+                32042L,
+                labels,
+                logLevelLabel,
+                logLevel,
+                ByteBuffer.wrap(logLine.getBytes())
+        );
+
+        // Then
+        LogBufferDeserializer deserializer = new LogBufferDeserializer();
+        Logproto.StreamAdapter stream = deserializer.deserialize(buffer, 0, Map.of());
+
+        assertThat(stream.getEntriesList()).hasSize(1);
+        assertThat(stream.getEntriesList().get(0).getTimestamp()).isEqualTo(
+                Timestamp.newBuilder()
+                        .setSeconds(32)
+                        .setNanos(42_000_000)
+                        .build()
+        );
+        assertThat(stream.getEntriesList().get(0).getLine()).isEqualTo(logLine);
+
+        Stream<String> logLevelStream = logLevelLabel == null ? Stream.of() : toPropertyStream(Map.of(logLevelLabel, logLevel));
+        Stream<String> incomingLabelsStream = toPropertyStream(labels);
+
+        assertThat(stream.getLabels()).isEqualToIgnoringWhitespace(
+                Streams.concat(
+                        incomingLabelsStream,
+                        logLevelStream
+                )
+                        .collect(joining(",", "{", "}"))
+        );
+    }
+
+    public static Stream<String> toPropertyStream(Map<String, String> labels) {
+        return labels.entrySet()
+                .stream()
+                .map(entry -> entry.getKey() + "=\"" + entry.getValue() + "\"");
+    }
+}

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

@@ -2,72 +2,23 @@ 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")
+    @MethodSource("pl.tkowalcz.tjahzi.LogBufferSerializerDeserializerTest#variousMessageConfigurations")
     void shouldCalculateSizeOfMessage(
             String testName,
             Map<String, String> labels,
             String logLevelLabel,
             String logLevel,
-            ByteBuffer logLine,
+            String logLine,
             int expectedSize) {
         // Given
         UnsafeBuffer buffer = new UnsafeBuffer(new byte[expectedSize]);
@@ -78,38 +29,10 @@ class LogBufferSerializerTest {
                 labels,
                 logLevelLabel,
                 logLevel,
-                logLine
+                ByteBuffer.wrap(logLine.getBytes())
         );
 
         // 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

@@ -57,9 +57,12 @@ class LoggingSystemSanityCheckTest {
         TjahziInitializer initializer = new TjahziInitializer();
         loggingSystem = initializer.createLoggingSystem(
                 httpClient,
+                Map.of("version", "0.43", "server", "127.0.0.1"),
                 1024 * 1024,
                 false
         );
+
+        loggingSystem.start();
     }
 
     @AfterEach
@@ -81,7 +84,7 @@ class LoggingSystemSanityCheckTest {
         // When
         logger.log(
                 timestamp,
-                Map.of("version", "0.43", "server", "127.0.0.1"),
+                Map.of(),
                 "level",
                 "warn",
                 ByteBuffer.wrap("Test".getBytes())

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

@@ -62,9 +62,12 @@ class LoggingSystemTest {
         TjahziInitializer initializer = new TjahziInitializer();
         loggingSystem = initializer.createLoggingSystem(
                 httpClient,
+                Map.of("version", "0.43", "server", "127.0.0.1"),
                 1024 * 1024,
                 false
         );
+
+        loggingSystem.start();
     }
 
     @AfterEach
@@ -87,7 +90,7 @@ class LoggingSystemTest {
         for (int i = 0; i < 1000; i++) {
             logger.log(
                     timestamp + i,
-                    Map.of("version", "0.43", "server", "127.0.0.1"),
+                    Map.of(),
                     "level",
                     "warn",
                     ByteBuffer.wrap(("Test" + i).getBytes())
@@ -119,6 +122,7 @@ class LoggingSystemTest {
                             .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", hasItems(new BaseMatcher<>() {
 
                                 int index = 999;

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

@@ -69,9 +69,11 @@ public class ReconnectTest {
 
         loggingSystem = initializer.createLoggingSystem(
                 httpClient,
+                Map.of(),
                 1024 * 1024,
                 false
         );
+        loggingSystem.start();
 
         // When
         TjahziLogger logger = loggingSystem.createLogger();
@@ -112,9 +114,11 @@ public class ReconnectTest {
 
         loggingSystem = initializer.createLoggingSystem(
                 httpClient,
+                Map.of(),
                 1024 * 1024,
                 false
         );
+        loggingSystem.start();
 
         TjahziLogger logger = loggingSystem.createLogger();
         logger.log(

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

@@ -13,6 +13,7 @@ import pl.tkowalcz.tjahzi.http.NettyHttpClient;
 import java.lang.management.ManagementFactory;
 import java.lang.management.ThreadInfo;
 import java.lang.management.ThreadMXBean;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
@@ -60,9 +61,11 @@ class ResourcesCleanupOnCloseTest {
 
         LoggingSystem loggingSystem = initializer.createLoggingSystem(
                 httpClient,
+                Map.of(),
                 1024 * 1024,
                 false
         );
+        loggingSystem.start();
 
         // Verify our assumptions that we can find threads started by Tjahzi
         Awaitility.await().untilAsserted(() -> {