anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add metrics for user channel

Browse Source
This commit is contained in:
heavyrain lee
2019-07-16 21:56:51 +08:00
parent 86e08f6151
commit b2cfc2b99b
6 changed files with 27 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/pom.xml
-17
View File
@@ -115,23 +115,6 @@
-->
</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
-4
View File
@@ -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";
broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
+1 -8
View File
@@ -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();
broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java
+13 -1
View File
@@ -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<BytesMetrics> ATTR_KEY_METRICS = AttributeKey.valueOf("BytesMetrics");
private static final AttributeKey<String> 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);
broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java
-95
View File
@@ -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();
}
}
broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java
+13 -1
View File
@@ -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<MessageMetrics> ATTR_KEY_METRICS = AttributeKey.valueOf("MessageMetrics");
private static final AttributeKey<String> 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);