IGNITE-27381 Support MessageService metrics

modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/CliIntegrationTest.java

@@ -83,7 +83,12 @@ public abstract class CliIntegrationTest extends ClusterPerClassIntegrationTest
             new MetricSource().name("resource.vacuum").enabled(true),
             new MetricSource().name("clock.service").enabled(true),
             new MetricSource().name("index.builder").enabled(true),
-            new MetricSource().name("raft.snapshots").enabled(true)
+            new MetricSource().name("raft.snapshots").enabled(true),
+            new MetricSource().name("network.messaging").enabled(true),
+            new MetricSource().name("network.messaging.executor.inbound.Default").enabled(true),
+            new MetricSource().name("network.messaging.executor.inbound.DeploymentUnits").enabled(true),
+            new MetricSource().name("network.messaging.executor.inbound.ScaleCube").enabled(true),
+            new MetricSource().name("network.messaging.executor.outbound").enabled(true),
     };
 
     /** Correct ignite jdbc url. */

modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/distributionzones/ItIgniteDistributionZoneManagerNodeRestartTest.java

@@ -239,6 +239,7 @@ private PartialNode startPartialNode(int idx, Consumer<MetaStorageManager> metaS
                 clusterIdService,
                 new NoOpCriticalWorkerRegistry(),
                 failureProcessor,
+                new NoOpMetricManager(),
                 defaultChannelTypeRegistry(),
                 new DefaultIgniteProductVersionSource()
         );

modules/metrics/src/main/java/org/apache/ignite/internal/metrics/MetricManagerImpl.java

@@ -21,6 +21,7 @@
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toUnmodifiableMap;
 import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
 import static org.apache.ignite.internal.util.IgniteUtils.closeAllManually;
 import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
 import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
@@ -44,6 +45,7 @@
 import org.apache.ignite.configuration.notifications.ConfigurationNotificationEvent;
 import org.apache.ignite.internal.close.ManuallyCloseable;
 import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.lang.NodeStoppingException;
 import org.apache.ignite.internal.logger.IgniteLogger;
 import org.apache.ignite.internal.logger.Loggers;
 import org.apache.ignite.internal.manager.ComponentContext;
@@ -186,18 +188,30 @@ public void registerSource(MetricSource src) {
 
     @Override
     public void unregisterSource(MetricSource src) {
-        inBusyLockSafe(busyLock, () -> {
-            disable(src);
-            registry.unregisterSource(src);
-        });
+        try {
+            inBusyLockSafe(busyLock, () -> {
+                disable(src);
+                registry.unregisterSource(src);
+            });
+        } catch (Exception e) {
+            if (!hasCause(e, NodeStoppingException.class)) {
+                log.error("Failed to unregister metrics source {}", e, src.name());
+            }
+        }
     }
 
     @Override
     public void unregisterSource(String srcName) {
-        inBusyLockSafe(busyLock, () -> {
-            disable(srcName);
-            registry.unregisterSource(srcName);
-        });
+        try {
+            inBusyLockSafe(busyLock, () -> {
+                disable(srcName);
+                registry.unregisterSource(srcName);
+            });
+        } catch (Exception e) {
+            if (!hasCause(e, NodeStoppingException.class)) {
+                log.warn("Failed to unregister metrics source {}", e, srcName);
+            }
+        }
     }
 
     @Override

modules/metrics/src/main/java/org/apache/ignite/internal/metrics/sources/StripedThreadPoolMetricSource.java

@@ -22,18 +22,21 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 import org.apache.ignite.internal.metrics.AbstractMetricSource;
 import org.apache.ignite.internal.metrics.IntGauge;
 import org.apache.ignite.internal.metrics.LongGauge;
 import org.apache.ignite.internal.metrics.Metric;
+import org.apache.ignite.internal.thread.AbstractStripedThreadPoolExecutor;
 import org.apache.ignite.internal.thread.StripedThreadPoolExecutor;
 import org.jetbrains.annotations.Nullable;
 
 /** Metric source for monitoring of {@link StripedThreadPoolExecutor}. */
-public class StripedThreadPoolMetricSource extends AbstractMetricSource<StripedThreadPoolMetricSource.Holder> {
+public class StripedThreadPoolMetricSource<T extends AbstractStripedThreadPoolExecutor<? extends ExecutorService>> extends
+        AbstractMetricSource<StripedThreadPoolMetricSource<T>.Holder> {
     /** Striped thread pool to be monitored. */
-    private final StripedThreadPoolExecutor exec;
+    private final T exec;
 
     /**
      * Creates a new thread pool metric source with the given {@code name} to monitor the provided striped executor {@code exec},
@@ -44,8 +47,8 @@ public class StripedThreadPoolMetricSource extends AbstractMetricSource<StripedT
      * @param exec Striped thread pool executor to monitor.
      * @see StripedThreadPoolExecutor
      */
-    public StripedThreadPoolMetricSource(String name, @Nullable String description, StripedThreadPoolExecutor exec) {
-        this(name, null, THREAD_POOLS_GROUP_NAME, exec);
+    public StripedThreadPoolMetricSource(String name, @Nullable String description, T exec) {
+        this(name, description, THREAD_POOLS_GROUP_NAME, exec);
     }
 
     /**
@@ -60,7 +63,7 @@ public StripedThreadPoolMetricSource(
             String name,
             @Nullable String description,
             @Nullable String group,
-            StripedThreadPoolExecutor exec
+            T exec
     ) {
         super(name, description, group);
 

modules/network/build.gradle

@@ -33,6 +33,7 @@ dependencies {
     implementation project(':ignite-vault')
     implementation project(':ignite-workers')
     implementation project(':ignite-failure-handler')
+    implementation project(':ignite-metrics')
     implementation libs.jetbrains.annotations
     implementation libs.scalecube.cluster
     implementation libs.fastutil.core
@@ -56,6 +57,7 @@ dependencies {
     testImplementation testFixtures(project(':ignite-configuration'))
     testImplementation testFixtures(project(':ignite-core'))
     testImplementation testFixtures(project(':ignite-failure-handler'))
+    testImplementation testFixtures(project(':ignite-metrics'))
     testImplementation libs.jmh.core
     testImplementation(libs.kryo) {
         //IDEA test runner don't apply Gradle dependency resolve strategy, this is just not implemented
@@ -72,6 +74,7 @@ dependencies {
     testFixturesImplementation project(':ignite-configuration')
     testFixturesImplementation project(':ignite-configuration-root')
     testFixturesImplementation project(':ignite-failure-handler')
+    testFixturesImplementation testFixtures(project(':ignite-metrics'))
     testFixturesImplementation testFixtures(project(':ignite-core'))
     testFixturesImplementation testFixtures(project(':ignite-configuration'))
     testFixturesImplementation testFixtures(project(':ignite-workers'))

modules/network/src/main/java/org/apache/ignite/internal/network/CriticalStripedExecutors.java

@@ -27,9 +27,11 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.ignite.internal.close.ManuallyCloseable;
 import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.metrics.MetricManager;
 import org.apache.ignite.internal.thread.StripedExecutor;
 import org.apache.ignite.internal.worker.CriticalWorker;
 import org.apache.ignite.internal.worker.CriticalWorkerRegistry;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * Collection of {@link StripedExecutor executors} for the network based on {@link ChannelType#id()}.
@@ -51,11 +53,15 @@ class CriticalStripedExecutors implements ManuallyCloseable {
             String poolNamePrefix,
             CriticalWorkerRegistry workerRegistry,
             ChannelTypeRegistry channelTypeRegistry,
-            IgniteLogger log
+            IgniteLogger log,
+            @Nullable MetricManager metricManager,
+            @Nullable String metricNamePrefix,
+            @Nullable String metricDescription
     ) {
         this.workerRegistry = workerRegistry;
 
-        var factory = new CriticalStripedThreadPoolExecutorFactory(nodeName, poolNamePrefix, log, workerRegistry, registeredWorkers);
+        var factory = new CriticalStripedThreadPoolExecutorFactory(nodeName, poolNamePrefix, log, workerRegistry, registeredWorkers,
+                metricManager, metricNamePrefix, metricDescription);
 
         executorByChannelTypeId = StripedExecutorByChannelTypeId.of(channelTypeRegistry, factory);
     }

modules/network/src/main/java/org/apache/ignite/internal/network/CriticalStripedThreadPoolExecutorFactory.java

@@ -21,9 +21,11 @@
 
 import java.util.List;
 import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.metrics.MetricManager;
 import org.apache.ignite.internal.worker.CriticalStripedThreadPoolExecutor;
 import org.apache.ignite.internal.worker.CriticalWorker;
 import org.apache.ignite.internal.worker.CriticalWorkerRegistry;
+import org.jetbrains.annotations.Nullable;
 
 /** Factory for creating {@link CriticalStripedThreadPoolExecutor}. */
 class CriticalStripedThreadPoolExecutorFactory {
@@ -43,18 +45,45 @@ class CriticalStripedThreadPoolExecutorFactory {
 
     private final List<CriticalWorker> registeredWorkers;
 
+    @Nullable
+    private final MetricManager metricManager;
+
+    @Nullable
+    private final String metricNamePrefix;
+
+    @Nullable
+    private final String metricDescription;
+
+    @SuppressWarnings("unused")
     CriticalStripedThreadPoolExecutorFactory(
             String nodeName,
             String poolNamePrefix,
             IgniteLogger log,
             CriticalWorkerRegistry workerRegistry,
             List<CriticalWorker> registeredWorkers
+    ) {
+        this(nodeName, poolNamePrefix, log, workerRegistry, registeredWorkers, null, null, null);
+    }
+
+    CriticalStripedThreadPoolExecutorFactory(
+            String nodeName,
+            String poolNamePrefix,
+            IgniteLogger log,
+            CriticalWorkerRegistry workerRegistry,
+            List<CriticalWorker> registeredWorkers,
+            @Nullable MetricManager metricManager,
+            @Nullable String metricNamePrefix,
+            @Nullable String metricDescription
     ) {
         this.nodeName = nodeName;
         this.poolNamePrefix = poolNamePrefix;
         this.log = log;
         this.workerRegistry = workerRegistry;
         this.registeredWorkers = registeredWorkers;
+
+        this.metricManager = metricManager;
+        this.metricNamePrefix = metricNamePrefix;
+        this.metricDescription = metricDescription;
     }
 
     CriticalStripedThreadPoolExecutor create(ChannelType channelType) {
@@ -64,6 +93,12 @@ CriticalStripedThreadPoolExecutor create(ChannelType channelType) {
         var threadFactory = IgniteMessageServiceThreadFactory.create(nodeName, poolName, log, NOTHING_ALLOWED);
         var executor = new CriticalStripedThreadPoolExecutor(stripeCountForIndex(channelTypeId), threadFactory, false, 0);
 
+        if (metricManager != null && metricNamePrefix != null) {
+            String metricName = String.format("%s.%s", metricNamePrefix, channelType.name());
+
+            executor.initMetricSource(metricManager, metricName, metricDescription);
+        }
+
         for (CriticalWorker worker : executor.workers()) {
             workerRegistry.register(worker);
             registeredWorkers.add(worker);