From 29f2dd1313c6cffe02da410272bce5c721bc1ffa Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Fri, 25 Dec 2020 23:02:35 +0100 Subject: [PATCH 1/5] Re #9: Added separate log level label and its configuration. Refactored serialization code (that writes into log buffer). --- .../tkowalcz/tjahzi/log4j2/AppenderLogic.java | 16 ++- .../tkowalcz/tjahzi/log4j2/LokiAppender.java | 14 +++ .../tkowalcz/tjahzi/LogBufferSerializer.java | 104 +++++++++++++++ .../java/pl/tkowalcz/tjahzi/TjahziLogger.java | 118 ++++-------------- .../java/pl/tkowalcz/tjahzi/HeadersTest.java | 17 ++- .../tjahzi/LoggingSystemSanityCheckTest.java | 11 +- .../pl/tkowalcz/tjahzi/LoggingSystemTest.java | 5 +- .../pl/tkowalcz/tjahzi/ReconnectTest.java | 13 +- 8 files changed, 187 insertions(+), 111 deletions(-) create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java 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(); From b2d5d1e0d4f98e50d83e5d359e18e9fbc223f559 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sat, 26 Dec 2020 18:00:44 +0100 Subject: [PATCH 2/5] Re #9: Fixed bugs and added test. --- .../tkowalcz/tjahzi/LogBufferSerializer.java | 111 +++++++++++------ .../tjahzi/LogBufferSerializerTest.java | 115 ++++++++++++++++++ 2 files changed, 191 insertions(+), 35 deletions(-) create mode 100644 tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerTest.java diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java index dfb2cd9..1362858 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java @@ -13,6 +13,19 @@ public class LogBufferSerializer { this.buffer = buffer; } + public int calculateRequiredSizeAscii( + Map labels, + String logLevelLabel, + String logLevel, + ByteBuffer line + ) { + int logLineSize = Integer.BYTES + line.remaining(); + int logLevelSize = calculateOptionalLabelSizeAscii(logLevelLabel, logLevel); + int headerAndLabelsSize = calculateRequiredSizeAscii(labels); + + return headerAndLabelsSize + logLineSize + logLevelSize; + } + public void writeTo( int cursor, long timestamp, @@ -21,16 +34,54 @@ public class LogBufferSerializer { String logLevel, ByteBuffer line ) { - cursor = writeHeader(cursor, timestamp, labels); + cursor = writeHeader( + cursor, + timestamp, + labels, + logLevelLabel != null + ); + cursor = writeLabels(cursor, labels); - cursor = writeLabel( + cursor = writeOptionalLabel( cursor, logLevelLabel, logLevel ); + writeLogLine(cursor, line); } + private int writeHeader( + int cursor, + long timestamp, + Map labels, + boolean hasLogLevelLabel + ) { + buffer.putLong(cursor, timestamp); + cursor += Long.BYTES; + + int labelsCount = labels.size(); + if (hasLogLevelLabel) { + labelsCount++; + } + + buffer.putInt(cursor, labelsCount); + cursor += Integer.BYTES; + return cursor; + } + + private int writeLabels(int cursor, Map labels) { + for (Map.Entry entry : labels.entrySet()) { + cursor = writeLabel( + cursor, + entry.getKey(), + entry.getValue() + ); + } + + return cursor; + } + private void writeLogLine(int cursor, ByteBuffer line) { buffer.putInt(cursor, line.remaining()); cursor += Integer.BYTES; @@ -42,27 +93,7 @@ public class LogBufferSerializer { ); } - 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( + private int writeLabel( int cursor, String key, String value @@ -73,32 +104,42 @@ public class LogBufferSerializer { return cursor; } + private int writeOptionalLabel( + int cursor, + String key, + String value + ) { + if (key != null) { + 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()); + sum += calculateLabelSizeAscii(entry.getKey(), entry.getValue()); } return sum; } - private int calculateLabeSizeAscii(String key, String value) { + private int calculateOptionalLabelSizeAscii(String key, String value) { + if (key == null) { + return 0; + } + 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; + private int calculateLabelSizeAscii(String key, String value) { + return Integer.BYTES + key.length() + + Integer.BYTES + value.length(); } } diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerTest.java new file mode 100644 index 0000000..d5c4759 --- /dev/null +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerTest.java @@ -0,0 +1,115 @@ +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 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") + void shouldCalculateSizeOfMessage( + String testName, + Map labels, + String logLevelLabel, + String logLevel, + ByteBuffer logLine, + int expectedSize) { + // Given + UnsafeBuffer buffer = new UnsafeBuffer(new byte[expectedSize]); + LogBufferSerializer serializer = new LogBufferSerializer(buffer); + + // When + int actual = serializer.calculateRequiredSizeAscii( + labels, + logLevelLabel, + logLevel, + logLine + ); + + // Then + assertThat(actual).isEqualTo(expectedSize); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("variousMessageConfigurations") + void shouldSerializeMessage( + String testName, + Map 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. + } +} From cdc1cf3eadab72112b2ad845f73027d0ef47c282 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sat, 26 Dec 2020 18:01:01 +0100 Subject: [PATCH 3/5] Re #9: Cleanup. --- .../tkowalcz/tjahzi/log4j2/BufferSizeConfigurationTest.java | 2 -- .../main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java | 5 ++--- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/BufferSizeConfigurationTest.java b/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/BufferSizeConfigurationTest.java index c160303..17ac79e 100644 --- a/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/BufferSizeConfigurationTest.java +++ b/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/BufferSizeConfigurationTest.java @@ -3,14 +3,12 @@ package pl.tkowalcz.tjahzi.log4j2; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.core.LoggerContext; import org.junit.jupiter.api.Test; -import org.testcontainers.junit.jupiter.Testcontainers; import java.net.URI; import java.net.URISyntaxException; import static org.assertj.core.api.Assertions.assertThat; -@Testcontainers class BufferSizeConfigurationTest { @Test diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java index 5a2e355..8ce408a 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java @@ -14,16 +14,15 @@ class ResponseHandler extends SimpleChannelInboundHandler { @Override protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) { - System.out.println("msg = " + msg); } @Override - public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { System.out.println("cause = " + cause); } @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { + public void channelInactive(ChannelHandlerContext ctx) { System.out.println("ResponseHandler.channelInactive"); } } From e96f28fb79204a4160332050f4c47f2c479095db Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sat, 26 Dec 2020 18:01:14 +0100 Subject: [PATCH 4/5] Re #9: Cleanup. --- .../java/pl/tkowalcz/tjahzi/TjahziLogger.java | 26 +++++++++++++++++-- 1 file changed, 24 insertions(+), 2 deletions(-) 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 e676392..f7e7832 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java @@ -30,6 +30,30 @@ public class TjahziLogger { ); int claim = logBuffer.tryClaim(LOG_MESSAGE_TYPE_ID, requiredSize); + if (claim > 0) { + putMessageOnRing( + timestamp, + labels, + logLevelLabel, + logLevel, + line, + claim + ); + } else { + // TODO: increment dropped lines metric + } + + return this; + } + + private void putMessageOnRing( + long timestamp, + Map labels, + String logLevelLabel, + String logLevel, + ByteBuffer line, + int claim + ) { try { serializer .writeTo( @@ -45,7 +69,5 @@ public class TjahziLogger { } catch (Throwable t) { logBuffer.abort(claim); } - - return this; } } From 49d5f91a2225dcd25dcff2c0136096846537af04 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sat, 26 Dec 2020 18:01:21 +0100 Subject: [PATCH 5/5] Re #9: Tests. --- .../LogLevelLabelConfigurationTest.java | 130 ++++++++++++++++++ .../appender-test-with-log-label-set.xml | 25 ++++ .../appender-test-with-log-label-unset.xml | 23 ++++ 3 files changed, 178 insertions(+) create mode 100644 loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LogLevelLabelConfigurationTest.java create mode 100644 loki-log4j2-appender/src/test/resources/appender-test-with-log-label-set.xml create mode 100644 loki-log4j2-appender/src/test/resources/appender-test-with-log-label-unset.xml diff --git a/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LogLevelLabelConfigurationTest.java b/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LogLevelLabelConfigurationTest.java new file mode 100644 index 0000000..36dddb2 --- /dev/null +++ b/loki-log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LogLevelLabelConfigurationTest.java @@ -0,0 +1,130 @@ +package pl.tkowalcz.tjahzi.log4j2; + +import io.restassured.RestAssured; +import io.restassured.http.ContentType; +import io.restassured.parsing.Parser; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +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.wait.strategy.Wait; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import java.net.URI; +import java.net.URISyntaxException; + +import static io.restassured.RestAssured.given; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.nullValue; + +@Testcontainers +class LogLevelLabelConfigurationTest { + + @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); + + @Test + void shouldWorkWIthNoLogLevelConfigured() throws URISyntaxException { + // Given + System.setProperty("loki.host", loki.getHost()); + System.setProperty("loki.port", loki.getFirstMappedPort().toString()); + + URI uri = getClass() + .getClassLoader() + .getResource("appender-test-with-log-label-unset.xml") + .toURI(); + + ((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false)) + .setConfigLocation(uri); + + // When + Logger logger = LogManager.getLogger(LokiAppenderTest.class); + logger.info("Test"); + + // Then + 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() + .log() + .all() + .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.log_level", nullValue()); + }); + } + + @Test + void shouldSendLogLevelAsConfigured() throws URISyntaxException { + // Given + System.setProperty("loki.host", loki.getHost()); + System.setProperty("loki.port", loki.getFirstMappedPort().toString()); + + URI uri = getClass() + .getClassLoader() + .getResource("appender-test-with-log-label-set.xml") + .toURI(); + + ((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false)) + .setConfigLocation(uri); + + // When + Logger logger = LogManager.getLogger(LokiAppenderTest.class); + logger.info("Test"); + + // Then + 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() + .log() + .all() + .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.log_level", equalTo("INFO")); + }); + } +} diff --git a/loki-log4j2-appender/src/test/resources/appender-test-with-log-label-set.xml b/loki-log4j2-appender/src/test/resources/appender-test-with-log-label-set.xml new file mode 100644 index 0000000..eb1ffa9 --- /dev/null +++ b/loki-log4j2-appender/src/test/resources/appender-test-with-log-label-set.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + ${sys:loki.host} + ${sys:loki.port} + + log_level + + + + %X{tid} [%t] %d{MM-dd HH:mm:ss.SSS} %5p %c{1} - %m%n%exception{full} + + +
+