diff --git a/README.md b/README.md index 46fae54..e44ad28 100644 --- a/README.md +++ b/README.md @@ -63,9 +63,9 @@ You can compare this with [design principles of Aeron](https://github.com/real-l │ . ▼ ┌────────────────────┐ ┌─────────┐ ┌─────────┐ - . Log │ │ │ Reading │ │ I/O │ - ---------------▶│ Log buffer ├--→---→--┤ agent ├--→---→--┤ thread │ - . │ │ │ thread │ │ (Netty) │ + . Log │ │ │ Log │ │ I/O │ + ---------------▶│ Log buffer ├--→---→--┤ shipper ├--→---→--┤ thread │ + . │ │ │ agent │ │ (Netty) │ └────────────────────┘ └─────────┘ └─────────┘ . ▲ │ diff --git a/core/pom.xml b/core/pom.xml index 9342dbf..6c2359f 100644 --- a/core/pom.xml +++ b/core/pom.xml @@ -1,4 +1,5 @@ - + 4.0.0 diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java b/core/src/main/java/pl/tkowalcz/tjahzi/LogShipperAgent.java similarity index 93% rename from core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java rename to core/src/main/java/pl/tkowalcz/tjahzi/LogShipperAgent.java index c5447af..98b3401 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/LogShipperAgent.java @@ -6,7 +6,7 @@ import pl.tkowalcz.tjahzi.http.NettyHttpClient; import java.io.IOException; -public class LogBufferAgent implements Agent { +public class LogShipperAgent implements Agent { public static final int MAX_MESSAGES_TO_RETRIEVE = 100; @@ -18,7 +18,7 @@ public class LogBufferAgent implements Agent { private final LogBufferMessageHandler messageHandler; private final TimeCappedBatchingStrategy batchStrategy; - public LogBufferAgent( + public LogShipperAgent( TimeCappedBatchingStrategy batchStrategy, ManyToOneRingBuffer logBuffer, OutputBuffer outputBuffer, @@ -69,6 +69,6 @@ public class LogBufferAgent implements Agent { @Override public String roleName() { - return "ReadingLogBufferAndSendingHttp"; + return "LogShipper"; } } diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java index 8931f3b..d71e1d2 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java @@ -23,8 +23,10 @@ public class TjahziInitializer { long batchSizeBytes, long batchWaitMillis, int bufferSizeBytes, + long logShipperWakeupIntervalMillis, long shutdownTimeoutMillis, - boolean offHeap) { + boolean offHeap + ) { bufferSizeBytes = findNearestPowerOfTwo(bufferSizeBytes); ByteBuffer javaBuffer = allocateJavaBuffer(bufferSizeBytes, offHeap); @@ -33,7 +35,7 @@ public class TjahziInitializer { ); OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); - LogBufferAgent agent = new LogBufferAgent( + LogShipperAgent agent = new LogShipperAgent( new TimeCappedBatchingStrategy( monitoringModule.getClock(), outputBuffer, @@ -52,7 +54,7 @@ public class TjahziInitializer { ); AgentRunner runner = new AgentRunner( - new SleepingMillisIdleStrategy(), + new SleepingMillisIdleStrategy(logShipperWakeupIntervalMillis), monitoringModule::addAgentError, null, agent diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java index 920feaf..696bf21 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java @@ -89,6 +89,7 @@ class HeadersTest { 0, 0, 1024 * 1024, + 250, 10_000, false ); @@ -141,6 +142,7 @@ class HeadersTest { 0, 0, 1024 * 1024, + 250, 10_000, false ); @@ -195,6 +197,7 @@ class HeadersTest { 0, 0, 1024 * 1024, + 250, 10_000, false ); diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java index c238161..8744b75 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java @@ -47,7 +47,7 @@ class LogBufferAgentTest { NettyHttpClient httpClient = mock(NettyHttpClient.class); OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); - LogBufferAgent agent = new LogBufferAgent( + LogShipperAgent agent = new LogShipperAgent( new TimeCappedBatchingStrategy( clock, outputBuffer, @@ -98,7 +98,7 @@ class LogBufferAgentTest { NettyHttpClient httpClient = mock(NettyHttpClient.class); OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); - LogBufferAgent agent = new LogBufferAgent( + LogShipperAgent agent = new LogShipperAgent( new TimeCappedBatchingStrategy( clock, outputBuffer, @@ -137,7 +137,7 @@ class LogBufferAgentTest { NettyHttpClient httpClient = mock(NettyHttpClient.class); OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); - LogBufferAgent agent = new LogBufferAgent( + LogShipperAgent agent = new LogShipperAgent( new TimeCappedBatchingStrategy( clock, outputBuffer, @@ -186,7 +186,7 @@ class LogBufferAgentTest { SettableClock clock = new SettableClock(); OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); - LogBufferAgent agent = new LogBufferAgent( + LogShipperAgent agent = new LogShipperAgent( new TimeCappedBatchingStrategy( clock, outputBuffer, @@ -242,7 +242,7 @@ class LogBufferAgentTest { SettableClock clock = new SettableClock(); OutputBuffer outputBuffer = new OutputBuffer(PooledByteBufAllocator.DEFAULT.buffer()); - LogBufferAgent agent = new LogBufferAgent( + LogShipperAgent agent = new LogShipperAgent( new TimeCappedBatchingStrategy( clock, outputBuffer, @@ -260,7 +260,7 @@ class LogBufferAgentTest { ) ); - for (int i = 0; i < LogBufferAgent.MAX_MESSAGES_TO_RETRIEVE * 2 + 1; i++) { + for (int i = 0; i < LogShipperAgent.MAX_MESSAGES_TO_RETRIEVE * 2 + 1; i++) { logger.log( 42L, LabelSerializerCreator.from(Map.of()), diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java index 778b48f..56de92d 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java @@ -30,7 +30,7 @@ import static org.hamcrest.CoreMatchers.hasItems; class LoggingSystemSanityCheckTest { @Container - public GenericContainer loki = new GenericContainer("grafana/loki:latest") + public GenericContainer loki = new GenericContainer("grafana/loki:2.3.0") .withCommand("-config.file=/etc/loki-config.yaml") .withClasspathResourceMapping("loki-config.yaml", "/etc/loki-config.yaml", @@ -67,6 +67,7 @@ class LoggingSystemSanityCheckTest { 0, 0, 1024 * 1024, + 250, 10_000, false ); diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java index 3863ae4..39a4ecc 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java @@ -33,7 +33,7 @@ import static org.hamcrest.CoreMatchers.hasItems; class LoggingSystemTest { @Container - public GenericContainer loki = new GenericContainer("grafana/loki:latest") + public GenericContainer loki = new GenericContainer("grafana/loki:2.3.0") .withCommand("-config.file=/etc/loki-config.yaml") .withClasspathResourceMapping("loki-config.yaml", "/etc/loki-config.yaml", @@ -71,6 +71,7 @@ class LoggingSystemTest { 1024 * 1024, 0, 0, + 250, 10_000, false ); diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java index 767888a..19e42f8 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java @@ -84,6 +84,7 @@ public class ReconnectTest { 10_000, 1000, 1024 * 1024, + 250, 10_000, false ); @@ -134,6 +135,7 @@ public class ReconnectTest { 0, 0, 1024 * 1024, + 250, 10_000, false ); diff --git a/core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java b/core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java index 89c3d6f..8ed4a1c 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java @@ -70,6 +70,7 @@ class ResourcesCleanupOnCloseTest { 0, 0, 1024 * 1024, + 250, 10_000, false ); @@ -83,7 +84,7 @@ class ResourcesCleanupOnCloseTest { assertThat(threadInfos) .extracting(ThreadInfo::getThreadName) .contains( - "ReadingLogBufferAndSendingHttp", + "LogShipper", "tjahzi-worker" ); }); @@ -102,7 +103,7 @@ class ResourcesCleanupOnCloseTest { assertThat(threadInfos) .extracting(ThreadInfo::getThreadName) .doesNotContain( - "ReadingLogBufferAndSendingHttp", + "LogShipper", "tjahzi-worker" ); }); diff --git a/log4j2-appender/README.md b/log4j2-appender/README.md index d94bbef..ffdefb1 100644 --- a/log4j2-appender/README.md +++ b/log4j2-appender/README.md @@ -275,6 +275,13 @@ Like in [promtail configuration](https://grafana.com/docs/loki/latest/clients/promtail/configuration/) `maximum amount of time to wait before sending a batch, even if that batch isn't full` . +#### logShipperWakeupIntervalMillis (optional, default = 10) + +The agent that reads data from log buffer, compresses it and sends to Loki via http is called `LogShipper`. This property +controls how often it wakes up to perform its duties. Other properties control how often the data should be sent to Loki +(`batchSize`, `batchWait`) this one just control how often to wake up and check for these conditions. In versions before +`0.9.17` it was left at default 1ms which caused high CPU usage on some setups. + #### shutdownTimeoutSeconds (optional, default = 10s) On logging system shutdown (or config reload) Tjahzi will flush its internal buffers so that no logs are lost. This 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 f161823..6f515fd 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 @@ -74,6 +74,9 @@ public class LokiAppenderBuilder> extends Abstr @PluginBuilderAttribute private long batchWait = 5; + @PluginBuilderAttribute + private long logShipperWakeupIntervalMillis = 10; + @PluginBuilderAttribute private int shutdownTimeoutSeconds = 10; @@ -141,6 +144,7 @@ public class LokiAppenderBuilder> extends Abstr batchSize, TimeUnit.SECONDS.toMillis(batchWait), bufferSizeBytes, + logShipperWakeupIntervalMillis, TimeUnit.SECONDS.toMillis(shutdownTimeoutSeconds), isUseOffHeapBuffer() ); @@ -247,6 +251,14 @@ public class LokiAppenderBuilder> extends Abstr this.batchWait = batchWait; } + public long getLogShipperWakeupIntervalMillis() { + return logShipperWakeupIntervalMillis; + } + + public void setLogShipperWakeupIntervalMillis(long logShipperWakeupIntervalMillis) { + this.logShipperWakeupIntervalMillis = logShipperWakeupIntervalMillis; + } + public int getShutdownTimeoutSeconds() { return shutdownTimeoutSeconds; } diff --git a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderMonitoringTest.java b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderMonitoringTest.java index f184084..946f9c7 100644 --- a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderMonitoringTest.java +++ b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderMonitoringTest.java @@ -5,6 +5,7 @@ import com.codahale.metrics.MetricRegistry; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.core.LoggerContext; +import org.awaitility.Durations; import org.junit.jupiter.api.Test; import pl.tkowalcz.tjahzi.stats.DropwizardMonitoringModule; @@ -43,9 +44,12 @@ class LokiAppenderMonitoringTest { .get("appender.loki.httpConnectAttempts"); assertThat(connectAttemptsCounter).isNotNull(); - await().untilAsserted(() -> { - assertThat(connectAttemptsCounter.getCount()).isEqualTo(1); - } - ); + await() + .atMost(Durations.ONE_MINUTE) + .pollInterval(Durations.ONE_SECOND) + .untilAsserted(() -> { + assertThat(connectAttemptsCounter.getCount()).isEqualTo(1); + } + ); } } diff --git a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/infra/IntegrationTest.java b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/infra/IntegrationTest.java index 6277826..b2cf198 100644 --- a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/infra/IntegrationTest.java +++ b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/infra/IntegrationTest.java @@ -18,7 +18,7 @@ import java.net.URL; public class IntegrationTest { @Container - public GenericContainer loki = new GenericContainer("grafana/loki:latest") + public GenericContainer loki = new GenericContainer("grafana/loki:2.3.0") .withCommand("-config.file=/etc/loki-config.yaml") .withClasspathResourceMapping("loki-config.yaml", "/etc/loki-config.yaml", @@ -51,10 +51,8 @@ public class IntegrationTest { return loadConfig(uri); } catch (URISyntaxException e) { - Assertions.fail(e); + return Assertions.fail(e); } - - return null; } public static LoggerContext loadConfig(URI uri) { diff --git a/log4j2-appender/src/test/resources/basic-appender-test-configuration.xml b/log4j2-appender/src/test/resources/basic-appender-test-configuration.xml index 6bb8074..09c84d6 100644 --- a/log4j2-appender/src/test/resources/basic-appender-test-configuration.xml +++ b/log4j2-appender/src/test/resources/basic-appender-test-configuration.xml @@ -16,7 +16,7 @@ - %X{ctx:tid} [%t] %d{DEFAULT} %5p %c{1} - %m%n%exception{full} + [%t] %d{DEFAULT} %5p %c{1} - %m%n%exception{full}
diff --git a/logback-appender/README.md b/logback-appender/README.md index d576833..e3e4f9b 100644 --- a/logback-appender/README.md +++ b/logback-appender/README.md @@ -202,4 +202,5 @@ or use Tjahzi optimized efficient, low allocation encoder: | UseOffHeapBuffer | true | Whether Tjahzi should allocate native buffer for `Log buffer` component. We can go into a rabbit hole of divagations what are the implications of this. Most important in our view is that having 10s or 100s of MB of space taken out of heap is not very friendly to garbage collector which might have to occasionally copy it around. | | BatchSize | 100 KB | Like in [promtail configuration](https://grafana.com/docs/loki/latest/clients/promtail/configuration/) `maximum batch size (in bytes) of logs to accumulate before sending the batch to Loki`.| | BatchWait | 5s | Like in [promtail configuration](https://grafana.com/docs/loki/latest/clients/promtail/configuration/) `maximum amount of time to wait before sending a batch, even if that batch isn't full`.| +| logShipperWakeupIntervalMillis | 10 | The agent that reads data from log buffer, compresses it and sends to Loki via http is called `LogShipper`. This property controls how often it wakes up to perform its duties. Other properties control how often the data should be sent to Loki (`batchSize`, `batchWait`) this one just control how often to wake up and check for these conditions. In versions before `0.9.17` it was left at default 1ms which caused high CPU usage on some setups. | | shutdownTimeoutSeconds | 10s | On logging system shutdown (or config reload) Tjahzi will flush its internal buffers so that no logs are lost. This property sets limit on how long to wait for this to complete before proceeding with shutdown. | diff --git a/logback-appender/pom.xml b/logback-appender/pom.xml index e14a2c4..2343763 100644 --- a/logback-appender/pom.xml +++ b/logback-appender/pom.xml @@ -1,4 +1,5 @@ - + 4.0.0 diff --git a/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppenderConfigurator.java b/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppenderConfigurator.java index fd276d3..0f29c55 100644 --- a/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppenderConfigurator.java +++ b/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppenderConfigurator.java @@ -26,6 +26,7 @@ public abstract class LokiAppenderConfigurator extends UnsynchronizedAppenderBas private long batchSize = 10_2400; private long batchWait = 5; private long shutdownTimeoutSeconds = 10; + private long logShipperWakeupIntervalMillis = 10; private int maxRequestsInFlight = 100; @@ -120,6 +121,14 @@ public abstract class LokiAppenderConfigurator extends UnsynchronizedAppenderBas this.shutdownTimeoutSeconds = shutdownTimeoutSeconds; } + public long getLogShipperWakeupIntervalMillis() { + return logShipperWakeupIntervalMillis; + } + + public void setLogShipperWakeupIntervalMillis(long logShipperWakeupIntervalMillis) { + this.logShipperWakeupIntervalMillis = logShipperWakeupIntervalMillis; + } + public int getMaxRequestsInFlight() { return maxRequestsInFlight; } diff --git a/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppenderFactory.java b/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppenderFactory.java index 7fdfd93..ce17acb 100644 --- a/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppenderFactory.java +++ b/logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppenderFactory.java @@ -80,6 +80,7 @@ public class LokiAppenderFactory { configurator.getBatchSize(), TimeUnit.SECONDS.toMillis(configurator.getBatchWait()), bufferSizeBytes, + configurator.getLogShipperWakeupIntervalMillis(), TimeUnit.SECONDS.toMillis(configurator.getShutdownTimeoutSeconds()), configurator.isUseOffHeapBuffer() ); diff --git a/logback-appender/src/test/java/pl/tkowalcz/tjahzi/logback/infra/IntegrationTest.java b/logback-appender/src/test/java/pl/tkowalcz/tjahzi/logback/infra/IntegrationTest.java index a717eb8..26e721f 100644 --- a/logback-appender/src/test/java/pl/tkowalcz/tjahzi/logback/infra/IntegrationTest.java +++ b/logback-appender/src/test/java/pl/tkowalcz/tjahzi/logback/infra/IntegrationTest.java @@ -20,7 +20,7 @@ import java.net.URISyntaxException; public class IntegrationTest { @Container - public GenericContainer loki = new GenericContainer("grafana/loki:latest") + public GenericContainer loki = new GenericContainer("grafana/loki:2.3.0") .withCommand("-config.file=/etc/loki-config.yaml") .withClasspathResourceMapping("loki-config.yaml", "/etc/loki-config.yaml",