Re #60: Refactored ByteBufferDestinations to be a factor of buffers with thread local caching and configurable buffer size.

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

@@ -12,7 +12,6 @@ import pl.tkowalcz.tjahzi.log4j2.labels.LabelPrinter;
 
 import java.io.Serializable;
 import java.nio.ByteBuffer;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
@@ -21,33 +20,36 @@ public class AppenderLogic implements BiConsumer<LogEvent, ByteBuffer> {
 
     private static final Logger LOGGER = StatusLogger.getLogger();
 
-    private static final Map<String, CharSequence> EMPTY_MAP = new HashMap<>();
-
     private final LoggingSystem loggingSystem;
 
     private final String logLevelLabel;
     private final TjahziLogger logger;
 
     private final Map<String, LabelPrinter> lokiLabels;
+    private final ByteBufferDestinationRepository destinationRepository;
 
     public AppenderLogic(
             LoggingSystem loggingSystem,
             String logLevelLabel,
-            Map<String, LabelPrinter> lokiLabels
+            Map<String, LabelPrinter> lokiLabels,
+            int maxLogLineSizeBytes
     ) {
         this.loggingSystem = loggingSystem;
         this.logLevelLabel = logLevelLabel;
 
-        logger = loggingSystem.createLogger();
+        this.logger = loggingSystem.createLogger();
         this.lokiLabels = lokiLabels;
+        this.destinationRepository = new ByteBufferDestinationRepository(maxLogLineSizeBytes);
     }
 
     public void append(
             Layout<? extends Serializable> layout,
             LogEvent event
     ) {
-        ByteBufferDestinations destination = ByteBufferDestinations.threadLocal();
-        destination.initialize(this, event);
+        TjahziByteBufferDestination destination = destinationRepository.threadLocalDestination(
+                this,
+                event
+        );
 
         layout.encode(event, destination);
         destination.drainRemaining();

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

@@ -0,0 +1,33 @@
+package pl.tkowalcz.tjahzi.log4j2;
+
+import org.apache.logging.log4j.core.LogEvent;
+
+import java.nio.ByteBuffer;
+import java.util.function.BiConsumer;
+
+public class ByteBufferDestinationRepository {
+
+    public static final int DEFAULT_MAX_LINE_SIZE_BYTES = 10 * 1024;
+
+    private final ThreadLocal<TjahziByteBufferDestination> threadLocal = new ThreadLocal<>();
+    private final int maxLogLineSizeBytes;
+
+    public ByteBufferDestinationRepository(int maxLogLineSizeBytes) {
+        this.maxLogLineSizeBytes = maxLogLineSizeBytes;
+    }
+
+    public TjahziByteBufferDestination threadLocalDestination(
+            BiConsumer<LogEvent, ByteBuffer> drain,
+            LogEvent context
+    ) {
+        TjahziByteBufferDestination result = threadLocal.get();
+
+        if (result == null) {
+            result = new TjahziByteBufferDestination(maxLogLineSizeBytes);
+            threadLocal.set(result);
+        }
+
+        result.initialize(drain, context);
+        return result;
+    }
+}

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

@@ -46,6 +46,7 @@ public class LokiAppender extends AbstractAppender {
             Property[] properties,
             String logLevelLabel,
             Map<String, LabelPrinter> lokiLabels,
+            int maxLogLineSizeKilobytes,
             LoggingSystem loggingSystem,
             MutableMonitoringModuleWrapper monitoringModuleWrapper
     ) {
@@ -63,7 +64,8 @@ public class LokiAppender extends AbstractAppender {
         this.appenderLogic = new AppenderLogic(
                 loggingSystem,
                 logLevelLabel,
-                lokiLabels
+                lokiLabels,
+                maxLogLineSizeKilobytes
         );
     }
 

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

@@ -34,6 +34,7 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
     private static final Logger LOGGER = StatusLogger.getLogger();
 
     private static final int BYTES_IN_MEGABYTE = 1024 * 1024;
+    private static final int BYTES_IN_KILOBYTE = 1024;
 
     @PluginBuilderAttribute
     @Required(message = "No Loki address provided for LokiAppender")
@@ -73,6 +74,9 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
     @PluginBuilderAttribute
     private final long batchWait = 5;
 
+    @PluginBuilderAttribute
+    private final int maxLogLineSizeKilobytes = 10;
+
     @PluginBuilderAttribute
     private int maxRequestsInFlight = 100;
 
@@ -132,6 +136,7 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
                 isUseOffHeapBuffer()
         );
 
+        int maxLogLineSizeBytes = Math.toIntExact(getMaxLogLineSizeKilobytes() * BYTES_IN_KILOBYTE);
         return new LokiAppender(
                 getName(),
                 getLayout(),
@@ -140,6 +145,7 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
                 getPropertyArray(),
                 logLevelLabel,
                 labelsDescriptor.getDynamicLabels(),
+                maxLogLineSizeBytes,
                 loggingSystem,
                 monitoringModuleWrapper
         );
@@ -224,6 +230,10 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
         return batchWait;
     }
 
+    public long getMaxLogLineSizeKilobytes() {
+        return maxLogLineSizeKilobytes;
+    }
+
     public void setMaxRequestsInFlight(int maxRequestsInFlight) {
         this.maxRequestsInFlight = maxRequestsInFlight;
     }

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

@@ -7,24 +7,15 @@ import org.apache.logging.log4j.core.layout.ByteBufferDestinationHelper;
 import java.nio.ByteBuffer;
 import java.util.function.BiConsumer;
 
-public class ByteBufferDestinations implements ByteBufferDestination {
+public class TjahziByteBufferDestination implements ByteBufferDestination {
 
-    private static final ThreadLocal<ByteBufferDestinations> THREAD_LOCAL = ThreadLocal.withInitial(ByteBufferDestinations::new);
-
-    /**
-     * 10kb should be enough for everyone. Even in case of 1000s of threads we will allocate at most 10s of MB of buffers.
-     * Larger messages will be fragmented.
-     */
-    private final ByteBuffer buffer = ByteBuffer.allocate(10 * 1024);
+    private final ByteBuffer buffer;
 
     private BiConsumer<LogEvent, ByteBuffer> drain;
     private LogEvent context;
 
-    public static ByteBufferDestinations threadLocal() {
-        ByteBufferDestinations result = THREAD_LOCAL.get();
-        result.getByteBuffer().clear();
-
-        return result;
+    public TjahziByteBufferDestination(int maxLogLineSizeBytes) {
+        buffer = ByteBuffer.allocate(maxLogLineSizeBytes);
     }
 
     public void drainRemaining() {
@@ -60,5 +51,7 @@ public class ByteBufferDestinations implements ByteBufferDestination {
     public void initialize(BiConsumer<LogEvent, ByteBuffer> drain, LogEvent context) {
         this.context = context;
         this.drain = drain;
+
+        buffer.clear();
     }
 }

log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/AppenderLogicTest.java

@@ -33,7 +33,8 @@ class AppenderLogicTest {
         AppenderLogic logic = new AppenderLogic(
                 loggingSystem,
                 logLevelLabel,
-                Map.of()
+                Map.of(),
+                ByteBufferDestinationRepository.DEFAULT_MAX_LINE_SIZE_BYTES
         );
 
         MutableLogEvent logEvent = new MutableLogEvent();
@@ -90,7 +91,8 @@ class AppenderLogicTest {
         AppenderLogic logic = new AppenderLogic(
                 loggingSystem,
                 logLevelLabel,
-                dynamicLabels
+                dynamicLabels,
+                ByteBufferDestinationRepository.DEFAULT_MAX_LINE_SIZE_BYTES
         );
 
         MutableLogEvent logEvent = new MutableLogEvent();

log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinationRepositoryTest.java

@@ -0,0 +1,100 @@
+package pl.tkowalcz.tjahzi.log4j2;
+
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.impl.MutableLogEvent;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ByteBufferDestinationRepositoryTest {
+
+    @Test
+    void shouldInitializeBufferToDesiredSize() {
+        // Given
+        int capacity = 42;
+        ByteBufferDestinationRepository repository = new ByteBufferDestinationRepository(capacity);
+
+        BiConsumer<LogEvent, ByteBuffer> drain = (logEvent, byteBuffer) -> {
+        };
+        MutableLogEvent logEvent = new MutableLogEvent();
+
+        // When
+        TjahziByteBufferDestination destination = repository.threadLocalDestination(drain, logEvent);
+
+        // Then
+        assertThat(destination.getByteBuffer().capacity()).isEqualTo(capacity);
+        assertThat(destination.getByteBuffer().remaining()).isEqualTo(capacity);
+    }
+
+    @Test
+    void shouldClearBufferBeforeUse() {
+        // Given
+        int capacity = 42;
+        ByteBufferDestinationRepository repository = new ByteBufferDestinationRepository(capacity);
+
+        BiConsumer<LogEvent, ByteBuffer> drain = (logEvent, byteBuffer) -> {
+        };
+        MutableLogEvent logEvent = new MutableLogEvent();
+
+        TjahziByteBufferDestination destination = repository.threadLocalDestination(drain, logEvent);
+        destination.writeBytes("blah".getBytes(), 0, 4);
+
+        // When
+        TjahziByteBufferDestination actual = repository.threadLocalDestination(drain, logEvent);
+
+        // Then
+        assertThat(actual.getByteBuffer().remaining()).isEqualTo(capacity);
+        assertThat(actual.getByteBuffer().capacity()).isEqualTo(capacity);
+    }
+
+    @Test
+    void shouldReturnSameBufferInsideSameThread() {
+        // Given
+        int capacity = 42;
+        ByteBufferDestinationRepository repository = new ByteBufferDestinationRepository(capacity);
+
+        BiConsumer<LogEvent, ByteBuffer> drain = (logEvent, byteBuffer) -> {
+        };
+        MutableLogEvent logEvent = new MutableLogEvent();
+
+        // When
+        TjahziByteBufferDestination actual1 = repository.threadLocalDestination(drain, logEvent);
+        TjahziByteBufferDestination actual2 = repository.threadLocalDestination(drain, logEvent);
+
+        // Then
+        assertThat(actual1).isSameAs(actual2);
+    }
+
+    @Test
+    void shouldReturnDifferentSerializerInsideDifferentThreads() throws InterruptedException {
+        // Given
+        int capacity = 42;
+        ByteBufferDestinationRepository repository = new ByteBufferDestinationRepository(capacity);
+
+        BiConsumer<LogEvent, ByteBuffer> drain = (logEvent, byteBuffer) -> {
+        };
+        MutableLogEvent logEvent = new MutableLogEvent();
+
+        AtomicReference<TjahziByteBufferDestination> captor1 = new AtomicReference<>();
+        AtomicReference<TjahziByteBufferDestination> captor2 = new AtomicReference<>();
+
+        // When
+        Thread thread1 = new Thread(() -> captor1.set(repository.threadLocalDestination(drain, logEvent)));
+        thread1.start();
+        thread1.join();
+
+        Thread thread2 = new Thread(() -> captor2.set(repository.threadLocalDestination(drain, logEvent)));
+        thread2.start();
+        thread2.join();
+
+        // Then
+        assertThat(captor1.get()).isNotNull();
+        assertThat(captor2.get()).isNotNull();
+
+        assertThat(captor1.get()).isNotSameAs(captor2.get());
+    }
+}