Merge pull request #79 from tkowalcz/77-agrona-sleeping-more

77 agrona sleeping more

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) │    
                           └────────────────────┘         └─────────┘         └─────────┘    
        .                             ▲                                    
                                      │                                    

core/pom.xml

@@ -1,4 +1,5 @@
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 
     <modelVersion>4.0.0</modelVersion>
 

core/src/main/java/pl/tkowalcz/tjahzi/LogBufferAgent.java → core/src/main/java/pl/tkowalcz/tjahzi/LogShipperAgent.java

@@ -6,7 +6,7 @@
 
 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 void onClose() {
 
     @Override
     public String roleName() {
-        return "ReadingLogBufferAndSendingHttp";
+        return "LogShipper";
     }
 }

core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java

@@ -23,8 +23,10 @@ public LoggingSystem createLoggingSystem(
             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 LoggingSystem createLoggingSystem(
         );
 
         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 LoggingSystem createLoggingSystem(
         );
 
         AgentRunner runner = new AgentRunner(
-                new SleepingMillisIdleStrategy(),
+                new SleepingMillisIdleStrategy(logShipperWakeupIntervalMillis),
                 monitoringModule::addAgentError,
                 null,
                 agent

core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java

@@ -89,6 +89,7 @@ void shouldIncludeAdditionalHeaders() {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );
@@ -141,6 +142,7 @@ void shouldHandleCaseWithNoAdditionalHeaders() {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );
@@ -195,6 +197,7 @@ void shouldNotOverrideCrucialHeaders() {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );

core/src/test/java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java

@@ -47,7 +47,7 @@ void shouldSendDataIfOverSizeLimit() throws IOException {
         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 @@ void shouldNotSendDataBelowSizeLimit() throws IOException {
 
         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 @@ void shouldNotAttemptSendingDataIfThereIsNothingToSend() throws IOException {
 
         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 @@ void shouldSendDataBelowSizeLimitIfTimeoutExpires() throws IOException {
         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 @@ void shouldDrainAllMessagesOnClose() throws IOException {
         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 @@ void shouldDrainAllMessagesOnClose() throws IOException {
                 )
         );
 
-        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()),

core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java

@@ -30,7 +30,7 @@
 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 @@ void setUp() {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );

core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java

@@ -33,7 +33,7 @@
 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 @@ void setUp() {
                 1024 * 1024,
                 0,
                 0,
+                250,
                 10_000,
                 false
         );

core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java

@@ -84,6 +84,7 @@ void shouldEventuallyReconnectIfLokiWasDownWhenStarting() {
                 10_000,
                 1000,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );
@@ -134,6 +135,7 @@ void shouldReconnectIfLokiFailed() {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );

core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java

@@ -70,6 +70,7 @@ void shouldStopThreads() {
                 0,
                 0,
                 1024 * 1024,
+                250,
                 10_000,
                 false
         );
@@ -83,7 +84,7 @@ void shouldStopThreads() {
             assertThat(threadInfos)
                     .extracting(ThreadInfo::getThreadName)
                     .contains(
-                            "ReadingLogBufferAndSendingHttp",
+                            "LogShipper",
                             "tjahzi-worker"
                     );
         });
@@ -102,7 +103,7 @@ void shouldStopThreads() {
             assertThat(threadInfos)
                     .extracting(ThreadInfo::getThreadName)
                     .doesNotContain(
-                            "ReadingLogBufferAndSendingHttp",
+                            "LogShipper",
                             "tjahzi-worker"
                     );
         });

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

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

@@ -74,6 +74,9 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
     @PluginBuilderAttribute
     private long batchWait = 5;
 
+    @PluginBuilderAttribute
+    private long logShipperWakeupIntervalMillis = 10;
+
     @PluginBuilderAttribute
     private int shutdownTimeoutSeconds = 10;
 
@@ -141,6 +144,7 @@ public LokiAppender build() {
                 batchSize,
                 TimeUnit.SECONDS.toMillis(batchWait),
                 bufferSizeBytes,
+                logShipperWakeupIntervalMillis,
                 TimeUnit.SECONDS.toMillis(shutdownTimeoutSeconds),
                 isUseOffHeapBuffer()
         );
@@ -247,6 +251,14 @@ public void setBatchWait(long batchWait) {
         this.batchWait = batchWait;
     }
 
+    public long getLogShipperWakeupIntervalMillis() {
+        return logShipperWakeupIntervalMillis;
+    }
+
+    public void setLogShipperWakeupIntervalMillis(long logShipperWakeupIntervalMillis) {
+        this.logShipperWakeupIntervalMillis = logShipperWakeupIntervalMillis;
+    }
+
     public int getShutdownTimeoutSeconds() {
         return shutdownTimeoutSeconds;
     }

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

@@ -5,6 +5,7 @@
 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 @@ void shouldInjectMonitoringAndUseIt() {
                 .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);
+                        }
+                );
     }
 }

log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/infra/IntegrationTest.java

@@ -18,7 +18,7 @@
 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 static LoggerContext loadConfig(String fileName) {
 
             return loadConfig(uri);
         } catch (URISyntaxException e) {
-            Assertions.fail(e);
+            return Assertions.fail(e);
         }
-
-        return null;
     }
 
     public static LoggerContext loadConfig(URI uri) {

log4j2-appender/src/test/resources/basic-appender-test-configuration.xml

@@ -16,7 +16,7 @@
 
             <ThresholdFilter level="ALL"/>
             <PatternLayout>
-                <Pattern>%X{ctx:tid} [%t] %d{DEFAULT} %5p %c{1} - %m%n%exception{full}</Pattern>
+                <Pattern>[%t] %d{DEFAULT} %5p %c{1} - %m%n%exception{full}</Pattern>
             </PatternLayout>
 
             <Header name="server" value="127.0.0.1"/>

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. |

logback-appender/pom.xml

@@ -1,4 +1,5 @@
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 
     <modelVersion>4.0.0</modelVersion>
 

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 void setShutdownTimeoutSeconds(long shutdownTimeoutSeconds) {
         this.shutdownTimeoutSeconds = shutdownTimeoutSeconds;
     }
 
+    public long getLogShipperWakeupIntervalMillis() {
+        return logShipperWakeupIntervalMillis;
+    }
+
+    public void setLogShipperWakeupIntervalMillis(long logShipperWakeupIntervalMillis) {
+        this.logShipperWakeupIntervalMillis = logShipperWakeupIntervalMillis;
+    }
+
     public int getMaxRequestsInFlight() {
         return maxRequestsInFlight;
     }

logback-appender/src/main/java/pl/tkowalcz/tjahzi/logback/LokiAppenderFactory.java

@@ -80,6 +80,7 @@ public LoggingSystem createAppender() {
                 configurator.getBatchSize(),
                 TimeUnit.SECONDS.toMillis(configurator.getBatchWait()),
                 bufferSizeBytes,
+                configurator.getLogShipperWakeupIntervalMillis(),
                 TimeUnit.SECONDS.toMillis(configurator.getShutdownTimeoutSeconds()),
                 configurator.isUseOffHeapBuffer()
         );

logback-appender/src/test/java/pl/tkowalcz/tjahzi/logback/infra/IntegrationTest.java

@@ -20,7 +20,7 @@
 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",