mirror of
https://github.com/tiennm99/tjahzi.git
synced 2026-10-14 22:17:47 +00:00
Merge pull request #61 from tkowalcz/60-configure-buffer-destination-size
60 configure buffer destination size
This commit is contained in:
10 files changed
+369
-80
No files matched your search
+69
-55
@@ -1,7 +1,7 @@
|
||||
# Log4j2 Appender
|
||||
|
||||
On top of a `Core` component with rather simplistic API we intend to build several layers that make it truly useful. Log4j2
|
||||
appender seemed like a good first.
|
||||
On top of a `Core` component with rather simplistic API we intend to build several layers that make it truly useful.
|
||||
Log4j2 appender seemed like a good first.
|
||||
|
||||
## Quick start guide
|
||||
|
||||
@@ -30,56 +30,57 @@ appender seemed like a good first.
|
||||
1. Include minimal appender configuration:
|
||||
|
||||
```xml
|
||||
<Loki name="loki-appender">
|
||||
<host>${sys:loki.host}</host>
|
||||
<port>${sys:loki.port}</port>
|
||||
<Loki name="loki-appender">
|
||||
<host>${sys:loki.host}</host>
|
||||
<port>${sys:loki.port}</port>
|
||||
|
||||
<PatternLayout>
|
||||
<Pattern>%X{tid} [%t] %d{MM-dd HH:mm:ss.SSS} %5p %c{1} - %m%n%exception{full}</Pattern>
|
||||
</PatternLayout>
|
||||
<PatternLayout>
|
||||
<Pattern>%X{tid} [%t] %d{MM-dd HH:mm:ss.SSS} %5p %c{1} - %m%n%exception{full}</Pattern>
|
||||
</PatternLayout>
|
||||
|
||||
<Label name="server" value="${sys:hostname}"/>
|
||||
</Loki>
|
||||
<Label name="server" value="${sys:hostname}"/>
|
||||
</Loki>
|
||||
```
|
||||
|
||||
1. Reference the appender from inside one of your logger definitions:
|
||||
|
||||
```xml
|
||||
<Root level="INFO">
|
||||
<AppenderRef ref="Loki"/>
|
||||
</Root>
|
||||
<Root level="INFO">
|
||||
<AppenderRef ref="Loki"/>
|
||||
</Root>
|
||||
```
|
||||
|
||||
### Note on Loki HTTP endpoint and host/port configuration
|
||||
|
||||
Tjahzi sends `POST` requests to `/loki/api/v1/push` HTTP endpoint. Specifying e.g. `<host>loki.mydomain.com</host><port>3100</port>`
|
||||
Tjahzi sends `POST` requests to `/loki/api/v1/push` HTTP endpoint. Specifying
|
||||
e.g. `<host>loki.mydomain.com</host><port>3100</port>`
|
||||
will configure the appender to call to URL: `http://loki.mydomain.com:3100/loki/api/v1/push`.
|
||||
|
||||
### Grafana Cloud configuration
|
||||
|
||||
Tjahzi can send logs to Grafana Cloud. It needs two things to be configured:
|
||||
- Set port number to `443` which switches HTTP client into HTTPS mode.
|
||||
- Specify `username` and `password` for HTTP basic authentication that Grafana uses.
|
||||
|
||||
- Specify `username` and `password` for HTTP basic authentication that Grafana uses.
|
||||
|
||||
Password is your "Grafana.com API Key" and can be generated in "Grafana datasource settings". The host in below example
|
||||
is just for illustrative purposes.
|
||||
|
||||
```xml
|
||||
<Loki name="loki-appender">
|
||||
<!-- example host -->
|
||||
<host>logs-prod-us-central1.grafana.net</host>
|
||||
<port>443</port>
|
||||
<Loki name="loki-appender">
|
||||
<!-- example host -->
|
||||
<host>logs-prod-us-central1.grafana.net</host>
|
||||
<port>443</port>
|
||||
|
||||
<username>...</username>
|
||||
<password>...</password>
|
||||
...
|
||||
</Loki>
|
||||
<username>...</username>
|
||||
<password>...</password>
|
||||
...
|
||||
</Loki>
|
||||
```
|
||||
|
||||
## Advanced configuration
|
||||
|
||||
This example sets up a root logger with a Loki appender. Note that `pl.tkowalcz.tjahzi.log4j2` is added to `packages` attribute
|
||||
of configuration so that the appender can be found.
|
||||
This example sets up a root logger with a Loki appender. Note that `pl.tkowalcz.tjahzi.log4j2` is added to `packages`
|
||||
attribute of configuration so that the appender can be found.
|
||||
|
||||
```xml
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
@@ -102,7 +103,7 @@ of configuration so that the appender can be found.
|
||||
|
||||
<Header name="X-Scope-OrgID" value="Circus"/>
|
||||
<Label name="server" value="127.0.0.1"/>
|
||||
|
||||
|
||||
<LogLevelLabel>log_level</LogLevelLabel>
|
||||
</Loki>
|
||||
</appenders>
|
||||
@@ -111,12 +112,13 @@ of configuration so that the appender can be found.
|
||||
|
||||
### Lookups / variable substitution
|
||||
|
||||
Contents of the properties are [automatically interpolated by Log4j2](https://logging.apache.org/log4j/log4j-2.2/manual/configuration.html#PropertySubstitution).
|
||||
All environment, system etc. variable references will be replaced by their values during initialization of the appender.
|
||||
The exception is context/MDC (`${ctx:foo}`) value lookup as it make sense only during runtime.
|
||||
Contents of the properties
|
||||
are [automatically interpolated by Log4j2](https://logging.apache.org/log4j/log4j-2.2/manual/configuration.html#PropertySubstitution)
|
||||
. All environment, system etc. variable references will be replaced by their values during initialization of the
|
||||
appender. The exception is context/MDC (`${ctx:foo}`) value lookup as it make sense only during runtime.
|
||||
|
||||
NOTE: This process could have been executed for every lookup type at runtime (for each log message). This approach was
|
||||
deemed too expensive. If you need a mechanism to replace a variable (other than context/MDC) after logging system
|
||||
NOTE: This process could have been executed for every lookup type at runtime (for each log message). This approach was
|
||||
deemed too expensive. If you need a mechanism to replace a variable (other than context/MDC) after logging system
|
||||
initialization I would love to hear your use case - please file an issue.
|
||||
|
||||
## Details
|
||||
@@ -125,8 +127,8 @@ Let's go through the example config above and analyze configuration options (**N
|
||||
|
||||
#### Host (required)
|
||||
|
||||
Network host address of Loki instance. Either IP address or host name. It will be passed to Netty and end up being resolved
|
||||
by call to `InetSocketAddress.createUnresolved`.
|
||||
Network host address of Loki instance. Either IP address or host name. It will be passed to Netty and end up being
|
||||
resolved by call to `InetSocketAddress.createUnresolved`.
|
||||
|
||||
#### Port (required)
|
||||
|
||||
@@ -143,48 +145,60 @@ Password for HTTP basic auth.
|
||||
|
||||
#### Header (optional)
|
||||
|
||||
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/).
|
||||
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 (optional)
|
||||
|
||||
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).
|
||||
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 (optional)
|
||||
|
||||
If defined then log level label of configured name will be added to each line sent to Loki. It will contain Log4j log level e.g. `INFO`, `WARN` etc. See also note about [label naming](https://github.com/tkowalcz/tjahzi/wiki/Label-naming).
|
||||
If defined then log level label of configured name will be added to each line sent to Loki. It will contain Log4j log
|
||||
level e.g. `INFO`, `WARN` etc. See also note about [label naming](https://github.com/tkowalcz/tjahzi/wiki/Label-naming).
|
||||
|
||||
#### bufferSizeMegabytes (optional, default = 32 MB)
|
||||
#### bufferSizeMegabytes (optional, default = 32)
|
||||
|
||||
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.
|
||||
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.
|
||||
|
||||
#### maxLogLineSizeKilobytes (optional, default = 10)
|
||||
|
||||
Size of an intermediate thread local buffer that is used by log4j to serialise single log message into. Log lines larger
|
||||
than that will be split into multiple log entries (
|
||||
see [wiki](https://github.com/tkowalcz/tjahzi/wiki/Log-buffer-sizing#note-on-thread-local-buffer) for discussion).
|
||||
|
||||
#### maxRetries (optional, default = 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).
|
||||
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 (optional, default = 5s)
|
||||
#### connectTimeoutMillis (optional, default = 5000)
|
||||
|
||||
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).
|
||||
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 (optional, default = 60s)
|
||||
#### readTimeoutMillis (optional, default = 60 000)
|
||||
|
||||
Sets socket read timeout on Loki connection.
|
||||
|
||||
#### useOffHeapBuffer (optional, default = 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.
|
||||
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 (optional, default = 10_2400)
|
||||
|
||||
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`.
|
||||
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 = (optional, default = 5s)
|
||||
#### batchWait (optional, default = 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`.
|
||||
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`
|
||||
.
|
||||
@@ -12,7 +12,6 @@ import pl.tkowalcz.tjahzi.log4j2.labels.LabelPrinter;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.BiConsumer;
|
||||
@@ -21,33 +20,36 @@ public class AppenderLogic implements BiConsumer<LogEvent, ByteBuffer> {
|
||||
|
||||
private static final Logger LOGGER = StatusLogger.getLogger();
|
||||
|
||||
private static final Map<String, CharSequence> EMPTY_MAP = new HashMap<>();
|
||||
|
||||
private final LoggingSystem loggingSystem;
|
||||
|
||||
private final String logLevelLabel;
|
||||
private final TjahziLogger logger;
|
||||
|
||||
private final Map<String, LabelPrinter> lokiLabels;
|
||||
private final ByteBufferDestinationRepository destinationRepository;
|
||||
|
||||
public AppenderLogic(
|
||||
LoggingSystem loggingSystem,
|
||||
String logLevelLabel,
|
||||
Map<String, LabelPrinter> lokiLabels
|
||||
Map<String, LabelPrinter> lokiLabels,
|
||||
int maxLogLineSizeBytes
|
||||
) {
|
||||
this.loggingSystem = loggingSystem;
|
||||
this.logLevelLabel = logLevelLabel;
|
||||
|
||||
logger = loggingSystem.createLogger();
|
||||
this.logger = loggingSystem.createLogger();
|
||||
this.lokiLabels = lokiLabels;
|
||||
this.destinationRepository = new ByteBufferDestinationRepository(maxLogLineSizeBytes);
|
||||
}
|
||||
|
||||
public void append(
|
||||
Layout<? extends Serializable> layout,
|
||||
LogEvent event
|
||||
) {
|
||||
ByteBufferDestinations destination = ByteBufferDestinations.threadLocal();
|
||||
destination.initialize(this, event);
|
||||
TjahziByteBufferDestination destination = destinationRepository.threadLocalDestination(
|
||||
this,
|
||||
event
|
||||
);
|
||||
|
||||
layout.encode(event, destination);
|
||||
destination.drainRemaining();
|
||||
|
||||
+33
@@ -0,0 +1,33 @@
|
||||
package pl.tkowalcz.tjahzi.log4j2;
|
||||
|
||||
import org.apache.logging.log4j.core.LogEvent;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.function.BiConsumer;
|
||||
|
||||
public class ByteBufferDestinationRepository {
|
||||
|
||||
public static final int DEFAULT_MAX_LINE_SIZE_BYTES = 10 * 1024;
|
||||
|
||||
private final ThreadLocal<TjahziByteBufferDestination> threadLocal = new ThreadLocal<>();
|
||||
private final int maxLogLineSizeBytes;
|
||||
|
||||
public ByteBufferDestinationRepository(int maxLogLineSizeBytes) {
|
||||
this.maxLogLineSizeBytes = maxLogLineSizeBytes;
|
||||
}
|
||||
|
||||
public TjahziByteBufferDestination threadLocalDestination(
|
||||
BiConsumer<LogEvent, ByteBuffer> drain,
|
||||
LogEvent context
|
||||
) {
|
||||
TjahziByteBufferDestination result = threadLocal.get();
|
||||
|
||||
if (result == null) {
|
||||
result = new TjahziByteBufferDestination(maxLogLineSizeBytes);
|
||||
threadLocal.set(result);
|
||||
}
|
||||
|
||||
result.initialize(drain, context);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@@ -46,6 +46,7 @@ public class LokiAppender extends AbstractAppender {
|
||||
Property[] properties,
|
||||
String logLevelLabel,
|
||||
Map<String, LabelPrinter> lokiLabels,
|
||||
int maxLogLineSizeKilobytes,
|
||||
LoggingSystem loggingSystem,
|
||||
MutableMonitoringModuleWrapper monitoringModuleWrapper
|
||||
) {
|
||||
@@ -63,7 +64,8 @@ public class LokiAppender extends AbstractAppender {
|
||||
this.appenderLogic = new AppenderLogic(
|
||||
loggingSystem,
|
||||
logLevelLabel,
|
||||
lokiLabels
|
||||
lokiLabels,
|
||||
maxLogLineSizeKilobytes
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -34,6 +34,7 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
|
||||
private static final Logger LOGGER = StatusLogger.getLogger();
|
||||
|
||||
private static final int BYTES_IN_MEGABYTE = 1024 * 1024;
|
||||
private static final int BYTES_IN_KILOBYTE = 1024;
|
||||
|
||||
@PluginBuilderAttribute
|
||||
@Required(message = "No Loki address provided for LokiAppender")
|
||||
@@ -68,10 +69,13 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
|
||||
private String logLevelLabel;
|
||||
|
||||
@PluginBuilderAttribute
|
||||
private final long batchSize = 10_2400;
|
||||
private long batchSize = 10_2400;
|
||||
|
||||
@PluginBuilderAttribute
|
||||
private final long batchWait = 5;
|
||||
private long batchWait = 5;
|
||||
|
||||
@PluginBuilderAttribute
|
||||
private int maxLogLineSizeKilobytes = 10;
|
||||
|
||||
@PluginBuilderAttribute
|
||||
private int maxRequestsInFlight = 100;
|
||||
@@ -132,6 +136,7 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
|
||||
isUseOffHeapBuffer()
|
||||
);
|
||||
|
||||
int maxLogLineSizeBytes = Math.toIntExact(getMaxLogLineSizeKilobytes() * BYTES_IN_KILOBYTE);
|
||||
return new LokiAppender(
|
||||
getName(),
|
||||
getLayout(),
|
||||
@@ -140,6 +145,7 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
|
||||
getPropertyArray(),
|
||||
logLevelLabel,
|
||||
labelsDescriptor.getDynamicLabels(),
|
||||
maxLogLineSizeBytes,
|
||||
loggingSystem,
|
||||
monitoringModuleWrapper
|
||||
);
|
||||
@@ -220,10 +226,26 @@ public class LokiAppenderBuilder<B extends LokiAppenderBuilder<B>> extends Abstr
|
||||
return batchSize;
|
||||
}
|
||||
|
||||
public void setBatchSize(long batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
}
|
||||
|
||||
public long getBatchWait() {
|
||||
return batchWait;
|
||||
}
|
||||
|
||||
public void setBatchWait(long batchWait) {
|
||||
this.batchWait = batchWait;
|
||||
}
|
||||
|
||||
public void setMaxLogLineSizeKilobytes(int maxLogLineSizeKilobytes) {
|
||||
this.maxLogLineSizeKilobytes = maxLogLineSizeKilobytes;
|
||||
}
|
||||
|
||||
public long getMaxLogLineSizeKilobytes() {
|
||||
return maxLogLineSizeKilobytes;
|
||||
}
|
||||
|
||||
public void setMaxRequestsInFlight(int maxRequestsInFlight) {
|
||||
this.maxRequestsInFlight = maxRequestsInFlight;
|
||||
}
|
||||
|
||||
+6
-13
@@ -7,24 +7,15 @@ import org.apache.logging.log4j.core.layout.ByteBufferDestinationHelper;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.function.BiConsumer;
|
||||
|
||||
public class ByteBufferDestinations implements ByteBufferDestination {
|
||||
public class TjahziByteBufferDestination implements ByteBufferDestination {
|
||||
|
||||
private static final ThreadLocal<ByteBufferDestinations> THREAD_LOCAL = ThreadLocal.withInitial(ByteBufferDestinations::new);
|
||||
|
||||
/**
|
||||
* 10kb should be enough for everyone. Even in case of 1000s of threads we will allocate at most 10s of MB of buffers.
|
||||
* Larger messages will be fragmented.
|
||||
*/
|
||||
private final ByteBuffer buffer = ByteBuffer.allocate(10 * 1024);
|
||||
private final ByteBuffer buffer;
|
||||
|
||||
private BiConsumer<LogEvent, ByteBuffer> drain;
|
||||
private LogEvent context;
|
||||
|
||||
public static ByteBufferDestinations threadLocal() {
|
||||
ByteBufferDestinations result = THREAD_LOCAL.get();
|
||||
result.getByteBuffer().clear();
|
||||
|
||||
return result;
|
||||
public TjahziByteBufferDestination(int maxLogLineSizeBytes) {
|
||||
buffer = ByteBuffer.allocate(maxLogLineSizeBytes);
|
||||
}
|
||||
|
||||
public void drainRemaining() {
|
||||
@@ -60,5 +51,7 @@ public class ByteBufferDestinations implements ByteBufferDestination {
|
||||
public void initialize(BiConsumer<LogEvent, ByteBuffer> drain, LogEvent context) {
|
||||
this.context = context;
|
||||
this.drain = drain;
|
||||
|
||||
buffer.clear();
|
||||
}
|
||||
}
|
||||
@@ -33,7 +33,8 @@ class AppenderLogicTest {
|
||||
AppenderLogic logic = new AppenderLogic(
|
||||
loggingSystem,
|
||||
logLevelLabel,
|
||||
Map.of()
|
||||
Map.of(),
|
||||
ByteBufferDestinationRepository.DEFAULT_MAX_LINE_SIZE_BYTES
|
||||
);
|
||||
|
||||
MutableLogEvent logEvent = new MutableLogEvent();
|
||||
@@ -90,7 +91,8 @@ class AppenderLogicTest {
|
||||
AppenderLogic logic = new AppenderLogic(
|
||||
loggingSystem,
|
||||
logLevelLabel,
|
||||
dynamicLabels
|
||||
dynamicLabels,
|
||||
ByteBufferDestinationRepository.DEFAULT_MAX_LINE_SIZE_BYTES
|
||||
);
|
||||
|
||||
MutableLogEvent logEvent = new MutableLogEvent();
|
||||
|
||||
+100
@@ -0,0 +1,100 @@
|
||||
package pl.tkowalcz.tjahzi.log4j2;
|
||||
|
||||
import org.apache.logging.log4j.core.LogEvent;
|
||||
import org.apache.logging.log4j.core.impl.MutableLogEvent;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.BiConsumer;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class ByteBufferDestinationRepositoryTest {
|
||||
|
||||
@Test
|
||||
void shouldInitializeBufferToDesiredSize() {
|
||||
// Given
|
||||
int capacity = 42;
|
||||
ByteBufferDestinationRepository repository = new ByteBufferDestinationRepository(capacity);
|
||||
|
||||
BiConsumer<LogEvent, ByteBuffer> drain = (logEvent, byteBuffer) -> {
|
||||
};
|
||||
MutableLogEvent logEvent = new MutableLogEvent();
|
||||
|
||||
// When
|
||||
TjahziByteBufferDestination destination = repository.threadLocalDestination(drain, logEvent);
|
||||
|
||||
// Then
|
||||
assertThat(destination.getByteBuffer().capacity()).isEqualTo(capacity);
|
||||
assertThat(destination.getByteBuffer().remaining()).isEqualTo(capacity);
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldClearBufferBeforeUse() {
|
||||
// Given
|
||||
int capacity = 42;
|
||||
ByteBufferDestinationRepository repository = new ByteBufferDestinationRepository(capacity);
|
||||
|
||||
BiConsumer<LogEvent, ByteBuffer> drain = (logEvent, byteBuffer) -> {
|
||||
};
|
||||
MutableLogEvent logEvent = new MutableLogEvent();
|
||||
|
||||
TjahziByteBufferDestination destination = repository.threadLocalDestination(drain, logEvent);
|
||||
destination.writeBytes("blah".getBytes(), 0, 4);
|
||||
|
||||
// When
|
||||
TjahziByteBufferDestination actual = repository.threadLocalDestination(drain, logEvent);
|
||||
|
||||
// Then
|
||||
assertThat(actual.getByteBuffer().remaining()).isEqualTo(capacity);
|
||||
assertThat(actual.getByteBuffer().capacity()).isEqualTo(capacity);
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldReturnSameBufferInsideSameThread() {
|
||||
// Given
|
||||
int capacity = 42;
|
||||
ByteBufferDestinationRepository repository = new ByteBufferDestinationRepository(capacity);
|
||||
|
||||
BiConsumer<LogEvent, ByteBuffer> drain = (logEvent, byteBuffer) -> {
|
||||
};
|
||||
MutableLogEvent logEvent = new MutableLogEvent();
|
||||
|
||||
// When
|
||||
TjahziByteBufferDestination actual1 = repository.threadLocalDestination(drain, logEvent);
|
||||
TjahziByteBufferDestination actual2 = repository.threadLocalDestination(drain, logEvent);
|
||||
|
||||
// Then
|
||||
assertThat(actual1).isSameAs(actual2);
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldReturnDifferentSerializerInsideDifferentThreads() throws InterruptedException {
|
||||
// Given
|
||||
int capacity = 42;
|
||||
ByteBufferDestinationRepository repository = new ByteBufferDestinationRepository(capacity);
|
||||
|
||||
BiConsumer<LogEvent, ByteBuffer> drain = (logEvent, byteBuffer) -> {
|
||||
};
|
||||
MutableLogEvent logEvent = new MutableLogEvent();
|
||||
|
||||
AtomicReference<TjahziByteBufferDestination> captor1 = new AtomicReference<>();
|
||||
AtomicReference<TjahziByteBufferDestination> captor2 = new AtomicReference<>();
|
||||
|
||||
// When
|
||||
Thread thread1 = new Thread(() -> captor1.set(repository.threadLocalDestination(drain, logEvent)));
|
||||
thread1.start();
|
||||
thread1.join();
|
||||
|
||||
Thread thread2 = new Thread(() -> captor2.set(repository.threadLocalDestination(drain, logEvent)));
|
||||
thread2.start();
|
||||
thread2.join();
|
||||
|
||||
// Then
|
||||
assertThat(captor1.get()).isNotNull();
|
||||
assertThat(captor2.get()).isNotNull();
|
||||
|
||||
assertThat(captor1.get()).isNotSameAs(captor2.get());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
package pl.tkowalcz.tjahzi.log4j2;
|
||||
|
||||
import io.restassured.RestAssured;
|
||||
import io.restassured.http.ContentType;
|
||||
import io.restassured.parsing.Parser;
|
||||
import org.apache.commons.lang3.RandomStringUtils;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
import org.awaitility.Awaitility;
|
||||
import org.awaitility.Durations;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.testcontainers.containers.BindMode;
|
||||
import org.testcontainers.containers.GenericContainer;
|
||||
import org.testcontainers.containers.wait.strategy.Wait;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
import org.testcontainers.junit.jupiter.Testcontainers;
|
||||
|
||||
import java.net.URI;
|
||||
|
||||
import static io.restassured.RestAssured.given;
|
||||
import static org.hamcrest.CoreMatchers.equalTo;
|
||||
import static org.hamcrest.CoreMatchers.hasItems;
|
||||
|
||||
@Testcontainers
|
||||
class LogLineFragmentationTest {
|
||||
|
||||
@Container
|
||||
public GenericContainer loki = new GenericContainer("grafana/loki:latest")
|
||||
.withCommand("-config.file=/etc/loki-config.yaml")
|
||||
.withClasspathResourceMapping("loki-config.yaml",
|
||||
"/etc/loki-config.yaml",
|
||||
BindMode.READ_ONLY
|
||||
)
|
||||
.waitingFor(
|
||||
Wait.forHttp("/ready")
|
||||
.forPort(3100)
|
||||
)
|
||||
.withExposedPorts(3100);
|
||||
|
||||
@Test
|
||||
void shouldFragmentMessageOverLimit() throws Exception {
|
||||
// Given
|
||||
System.setProperty("loki.host", loki.getHost());
|
||||
System.setProperty("loki.port", loki.getFirstMappedPort().toString());
|
||||
|
||||
URI uri = getClass()
|
||||
.getClassLoader()
|
||||
.getResource("appender-test-large-log-line-fragmentation.xml")
|
||||
.toURI();
|
||||
|
||||
((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false))
|
||||
.setConfigLocation(uri);
|
||||
|
||||
String unfragmentedLogLine = RandomStringUtils.randomAlphanumeric(100 * 1024 - 1);
|
||||
String fragmentedLogLine = RandomStringUtils.randomAlphanumeric(110 * 1024);
|
||||
|
||||
long expectedTimestamp = System.currentTimeMillis();
|
||||
Logger logger = LogManager.getLogger(LogLineFragmentationTest.class);
|
||||
|
||||
// When
|
||||
logger.info(unfragmentedLogLine);
|
||||
logger.info(fragmentedLogLine);
|
||||
|
||||
// Then
|
||||
RestAssured.port = loki.getFirstMappedPort();
|
||||
RestAssured.baseURI = "http://" + loki.getHost();
|
||||
RestAssured.registerParser("text/plain", Parser.JSON);
|
||||
|
||||
Awaitility
|
||||
.await()
|
||||
.atMost(Durations.TEN_SECONDS)
|
||||
.pollInterval(Durations.ONE_SECOND)
|
||||
.ignoreExceptions()
|
||||
.untilAsserted(() -> given()
|
||||
.contentType(ContentType.URLENC)
|
||||
.urlEncodingEnabled(false)
|
||||
.formParam("&start=" + expectedTimestamp + "&limit=1000&query=%7Bserver%3D%22127.0.0.1%22%7D")
|
||||
.when()
|
||||
.get("/loki/api/v1/query_range")
|
||||
.then()
|
||||
.statusCode(200)
|
||||
.body("status", equalTo("success"))
|
||||
.body("data.result.size()", equalTo(1))
|
||||
.body("data.result[0].stream.server", equalTo("127.0.0.1"))
|
||||
.body("data.result[0].values.size()", equalTo(3))
|
||||
.body(
|
||||
"data.result.values",
|
||||
hasItems(
|
||||
hasItems(hasItems(fragmentedLogLine.substring(102_400, 110 * 1024))),
|
||||
hasItems(hasItems(fragmentedLogLine.substring(0, 102_400))),
|
||||
hasItems(hasItems(unfragmentedLogLine))
|
||||
)
|
||||
)
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<configuration packages="pl.tkowalcz.tjahzi.log4j2">
|
||||
<Loggers>
|
||||
<Root level="INFO">
|
||||
<AppenderRef ref="Loki"/>
|
||||
</Root>
|
||||
</Loggers>
|
||||
|
||||
<appenders>
|
||||
<Loki name="Loki" bufferSizeMegabytes="2" batchWait="10">
|
||||
<host>${sys:loki.host}</host>
|
||||
<port>${sys:loki.port}</port>
|
||||
|
||||
<maxLogLineSizeKilobytes>100</maxLogLineSizeKilobytes>
|
||||
|
||||
<ThresholdFilter level="ALL"/>
|
||||
<PatternLayout>
|
||||
<Pattern>%m</Pattern>
|
||||
</PatternLayout>
|
||||
|
||||
<Header name="server" value="127.0.0.1"/>
|
||||
<Label name="server" value="127.0.0.1"/>
|
||||
</Loki>
|
||||
</appenders>
|
||||
</configuration>
|
||||
Reference in new issue
Block a user