Logback Encoder Support for HEC

.github/workflows/test.yml

@@ -41,20 +41,6 @@ jobs:
     runs-on: ubuntu-latest
     container: maven:3-jdk-8
 
-    services:
-      splunk:
-        image: splunk/splunk:${{matrix.splunk-version}}
-        env:
-          SPLUNK_START_ARGS: --accept-license
-          SPLUNK_PASSWORD: changed!
-        ports:
-          - 8089
-          - 8088
-          - 5555
-          - 15000
-          - 10667
-          - 10668/udp
-
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK
@@ -72,4 +58,5 @@ jobs:
         run: mvn -P AcceptanceTest -B verify --file pom.xml
         env:
           SPLUNK_PASSWORD: changed!
-          SPLUNK_HOST: splunk
+          SPLUNK_HOST: splunk
+          USE_IMAGE_VERSION: ${{matrix.splunk-version}}

pom.xml

@@ -126,7 +126,7 @@
                         <version>2.22.2</version>
                         <configuration>
                             <includes>
-                                <include>**/HttpEventCollector_*.class</include>
+                                <include>**/TestSuiteAcceptanceTests.class</include>
                             </includes>
                         </configuration>
                     </plugin>
@@ -160,7 +160,7 @@
                         <version>2.22.2</version>
                         <configuration>
                             <includes>
-                                <include>**/HttpLoggerStressTest.class</include>
+                                <include>**/TestSuiteStressTests.class</include>
                             </includes>
                         </configuration>
                     </plugin>
@@ -198,6 +198,18 @@
             <version>1.7.36</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>1.20.1</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>net.logstash.logback</groupId>
+            <artifactId>logstash-logback-encoder</artifactId>
+            <version>7.3</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-classic</artifactId>

src/main/java/com/splunk/logging/HttpEventCollectorLogbackAppender.java

@@ -22,19 +22,27 @@
 import ch.qos.logback.classic.spi.StackTraceElementProxy;
 import ch.qos.logback.core.AppenderBase;
 import ch.qos.logback.core.Layout;
+import ch.qos.logback.core.encoder.Encoder;
+import ch.qos.logback.core.encoder.LayoutWrappingEncoder;
+import ch.qos.logback.core.spi.DeferredProcessingAware;
 import com.google.gson.Gson;
 import com.splunk.logging.hec.MetadataTags;
 
+import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
 /**
  * Logback Appender which writes its events to Splunk http event collector rest endpoint.
+ * <b>Remarks: </b>
+ * {@link ch.qos.logback.core.spi.DeferredProcessingAware} is the super-interface of all loggable events.
  */
-public class HttpEventCollectorLogbackAppender<E> extends AppenderBase<E> {
+public class HttpEventCollectorLogbackAppender<E extends DeferredProcessingAware>
+        extends AppenderBase<E> {
+
     private HttpEventCollectorSender sender = null;
-    private Layout<E> _layout;
+    private Encoder<E> _encoder;
     private boolean _includeLoggerName = true;
     private boolean _includeThreadName = true;
     private boolean _includeMDC = true;
@@ -59,11 +67,17 @@ public class HttpEventCollectorLogbackAppender<E> extends AppenderBase<E> {
     private long _batchSize = 0;
     private String _sendMode;
     private long _retriesOnError = 0;
-    private Map<String, String> _metadata = new HashMap<>();
+    private final Map<String, String> _metadata = new HashMap<>();
     private boolean _batchingConfigured = false;
 
 
-    private HttpEventCollectorSender.TimeoutSettings timeoutSettings = new HttpEventCollectorSender.TimeoutSettings();
+    private final HttpEventCollectorSender.TimeoutSettings timeoutSettings = new HttpEventCollectorSender.TimeoutSettings();
+
+    private static class EncodeFailException extends Exception {
+        EncodeFailException(Throwable cause) {
+            super(cause);
+        }
+    }
 
     @Override
     public void start() {
@@ -92,6 +106,16 @@ public void start() {
             throw new IllegalArgumentException("Batching configuration and sending type of raw are incompatible.");
         }
 
+        // encoder
+        if (this._encoder == null) {
+            addError("No encoder set for the appender named [" + name + "].");
+        } else {
+            _encoder.setContext(this.context);
+            if (!_encoder.isStarted()) {
+                _encoder.start();
+            }
+        }
+
         this.sender = new HttpEventCollectorSender(
                 _url, _token, _channel, _type, _batchInterval, _batchCount, _batchSize, _sendMode, metadata, timeoutSettings);
 
@@ -141,6 +165,7 @@ public void stop() {
         if (!started)
             return;
         this.sender.close();
+        _encoder.stop();
         super.stop();
     }
 
@@ -153,6 +178,19 @@ protected void append(E e) {
         }
     }
 
+    private String encodeMessage(E message) throws EncodeFailException {
+        byte[] encoded;
+
+        try {
+            encoded = _encoder.encode(message);
+        } catch (Exception e) {
+            throw new EncodeFailException(e);
+        }
+
+        return new String(encoded, StandardCharsets.UTF_8);
+    }
+
+    @SuppressWarnings("unchecked")
     private void sendEvent(ILoggingEvent event) {
         event.prepareForDeferredProcessing();
         if (event.hasCallerData()) {
@@ -189,31 +227,32 @@ private void sendEvent(ILoggingEvent event) {
             // No actions here
         }
 
-        MarkerConverter c = new MarkerConverter();
         if (this.started) {
-            this.sender.send(
-            		event.getTimeStamp(),
-                    event.getLevel().toString(),
-                    _layout.doLayout((E) event),
-                    _includeLoggerName ? event.getLoggerName() : null,
-                    _includeThreadName ? event.getThreadName() : null,
-                    _includeMDC ? event.getMDCPropertyMap() : null,
-                    (_includeException && isExceptionOccured) ? exceptionDetail : null,
-                    c.convert(event)
-            );
+            MarkerConverter c = new MarkerConverter();
+            try {
+                this.sender.send(
+                        event.getTimeStamp(),
+                        event.getLevel().toString(),
+                        this.encodeMessage((E) event), // Fine. E is super of ILoggingEvent
+                        _includeLoggerName ? event.getLoggerName() : null,
+                        _includeThreadName ? event.getThreadName() : null,
+                        _includeMDC ? event.getMDCPropertyMap() : null,
+                        (_includeException && isExceptionOccured) ? exceptionDetail : null,
+                        c.convert(event)
+                );
+            } catch (EncodeFailException e) {
+                addWarn("Failed to encode event. Dropping event.", e.getCause());
+            }
         }
     }
 
     // send non ILoggingEvent such as ch.qos.logback.access.spi.IAccessEvent
     private void sendEvent(E e) {
-        String message = _layout.doLayout(e);
-        if (message == null) {
-            throw new IllegalArgumentException(String.format(
-                    "The logback layout %s is probably incorrect, " +
-                            "and fails to format the message.",
-                    _layout.toString()));
+        try {
+            this.sender.send(this.encodeMessage(e));
+        } catch (EncodeFailException ex) {
+            addWarn("Failed to encode event. Dropping event.", ex.getCause());
         }
-        this.sender.send(message);
     }
 
     public void setUrl(String url) {
@@ -261,11 +300,22 @@ public String getType() {
     }
 
     public void setLayout(Layout<E> layout) {
-        this._layout = layout;
+        this.addWarn("This appender no longer admits a layout as a sub-component, set an encoder instead.");
+        this.addWarn("To ensure compatibility, wrapping your layout in LayoutWrappingEncoder.");
+        this.addWarn("See also http://logback.qos.ch/codes.html#layoutInsteadOfEncoder for details");
+
+        LayoutWrappingEncoder<E> layoutWrappingEncoder = new LayoutWrappingEncoder<>();
+        layoutWrappingEncoder.setLayout(layout);
+        layoutWrappingEncoder.setContext(this.context);
+        this._encoder = layoutWrappingEncoder;
+    }
+
+    public Encoder<E> getEncoder() {
+        return this._encoder;
     }
 
-    public Layout<E> getLayout() {
-        return this._layout;
+    public void setEncoder(Encoder<E> encoder) {
+        this._encoder = encoder;
     }
 
     public boolean getIncludeLoggerName() {

src/test/java/HttpEventCollector_Log4j2Test.java

@@ -30,6 +30,7 @@
 import java.util.List;
 
 public final class HttpEventCollector_Log4j2Test {
+
     private String httpEventCollectorName = "Log4j2Test";
     final List<List<HttpEventCollectorEventInfo>> errors = new ArrayList<>();
     final List<Exception> logEx = new ArrayList<>();

src/test/java/HttpEventCollector_Test.java

@@ -17,22 +17,25 @@
  */
 
 import ch.qos.logback.core.joran.spi.JoranException;
+import com.splunk.*;
 import com.splunk.logging.HttpEventCollectorErrorHandler;
 import com.splunk.logging.HttpEventCollectorEventInfo;
 import org.junit.Assert;
 import org.junit.Test;
 
-import java.io.*;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Method;
 import java.net.URI;
 import java.net.URL;
 import java.net.URLClassLoader;
-import java.util.*;
-import java.lang.reflect.*;
-
-import com.splunk.*;
-import org.slf4j.*;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
 
 public class HttpEventCollector_Test {
+
     public static void addPath(String s) throws Exception {
         File f = new File(s);
         URI u = f.toURI();

src/test/java/TestRuleSplunkContainer.java

@@ -0,0 +1,70 @@
+
+import org.junit.rules.TestRule;
+import org.junit.runner.Description;
+import org.junit.runners.model.MultipleFailureException;
+import org.junit.runners.model.Statement;
+import org.testcontainers.containers.FixedHostPortGenericContainer;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.InternetProtocol;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+public class TestRuleSplunkContainer implements TestRule {
+    // Allow choosing image version from CI
+    private static String dockerImageName() {
+        String version = System.getenv().getOrDefault("USE_IMAGE_VERSION", "latest");
+        return String.join(
+                ":",
+                "splunk/splunk",
+                version);
+    }
+
+    @Override
+    public Statement apply(Statement base, Description description) {
+        try (
+                GenericContainer<?> splunk = new FixedHostPortGenericContainer<>(dockerImageName())
+                        .withFixedExposedPort(8000, 8000)
+                        .withFixedExposedPort(5555, 5555)
+                        .withFixedExposedPort(8088, 8088)
+                        .withFixedExposedPort(8089, 8089)
+                        .withFixedExposedPort(15000, 15000)
+                        .withFixedExposedPort(10667, 10667)
+                        .withFixedExposedPort(10668, 10668, InternetProtocol.UDP)
+                        .withExposedPorts(9997)
+                        .withEnv("SPLUNK_START_ARGS", "--accept-license")
+                        .withEnv("SPLUNK_PASSWORD", "changed!")
+                        .withStartupTimeout(Duration.ofMinutes(2L))
+                        .waitingFor(Wait.forListeningPorts(9997)) // must wait on an unfixed port
+        ) {
+            return wrapTestCase(base, splunk);
+        }
+    }
+
+    private Statement wrapTestCase(final Statement base, final GenericContainer<?> container) {
+        return new Statement() {
+            public void evaluate() throws Throwable {
+                List<Throwable> errors = new ArrayList<>();
+
+                try {
+                    // Pre-Test
+                    container.start();
+                    base.evaluate();
+                } catch (Throwable preError) {
+                    errors.add(preError);
+                } finally {
+                    // Post-Test
+                    try {
+                        container.stop();
+                    } catch (Throwable postError) {
+                        errors.add(postError);
+                    }
+                }
+
+                MultipleFailureException.assertEmpty(errors);
+            }
+        };
+    }
+}

src/test/java/TestSuiteAcceptanceTests.java

@@ -0,0 +1,15 @@
+import org.junit.ClassRule;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses({
+        HttpEventCollector_JavaLoggingTest.class,
+        HttpEventCollector_Log4j2Test.class,
+        HttpEventCollector_LogbackTest.class,
+        HttpEventCollector_Test.class
+})
+public class TestSuiteAcceptanceTests {
+    @ClassRule
+    public static final TestRuleSplunkContainer TEST_CONTAINER = new TestRuleSplunkContainer();
+}

src/test/java/TestSuiteStressTests.java

@@ -0,0 +1,12 @@
+import org.junit.ClassRule;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses({
+        HttpLoggerStressTest.class
+})
+public class TestSuiteStressTests {
+    @ClassRule
+    public static final TestRuleSplunkContainer TEST_CONTAINER = new TestRuleSplunkContainer();
+}

src/test/resources/logback.xml

@@ -57,13 +57,11 @@ under the License.
         <token>11111111-2222-3333-4444-555555555555</token>
         <source>splunktest</source>
         <sourcetype>battlecat</sourcetype>
-        <messageFormat>text</messageFormat>
+        <messageFormat>json</messageFormat>
         <middleware>HttpEventCollectorUnitTestMiddleware</middleware>
         <connectTimeout>5000</connectTimeout>
         <terminationTimeout>2000</terminationTimeout>
-        <layout class="ch.qos.logback.classic.PatternLayout">
-            <pattern>%msg</pattern>
-        </layout>
+        <encoder class="net.logstash.logback.encoder.LogstashEncoder" />
     </appender>
 
     <logger name="splunk.logger" additivity="false" level="INFO">