IGNITE-27573 Adapt jraft node metrics to ignite metric framework

modules/cli/build.gradle

@@ -104,6 +104,10 @@ dependencies {
     integrationTestImplementation project(':ignite-sql-engine')
     integrationTestImplementation project(':ignite-metrics')
     integrationTestImplementation project(':ignite-network-api')
+    integrationTestImplementation project(':ignite-raft')
+    integrationTestImplementation project(':ignite-raft-api')
+    integrationTestImplementation project(':ignite-metastorage')
+    integrationTestImplementation project(':ignite-cluster-management')
     integrationTestImplementation testFixtures(project(':ignite-core'))
     integrationTestImplementation testFixtures(project(':ignite-schema'))
     integrationTestImplementation testFixtures(project(':ignite-api'))

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

@@ -17,7 +17,7 @@
 
 package org.apache.ignite.internal.cli;
 
-import static org.apache.ignite.internal.metrics.sources.ThreadPoolMetricSource.THREAD_POOLS_METRICS_SOURCE_NAME;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import io.micronaut.configuration.picocli.MicronautFactory;
@@ -27,10 +27,13 @@
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.io.Writer;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
+import org.apache.ignite.Ignite;
 import org.apache.ignite.internal.ClusterConfiguration;
 import org.apache.ignite.internal.ClusterPerClassIntegrationTest;
 import org.apache.ignite.internal.cli.call.connect.ConnectCall;
@@ -61,41 +64,6 @@
 @MicronautTest(rebuildContext = true)
 public abstract class CliIntegrationTest extends ClusterPerClassIntegrationTest {
 
-    public static final MetricSource[] ALL_METRIC_SOURCES = {
-            new MetricSource().name("jvm").enabled(true),
-            new MetricSource().name("os").enabled(true),
-            new MetricSource().name("raft").enabled(true),
-            new MetricSource().name("metastorage").enabled(true),
-            new MetricSource().name("schema.sync").enabled(true),
-            new MetricSource().name("client.handler").enabled(true),
-            new MetricSource().name("sql.client").enabled(true),
-            new MetricSource().name("sql.plan.cache").enabled(true),
-            new MetricSource().name("sql.queries").enabled(true),
-            new MetricSource().name("storage.aipersist").enabled(true),
-            new MetricSource().name("storage.aipersist.default").enabled(true),
-            new MetricSource().name("storage.aipersist.default_aipersist").enabled(true),
-            new MetricSource().name("storage.aipersist.checkpoint").enabled(true),
-            new MetricSource().name("storage.aipersist.io").enabled(true),
-            new MetricSource().name("topology.cluster").enabled(true),
-            new MetricSource().name("topology.local").enabled(true),
-            new MetricSource().name("thread.pools.partitions-executor").enabled(true),
-            new MetricSource().name("thread.pools.sql-executor").enabled(true),
-            new MetricSource().name("thread.pools.sql-planning-executor").enabled(true),
-            new MetricSource().name("transactions").enabled(true),
-            new MetricSource().name("placement-driver").enabled(true),
-            new MetricSource().name("resource.vacuum").enabled(true),
-            new MetricSource().name("zones.Default").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("messaging").enabled(true),
-            new MetricSource().name("log.storage").enabled(true),
-            new MetricSource().name(THREAD_POOLS_METRICS_SOURCE_NAME + "striped.messaging.inbound.default").enabled(true),
-            new MetricSource().name(THREAD_POOLS_METRICS_SOURCE_NAME + "striped.messaging.inbound.deploymentunits").enabled(true),
-            new MetricSource().name(THREAD_POOLS_METRICS_SOURCE_NAME + "striped.messaging.inbound.scalecube").enabled(true),
-            new MetricSource().name(THREAD_POOLS_METRICS_SOURCE_NAME + "messaging.outbound").enabled(true),
-    };
-
     /** Correct ignite jdbc url. */
     protected static final String JDBC_URL = "jdbc:ignite:thin://127.0.0.1:" + ClusterConfiguration.DEFAULT_BASE_CLIENT_PORT;
 
@@ -373,6 +341,65 @@ protected static void createAndPopulateTable() {
         );
     }
 
+    /** Returns all metric sources that are expected to be present on all nodes combined. */
+    public static MetricSource[] getExpectedClusterMetrics() {
+        Set<MetricSource> result = new HashSet<>();
+
+        for (var node : CLUSTER.nodes()) {
+            result.addAll(List.of(getExpectedNodeMetrics(node)));
+        }
+
+        return result.toArray(MetricSource[]::new);
+    }
+
+    /** Returns all metric sources that are expected to be present on a specific node. */
+    public static MetricSource[] getExpectedNodeMetrics(Ignite ignite) {
+        MetricSource[] commonMetrics = {new MetricSource().name("client.handler").enabled(true),
+                new MetricSource().name("clock.service").enabled(true),
+                new MetricSource().name("index.builder").enabled(true),
+                new MetricSource().name("jvm").enabled(true),
+                new MetricSource().name("log.storage").enabled(true),
+                new MetricSource().name("messaging").enabled(true),
+                new MetricSource().name("metastorage").enabled(true),
+                new MetricSource().name("os").enabled(true),
+                new MetricSource().name("placement-driver").enabled(true),
+                new MetricSource().name("raft").enabled(true),
+                new MetricSource().name("raft.snapshots").enabled(true),
+                new MetricSource().name("resource.vacuum").enabled(true),
+                new MetricSource().name("sql.client").enabled(true),
+                new MetricSource().name("sql.plan.cache").enabled(true),
+                new MetricSource().name("sql.queries").enabled(true),
+                new MetricSource().name("schema.sync").enabled(true),
+                new MetricSource().name("storage.aipersist").enabled(true),
+                new MetricSource().name("storage.aipersist.checkpoint").enabled(true),
+                new MetricSource().name("storage.aipersist.default").enabled(true),
+                new MetricSource().name("storage.aipersist.default_aipersist").enabled(true),
+                new MetricSource().name("storage.aipersist.io").enabled(true),
+                new MetricSource().name("thread.pools.messaging.outbound").enabled(true),
+                new MetricSource().name("thread.pools.partitions-executor").enabled(true),
+                new MetricSource().name("thread.pools.sql-executor").enabled(true),
+                new MetricSource().name("thread.pools.sql-planning-executor").enabled(true),
+                new MetricSource().name("thread.pools.striped.messaging.inbound.default").enabled(true),
+                new MetricSource().name("thread.pools.striped.messaging.inbound.deploymentunits").enabled(true),
+                new MetricSource().name("thread.pools.striped.messaging.inbound.scalecube").enabled(true),
+                new MetricSource().name("topology.cluster").enabled(true),
+                new MetricSource().name("topology.local").enabled(true),
+                new MetricSource().name("transactions").enabled(true),
+                new MetricSource().name("zones.Default").enabled(true)
+        };
+
+        List<MetricSource> metrics = new ArrayList<>(Arrays.asList(commonMetrics));
+
+        for (var node : unwrapIgniteImpl(ignite).raftManager().localNodes()) {
+            metrics.add(new MetricSource().name("raft.fsmcaller." + node.groupId().toString()).enabled(true));
+            metrics.add(new MetricSource().name("raft.logmanager." + node.groupId().toString()).enabled(true));
+            metrics.add(new MetricSource().name("raft.node." + node.groupId().toString()).enabled(true));
+            metrics.add(new MetricSource().name("raft.readonlyservice." + node.groupId().toString()).enabled(true));
+        }
+
+        return metrics.toArray(MetricSource[]::new);
+    }
+
     protected static PrintWriter output(List<Character> buffer) {
         return new PrintWriter(new Writer() {
             @Override

modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/call/metric/ItEnabledMetricCallsTest.java

@@ -52,7 +52,7 @@ class ItEnabledMetricCallsTest extends CliIntegrationTest {
     @BeforeAll
     void beforeAll() {
         // Disable all metrics except "jvm".
-        Arrays.stream(ALL_METRIC_SOURCES).map(MetricSource::getName).forEach(name -> {
+        Arrays.stream(getExpectedClusterMetrics()).map(MetricSource::getName).forEach(name -> {
             if ("jvm".equals(name)) {
                 return;
             }

modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/call/metric/ItMetricCallsTest.java

@@ -68,8 +68,9 @@ void nodeMetricSourcesList() {
         assertThat(output.hasError()).isFalse();
 
         // And
-        assertThat(output.body()).contains(ALL_METRIC_SOURCES);
-        assertThat(output.body()).hasSize(ALL_METRIC_SOURCES.length);
+        MetricSource[] expectedMetrics = getExpectedNodeMetrics(node(0));
+
+        assertThat(output.body()).containsExactlyInAnyOrder(expectedMetrics);
     }
 
     @Test
@@ -86,7 +87,7 @@ void clusterMetricSourcesList() {
         ThrowingConsumer<NodeMetricSources>[] assertions = CLUSTER.runningNodes()
                 .map(ignite -> (ThrowingConsumer<NodeMetricSources>) input -> {
                     assertThat(input.getNode()).isEqualTo(ignite.name());
-                    assertThat(input.getSources()).containsExactlyInAnyOrder(ALL_METRIC_SOURCES);
+                    assertThat(input.getSources()).containsExactlyInAnyOrder(getExpectedNodeMetrics(ignite));
                 })
                 .toArray(ThrowingConsumer[]::new);
 

modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/metric/ItClusterMetricCommandTest.java

@@ -81,12 +81,6 @@ void metricList() {
         // When list cluster metrics with valid url
         execute("cluster", "metric", "source", "list", "--plain", "--url", NODE_URL);
 
-        // Then
-        List<String> sourceNames = Arrays.stream(ALL_METRIC_SOURCES)
-                .map(org.apache.ignite.rest.client.model.MetricSource::getName)
-                .sorted()
-                .collect(toList());
-
         List<String> nodeNames = CLUSTER.nodes().stream()
                 .map(Ignite::name)
                 .sorted()
@@ -97,14 +91,17 @@ void metricList() {
         assertions.add(this::assertExitCodeIsZero);
         assertions.add(this::assertErrOutputIsEmpty);
         assertions.add(() -> assertOutputContains("Node\tSource name\tAvailability"));
-        for (String sourceName : sourceNames) {
-            assertions.add(() -> assertOutputContains(sourceName + "\tenabled" + NL));
-        }
-        // Header + number of nodes * (node name header + metric sources)
-        assertions.add(() -> assertOutputHasLineCount(1 + initialNodes() * (ALL_METRIC_SOURCES.length + 1)));
         // Let's check that the substrings are in the correct order.
         assertions.add(() -> assertOutputContainsSubsequence(nodeNames));
-        assertions.add(() -> assertOutputContainsSubsequence(sourceNames));
+
+        for (Ignite node : CLUSTER.nodes()) {
+            List<String> expectedMetrics = Arrays.stream(getExpectedNodeMetrics(node))
+                            .map(org.apache.ignite.rest.client.model.MetricSource::getName)
+                            .sorted()
+                            .collect(toList());
+
+            assertions.add(() -> assertOutputContainsSubsequence(expectedMetrics));
+        }
 
         assertAll(assertions);
     }

modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/metric/ItEnabledNodeMetricCommandTest.java

@@ -40,7 +40,7 @@ class ItEnabledNodeMetricCommandTest extends CliIntegrationTest {
     @BeforeAll
     void beforeAll() {
         // Disable all metrics except "jvm".
-        Arrays.stream(ALL_METRIC_SOURCES).map(MetricSource::getName).forEach(name -> {
+        Arrays.stream(getExpectedClusterMetrics()).map(MetricSource::getName).forEach(name -> {
             if ("jvm".equals(name)) {
                 return;
             }

modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/metric/ItNodeMetricCommandTest.java

@@ -79,7 +79,7 @@ void metricList() {
         List<Executable> assertions = new ArrayList<>();
         assertions.add(this::assertExitCodeIsZero);
         assertions.add(this::assertErrOutputIsEmpty);
-        for (org.apache.ignite.rest.client.model.MetricSource source : ALL_METRIC_SOURCES) {
+        for (org.apache.ignite.rest.client.model.MetricSource source : getExpectedNodeMetrics(CLUSTER.node(0))) {
             assertions.add(() -> assertOutputContains(source.getName() + "\tenabled" + NL));
         }
         assertAll(assertions);

modules/cli/src/integrationTest/java/org/apache/ignite/internal/rest/ItGeneratedRestClientTest.java

@@ -263,7 +263,7 @@ void nodeVersion() throws ApiException {
 
     @Test
     void nodeMetricSourcesList() throws ApiException {
-        assertThat(nodeMetricApi.listNodeMetricSources(), containsInAnyOrder(CliIntegrationTest.ALL_METRIC_SOURCES));
+        assertThat(nodeMetricApi.listNodeMetricSources(), containsInAnyOrder(CliIntegrationTest.getExpectedNodeMetrics(CLUSTER.node(0))));
     }
 
     @Test

modules/metrics/src/main/java/org/apache/ignite/internal/metrics/exporters/configuration/LogPushExporterConfigurationSchema.java

@@ -50,7 +50,8 @@ public class LogPushExporterConfigurationSchema extends ExporterConfigurationSch
             "metastorage",
             "os",
             "placement-driver",
-            "raft*",
+            "raft",
+            "raft.snapshots",
             "resource.vacuum",
             "sql.plan.cache",
             "storage.aipersist",

modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java

@@ -56,7 +56,6 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
-import com.codahale.metrics.ConsoleReporter;
 import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.RingBuffer;
 import java.io.File;
@@ -101,6 +100,7 @@
 import org.apache.ignite.internal.logger.IgniteLogger;
 import org.apache.ignite.internal.logger.Loggers;
 import org.apache.ignite.internal.manager.ComponentContext;
+import org.apache.ignite.internal.metrics.TestMetricManager;
 import org.apache.ignite.internal.network.ClusterService;
 import org.apache.ignite.internal.network.StaticNodeFinder;
 import org.apache.ignite.internal.network.utils.ClusterServiceTestUtils;
@@ -1667,40 +1667,6 @@ public void run(Status status, long index, byte[] reqCtx) {
             assertEquals(10000, fsm.getLogs().size());
     }
 
-    @Test
-    public void testNodeMetrics() throws Exception {
-        List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
-
-        cluster = new TestCluster("unittest", dataPath, peers, testInfo);
-        for (TestPeer peer : peers)
-            assertTrue(cluster.start(peer, false, 300, true));
-
-        //elect and get leader
-        Node leader = cluster.waitAndGetLeader();
-        assertNotNull(leader);
-        assertEquals(3, leader.listPeers().size());
-        // apply tasks to leader
-        sendTestTaskAndWait(leader);
-
-        {
-            ByteBuffer data = ByteBuffer.wrap("no closure".getBytes(UTF_8));
-            Task task = new Task(data, null);
-            leader.apply(task);
-        }
-
-        cluster.ensureSame();
-        for (Node node : cluster.getNodes()) {
-            System.out.println("-------------" + node.getNodeId() + "-------------");
-            ConsoleReporter reporter = ConsoleReporter.forRegistry(node.getNodeMetrics().getMetricRegistry())
-                .build();
-            reporter.report();
-            reporter.close();
-            System.out.println();
-        }
-        // TODO check http status https://issues.apache.org/jira/browse/IGNITE-14832
-        assertEquals(2, cluster.getFollowers().size());
-    }
-
     @Test
     public void testLeaderFail() throws Exception {
         List<TestPeer> peers = TestUtils.generatePeers(testInfo, 3);
@@ -3331,7 +3297,7 @@ public void testBootStrapWithSnapshot() throws Exception {
         opts.setGroupConf(JRaftUtils.getConfiguration(peer.getPeerId().toString()));
         opts.setFsm(fsm);
 
-        assertTrue(JRaftUtils.bootstrap(opts));
+        assertTrue(JRaftUtils.bootstrap(opts, new TestMetricManager()));
         assertThat(logStorageProvider.stopAsync(new ComponentContext()), willCompleteSuccessfully());
 
         NodeOptions nodeOpts = new NodeOptions();
@@ -3378,7 +3344,7 @@ public void testBootStrapWithoutSnapshot() throws Exception {
         opts.setGroupConf(JRaftUtils.getConfiguration(peer.getPeerId().toString()));
         opts.setFsm(fsm);
 
-        assertTrue(JRaftUtils.bootstrap(opts));
+        assertTrue(JRaftUtils.bootstrap(opts, new TestMetricManager()));
         assertThat(logStorageProvider.stopAsync(new ComponentContext()), willCompleteSuccessfully());
 
         NodeOptions nodeOpts = new NodeOptions();
@@ -4973,7 +4939,7 @@ private RaftGroupService createService(String groupId, TestPeer peer, NodeOption
 
         assertThat(clusterService.startAsync(new ComponentContext()), willCompleteSuccessfully());
 
-        var service = new RaftGroupService(groupId, peer.getPeerId(), nodeOptions, rpcServer) {
+        var service = new RaftGroupService(groupId, peer.getPeerId(), nodeOptions, rpcServer, new TestMetricManager()) {
             @Override
             public synchronized void shutdown() {
                 rpcServer.shutdown();

modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/TestCluster.java

@@ -53,6 +53,7 @@
 import org.apache.ignite.internal.logger.IgniteLogger;
 import org.apache.ignite.internal.logger.Loggers;
 import org.apache.ignite.internal.manager.ComponentContext;
+import org.apache.ignite.internal.metrics.TestMetricManager;
 import org.apache.ignite.internal.network.ClusterService;
 import org.apache.ignite.internal.network.StaticNodeFinder;
 import org.apache.ignite.internal.raft.JraftGroupEventsListener;
@@ -217,7 +218,6 @@ public boolean start(TestPeer peer, boolean emptyPeers, int snapshotIntervalSecs
             nodeOptions.setServerName(peer.getPeerId().toString());
 
             nodeOptions.setElectionTimeoutMs(this.electionTimeoutMs);
-            nodeOptions.setEnableMetrics(enableMetrics);
             nodeOptions.setSnapshotThrottle(snapshotThrottle);
             nodeOptions.setSnapshotIntervalSecs(snapshotIntervalSecs);
             nodeOptions.setServiceFactory(this.raftServiceFactories.apply(peer.getPeerId()));
@@ -285,7 +285,7 @@ public boolean start(TestPeer peer, boolean emptyPeers, int snapshotIntervalSecs
                 optsClo.accept(peer.getPeerId(), nodeOptions);
 
             RaftGroupService server = new RaftGroupService(this.name, peer.getPeerId(),
-                nodeOptions, rpcServer) {
+                nodeOptions, rpcServer, new TestMetricManager()) {
                 @Override public synchronized void shutdown() {
                     nodeManager.shutdown();
                     // This stop order is consistent with JRaftServerImpl