From 4b8d644422c2e5118d0c098be5517e61a257832f Mon Sep 17 00:00:00 2001 From: Tomasz Kowalczewski Date: Sat, 27 Aug 2022 01:05:31 +0200 Subject: [PATCH] 96 greater timestamp precision (#97) * Re 96: Using fine precision event time provided by log4j for log line timestamps. Loki as well as our serializer support it. * Re #96: Fixed tests. * Re #96: Fixed tests. * Re #96: Improved tests. --- .../tkowalcz/tjahzi/LogBufferSerializer.java | 20 +++++++++++++------ .../tkowalcz/tjahzi/LogBufferTranscoder.java | 12 +++++++++-- .../java/pl/tkowalcz/tjahzi/OutputBuffer.java | 10 ++++++++-- .../java/pl/tkowalcz/tjahzi/TjahziLogger.java | 18 +++++++++++------ .../tjahzi/protobuf/EntrySerializer.java | 5 +++-- .../tjahzi/protobuf/StreamSerializer.java | 5 +++-- .../tjahzi/protobuf/TimestampSerializer.java | 7 ++++--- .../thjazi/protobuf/ProtobufDeserializer.java | 1 + .../java/pl/tkowalcz/tjahzi/HeadersTest.java | 3 +++ .../tkowalcz/tjahzi/LogBufferAgentTest.java | 5 +++++ .../LogBufferSerializerDeserializerTest.java | 18 ++++++++++------- .../tjahzi/LogBufferTranscoderTest.java | 4 +++- .../tjahzi/LoggingSystemSanityCheckTest.java | 3 ++- .../pl/tkowalcz/tjahzi/LoggingSystemTest.java | 3 ++- .../pl/tkowalcz/tjahzi/ReconnectTest.java | 3 +++ .../TimeCappedBatchingStrategyTest.java | 5 +++-- .../tkowalcz/tjahzi/log4j2/AppenderLogic.java | 3 ++- .../tjahzi/log4j2/AppenderLogicTest.java | 2 ++ .../tkowalcz/tjahzi/logback/LokiAppender.java | 1 + 19 files changed, 92 insertions(+), 36 deletions(-) diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java b/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java index 1d6a006..788efd4 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferSerializer.java @@ -18,22 +18,26 @@ public class LogBufferSerializer { ) { int logLineSize = Integer.BYTES + line.remaining(); - int timestampSize = Long.BYTES; + int nanosecondResolutionTimestampSize = 2 * Long.BYTES; int labelsCountSize = Integer.BYTES; - int headerAndLabelsSize = timestampSize + labelsCountSize + labelSerializer.sizeBytes(); + int headerAndLabelsSize = nanosecondResolutionTimestampSize + + labelsCountSize + + labelSerializer.sizeBytes(); return headerAndLabelsSize + logLineSize; } public void writeTo( int cursor, - long timestamp, + long epochMillisecond, + long nanoOfMillisecond, LabelSerializer serializedLabels, ByteBuffer line ) { cursor = writeHeader( cursor, - timestamp, + epochMillisecond, + nanoOfMillisecond, serializedLabels ); @@ -43,10 +47,14 @@ public class LogBufferSerializer { private int writeHeader( int cursor, - long timestamp, + long epochMillisecond, + long nanoOfMillisecond, LabelSerializer serializedLabels ) { - buffer.putLong(cursor, timestamp); + buffer.putLong(cursor, epochMillisecond); + cursor += Long.BYTES; + + buffer.putLong(cursor, nanoOfMillisecond); cursor += Long.BYTES; buffer.putInt(cursor, serializedLabels.getLabelsCount()); diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferTranscoder.java b/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferTranscoder.java index 3b03172..683e73c 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferTranscoder.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferTranscoder.java @@ -30,7 +30,10 @@ public class LogBufferTranscoder { } public void deserializeIntoByteBuf(DirectBuffer buffer, int index, OutputBuffer outputBuffer) { - long timestamp = buffer.getLong(index); + long epochMillisecond = buffer.getLong(index); + index += Long.BYTES; + + long nanoOfMillisecond = buffer.getLong(index); index += Long.BYTES; TextBuilder labelsBuilder = TextBuilders.threadLocal(); @@ -49,7 +52,12 @@ public class LogBufferTranscoder { ); logLineHolder.readerIndex(index); - outputBuffer.addLogLine(actualLabels, timestamp, logLineHolder); + outputBuffer.addLogLine( + actualLabels, + epochMillisecond, + nanoOfMillisecond, + logLineHolder + ); } private int readLabels( diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/OutputBuffer.java b/core/src/main/java/pl/tkowalcz/tjahzi/OutputBuffer.java index b49309d..9f520f0 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/OutputBuffer.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/OutputBuffer.java @@ -16,10 +16,16 @@ public class OutputBuffer { target.clear(); } - public void addLogLine(CharSequence labels, long timestamp, ByteBuf logLine) { + public void addLogLine( + CharSequence labels, + long epochMillisecond, + long nanoOfMillisecond, + ByteBuf logLine + ) { PushRequestSerializer.serialize(target); StreamSerializer.serialize( - timestamp, + epochMillisecond, + nanoOfMillisecond, logLine, labels, target diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java index 8822707..a82b7cc 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java @@ -21,9 +21,12 @@ public class TjahziLogger { this.serializer = new LogBufferSerializer(logBuffer.buffer()); } - public TjahziLogger log(long timestamp, - LabelSerializer serializedLabels, - ByteBuffer line) { + public TjahziLogger log( + long epochMillisecond, + long nanoOfMillisecond, + LabelSerializer serializedLabels, + ByteBuffer line + ) { int requiredSize = serializer.calculateRequiredSize( serializedLabels, line @@ -32,7 +35,8 @@ public class TjahziLogger { int claim = logBuffer.tryClaim(LOG_MESSAGE_TYPE_ID, requiredSize); if (claim > 0) { putMessageOnRing( - timestamp, + epochMillisecond, + nanoOfMillisecond, serializedLabels, line, claim @@ -45,7 +49,8 @@ public class TjahziLogger { } private void putMessageOnRing( - long timestamp, + long epochMillisecond, + long nanoOfMillisecond, LabelSerializer serializedLabels, ByteBuffer line, int claim @@ -54,7 +59,8 @@ public class TjahziLogger { serializer .writeTo( claim, - timestamp, + epochMillisecond, + nanoOfMillisecond, serializedLabels, line ); diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/EntrySerializer.java b/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/EntrySerializer.java index 1f45b54..6d4aa1e 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/EntrySerializer.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/EntrySerializer.java @@ -11,7 +11,8 @@ public class EntrySerializer { public static final int LOG_LINE_FIELD_NUMBER = 2; public static void serialize( - long timestamp, + long epochMillisecond, + long nanoOfMillisecond, ByteBuf logLine, ByteBuf target ) { @@ -19,7 +20,7 @@ public class EntrySerializer { target.writeInt(0); target.writeByte(TIMESTAMP_FIELD_NUMBER << 3 | LENGTH_DELIMITED_TYPE); - TimestampSerializer.serialize(timestamp, target); + TimestampSerializer.serialize(epochMillisecond, nanoOfMillisecond, target); target.writeByte(LOG_LINE_FIELD_NUMBER << 3 | LENGTH_DELIMITED_TYPE); StringSerializer.serialize(logLine, target); diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/StreamSerializer.java b/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/StreamSerializer.java index dbb0530..3d31be0 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/StreamSerializer.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/StreamSerializer.java @@ -10,7 +10,8 @@ public class StreamSerializer { public static final int ENTRY_FIELD_NUMBER = 2; public static void serialize( - long timestamp, + long epochMillisecond, + long nanoOfMillisecond, ByteBuf logLine, CharSequence labels, ByteBuf target @@ -22,7 +23,7 @@ public class StreamSerializer { StringSerializer.serialize(labels, target); target.writeByte(ENTRY_FIELD_NUMBER << 3 | LENGTH_DELIMITED_TYPE); - EntrySerializer.serialize(timestamp, logLine, target); + EntrySerializer.serialize(epochMillisecond, nanoOfMillisecond, logLine, target); Protobuf.writeSize(target, messageStartIndex); } diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/TimestampSerializer.java b/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/TimestampSerializer.java index 6901f49..53fb54e 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/TimestampSerializer.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/protobuf/TimestampSerializer.java @@ -8,11 +8,12 @@ import static pl.tkowalcz.tjahzi.protobuf.Protobuf.writeUnsignedVarint; public class TimestampSerializer { public static void serialize( - long timestamp, + long epochMillisecond, + long nanoOfMillisecond, ByteBuf target ) { - long timestampSeconds = timestamp / 1000; - int timestampNanos = (int) (timestamp % 1000) * 1000_000; + long timestampSeconds = epochMillisecond / 1000; + long timestampNanos = (epochMillisecond % 1000) * 1000_000 + nanoOfMillisecond; int messageStartIndex = target.writerIndex(); diff --git a/core/src/test/java/pl/tkowalcz/thjazi/protobuf/ProtobufDeserializer.java b/core/src/test/java/pl/tkowalcz/thjazi/protobuf/ProtobufDeserializer.java index 6c4fb18..8091367 100644 --- a/core/src/test/java/pl/tkowalcz/thjazi/protobuf/ProtobufDeserializer.java +++ b/core/src/test/java/pl/tkowalcz/thjazi/protobuf/ProtobufDeserializer.java @@ -38,6 +38,7 @@ public class ProtobufDeserializer { serializer.writeTo( 0, System.currentTimeMillis(), + 0, labelSerializer, ByteBuffer.wrap("Test".getBytes()) ); diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java index 696bf21..165c23c 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java @@ -99,6 +99,7 @@ class HeadersTest { TjahziLogger logger = loggingSystem.createLogger(); logger.log( System.currentTimeMillis(), + 0, labelSerializer, ByteBuffer.wrap("Test".getBytes()) ); @@ -152,6 +153,7 @@ class HeadersTest { TjahziLogger logger = loggingSystem.createLogger(); logger.log( System.currentTimeMillis(), + 0, labelSerializer, ByteBuffer.wrap("Test".getBytes()) ); @@ -207,6 +209,7 @@ class HeadersTest { TjahziLogger logger = loggingSystem.createLogger(); logger.log( System.currentTimeMillis(), + 0, labelSerializer, ByteBuffer.wrap("Test".getBytes()) ); diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java index 8744b75..f1f9df2 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java @@ -68,6 +68,7 @@ class LogBufferAgentTest { for (int i = 0; i < 100; i++) { logger.log( 42L, + 0, LabelSerializerCreator.from(Map.of()), ByteBuffer.wrap(( "Cupcake ipsum dolor sit amet cake wafer. " + @@ -118,6 +119,7 @@ class LogBufferAgentTest { logger.log( 42L, + 0, LabelSerializerCreator.from(Map.of()), ByteBuffer.wrap("Test".getBytes()) ); @@ -207,6 +209,7 @@ class LogBufferAgentTest { // When logger.log( 42L, + 0, LabelSerializerCreator.from(Map.of()), ByteBuffer.wrap("Test".getBytes()) ); @@ -215,6 +218,7 @@ class LogBufferAgentTest { logger.log( 42L, + 0, LabelSerializerCreator.from(Map.of()), ByteBuffer.wrap("Test".getBytes()) ); @@ -263,6 +267,7 @@ class LogBufferAgentTest { for (int i = 0; i < LogShipperAgent.MAX_MESSAGES_TO_RETRIEVE * 2 + 1; i++) { logger.log( 42L, + 0, LabelSerializerCreator.from(Map.of()), ByteBuffer.wrap("Test".getBytes()) ); diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerDeserializerTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerDeserializerTest.java index 44c76f5..9c70431 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerDeserializerTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferSerializerDeserializerTest.java @@ -37,7 +37,8 @@ class LogBufferSerializerDeserializerTest { + 21 // log level + 45 // labels string sizes + 6 * 4 // labels string length indicators - + 8 // timestamp + + 8 // timestamp millis + + 8 // timestamp nanos + 4 // labels count ), Arguments.of( @@ -48,7 +49,8 @@ class LogBufferSerializerDeserializerTest { "[Mando] You have something I want.", 38 // log line + 21 // log level - + 8 // timestamp + + 8 // timestamp millis + + 8 // timestamp nanos + 4 // labels count ), Arguments.of( @@ -63,7 +65,8 @@ class LogBufferSerializerDeserializerTest { 38 // log line + 45 // labels string sizes + 6 * 4 // labels string length indicators - + 8 // timestamp + + 8 // timestamp millis + + 8 // timestamp nanos + 4 // labels count ) ); @@ -93,6 +96,7 @@ class LogBufferSerializerDeserializerTest { serializer.writeTo( 0, 32042L, + 882L, labelSerializer, ByteBuffer.wrap(logLine.getBytes()) ); @@ -120,7 +124,7 @@ class LogBufferSerializerDeserializerTest { assertThat(stream.getEntriesList().get(0).getTimestamp()).isEqualTo( Timestamp.newBuilder() .setSeconds(32) - .setNanos(42_000_000) + .setNanos(42_000_882) .build() ); assertThat(stream.getEntriesList().get(0).getLine()).isEqualTo(logLine); @@ -130,9 +134,9 @@ class LogBufferSerializerDeserializerTest { assertThat(stream.getLabels()).isEqualToIgnoringWhitespace( Streams.concat( - incomingLabelsStream, - logLevelStream - ) + incomingLabelsStream, + logLevelStream + ) .collect(joining(",", "{", "}")) ); } diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferTranscoderTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferTranscoderTest.java index a9fd2e6..ff19f90 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferTranscoderTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferTranscoderTest.java @@ -48,6 +48,7 @@ class LogBufferTranscoderTest { serializer.writeTo( 0, 32042L, + 882L, labelSerializer, ByteBuffer.wrap(logLine.getBytes()) ); @@ -80,7 +81,7 @@ class LogBufferTranscoderTest { assertThat(stream.getEntriesList().get(0).getTimestamp()).isEqualTo( Timestamp.newBuilder() .setSeconds(32) - .setNanos(42_000_000) + .setNanos(42_000_882) .build() ); assertThat(stream.getEntriesList().get(0).getLine()).isEqualTo(logLine); @@ -122,6 +123,7 @@ class LogBufferTranscoderTest { serializer.writeTo( 0, 32042L, + 0, labelSerializer, ByteBuffer.wrap("[Mando] You have something I want.".getBytes()) ); diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java index 56de92d..792c3e4 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java @@ -96,6 +96,7 @@ class LoggingSystemSanityCheckTest { // When logger.log( timestamp, + 882L, labelSerializer, ByteBuffer.wrap("Test".getBytes()) ); @@ -124,7 +125,7 @@ class LoggingSystemSanityCheckTest { .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")); + .body("data.result[0].values[0]", hasItems("" + (timestamp * 1000_000 +882), "Test")); }); } } diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java index 39a4ecc..e5fd61e 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java @@ -99,6 +99,7 @@ class LoggingSystemTest { for (int i = 0; i < 1000; i++) { logger.log( timestamp + i, + 9974, LabelSerializerCreator.from( Map.of("level", "warn") ), @@ -149,7 +150,7 @@ class LoggingSystemTest { } long actualTimestamp = Long.parseLong(list.get(0).toString()); - long expectedTimestamp = (timestamp + index) * 1000_000; + long expectedTimestamp = (timestamp + index) * 1000_000 + 9974; String actualLogLine = list.get(1).toString(); Object expectedLogLine = "Test" + index; diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java index 19e42f8..1fcdabd 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java @@ -94,6 +94,7 @@ public class ReconnectTest { TjahziLogger logger = loggingSystem.createLogger(); logger.log( System.currentTimeMillis(), + 0, labelSerializer, ByteBuffer.wrap("Test".getBytes()) ); @@ -144,6 +145,7 @@ public class ReconnectTest { TjahziLogger logger = loggingSystem.createLogger(); logger.log( System.currentTimeMillis(), + 0, labelSerializer, ByteBuffer.wrap("Test".getBytes()) ); @@ -163,6 +165,7 @@ public class ReconnectTest { // When logger.log( System.currentTimeMillis(), + 0, labelSerializer, ByteBuffer.wrap("Test".getBytes()) ); diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategyTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategyTest.java index 1afdb6f..fe89ac2 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategyTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategyTest.java @@ -27,6 +27,7 @@ class TimeCappedBatchingStrategyTest { outputBufferWthData.addLogLine( "foobar", 44L, + 0, logLine.duplicate() ); } @@ -119,7 +120,7 @@ class TimeCappedBatchingStrategyTest { ); // When - outputBufferWithNoData.addLogLine("foobar", 44L, logLine); + outputBufferWithNoData.addLogLine("foobar", 44L, 0, logLine); // Then assertThat(strategy.shouldProceed()).isTrue(); @@ -143,7 +144,7 @@ class TimeCappedBatchingStrategyTest { ); // When - outputBufferWithNoData.addLogLine("foobar", 44L, logLine); + outputBufferWithNoData.addLogLine("foobar", 44L, 0, logLine); clock.setMillis(clock.millis() + batchWaitMillis); // Then diff --git a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java index fc42c4f..16b7fdc 100644 --- a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java +++ b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java @@ -68,7 +68,8 @@ public class AppenderLogic implements BiConsumer { processDynamicLabelsIfAny(labelSerializer, event); logger.log( - event.getTimeMillis(), + event.getInstant().getEpochMillisecond(), + event.getInstant().getNanoOfMillisecond(), labelSerializer, byteBuffer ); diff --git a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogicTest.java b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogicTest.java index 62eda7e..158d649 100644 --- a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogicTest.java +++ b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogicTest.java @@ -52,6 +52,7 @@ class AppenderLogicTest { // Then verify(tjahziLogger).log( eq(timestamp), + eq(0L), eq(labelSerializer), any() ); @@ -110,6 +111,7 @@ class AppenderLogicTest { // Then verify(tjahziLogger).log( eq(timestamp), + eq(0L), eq(labelSerializer), any() ); diff --git a/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppender.java b/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppender.java index b8cb71f..f1a6bfe 100644 --- a/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppender.java +++ b/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppender.java @@ -71,6 +71,7 @@ public class LokiAppender extends LokiAppenderConfigurator { logger.log( event.getTimeStamp(), + 0L, labelSerializer, logLine );