mirror of
https://github.com/tiennm99/tjahzi.git
synced 2026-10-15 16:15:22 +00:00
Merge pull request #2 from tkowalcz/circleci-project-setup
Circleci project setup
This commit is contained in:
12 files changed
+428
-208
No files matched your search
+1
-29
@@ -8,7 +8,7 @@
|
||||
<version>1.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>thjazi-core</artifactId>
|
||||
<artifactId>core</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<dependencies>
|
||||
@@ -34,40 +34,12 @@
|
||||
<version>6.0.0</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
<version>1.7.30</version>
|
||||
</dependency>
|
||||
<!-- Binding for Log4J -->
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-slf4j-impl</artifactId>
|
||||
<version>2.12.0</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-api</artifactId>
|
||||
<version>2.11.2</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-core</artifactId>
|
||||
<version>2.11.2</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.rest-assured</groupId>
|
||||
<artifactId>rest-assured</artifactId>
|
||||
<version>4.3.1</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.github.dzieciou.testing</groupId>
|
||||
<artifactId>curl-logger</artifactId>
|
||||
<version>1.0.5</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.awaitility</groupId>
|
||||
<artifactId>awaitility</artifactId>
|
||||
|
||||
@@ -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";
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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<SocketChannel> {
|
||||
p.addLast(new HttpClientCodec());
|
||||
p.addLast(new HttpObjectAggregator(1048576));
|
||||
p.addLast(new HttpContentDecompressor());
|
||||
p.addLast(new SimpleChannelInboundHandler<FullHttpResponse>() {
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
|
||||
System.out.println(msg);
|
||||
}
|
||||
});
|
||||
p.addLast(new InactiveConnectionsHandler());
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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"));
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}));
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0">
|
||||
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>pl.tkowalcz</groupId>
|
||||
<artifactId>thjazi</artifactId>
|
||||
<version>1.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>log4j2-appender</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>pl.tkowalcz</groupId>
|
||||
<artifactId>core</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- Binding for Log4J -->
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-api</artifactId>
|
||||
<version>2.13.3</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.logging.log4j</groupId>
|
||||
<artifactId>log4j-core</artifactId>
|
||||
<version>2.13.3</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.rest-assured</groupId>
|
||||
<artifactId>rest-assured</artifactId>
|
||||
<version>4.3.1</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.awaitility</groupId>
|
||||
<artifactId>awaitility</artifactId>
|
||||
<version>4.0.3</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
|
||||
<!-- Testcontainers -->
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter-api</artifactId>
|
||||
<version>5.4.2</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter-params</artifactId>
|
||||
<version>5.4.2</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter-engine</artifactId>
|
||||
<version>5.4.2</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>testcontainers</artifactId>
|
||||
<version>1.14.3</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
<version>1.14.3</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -11,8 +11,8 @@
|
||||
|
||||
<modules>
|
||||
<module>loki-protobuf</module>
|
||||
<module>log4j2-appender</module>
|
||||
<module>core</module>
|
||||
<module>slf4j-binding</module>
|
||||
</modules>
|
||||
|
||||
<scm>
|
||||
@@ -20,6 +20,22 @@
|
||||
<url>https://github.com/tkowalcz/thjazi</url>
|
||||
</scm>
|
||||
|
||||
<licenses>
|
||||
<license>
|
||||
<name>MIT License</name>
|
||||
<url>https://github.com/tkowalcz/thjazi/blob/master/LICENSE</url>
|
||||
<distribution>repo</distribution>
|
||||
<comments>A short and simple permissive license</comments>
|
||||
</license>
|
||||
</licenses>
|
||||
|
||||
<developers>
|
||||
<developer>
|
||||
<id>tkowalcz</id>
|
||||
<name>Tomasz Kowalczewski</name>
|
||||
</developer>
|
||||
</developers>
|
||||
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
|
||||
|
||||
@@ -1,27 +0,0 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0">
|
||||
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>pl.tkowalcz</groupId>
|
||||
<artifactId>thjazi</artifactId>
|
||||
<version>1.0-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>thjazi-slf4j-impl</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>pl.tkowalcz</groupId>
|
||||
<artifactId>thjazi-core</artifactId>
|
||||
<version>1.0-SNAPSHOT</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
<version>1.7.30</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
Reference in new issue
Block a user