From e0104d99f911413f05cf92a8dffe3110823bbc29 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 18 Jun 2017 11:04:24 +0200 Subject: [PATCH] Introduced DropwizardMetrics to push data on Librato --- broker/pom.xml | 17 +++++ .../java/io/moquette/BrokerConstants.java | 4 + .../moquette/server/netty/NettyAcceptor.java | 31 ++++++-- .../metrics/DropWizardMetricsHandler.java | 73 +++++++++++++++++++ distribution/src/main/resources/moquette.conf | 18 ++++- 5 files changed, 133 insertions(+), 10 deletions(-) create mode 100644 broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java diff --git a/broker/pom.xml b/broker/pom.xml index 322f8283..5adad2b1 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -83,6 +83,23 @@ --> + + io.dropwizard.metrics + metrics-core + 3.2.2 + + + io.dropwizard.metrics + metrics-jvm + 3.2.2 + + + + com.librato.metrics + metrics-librato + 5.1.0 + + commons-codec commons-codec diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 220dd4a8..c29c0149 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -57,6 +57,10 @@ public final class BrokerConstants { public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive"; public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds"; public static final String NETTY_EPOLL_PROPERTY_NAME = "netty.epoll"; + public static final String METRICS_ENABLE_PROPERTY_NAME = "use_metrics"; + public static final String METRICS_LIBRATO_EMAIL_PROPERTY_NAME = "metrics.librato.email"; + public static final String METRICS_LIBRATO_TOKEN_PROPERTY_NAME = "metrics.librato.token"; + public static final String METRICS_LIBRATO_SOURCE_PROPERTY_NAME = "metrics.librato.source"; public static final String STORAGE_CLASS_NAME = "storage_class"; private BrokerConstants() { diff --git a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java index 71bf48dc..4da62609 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -50,6 +50,7 @@ import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; import java.io.IOException; import java.util.List; +import java.util.Optional; import java.util.concurrent.TimeUnit; import static io.moquette.BrokerConstants.*; @@ -95,6 +96,7 @@ public class NettyAcceptor implements ServerAcceptor { EventLoopGroup m_workerGroup; BytesMetricsCollector m_bytesMetricsCollector = new BytesMetricsCollector(); MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector(); + private Optional metrics; private int nettySoBacklog; private boolean nettySoReuseaddr; @@ -121,21 +123,30 @@ public class NettyAcceptor implements ServerAcceptor { boolean epoll = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_EPOLL_PROPERTY_NAME, "false")); if (epoll) { - LOG.info("Netty is using epoll."); + LOG.info("Netty is using Epoll"); m_bossGroup = new EpollEventLoopGroup(); m_workerGroup = new EpollEventLoopGroup(); channelClass = EpollServerSocketChannel.class; } else { - LOG.info("Netty is using nio."); + LOG.info("Netty is using NIO"); m_bossGroup = new NioEventLoopGroup(); m_workerGroup = new NioEventLoopGroup(); channelClass = NioServerSocketChannel.class; } - final NettyMQTTHandler handler = new NettyMQTTHandler(processor); + final NettyMQTTHandler mqttHandler = new NettyMQTTHandler(processor); - initializePlainTCPTransport(handler, props); - initializeWebSocketTransport(handler, props); + final boolean useFineMetrics = Boolean.parseBoolean(props.getProperty(METRICS_ENABLE_PROPERTY_NAME, "false")); + if (useFineMetrics) { + DropWizardMetricsHandler metricsHandler = new DropWizardMetricsHandler(); + metricsHandler.init(props); + this.metrics = Optional.of(metricsHandler); + } else { + this.metrics = Optional.empty(); + } + + initializePlainTCPTransport(mqttHandler, props); + initializeWebSocketTransport(mqttHandler, props); String sslTcpPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME); String wssPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME); if (sslTcpPortProp != null || wssPortProp != null) { @@ -144,8 +155,8 @@ public class NettyAcceptor implements ServerAcceptor { LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks"); return; } - initializeSSLTCPTransport(handler, props, sslContext); - initializeWSSTransport(handler, props, sslContext); + initializeSSLTCPTransport(mqttHandler, props, sslContext); + initializeWSSTransport(mqttHandler, props, sslContext); } } @@ -179,7 +190,8 @@ public class NettyAcceptor implements ServerAcceptor { } } - private void initializePlainTCPTransport(final NettyMQTTHandler handler, IConfig props) throws IOException { + private void initializePlainTCPTransport(final NettyMQTTHandler handler, + IConfig props) throws IOException { LOG.info("Configuring TCP MQTT transport"); final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler(); String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME); @@ -202,6 +214,9 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addLast("encoder", MqttEncoder.INSTANCE); pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); pipeline.addLast("messageLogger", new MQTTMessageLogger()); + if (metrics.isPresent()) { + pipeline.addLast("wizardMetrics", metrics.get()); + } pipeline.addLast("handler", handler); } }); diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java new file mode 100644 index 00000000..42626b6d --- /dev/null +++ b/broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java @@ -0,0 +1,73 @@ +/* + * Copyright (c) 2012-2017 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package io.moquette.server.netty.metrics; + +import com.codahale.metrics.Meter; +import com.codahale.metrics.MetricRegistry; +import com.librato.metrics.reporter.Librato; +import io.moquette.server.config.IConfig; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.handler.codec.mqtt.MqttMessage; +import io.netty.handler.codec.mqtt.MqttMessageType; + +import java.util.concurrent.TimeUnit; + +import static io.moquette.BrokerConstants.METRICS_LIBRATO_EMAIL_PROPERTY_NAME; +import static io.moquette.BrokerConstants.METRICS_LIBRATO_TOKEN_PROPERTY_NAME; +import static io.moquette.BrokerConstants.METRICS_LIBRATO_SOURCE_PROPERTY_NAME; +import static io.netty.channel.ChannelHandler.Sharable; + +/** + * Pipeline handler use to track some MQTT metrics. + */ +@Sharable +public final class DropWizardMetricsHandler extends ChannelInboundHandlerAdapter { + private MetricRegistry metrics = new MetricRegistry(); + private Meter publishesMetrics = metrics.meter("publish.requests"); + + public void init(IConfig props) { + this.metrics = new MetricRegistry(); + this.publishesMetrics = metrics.meter("publish.requests"); +// ConsoleReporter reporter = ConsoleReporter.forRegistry(metrics) +// .convertRatesTo(TimeUnit.SECONDS) +// .convertDurationsTo(TimeUnit.MILLISECONDS) +// .build(); +// reporter.start(1, TimeUnit.MINUTES); + final String email = props.getProperty(METRICS_LIBRATO_EMAIL_PROPERTY_NAME); + final String token = props.getProperty(METRICS_LIBRATO_TOKEN_PROPERTY_NAME); + final String source = props.getProperty(METRICS_LIBRATO_SOURCE_PROPERTY_NAME); + + Librato.reporter(this.metrics, email, token) + .setSource(source) + .start(10, TimeUnit.SECONDS); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object message) { + MqttMessage msg = (MqttMessage) message; + MqttMessageType messageType = msg.fixedHeader().messageType(); + switch (messageType) { + case PUBLISH: + publishesMetrics.mark(); + break; + default: + break; + } + ctx.fireChannelRead(message); + } + +} diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf index c3fd7b9c..87bad321 100644 --- a/distribution/src/main/resources/moquette.conf +++ b/distribution/src/main/resources/moquette.conf @@ -103,10 +103,24 @@ password_file config/password_file.conf #********************************************************************* # Netty Configuration -#********************************************************************* # # Linux systems can use epoll instead of nio. To get a performance # gain and reduced GC. # http://netty.io/wiki/native-transports.html for more information +#********************************************************************* +# netty.epoll true + +#********************************************************************* +# Metrics Configuration # -# netty.epoll=true +# use_metrics: used to enable Dropwizard Metrics sampling metrics. +# metrics.librato.email: Librato account's email +# metrics.librato.token: Librato account's security token +# metrics.librato.source: your application as source in Librato +# definitions +#********************************************************************* +# use_metrics true +# metrics.librato.email john@doe.com +# metrics.librato.token sdfergheghliuhyr2283ehd9827398h +# metrics.librato.source My Fantastic Service +