You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2020/01/09 16:54:26 UTC
[GitHub] [ignite] nizhikov commented on a change in pull request #6814:
IGNITE-12108 TCP Communication Metrics ported to a new framework.
nizhikov commented on a change in pull request #6814: IGNITE-12108 TCP Communication Metrics ported to a new framework.
URL: https://github.com/apache/ignite/pull/6814#discussion_r364851667
##########
File path: modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java
##########
@@ -22,60 +22,145 @@
import java.util.Map;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.LongAdder;
-
+import java.util.function.Function;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
-import org.apache.ignite.internal.util.nio.GridNioMetricsListener;
-import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.processors.metric.GridMetricManager;
+import org.apache.ignite.internal.processors.metric.MetricRegistry;
+import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
+import org.apache.ignite.internal.processors.metric.impl.MetricUtils;
import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.metric.LongMetric;
+import org.apache.ignite.spi.metric.Metric;
+
+import static org.apache.ignite.internal.util.nio.GridNioServer.RECEIVED_BYTES_METRIC_DESC;
+import static org.apache.ignite.internal.util.nio.GridNioServer.RECEIVED_BYTES_METRIC_NAME;
+import static org.apache.ignite.internal.util.nio.GridNioServer.SENT_BYTES_METRIC_DESC;
+import static org.apache.ignite.internal.util.nio.GridNioServer.SENT_BYTES_METRIC_NAME;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.COMMUNICATION_METRICS_GROUP_NAME;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.RECEIVED_MESSAGES_BY_NODE_ID_METRIC_DESC;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.RECEIVED_MESSAGES_BY_NODE_ID_METRIC_NAME;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.RECEIVED_MESSAGES_BY_TYPE_METRIC_DESC;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.RECEIVED_MESSAGES_BY_TYPE_METRIC_NAME;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.RECEIVED_MESSAGES_METRIC_DESC;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.RECEIVED_MESSAGES_METRIC_NAME;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.SENT_MESSAGES_BY_NODE_ID_METRIC_DESC;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.SENT_MESSAGES_BY_NODE_ID_METRIC_NAME;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.SENT_MESSAGES_BY_TYPE_METRIC_DESC;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.SENT_MESSAGES_BY_TYPE_METRIC_NAME;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.SENT_MESSAGES_METRIC_DESC;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.SENT_MESSAGES_METRIC_NAME;
/**
* Statistics for {@link org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi}.
*/
-public class TcpCommunicationMetricsListener implements GridNioMetricsListener{
- /** Counter factory. */
- private static final Callable<LongHolder> HOLDER_FACTORY = new Callable<LongHolder>() {
- @Override public LongHolder call() {
- return new LongHolder();
- }
- };
+class TcpCommunicationMetricsListener {
+ /** Metrics manager. */
+ private final GridMetricManager mmgr;
- /** Received bytes count. */
- private final LongAdder rcvdBytesCnt = new LongAdder();
-
- /** Sent bytes count.*/
- private final LongAdder sentBytesCnt = new LongAdder();
+ /** Metrics registry. */
+ private final MetricRegistry mreg;
/** All registered metrics. */
private final Set<ThreadMetrics> allMetrics = Collections.newSetFromMap(new ConcurrentHashMap<>());
/** Thread-local metrics. */
- private final ThreadLocal<ThreadMetrics> threadMetrics = new ThreadLocal<ThreadMetrics>() {
- @Override protected ThreadMetrics initialValue() {
- ThreadMetrics metrics = new ThreadMetrics();
+ private final ThreadLocal<ThreadMetrics> threadMetrics = ThreadLocal.withInitial(() -> {
+ ThreadMetrics metrics = new ThreadMetrics();
- allMetrics.add(metrics);
+ allMetrics.add(metrics);
- return metrics;
- }
- };
+ return metrics;
+ });
+
+ /** Function to be used in {@link Map#computeIfAbsent(Object, Function)} of {@code sentMsgsMetricsByType}. */
+ private final Function<Short, LongAdderMetric> sentMsgsCntByTypeMetricFactory;
+
+ /** Function to be used in {@link Map#computeIfAbsent(Object, Function)} of {@code rcvdMsgsMetricsByType}. */
+ private final Function<Short, LongAdderMetric> rcvdMsgsCntByTypeMetricFactory;
+
+ /** Function to be used in {@link Map#computeIfAbsent(Object, Function)} of {@code sentMsgsMetricsByNodeId}. */
+ private final Function<UUID, LongAdderMetric> sentMsgsCntByNodeIdMetricFactory;
+
+ /** Function to be used in {@link Map#computeIfAbsent(Object, Function)} of {@code rcvdMsgsMetricsByNodeId}. */
+ private final Function<UUID, LongAdderMetric> rcvdMsgsCntByNodeIdMetricFactory;
+
+ /** Sent bytes count metric.*/
+ private final LongAdderMetric sentBytesMetric;
+
+ /** Received bytes count metric. */
+ private final LongAdderMetric rcvdBytesMetric;
+
+ /** Sent messages count metric. */
+ private final LongAdderMetric sentMsgsMetric;
+
+ /** Received messages count metric. */
+ private final LongAdderMetric rcvdMsgsMetric;
/** Method to synchronize access to message type map. */
private final Object msgTypMapMux = new Object();
/** Message type map. */
private volatile Map<Short, String> msgTypMap;
- /** {@inheritDoc} */
- @Override public void onBytesSent(int bytesCnt) {
- sentBytesCnt.add(bytesCnt);
+ /** Generate metric name by message direct type id. */
+ public static String sentMessagesByTypeMetricName(Short directType) {
+ return MetricUtils.metricName(SENT_MESSAGES_BY_TYPE_METRIC_NAME, directType.toString());
+ }
+
+ /** Generate metric name by message direct type id. */
+ public static String receivedMessagesByTypeMetricName(Short directType) {
+ return MetricUtils.metricName(RECEIVED_MESSAGES_BY_TYPE_METRIC_NAME, directType.toString());
+ }
+
+ /** */
+ public TcpCommunicationMetricsListener(GridMetricManager mmgr) {
+ this.mmgr = mmgr;
+
+ mreg = mmgr.registry(COMMUNICATION_METRICS_GROUP_NAME);
+
+ sentMsgsCntByTypeMetricFactory = directType -> mreg.longAdderMetric(
+ sentMessagesByTypeMetricName(directType),
+ SENT_MESSAGES_BY_TYPE_METRIC_DESC
+ );
+ rcvdMsgsCntByTypeMetricFactory = directType -> mreg.longAdderMetric(
+ receivedMessagesByTypeMetricName(directType),
+ RECEIVED_MESSAGES_BY_TYPE_METRIC_DESC
+ );
+
+ sentMsgsCntByNodeIdMetricFactory = nodeId -> getOrCreateMetricRegistry(mmgr, nodeId)
+ .findMetric(SENT_MESSAGES_BY_NODE_ID_METRIC_NAME);
+
+ rcvdMsgsCntByNodeIdMetricFactory = nodeId -> getOrCreateMetricRegistry(mmgr, nodeId)
+ .findMetric(RECEIVED_MESSAGES_BY_NODE_ID_METRIC_NAME);
+
+ sentBytesMetric = mreg.longAdderMetric(SENT_BYTES_METRIC_NAME, SENT_BYTES_METRIC_DESC);
+ rcvdBytesMetric = mreg.longAdderMetric(RECEIVED_BYTES_METRIC_NAME, RECEIVED_BYTES_METRIC_DESC);
+
+ sentMsgsMetric = mreg.longAdderMetric(SENT_MESSAGES_METRIC_NAME, SENT_MESSAGES_METRIC_DESC);
+ rcvdMsgsMetric = mreg.longAdderMetric(RECEIVED_MESSAGES_METRIC_NAME, RECEIVED_MESSAGES_METRIC_DESC);
+ }
+
+ /** */
+ private static synchronized MetricRegistry getOrCreateMetricRegistry(GridMetricManager mmgr, UUID nodeId) {
+ String regName = MetricUtils.metricName(COMMUNICATION_METRICS_GROUP_NAME, nodeId.toString());
+
+ for (MetricRegistry mreg : mmgr) {
Review comment:
We can eliminate iteration on all registries if we use `GridMetricManager#addMetricRegistryCreationListener` to implement registry initialization logic.
```
cctx.kernalContext().metric().addMetricRegistryCreationListener(mreg -> {
if (!mreg.name().startsWith(COMMUNICATION_METRICS_GROUP_NAME))
return;
mreg.longAdderMetric(SENT_MESSAGES_BY_NODE_ID_METRIC_NAME, SENT_MESSAGES_BY_NODE_ID_METRIC_DESC);
mreg.longAdderMetric(RECEIVED_MESSAGES_BY_NODE_ID_METRIC_NAME, RECEIVED_MESSAGES_BY_NODE_ID_METRIC_DESC);
});
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services