From bc4211272d9155fce0678528b637caed00458908 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Tue, 6 Jul 2021 18:38:30 +0200 Subject: [PATCH] Re #60: Refactored ByteBufferDestinations to be a factor of buffers with thread local caching and configurable buffer size. --- .../tkowalcz/tjahzi/log4j2/AppenderLogic.java | 16 +-- .../ByteBufferDestinationRepository.java | 33 ++++++ .../tkowalcz/tjahzi/log4j2/LokiAppender.java | 4 +- .../tjahzi/log4j2/LokiAppenderBuilder.java | 10 ++ ....java => TjahziByteBufferDestination.java} | 19 ++-- .../tjahzi/log4j2/AppenderLogicTest.java | 6 +- .../ByteBufferDestinationRepositoryTest.java | 100 ++++++++++++++++++ 7 files changed, 165 insertions(+), 23 deletions(-) create mode 100644 log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinationRepository.java rename log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/{ByteBufferDestinations.java => TjahziByteBufferDestination.java} (65%) create mode 100644 log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinationRepositoryTest.java 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 0fa33ab..fc42c4f 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 @@ -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 { private static final Logger LOGGER = StatusLogger.getLogger(); - private static final Map EMPTY_MAP = new HashMap<>(); - private final LoggingSystem loggingSystem; private final String logLevelLabel; private final TjahziLogger logger; private final Map lokiLabels; + private final ByteBufferDestinationRepository destinationRepository; public AppenderLogic( LoggingSystem loggingSystem, String logLevelLabel, - Map lokiLabels + Map 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 layout, LogEvent event ) { - ByteBufferDestinations destination = ByteBufferDestinations.threadLocal(); - destination.initialize(this, event); + TjahziByteBufferDestination destination = destinationRepository.threadLocalDestination( + this, + event + ); layout.encode(event, destination); destination.drainRemaining(); diff --git a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinationRepository.java b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinationRepository.java new file mode 100644 index 0000000..3caa8d0 --- /dev/null +++ b/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 threadLocal = new ThreadLocal<>(); + private final int maxLogLineSizeBytes; + + public ByteBufferDestinationRepository(int maxLogLineSizeBytes) { + this.maxLogLineSizeBytes = maxLogLineSizeBytes; + } + + public TjahziByteBufferDestination threadLocalDestination( + BiConsumer drain, + LogEvent context + ) { + TjahziByteBufferDestination result = threadLocal.get(); + + if (result == null) { + result = new TjahziByteBufferDestination(maxLogLineSizeBytes); + threadLocal.set(result); + } + + result.initialize(drain, context); + return result; + } +} diff --git a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java index 947d5be..a462b90 100644 --- a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java +++ b/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 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 ); } diff --git a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java index 9e8bf6e..2c61e09 100644 --- a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java +++ b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderBuilder.java @@ -34,6 +34,7 @@ public class LokiAppenderBuilder> 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> 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> extends Abstr isUseOffHeapBuffer() ); + int maxLogLineSizeBytes = Math.toIntExact(getMaxLogLineSizeKilobytes() * BYTES_IN_KILOBYTE); return new LokiAppender( getName(), getLayout(), @@ -140,6 +145,7 @@ public class LokiAppenderBuilder> extends Abstr getPropertyArray(), logLevelLabel, labelsDescriptor.getDynamicLabels(), + maxLogLineSizeBytes, loggingSystem, monitoringModuleWrapper ); @@ -224,6 +230,10 @@ public class LokiAppenderBuilder> extends Abstr return batchWait; } + public long getMaxLogLineSizeKilobytes() { + return maxLogLineSizeKilobytes; + } + public void setMaxRequestsInFlight(int maxRequestsInFlight) { this.maxRequestsInFlight = maxRequestsInFlight; } diff --git a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinations.java b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/TjahziByteBufferDestination.java similarity index 65% rename from log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinations.java rename to log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/TjahziByteBufferDestination.java index b2b0cfa..a6227db 100644 --- a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinations.java +++ b/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 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 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 drain, LogEvent context) { this.context = context; this.drain = drain; + + buffer.clear(); } } 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 cefdc9a..62eda7e 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 @@ -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(); diff --git a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinationRepositoryTest.java b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/ByteBufferDestinationRepositoryTest.java new file mode 100644 index 0000000..a8d9dbb --- /dev/null +++ b/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 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 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 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 drain = (logEvent, byteBuffer) -> { + }; + MutableLogEvent logEvent = new MutableLogEvent(); + + AtomicReference captor1 = new AtomicReference<>(); + AtomicReference 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()); + } +}