From 04478e77e1c674edbf01b62296e6e73b6839c3aa Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 20 Sep 2013 19:25:19 +0200 Subject: [PATCH] Added Message and Bytes read/write metrics --- .../moquette/server/netty/NettyAcceptor.java | 18 +++++-- .../server/netty/metrics/BytesMetrics.java | 22 ++++++++ .../netty/metrics/BytesMetricsCollector.java | 24 +++++++++ .../netty/metrics/BytesMetricsHandler.java | 50 +++++++++++++++++++ .../server/netty/metrics/MessageMetrics.java | 22 ++++++++ .../metrics/MessageMetricsCollector.java | 24 +++++++++ .../netty/metrics/MessageMetricsHandler.java | 50 +++++++++++++++++++ 7 files changed, 205 insertions(+), 5 deletions(-) create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetrics.java create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsCollector.java create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsHandler.java create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetrics.java create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsCollector.java create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsHandler.java diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java index dab9fc73..22c65a2c 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java @@ -9,8 +9,7 @@ import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; -import io.netty.handler.logging.LogLevel; -import io.netty.handler.logging.LoggingHandler; + import java.io.IOException; import io.netty.handler.timeout.IdleStateHandler; @@ -19,6 +18,7 @@ import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.parser.netty.MQTTDecoder; import org.dna.mqtt.moquette.parser.netty.MQTTEncoder; import org.dna.mqtt.moquette.server.ServerAcceptor; +import org.dna.mqtt.moquette.server.netty.metrics.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,7 +32,9 @@ public class NettyAcceptor implements ServerAcceptor { EventLoopGroup m_bossGroup; EventLoopGroup m_workerGroup; - + //BytesMetricsCollector m_metricsCollector = new BytesMetricsCollector(); + MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector(); + public void initialize(IMessaging messaging) throws IOException { m_bossGroup = new NioEventLoopGroup(); @@ -40,19 +42,21 @@ public class NettyAcceptor implements ServerAcceptor { final NettyMQTTHandler handler = new NettyMQTTHandler(); handler.setMessaging(messaging); - - ServerBootstrap b = new ServerBootstrap(); + + ServerBootstrap b = new ServerBootstrap(); b.group(m_bossGroup, m_workerGroup) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer() { @Override public void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); + //pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector)); pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); pipeline.addLast("decoder", new MQTTDecoder()); pipeline.addLast("encoder", new MQTTEncoder()); + pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); pipeline.addLast("handler", handler); } }) @@ -78,6 +82,10 @@ public class NettyAcceptor implements ServerAcceptor { } m_workerGroup.shutdownGracefully(); m_bossGroup.shutdownGracefully(); + + MessageMetrics metrics = m_metricsCollector.computeMetrics(); + //LOG.info(String.format("Bytes read: %d, bytes wrote: %d", metrics.readBytes(), metrics.wroteBytes())); + LOG.info(String.format("Msg read: %d, msg wrote: %d", metrics.messagesRead(), metrics.messagesWrote())); } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetrics.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetrics.java new file mode 100644 index 00000000..c18dc87c --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetrics.java @@ -0,0 +1,22 @@ +package org.dna.mqtt.moquette.server.netty.metrics; + +public class BytesMetrics { + private long m_readBytes = 0; + private long m_wroteBytes = 0; + + void incrementRead(long numBytes) { + m_readBytes += numBytes; + } + + void incrementWrote(long numBytes) { + m_wroteBytes += numBytes; + } + + public long readBytes() { + return m_readBytes; + } + + public long wroteBytes() { + return m_wroteBytes; + } +} \ No newline at end of file diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsCollector.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsCollector.java new file mode 100644 index 00000000..409d383e --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsCollector.java @@ -0,0 +1,24 @@ +package org.dna.mqtt.moquette.server.netty.metrics; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * Collects all the metrics from the various pipeline. + */ +public class BytesMetricsCollector { + private Queue m_allMetrics = new ConcurrentLinkedQueue(); + + void addMetrics(BytesMetrics metrics) { + m_allMetrics.add(metrics); + } + + public BytesMetrics computeMetrics() { + BytesMetrics allMetrics = new BytesMetrics(); + for (BytesMetrics m : m_allMetrics) { + allMetrics.incrementRead(m.readBytes()); + allMetrics.incrementWrote(m.wroteBytes()); + } + return allMetrics; + } +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsHandler.java new file mode 100644 index 00000000..5b4be068 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsHandler.java @@ -0,0 +1,50 @@ +package org.dna.mqtt.moquette.server.netty.metrics; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import io.netty.util.Attribute; +import io.netty.util.AttributeKey; + +public class BytesMetricsHandler extends ChannelDuplexHandler { + + private static final AttributeKey ATTR_KEY_METRICS = new AttributeKey("BytesMetrics"); + + private BytesMetricsCollector m_collector; + + public BytesMetricsHandler(BytesMetricsCollector collector) { + m_collector = collector; + } + + @Override + public void channelActive(ChannelHandlerContext ctx) throws Exception { + Attribute attr = ctx.attr(ATTR_KEY_METRICS); + attr.set(new BytesMetrics()); + + super.channelActive(ctx); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get(); + metrics.incrementRead(((ByteBuf)msg).readableBytes()); + ctx.fireChannelRead(msg); + } + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get(); + metrics.incrementWrote(((ByteBuf)msg).writableBytes()); + ctx.write(msg, promise); + } + + + @Override + public void close(ChannelHandlerContext ctx, + ChannelPromise promise) throws Exception { + BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get(); + m_collector.addMetrics(metrics); + super.close(ctx, promise); + } +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetrics.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetrics.java new file mode 100644 index 00000000..dda3a19e --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetrics.java @@ -0,0 +1,22 @@ +package org.dna.mqtt.moquette.server.netty.metrics; + +public class MessageMetrics { + private long m_messagesRead = 0; + private long m_messageWrote = 0; + + void incrementRead(long numMessages) { + m_messagesRead += numMessages; + } + + void incrementWrote(long numMessages) { + m_messageWrote += numMessages; + } + + public long messagesRead() { + return m_messagesRead; + } + + public long messagesWrote() { + return m_messageWrote; + } +} \ No newline at end of file diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsCollector.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsCollector.java new file mode 100644 index 00000000..044630de --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsCollector.java @@ -0,0 +1,24 @@ +package org.dna.mqtt.moquette.server.netty.metrics; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * Collects all the metrics from the various pipeline. + */ +public class MessageMetricsCollector { + private Queue m_allMetrics = new ConcurrentLinkedQueue(); + + void addMetrics(MessageMetrics metrics) { + m_allMetrics.add(metrics); + } + + public MessageMetrics computeMetrics() { + MessageMetrics allMetrics = new MessageMetrics(); + for (MessageMetrics m : m_allMetrics) { + allMetrics.incrementRead(m.messagesRead()); + allMetrics.incrementWrote(m.messagesWrote()); + } + return allMetrics; + } +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsHandler.java new file mode 100644 index 00000000..98d5044e --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsHandler.java @@ -0,0 +1,50 @@ +package org.dna.mqtt.moquette.server.netty.metrics; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import io.netty.util.Attribute; +import io.netty.util.AttributeKey; + +public class MessageMetricsHandler extends ChannelDuplexHandler { + + private static final AttributeKey ATTR_KEY_METRICS = new AttributeKey("MessageMetrics"); + + private MessageMetricsCollector m_collector; + + public MessageMetricsHandler(MessageMetricsCollector collector) { + m_collector = collector; + } + + @Override + public void channelActive(ChannelHandlerContext ctx) throws Exception { + Attribute attr = ctx.attr(ATTR_KEY_METRICS); + attr.set(new MessageMetrics()); + + super.channelActive(ctx); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get(); + metrics.incrementRead(1); + ctx.fireChannelRead(msg); + } + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get(); + metrics.incrementWrote(1); + ctx.write(msg, promise); + } + + + @Override + public void close(ChannelHandlerContext ctx, + ChannelPromise promise) throws Exception { + MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get(); + m_collector.addMetrics(metrics); + super.close(ctx, promise); + } +}