You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2018/10/19 14:18:20 UTC
[GitHub] mkuthan closed pull request #5782: Support for request logging in
kafka emitter.
mkuthan closed pull request #5782: Support for request logging in kafka emitter.
URL: https://github.com/apache/incubator-druid/pull/5782
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/docs/content/development/extensions-contrib/kafka-emitter.md b/docs/content/development/extensions-contrib/kafka-emitter.md
index fb1371db5c7..2e5b974cc6a 100644
--- a/docs/content/development/extensions-contrib/kafka-emitter.md
+++ b/docs/content/development/extensions-contrib/kafka-emitter.md
@@ -22,6 +22,7 @@ All the configuration parameters for the Kafka emitter are under `druid.emitter.
|`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.request.topic`|Kafka topic name for emitter's target to emit request logs.|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|
@@ -31,5 +32,6 @@ All the configuration parameters for the Kafka emitter are under `druid.emitter.
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.request.topic=druid-request
druid.emitter.kafka.producer.config={"max.block.ms":10000}
```
diff --git a/extensions-contrib/kafka-emitter/pom.xml b/extensions-contrib/kafka-emitter/pom.xml
index d2aabf3fa67..62d04f056cd 100644
--- a/extensions-contrib/kafka-emitter/pom.xml
+++ b/extensions-contrib/kafka-emitter/pom.xml
@@ -52,6 +52,12 @@
<version>${project.parent.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>io.druid</groupId>
+ <artifactId>druid-server</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>io.druid</groupId>
<artifactId>java-util</artifactId>
diff --git a/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/ClusterNameMixin.java b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/ClusterNameMixin.java
new file mode 100644
index 00000000000..6893fdfbc16
--- /dev/null
+++ b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/ClusterNameMixin.java
@@ -0,0 +1,31 @@
+/*
+ * 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.databind.annotation.JsonAppend;
+
+@JsonAppend(
+ attrs = {
+ @JsonAppend.Attr(value = "clusterName")
+ }
+)
+public class ClusterNameMixin
+{
+}
diff --git a/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java
index 92c000a4d55..c71966bdb96 100644
--- a/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java
+++ b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java
@@ -21,6 +21,7 @@
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.collect.ImmutableMap;
import io.druid.emitter.kafka.MemoryBoundLinkedBlockingQueue.ObjectContainer;
import io.druid.java.util.common.StringUtils;
@@ -31,6 +32,7 @@
import io.druid.java.util.emitter.core.Event;
import io.druid.java.util.emitter.service.AlertEvent;
import io.druid.java.util.emitter.service.ServiceMetricEvent;
+import io.druid.server.log.EmittingRequestLogger;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
@@ -38,7 +40,6 @@
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
-import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -52,6 +53,7 @@
private static final int DEFAULT_RETRIES = 3;
private final AtomicLong metricLost;
private final AtomicLong alertLost;
+ private final AtomicLong requestLost;
private final AtomicLong invalidLost;
private final KafkaEmitterConfig config;
@@ -60,12 +62,16 @@
private final ObjectMapper jsonMapper;
private final MemoryBoundLinkedBlockingQueue<String> metricQueue;
private final MemoryBoundLinkedBlockingQueue<String> alertQueue;
+ private final MemoryBoundLinkedBlockingQueue<String> requestQueue;
private final ScheduledExecutorService scheduler;
public KafkaEmitter(KafkaEmitterConfig config, ObjectMapper jsonMapper)
{
this.config = config;
+
this.jsonMapper = jsonMapper;
+ this.jsonMapper.addMixIn(EmittingRequestLogger.RequestLogEvent.class, ClusterNameMixin.class);
+
this.producer = setKafkaProducer();
this.producerCallback = setProducerCallback();
// same with kafka producer's buffer.memory
@@ -73,9 +79,11 @@ public KafkaEmitter(KafkaEmitterConfig config, ObjectMapper jsonMapper)
.getOrDefault(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432"));
this.metricQueue = new MemoryBoundLinkedBlockingQueue<>(queueMemoryBound);
this.alertQueue = new MemoryBoundLinkedBlockingQueue<>(queueMemoryBound);
- this.scheduler = Executors.newScheduledThreadPool(3);
+ this.requestQueue = new MemoryBoundLinkedBlockingQueue<>(queueMemoryBound);
+ this.scheduler = Executors.newScheduledThreadPool(4);
this.metricLost = new AtomicLong(0L);
this.alertLost = new AtomicLong(0L);
+ this.requestLost = new AtomicLong(0L);
this.invalidLost = new AtomicLong(0L);
}
@@ -88,6 +96,8 @@ private Callback setProducerCallback()
metricLost.incrementAndGet();
} else if (recordMetadata.topic().equals(config.getAlertTopic())) {
alertLost.incrementAndGet();
+ } else if (recordMetadata.topic().equals(config.getRequestTopic())) {
+ requestLost.incrementAndGet();
} else {
invalidLost.incrementAndGet();
}
@@ -121,9 +131,10 @@ public void start()
{
scheduler.scheduleWithFixedDelay(this::sendMetricToKafka, 10, 10, TimeUnit.SECONDS);
scheduler.scheduleWithFixedDelay(this::sendAlertToKafka, 10, 10, TimeUnit.SECONDS);
+ scheduler.scheduleWithFixedDelay(this::sendRequestToKafka, 10, 10, TimeUnit.SECONDS);
scheduler.scheduleWithFixedDelay(() -> {
- log.info("Message lost counter: metricLost=[%d], alertLost=[%d], invalidLost=[%d]",
- metricLost.get(), alertLost.get(), invalidLost.get());
+ log.info("Message lost counter: metricLost=[%d], alertLost=[%d], requestLost=[%d], invalidLost=[%d]",
+ metricLost.get(), alertLost.get(), requestLost.get(), invalidLost.get());
}, 5, 5, TimeUnit.MINUTES);
log.info("Starting Kafka Emitter.");
}
@@ -138,6 +149,11 @@ private void sendAlertToKafka()
sendToKafka(config.getAlertTopic(), alertQueue);
}
+ private void sendRequestToKafka()
+ {
+ sendToKafka(config.getRequestTopic(), requestQueue);
+ }
+
private void sendToKafka(final String topic, MemoryBoundLinkedBlockingQueue<String> recordQueue)
{
ObjectContainer<String> objectToSend;
@@ -156,18 +172,33 @@ private void sendToKafka(final String topic, MemoryBoundLinkedBlockingQueue<Stri
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);
+ String resultJson;
+
+ if (event instanceof EmittingRequestLogger.RequestLogEvent) {
+ ObjectWriter objectWriter = jsonMapper.writerFor(EmittingRequestLogger.RequestLogEvent.class);
+
+ if (config.getClusterName() != null) {
+ objectWriter.withAttribute("clusterName", config.getClusterName());
+ }
+
+ resultJson = objectWriter.writeValueAsString(event);
+ } else {
+ ImmutableMap.Builder<String, Object> resultBuilder = ImmutableMap.builder();
+ resultBuilder.putAll(event.toMap());
+
+ if (config.getClusterName() != null) {
+ resultBuilder.put("clusterName", config.getClusterName());
+ }
+
+ resultJson = jsonMapper.writeValueAsString(resultBuilder.build());
+ }
+
ObjectContainer<String> objectContainer = new ObjectContainer<>(
resultJson,
StringUtils.toUtf8(resultJson).length
);
+
if (event instanceof ServiceMetricEvent) {
if (!metricQueue.offer(objectContainer)) {
metricLost.incrementAndGet();
@@ -176,6 +207,10 @@ public void emit(final Event event)
if (!alertQueue.offer(objectContainer)) {
alertLost.incrementAndGet();
}
+ } else if (event instanceof EmittingRequestLogger.RequestLogEvent) {
+ if (!requestQueue.offer(objectContainer)) {
+ requestLost.incrementAndGet();
+ }
} else {
invalidLost.incrementAndGet();
}
diff --git a/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java
index 24074a4638d..76d9dc3f521 100644
--- a/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java
+++ b/extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java
@@ -37,6 +37,8 @@
private final String metricTopic;
@JsonProperty("alert.topic")
private final String alertTopic;
+ @JsonProperty("request.topic")
+ private final String requestTopic;
@JsonProperty
private final String clusterName;
@JsonProperty("producer.config")
@@ -47,6 +49,7 @@ public KafkaEmitterConfig(
@JsonProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG) String bootstrapServers,
@JsonProperty("metric.topic") String metricTopic,
@JsonProperty("alert.topic") String alertTopic,
+ @JsonProperty("request.topic") String requestTopic,
@JsonProperty("clusterName") String clusterName,
@JsonProperty("producer.config") @Nullable Map<String, String> kafkaProducerConfig
)
@@ -54,6 +57,7 @@ public KafkaEmitterConfig(
this.bootstrapServers = Preconditions.checkNotNull(bootstrapServers, "bootstrap.servers can not be null");
this.metricTopic = Preconditions.checkNotNull(metricTopic, "metric.topic can not be null");
this.alertTopic = Preconditions.checkNotNull(alertTopic, "alert.topic can not be null");
+ this.requestTopic = Preconditions.checkNotNull(requestTopic, "request.topic can not be null");
this.clusterName = clusterName;
this.kafkaProducerConfig = kafkaProducerConfig == null ? ImmutableMap.of() : kafkaProducerConfig;
}
@@ -76,6 +80,12 @@ public String getAlertTopic()
return alertTopic;
}
+ @JsonProperty
+ public String getRequestTopic()
+ {
+ return requestTopic;
+ }
+
@JsonProperty
public String getClusterName()
{
@@ -109,6 +119,9 @@ public boolean equals(Object o)
if (!getAlertTopic().equals(that.getAlertTopic())) {
return false;
}
+ if (!getRequestTopic().equals(that.getRequestTopic())) {
+ return false;
+ }
if (getClusterName() != null ? !getClusterName().equals(that.getClusterName()) : that.getClusterName() != null) {
return false;
}
@@ -121,6 +134,7 @@ public int hashCode()
int result = getBootstrapServers().hashCode();
result = 31 * result + getMetricTopic().hashCode();
result = 31 * result + getAlertTopic().hashCode();
+ result = 31 * result + getRequestTopic().hashCode();
result = 31 * result + (getClusterName() != null ? getClusterName().hashCode() : 0);
result = 31 * result + getKafkaProducerConfig().hashCode();
return result;
@@ -133,6 +147,7 @@ public String toString()
"bootstrap.servers='" + bootstrapServers + '\'' +
", metric.topic='" + metricTopic + '\'' +
", alert.topic='" + alertTopic + '\'' +
+ ", request.topic='" + requestTopic + '\'' +
", clusterName='" + clusterName + '\'' +
", Producer.config=" + kafkaProducerConfig +
'}';
diff --git a/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java b/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java
index 7526de28759..8882c8d4a1c 100644
--- a/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java
+++ b/extensions-contrib/kafka-emitter/src/test/java/io/druid/emitter/kafka/KafkaEmitterConfigTest.java
@@ -43,7 +43,7 @@ public void setUp()
public void testSerDeserKafkaEmitterConfig() throws IOException
{
KafkaEmitterConfig kafkaEmitterConfig = new KafkaEmitterConfig("hostname", "metricTest",
- "alertTest", "clusterNameTest",
+ "alertTest", "requestTest", "clusterNameTest",
ImmutableMap.<String, String>builder()
.put("testKey", "testValue").build()
);
@@ -57,7 +57,7 @@ public void testSerDeserKafkaEmitterConfig() throws IOException
public void testSerDeNotRequiredKafkaProducerConfig()
{
KafkaEmitterConfig kafkaEmitterConfig = new KafkaEmitterConfig("localhost:9092", "metricTest",
- "alertTest", "clusterNameTest",
+ "alertTest", "requestTest", "clusterNameTest",
null
);
try {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org