IGNITE-6868 Implement new JMX metrics for TcpCommunicationSpi monitoring HotFix
Signed-off-by: Anton Vinogradov <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/8da4fc95 Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/8da4fc95 Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/8da4fc95 Branch: refs/heads/ignite-zk Commit: 8da4fc95a2dc6d3c89a799795d6cc15f3536cce7 Parents: 80801cb Author: Aleksey Plekhanov <[email protected]> Authored: Fri Dec 8 17:17:06 2017 +0300 Committer: Anton Vinogradov <[email protected]> Committed: Mon Dec 11 14:09:32 2017 +0300 ---------------------------------------------------------------------- .../tcp/TcpCommunicationMetricsListener.java | 19 +++++++------- .../communication/tcp/TcpCommunicationSpi.java | 18 +++++++------- .../tcp/TcpCommunicationSpiMBean.java | 5 ++-- .../tcp/TcpCommunicationStatisticsTest.java | 26 ++++++-------------- 4 files changed, 29 insertions(+), 39 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/8da4fc95/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java index 8981e17..f9e35a5 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java @@ -56,13 +56,13 @@ public class TcpCommunicationMetricsListener implements GridNioMetricsListener{ private final ConcurrentMap<String, LongAdder8> rcvdMsgsCntByType = new ConcurrentHashMap<>(); /** Received messages count grouped by sender. */ - private final ConcurrentMap<String, LongAdder8> rcvdMsgsCntByNode = new ConcurrentHashMap<>(); + private final ConcurrentMap<UUID, LongAdder8> rcvdMsgsCntByNode = new ConcurrentHashMap<>(); /** Sent messages count grouped by message type. */ private final ConcurrentMap<String, LongAdder8> sentMsgsCntByType = new ConcurrentHashMap<>(); /** Sent messages count grouped by receiver. */ - private final ConcurrentMap<String, LongAdder8> sentMsgsCntByNode = new ConcurrentHashMap<>(); + private final ConcurrentMap<UUID, LongAdder8> sentMsgsCntByNode = new ConcurrentHashMap<>(); /** {@inheritDoc} */ @Override public void onBytesSent(int bytesCnt) { @@ -90,7 +90,7 @@ public class TcpCommunicationMetricsListener implements GridNioMetricsListener{ msg = ((GridIoMessage)msg).message(); LongAdder8 cntByType = F.addIfAbsent(sentMsgsCntByType, msg.getClass().getSimpleName(), LONG_ADDER_FACTORY); - LongAdder8 cntByNode = F.addIfAbsent(sentMsgsCntByNode, nodeId.toString(), LONG_ADDER_FACTORY); + LongAdder8 cntByNode = F.addIfAbsent(sentMsgsCntByNode, nodeId, LONG_ADDER_FACTORY); cntByType.increment(); cntByNode.increment(); @@ -112,7 +112,7 @@ public class TcpCommunicationMetricsListener implements GridNioMetricsListener{ msg = ((GridIoMessage)msg).message(); LongAdder8 cntByType = F.addIfAbsent(rcvdMsgsCntByType, msg.getClass().getSimpleName(), LONG_ADDER_FACTORY); - LongAdder8 cntByNode = F.addIfAbsent(rcvdMsgsCntByNode, nodeId.toString(), LONG_ADDER_FACTORY); + LongAdder8 cntByNode = F.addIfAbsent(rcvdMsgsCntByNode, nodeId, LONG_ADDER_FACTORY); cntByType.increment(); cntByNode.increment(); @@ -160,12 +160,11 @@ public class TcpCommunicationMetricsListener implements GridNioMetricsListener{ * @param srcStat Internal statistics representation. * @return Result map. */ - private Map<String, Long> convertStatistics(Map<String, LongAdder8> srcStat) { - Map<String, Long> destStat = U.newHashMap(srcStat.size()); + private <T> Map<T, Long> convertStatistics(Map<T, LongAdder8> srcStat) { + Map<T, Long> destStat = U.newHashMap(srcStat.size()); - for (Map.Entry<String, LongAdder8> entry : srcStat.entrySet()) { + for (Map.Entry<T, LongAdder8> entry : srcStat.entrySet()) destStat.put(entry.getKey(), entry.getValue().longValue()); - } return destStat; } @@ -184,7 +183,7 @@ public class TcpCommunicationMetricsListener implements GridNioMetricsListener{ * * @return Map containing sender nodes and respective counts. */ - public Map<String, Long> receivedMessagesByNode() { + public Map<UUID, Long> receivedMessagesByNode() { return convertStatistics(rcvdMsgsCntByNode); } @@ -202,7 +201,7 @@ public class TcpCommunicationMetricsListener implements GridNioMetricsListener{ * * @return Map containing receiver nodes and respective counts. */ - public Map<String, Long> sentMessagesByNode() { + public Map<UUID, Long> sentMessagesByNode() { return convertStatistics(sentMsgsCntByNode); } http://git-wip-us.apache.org/repos/asf/ignite/blob/8da4fc95/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java index e68797e..9290f24 100755 --- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java @@ -1831,7 +1831,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati * * @return Map containing sender nodes and respective counts. */ - public Map<String, Long> getReceivedMessagesByNode() { + public Map<UUID, Long> getReceivedMessagesByNode() { return metricsLsnr.receivedMessagesByNode(); } @@ -1849,8 +1849,8 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati * * @return Map containing receiver nodes and respective counts. */ - public Map<String, Long> getSentMessagesByNode() { - return metricsLsnr.receivedMessagesByNode(); + public Map<UUID, Long> getSentMessagesByNode() { + return metricsLsnr.sentMessagesByNode(); } /** {@inheritDoc} */ @@ -5156,22 +5156,22 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati /** {@inheritDoc} */ @Override public Map<String, Long> getReceivedMessagesByType() { - return TcpCommunicationSpi.this.metricsLsnr.receivedMessagesByType(); + return TcpCommunicationSpi.this.getReceivedMessagesByType(); } /** {@inheritDoc} */ - @Override public Map<String, Long> getReceivedMessagesByNode() { - return TcpCommunicationSpi.this.metricsLsnr.receivedMessagesByNode(); + @Override public Map<UUID, Long> getReceivedMessagesByNode() { + return TcpCommunicationSpi.this.getReceivedMessagesByNode(); } /** {@inheritDoc} */ @Override public Map<String, Long> getSentMessagesByType() { - return TcpCommunicationSpi.this.metricsLsnr.sentMessagesByType(); + return TcpCommunicationSpi.this.getSentMessagesByType(); } /** {@inheritDoc} */ - @Override public Map<String, Long> getSentMessagesByNode() { - return TcpCommunicationSpi.this.metricsLsnr.sentMessagesByNode(); + @Override public Map<UUID, Long> getSentMessagesByNode() { + return TcpCommunicationSpi.this.getSentMessagesByNode(); } /** {@inheritDoc} */ http://git-wip-us.apache.org/repos/asf/ignite/blob/8da4fc95/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiMBean.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiMBean.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiMBean.java index f4aba01..ca144c0 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiMBean.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpiMBean.java @@ -18,6 +18,7 @@ package org.apache.ignite.spi.communication.tcp; import java.util.Map; +import java.util.UUID; import org.apache.ignite.mxbean.MXBeanDescription; import org.apache.ignite.spi.IgniteSpiManagementMBean; @@ -161,7 +162,7 @@ public interface TcpCommunicationSpiMBean extends IgniteSpiManagementMBean { * @return Map containing sender nodes and respective counts. */ @MXBeanDescription("Received messages count grouped by sender node.") - public Map<String, Long> getReceivedMessagesByNode(); + public Map<UUID, Long> getReceivedMessagesByNode(); /** * Gets sent messages counts (grouped by type). @@ -177,7 +178,7 @@ public interface TcpCommunicationSpiMBean extends IgniteSpiManagementMBean { * @return Map containing receiver nodes and respective counts. */ @MXBeanDescription("Sent messages count grouped by receiver node.") - public Map<String, Long> getSentMessagesByNode(); + public Map<UUID, Long> getSentMessagesByNode(); /** * Gets outbound messages queue size. http://git-wip-us.apache.org/repos/asf/ignite/blob/8da4fc95/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationStatisticsTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationStatisticsTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationStatisticsTest.java index f0a8d71..e06a4bf 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationStatisticsTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationStatisticsTest.java @@ -128,16 +128,6 @@ public class TcpCommunicationStatisticsTest extends GridCommonAbstractTest { } /** - * Compares two maps for equality. - */ - private static <K, V> boolean mapsEquals(Map<K, V> map1, Map<K, V> map2) { - assert map1 != null; - assert map2 != null; - - return map1.size() == map2.size() && map1.entrySet().containsAll(map2.entrySet()); - } - - /** * @throws Exception If failed. */ public void testStatistics() throws Exception { @@ -163,13 +153,13 @@ public class TcpCommunicationStatisticsTest extends GridCommonAbstractTest { TcpCommunicationSpiMBean mbean0 = mbean(0); TcpCommunicationSpiMBean mbean1 = mbean(1); - Map<String, Long> msgsSentByNode0 = mbean0.getSentMessagesByNode(); - Map<String, Long> msgsSentByNode1 = mbean1.getSentMessagesByNode(); - Map<String, Long> msgsReceivedByNode0 = mbean0.getReceivedMessagesByNode(); - Map<String, Long> msgsReceivedByNode1 = mbean1.getReceivedMessagesByNode(); + Map<UUID, Long> msgsSentByNode0 = mbean0.getSentMessagesByNode(); + Map<UUID, Long> msgsSentByNode1 = mbean1.getSentMessagesByNode(); + Map<UUID, Long> msgsReceivedByNode0 = mbean0.getReceivedMessagesByNode(); + Map<UUID, Long> msgsReceivedByNode1 = mbean1.getReceivedMessagesByNode(); - String nodeId0 = grid(0).localNode().id().toString(); - String nodeId1 = grid(1).localNode().id().toString(); + UUID nodeId0 = grid(0).localNode().id(); + UUID nodeId1 = grid(1).localNode().id(); assertEquals(msgsReceivedByNode0.get(nodeId1).longValue(), mbean0.getReceivedMessagesCount()); assertEquals(msgsReceivedByNode1.get(nodeId0).longValue(), mbean1.getReceivedMessagesCount()); @@ -185,10 +175,10 @@ public class TcpCommunicationStatisticsTest extends GridCommonAbstractTest { Map<String, Long> msgsReceivedByType1 = mbean1.getReceivedMessagesByType(); // Node0 sent exactly the same types and count of messages as node1 received. - assertTrue(mapsEquals(msgsSentByType0, msgsReceivedByType1)); + assertEquals(msgsSentByType0, msgsReceivedByType1); // Node1 sent exactly the same types and count of messages as node0 received. - assertTrue(mapsEquals(msgsSentByType1, msgsReceivedByType0)); + assertEquals(msgsSentByType1, msgsReceivedByType0); assertEquals(1, msgsSentByType0.get(GridTestMessage.class.getSimpleName()).longValue()); assertEquals(1, msgsReceivedByType1.get(GridTestMessage.class.getSimpleName()).longValue());
