Adding Kafka-emitter

docs/content/development/extensions-contrib/kafka-emitter.md

@@ -0,0 +1,35 @@
+---
+layout: doc_page
+---
+
+# Kafka Emitter
+
+To use this extension, make sure to [include](../../operations/including-extensions.html) `kafka-emitter` extension.
+
+## Introduction
+
+This extension emits Druid metrics to a [Kafka](https://kafka.apache.org) directly with JSON format.<br>
+Currently, Kafka has not only their nice ecosystem but also consumer API readily available. 
+So, If you currently use Kafka, It's easy to integrate various tool or UI 
+to monitor the status of your Druid cluster with this extension.
+
+## Configuration
+
+All the configuration parameters for the Kafka emitter are under `druid.emitter.kafka`.
+
+|property|description|required?|default|
+|--------|-----------|---------|-------|
+|`druid.emitter.kafka.bootstrap.servers`|Comma-separated Kafka broker. (`[hostname:port],[hostname:port]...`)|yes|none|
+|`druid.emitter.kafka.metric.topic`|Kafka topic name for emitter's target to emit service metric.|yes|none|
+|`druid.emitter.kafka.alert.topic`|Kafka topic name for emitter's target to emit alert.|yes|none|
+|`druid.emitter.kafka.producer.config`|JSON formatted configuration which user want to set additional properties to Kafka producer.|no|none|
+|`druid.emitter.kafka.clusterName`|Optional value to specify name of your druid cluster. It can help make groups in your monitoring environment. |no|none|
+
+### Example
+
+```
+druid.emitter.kafka.bootstrap.servers=hostname1:9092,hostname2:9092
+druid.emitter.kafka.metric.topic=druid-metric
+druid.emitter.kafka.alert.topic=druid-alert
+druid.emitter.kafka.producer.config={"max.block.ms":10000}
+```

docs/content/development/extensions.md

@@ -65,6 +65,7 @@ All of these community extensions can be downloaded using *pull-deps* with the c
 |sqlserver-metadata-storage|Microsoft SqlServer deep storage.|[link](../development/extensions-contrib/sqlserver.html)|
 |graphite-emitter|Graphite metrics emitter|[link](../development/extensions-contrib/graphite.html)|
 |statsd-emitter|StatsD metrics emitter|[link](../development/extensions-contrib/statsd.html)|
+|kafka-emitter|Kafka metrics emitter|[link](../development/extensions-contrib/kafka-emitter.html)|
 |druid-thrift-extensions|Support thrift ingestion |[link](../development/extensions-contrib/thrift.html)|
 |scan-query|Scan query|[link](../development/extensions-contrib/scan-query.html)|
 

extensions-contrib/kafka-emitter/pom.xml

@@ -0,0 +1,91 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. Metamarkets licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+  -->
+
+<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 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>io.druid</groupId>
+    <artifactId>druid</artifactId>
+    <version>0.10.0-SNAPSHOT</version>
+    <relativePath>../../pom.xml</relativePath>
+  </parent>
+
+  <groupId>io.druid.extensions.contrib</groupId>
+  <artifactId>kafka-emitter</artifactId>
+  <name>kafka-emitter</name>
+  <description>Druid emitter extension to support kafka</description>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>0.10.2.0</version>
+    </dependency>
+    <dependency>
+      <groupId>io.druid</groupId>
+      <artifactId>druid-common</artifactId>
+      <version>${project.parent.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.druid</groupId>
+      <artifactId>druid-api</artifactId>
+      <version>${project.parent.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.metamx</groupId>
+      <artifactId>emitter</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.easymock</groupId>
+      <artifactId>easymock</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>pl.pragmatists</groupId>
+      <artifactId>JUnitParams</artifactId>
+      <version>1.0.4</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.druid</groupId>
+      <artifactId>druid-server</artifactId>
+      <version>${project.parent.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.druid</groupId>
+      <artifactId>druid-processing</artifactId>
+      <version>${project.parent.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>

extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java

@@ -0,0 +1,199 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.emitter.kafka;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.metamx.emitter.core.Emitter;
+import com.metamx.emitter.core.Event;
+import com.metamx.emitter.service.AlertEvent;
+import com.metamx.emitter.service.ServiceMetricEvent;
+import io.druid.emitter.kafka.MemoryBoundLinkedBlockingQueue.ObjectContainer;
+import io.druid.java.util.common.lifecycle.LifecycleStart;
+import io.druid.java.util.common.lifecycle.LifecycleStop;
+import io.druid.java.util.common.logger.Logger;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class KafkaEmitter implements Emitter
+{
+  private static Logger log = new Logger(KafkaEmitter.class);
+
+  private final static int DEFAULT_RETRIES = 3;
+  private final AtomicLong metricLost;
+  private final AtomicLong alertLost;
+  private final AtomicLong invalidLost;
+
+  private final KafkaEmitterConfig config;
+  private final Producer<String, String> producer;
+  private final Callback producerCallback;
+  private final ObjectMapper jsonMapper;
+  private final MemoryBoundLinkedBlockingQueue<String> metricQueue;
+  private final MemoryBoundLinkedBlockingQueue<String> alertQueue;
+  private final ScheduledExecutorService scheduler;
+
+  public KafkaEmitter(KafkaEmitterConfig config, ObjectMapper jsonMapper)
+  {
+    this.config = config;
+    this.jsonMapper = jsonMapper;
+    this.producer = setKafkaProducer();
+    this.producerCallback = setProducerCallback();
+    // same with kafka producer's buffer.memory
+    long queueMemoryBound = Long.parseLong(this.config.getKafkaProducerConfig()
+                                                      .getOrDefault(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432"));
+    this.metricQueue = new MemoryBoundLinkedBlockingQueue<>(queueMemoryBound);
+    this.alertQueue = new MemoryBoundLinkedBlockingQueue<>(queueMemoryBound);
+    this.scheduler = Executors.newScheduledThreadPool(3);
+    this.metricLost = new AtomicLong(0L);
+    this.alertLost = new AtomicLong(0L);
+    this.invalidLost = new AtomicLong(0L);
+  }
+
+  private Callback setProducerCallback()
+  {
+    return (recordMetadata, e) -> {
+      if (e != null) {
+        log.debug("Event send failed [%s]", e.getMessage());
+        if (recordMetadata.topic().equals(config.getMetricTopic())) {
+          metricLost.incrementAndGet();
+        } else if (recordMetadata.topic().equals(config.getAlertTopic())) {
+          alertLost.incrementAndGet();
+        } else {
+          invalidLost.incrementAndGet();
+        }
+      }
+    };
+  }
+
+  private Producer<String, String> setKafkaProducer()
+  {
+    ClassLoader currCtxCl = Thread.currentThread().getContextClassLoader();
+    try {
+      Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
+
+      Properties props = new Properties();
+      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers());
+      props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+      props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+      props.put(ProducerConfig.RETRIES_CONFIG, DEFAULT_RETRIES);
+      props.putAll(config.getKafkaProducerConfig());
+
+      return new KafkaProducer<>(props);
+    }
+    finally {
+      Thread.currentThread().setContextClassLoader(currCtxCl);
+    }
+  }
+
+  @Override
+  @LifecycleStart
+  public void start()
+  {
+    scheduler.scheduleWithFixedDelay(this::sendMetricToKafka, 10, 10, TimeUnit.SECONDS);
+    scheduler.scheduleWithFixedDelay(this::sendAlertToKafka, 10, 10, TimeUnit.SECONDS);
+    scheduler.scheduleWithFixedDelay(() -> {
+      log.info("Message lost counter: metricLost=[%d], alertLost=[%d], invalidLost=[%d]",
+               metricLost.get(), alertLost.get(), invalidLost.get());
+    }, 5, 5, TimeUnit.MINUTES);
+    log.info("Starting Kafka Emitter.");
+  }
+
+  private void sendMetricToKafka()
+  {
+    sendToKafka(config.getMetricTopic(), metricQueue);
+  }
+
+  private void sendAlertToKafka()
+  {
+    sendToKafka(config.getAlertTopic(), alertQueue);
+  }
+
+  private void sendToKafka(final String topic, MemoryBoundLinkedBlockingQueue<String> recordQueue)
+  {
+    ObjectContainer<String> objectToSend;
+    try {
+      while (true) {
+        objectToSend = recordQueue.take();
+        producer.send(new ProducerRecord<>(topic, objectToSend.getData()), producerCallback);
+      }
+    }
+    catch (InterruptedException e) {
+      log.warn(e, "Failed to take record from queue!");
+    }
+  }
+
+  @Override
+  public void emit(final Event event)
+  {
+    if (event != null) {
+      ImmutableMap.Builder<String, Object> resultBuilder = ImmutableMap.<String, Object>builder().putAll(event.toMap());
+      if (config.getClusterName() != null) {
+        resultBuilder.put("clusterName", config.getClusterName());
+      }
+      Map<String, Object> result = resultBuilder.build();
+
+      try {
+        String resultJson = jsonMapper.writeValueAsString(result);
+        ObjectContainer<String> objectContainer = new ObjectContainer<>(resultJson, resultJson.getBytes().length);
+        if (event instanceof ServiceMetricEvent) {
+          if (!metricQueue.offer(objectContainer)) {
+            metricLost.incrementAndGet();
+          }
+        } else if (event instanceof AlertEvent) {
+          if (!alertQueue.offer(objectContainer)) {
+            alertLost.incrementAndGet();
+          }
+        } else {
+          invalidLost.incrementAndGet();
+        }
+      }
+      catch (JsonProcessingException e) {
+        invalidLost.incrementAndGet();
+      }
+    }
+  }
+
+  @Override
+  public void flush() throws IOException
+  {
+    producer.flush();
+  }
+
+  @Override
+  @LifecycleStop
+  public void close() throws IOException
+  {
+    scheduler.shutdownNow();
+    producer.close();
+  }
+}