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/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";
     }
 }

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

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
         );

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()),

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
         );

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
         );

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
         );

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"
                     );
         });

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 class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
                 batchSize,
                 TimeUnit.SECONDS.toMillis(batchWait),
                 bufferSizeBytes,
+                logShipperWakeupIntervalMillis,
                 TimeUnit.SECONDS.toMillis(shutdownTimeoutSeconds),
                 isUseOffHeapBuffer()
         );
@@ -247,6 +251,14 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
         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 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);
+                        }
+                );
     }
 }

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) {

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 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;
     }

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()
         );

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",