diff --git a/core/pom.xml b/core/pom.xml
index 389dbf0..f2da192 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -8,7 +8,7 @@
1.0-SNAPSHOT
- thjazi-core
+ core
jar
@@ -34,40 +34,12 @@
6.0.0
-
- org.slf4j
- slf4j-api
- 1.7.30
-
-
-
- org.apache.logging.log4j
- log4j-slf4j-impl
- 2.12.0
-
-
- org.apache.logging.log4j
- log4j-api
- 2.11.2
-
-
- org.apache.logging.log4j
- log4j-core
- 2.11.2
-
-
io.rest-assured
rest-assured
4.3.1
test
-
- com.github.dzieciou.testing
- curl-logger
- 1.0.5
- test
-
org.awaitility
awaitility
diff --git a/core/src/main/java/pl/tkowalcz/thjazi/LogBufferAgent.java b/core/src/main/java/pl/tkowalcz/thjazi/LogBufferAgent.java
new file mode 100644
index 0000000..e770439
--- /dev/null
+++ b/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 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 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";
+ }
+}
diff --git a/core/src/main/java/pl/tkowalcz/thjazi/ThjaziInitializer.java b/core/src/main/java/pl/tkowalcz/thjazi/ThjaziInitializer.java
index c6558a9..445d92d 100644
--- a/core/src/main/java/pl/tkowalcz/thjazi/ThjaziInitializer.java
+++ b/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
);
}
diff --git a/core/src/main/java/pl/tkowalcz/thjazi/ThjaziLogger.java b/core/src/main/java/pl/tkowalcz/thjazi/ThjaziLogger.java
index dad0225..ee0f47c 100644
--- a/core/src/main/java/pl/tkowalcz/thjazi/ThjaziLogger.java
+++ b/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);
diff --git a/core/src/main/java/pl/tkowalcz/thjazi/http/HttpClientFactory.java b/core/src/main/java/pl/tkowalcz/thjazi/http/HttpClientFactory.java
index a36fa5e..19cb957 100644
--- a/core/src/main/java/pl/tkowalcz/thjazi/http/HttpClientFactory.java
+++ b/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() {
diff --git a/core/src/main/java/pl/tkowalcz/thjazi/http/HttpClientInitializer.java b/core/src/main/java/pl/tkowalcz/thjazi/http/HttpClientInitializer.java
index 65e769f..a776b9c 100644
--- a/core/src/main/java/pl/tkowalcz/thjazi/http/HttpClientInitializer.java
+++ b/core/src/main/java/pl/tkowalcz/thjazi/http/HttpClientInitializer.java
@@ -1,11 +1,8 @@
package pl.tkowalcz.thjazi.http;
-import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
-import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpContentDecompressor;
import io.netty.handler.codec.http.HttpObjectAggregator;
@@ -26,13 +23,5 @@ class HttpClientInitializer extends ChannelInitializer {
p.addLast(new HttpClientCodec());
p.addLast(new HttpObjectAggregator(1048576));
p.addLast(new HttpContentDecompressor());
- p.addLast(new SimpleChannelInboundHandler() {
-
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
- System.out.println(msg);
- }
- });
- p.addLast(new InactiveConnectionsHandler());
}
}
diff --git a/core/src/main/java/pl/tkowalcz/thjazi/http/NettyHttpClient.java b/core/src/main/java/pl/tkowalcz/thjazi/http/NettyHttpClient.java
index d4f2efa..8912dbb 100644
--- a/core/src/main/java/pl/tkowalcz/thjazi/http/NettyHttpClient.java
+++ b/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 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 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
diff --git a/core/src/test/java/pl/tkowalcz/thjazi/LoggingSystemSanityCheckTest.java b/core/src/test/java/pl/tkowalcz/thjazi/LoggingSystemSanityCheckTest.java
new file mode 100644
index 0000000..c47c6bd
--- /dev/null
+++ b/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 -> {
+ 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"));
+ });
+ }
+}
diff --git a/core/src/test/java/pl/tkowalcz/thjazi/LoggingSystemTest.java b/core/src/test/java/pl/tkowalcz/thjazi/LoggingSystemTest.java
index bdee156..2028ddd 100644
--- a/core/src/test/java/pl/tkowalcz/thjazi/LoggingSystemTest.java
+++ b/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 -> {
-// 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 -> {
+ 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