Re #36 - initial stab at implementing logback logger.

This commit is contained in:
tomasz.kowalczewski committed 2021-02-28 19:04:07 +01:00
1 parent 0f65c24442
commit a834b22c8f
12 files changed
+738 -3

No files matched your search

@@ -52,7 +52,7 @@ public class LabelFactory {
if (!duplicatedLabels.isEmpty()) {
LOGGER.error(
"There are duplicated labels which is not allowed by Loki. " +
"These labels will be deduplicated undeterministically: {}",
"These labels will be deduplicated non-deterministically: {}",
duplicatedLabels
);
}
@@ -67,7 +67,8 @@ public class LabelFactory {
return Stream.of(label);
}
LOGGER.error("Label '{}' contains invalid characters - ignoring it. {}",
LOGGER.error(
"Ignoring label '{}' - contains invalid characters. {}",
label.getName(),
GitHubDocs.LABEL_NAMING.getLogMessage()
);
@@ -85,7 +86,8 @@ public class LabelFactory {
String logLevelLabel
) {
if (!Label.hasValidName(logLevelLabel)) {
LOGGER.error("Log level label '{}' contains invalid characters - ignoring it. {}",
LOGGER.error(
"Ignoring log level label '{}' - contains invalid characters. {}",
logLevelLabel,
GitHubDocs.LABEL_NAMING.getLogMessage()
);
+86
View File
@@ -0,0 +1,86 @@
<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>
<parent>
<groupId>pl.tkowalcz.tjahzi</groupId>
<artifactId>tjahzi-parent</artifactId>
<version>0.9.5-SNAPSHOT</version>
</parent>
<name>logback-appender</name>
<artifactId>logback-appender</artifactId>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>pl.tkowalcz.tjahzi</groupId>
<artifactId>core</artifactId>
<version>0.9.5-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.2.3</version>
</dependency>
<dependency>
<groupId>com.google.code.java-allocation-instrumenter</groupId>
<artifactId>java-allocation-instrumenter</artifactId>
<version>3.3.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.rest-assured</groupId>
<artifactId>rest-assured</artifactId>
<version>4.3.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<version>4.0.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<version>3.15.0</version>
<scope>test</scope>
</dependency>
<!-- Testcontainers -->
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<version>5.6.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-params</artifactId>
<version>5.6.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
<version>5.6.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<version>1.14.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<version>1.14.3</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>
@@ -0,0 +1,23 @@
package pl.tkowalcz.tjahzi.logback;
public class Header {
private String name;
private String value;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getValue() {
return value;
}
public void setValue(String value) {
this.value = value;
}
}
@@ -0,0 +1,35 @@
package pl.tkowalcz.tjahzi.logback;
import java.util.regex.Pattern;
public class Label {
private static final Pattern LABEL_NAME_PATTER = Pattern.compile("[a-zA-Z_][a-zA-Z0-9_]*");
private String name;
private String value;
public boolean hasValidName() {
return hasValidName(getName());
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getValue() {
return value;
}
public void setValue(String value) {
this.value = value;
}
public static boolean hasValidName(String label) {
return LABEL_NAME_PATTER.matcher(label).matches();
}
}
@@ -0,0 +1,103 @@
package pl.tkowalcz.tjahzi.logback;
import pl.tkowalcz.tjahzi.github.GitHubDocs;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.util.Arrays.stream;
import static java.util.stream.Collectors.counting;
public class LabelFactory {
private final String logLevelLabel;
private final Label[] labels;
public LabelFactory(String logLevelLabel, Label... labels) {
this.logLevelLabel = logLevelLabel;
this.labels = labels;
}
public HashMap<String, String> convertLabelsDroppingInvalid() {
detectAndLogDuplicateLabels();
return convertAndLogViolations();
}
public String validateLogLevelLabel(HashMap<String, String> existingLabels) {
if (logLevelLabel != null) {
return validateLogLevelLabelAgainst(
existingLabels,
logLevelLabel
);
}
return null;
}
private void detectAndLogDuplicateLabels() {
List<String> duplicatedLabels = stream(labels)
.collect(Collectors.groupingBy(Label::getName, counting()))
.entrySet()
.stream().filter(entry -> entry.getValue() > 1)
.map(Map.Entry::getKey)
.collect(Collectors.toList());
if (!duplicatedLabels.isEmpty()) {
System.out.printf(
"There are duplicated labels which is not allowed by Loki. " +
"These labels will be deduplicated non-deterministically: %s\n",
duplicatedLabels
);
}
}
private HashMap<String, String> convertAndLogViolations() {
HashMap<String, String> lokiLabels = new HashMap<>();
stream(labels)
.flatMap(label -> {
if (label.hasValidName()) {
return Stream.of(label);
}
System.out.printf(
"Ignoring label '%s' - contains invalid characters. %s\n",
label.getName(),
GitHubDocs.LABEL_NAMING.getLogMessage()
);
return Stream.of();
}
)
.forEach(__ -> lokiLabels.put(__.getName(), __.getValue()));
return lokiLabels;
}
private String validateLogLevelLabelAgainst(
Map<String, String> existingLabels,
String logLevelLabel
) {
if (!Label.hasValidName(logLevelLabel)) {
System.out.printf(
"Ignoring log level label '%s' - contains invalid characters. %s\n",
logLevelLabel,
GitHubDocs.LABEL_NAMING.getLogMessage()
);
return null;
}
if (existingLabels.remove(logLevelLabel) != null) {
System.out.printf(
"Ignoring log level label '%s' - conflicts with label defined in configuration.\n",
logLevelLabel
);
}
return logLevelLabel;
}
}
@@ -0,0 +1,68 @@
package pl.tkowalcz.tjahzi.logback;
import ch.qos.logback.classic.encoder.PatternLayoutEncoder;
import ch.qos.logback.classic.spi.ILoggingEvent;
import pl.tkowalcz.tjahzi.LoggingSystem;
import pl.tkowalcz.tjahzi.TjahziLogger;
import java.nio.ByteBuffer;
import java.util.Collections;
public class LokiAppender extends LokiAppenderConfigurator {
private PatternLayoutEncoder encoder;
private LoggingSystem loggingSystem;
private TjahziLogger logger;
private String logLevelLabel;
public PatternLayoutEncoder getEncoder() {
return encoder;
}
public void setEncoder(PatternLayoutEncoder encoder) {
this.encoder = encoder;
}
@Override
protected void append(ILoggingEvent event) {
String logLevel = event.getLevel().toString();
byte[] logLine = encoder.encode(event);
logger.log(
event.getTimeStamp(),
Collections.emptyMap(),
logLevelLabel,
logLevel,
ByteBuffer.wrap(logLine)
);
}
@Override
public void start() {
if (encoder == null) {
addError("No encoder set for the appender named [" + name + "].");
return;
}
encoder.start();
LokiAppenderFactory lokiAppenderFactory = new LokiAppenderFactory(this);
loggingSystem = lokiAppenderFactory.createAppender();
logLevelLabel = lokiAppenderFactory.getLogLevelLabel();
logger = loggingSystem.createLogger();
loggingSystem.start();
super.start();
}
@Override
public void stop() {
loggingSystem.close(
1000,
thread -> System.out.println("Loki appender was unable to stop thread on shutdown: " + thread)
);
super.stop();
}
}
@@ -0,0 +1,136 @@
package pl.tkowalcz.tjahzi.logback;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.UnsynchronizedAppenderBase;
import java.util.ArrayList;
import java.util.List;
public abstract class LokiAppenderConfigurator extends UnsynchronizedAppenderBase<ILoggingEvent> {
static final int BYTES_IN_MEGABYTE = 1024 * 1024;
private String host;
private int port;
private int connectTimeoutMillis = 5000;
private int readTimeoutMillis = 60_000;
private int maxRetries = 3;
private int bufferSizeMegabytes = 32;
private boolean useOffHeapBuffer = true;
private String logLevelLabel;
private long batchSize = 10_2400;
private long batchWait = 5;
private int maxRequestsInFlight = 100;
private final List<Header> headers = new ArrayList<>();
private final List<Label> labels = new ArrayList<>();
public String getHost() {
return host;
}
public void setHost(String host) {
this.host = host;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
public int getConnectTimeoutMillis() {
return connectTimeoutMillis;
}
public void setConnectTimeoutMillis(int connectTimeoutMillis) {
this.connectTimeoutMillis = connectTimeoutMillis;
}
public int getReadTimeoutMillis() {
return readTimeoutMillis;
}
public void setReadTimeoutMillis(int readTimeoutMillis) {
this.readTimeoutMillis = readTimeoutMillis;
}
public int getMaxRetries() {
return maxRetries;
}
public void setMaxRetries(int maxRetries) {
this.maxRetries = maxRetries;
}
public int getBufferSizeMegabytes() {
return bufferSizeMegabytes;
}
public void setBufferSizeMegabytes(int bufferSizeMegabytes) {
this.bufferSizeMegabytes = bufferSizeMegabytes;
}
public boolean isUseOffHeapBuffer() {
return useOffHeapBuffer;
}
public void setUseOffHeapBuffer(boolean useOffHeapBuffer) {
this.useOffHeapBuffer = useOffHeapBuffer;
}
public String getLogLevelLabel() {
return logLevelLabel;
}
public void setLogLevelLabel(String logLevelLabel) {
this.logLevelLabel = logLevelLabel;
}
public void setBatchSize(long batchSize) {
this.batchSize = batchSize;
}
public void setBatchWait(long batchWait) {
this.batchWait = batchWait;
}
public long getBatchSize() {
return batchSize;
}
public long getBatchWait() {
return batchWait;
}
public int getMaxRequestsInFlight() {
return maxRequestsInFlight;
}
public void setMaxRequestsInFlight(int maxRequestsInFlight) {
this.maxRequestsInFlight = maxRequestsInFlight;
}
public void addHeader(Header header) {
headers.add(header);
}
public List<Header> getHeaders() {
return headers;
}
public void addLabel(Label label) {
labels.add(label);
}
public List<Label> getLabels() {
return labels;
}
}
@@ -0,0 +1,79 @@
package pl.tkowalcz.tjahzi.logback;
import pl.tkowalcz.tjahzi.LoggingSystem;
import pl.tkowalcz.tjahzi.TjahziInitializer;
import pl.tkowalcz.tjahzi.github.GitHubDocs;
import pl.tkowalcz.tjahzi.http.ClientConfiguration;
import pl.tkowalcz.tjahzi.http.HttpClientFactory;
import pl.tkowalcz.tjahzi.http.NettyHttpClient;
import pl.tkowalcz.tjahzi.stats.MutableMonitoringModuleWrapper;
import pl.tkowalcz.tjahzi.stats.StandardMonitoringModule;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
public class LokiAppenderFactory {
private final LokiAppenderConfigurator configurator;
private final HashMap<String, String> lokiLabels;
private final String logLevelLabel;
public LokiAppenderFactory(LokiAppenderConfigurator configurator) {
this.configurator = configurator;
LabelFactory labelFactory = new LabelFactory(configurator.getLogLevelLabel(), configurator.getLabels().toArray(new Label[0]));
lokiLabels = labelFactory.convertLabelsDroppingInvalid();
logLevelLabel = labelFactory.validateLogLevelLabel(lokiLabels);
}
public LoggingSystem createAppender() {
ClientConfiguration configurationBuilder = ClientConfiguration.builder()
.withHost(configurator.getHost())
.withConnectionTimeoutMillis(configurator.getConnectTimeoutMillis())
.withPort(configurator.getPort())
.withMaxRetries(configurator.getMaxRetries())
.withRequestTimeoutMillis(configurator.getReadTimeoutMillis())
.withMaxRequestsInFlight(configurator.getMaxRequestsInFlight())
.build();
String[] additionalHeaders = configurator.getHeaders().stream()
.flatMap(header -> Stream.of(header.getName(), header.getValue()))
.toArray(String[]::new);
MutableMonitoringModuleWrapper monitoringModuleWrapper = new MutableMonitoringModuleWrapper();
monitoringModuleWrapper.setMonitoringModule(new StandardMonitoringModule());
NettyHttpClient httpClient = HttpClientFactory
.defaultFactory()
.getHttpClient(
configurationBuilder,
monitoringModuleWrapper,
additionalHeaders
);
int bufferSizeBytes = configurator.getBufferSizeMegabytes() * LokiAppenderConfigurator.BYTES_IN_MEGABYTE;
if (TjahziInitializer.isCorrectSize(bufferSizeBytes)) {
System.out.printf(
"Invalid log buffer size %d - using nearest power of two greater than provided value, no less than 1MB. %s\n",
bufferSizeBytes,
GitHubDocs.LOG_BUFFER_SIZING.getLogMessage()
);
}
return new TjahziInitializer().createLoggingSystem(
httpClient,
monitoringModuleWrapper,
lokiLabels,
configurator.getBatchSize(),
TimeUnit.SECONDS.toMillis(configurator.getBatchWait()),
bufferSizeBytes,
configurator.isUseOffHeapBuffer()
);
}
public String getLogLevelLabel() {
return logLevelLabel;
}
}
@@ -0,0 +1,121 @@
package pl.tkowalcz.tjahzi.logback;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.joran.spi.JoranException;
import io.restassured.RestAssured;
import io.restassured.http.ContentType;
import io.restassured.parsing.Parser;
import org.awaitility.Awaitility;
import org.awaitility.Durations;
import org.hamcrest.BaseMatcher;
import org.hamcrest.Description;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static io.restassured.RestAssured.given;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.hasItems;
@Testcontainers
class LokiAppenderTest {
@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 shouldSendData() throws URISyntaxException, MalformedURLException, JoranException {
// Given
System.setProperty("loki.host", loki.getHost());
System.setProperty("loki.port", loki.getFirstMappedPort().toString());
URI uri = getClass()
.getClassLoader()
.getResource("logback-configuration.xml")
.toURI();
LoggerContext context = (LoggerContext) LoggerFactory.getILoggerFactory();
JoranConfigurator configurator = new JoranConfigurator();
configurator.setContext(context);
context.reset();
configurator.doConfigure(uri.toURL());
String expectedLogLine = "Test";
long expectedTimestamp = System.currentTimeMillis();
// When
Logger logger = context.getLogger(LokiAppenderTest.class);
logger.info(expectedLogLine);
// 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("query=%7Bserver%3D%22127.0.0.1%22%7D")
.when()
.get("/loki/api/v1/query_range")
.then()
.log()
.all()
.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", hasItems(new BaseMatcher<>() {
@Override
public boolean matches(Object o) {
List<Object> list = (List<Object>) o;
if (list.size() != 2) {
return false;
}
long actualTimestamp = Long.parseLong(list.get(0).toString());
String actualLogLine = list.get(1).toString();
return actualLogLine.contains(expectedLogLine)
&& (expectedTimestamp - actualTimestamp) < TimeUnit.MINUTES.toMillis(1);
}
@Override
public void describeTo(Description description) {
}
}
));
});
}
}
@@ -0,0 +1,33 @@
<configuration debug="true">
<appender name="Loki" class="pl.tkowalcz.tjahzi.logback.LokiAppender">
<host>${loki.host}</host>
<port>${loki.port}</port>
<encoder>
<pattern>%-4relative [%thread] %-5level %logger{35} - %msg%n</pattern>
</encoder>
<header>
<name>server</name>
<value>127.0.0.1</value>
</header>
<header>
<name>X-Org-Id</name>
<value>Codewise</value>
</header>
<label>
<name>server</name>
<value>127.0.0.1</value>
</label>
<logLevelLabel>
log_level
</logLevelLabel>
</appender>
<root level="debug">
<appender-ref ref="Loki"/>
</root>
</configuration>
@@ -0,0 +1,48 @@
auth_enabled: false
server:
http_listen_port: 3100
log_level: debug
ingester:
lifecycler:
address: 127.0.0.1
ring:
kvstore:
store: inmemory
replication_factor: 1
min_ready_duration: 0s
final_sleep: 0s
chunk_idle_period: 5m
chunk_retain_period: 30s
max_transfer_retries: 0
schema_config:
configs:
- from: 2018-04-15
store: boltdb
object_store: filesystem
schema: v11
index:
prefix: index_
period: 168h
storage_config:
boltdb:
directory: /loki/index
filesystem:
directory: /loki/chunks
limits_config:
enforce_metric_name: false
reject_old_samples: true
reject_old_samples_max_age: 168h
max_entries_limit_per_query: 10000
chunk_store_config:
max_look_back_period: 0s
table_manager:
retention_deletes_enabled: false
retention_period: 0s
+1
View File
@@ -15,6 +15,7 @@
<module>loki-protobuf</module>
<module>log4j2-appender</module>
<module>log4j2-appender-nodep</module>
<module>logback-appender</module>
<module>core</module>
</modules>