From 7883f4851f081cc438bcaeb869f3d3fec63a7095 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Thu, 21 Oct 2021 18:50:13 +0200 Subject: [PATCH] Re #77: Renamed LogBufferAgent and its thread (ReadingLogBufferAndSendingHttp) to LogShipper. Added option to configure time it sleeps between reading from log buffer to not hog CPU. Also now default is 25ms instead of hardcoded 1ms previously. --- .../{LogBufferAgent.java => LogShipperAgent.java} | 6 +++--- .../java/pl/tkowalcz/tjahzi/TjahziInitializer.java | 8 +++++--- .../test/java/pl/tkowalcz/tjahzi/HeadersTest.java | 3 +++ .../java/pl/tkowalcz/tjahzi/LogBufferAgentTest.java | 12 ++++++------ .../tjahzi/LoggingSystemSanityCheckTest.java | 1 + .../java/pl/tkowalcz/tjahzi/LoggingSystemTest.java | 1 + .../test/java/pl/tkowalcz/tjahzi/ReconnectTest.java | 2 ++ .../tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java | 5 +++-- 8 files changed, 24 insertions(+), 14 deletions(-) rename core/src/main/java/pl/tkowalcz/tjahzi/{LogBufferAgent.java => LogShipperAgent.java} (93%) 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..11a2999 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 logShipperIntervalMillis, 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(logShipperIntervalMillis), 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..9ba2ae0 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemSanityCheckTest.java @@ -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..cc681ec 100644 --- a/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java +++ b/core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java @@ -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" ); });