diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/LabelSerializers.java b/core/src/main/java/pl/tkowalcz/tjahzi/LabelSerializers.java index e691e19..d8e57ea 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/LabelSerializers.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/LabelSerializers.java @@ -4,7 +4,6 @@ public class LabelSerializers { private static final ThreadLocal THREAD_LOCAL = ThreadLocal.withInitial(LabelSerializer::new); - public static LabelSerializer threadLocal() { LabelSerializer result = THREAD_LOCAL.get(); result.clear(); diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java b/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java index 1651a09..c5447af 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java @@ -1,65 +1,46 @@ package pl.tkowalcz.tjahzi; -import io.netty.buffer.PooledByteBufAllocator; -import org.agrona.MutableDirectBuffer; import org.agrona.concurrent.Agent; -import org.agrona.concurrent.MessageHandler; import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer; import pl.tkowalcz.tjahzi.http.NettyHttpClient; import java.io.IOException; -import java.time.Clock; -import java.util.Map; -public class LogBufferAgent implements Agent, MessageHandler { +public class LogBufferAgent implements Agent { public static final int MAX_MESSAGES_TO_RETRIEVE = 100; - private final Clock clock; - private final ManyToOneRingBuffer logBuffer; private final NettyHttpClient httpClient; - private final long batchSize; - private final long batchWaitMillis; private final OutputBuffer outputBuffer; - private final LogBufferTranscoder logBufferTranscoder; - private long timeoutDeadline; + private final LogBufferMessageHandler messageHandler; + private final TimeCappedBatchingStrategy batchStrategy; public LogBufferAgent( - Clock clock, + TimeCappedBatchingStrategy batchStrategy, ManyToOneRingBuffer logBuffer, + OutputBuffer outputBuffer, NettyHttpClient httpClient, - long batchSizeBytes, - long batchWaitMillis, - Map staticLabels - ) { - this.clock = clock; + LogBufferMessageHandler messageHandler) { + this.batchStrategy = batchStrategy; this.logBuffer = logBuffer; - this.batchSize = batchSizeBytes; - this.batchWaitMillis = batchWaitMillis; this.httpClient = httpClient; + this.messageHandler = messageHandler; - this.outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); - this.logBufferTranscoder = new LogBufferTranscoder( - staticLabels, - logBuffer.buffer() - ); + this.outputBuffer = outputBuffer; } - @Override - public int doWork() throws IOException { - int workDone = logBuffer.read(this, MAX_MESSAGES_TO_RETRIEVE); + private int doWork(boolean isTerminating) throws IOException { + int workDone = logBuffer.read(messageHandler, MAX_MESSAGES_TO_RETRIEVE); - long currentTimeMillis = clock.millis(); - if (exceededBatchSizeThreshold() || exceededWaitTimeThreshold(currentTimeMillis)) { + if (isTerminating || batchStrategy.shouldProceed()) { try { httpClient.log(outputBuffer); } finally { outputBuffer.clear(); - timeoutDeadline = currentTimeMillis + batchWaitMillis; } } @@ -67,37 +48,27 @@ public class LogBufferAgent implements Agent, MessageHandler { } @Override - public void onMessage( - int msgTypeId, - MutableDirectBuffer buffer, - int index, - int length - ) { - if (msgTypeId == TjahziLogger.LOG_MESSAGE_TYPE_ID) { - processMessage(buffer, index); - } else { - // Ignore - } + public int doWork() throws IOException { + return doWork(false); + } + + @Override + public void onClose() { + batchStrategy.initShutdown(); + + int workDone = Integer.MAX_VALUE; + do { + try { + workDone = doWork(true); + } catch (Exception ignore) { + // We are shutting down, what else can we do? + } + + } while (workDone != 0 && batchStrategy.shouldContinueShutdown()); } @Override public String roleName() { return "ReadingLogBufferAndSendingHttp"; } - - private boolean exceededWaitTimeThreshold(long currentTimeMillis) { - return currentTimeMillis > timeoutDeadline & outputBuffer.getBytesPending() > 0; - } - - private boolean exceededBatchSizeThreshold() { - return outputBuffer.getBytesPending() > batchSize; - } - - private void processMessage(MutableDirectBuffer buffer, int index) { - logBufferTranscoder.deserializeIntoByteBuf( - buffer, - index, - outputBuffer - ); - } } diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferMessageHandler.java b/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferMessageHandler.java new file mode 100644 index 0000000..f916e57 --- /dev/null +++ b/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferMessageHandler.java @@ -0,0 +1,48 @@ +package pl.tkowalcz.tjahzi; + +import org.agrona.MutableDirectBuffer; +import org.agrona.concurrent.MessageHandler; +import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer; + +import java.util.Map; + +public class LogBufferMessageHandler implements MessageHandler { + + private final LogBufferTranscoder logBufferTranscoder; + private final OutputBuffer outputBuffer; + + public LogBufferMessageHandler( + ManyToOneRingBuffer logBuffer, + Map staticLabels, + OutputBuffer outputBuffer + ) { + this.outputBuffer = outputBuffer; + + this.logBufferTranscoder = new LogBufferTranscoder( + staticLabels, + logBuffer.buffer() + ); + } + + @Override + public void onMessage( + int msgTypeId, + MutableDirectBuffer buffer, + int index, + int length + ) { + if (msgTypeId == TjahziLogger.LOG_MESSAGE_TYPE_ID) { + processMessage(buffer, index); + } else { + // Ignore + } + } + + private void processMessage(MutableDirectBuffer buffer, int index) { + logBufferTranscoder.deserializeIntoByteBuf( + buffer, + index, + outputBuffer + ); + } +} diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategy.java b/core/src/main/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategy.java new file mode 100644 index 0000000..705ceda --- /dev/null +++ b/core/src/main/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategy.java @@ -0,0 +1,59 @@ +package pl.tkowalcz.tjahzi; + +import java.time.Clock; + +public class TimeCappedBatchingStrategy { + + private final Clock clock; + private final OutputBuffer outputBuffer; + + private final long batchSize; + private final long batchWaitMillis; + private final long shutdownTimeout; + + private long timeoutDeadline; + private long shutdownDeadline; + + public TimeCappedBatchingStrategy( + Clock clock, + OutputBuffer outputBuffer, + long batchSize, + long batchWaitMillis, + long shutdownTimeout + ) { + this.clock = clock; + this.outputBuffer = outputBuffer; + this.batchSize = batchSize; + this.batchWaitMillis = batchWaitMillis; + this.shutdownTimeout = shutdownTimeout; + + this.timeoutDeadline = clock.millis() + batchWaitMillis; + } + + public boolean shouldProceed() { + long currentTimeMillis = clock.millis(); + + boolean shouldProceed = exceededBatchSizeThreshold() || exceededWaitTimeThreshold(currentTimeMillis); + if (shouldProceed) { + timeoutDeadline = currentTimeMillis + batchWaitMillis; + } + + return shouldProceed; + } + + public boolean shouldContinueShutdown() { + return clock.millis() < shutdownDeadline; + } + + public void initShutdown() { + shutdownDeadline = clock.millis() + shutdownTimeout; + } + + private boolean exceededWaitTimeThreshold(long currentTimeMillis) { + return currentTimeMillis > timeoutDeadline & outputBuffer.getBytesPending() > 0; + } + + private boolean exceededBatchSizeThreshold() { + return outputBuffer.getBytesPending() > batchSize; + } +} diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java index 983ed80..5e08450 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java @@ -1,5 +1,6 @@ package pl.tkowalcz.tjahzi; +import io.netty.buffer.PooledByteBufAllocator; import org.agrona.concurrent.AgentRunner; import org.agrona.concurrent.SleepingMillisIdleStrategy; import org.agrona.concurrent.UnsafeBuffer; @@ -30,13 +31,23 @@ public class TjahziInitializer { new UnsafeBuffer(javaBuffer) ); + OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); LogBufferAgent agent = new LogBufferAgent( - monitoringModule.getClock(), + new TimeCappedBatchingStrategy( + monitoringModule.getClock(), + outputBuffer, + batchSizeBytes, + batchWaitMillis, + 10_000 + ), logBuffer, + outputBuffer, httpClient, - batchSizeBytes, - batchWaitMillis, - staticLabels + new LogBufferMessageHandler( + logBuffer, + staticLabels, + outputBuffer + ) ); AgentRunner runner = new AgentRunner( diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java index f7b9f89..c238161 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java @@ -1,5 +1,6 @@ package pl.tkowalcz.tjahzi; +import io.netty.buffer.PooledByteBufAllocator; import org.agrona.concurrent.UnsafeBuffer; import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer; import org.agrona.concurrent.ringbuffer.RingBufferDescriptor; @@ -13,6 +14,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.Map; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Matchers.any; import static org.mockito.Mockito.*; @@ -43,13 +45,24 @@ class LogBufferAgentTest { int sentBatchAfter5kb = 5 * 1024; NettyHttpClient httpClient = mock(NettyHttpClient.class); + + OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); LogBufferAgent agent = new LogBufferAgent( - clock, + new TimeCappedBatchingStrategy( + clock, + outputBuffer, + sentBatchAfter5kb, + waitForever, + 10_000 + ), logBuffer, + outputBuffer, httpClient, - sentBatchAfter5kb, - waitForever, - Map.of() + new LogBufferMessageHandler( + logBuffer, + Map.of(), + outputBuffer + ) ); for (int i = 0; i < 100; i++) { @@ -84,13 +97,23 @@ class LogBufferAgentTest { int sentBatchAfter5kb = 5 * 1024; NettyHttpClient httpClient = mock(NettyHttpClient.class); + OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); LogBufferAgent agent = new LogBufferAgent( - clock, + new TimeCappedBatchingStrategy( + clock, + outputBuffer, + sentBatchAfter5kb, + wait5s, + 10_000 + ), logBuffer, + outputBuffer, httpClient, - sentBatchAfter5kb, - wait5s, - Map.of() + new LogBufferMessageHandler( + logBuffer, + Map.of(), + outputBuffer + ) ); logger.log( @@ -113,13 +136,23 @@ class LogBufferAgentTest { int sentBatchAfter5kb = 5 * 1024; NettyHttpClient httpClient = mock(NettyHttpClient.class); + OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); LogBufferAgent agent = new LogBufferAgent( - clock, + new TimeCappedBatchingStrategy( + clock, + outputBuffer, + sentBatchAfter5kb, + wait5s, + 10_000 + ), logBuffer, + outputBuffer, httpClient, - sentBatchAfter5kb, - wait5s, - Map.of() + new LogBufferMessageHandler( + logBuffer, + Map.of(), + outputBuffer + ) ); // When @@ -152,13 +185,23 @@ class LogBufferAgentTest { NettyHttpClient httpClient = mock(NettyHttpClient.class); SettableClock clock = new SettableClock(); + OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); LogBufferAgent agent = new LogBufferAgent( - clock, + new TimeCappedBatchingStrategy( + clock, + outputBuffer, + sentBatchAfter5kb, + wait5s, + 10_000 + ), logBuffer, + outputBuffer, httpClient, - sentBatchAfter5kb, - wait5s, - Map.of() + new LogBufferMessageHandler( + logBuffer, + Map.of(), + outputBuffer + ) ); // When @@ -181,4 +224,55 @@ class LogBufferAgentTest { // Then verify(httpClient, times(2)).log((OutputBuffer) any()); } + + @Test + void shouldDrainAllMessagesOnClose() throws IOException { + // Given + ManyToOneRingBuffer logBuffer = new ManyToOneRingBuffer( + new UnsafeBuffer( + ByteBuffer.wrap( + new byte[64 * 1024 + RingBufferDescriptor.TRAILER_LENGTH] + ) + ) + ); + + TjahziLogger logger = new TjahziLogger(logBuffer, new StandardMonitoringModule()); + + NettyHttpClient httpClient = mock(NettyHttpClient.class); + SettableClock clock = new SettableClock(); + + OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); + LogBufferAgent agent = new LogBufferAgent( + new TimeCappedBatchingStrategy( + clock, + outputBuffer, + Integer.MAX_VALUE, + Integer.MAX_VALUE, + 10_000 + ), + logBuffer, + outputBuffer, + httpClient, + new LogBufferMessageHandler( + logBuffer, + Map.of(), + outputBuffer + ) + ); + + for (int i = 0; i < LogBufferAgent.MAX_MESSAGES_TO_RETRIEVE * 2 + 1; i++) { + logger.log( + 42L, + LabelSerializerCreator.from(Map.of()), + ByteBuffer.wrap("Test".getBytes()) + ); + } + + // When + agent.onClose(); + + // Then + verify(httpClient, times(4)).log((OutputBuffer) any()); + assertThat(logBuffer.size()).isZero(); + } } diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java index aa64a18..0164184 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java @@ -19,7 +19,6 @@ import java.util.concurrent.TimeUnit; import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.mock; @Testcontainers class ResourcesCleanupOnCloseTest { @@ -38,7 +37,7 @@ class ResourcesCleanupOnCloseTest { wireMockServer.start(); initializer = new TjahziInitializer(); - monitoringModule = mock(StandardMonitoringModule.class); + monitoringModule = new StandardMonitoringModule(); } @AfterEach diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategyTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategyTest.java new file mode 100644 index 0000000..1afdb6f --- /dev/null +++ b/core/src/test/java/pl/tkowalcz/tjahzi/TimeCappedBatchingStrategyTest.java @@ -0,0 +1,186 @@ +package pl.tkowalcz.tjahzi; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufUtil; +import io.netty.buffer.PooledByteBufAllocator; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import pl.tkowalcz.tjahzi.stats.SettableClock; + +import static org.assertj.core.api.Assertions.assertThat; + +class TimeCappedBatchingStrategyTest { + + private OutputBuffer outputBufferWthData; + private OutputBuffer outputBufferWithNoData; + private ByteBuf logLine; + + @BeforeEach + void setUp() { + outputBufferWithNoData = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); + outputBufferWthData = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); + + logLine = PooledByteBufAllocator.DEFAULT.buffer(); + logLine.writeIntLE(4); + ByteBufUtil.writeAscii(logLine, "aaba"); + + outputBufferWthData.addLogLine( + "foobar", + 44L, + logLine.duplicate() + ); + } + + @Test + void shouldProceedWhenWaitTimeIsOver() { + // Given + int batchWaitMillis = 1000; + + SettableClock clock = new SettableClock(); + clock.setMillis(42L); + + TimeCappedBatchingStrategy strategy = new TimeCappedBatchingStrategy( + clock, + outputBufferWthData, + 10, + batchWaitMillis, + 10_000 + ); + + // When + clock.setMillis(clock.millis() + batchWaitMillis + 1); + + // Then + assertThat(strategy.shouldProceed()).isTrue(); + } + + @Test + void shouldNotProceedWhenWaitTimeIsOverButThereIsNothingToSend() { + // Given + int batchWaitMillis = 1000; + + SettableClock clock = new SettableClock(); + clock.setMillis(42L); + + TimeCappedBatchingStrategy strategy = new TimeCappedBatchingStrategy( + clock, + outputBufferWithNoData, + 10, + batchWaitMillis, + 10_000 + ); + + // When + clock.setMillis(clock.millis() + batchWaitMillis); + + // Then + assertThat(strategy.shouldProceed()).isFalse(); + } + + @Test + void shouldNotProceedWhenWaitTimeIsNotOver() { + // Given + int batchSize = 100; + int batchWaitMillis = 1000; + + SettableClock clock = new SettableClock(); + clock.setMillis(42L); + + TimeCappedBatchingStrategy strategy = new TimeCappedBatchingStrategy( + clock, + outputBufferWthData, + batchSize, + batchWaitMillis, + 10_000 + ); + + // When + clock.setMillis(clock.millis() + batchWaitMillis - 10); + + // Then + assertThat(strategy.shouldProceed()).isFalse(); + } + + @Test + void shouldProceedWhenBatchSizeIsReached() { + // Given + int batchWaitMillis = 1000; + int batchSize = 4; + + SettableClock clock = new SettableClock(); + clock.setMillis(42L); + + TimeCappedBatchingStrategy strategy = new TimeCappedBatchingStrategy( + clock, + outputBufferWithNoData, + batchSize, + batchWaitMillis, + 10_000 + ); + + // When + outputBufferWithNoData.addLogLine("foobar", 44L, logLine); + + // Then + assertThat(strategy.shouldProceed()).isTrue(); + } + + @Test + void shouldProceedWhenBatchSizeReachedAndWaitTimeIsOver() { + // Given + int batchWaitMillis = 1000; + int batchSize = 4; + + SettableClock clock = new SettableClock(); + clock.setMillis(42L); + + TimeCappedBatchingStrategy strategy = new TimeCappedBatchingStrategy( + clock, + outputBufferWithNoData, + batchSize, + batchWaitMillis, + 10_000 + ); + + // When + outputBufferWithNoData.addLogLine("foobar", 44L, logLine); + clock.setMillis(clock.millis() + batchWaitMillis); + + // Then + assertThat(strategy.shouldProceed()).isTrue(); + } + + @Test + void shouldAllowShutdownUntilDeadlinePasses() { + // Given + int batchSize = 4; + int batchWaitMillis = 1000; + int shutdownTimeout = 10_000; + + long initialTime = 42L; + + SettableClock clock = new SettableClock(); + clock.setMillis(initialTime); + + TimeCappedBatchingStrategy strategy = new TimeCappedBatchingStrategy( + clock, + outputBufferWthData, + batchSize, + batchWaitMillis, + shutdownTimeout + ); + + // When & Then + strategy.initShutdown(); + assertThat(strategy.shouldContinueShutdown()).isTrue(); + + clock.setMillis(initialTime + 1000); + assertThat(strategy.shouldContinueShutdown()).isTrue(); + + clock.setMillis(initialTime + 9999); + assertThat(strategy.shouldContinueShutdown()).isTrue(); + + clock.setMillis(initialTime + 10_000); + assertThat(strategy.shouldContinueShutdown()).isFalse(); + } +} diff --git a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderReloadTest.java b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderReloadTest.java new file mode 100644 index 0000000..c08bdcb --- /dev/null +++ b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderReloadTest.java @@ -0,0 +1,107 @@ +package pl.tkowalcz.tjahzi.log4j2; + +import io.restassured.RestAssured; +import io.restassured.http.ContentType; +import io.restassured.parsing.Parser; +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 java.nio.file.Files; +import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; + +import static io.restassured.RestAssured.given; +import static org.hamcrest.CoreMatchers.*; + +@Testcontainers +class LokiAppenderReloadTest { + + @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 shouldNotLooseDataWhenConfigChanges() throws Exception { + // Given + System.setProperty("loki.host", loki.getHost()); + System.setProperty("loki.port", loki.getFirstMappedPort().toString()); + + URI uriBefore = getClass() + .getClassLoader() + .getResource("appender-test-reload-before.xml") + .toURI(); + + URI uriAfter = getClass() + .getClassLoader() + .getResource("appender-test-reload-after.xml") + .toURI(); + + URI uri = new URI("file://" + Paths.get(uriBefore).toFile().getParent() + "/appender-test-reload.xml"); + Files.copy(Paths.get(uriBefore), Paths.get(uri), StandardCopyOption.REPLACE_EXISTING); + + ((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false)) + .setConfigLocation(uri); + + String expectedLogLine = "Test"; + + long expectedTimestamp = System.currentTimeMillis(); + Logger logger = LogManager.getLogger(LokiAppenderReloadTest.class); + + // When + logger.info(expectedLogLine); + Files.copy(Paths.get(uriAfter), Paths.get(uri), StandardCopyOption.REPLACE_EXISTING); + + // Then + RestAssured.port = loki.getFirstMappedPort(); + RestAssured.baseURI = "http://" + loki.getHost(); + RestAssured.registerParser("text/plain", Parser.JSON); + + Awaitility + .await() + .atMost(Durations.ONE_MINUTE) + .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() + .log() + .all() + .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(1)) + .body( + "data.result.values", + hasItems( + hasItems( + hasItems( + containsString("INFO LokiAppenderReloadTest - Test") + ) + ) + ) + ) + ); + } +} diff --git a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderShutdownTest.java b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderShutdownTest.java index 7dd0744..67fbcf9 100644 --- a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderShutdownTest.java +++ b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderShutdownTest.java @@ -1,7 +1,12 @@ package pl.tkowalcz.tjahzi.log4j2; +import io.restassured.RestAssured; +import io.restassured.http.ContentType; +import io.restassured.parsing.Parser; 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; @@ -9,13 +14,10 @@ import org.testcontainers.containers.wait.strategy.Wait; import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; -import java.lang.management.ManagementFactory; -import java.lang.management.ThreadInfo; -import java.lang.management.ThreadMXBean; import java.net.URI; -import java.net.URISyntaxException; -import static org.assertj.core.api.Assertions.assertThat; +import static io.restassured.RestAssured.given; +import static org.hamcrest.CoreMatchers.*; @Testcontainers class LokiAppenderShutdownTest { @@ -33,46 +35,62 @@ class LokiAppenderShutdownTest { .withExposedPorts(3100); @Test - void shouldSendData() throws URISyntaxException { + void shouldNotLooseDataWhenShuttingDown() throws Exception { // Given System.setProperty("loki.host", loki.getHost()); System.setProperty("loki.port", loki.getFirstMappedPort().toString()); URI uri = getClass() .getClassLoader() - .getResource("basic-appender-test-configuration.xml") + .getResource("appender-test-shutdown.xml") .toURI(); ((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false)) .setConfigLocation(uri); - // Verify our assumptions that we can find threads started by Tjahzi - Awaitility.await().untilAsserted(() -> { - ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean(); - ThreadInfo[] threadInfos = threadMXBean.dumpAllThreads(false, false); + String expectedLogLine = "Test"; - assertThat(threadInfos) - .extracting(ThreadInfo::getThreadName) - .contains( - "ReadingLogBufferAndSendingHttp", - "tjahzi-worker" - ); - }); + long expectedTimestamp = System.currentTimeMillis(); + Logger logger = LogManager.getLogger(LokiAppenderShutdownTest.class); // When + logger.info(expectedLogLine); LogManager.shutdown(); // Then - Awaitility.await().untilAsserted(() -> { - ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean(); - ThreadInfo[] threadInfos = threadMXBean.dumpAllThreads(false, false); + RestAssured.port = loki.getFirstMappedPort(); + RestAssured.baseURI = "http://" + loki.getHost(); + RestAssured.registerParser("text/plain", Parser.JSON); - assertThat(threadInfos) - .extracting(ThreadInfo::getThreadName) - .doesNotContain( - "ReadingLogBufferAndSendingHttp", - "tjahzi-worker" - ); - }); + Awaitility + .await() + .atMost(Durations.ONE_MINUTE) + .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() + .log() + .all() + .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(1)) + .body( + "data.result.values", + hasItems( + hasItems( + hasItems( + containsString("INFO LokiAppenderShutdownTest - Test") + ) + ) + ) + ) + ); } } diff --git a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/labels/LogLevelLabelConfigurationTest.java b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/labels/LogLevelLabelConfigurationTest.java index 5a0a3d8..beb4e47 100644 --- a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/labels/LogLevelLabelConfigurationTest.java +++ b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/labels/LogLevelLabelConfigurationTest.java @@ -68,7 +68,7 @@ class LogLevelLabelConfigurationTest { given() .contentType(ContentType.URLENC) .urlEncodingEnabled(false) - .formParam("query=%7Bserver%3D%22127.0.0.1%22%7D") + .formParam("query=%7Btest%3D%22shouldWorkWithNoLogLevelConfigured%22%7D") .when() .get("/loki/api/v1/query_range") .then() @@ -77,7 +77,6 @@ class LogLevelLabelConfigurationTest { .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].stream.log_level", nullValue()); }); } @@ -114,7 +113,7 @@ class LogLevelLabelConfigurationTest { given() .contentType(ContentType.URLENC) .urlEncodingEnabled(false) - .formParam("query=%7Bserver%3D%22127.0.0.1%22%7D") + .formParam("query=%7Btest%3D%22shouldSendLogLevelAsConfigured%22%7D") .when() .get("/loki/api/v1/query_range") .then() @@ -123,7 +122,6 @@ class LogLevelLabelConfigurationTest { .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].stream.log_level", equalTo("INFO")); }); } diff --git a/log4j2-appender/src/test/resources/appender-test-large-batches.xml b/log4j2-appender/src/test/resources/appender-test-large-batches.xml index b418b5a..0e937be 100644 --- a/log4j2-appender/src/test/resources/appender-test-large-batches.xml +++ b/log4j2-appender/src/test/resources/appender-test-large-batches.xml @@ -1,9 +1,12 @@ - - + + + + + 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 index 6cd15e7..4ecf519 100644 --- 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 @@ -1,9 +1,12 @@ - - + + + + + diff --git a/log4j2-appender/src/test/resources/appender-test-reload-after.xml b/log4j2-appender/src/test/resources/appender-test-reload-after.xml new file mode 100644 index 0000000..916fd84 --- /dev/null +++ b/log4j2-appender/src/test/resources/appender-test-reload-after.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + ${sys:loki.host} + ${sys:loki.port} + + + + %X{tid} [%t] %d{MM-dd HH:mm:ss.SSS} %5p %c{1} - %m%n%exception{full} + + +
+