Cleaned up API and added tests.

core/src/main/java/pl/tkowalcz/thjazi/LogBufferAgent.java

@@ -0,0 +1,118 @@
+package pl.tkowalcz.thjazi;
+
+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.thjazi.http.NettyHttpClient;
+import pl.tkowalcz.thjazi.http.TextBuilders;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class LogBufferAgent implements Agent, MessageHandler {
+
+    public static final int MESSAGE_COUNT_LIMIT = 100;
+
+    private final ManyToOneRingBuffer logBuffer;
+    private final NettyHttpClient httpClient;
+
+    private Logproto.PushRequest.Builder request = Logproto.PushRequest.newBuilder();
+
+    public LogBufferAgent(
+            ManyToOneRingBuffer logBuffer,
+            NettyHttpClient httpClient
+    ) {
+        this.logBuffer = logBuffer;
+        this.httpClient = httpClient;
+    }
+
+    @Override
+    public int doWork() throws IOException {
+        int workDone = logBuffer.read(this, MESSAGE_COUNT_LIMIT);
+
+        if (workDone > 0) {
+            ByteBuf buffer = PooledByteBufAllocator.DEFAULT.buffer();
+            request.build().writeTo(
+                    new ByteBufOutputStream(buffer)
+            );
+
+//            StringBuilder dump = new StringBuilder();
+//            ByteBufUtil.appendPrettyHexDump(dump, buffer.duplicate());
+//            System.out.println(dump);
+
+            httpClient.log(buffer);
+            request = Logproto.PushRequest.newBuilder();
+        }
+
+        return workDone;
+    }
+
+    @Override
+    public void onMessage(int msgTypeId, MutableDirectBuffer buffer, int index, int length) {
+        if (msgTypeId == ThjaziLogger.LOG_MESSAGE_TYPE_ID) {
+            processMessage(buffer, index);
+        } else {
+            // Ignore
+        }
+    }
+
+    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();
+
+        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";
+    }
+}

core/src/main/java/pl/tkowalcz/thjazi/ThjaziInitializer.java

@@ -1,24 +1,44 @@
 package pl.tkowalcz.thjazi;
 
+import org.agrona.concurrent.AgentRunner;
+import org.agrona.concurrent.SleepingMillisIdleStrategy;
 import org.agrona.concurrent.UnsafeBuffer;
 import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
 import org.agrona.concurrent.ringbuffer.RingBufferDescriptor;
+import pl.tkowalcz.thjazi.http.NettyHttpClient;
 
 import java.nio.ByteBuffer;
 
 public class ThjaziInitializer {
 
     public LoggingSystem createLoggingSystem(
+            NettyHttpClient httpClient,
             int bufferSizeBytes,
             boolean offHeap) {
 
         verifyIsPowerOfTwo(bufferSizeBytes);
         ByteBuffer javaBuffer = allocateJavaBuffer(bufferSizeBytes, offHeap);
 
+        ManyToOneRingBuffer logBuffer = new ManyToOneRingBuffer(
+                new UnsafeBuffer(javaBuffer)
+        );
+
+        LogBufferAgent agent = new LogBufferAgent(
+                logBuffer,
+                httpClient
+        );
+
+        AgentRunner runner = new AgentRunner(
+                new SleepingMillisIdleStrategy(),
+                Throwable::printStackTrace,
+                null,
+                agent
+        );
+
+        AgentRunner.startOnThread(runner);
+
         return new LoggingSystem(
-                new ManyToOneRingBuffer(
-                        new UnsafeBuffer(javaBuffer)
-                )
+                logBuffer
         );
     }
 

core/src/main/java/pl/tkowalcz/thjazi/ThjaziLogger.java

@@ -40,7 +40,7 @@ public class ThjaziLogger {
                     cursor += buffer.putStringAscii(cursor, value);
                 }
 
-                claim += buffer.putStringAscii(cursor, line);
+                buffer.putStringAscii(cursor, line);
                 logBuffer.commit(claim);
             } catch (Throwable t) {
                 logBuffer.abort(claim);

core/src/main/java/pl/tkowalcz/thjazi/http/HttpClientFactory.java

@@ -1,16 +1,11 @@
 package pl.tkowalcz.thjazi.http;
 
-import java.time.Clock;
-
 public class HttpClientFactory {
 
     private static final HttpClientFactory INSTANCE = new HttpClientFactory();
 
     public NettyHttpClient getHttpClient(ClientConfiguration clientConfiguration) {
-        return new NettyHttpClient(
-                Clock.systemUTC(),
-                clientConfiguration
-        );
+        return new NettyHttpClient(clientConfiguration);
     }
 
     public static HttpClientFactory defaultFactory() {

core/src/main/java/pl/tkowalcz/thjazi/http/NettyHttpClient.java

@@ -1,24 +1,16 @@
 package pl.tkowalcz.thjazi.http;
 
-import com.google.protobuf.Timestamp;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufOutputStream;
 import io.netty.buffer.PooledByteBufAllocator;
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.compression.Snappy;
 import io.netty.handler.codec.http.*;
-import javolution.text.TextBuilder;
-import logproto.Logproto;
 
-import java.io.IOException;
-import java.time.Clock;
-import java.util.Map;
 import java.util.concurrent.ThreadFactory;
 
 public class NettyHttpClient implements AutoCloseable {
@@ -27,25 +19,19 @@ public class NettyHttpClient implements AutoCloseable {
     private final String logEndpoint;
 
     private final Channel channel;
-    private final Bootstrap bootstrap;
-
-    private final Clock clock;
     private final String host;
-    private final int port;
 
-    public NettyHttpClient(
-            Clock clock,
-            ClientConfiguration clientConfiguration
-    ) {
-        this.clock = clock;
+    private final Snappy snappy = new Snappy();
+
+    public NettyHttpClient(ClientConfiguration clientConfiguration) {
         host = clientConfiguration.getHost();
-        port = clientConfiguration.getPort();
+        logEndpoint = clientConfiguration.getLogEndpoint();
 
         ThreadGroup threadGroup = new ThreadGroup("Thjazi Loki client");
         ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "thjazi-worker");
         group = new NioEventLoopGroup(1, threadFactory);
 
-        bootstrap = new Bootstrap();
+        Bootstrap bootstrap = new Bootstrap();
         channel = bootstrap.group(group)
                 .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                 .option(ChannelOption.SO_KEEPALIVE, true)
@@ -60,41 +46,16 @@ public class NettyHttpClient implements AutoCloseable {
                 .connect()
                 .syncUninterruptibly()
                 .channel();
-
-        logEndpoint = clientConfiguration.getLogEndpoint();
     }
 
-    public void log(
-            long timestamp,
-            Map<String, String> labels,
-            String line
-    ) throws IOException {
-        CharSequence labelsString = buildLabelsString(labels);
-        Logproto.PushRequest pushRequest = Logproto.PushRequest.newBuilder()
-                .addStreams(Logproto.StreamAdapter.newBuilder()
-                        .setLabels(labelsString.toString())
-                        .addEntries(Logproto.EntryAdapter.newBuilder()
-                                .setTimestamp(Timestamp.newBuilder()
-                                        .setSeconds(timestamp / 1000)
-                                        .setNanos((int) (timestamp % 1000) * 1000_000))
-                                .setLine(line)
-                        )
-                )
-                .build();
-
-        ByteBuf buffer = PooledByteBufAllocator.DEFAULT.buffer();
-        pushRequest.writeTo(
-                new ByteBufOutputStream(buffer)
-        );
-
+    public void log(ByteBuf dataBuffer) {
         ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer();
-        Snappy snappy = new Snappy();
-        snappy.encode(buffer, output, buffer.readableBytes());
+        snappy.encode(dataBuffer, output, dataBuffer.readableBytes());
 
         FullHttpRequest request = new DefaultFullHttpRequest(
                 HttpVersion.HTTP_1_1,
                 HttpMethod.POST,
-               logEndpoint,
+                logEndpoint,
                 output
         );
 
@@ -102,23 +63,7 @@ public class NettyHttpClient implements AutoCloseable {
         request.headers().set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes());
         request.headers().set(HttpHeaderNames.HOST, host);
 
-        ChannelFuture channelFuture = channel.writeAndFlush(request);
-        System.out.println(channelFuture.awaitUninterruptibly()
-                .isSuccess());
-    }
-
-    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;
+        channel.writeAndFlush(request);
     }
 
     @Override

core/src/test/java/pl/tkowalcz/thjazi/LoggingSystemSanityCheckTest.java

@@ -0,0 +1,100 @@
+package pl.tkowalcz.thjazi;
+
+import io.restassured.RestAssured;
+import io.restassured.http.ContentType;
+import io.restassured.parsing.Parser;
+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.output.OutputFrame;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import pl.tkowalcz.thjazi.http.ClientConfiguration;
+import pl.tkowalcz.thjazi.http.HttpClientFactory;
+import pl.tkowalcz.thjazi.http.NettyHttpClient;
+
+import java.util.Map;
+import java.util.function.Consumer;
+
+import static io.restassured.RestAssured.given;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.hasItems;
+
+@Testcontainers
+class LoggingSystemSanityCheckTest {
+
+    @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)
+            .withLogConsumer((Consumer<OutputFrame>) outputFrame -> {
+                        if (outputFrame.getBytes() != null) {
+                            System.out.print(new String(outputFrame.getBytes()));
+                        }
+                    }
+            );
+
+    @Test
+    void sendData() {
+        ClientConfiguration clientConfiguration = ClientConfiguration.builder()
+                .withConnectionTimeoutMillis(10_000)
+                .withHost(loki.getHost())
+                .withPort(loki.getFirstMappedPort())
+                .withMaxRetries(1)
+                .build();
+
+        NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
+                .getHttpClient(clientConfiguration);
+
+        ThjaziInitializer initializer = new ThjaziInitializer();
+        LoggingSystem loggingSystem = initializer.createLoggingSystem(
+                httpClient,
+                1024 * 1024,
+                false
+        );
+
+        long timestamp = System.currentTimeMillis();
+
+        ThjaziLogger logger = loggingSystem.createLogger();
+        logger.log(
+                timestamp,
+                Map.of("version", "0.43", "server", "127.0.0.1"),
+                "Test"
+        );
+
+        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()
+                            .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.version", equalTo("0.43"))
+                            .body("data.result[0].values[0]", hasItems("" + (timestamp * 1000_000), "Test"));
+                });
+    }
+}

core/src/test/java/pl/tkowalcz/thjazi/LoggingSystemTest.java

@@ -3,9 +3,14 @@ package pl.tkowalcz.thjazi;
 import io.restassured.RestAssured;
 import io.restassured.http.ContentType;
 import io.restassured.parsing.Parser;
+import org.awaitility.Awaitility;
+import org.awaitility.Durations;
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.BindMode;
 import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.OutputFrame;
 import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
@@ -13,9 +18,9 @@ import pl.tkowalcz.thjazi.http.ClientConfiguration;
 import pl.tkowalcz.thjazi.http.HttpClientFactory;
 import pl.tkowalcz.thjazi.http.NettyHttpClient;
 
-import java.io.IOException;
-import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.function.Consumer;
 
 import static io.restassured.RestAssured.given;
 import static org.hamcrest.CoreMatchers.equalTo;
@@ -34,46 +39,16 @@ class LoggingSystemTest {
                     Wait.forHttp("/ready")
                             .forPort(3100)
             )
-            .withExposedPorts(3100);
-//            .withLogConsumer((Consumer<OutputFrame>) outputFrame -> {
-//                        if (outputFrame.getBytes() != null) {
-//                            System.out.print(new String(outputFrame.getBytes()));
-//                        }
-//                    }
-//            );
-
-//    public static void main(String[] args) throws IOException {
-//        ClientConfiguration clientConfiguration = ClientConfiguration.builder()
-//                .withConnectionTimeoutMillis(10_000)
-//                .withHost("localhost")
-//                .withPort(3100)
-//                .withMaxRetries(1)
-//                .build();
-//
-//        NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
-//                .getHttpClient(clientConfiguration);
-//
-//        httpClient.log(
-//                System.currentTimeMillis(),
-//                "Test"
-//        );
-//    }
+            .withExposedPorts(3100)
+            .withLogConsumer((Consumer<OutputFrame>) outputFrame -> {
+                        if (outputFrame.getBytes() != null) {
+                            System.out.print(new String(outputFrame.getBytes()));
+                        }
+                    }
+            );
 
     @Test
-    void sendData() throws IOException {
-        ThjaziInitializer initializer = new ThjaziInitializer();
-        LoggingSystem loggingSystem = initializer.createLoggingSystem(
-                1024 * 1024,
-                false
-        );
-
-        ThjaziLogger logger = loggingSystem.createLogger();
-        logger.log(
-                System.currentTimeMillis(),
-                new HashMap<>(),
-                "Test"
-        );
-
+    void sendData() {
         ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                 .withConnectionTimeoutMillis(10_000)
                 .withHost(loki.getHost())
@@ -84,38 +59,76 @@ class LoggingSystemTest {
         NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
                 .getHttpClient(clientConfiguration);
 
+        ThjaziInitializer initializer = new ThjaziInitializer();
+        LoggingSystem loggingSystem = initializer.createLoggingSystem(
+                httpClient,
+                1024 * 1024,
+                false
+        );
+
         long timestamp = System.currentTimeMillis();
 
-        httpClient.log(
-                timestamp,
-                Map.of("foo", "bar2", "app", "lokki"),
-                "Test"
-        );
+        ThjaziLogger logger = loggingSystem.createLogger();
+        for (int i = 0; i < 1000; i++) {
+            logger.log(
+                    timestamp + i,
+                    Map.of("version", "0.43", "server", "127.0.0.1"),
+                    "Test" + i
+            );
+        }
 
         RestAssured.port = loki.getFirstMappedPort();
         RestAssured.baseURI = "http://" + loki.getHost();
         RestAssured.registerParser("text/plain", Parser.JSON);
 
-//        String response = given()
-        given()
-                .contentType(ContentType.URLENC)
-                .urlEncodingEnabled(false)
-                .formParam("query=%7Bfoo%3D%22bar2%22%7D")
-                .when()
-                .get("/loki/api/v1/query_range")
-                .then()
-//                .extract().asString();
-//
-//        Map<Object, Object> list = JsonPath.from(response)
-//                .getMap("data.result[0]");
-//
-//        System.out.println("list = " + list);
-                .log()
-                .all()
-                .statusCode(200)
-                .body("status", equalTo("success"))
-                .body("data.result.size()", equalTo(1))
-                .body("data.result[0].stream.foo", equalTo("bar2"))
-                .body("data.result[0].values[0]", hasItems("" + (timestamp * 1000_000), "Test"));
+        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()
+                            .statusCode(200)
+                            .log()
+                            .all()
+                            .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.version", equalTo("0.43"))
+                            .body("data.result[0].values", hasItems(new BaseMatcher<>() {
+
+                                int index = 999;
+
+                                @Override
+                                public void describeTo(Description description) {
+
+                                }
+
+                                @Override
+                                public boolean matches(Object o) {
+                                    List<Object> list = (List<Object>) o;
+                                    if (list.size() != 2) {
+                                        return false;
+                                    }
+
+                                    long actualTimestamp = Long.parseLong(list.get(0).toString());
+                                    long expectedTimestamp = (timestamp + index) * 1000_000;
+
+                                    String actualLogLine = list.get(1).toString();
+                                    Object expectedLogLine = "Test" + index;
+
+                                    index--;
+
+                                    return actualTimestamp == expectedTimestamp
+                                            && actualLogLine.equals(expectedLogLine);
+                                }
+                            }));
+                });
     }
 }