Issue #30: Solution for duplicated dropped logs.

* ITimestampProvider introduced.
* Refactoring.
* v1.1.3

Author: mjfryc

src/main/java/pl/mjaron/tinyloki/CurrentTimestampProvider.java

@@ -0,0 +1,32 @@
+package pl.mjaron.tinyloki;
+
+/**
+ * Basic implementation.
+ * <p>
+ * Provides current timestamp provided by {@link System#currentTimeMillis()}
+ * and converted to nanoseconds by multiplying by <code>1_000_000</code>.
+ *
+ * @since 1.1.3
+ */
+public class CurrentTimestampProvider implements ITimestampProvider {
+
+    @Override
+    public long next(String message) {
+        return Utils.Nanoseconds.currentTime();
+    }
+
+    /**
+     * Factory of {@link CurrentTimestampProvider}.
+     */
+    public static class Factory implements ITimestampProviderFactory {
+
+        @Override
+        public ITimestampProvider create() {
+            return new CurrentTimestampProvider();
+        }
+    }
+
+    public static ITimestampProviderFactory factory() {
+        return new Factory();
+    }
+}

src/main/java/pl/mjaron/tinyloki/IExecutor.java

@@ -6,7 +6,7 @@
  *
  * @since 1.0.0
  */
-public interface IExecutor {
+public interface IExecutor extends ILogListener {
 
     /**
      * Performs the configuration before using the object. Called internally by the {@link TinyLoki} constructor.

src/main/java/pl/mjaron/tinyloki/ILogCollector.java

@@ -6,40 +6,15 @@
 public interface ILogCollector {
 
     /**
-     * Called before using the object.
-     *
-     * @param logListener The object used as callback when new log occurs.
-     * @since 1.0.0
-     */
-    void configureLogListener(ILogListener logListener);
-
-    /**
-     * Called before using the object.
-     *
-     * @param bufferingManager The object which determines sent data buffering policy.
-     * @since 1.0.0
-     */
-    void configureBufferingManager(IBuffering bufferingManager);
-
-    /**
-     * If label settings are not <code>null</code>, the structured metadata is supported, else structured metadata should not be collected.
+     * Called internally by {@link TinyLoki} to configure the object before using it.
      *
+     * @param logListener                     The object used as callback when new log occurs.
+     * @param bufferingManager                The object which determines sent data buffering policy.
      * @param structuredMetadataLabelSettings Settings for structured metadata validation.
-     * @since 1.1.0
-     */
-    void configureStructuredMetadata(LabelSettings structuredMetadataLabelSettings);
-
-    /**
-     * Configures all values in single command.
-     *
-     * @param logListener      The object used as callback when new log occurs.
-     * @param bufferingManager The object which determines sent data buffering policy.
-     * @since 1.0.0
+     * @param timestampProviderFactory        The factory creating {@link ITimestampProvider} objects.
+     * @since 1.1.3
      */
-    default void configure(ILogListener logListener, IBuffering bufferingManager) {
-        configureLogListener(logListener);
-        configureBufferingManager(bufferingManager);
-    }
+    void configure(ILogListener logListener, IBuffering bufferingManager, LabelSettings structuredMetadataLabelSettings, ITimestampProviderFactory timestampProviderFactory);
 
     /**
      * Creates a new stream.

src/main/java/pl/mjaron/tinyloki/ILogStream.java

@@ -10,25 +10,32 @@
  */
 public interface ILogStream {
 
+    /**
+     * Specifies that the stream should determine the current timestamp itself.
+     *
+     * @since 1.1.3
+     */
+    long TIMESTAMP_NONE = -1;
+
     /**
      * Thread-safe method used to write log messages to a stream.
      *
-     * @param timestampMs        Usually System.currentTimeMillis().
+     * @param timestampNs        Log timestamp in <b>nanoseconds</b>.
      * @param line               Log content.
      * @param structuredMetadata The optional (nullable) {@link Labels} containing <a href="https://grafana.com/docs/loki/v3.4.x/get-started/labels/structured-metadata/">structured metadata</a> values.
-     * @since 1.1.0
+     * @since 1.1.3
      */
-    void log(final long timestampMs, final String line, final Labels structuredMetadata);
+    void log(final long timestampNs, final String line, final Labels structuredMetadata);
 
     /**
      * Thread-safe log line with custom time.
      *
-     * @param timestampMs Log timestamp in milliseconds.
+     * @param timestampNs Log timestamp in <b>nanoseconds</b>.
      * @param line        Log content.
-     * @since 1.1.0
+     * @since 1.1.3
      */
-    default void log(final long timestampMs, final String line) {
-        log(timestampMs, line, null);
+    default void log(final long timestampNs, final String line) {
+        log(timestampNs, line, null);
     }
 
     /**
@@ -37,7 +44,7 @@ default void log(final long timestampMs, final String line) {
      * @param line Log content.
      */
     default void log(final String line) {
-        log(System.currentTimeMillis(), line, null);
+        log(TIMESTAMP_NONE, line, null);
     }
 
     /**
@@ -48,7 +55,7 @@ default void log(final String line) {
      * @since 1.1.0
      */
     default void log(final String line, final Labels structuredMetadata) {
-        log(System.currentTimeMillis(), line, structuredMetadata);
+        log(TIMESTAMP_NONE, line, structuredMetadata);
     }
 
     /**

src/main/java/pl/mjaron/tinyloki/ITimestampProvider.java

@@ -0,0 +1,34 @@
+package pl.mjaron.tinyloki;
+
+/**
+ * Provides the log occurrence timestamp in <b>nanoseconds</b>.
+ * <p>
+ * <b>Motivation</b><br>
+ * Grafana Loki treats two identical log messages with the same timestamp as a single log.
+ * The second log is ignored, even if the structured metadata is different.
+ * <p>
+ * User may want to differentiate the timestamps.
+ * This is the reason why the creation of timestamp values is configurable.
+ *
+ * @see ITimestampProviderFactory
+ * @see CurrentTimestampProvider
+ * @see IncrementingTimestampProvider
+ * @since 1.1.3
+ */
+public interface ITimestampProvider {
+
+    /**
+     * Provides log timestamp value in nanoseconds.
+     * The point is to optionally avoid two identical log messages
+     * having the same timestamp.
+     * <p>
+     * <b>Thread safety</b><br>
+     * This method may not be thread safe and should be synchronized internally
+     * by {@link ILogStream} or {@link ILogCollector} implementation.
+     *
+     * @param message Original log message. It may be analyzed by any {@link ITimestampProvider} implementation.
+     * @return The calculated timestamp.
+     * @since 1.1.3
+     */
+    long next(String message);
+}

src/main/java/pl/mjaron/tinyloki/ITimestampProviderFactory.java

@@ -0,0 +1,52 @@
+package pl.mjaron.tinyloki;
+
+/**
+ * Creates new instance of {@link ITimestampProvider} implementation.
+ *
+ * @since 1.1.3
+ */
+public interface ITimestampProviderFactory {
+
+    /**
+     * Creates new instance of {@link ITimestampProvider} implementation.
+     *
+     * @return New instance of {@link ITimestampProvider} implementation.
+     * @since 1.1.3
+     */
+    ITimestampProvider create();
+
+    /**
+     * Provides given <code>factory</code> or the {@link #getDefault() default} implementation if parameter is <code>null</code>.
+     *
+     * @param factory The original factory or <code>null</code> value.
+     * @return If not a <code>null</code>, the <code>factory</code> parameter, otherwise the {@link #getDefault() default} implementation.
+     * @since 1.1.3
+     */
+    static ITimestampProviderFactory orDefault(ITimestampProviderFactory factory) {
+        if (factory == null) {
+            return getDefault();
+        }
+        return factory;
+    }
+
+    /**
+     * @return New factory of {@link CurrentTimestampProvider}.
+     */
+    static ITimestampProviderFactory current() {
+        return CurrentTimestampProvider.factory();
+    }
+
+    /**
+     * @return New factory of {@link IncrementingTimestampProvider}.
+     */
+    static ITimestampProviderFactory incrementing() {
+        return IncrementingTimestampProvider.factory();
+    }
+
+    /**
+     * @return {@link CurrentTimestampProvider}.
+     */
+    static ITimestampProviderFactory getDefault() {
+        return current();
+    }
+}

src/main/java/pl/mjaron/tinyloki/IncrementingTimestampProvider.java

@@ -0,0 +1,43 @@
+package pl.mjaron.tinyloki;
+
+/**
+ * Provides uniqueness of logged time for first 999999 logs to the same
+ * millisecond with little impact on logging performance.
+ *
+ * @since 1.1.3
+ */
+public class IncrementingTimestampProvider implements ITimestampProvider {
+
+    private long lastCurrentTimeMillis = 0;
+    private int nanoseconds = 0;
+
+    @Override
+    public long next(String message) {
+        final long current = System.currentTimeMillis();
+        if (current == lastCurrentTimeMillis) {
+            if (nanoseconds < 999_999) {
+                ++nanoseconds;
+            }
+            return lastCurrentTimeMillis * 1_000_000 + nanoseconds;
+        } else {
+            lastCurrentTimeMillis = current;
+            nanoseconds = 0;
+            return lastCurrentTimeMillis * 1_000_000;
+        }
+    }
+
+    /**
+     * Factory of {@link IncrementingTimestampProvider}.
+     */
+    public static class Factory implements ITimestampProviderFactory {
+
+        @Override
+        public ITimestampProvider create() {
+            return new IncrementingTimestampProvider();
+        }
+    }
+
+    public static ITimestampProviderFactory factory() {
+        return new Factory();
+    }
+}

src/main/java/pl/mjaron/tinyloki/JsonLogCollector.java

@@ -17,20 +17,14 @@ public class JsonLogCollector implements ILogCollector {
     private ILogListener logListener = null;
     private IBuffering bufferingManager = null;
     private LabelSettings structuredMetadataLabelSettings = null;
+    private ITimestampProvider timestampProvider = null;
 
     @Override
-    public void configureLogListener(final ILogListener logListener) {
+    public void configure(ILogListener logListener, IBuffering bufferingManager, LabelSettings structuredMetadataLabelSettings, ITimestampProviderFactory timestampProviderFactory) {
         this.logListener = logListener;
-    }
-
-    @Override
-    public void configureBufferingManager(final IBuffering bufferingManager) {
         this.bufferingManager = bufferingManager;
-    }
-
-    @Override
-    public void configureStructuredMetadata(LabelSettings structuredMetadataLabelSettings) {
         this.structuredMetadataLabelSettings = structuredMetadataLabelSettings;
+        this.timestampProvider = ITimestampProviderFactory.orDefault(timestampProviderFactory).create();
     }
 
     /**
@@ -126,13 +120,16 @@ public String contentType() {
         return CONTENT_TYPE;
     }
 
-    synchronized void logImplementation(JsonLogStream stream, final long timestampMs, final String line, final Labels structuredMetadata) {
+    synchronized void logImplementation(JsonLogStream stream, long timestamp, final String line, final Labels structuredMetadata) {
         final int logCandidateSize = 25 + line.length();
         if (!bufferingManager.beforeLog(logCandidateSize)) {
             return;
         }
 
-        final int acceptedLogSize = stream.logUnsafe(timestampMs, line, structuredMetadata);
+        if (timestamp == ILogStream.TIMESTAMP_NONE) {
+            timestamp = timestampProvider.next(line);
+        }
+        final int acceptedLogSize = stream.logUnsafe(timestamp, line, structuredMetadata);
 
         bufferingManager.logAccepted(acceptedLogSize);
         ++logEntriesCount;

src/main/java/pl/mjaron/tinyloki/JsonLogStream.java

@@ -42,15 +42,15 @@ public JsonLogStream(final JsonLogCollector collector, final Labels labels) {
         initialSequenceWithHeaders = b.toString();
     }
 
-    int logUnsafe(final long timestampMs, final String line, final Labels structuredMetadata) {
+    int logUnsafe(final long timestamp, final String line, final Labels structuredMetadata) {
         final int beforeSize = b.length();
         if (cachedLogsCount != 0) {
             b.append(',');
         }
         ++cachedLogsCount;
         b.append("[\"");
-        b.append(timestampMs);
-        b.append("000000\",\"");
+        b.append(timestamp);
+        b.append("\",\"");
         Utils.escapeJsonString(b, line);
         b.append('"');
 
@@ -78,8 +78,8 @@ int logUnsafe(final long timestampMs, final String line, final Labels structured
     }
 
     @Override
-    public void log(final long timestampMs, final String line, final Labels structuredMetadata) {
-        collector.logImplementation(this, timestampMs, line, structuredMetadata);
+    public void log(final long timestampNs, final String line, final Labels structuredMetadata) {
+        collector.logImplementation(this, timestampNs, line, structuredMetadata);
     }
 
     @Override