From bc4211272d9155fce0678528b637caed00458908 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Tue, 6 Jul 2021 18:38:30 +0200 Subject: [PATCH 1/4] 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()); + } +} From 825844330f85d1f02b302a138e356d7e74e8970e Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Tue, 6 Jul 2021 23:04:08 +0200 Subject: [PATCH 2/4] Re #60: Added test for message fragmentation and maxLogLineSizeKilobytes setting. --- .../tjahzi/log4j2/LokiAppenderBuilder.java | 6 +- .../log4j2/LogLineFragmentationTest.java | 96 +++++++++++++++++++ ...nder-test-large-log-line-fragmentation.xml | 25 +++++ 3 files changed, 126 insertions(+), 1 deletion(-) create mode 100644 log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LogLineFragmentationTest.java create mode 100644 log4j2-appender/src/test/resources/appender-test-large-log-line-fragmentation.xml 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 2c61e09..767bfea 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 @@ -75,7 +75,7 @@ public class LokiAppenderBuilder> extends Abstr private final long batchWait = 5; @PluginBuilderAttribute - private final int maxLogLineSizeKilobytes = 10; + private int maxLogLineSizeKilobytes = 10; @PluginBuilderAttribute private int maxRequestsInFlight = 100; @@ -230,6 +230,10 @@ public class LokiAppenderBuilder> extends Abstr return batchWait; } + public void setMaxLogLineSizeKilobytes(int maxLogLineSizeKilobytes) { + this.maxLogLineSizeKilobytes = maxLogLineSizeKilobytes; + } + public long getMaxLogLineSizeKilobytes() { return maxLogLineSizeKilobytes; } diff --git a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LogLineFragmentationTest.java b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LogLineFragmentationTest.java new file mode 100644 index 0000000..b7cceed --- /dev/null +++ b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LogLineFragmentationTest.java @@ -0,0 +1,96 @@ +package pl.tkowalcz.tjahzi.log4j2; + +import io.restassured.RestAssured; +import io.restassured.http.ContentType; +import io.restassured.parsing.Parser; +import org.apache.commons.lang3.RandomStringUtils; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.awaitility.Awaitility; +import org.awaitility.Durations; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.BindMode; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import java.net.URI; + +import static io.restassured.RestAssured.given; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.CoreMatchers.hasItems; + +@Testcontainers +class LogLineFragmentationTest { + + @Container + public GenericContainer loki = new GenericContainer("grafana/loki:latest") + .withCommand("-config.file=/etc/loki-config.yaml") + .withClasspathResourceMapping("loki-config.yaml", + "/etc/loki-config.yaml", + BindMode.READ_ONLY + ) + .waitingFor( + Wait.forHttp("/ready") + .forPort(3100) + ) + .withExposedPorts(3100); + + @Test + void shouldFragmentMessageOverLimit() throws Exception { + // Given + System.setProperty("loki.host", loki.getHost()); + System.setProperty("loki.port", loki.getFirstMappedPort().toString()); + + URI uri = getClass() + .getClassLoader() + .getResource("appender-test-large-log-line-fragmentation.xml") + .toURI(); + + ((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false)) + .setConfigLocation(uri); + + String unfragmentedLogLine = RandomStringUtils.randomAlphanumeric(100 * 1024 - 1); + String fragmentedLogLine = RandomStringUtils.randomAlphanumeric(110 * 1024); + + long expectedTimestamp = System.currentTimeMillis(); + Logger logger = LogManager.getLogger(LogLineFragmentationTest.class); + + // When + logger.info(unfragmentedLogLine); + logger.info(fragmentedLogLine); + + // Then + RestAssured.port = loki.getFirstMappedPort(); + RestAssured.baseURI = "http://" + loki.getHost(); + RestAssured.registerParser("text/plain", Parser.JSON); + + Awaitility + .await() + .atMost(Durations.TEN_SECONDS) + .pollInterval(Durations.ONE_SECOND) + .ignoreExceptions() + .untilAsserted(() -> given() + .contentType(ContentType.URLENC) + .urlEncodingEnabled(false) + .formParam("&start=" + expectedTimestamp + "&limit=1000&query=%7Bserver%3D%22127.0.0.1%22%7D") + .when() + .get("/loki/api/v1/query_range") + .then() + .statusCode(200) + .body("status", equalTo("success")) + .body("data.result.size()", equalTo(1)) + .body("data.result[0].stream.server", equalTo("127.0.0.1")) + .body("data.result[0].values.size()", equalTo(3)) + .body( + "data.result.values", + hasItems( + hasItems(hasItems(fragmentedLogLine.substring(102_400, 110 * 1024))), + hasItems(hasItems(fragmentedLogLine.substring(0, 102_400))), + hasItems(hasItems(unfragmentedLogLine)) + ) + ) + ); + } +} diff --git a/log4j2-appender/src/test/resources/appender-test-large-log-line-fragmentation.xml b/log4j2-appender/src/test/resources/appender-test-large-log-line-fragmentation.xml new file mode 100644 index 0000000..6cd15e7 --- /dev/null +++ b/log4j2-appender/src/test/resources/appender-test-large-log-line-fragmentation.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + ${sys:loki.host} + ${sys:loki.port} + + 100 + + + + %m + + +
+