diff --git a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java index 5dbc2d5..2b6c369 100644 --- a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java +++ b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java @@ -8,6 +8,7 @@ import pl.tkowalcz.tjahzi.TjahziLogger; import java.io.Serializable; import java.nio.ByteBuffer; +import java.util.Collections; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; @@ -17,18 +18,23 @@ public class AppenderLogic implements BiConsumer { protected static final org.apache.logging.log4j.Logger LOGGER = StatusLogger.getLogger(); private final LoggingSystem loggingSystem; - private final TjahziLogger logger; + private final String logLevelLabel; private final Map lokiLabels; + private final TjahziLogger logger; + public AppenderLogic( LoggingSystem loggingSystem, + String logLevelLabel, Map lokiLabels ) { this.loggingSystem = loggingSystem; - logger = loggingSystem.createLogger(); - this.lokiLabels = lokiLabels; + this.lokiLabels = Collections.unmodifiableMap(lokiLabels); + this.logLevelLabel = logLevelLabel; + + logger = loggingSystem.createLogger(); } public void append( @@ -51,9 +57,13 @@ public class AppenderLogic implements BiConsumer { @Override public void accept(LogEvent event, ByteBuffer byteBuffer) { + String logLevel = event.getLevel().toString(); + logger.log( event.getTimeMillis(), lokiLabels, + logLevelLabel, + logLevel, byteBuffer ); } diff --git a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java index 53cb9bc..bfc6110 100644 --- a/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java +++ b/loki-log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java @@ -68,6 +68,9 @@ public class LokiAppender extends AbstractAppender { @PluginBuilderAttribute private boolean useOffHeapBuffer = true; + @PluginBuilderAttribute + private String logLevelLabel; + @PluginElement("Headers") private Header[] headers; @@ -118,6 +121,7 @@ public class LokiAppender extends AbstractAppender { getFilter(), isIgnoreExceptions(), getPropertyArray(), + logLevelLabel, lokiLabels, loggingSystem ); @@ -186,6 +190,14 @@ public class LokiAppender extends AbstractAppender { this.useOffHeapBuffer = useOffHeapBuffer; } + public String getLogLevelLabel() { + return logLevelLabel; + } + + public void setLogLevelLabel(String logLevelLabel) { + this.logLevelLabel = logLevelLabel; + } + public void setHeaders(Header[] headers) { this.headers = headers; } @@ -216,6 +228,7 @@ public class LokiAppender extends AbstractAppender { Filter filter, boolean ignoreExceptions, Property[] properties, + String logLevelLabel, Map lokiLabels, LoggingSystem loggingSystem) { super( @@ -230,6 +243,7 @@ public class LokiAppender extends AbstractAppender { this.loggingSystem = loggingSystem; this.appenderLogic = new AppenderLogic( loggingSystem, + logLevelLabel, lokiLabels ); } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java new file mode 100644 index 0000000..dfb2cd9 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java @@ -0,0 +1,104 @@ +package pl.tkowalcz.tjahzi; + +import org.agrona.concurrent.AtomicBuffer; + +import java.nio.ByteBuffer; +import java.util.Map; + +public class LogBufferSerializer { + + private final AtomicBuffer buffer; + + public LogBufferSerializer(AtomicBuffer buffer) { + this.buffer = buffer; + } + + public void writeTo( + int cursor, + long timestamp, + Map labels, + String logLevelLabel, + String logLevel, + ByteBuffer line + ) { + cursor = writeHeader(cursor, timestamp, labels); + cursor = writeLabels(cursor, labels); + cursor = writeLabel( + cursor, + logLevelLabel, + logLevel + ); + writeLogLine(cursor, line); + } + + private void writeLogLine(int cursor, ByteBuffer line) { + buffer.putInt(cursor, line.remaining()); + cursor += Integer.BYTES; + + buffer.putBytes( + cursor, + line, + line.remaining() + ); + } + + private int writeLabels(int cursor, Map labels) { + for (Map.Entry entry : labels.entrySet()) { + cursor = writeLabel( + cursor, + entry.getKey(), + entry.getValue() + ); + } + return cursor; + } + + private int writeHeader(int cursor, long timestamp, Map labels) { + buffer.putLong(cursor, timestamp); + cursor += Long.BYTES; + + buffer.putInt(cursor, labels.size() + 1); + cursor += Integer.BYTES; + return cursor; + } + + public int writeLabel( + int cursor, + String key, + String value + ) { + cursor += buffer.putStringAscii(cursor, key); + cursor += buffer.putStringAscii(cursor, value); + + return cursor; + } + + private int calculateRequiredSizeAscii(Map labels) { + int timestampSize = Long.BYTES; + int labelsCountSize = Integer.BYTES; + int sum = timestampSize + labelsCountSize; + + for (Map.Entry entry : labels.entrySet()) { + sum += calculateLabeSizeAscii(entry.getKey(), entry.getValue()); + } + + return sum; + } + + private int calculateLabeSizeAscii(String key, String value) { + return Integer.BYTES + key.length() + + Integer.BYTES + value.length(); + } + + public int calculateRequiredSizeAscii( + Map labels, + String logLevelLabel, + String logLevel, + ByteBuffer line + ) { + int lineSize = Integer.BYTES + line.remaining(); + int logLevelSize = calculateLabeSizeAscii(logLevelLabel, logLevel); + + return calculateRequiredSizeAscii(labels) + lineSize + logLevelSize; + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java index 370085c..e676392 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java @@ -1,6 +1,5 @@ package pl.tkowalcz.tjahzi; -import org.agrona.concurrent.AtomicBuffer; import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer; import java.nio.ByteBuffer; @@ -11,111 +10,42 @@ public class TjahziLogger { public static final int LOG_MESSAGE_TYPE_ID = 5; private final ManyToOneRingBuffer logBuffer; + private final LogBufferSerializer serializer; public TjahziLogger(ManyToOneRingBuffer logBuffer) { this.logBuffer = logBuffer; + this.serializer = new LogBufferSerializer(logBuffer.buffer()); } public TjahziLogger log(long timestamp, Map labels, - String line) { - int requiredSize = calculateRequiredSizeAscii(labels, line); - int claim = logBuffer.tryClaim(LOG_MESSAGE_TYPE_ID, requiredSize); - - int cursor = claim; - if (requiredSize > 0) { - try { - AtomicBuffer buffer = this.logBuffer.buffer(); - - buffer.putLong(cursor, timestamp); - cursor += Long.BYTES; - - buffer.putInt(cursor, labels.size()); - cursor += Integer.BYTES; - - for (Map.Entry entry : labels.entrySet()) { - String key = entry.getKey(); - String value = entry.getValue(); - - cursor += buffer.putStringAscii(cursor, key); - cursor += buffer.putStringAscii(cursor, value); - } - - buffer.putStringAscii(cursor, line); - logBuffer.commit(claim); - } catch (Throwable t) { - logBuffer.abort(claim); - } - } - - return this; - } - - public TjahziLogger log(long timestamp, - Map labels, + String logLevelLabel, + String logLevel, ByteBuffer line) { - int requiredSize = calculateRequiredSizeAscii(labels, line); + int requiredSize = serializer.calculateRequiredSizeAscii( + labels, + logLevelLabel, + logLevel, + line + ); + int claim = logBuffer.tryClaim(LOG_MESSAGE_TYPE_ID, requiredSize); + try { + serializer + .writeTo( + claim, + timestamp, + labels, + logLevelLabel, + logLevel, + line + ); - int cursor = claim; - if (requiredSize > 0) { - try { - AtomicBuffer buffer = this.logBuffer.buffer(); - - buffer.putLong(cursor, timestamp); - cursor += Long.BYTES; - - buffer.putInt(cursor, labels.size()); - cursor += Integer.BYTES; - - for (Map.Entry entry : labels.entrySet()) { - String key = entry.getKey(); - String value = entry.getValue(); - - cursor += buffer.putStringAscii(cursor, key); - cursor += buffer.putStringAscii(cursor, value); - } - - buffer.putInt(cursor, line.remaining()); - cursor += Integer.BYTES; - buffer.putBytes( - cursor, - line, - line.remaining() - ); - logBuffer.commit(claim); - } catch (Throwable t) { - logBuffer.abort(claim); - } + logBuffer.commit(claim); + } catch (Throwable t) { + logBuffer.abort(claim); } return this; } - - private int calculateRequiredSizeAscii(Map labels) { - return Long.BYTES + - Integer.BYTES + - labels - .entrySet() - .stream() - .mapToInt(entry -> entry.getKey().length() + entry.getValue().length()) - .map(size -> size + 2 * Integer.BYTES) - .sum(); - } - - private int calculateRequiredSizeAscii( - Map labels, - String line - ) { - return calculateRequiredSizeAscii(labels) + - line.length() + Integer.BYTES; - } - - private int calculateRequiredSizeAscii( - Map labels, - ByteBuffer line - ) { - return calculateRequiredSizeAscii(labels) + - line.remaining() + Integer.BYTES; - } } diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java index b165b48..20283e8 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java @@ -9,6 +9,7 @@ import pl.tkowalcz.tjahzi.http.ClientConfiguration; import pl.tkowalcz.tjahzi.http.HttpClientFactory; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import java.nio.ByteBuffer; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -84,7 +85,9 @@ class HeadersTest { logger.log( System.currentTimeMillis(), Map.of(), - "Test" + "level", + "warn", + ByteBuffer.wrap("Test".getBytes()) ); // Then @@ -129,7 +132,9 @@ class HeadersTest { logger.log( System.currentTimeMillis(), Map.of(), - "Test" + "level", + "warn", + ByteBuffer.wrap("Test".getBytes()) ); // Then @@ -145,7 +150,7 @@ class HeadersTest { void shouldNotOverrideCrucialHeaders() { // Given wireMockServer.stubFor( - get(urlEqualTo("/loki/api/v1/push")) + post(urlEqualTo("/loki/api/v1/push")) .willReturn( aResponse().withStatus(200) )); @@ -176,7 +181,9 @@ class HeadersTest { logger.log( System.currentTimeMillis(), Map.of(), - "Test" + "level", + "warn", + ByteBuffer.wrap("Test".getBytes()) ); // Then @@ -186,7 +193,7 @@ class HeadersTest { wireMockServer.verify( postRequestedFor(urlMatching("/loki/api/v1/push")) .withHeader("content-type", matching("application/x-protobuf")) - .withHeader("content-length", matching("29|30")) + .withHeader("content-length", matching("42|43")) .withHeader("host", matching("localhost")) )); } diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java index 09069e1..e0bff6c 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java @@ -17,6 +17,7 @@ import pl.tkowalcz.tjahzi.http.ClientConfiguration; import pl.tkowalcz.tjahzi.http.HttpClientFactory; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import java.nio.ByteBuffer; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -51,10 +52,7 @@ class LoggingSystemSanityCheckTest { .build(); NettyHttpClient httpClient = HttpClientFactory.defaultFactory() - .getHttpClient( - clientConfiguration, - new String[0] - ); + .getHttpClient(clientConfiguration); TjahziInitializer initializer = new TjahziInitializer(); loggingSystem = initializer.createLoggingSystem( @@ -84,7 +82,9 @@ class LoggingSystemSanityCheckTest { logger.log( timestamp, Map.of("version", "0.43", "server", "127.0.0.1"), - "Test" + "level", + "warn", + ByteBuffer.wrap("Test".getBytes()) ); // Then @@ -110,6 +110,7 @@ class LoggingSystemSanityCheckTest { .body("data.result.size()", equalTo(1)) .body("data.result[0].stream.server", equalTo("127.0.0.1")) .body("data.result[0].stream.version", equalTo("0.43")) + .body("data.result[0].stream.level", equalTo("warn")) .body("data.result[0].values[0]", hasItems("" + (timestamp * 1000_000), "Test")); }); } diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java index 17e813c..7cd0f2c 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java @@ -19,6 +19,7 @@ import pl.tkowalcz.tjahzi.http.ClientConfiguration; import pl.tkowalcz.tjahzi.http.HttpClientFactory; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import java.nio.ByteBuffer; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -87,7 +88,9 @@ class LoggingSystemTest { logger.log( timestamp + i, Map.of("version", "0.43", "server", "127.0.0.1"), - "Test" + i + "level", + "warn", + ByteBuffer.wrap(("Test" + i).getBytes()) ); } diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java index 73f51ab..e9be92e 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java @@ -9,6 +9,7 @@ import pl.tkowalcz.tjahzi.http.ClientConfiguration; import pl.tkowalcz.tjahzi.http.HttpClientFactory; import pl.tkowalcz.tjahzi.http.NettyHttpClient; +import java.nio.ByteBuffer; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -77,7 +78,9 @@ public class ReconnectTest { logger.log( System.currentTimeMillis(), Map.of(), - "Test" + "level", + "warn", + ByteBuffer.wrap("Test".getBytes()) ); wireMockServer.start(); @@ -117,7 +120,9 @@ public class ReconnectTest { logger.log( System.currentTimeMillis(), Map.of(), - "Test" + "level", + "warn", + ByteBuffer.wrap("Test".getBytes()) ); await() @@ -136,7 +141,9 @@ public class ReconnectTest { logger.log( System.currentTimeMillis(), Map.of(), - "Test" + "level", + "warn", + ByteBuffer.wrap("Test".getBytes()) ); wireMockServer.start();