diff --git a/broker/pom.xml b/broker/pom.xml index c7ababd3..28fd99bd 100755 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -115,23 +115,6 @@ --> - - 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 64ee53f0..faf95dc0 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -60,10 +60,6 @@ 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"; 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 8e161d62..f9b0e7fa 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -139,14 +139,7 @@ public class NettyAcceptor implements ServerAcceptor { final NettyMQTTHandler mqttHandler = new NettyMQTTHandler(processor); - 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(); - } + this.metrics = Optional.empty(); this.errorsCather = Optional.empty(); diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java index d93c0e17..554c71e9 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java @@ -16,6 +16,7 @@ package io.moquette.server.netty.metrics; +import io.moquette.server.netty.NettyUtils; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; import io.netty.channel.ChannelDuplexHandler; @@ -23,10 +24,15 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; import io.netty.util.Attribute; import io.netty.util.AttributeKey; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static io.moquette.server.netty.NettyUtils.ATTR_USERNAME; public class BytesMetricsHandler extends ChannelDuplexHandler { - + private static final Logger LOG = LoggerFactory.getLogger(BytesMetricsHandler.class); private static final AttributeKey ATTR_KEY_METRICS = AttributeKey.valueOf("BytesMetrics"); + private static final AttributeKey ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME); private BytesMetricsCollector m_collector; @@ -59,6 +65,12 @@ public class BytesMetricsHandler extends ChannelDuplexHandler { @Override public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { BytesMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get(); + String userId = ctx.channel().attr(ATTR_KEY_USERNAME).get(); + if (userId == null) { + userId = ""; + } + + LOG.info("channel<{}> closing after read {} bytes and wrote {} bytes", userId, metrics.readBytes(), metrics.wroteBytes()); m_collector.sumReadBytes(metrics.readBytes()); m_collector.sumWroteBytes(metrics.wroteBytes()); super.close(ctx, promise); 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 deleted file mode 100644 index 0a846f0a..00000000 --- a/broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * 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.Counter; -import com.codahale.metrics.Meter; -import com.codahale.metrics.MetricRegistry; -import com.librato.metrics.reporter.Librato; -import io.moquette.server.config.IConfig; -import io.moquette.server.netty.NettyUtils; -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.*; -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; - private Meter publishesMetrics; - private Meter subscribeMetrics; - private Counter connectedClientsMetrics; - - public void init(IConfig props) { - this.metrics = new MetricRegistry(); - this.publishesMetrics = metrics.meter("publish.requests"); - this.subscribeMetrics = metrics.meter("subscribe.requests"); - this.connectedClientsMetrics = metrics.counter("connect.num_clients"); -// 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: - this.publishesMetrics.mark(); - break; - case SUBSCRIBE: - this.subscribeMetrics.mark(); - break; - case CONNECT: - this.connectedClientsMetrics.inc(); - break; - case DISCONNECT: - this.connectedClientsMetrics.dec(); - break; - default: - break; - } - ctx.fireChannelRead(message); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { - String clientID = NettyUtils.clientID(ctx.channel()); - if (clientID != null && !clientID.isEmpty()) { - this.connectedClientsMetrics.dec(); - } - ctx.fireChannelInactive(); - } - -} diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java index c4894d12..c90632cf 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java @@ -22,10 +22,16 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; import io.netty.util.Attribute; import io.netty.util.AttributeKey; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static io.moquette.server.netty.NettyUtils.ATTR_USERNAME; public class MessageMetricsHandler extends ChannelDuplexHandler { - + private static final Logger LOG = LoggerFactory.getLogger(MessageMetricsHandler.class); private static final AttributeKey ATTR_KEY_METRICS = AttributeKey.valueOf("MessageMetrics"); + private static final AttributeKey ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME); + private MessageMetricsCollector m_collector; @@ -58,6 +64,12 @@ public class MessageMetricsHandler extends ChannelDuplexHandler { @Override public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { MessageMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get(); + String userId = ctx.channel().attr(ATTR_KEY_USERNAME).get(); + if (userId == null) { + userId = ""; + } + + LOG.info("channel<{}> closing after read {} messages and wrote {} messages", userId, metrics.messagesRead(), metrics.messagesWrote()); m_collector.sumReadMessages(metrics.messagesRead()); m_collector.sumWroteMessages(metrics.messagesWrote()); super.close(ctx, promise);