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.

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());

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(

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

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
                     );

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);

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);
     }

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();
 

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())
         );

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())
         );

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())
             );

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(",", "{", "}"))
         );
     }

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())
         );

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"));
                 });
     }
 }

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;

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())
         );

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

log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogic.java

@@ -68,7 +68,8 @@ public class AppenderLogic implements BiConsumer<LogEvent, ByteBuffer> {
         processDynamicLabelsIfAny(labelSerializer, event);
 
         logger.log(
-                event.getTimeMillis(),
+                event.getInstant().getEpochMillisecond(),
+                event.getInstant().getNanoOfMillisecond(),
                 labelSerializer,
                 byteBuffer
         );

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()
         );

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
         );