mirror of
https://github.com/tiennm99/tjahzi.git
synced 2026-10-16 02:15:51 +00:00
104 daemon threads (#105)
* Revert "Update LICENSE" * Added new option that instructs Tjahzi to use daemon threads (and thus potentially loose data when idle). * Re #104: Added documentation of useDaemonThreads option to log4j2 and logback. Always using daemon threads for netty event loop group. * Re #104: Fixed test. * Re #104: Fixed test. * Re #104: Added test verifying we start threads as daemons. * Re #104: Corrected a typo in the docs. * Re #104: Rewrite of the readme section about useDaemonThreads option.
This commit is contained in:
Tomasz Kowalczewski authored and GitHub committed
1 parent
082fe7e3a7
commit
32ee5d04a9
15 files changed
+150
-34
No files matched your search
@@ -83,5 +83,5 @@ Check out [this wiki article](https://github.com/tkowalcz/tjahzi/wiki/Monitoring
|
||||
|
||||
# LICENSE
|
||||
|
||||
This work is released under MIT license with the provision that any company registered or conducting operations mainly in Russian Federation IS PROHIBITED from using this software.
|
||||
This work is released under MIT license. Feel free to use, copy and modify this work as long as you credit original authors.
|
||||
Pull and feature requests are welcome.
|
||||
@@ -13,18 +13,21 @@ public class LoggingSystem {
|
||||
private final AgentRunner runner;
|
||||
|
||||
private final MonitoringModule monitoringModule;
|
||||
private final boolean useDaemonThreads;
|
||||
private final Closeable[] resourcesToCleanup;
|
||||
|
||||
public LoggingSystem(
|
||||
ManyToOneRingBuffer logBuffer,
|
||||
AgentRunner runner,
|
||||
MonitoringModule monitoringModule,
|
||||
boolean useDaemonThreads,
|
||||
Closeable... resourcesToCleanup
|
||||
) {
|
||||
this.logBuffer = logBuffer;
|
||||
this.runner = runner;
|
||||
|
||||
this.monitoringModule = monitoringModule;
|
||||
this.useDaemonThreads = useDaemonThreads;
|
||||
this.resourcesToCleanup = resourcesToCleanup;
|
||||
}
|
||||
|
||||
@@ -33,7 +36,15 @@ public class LoggingSystem {
|
||||
}
|
||||
|
||||
public void start() {
|
||||
AgentRunner.startOnThread(runner);
|
||||
AgentRunner.startOnThread(
|
||||
runner,
|
||||
runnable -> {
|
||||
Thread result = new Thread(runnable);
|
||||
result.setDaemon(useDaemonThreads);
|
||||
|
||||
return result;
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
public void close(
|
||||
|
||||
@@ -25,7 +25,8 @@ public class TjahziInitializer {
|
||||
int bufferSizeBytes,
|
||||
long logShipperWakeupIntervalMillis,
|
||||
long shutdownTimeoutMillis,
|
||||
boolean offHeap
|
||||
boolean offHeap,
|
||||
boolean useDaemonThreads
|
||||
) {
|
||||
bufferSizeBytes = findNearestPowerOfTwo(bufferSizeBytes);
|
||||
ByteBuffer javaBuffer = allocateJavaBuffer(bufferSizeBytes, offHeap);
|
||||
@@ -64,6 +65,7 @@ public class TjahziInitializer {
|
||||
logBuffer,
|
||||
runner,
|
||||
monitoringModule,
|
||||
useDaemonThreads,
|
||||
httpClient
|
||||
);
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package pl.tkowalcz.tjahzi.http;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.handler.codec.http.FullHttpRequest;
|
||||
import io.netty.util.concurrent.DefaultThreadFactory;
|
||||
import pl.tkowalcz.tjahzi.stats.MonitoringModule;
|
||||
|
||||
import java.io.Closeable;
|
||||
@@ -20,9 +21,7 @@ public class HttpConnection implements Closeable {
|
||||
this.clientConfiguration = clientConfiguration;
|
||||
this.monitoringModule = monitoringModule;
|
||||
|
||||
ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client");
|
||||
ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker");
|
||||
|
||||
ThreadFactory threadFactory = new DefaultThreadFactory("tjahzi-worker", true);
|
||||
this.group = new NioEventLoopGroup(1, threadFactory);
|
||||
|
||||
EventLoopGroupRetry retry = new EventLoopGroupRetry(
|
||||
|
||||
@@ -91,6 +91,7 @@ class HeadersTest {
|
||||
1024 * 1024,
|
||||
250,
|
||||
10_000,
|
||||
false,
|
||||
false
|
||||
);
|
||||
loggingSystem.start();
|
||||
@@ -145,6 +146,7 @@ class HeadersTest {
|
||||
1024 * 1024,
|
||||
250,
|
||||
10_000,
|
||||
false,
|
||||
false
|
||||
);
|
||||
loggingSystem.start();
|
||||
@@ -201,6 +203,7 @@ class HeadersTest {
|
||||
1024 * 1024,
|
||||
250,
|
||||
10_000,
|
||||
false,
|
||||
false
|
||||
);
|
||||
loggingSystem.start();
|
||||
|
||||
@@ -69,6 +69,7 @@ class LoggingSystemSanityCheckTest {
|
||||
1024 * 1024,
|
||||
250,
|
||||
10_000,
|
||||
false,
|
||||
false
|
||||
);
|
||||
|
||||
|
||||
@@ -73,6 +73,7 @@ class LoggingSystemTest {
|
||||
0,
|
||||
250,
|
||||
10_000,
|
||||
false,
|
||||
false
|
||||
);
|
||||
|
||||
|
||||
@@ -86,6 +86,7 @@ public class ReconnectTest {
|
||||
1024 * 1024,
|
||||
250,
|
||||
10_000,
|
||||
false,
|
||||
false
|
||||
);
|
||||
loggingSystem.start();
|
||||
@@ -138,6 +139,7 @@ public class ReconnectTest {
|
||||
1024 * 1024,
|
||||
250,
|
||||
10_000,
|
||||
false,
|
||||
false
|
||||
);
|
||||
loggingSystem.start();
|
||||
|
||||
@@ -14,7 +14,9 @@ import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule;
|
||||
import java.lang.management.ManagementFactory;
|
||||
import java.lang.management.ThreadInfo;
|
||||
import java.lang.management.ThreadMXBean;
|
||||
import java.util.Arrays;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
|
||||
@@ -26,6 +28,7 @@ class ResourcesCleanupOnCloseTest {
|
||||
private WireMockServer wireMockServer;
|
||||
private TjahziInitializer initializer;
|
||||
private StandardMonitoringModule monitoringModule;
|
||||
private LoggingSystem loggingSystem;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
@@ -43,6 +46,13 @@ class ResourcesCleanupOnCloseTest {
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
wireMockServer.stop();
|
||||
|
||||
if (loggingSystem != null) {
|
||||
loggingSystem.close(
|
||||
(int) TimeUnit.SECONDS.toMillis(10),
|
||||
System.out::println
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -72,6 +82,7 @@ class ResourcesCleanupOnCloseTest {
|
||||
1024 * 1024,
|
||||
250,
|
||||
10_000,
|
||||
false,
|
||||
false
|
||||
);
|
||||
loggingSystem.start();
|
||||
@@ -83,10 +94,8 @@ class ResourcesCleanupOnCloseTest {
|
||||
|
||||
assertThat(threadInfos)
|
||||
.extracting(ThreadInfo::getThreadName)
|
||||
.contains(
|
||||
"LogShipper",
|
||||
"tjahzi-worker"
|
||||
);
|
||||
.anyMatch(threadName -> threadName.contains("LogShipper"))
|
||||
.anyMatch(threadName -> threadName.contains("tjahzi-worker"));
|
||||
});
|
||||
|
||||
//When
|
||||
@@ -108,4 +117,58 @@ class ResourcesCleanupOnCloseTest {
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldStartThreadsAsDaemon() {
|
||||
// Given
|
||||
ClientConfiguration clientConfiguration = ClientConfiguration.builder()
|
||||
.withConnectionTimeoutMillis(10_000)
|
||||
.withHost("localhost")
|
||||
.withPort(wireMockServer.port())
|
||||
.withMaxRetries(1)
|
||||
.build();
|
||||
|
||||
NettyHttpClient httpClient = HttpClientFactory.defaultFactory()
|
||||
.getHttpClient(
|
||||
clientConfiguration,
|
||||
monitoringModule,
|
||||
"X-Scope-OrgID", "Circus",
|
||||
"C", "Control"
|
||||
);
|
||||
|
||||
// When
|
||||
loggingSystem = initializer.createLoggingSystem(
|
||||
httpClient,
|
||||
monitoringModule,
|
||||
Map.of(),
|
||||
0,
|
||||
0,
|
||||
1024 * 1024,
|
||||
250,
|
||||
10_000,
|
||||
false,
|
||||
true
|
||||
);
|
||||
loggingSystem.start();
|
||||
|
||||
// Then
|
||||
Awaitility.await().untilAsserted(() -> {
|
||||
ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
|
||||
ThreadInfo[] threadInfos = threadMXBean.dumpAllThreads(false, false);
|
||||
|
||||
Optional<ThreadInfo> maybeLogShipper = Arrays.stream(threadInfos)
|
||||
.filter(thread -> thread.getThreadName().contains("LogShipper"))
|
||||
.findAny();
|
||||
|
||||
Optional<ThreadInfo> maybeTjahziWorker = Arrays.stream(threadInfos)
|
||||
.filter(thread -> thread.getThreadName().contains("tjahzi-worker"))
|
||||
.findAny();
|
||||
|
||||
assertThat(maybeLogShipper)
|
||||
.hasValueSatisfying(ThreadInfo::isDaemon);
|
||||
|
||||
assertThat(maybeTjahziWorker)
|
||||
.hasValueSatisfying(ThreadInfo::isDaemon);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -356,3 +356,11 @@ controls how often it wakes up to perform its duties. Other properties control h
|
||||
|
||||
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.
|
||||
|
||||
#### useDaemonThreads (optional, default = false)
|
||||
|
||||
If set to true Tjahzi will run all it's threads as daemon threads.
|
||||
|
||||
Use this option if you do not want to explicitly close the logging system and still want to make sure Tjahzi internal
|
||||
threads will not prevent JVM from closing down. Note that this can result in unflushed logs not being delivered when the
|
||||
JVM is closed.
|
||||
@@ -50,6 +50,9 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
|
||||
@PluginBuilderAttribute
|
||||
private boolean useSSL;
|
||||
|
||||
@PluginBuilderAttribute
|
||||
private boolean useDaemonThreads;
|
||||
|
||||
@PluginBuilderAttribute
|
||||
private String username;
|
||||
|
||||
@@ -155,7 +158,8 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
|
||||
bufferSizeBytes,
|
||||
logShipperWakeupIntervalMillis,
|
||||
TimeUnit.SECONDS.toMillis(shutdownTimeoutSeconds),
|
||||
isUseOffHeapBuffer()
|
||||
isUseOffHeapBuffer(),
|
||||
useDaemonThreads
|
||||
);
|
||||
|
||||
int maxLogLineSizeBytes = Math.toIntExact(getMaxLogLineSizeKilobytes() * BYTES_IN_KILOBYTE);
|
||||
@@ -201,6 +205,14 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
|
||||
this.useSSL = useSSL;
|
||||
}
|
||||
|
||||
public boolean isUseDaemonThreads() {
|
||||
return useDaemonThreads;
|
||||
}
|
||||
|
||||
public void setUseDaemonThreads(boolean useDaemonThreads) {
|
||||
this.useDaemonThreads = useDaemonThreads;
|
||||
}
|
||||
|
||||
public String getUsername() {
|
||||
return username;
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package pl.tkowalcz.tjahzi.log4j2;
|
||||
import org.apache.commons.lang3.RandomStringUtils;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
import org.junit.jupiter.api.Disabled;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.hamcrest.CoreMatchers.*;
|
||||
@@ -12,6 +13,7 @@ import static pl.tkowalcz.tjahzi.log4j2.infra.LokiAssert.assertThat;
|
||||
class GrafanaCloudAppenderTest {
|
||||
|
||||
@Test
|
||||
@Disabled
|
||||
void shouldSendData() {
|
||||
// Given
|
||||
System.setProperty("loki.host", "logs-prod-us-central1.grafana.net");
|
||||
|
||||
+23
-22
@@ -152,12 +152,12 @@ Note that all previously mentioned tags (host, port, useSSL, logEndpoint) cannot
|
||||
URL consists of four parts: protocol, host, port, path. Some of them may be omitted and there are defaults that depend
|
||||
on contents of other parts of the URL. This table has a rundown of all viable configurations:
|
||||
|
||||
| Section | Default | Comment |
|
||||
|----------|----------------------------|-------------------------------------------------------------------------------------|
|
||||
| Protocol | None (must be provided) | Supported protocols are `http` and `https`. Https is equivlent to setting `useUSSL` |
|
||||
| Host | None (must be provided) | |
|
||||
| Port | 80 for http, 443 for https | You can use any port and SSL will still be used if protocol is set to https |
|
||||
| Path | '/loki/api/v1/push' | |
|
||||
| Section | Default | Comment |
|
||||
|----------|----------------------------|--------------------------------------------------------------------------------------|
|
||||
| Protocol | None (must be provided) | Supported protocols are `http` and `https`. Https is equivalent to setting `useUSSL` |
|
||||
| Host | None (must be provided) | |
|
||||
| Port | 80 for http, 443 for https | You can use any port and SSL will still be used if protocol is set to https |
|
||||
| Path | '/loki/api/v1/push' | |
|
||||
|
||||
Some examples of correct URLs:
|
||||
|
||||
@@ -249,20 +249,21 @@ or use Tjahzi optimized efficient, low allocation encoder:
|
||||
|
||||
#### Optional configuration parameters
|
||||
|
||||
| Tag | Default value | Description |
|
||||
|--------------------------------|-----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
|
||||
| Url | - | Configure connection in one place instead of using host, port etc. See [this section](#configuring-connection-parameters-individually-and-using-URL). |
|
||||
| UseSSL | true if port == '443' | Enable secure (HTTPS) communication regardless of configured port number. |
|
||||
| LogEndpoint | `/loki/api/v1/push` | Overrides the default endpoint to which Tjahzi sends data. This can be useful if Loki is behind reverse proxy and additional path mapping is used. |
|
||||
| Header | - | This tag can be used multiple times to specify additional headers that are passed to Loki instance. One example is to pass a `X-Scope-OrgID` header when [running Loki in multi-tenant mode](https://grafana.com/docs/loki/latest/operations/authentication/). |
|
||||
| Label | - | Specify additional labels attached to each log line sent via this appender instance. See also note about [label naming](https://github.com/tkowalcz/tjahzi/wiki/Label-naming). |
|
||||
| LogLevelLabel | - | If defined then log level label of configured name will be added to each line sent to Loki. It will contain Logback log level e.g. `INFO`, `WARN` etc. See also note about [label naming](https://github.com/tkowalcz/tjahzi/wiki/Label-naming). |
|
||||
| BufferSizeMegabytes | 32 MB | Size of the `log buffer`. Must be power of two between 1MB and 1GB. See [log buffer sizing](https://github.com/tkowalcz/tjahzi/wiki/Log-buffer-sizing) for more explanations. |
|
||||
| MaxRetries | 3 | Maximum number of retries to perform when delivering log message to Loki. Log buffer data is delivered in order, one batch after the other, so too much retries will block delivery of subsequent log batches (on the other hand if we need to retry many times then next batches will probably fail too). |
|
||||
| ConnectTimeoutMillis | 5s | This configures socket connect timeout when connecting to Loki. After unsuccessful connection attempt it will continue to retry indefinitely employing exponential backoff (initial backoff = 250ms, maximum backoff = 30s, multiplier = 3). |
|
||||
| ReadTimeoutMillis | 60s | Sets socket read timeout on Loki connection. |
|
||||
| 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`. |
|
||||
| Tag | Default value | Description |
|
||||
|--------------------------------|-----------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
|
||||
| Url | - | Configure connection in one place instead of using host, port etc. See [this section](#configuring-connection-parameters-individually-and-using-URL). |
|
||||
| UseSSL | true if port == '443' | Enable secure (HTTPS) communication regardless of configured port number. |
|
||||
| LogEndpoint | `/loki/api/v1/push` | Overrides the default endpoint to which Tjahzi sends data. This can be useful if Loki is behind reverse proxy and additional path mapping is used. |
|
||||
| Header | - | This tag can be used multiple times to specify additional headers that are passed to Loki instance. One example is to pass a `X-Scope-OrgID` header when [running Loki in multi-tenant mode](https://grafana.com/docs/loki/latest/operations/authentication/). |
|
||||
| Label | - | Specify additional labels attached to each log line sent via this appender instance. See also note about [label naming](https://github.com/tkowalcz/tjahzi/wiki/Label-naming). |
|
||||
| LogLevelLabel | - | If defined then log level label of configured name will be added to each line sent to Loki. It will contain Logback log level e.g. `INFO`, `WARN` etc. See also note about [label naming](https://github.com/tkowalcz/tjahzi/wiki/Label-naming). |
|
||||
| BufferSizeMegabytes | 32 MB | Size of the `log buffer`. Must be power of two between 1MB and 1GB. See [log buffer sizing](https://github.com/tkowalcz/tjahzi/wiki/Log-buffer-sizing) for more explanations. |
|
||||
| MaxRetries | 3 | Maximum number of retries to perform when delivering log message to Loki. Log buffer data is delivered in order, one batch after the other, so too much retries will block delivery of subsequent log batches (on the other hand if we need to retry many times then next batches will probably fail too). |
|
||||
| ConnectTimeoutMillis | 5s | This configures socket connect timeout when connecting to Loki. After unsuccessful connection attempt it will continue to retry indefinitely employing exponential backoff (initial backoff = 250ms, maximum backoff = 30s, multiplier = 3). |
|
||||
| ReadTimeoutMillis | 60s | Sets socket read timeout on Loki connection. |
|
||||
| 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. |
|
||||
| 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. |
|
||||
| useDaemonThreads | false | If set to true Tjahzi will run all it's threads as daemon threads. Use this option if you do not want to explicitly close the logging system and still want to make sure Tjahzi internal threads will not prevent JVM from closing down. Note that this can result in unflushed logs not being delivered when the JVM is closed. |
|
||||
+10
@@ -18,6 +18,8 @@ public abstract class LokiAppenderConfigurator extends UnsynchronizedAppenderBas
|
||||
|
||||
private boolean useSSL;
|
||||
|
||||
private boolean useDaemonThreads;
|
||||
|
||||
private String username;
|
||||
|
||||
private String password;
|
||||
@@ -82,6 +84,14 @@ public abstract class LokiAppenderConfigurator extends UnsynchronizedAppenderBas
|
||||
this.useSSL = useSSL;
|
||||
}
|
||||
|
||||
public boolean isUseDaemonThreads() {
|
||||
return useDaemonThreads;
|
||||
}
|
||||
|
||||
public void setUseDaemonThreads(boolean useDaemonThreads) {
|
||||
this.useDaemonThreads = useDaemonThreads;
|
||||
}
|
||||
|
||||
public String getUsername() {
|
||||
return username;
|
||||
}
|
||||
|
||||
@@ -87,7 +87,8 @@ public class LokiAppenderFactory {
|
||||
bufferSizeBytes,
|
||||
configurator.getLogShipperWakeupIntervalMillis(),
|
||||
TimeUnit.SECONDS.toMillis(configurator.getShutdownTimeoutSeconds()),
|
||||
configurator.isUseOffHeapBuffer()
|
||||
configurator.isUseOffHeapBuffer(),
|
||||
configurator.isUseDaemonThreads()
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user