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
+