Add Emitter monitoring (#4973)

* Add Emitter monitoring

* Fix typo

* Fixes

* testing new emitter

* Fix failed test (#71)

* testing new emitter

* fix on failed test

* Remove emitter's readTimeout from docs

* Update docs

* Add HttpEmittingMonitor

* Update java-util to 1.3.2

docs/content/configuration/index.md

@@ -196,6 +196,7 @@ The following monitors are available:
 |`io.druid.segment.realtime.RealtimeMetricsMonitor`|Reports statistics on Realtime nodes.|
 |`io.druid.server.metrics.EventReceiverFirehoseMonitor`|Reports how many events have been queued in the EventReceiverFirehose.|
 |`io.druid.server.metrics.QueryCountStatsMonitor`|Reports how many queries have been successful/failed/interrupted.|
+|`io.druid.server.emitter.HttpEmitterMonitor`|Reports internal metrics of `http` or `parametrized` emitter (see below). Must not be used with another emitter type. See the description of the metrics here: https://github.com/druid-io/druid/pull/4973.|
 
 ### Emitting Metrics
 
@@ -216,26 +217,26 @@ The Druid servers [emit various metrics](../operations/metrics.html) and alerts
 
 |Property|Description|Default|
 |--------|-----------|-------|
-|`druid.emitter.http.readTimeout`|The timeout for data reads.|PT5M|
 |`druid.emitter.http.flushMillis`|How often the internal message buffer is flushed (data is sent).|60000|
 |`druid.emitter.http.flushCount`|How many messages the internal message buffer can hold before flushing (sending).|500|
 |`druid.emitter.http.basicAuthentication`|Login and password for authentification in "login:password" form, e. g. `druid.emitter.http.basicAuthentication=admin:adminpassword`|not specified = no authentification|
 |`druid.emitter.http.flushTimeOut|The timeout after which an event should be sent to the endpoint, even if internal buffers are not filled, in milliseconds.|not specified = no timeout|
 |`druid.emitter.http.batchingStrategy`|The strategy of how the batch is formatted. "ARRAY" means `[event1,event2]`, "NEWLINES" means `event1\nevent2`, ONLY_EVENTS means `event1event2`.|ARRAY|
 |`druid.emitter.http.maxBatchSize`|The maximum batch size, in bytes.|5191680 (i. e. 5 MB)|
+|`druid.emitter.http.batchQueueSizeLimit`|The maximum number of batches in emitter queue, if there are problems with emitting.|50|
+|`druid.emitter.http.minHttpTimeoutMillis`|If the speed of filling batches imposes timeout smaller than that, not even trying to send batch to endpoint, because it will likely fail, not being able to send the data that fast. Configure this depending based on emitter/successfulSending/minTimeMs metric. Reasonable values are 10ms..100ms.|0|
 |`druid.emitter.http.recipientBaseUrl`|The base URL to emit messages to. Druid will POST JSON to be consumed at the HTTP endpoint specified by this property.|none, required config|
 
 #### Parametrized Http Emitter Module
 
 `druid.emitter.parametrized.httpEmitting.*` configs correspond to the configs of Http Emitter Modules, see above.
-Except `readTimeout` and `recipientBaseUrl`. E. g. `druid.emitter.parametrized.httpEmitting.flushMillis`,
+Except `recipientBaseUrl`. E. g. `druid.emitter.parametrized.httpEmitting.flushMillis`,
 `druid.emitter.parametrized.httpEmitting.flushCount`, etc.
 
 The additional configs are:
 
 |Property|Description|Default|
 |--------|-----------|-------|
-|`druid.emitter.parametrized.readTimeout`|The timeout for data reads.|PT5M|
 |`druid.emitter.parametrized.recipientBaseUrlPattern`|The URL pattern to send an event to, based on the event's feed. E. g. `http://foo.bar/{feed}`, that will send event to `http://foo.bar/metrics` if the event's feed is "metrics".|none, required config|
 
 #### Composing Emitter Module

extensions-contrib/ambari-metrics-emitter/pom.xml

@@ -49,7 +49,7 @@
     </dependency>
     <dependency>
       <groupId>com.metamx</groupId>
-      <artifactId>emitter</artifactId>
+      <artifactId>java-util</artifactId>
       <scope>provided</scope>
     </dependency>
     <dependency>

extensions-contrib/graphite-emitter/pom.xml

@@ -55,7 +55,7 @@
     </dependency>
     <dependency>
       <groupId>com.metamx</groupId>
-      <artifactId>emitter</artifactId>
+      <artifactId>java-util</artifactId>
       <scope>provided</scope>
     </dependency>
     <dependency>

extensions-contrib/kafka-eight-simpleConsumer/pom.xml

@@ -40,7 +40,7 @@
     </dependency>
     <dependency>
       <groupId>com.metamx</groupId>
-      <artifactId>emitter</artifactId>
+      <artifactId>java-util</artifactId>
       <scope>provided</scope>
     </dependency>
     <dependency>

extensions-contrib/kafka-emitter/pom.xml

@@ -54,7 +54,7 @@
     </dependency>
     <dependency>
       <groupId>com.metamx</groupId>
-      <artifactId>emitter</artifactId>
+      <artifactId>java-util</artifactId>
       <scope>provided</scope>
     </dependency>
     <dependency>

extensions-contrib/statsd-emitter/pom.xml

@@ -48,7 +48,7 @@
     </dependency>
     <dependency>
       <groupId>com.metamx</groupId>
-      <artifactId>emitter</artifactId>
+      <artifactId>java-util</artifactId>
       <scope>provided</scope>
     </dependency>
     <dependency>

extensions-core/hdfs-storage/pom.xml

@@ -137,7 +137,7 @@
         </dependency>
         <dependency>
             <groupId>com.metamx</groupId>
-            <artifactId>emitter</artifactId>
+            <artifactId>java-util</artifactId>
             <scope>provided</scope>
         </dependency>
         <dependency>

extensions-core/s3-extensions/pom.xml

@@ -53,7 +53,7 @@
         </dependency>
         <dependency>
             <groupId>com.metamx</groupId>
-            <artifactId>emitter</artifactId>
+            <artifactId>java-util</artifactId>
             <scope>provided</scope>
         </dependency>
         <dependency>

extensions-core/simple-client-sslcontext/pom.xml

@@ -36,7 +36,7 @@
     </dependency>
     <dependency>
       <groupId>com.metamx</groupId>
-      <artifactId>emitter</artifactId>
+      <artifactId>java-util</artifactId>
       <scope>provided</scope>
     </dependency>
   </dependencies>

java-util/src/main/java/io/druid/java/util/common/lifecycle/Lifecycle.java

@@ -23,6 +23,7 @@ import com.google.common.collect.Lists;
 import io.druid.java.util.common.ISE;
 import io.druid.java.util.common.logger.Logger;
 
+import java.io.Closeable;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
 import java.util.List;
@@ -238,6 +239,16 @@ public class Lifecycle
     return o;
   }
 
+  /**
+   * Adds a Closeable instance to the lifecycle at {@link Stage#NORMAL} stage, doesn't try to call any "start" method on
+   * it, use {@link #addStartCloseInstance(Object)} instead if you need the latter behaviour.
+   */
+  public <T extends Closeable> T addCloseableInstance(T o)
+  {
+    addHandler(new CloseableHandler(o));
+    return o;
+  }
+
   /**
    * Adds a handler to the Lifecycle at the Stage.NORMAL stage and starts it if the lifecycle has already been started.
    *
@@ -472,4 +483,33 @@ public class Lifecycle
       }
     }
   }
+
+  private static class CloseableHandler implements Handler
+  {
+    private static final Logger log = new Logger(CloseableHandler.class);
+    private final Closeable o;
+
+    private CloseableHandler(Closeable o)
+    {
+      this.o = o;
+    }
+
+    @Override
+    public void start() throws Exception
+    {
+      // do nothing
+    }
+
+    @Override
+    public void stop()
+    {
+      log.info("Closing object[%s]", o);
+      try {
+        o.close();
+      }
+      catch (Exception e) {
+        log.error(e, "Exception when closing object [%s]", o);
+      }
+    }
+  }
 }

pom.xml

@@ -145,18 +145,8 @@
             <!-- Compile Scope -->
             <dependency>
                 <groupId>com.metamx</groupId>
-                <artifactId>emitter</artifactId>
+                <artifactId>java-util</artifactId>
-                <version>0.6.0</version>
+                <version>1.3.2</version>
-            </dependency>
-            <dependency>
-                <groupId>com.metamx</groupId>
-                <artifactId>http-client</artifactId>
-                <version>1.1.0</version>
-            </dependency>
-            <dependency>
-                <groupId>com.metamx</groupId>
-                <artifactId>server-metrics</artifactId>
-                <version>0.5.2</version>
             </dependency>
             <dependency>
                 <groupId>commons-codec</groupId>

processing/pom.xml

@@ -51,11 +51,7 @@
         </dependency>
         <dependency>
             <groupId>com.metamx</groupId>
-            <artifactId>emitter</artifactId>
+            <artifactId>java-util</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.metamx</groupId>
-            <artifactId>server-metrics</artifactId>
         </dependency>
         <dependency>
             <groupId>com.ning</groupId>

server/pom.xml

@@ -47,11 +47,7 @@
         </dependency>
         <dependency>
             <groupId>com.metamx</groupId>
-            <artifactId>http-client</artifactId>
+            <artifactId>java-util</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.metamx</groupId>
-            <artifactId>server-metrics</artifactId>
         </dependency>
         <dependency>
             <groupId>commons-cli</groupId>

server/src/main/java/io/druid/server/emitter/HttpEmitterConfig.java

@@ -1,36 +0,0 @@
-/*
- * 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.server.emitter;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import org.joda.time.Period;
-
-/**
- */
-public class HttpEmitterConfig extends com.metamx.emitter.core.HttpEmitterConfig
-{
-  @JsonProperty
-  private Period readTimeout = new Period("PT5M");
-
-  public Period getReadTimeout()
-  {
-    return readTimeout;
-  }
-}

server/src/main/java/io/druid/server/emitter/HttpEmitterModule.java

@@ -27,14 +27,17 @@ import com.google.inject.Provides;
 import com.google.inject.name.Named;
 import com.google.inject.util.Providers;
 import com.metamx.emitter.core.Emitter;
+import com.metamx.emitter.core.HttpEmitterConfig;
 import com.metamx.emitter.core.HttpPostEmitter;
-import com.metamx.http.client.HttpClientConfig;
-import com.metamx.http.client.HttpClientInit;
 import io.druid.guice.JsonConfigProvider;
 import io.druid.guice.LazySingleton;
 import io.druid.guice.ManageLifecycle;
-import io.druid.guice.http.LifecycleUtils;
 import io.druid.java.util.common.lifecycle.Lifecycle;
+import io.netty.handler.ssl.ClientAuth;
+import io.netty.handler.ssl.JdkSslContext;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.DefaultAsyncHttpClient;
+import org.asynchttpclient.DefaultAsyncHttpClientConfig;
 
 import javax.annotation.Nullable;
 import javax.net.ssl.SSLContext;
@@ -75,14 +78,13 @@ public class HttpEmitterModule implements Module
       ObjectMapper jsonMapper
   )
   {
-    final HttpClientConfig.Builder builder = HttpClientConfig
+    final DefaultAsyncHttpClientConfig.Builder builder = new DefaultAsyncHttpClientConfig.Builder();
-        .builder()
-        .withNumConnections(1)
-        .withReadTimeout(config.get().getReadTimeout().toStandardDuration());
-
     if (sslContext != null) {
-      builder.withSslContext(sslContext);
+      builder.setSslContext(new JdkSslContext(sslContext, true, ClientAuth.NONE));
     }
-    return new HttpPostEmitter(config.get(), HttpClientInit.createClient(builder.build(), LifecycleUtils.asMmxLifecycle(lifecycle)), jsonMapper);
+    final AsyncHttpClient client = new DefaultAsyncHttpClient(builder.build());
+    lifecycle.addCloseableInstance(client);
+
+    return new HttpPostEmitter(config.get(), client, jsonMapper);
   }
 }

server/src/main/java/io/druid/server/emitter/HttpEmittingMonitor.java

@@ -0,0 +1,67 @@
+/*
+ * 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.server.emitter;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import com.metamx.emitter.core.Emitter;
+import com.metamx.emitter.core.HttpPostEmitter;
+import com.metamx.emitter.core.ParametrizedUriEmitter;
+import com.metamx.emitter.service.ServiceEmitter;
+import com.metamx.metrics.AbstractMonitor;
+import com.metamx.metrics.FeedDefiningMonitor;
+import com.metamx.metrics.HttpPostEmitterMonitor;
+import com.metamx.metrics.ParametrizedUriEmitterMonitor;
+
+/**
+ * Able to monitor {@link HttpPostEmitter} or {@link ParametrizedUriEmitter}, which is based on the former.
+ */
+public class HttpEmittingMonitor extends AbstractMonitor
+{
+  private AbstractMonitor delegate;
+
+  @Inject
+  public HttpEmittingMonitor(Emitter emitter)
+  {
+    if (emitter instanceof HttpPostEmitter) {
+      delegate = new HttpPostEmitterMonitor(
+          FeedDefiningMonitor.DEFAULT_METRICS_FEED,
+          (HttpPostEmitter) emitter,
+          ImmutableMap.of()
+      );
+    } else if (emitter instanceof ParametrizedUriEmitter) {
+      delegate = new ParametrizedUriEmitterMonitor(
+          FeedDefiningMonitor.DEFAULT_METRICS_FEED,
+          (ParametrizedUriEmitter) emitter
+      );
+    } else {
+      throw new IllegalStateException(
+          "Unable to use HttpEmittingMonitor with emitter other than HttpPostEmitter or ParametrizedUriEmitter, " +
+          emitter.getClass() + " is used"
+      );
+    }
+  }
+
+  @Override
+  public boolean doMonitor(ServiceEmitter serviceEmitter)
+  {
+    return delegate.doMonitor(serviceEmitter);
+  }
+}

server/src/main/java/io/druid/server/emitter/ParametrizedUriEmitterConfig.java

@@ -1,34 +0,0 @@
-/*
- * 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.server.emitter;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import org.joda.time.Period;
-
-public class ParametrizedUriEmitterConfig extends com.metamx.emitter.core.ParametrizedUriEmitterConfig
-{
-  @JsonProperty
-  private Period readTimeout = new Period("PT5M");
-
-  public Period getReadTimeout()
-  {
-    return readTimeout;
-  }
-}

server/src/main/java/io/druid/server/emitter/ParametrizedUriEmitterModule.java

@@ -27,12 +27,15 @@ import com.google.inject.Provides;
 import com.google.inject.name.Named;
 import com.metamx.emitter.core.Emitter;
 import com.metamx.emitter.core.ParametrizedUriEmitter;
-import com.metamx.http.client.HttpClientConfig;
+import com.metamx.emitter.core.ParametrizedUriEmitterConfig;
-import com.metamx.http.client.HttpClientInit;
 import io.druid.guice.JsonConfigProvider;
 import io.druid.guice.ManageLifecycle;
-import io.druid.guice.http.LifecycleUtils;
 import io.druid.java.util.common.lifecycle.Lifecycle;
+import io.netty.handler.ssl.ClientAuth;
+import io.netty.handler.ssl.JdkSslContext;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.DefaultAsyncHttpClient;
+import org.asynchttpclient.DefaultAsyncHttpClientConfig;
 
 import javax.annotation.Nullable;
 import javax.net.ssl.SSLContext;
@@ -56,17 +59,13 @@ public class ParametrizedUriEmitterModule implements Module
       ObjectMapper jsonMapper
   )
   {
-    final HttpClientConfig.Builder builder = HttpClientConfig
+    final DefaultAsyncHttpClientConfig.Builder builder = new DefaultAsyncHttpClientConfig.Builder();
-        .builder()
-        .withNumConnections(1)
-        .withReadTimeout(config.get().getReadTimeout().toStandardDuration());
     if (sslContext != null) {
-      builder.withSslContext(sslContext);
+      builder.setSslContext(new JdkSslContext(sslContext, true, ClientAuth.NONE));
     }
-    return new ParametrizedUriEmitter(
+    final AsyncHttpClient client = new DefaultAsyncHttpClient(builder.build());
-        config.get(),
+    lifecycle.addCloseableInstance(client);
-        HttpClientInit.createClient(builder.build(), LifecycleUtils.asMmxLifecycle(lifecycle)),
+
-        jsonMapper
+    return new ParametrizedUriEmitter(config.get(), client, jsonMapper);
-    );
   }
 }

server/src/main/java/io/druid/server/metrics/MetricsModule.java

@@ -34,11 +34,11 @@ import com.metamx.metrics.JvmMonitor;
 import com.metamx.metrics.Monitor;
 import com.metamx.metrics.MonitorScheduler;
 import com.metamx.metrics.SysMonitor;
-import io.druid.java.util.common.concurrent.Execs;
 import io.druid.guice.DruidBinders;
 import io.druid.guice.JsonConfigProvider;
 import io.druid.guice.LazySingleton;
 import io.druid.guice.ManageLifecycle;
+import io.druid.java.util.common.concurrent.Execs;
 import io.druid.java.util.common.logger.Logger;
 import io.druid.query.ExecutorServiceMonitor;
 

server/src/test/java/io/druid/server/emitter/EmitterModuleTest.java

@@ -52,7 +52,6 @@ public class EmitterModuleTest
     props.setProperty("druid.emitter.parametrized.httpEmitting.basicAuthentication", "a:b");
     props.setProperty("druid.emitter.parametrized.httpEmitting.batchingStrategy", "NEWLINES");
     props.setProperty("druid.emitter.parametrized.httpEmitting.maxBatchSize", "4");
-    props.setProperty("druid.emitter.parametrized.httpEmitting.maxBufferSize", "8");
     props.setProperty("druid.emitter.parametrized.httpEmitting.flushTimeOut", "1000");
 
     final Emitter emitter =