From e07322c25adfdfa978e8e6dc6722e672aad0427f Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Sun, 18 Oct 2020 22:00:37 +0200 Subject: [PATCH] Added Log4j2 Appender (still WiP but seems to be working. Added tests. --- .../pl/tkowalcz/tjahzi/LoggingSystem.java | 19 +- .../pl/tkowalcz/tjahzi/TjahziInitializer.java | 3 +- .../java/pl/tkowalcz/tjahzi/TjahziLogger.java | 60 +++- core/src/test/resources/log4j2.xml | 17 -- log4j2-appender/pom.xml | 8 +- .../pl/tkowalcz/tjahzi/log4j2/Header.java | 30 ++ .../java/pl/tkowalcz/tjahzi/log4j2/Label.java | 30 ++ .../tkowalcz/tjahzi/log4j2/LokiAppender.java | 269 ++++++++++++++++++ .../pl/tkowalcz/tjahzi/log4j2/Property.java | 54 ++++ .../tjahzi/log4j2/LokiAppenderTest.java | 100 +++++++ log4j2-appender/src/test/resources/log4j2.xml | 23 ++ .../src/test/resources/loki-config.yaml | 47 +++ 12 files changed, 633 insertions(+), 27 deletions(-) delete mode 100644 core/src/test/resources/log4j2.xml create mode 100644 log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Header.java create mode 100644 log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Label.java create mode 100644 log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java create mode 100644 log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Property.java create mode 100644 log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderTest.java create mode 100644 log4j2-appender/src/test/resources/log4j2.xml create mode 100644 log4j2-appender/src/test/resources/loki-config.yaml diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/LoggingSystem.java b/core/src/main/java/pl/tkowalcz/tjahzi/LoggingSystem.java index 4ce1677..9dc3a2a 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/LoggingSystem.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/LoggingSystem.java @@ -1,16 +1,33 @@ package pl.tkowalcz.tjahzi; +import org.agrona.concurrent.AgentRunner; import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer; +import java.util.function.Consumer; + public class LoggingSystem { private final ManyToOneRingBuffer logBuffer; + private final AgentRunner runner; - public LoggingSystem(ManyToOneRingBuffer logBuffer) { + public LoggingSystem( + ManyToOneRingBuffer logBuffer, + AgentRunner runner + ) { this.logBuffer = logBuffer; + this.runner = runner; } public TjahziLogger createLogger() { return new TjahziLogger(logBuffer); } + + public void close( + int retryCloseTimeoutMs, + Consumer closeFailAction) { + runner.close( + retryCloseTimeoutMs, + closeFailAction + ); + } } diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java index 6c2aac4..495011c 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziInitializer.java @@ -38,7 +38,8 @@ public class TjahziInitializer { AgentRunner.startOnThread(runner); return new LoggingSystem( - logBuffer + logBuffer, + runner ); } diff --git a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java index 820c0bb..6b7f90c 100644 --- a/core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java +++ b/core/src/main/java/pl/tkowalcz/tjahzi/TjahziLogger.java @@ -50,10 +50,44 @@ public class TjahziLogger { return this; } - private int calculateRequiredSizeAscii( - Map labels, - String line - ) { + public TjahziLogger log(long timestamp, + Map labels, + byte[] line) { + int requiredSize = calculateRequiredSizeAscii(labels, line); + int claim = logBuffer.tryClaim(LOG_MESSAGE_TYPE_ID, requiredSize); + + int cursor = claim; + if (requiredSize > 0) { + try { + AtomicBuffer buffer = this.logBuffer.buffer(); + + buffer.putLong(cursor, timestamp); + cursor += Long.BYTES; + + buffer.putInt(cursor, labels.size()); + cursor += Integer.BYTES; + + for (Map.Entry entry : labels.entrySet()) { + String key = entry.getKey(); + String value = entry.getValue(); + + cursor += buffer.putStringAscii(cursor, key); + cursor += buffer.putStringAscii(cursor, value); + } + + buffer.putInt(cursor, line.length); + cursor += Integer.BYTES; + buffer.putBytes(cursor, line); + logBuffer.commit(claim); + } catch (Throwable t) { + logBuffer.abort(claim); + } + } + + return this; + } + + private int calculateRequiredSizeAscii(Map labels) { return Long.BYTES + Integer.BYTES + labels @@ -61,8 +95,22 @@ public class TjahziLogger { .stream() .mapToInt(entry -> entry.getKey().length() + entry.getValue().length()) .map(size -> size + 2 * Integer.BYTES) - .sum() + - line.length() + Integer.BYTES; + .sum(); + } + private int calculateRequiredSizeAscii( + Map labels, + String line + ) { + return calculateRequiredSizeAscii(labels) + + line.length() + Integer.BYTES; + } + + private int calculateRequiredSizeAscii( + Map labels, + byte[] line + ) { + return calculateRequiredSizeAscii(labels) + + line.length + Integer.BYTES; } } diff --git a/core/src/test/resources/log4j2.xml b/core/src/test/resources/log4j2.xml deleted file mode 100644 index 12e6c62..0000000 --- a/core/src/test/resources/log4j2.xml +++ /dev/null @@ -1,17 +0,0 @@ - - - - - - - - - - - - - - - - - diff --git a/log4j2-appender/pom.xml b/log4j2-appender/pom.xml index 41af363..3a841fc 100644 --- a/log4j2-appender/pom.xml +++ b/log4j2-appender/pom.xml @@ -18,7 +18,6 @@ ${project.version} - org.apache.logging.log4j log4j-api @@ -42,7 +41,12 @@ 4.0.3 test - + + org.apache.logging.log4j + log4j-slf4j-impl + 2.13.3 + test + diff --git a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Header.java b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Header.java new file mode 100644 index 0000000..d95f1d7 --- /dev/null +++ b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Header.java @@ -0,0 +1,30 @@ +package pl.tkowalcz.tjahzi.log4j2; + +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.core.config.Node; +import org.apache.logging.log4j.core.config.plugins.Plugin; +import org.apache.logging.log4j.core.config.plugins.PluginAttribute; +import org.apache.logging.log4j.core.config.plugins.PluginFactory; +import org.apache.logging.log4j.core.config.plugins.PluginValue; +import org.apache.logging.log4j.status.StatusLogger; + +@Plugin(name = "header", category = Node.CATEGORY, printObject = true) +public class Header extends Property { + + private static final Logger LOGGER = StatusLogger.getLogger(); + + protected Header(String name, String value) { + super(name, value); + } + + @PluginFactory + public static Header createHeader( + @PluginAttribute("name") String name, + @PluginValue("value") String value) { + if (name == null) { + LOGGER.error("Property name cannot be null"); + } + + return new Header(name, value); + } +} diff --git a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Label.java b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Label.java new file mode 100644 index 0000000..32ed73e --- /dev/null +++ b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Label.java @@ -0,0 +1,30 @@ +package pl.tkowalcz.tjahzi.log4j2; + +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.core.config.Node; +import org.apache.logging.log4j.core.config.plugins.Plugin; +import org.apache.logging.log4j.core.config.plugins.PluginAttribute; +import org.apache.logging.log4j.core.config.plugins.PluginFactory; +import org.apache.logging.log4j.core.config.plugins.PluginValue; +import org.apache.logging.log4j.status.StatusLogger; + +@Plugin(name = "label", category = Node.CATEGORY, printObject = true) +public class Label extends Property { + + private static final Logger LOGGER = StatusLogger.getLogger(); + + private Label(String name, String value) { + super(name, value); + } + + @PluginFactory + public static Label createLabel( + @PluginAttribute("name") String name, + @PluginValue("value") String value) { + if (name == null) { + LOGGER.error("Property name cannot be null"); + } + + return new Label(name, value); + } +} diff --git a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java new file mode 100644 index 0000000..02d0386 --- /dev/null +++ b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/LokiAppender.java @@ -0,0 +1,269 @@ +package pl.tkowalcz.tjahzi.log4j2; + +import com.google.common.collect.Maps; +import org.apache.logging.log4j.core.Appender; +import org.apache.logging.log4j.core.Filter; +import org.apache.logging.log4j.core.Layout; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Node; +import org.apache.logging.log4j.core.config.Property; +import org.apache.logging.log4j.core.config.plugins.Plugin; +import org.apache.logging.log4j.core.config.plugins.PluginBuilderAttribute; +import org.apache.logging.log4j.core.config.plugins.PluginBuilderFactory; +import org.apache.logging.log4j.core.config.plugins.PluginElement; +import org.apache.logging.log4j.core.config.plugins.validation.constraints.Required; +import pl.tkowalcz.tjahzi.LoggingSystem; +import pl.tkowalcz.tjahzi.TjahziInitializer; +import pl.tkowalcz.tjahzi.http.ClientConfiguration; +import pl.tkowalcz.tjahzi.http.HttpClientFactory; +import pl.tkowalcz.tjahzi.http.NettyHttpClient; + +import java.io.Serializable; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.TimeUnit; + +/** + * Loki Appender. + */ +@Plugin(name = "Loki", category = Node.CATEGORY, elementType = Appender.ELEMENT_TYPE, printObject = true) +public class LokiAppender extends AbstractAppender { + + /** + * Builds LokiAppender instances. + * + * @param The type to build + */ + public static class Builder> extends AbstractAppender.Builder + implements org.apache.logging.log4j.core.util.Builder { + + public static final int BYTES_IN_MEGABYTE = 1024 * 1024; + + @PluginBuilderAttribute + @Required(message = "No Loki address provided for LokiAppender") + private String host; + + @PluginBuilderAttribute + @Required(message = "No Loki port provided for LokiAppender") + private int port; + + @PluginBuilderAttribute + private int connectTimeoutMillis = 0; + + @PluginBuilderAttribute + private int readTimeoutMillis = 0; + + @PluginBuilderAttribute + private int maxRetries = 0; + + @PluginBuilderAttribute + private int bufferSizeMegabytes = 32; + + @PluginBuilderAttribute + private boolean useOffHeapBuffer = true; + + @PluginElement("Headers") + private Header[] headers; + + @PluginElement("Labels") + private Label[] labels; + + @Override + public LokiAppender build() { + ClientConfiguration configurationBuilder = ClientConfiguration.builder() + .withHost(host) + .withConnectionTimeoutMillis(connectTimeoutMillis) + .withPort(port) + .withMaxRetries(maxRetries) + .withRequestTimeoutMillis(readTimeoutMillis) + .build(); + + NettyHttpClient httpClient = HttpClientFactory + .defaultFactory() + .getHttpClient(configurationBuilder); + + LoggingSystem loggingSystem = new TjahziInitializer().createLoggingSystem( + httpClient, + getBufferSizeMegabytes() * BYTES_IN_MEGABYTE, + isUseOffHeapBuffer() + ); + + HashMap lokiLabels = Maps.newHashMap(); + Arrays.stream(labels).forEach(property -> { + String value = getConfiguration() + .getStrSubstitutor() + .replace(property.getValue()); + + lokiLabels.put( + property.getName(), + value + ); + } + ); + + return new LokiAppender( + getName(), + getLayout(), + getFilter(), + isIgnoreExceptions(), + getPropertyArray(), + lokiLabels, + loggingSystem + ); + } + + public String getHost() { + return host; + } + + public int getConnectTimeoutMillis() { + return connectTimeoutMillis; + } + + public int getReadTimeoutMillis() { + return readTimeoutMillis; + } + + public Header[] getHeaders() { + return headers; + } + + public B setHost(String host) { + this.host = host; + return asBuilder(); + } + + public B setConnectTimeoutMillis(final int connectTimeoutMillis) { + this.connectTimeoutMillis = connectTimeoutMillis; + return asBuilder(); + } + + public B setReadTimeoutMillis(final int readTimeoutMillis) { + this.readTimeoutMillis = readTimeoutMillis; + return asBuilder(); + } + + public int getPort() { + return port; + } + + public void setPort(int port) { + this.port = port; + } + + 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 void setHeaders(Header[] headers) { + this.headers = headers; + } + + public Label[] getLabels() { + return labels; + } + + public void setLabels(Label[] labels) { + this.labels = labels; + } + } + + /** + * @return a builder for a LokiAppender. + */ + @PluginBuilderFactory + public static > B newBuilder() { + return new LokiAppender.Builder().asBuilder(); + } + + private final Map lokiLabels; + private final LoggingSystem loggingSystem; + + private LokiAppender( + String name, + Layout layout, + Filter filter, + boolean ignoreExceptions, + Property[] properties, + Map lokiLabels, + LoggingSystem loggingSystem) { + super( + name, + filter, + layout, + ignoreExceptions, + properties + ); + Objects.requireNonNull(layout, "layout"); + + this.lokiLabels = lokiLabels; + this.loggingSystem = loggingSystem; + } + + @Override + public void start() { + super.start(); + } + + @Override + public void append(final LogEvent event) { + byte[] bytes = getLayout().toByteArray(event); + loggingSystem.createLogger().log( + event.getTimeMillis(), + lokiLabels, + bytes + ); + } + + @Override + public boolean stop(long timeout, TimeUnit timeUnit) { + setStopping(); + + boolean stopped = super.stop( + timeout, + timeUnit, + false + ); + + loggingSystem.close( + (int) timeUnit.toMillis(timeout), + thread -> { + error(""); + } + ); + + setStopped(); + return stopped; + } + + @Override + public String toString() { + return "LokiAppender{" + + "name=" + getName() + + ", state=" + getState() + + '}'; + } +} diff --git a/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Property.java b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Property.java new file mode 100644 index 0000000..fc5c6ca --- /dev/null +++ b/log4j2-appender/src/main/java/pl/tkowalcz/tjahzi/log4j2/Property.java @@ -0,0 +1,54 @@ +package pl.tkowalcz.tjahzi.log4j2; + +import org.apache.logging.log4j.util.Strings; + +import java.util.Objects; + +/** + * Class copied from org.apache.logging.log4j.core.config.Property. I have no idea how to reuse that class while assigning + * it a custom tag name e.g. headers/labels. + */ +public class Property { + + private final String name; + private final String value; + private final boolean valueNeedsLookup; + + protected Property(final String name, final String value) { + this.name = name; + this.value = value; + this.valueNeedsLookup = value != null && value.contains("${"); + } + + /** + * Returns the property name. + * + * @return the property name. + */ + public String getName() { + return name; + } + + /** + * Returns the property value. + * + * @return the value of the property. + */ + public String getValue() { + return Objects.toString(value, Strings.EMPTY); // LOG4J2-1313 null would be same as Property not existing + } + + /** + * Returns {@code true} if the value contains a substitutable property that requires a lookup to be resolved. + * + * @return {@code true} if the value contains {@code "${"}, {@code false} otherwise + */ + public boolean isValueNeedsLookup() { + return valueNeedsLookup; + } + + @Override + public String toString() { + return name + '=' + getValue(); + } +} diff --git a/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderTest.java b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderTest.java new file mode 100644 index 0000000..2f38c92 --- /dev/null +++ b/log4j2-appender/src/test/java/pl/tkowalcz/tjahzi/log4j2/LokiAppenderTest.java @@ -0,0 +1,100 @@ +package pl.tkowalcz.tjahzi.log4j2; + +import io.restassured.RestAssured; +import io.restassured.http.ContentType; +import io.restassured.parsing.Parser; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.awaitility.Awaitility; +import org.awaitility.Durations; +import org.hamcrest.BaseMatcher; +import org.hamcrest.Description; +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.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 sendData() { + System.setProperty("loki.host", loki.getHost()); + System.setProperty("loki.port", loki.getFirstMappedPort().toString()); + ((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false)).reconfigure(); + + String expectedLogLine = "Test"; + long expectedTimestamp = System.currentTimeMillis(); + + Logger logger = LogManager.getLogger(LokiAppenderTest.class); + logger.info(expectedLogLine); + + 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 list = (List) 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) { + + } + } + + )); + }); + } +} diff --git a/log4j2-appender/src/test/resources/log4j2.xml b/log4j2-appender/src/test/resources/log4j2.xml new file mode 100644 index 0000000..663c75b --- /dev/null +++ b/log4j2-appender/src/test/resources/log4j2.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + ${sys:loki.host} + ${sys:loki.port} + + + + %X{tid} [%t] %d{MM-dd HH:mm:ss.SSS} %5p %c{1} - %m%n%exception{full} + + +
+