diff --git a/docs/content/configuration/index.md b/docs/content/configuration/index.md index 9bd7b97ce49..66b8ca41f76 100644 --- a/docs/content/configuration/index.md +++ b/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 diff --git a/extensions-contrib/ambari-metrics-emitter/pom.xml b/extensions-contrib/ambari-metrics-emitter/pom.xml index 702eee4a7c9..bdf4b543cf2 100644 --- a/extensions-contrib/ambari-metrics-emitter/pom.xml +++ b/extensions-contrib/ambari-metrics-emitter/pom.xml @@ -49,7 +49,7 @@ com.metamx - emitter + java-util provided diff --git a/extensions-contrib/graphite-emitter/pom.xml b/extensions-contrib/graphite-emitter/pom.xml index d81450f7185..6e4b433305f 100644 --- a/extensions-contrib/graphite-emitter/pom.xml +++ b/extensions-contrib/graphite-emitter/pom.xml @@ -55,7 +55,7 @@ com.metamx - emitter + java-util provided diff --git a/extensions-contrib/kafka-eight-simpleConsumer/pom.xml b/extensions-contrib/kafka-eight-simpleConsumer/pom.xml index 018e2b30159..1f351b2f578 100644 --- a/extensions-contrib/kafka-eight-simpleConsumer/pom.xml +++ b/extensions-contrib/kafka-eight-simpleConsumer/pom.xml @@ -40,7 +40,7 @@ com.metamx - emitter + java-util provided diff --git a/extensions-contrib/kafka-emitter/pom.xml b/extensions-contrib/kafka-emitter/pom.xml index 01766f6c43c..24ce7f36333 100644 --- a/extensions-contrib/kafka-emitter/pom.xml +++ b/extensions-contrib/kafka-emitter/pom.xml @@ -54,7 +54,7 @@ com.metamx - emitter + java-util provided diff --git a/extensions-contrib/statsd-emitter/pom.xml b/extensions-contrib/statsd-emitter/pom.xml index 475fae574d6..48878211123 100644 --- a/extensions-contrib/statsd-emitter/pom.xml +++ b/extensions-contrib/statsd-emitter/pom.xml @@ -48,7 +48,7 @@ com.metamx - emitter + java-util provided diff --git a/extensions-core/hdfs-storage/pom.xml b/extensions-core/hdfs-storage/pom.xml index b066f205053..1a842ac76d4 100644 --- a/extensions-core/hdfs-storage/pom.xml +++ b/extensions-core/hdfs-storage/pom.xml @@ -137,7 +137,7 @@ com.metamx - emitter + java-util provided diff --git a/extensions-core/s3-extensions/pom.xml b/extensions-core/s3-extensions/pom.xml index 16e5974338b..2c89929620a 100644 --- a/extensions-core/s3-extensions/pom.xml +++ b/extensions-core/s3-extensions/pom.xml @@ -53,7 +53,7 @@ com.metamx - emitter + java-util provided diff --git a/extensions-core/simple-client-sslcontext/pom.xml b/extensions-core/simple-client-sslcontext/pom.xml index 4446b377723..96b33eaafc3 100644 --- a/extensions-core/simple-client-sslcontext/pom.xml +++ b/extensions-core/simple-client-sslcontext/pom.xml @@ -36,7 +36,7 @@ com.metamx - emitter + java-util provided diff --git a/java-util/src/main/java/io/druid/java/util/common/lifecycle/Lifecycle.java b/java-util/src/main/java/io/druid/java/util/common/lifecycle/Lifecycle.java index 74eef87c8af..05999be5456 100644 --- a/java-util/src/main/java/io/druid/java/util/common/lifecycle/Lifecycle.java +++ b/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 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); + } + } + } } diff --git a/pom.xml b/pom.xml index b9748ab36d3..4fb02b7fbed 100644 --- a/pom.xml +++ b/pom.xml @@ -145,18 +145,8 @@ com.metamx - emitter - 0.6.0 - - - com.metamx - http-client - 1.1.0 - - - com.metamx - server-metrics - 0.5.2 + java-util + 1.3.2 commons-codec diff --git a/processing/pom.xml b/processing/pom.xml index 8f6ce7c5283..54963de0fb8 100644 --- a/processing/pom.xml +++ b/processing/pom.xml @@ -51,11 +51,7 @@ com.metamx - emitter - - - com.metamx - server-metrics + java-util com.ning diff --git a/server/pom.xml b/server/pom.xml index 67e0268782a..7c9b7f7b8c9 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -47,11 +47,7 @@ com.metamx - http-client - - - com.metamx - server-metrics + java-util commons-cli diff --git a/server/src/main/java/io/druid/server/emitter/HttpEmitterConfig.java b/server/src/main/java/io/druid/server/emitter/HttpEmitterConfig.java deleted file mode 100644 index 55dc4f8a092..00000000000 --- a/server/src/main/java/io/druid/server/emitter/HttpEmitterConfig.java +++ /dev/null @@ -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; - } -} diff --git a/server/src/main/java/io/druid/server/emitter/HttpEmitterModule.java b/server/src/main/java/io/druid/server/emitter/HttpEmitterModule.java index 4dfe7166d57..7497c21f8f1 100644 --- a/server/src/main/java/io/druid/server/emitter/HttpEmitterModule.java +++ b/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 - .builder() - .withNumConnections(1) - .withReadTimeout(config.get().getReadTimeout().toStandardDuration()); - + final DefaultAsyncHttpClientConfig.Builder builder = new DefaultAsyncHttpClientConfig.Builder(); 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); } } diff --git a/server/src/main/java/io/druid/server/emitter/HttpEmittingMonitor.java b/server/src/main/java/io/druid/server/emitter/HttpEmittingMonitor.java new file mode 100644 index 00000000000..0bb85ef55ce --- /dev/null +++ b/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); + } +} diff --git a/server/src/main/java/io/druid/server/emitter/ParametrizedUriEmitterConfig.java b/server/src/main/java/io/druid/server/emitter/ParametrizedUriEmitterConfig.java deleted file mode 100644 index 665c6f2fb70..00000000000 --- a/server/src/main/java/io/druid/server/emitter/ParametrizedUriEmitterConfig.java +++ /dev/null @@ -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; - } -} diff --git a/server/src/main/java/io/druid/server/emitter/ParametrizedUriEmitterModule.java b/server/src/main/java/io/druid/server/emitter/ParametrizedUriEmitterModule.java index 7e9ea094614..6d8633c9f6c 100644 --- a/server/src/main/java/io/druid/server/emitter/ParametrizedUriEmitterModule.java +++ b/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.http.client.HttpClientInit; +import com.metamx.emitter.core.ParametrizedUriEmitterConfig; 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 - .builder() - .withNumConnections(1) - .withReadTimeout(config.get().getReadTimeout().toStandardDuration()); + final DefaultAsyncHttpClientConfig.Builder builder = new DefaultAsyncHttpClientConfig.Builder(); if (sslContext != null) { - builder.withSslContext(sslContext); + builder.setSslContext(new JdkSslContext(sslContext, true, ClientAuth.NONE)); } - return new ParametrizedUriEmitter( - config.get(), - HttpClientInit.createClient(builder.build(), LifecycleUtils.asMmxLifecycle(lifecycle)), - jsonMapper - ); + final AsyncHttpClient client = new DefaultAsyncHttpClient(builder.build()); + lifecycle.addCloseableInstance(client); + + return new ParametrizedUriEmitter(config.get(), client, jsonMapper); } } diff --git a/server/src/main/java/io/druid/server/metrics/MetricsModule.java b/server/src/main/java/io/druid/server/metrics/MetricsModule.java index 6d3c4f37404..1f24942c935 100644 --- a/server/src/main/java/io/druid/server/metrics/MetricsModule.java +++ b/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; diff --git a/server/src/test/java/io/druid/server/emitter/EmitterModuleTest.java b/server/src/test/java/io/druid/server/emitter/EmitterModuleTest.java index d693d989d95..52a6876a5b6 100644 --- a/server/src/test/java/io/druid/server/emitter/EmitterModuleTest.java +++ b/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 =