Gelf Kafka sender implementation #173

Log events can now be shipped using Kafka. Appenders can configure a Kafka URL in the host field according to the scheme:

kafka://broker[:port]?[producer_properties]#[log-topic]

e.g. kafka://localhost#topic-log

Kafka uses internally logging and JMX components with logging so these logging categories should be excluded to prevent circular, recursive log calls.

Original pull request: #173.

Author: Rifat Döver

README.md

@@ -418,7 +418,168 @@ logback.xml Example:
     </root>
 </configuration>
 ```
+# Kafka transport for logstash-gelf 
 
+
+logstash-gelf can be used together with [Kafka](https://kafka.apache.org/) for shipping log events. 
+
+ The URI used as connection property is a java.net.URI. The minimal URI must contain at least a host and
+ the Fragment (Topic Name). 
+ The URL allows to specify one or more brokers but in this case you must define ports inside URL.
+
+    kafka://broker[:port]?[producer_properties]#[log-topic]
+
+Example:
+
+    kafka://localhost#topic-log
+    kafka://localhost:9092#topic-log
+    kafka://localhost:9092,localhost:9093,localhost:9094#topic-log
+    kafka://localhost?acks=all#topic-log
+    kafka://localhost:19092?acks=1&max.block.ms=1000&transaction.timeout.ms=1000&request.timeout.ms=1000#kafka-log-topic
+
+   * scheme    (fixed: Kafka, directly used to determine the to be used sender class)
+   * host      (variable: the host your Kafka broker runs on)
+   * port      (variable: the port your Kafka broker runs on)
+   * query     (variable: kafka producer config properties which is usually defined inside producer.properties file)
+   * fragment  (variable: the topic we send log messages on)
+   
+Sample log4J2 configuration:
+```xml
+<Configuration packages="biz.paluch.logging.gelf.log4j2">
+    <Appenders>
+        <Gelf name="gelf" host="kafka://localhost#kafka-log-topic" port="9093" version="1.1" extractStackTrace="true"
+              filterStackTrace="true" mdcProfiling="true" includeFullMdc="true" maximumMessageSize="8192"
+              originHost="%host{fqdn}" additionalFieldTypes="fieldName1=String,fieldName2=Double,fieldName3=Long">
+            <Field name="timestamp" pattern="%d{dd MMM yyyy HH:mm:ss,SSS}" />
+            <Field name="level" pattern="%level" />
+            <Field name="simpleClassName" pattern="%C{1}" />
+            <Field name="className" pattern="%C" />
+            <Field name="server" pattern="%host" />
+            <Field name="server.fqdn" pattern="%host{fqdn}" />
+
+
+            <!-- This is a static field -->
+            <Field name="fieldName2" literal="fieldValue2" />
+
+            <!-- This is a field using MDC -->
+            <Field name="mdcField2" mdc="mdcField2" />
+            <DynamicMdcFields regex="mdc.*" />
+            <DynamicMdcFields regex="(mdc|MDC)fields" />
+        </Gelf>
+        <Console name="Console" target="SYSTEM_OUT">
+            <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+        </Console>
+    </Appenders>
+    <Loggers>
+        <Root level="INFO">
+            <AppenderRef ref="gelf" />
+            <AppenderRef ref="Console"/>
+        </Root>
+    </Loggers>
+</Configuration>
+```
+Sample spring logback configuration:
+```xml
+<!DOCTYPE configuration>
+
+<configuration>
+    <include resource="org/springframework/boot/logging/logback/base.xml"/>
+    <contextName>test</contextName>
+    <jmxConfigurator/>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
+            <pattern>
+                %d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n
+            </pattern>
+        </encoder>
+    </appender>
+    <appender name="gelfUdp" class="biz.paluch.logging.gelf.logback.GelfLogbackAppender">
+        <host>udp:localhost</host>
+        <port>12201</port>
+        <version>1.1</version>
+        <facility>java-test</facility>
+        <extractStackTrace>true</extractStackTrace>
+        <filterStackTrace>true</filterStackTrace>
+        <mdcProfiling>false</mdcProfiling>
+        <timestampPattern>yyyy-MM-dd HH:mm:ss,SSS</timestampPattern>
+        <maximumMessageSize>8192</maximumMessageSize>
+        <!-- This are static fields -->
+        <additionalFields>fieldName1=fieldValue1,fieldName2=fieldValue2</additionalFields>
+        <!-- Optional: Specify field types -->
+        <additionalFieldTypes>fieldName1=String,fieldName2=Double,fieldName3=Long</additionalFieldTypes>
+
+        <!-- This are fields using MDC -->
+        <mdcFields>mdcField1,mdcField2</mdcFields>
+        <dynamicMdcFields>mdc.*,(mdc|MDC)fields</dynamicMdcFields>
+        <includeFullMdc>true</includeFullMdc>
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>ALL</level>
+        </filter>
+    </appender>
+
+    <appender name="gelfKafka" class="biz.paluch.logging.gelf.logback.GelfLogbackAppender">
+        <host>kafka://localhost:9092?acks=all&amp;max.block.ms=1000&amp;transaction.timeout.ms=1000&amp;request.timeout.ms=1000#kafka-log-topic</host>
+        <port></port>
+        <version>1.1</version>
+        <facility>java-test</facility>
+        <extractStackTrace>true</extractStackTrace>
+        <filterStackTrace>true</filterStackTrace>
+        <mdcProfiling>false</mdcProfiling>
+        <timestampPattern>yyyy-MM-dd HH:mm:ss,SSS</timestampPattern>
+        <maximumMessageSize>8192</maximumMessageSize>
+
+        <!-- This are static fields -->
+        <additionalFields>fieldName1=fieldValue1,fieldName2=fieldValue2</additionalFields>
+        <!-- Optional: Specify field types -->
+        <additionalFieldTypes>fieldName1=String,fieldName2=Double,fieldName3=Long</additionalFieldTypes>
+
+        <!-- This are fields using MDC -->
+        <mdcFields>mdcField1,mdcField2</mdcFields>
+        <dynamicMdcFields>mdc.*,(mdc|MDC)fields</dynamicMdcFields>
+        <includeFullMdc>true</includeFullMdc>
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>INFO</level>
+        </filter>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="gelfKafka" />
+    </root>
+    <logger name="org.apache.kafka" level="ALL" additivity="false">
+        <appender-ref ref="gelfUdp" />
+        <appender-ref ref="STDOUT" />
+    </logger>
+    <logger name="javax.management" level="ALL" additivity="false">
+        <appender-ref ref="gelfUdp" />
+        <appender-ref ref="STDOUT" />
+    </logger>
+</configuration>
+```
+**Limitations**
+
+Some configurations will be overridden or set by default:
+- acks (If you set it to 0 it will be set to 1 by default for log message acknowledgements) defaultValue = all
+- retries defaultValue=2
+- value.serializer ByteArraySerializer (will always override)
+- key.serializer ByteArraySerializer (will always override)
+
+**When using with SL4J/Logback/Spring:**
+
+When you are using logback/sl4j/spring you must not use kafka sender for loggers of `org.apache.kafka` and `javax.management` 
+packages this will create a cyclic dependency for [KafkaProducer](https://kafka.apache.org/20/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html) which is also using loggers from these package. 
+You can use other logstash-gelf appender such as UDP or TCP to log these logs to Graylog.
+
+You can set these loggers by changing appender-ref this way:
+```xml
+<logger name="org.apache.kafka" level="ALL" additivity="false">
+     <appender-ref ref="gelfUdp" />
+     <appender-ref ref="STDOUT" />
+</logger>
+<logger name="javax.management" level="ALL" additivity="false">
+     <appender-ref ref="gelfUdp" />
+     <appender-ref ref="STDOUT" />
+</logger>
+```
 Versions/Dependencies
 ---------------------
 This project is built against following dependencies/versions:
@@ -427,8 +588,9 @@ This project is built against following dependencies/versions:
 * log4j2 2.9.1
 * Java Util Logging JDK Version 1.6
 * logback 1.1.3
-* slf4j-api 1.7.13
+* slf4j-api 1.7.25
 * jedis 2.9.0 (includes commons-pool2 2.4.3)
+* kafka-clients 2.1.0
 
 License
 -------

pom.xml

@@ -63,7 +63,7 @@
         <jackson.version>2.9.8</jackson.version>
         <jedis.version>3.0.1</jedis.version>
         <jedis-commons-pool2.version>2.4.3</jedis-commons-pool2.version>
-        <slf4j-api.version>1.7.13</slf4j-api.version>
+        <slf4j-api.version>1.7.25</slf4j-api.version>
         <logback-classic.version>1.1.3</logback-classic.version>
         <org.jboss.logmanager.version>1.5.2.Final</org.jboss.logmanager.version>
         <arquillian.version>1.4.0.Final</arquillian.version>
@@ -85,6 +85,7 @@
         <maven-scm-provider-gitexe.version>1.9.5</maven-scm-provider-gitexe.version>
         <maven-resources-plugin.version>3.0.2</maven-resources-plugin.version>
         <maven.jacoco.version>0.7.9</maven.jacoco.version>
+        <kafka-clients.version>2.1.0</kafka-clients.version>
     </properties>
 
     <developers>
@@ -391,6 +392,13 @@
             <version>${jackson.version}</version>
             <scope>test</scope>
         </dependency>
+        <!-- kafka-junit dependency -->
+        <dependency>
+            <groupId>com.github.charithe</groupId>
+            <artifactId>kafka-junit</artifactId>
+            <version>4.1.3</version>
+            <scope>test</scope>
+        </dependency>
 
         <!-- Redis Client -->
         <dependency>
@@ -399,6 +407,13 @@
             <version>${jedis.version}</version>
             <optional>true</optional>
         </dependency>
+        <!-- Kafka Client -->
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>${kafka-clients.version}</version>
+            <optional>true</optional>
+        </dependency>
 
         <dependency>
             <groupId>org.apache.commons</groupId>

src/main/java/biz/paluch/logging/gelf/intern/GelfSenderFactory.java

@@ -1,16 +1,13 @@
 package biz.paluch.logging.gelf.intern;
 
+import biz.paluch.logging.gelf.intern.sender.DefaultGelfSenderProvider;
+import biz.paluch.logging.gelf.intern.sender.KafkaGelfSenderProvider;
+import biz.paluch.logging.gelf.intern.sender.RedisGelfSenderProvider;
+
 import java.io.IOException;
 import java.net.SocketException;
 import java.net.UnknownHostException;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.ServiceLoader;
-
-import biz.paluch.logging.gelf.intern.sender.DefaultGelfSenderProvider;
-import biz.paluch.logging.gelf.intern.sender.RedisGelfSenderProvider;
+import java.util.*;
 
 /**
  * Factory to create a {@link GelfSender} based on the host and protocol details. This factory uses Java's {@link ServiceLoader}
@@ -118,6 +115,7 @@ private static class SenderProviderHolder {
                 providerList.add(iter.next());
             }
             providerList.add(new RedisGelfSenderProvider());
+            providerList.add(new KafkaGelfSenderProvider());
             providerList.add(new DefaultGelfSenderProvider());
         }
 

src/main/java/biz/paluch/logging/gelf/intern/sender/KafkaContants.java

@@ -0,0 +1,9 @@
+package biz.paluch.logging.gelf.intern.sender;
+
+/**
+ * @author Rifat Döver
+ */
+public class KafkaContants {
+    public static final String KAFKA_SCHEME = "kafka";
+    public static final String KAFKA_LOG_TOPIC = "kafka.log.topic";
+}

src/main/java/biz/paluch/logging/gelf/intern/sender/KafkaGelfSender.java

@@ -0,0 +1,47 @@
+package biz.paluch.logging.gelf.intern.sender;
+
+import biz.paluch.logging.gelf.intern.ErrorReporter;
+import biz.paluch.logging.gelf.intern.GelfMessage;
+import biz.paluch.logging.gelf.intern.GelfSender;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+
+import java.util.concurrent.Future;
+
+/**
+ * Gelf sender for kafka. This sender uses Kafka.
+ *
+ * @author Rifat Döver
+ * @since 1.13
+ */
+public class KafkaGelfSender implements GelfSender {
+    private KafkaProducer<byte[], byte[]> kafkaProducer;
+    private String topicName;
+    private ErrorReporter errorReporter;
+
+    public KafkaGelfSender(KafkaProducer<byte[], byte[]> kafkaProducer, String topicName, ErrorReporter errorReporter) {
+        this.kafkaProducer = kafkaProducer;
+        this.topicName = topicName;
+        this.errorReporter = errorReporter;
+    }
+
+    @Override
+    public boolean sendMessage(GelfMessage message) {
+        ProducerRecord<byte[], byte[]> record = new ProducerRecord<byte[], byte[]>(topicName, message.toJson().getBytes());
+        boolean hasOfset;
+        try {
+            Future<RecordMetadata> metadata = kafkaProducer.send(record);
+            hasOfset = metadata.get().hasOffset();
+        } catch (Exception e) {
+            errorReporter.reportError("Error sending log to kafka", e);
+            return false;
+        }
+        return hasOfset;
+    }
+
+    @Override
+    public void close() {
+        kafkaProducer.close();
+    }
+}

src/main/java/biz/paluch/logging/gelf/intern/sender/KafkaGelfSenderProvider.java

@@ -0,0 +1,85 @@
+package biz.paluch.logging.gelf.intern.sender;
+
+import biz.paluch.logging.gelf.intern.GelfSender;
+import biz.paluch.logging.gelf.intern.GelfSenderConfiguration;
+import biz.paluch.logging.gelf.intern.GelfSenderProvider;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.logging.log4j.util.Strings;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * @author Rifat Döver
+ * @since 1.13
+ *
+ */
+public class KafkaGelfSenderProvider implements GelfSenderProvider {
+    private static final int BROKER_DEFAULT_PORT = 9092;
+
+    @Override
+    public boolean supports(String host) {
+        return host != null && host.startsWith(KafkaContants.KAFKA_SCHEME + ":");
+    }
+
+    @Override
+    public GelfSender create(GelfSenderConfiguration configuration) throws IOException {
+        URI uri = URI.create(configuration.getHost());
+        Map<String, String> options = QueryStringParser.parse(uri);
+
+        Properties props = new Properties();
+        for (String key : options.keySet()) {
+            props.setProperty(key, options.get(key));
+        }
+
+        String brokers = getBrokerServers(configuration);
+
+        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
+
+        String kafkaLogTopic = getTopic(uri);
+
+        // Default Configurations
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+
+        if (!props.containsKey(ProducerConfig.ACKS_CONFIG)) {
+            props.put(ProducerConfig.ACKS_CONFIG, "all");
+        } else {
+            String acks = props.getProperty(ProducerConfig.ACKS_CONFIG);
+            acks = acks.equalsIgnoreCase("0") ? "1" : acks;
+            props.put(ProducerConfig.ACKS_CONFIG, acks);
+        }
+        if (!props.containsKey(ProducerConfig.RETRIES_CONFIG))
+            props.put(ProducerConfig.RETRIES_CONFIG, 2);
+
+        KafkaProducer<byte[], byte[]> kafkaProducer = new KafkaProducer<byte[], byte[]>(props);
+        return new KafkaGelfSender(kafkaProducer, kafkaLogTopic, configuration.getErrorReporter());
+    }
+
+    private String getBrokerServers(GelfSenderConfiguration configuration) {
+        URI uri = URI.create(configuration.getHost());
+        String brokers;
+        if (uri.getHost() != null) {
+            brokers = uri.getHost();
+            int port = uri.getPort() > 0 ? uri.getPort()
+                    : configuration.getPort() > 0 ? configuration.getPort() : BROKER_DEFAULT_PORT;
+            brokers += ":" + port;
+
+        } else {
+            brokers = uri.getAuthority();
+        }
+        if (Strings.isEmpty(brokers))
+            throw new IllegalArgumentException("Kafka URI must specify bootstrap.servers.");
+        return brokers;
+    }
+
+    private String getTopic(URI uri) {
+        if (Strings.isEmpty(uri.getFragment()))
+            throw new IllegalArgumentException("Kafka URI must specify log topic as fragment.");
+        return uri.getFragment();
+    }
+}