Introduced DropwizardMetrics to push data on Librato

broker/pom.xml

@@ -83,6 +83,23 @@
              -->
         </dependency>
 
+        <dependency>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+            <version>3.2.2</version>
+        </dependency>
+        <dependency>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-jvm</artifactId>
+            <version>3.2.2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.librato.metrics</groupId>
+            <artifactId>metrics-librato</artifactId>
+            <version>5.1.0</version>
+        </dependency>
+
         <dependency>
             <groupId>commons-codec</groupId>
             <artifactId>commons-codec</artifactId>

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() {

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<? extends ChannelInboundHandler> 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);
+        final boolean useFineMetrics = Boolean.parseBoolean(props.getProperty(METRICS_ENABLE_PROPERTY_NAME, "false"));
-        initializeWebSocketTransport(handler, props);
+        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);
+            initializeSSLTCPTransport(mqttHandler, props, sslContext);
-            initializeWSSTransport(handler, 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);
             }
         });

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);
+    }
+
+}

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 [email protected]
+# metrics.librato.token sdfergheghliuhyr2283ehd9827398h
+# metrics.librato.source My Fantastic Service
+