IGNITE-28919 Add communication and discovery IO test commands

docs/_docs/tools/control-script.adoc

@@ -1187,6 +1187,56 @@ TASKS                          SYS       Running compute tasks
 Command [SYSTEM-VIEW] finished with code: 0
 
 
+== Network I/O Diagnostics
+
+The `--io-test` commands actively measure network latency between cluster nodes.
+
+=== Communication SPI Diagnostic
+
+The `communication` test sends request-response messages from the selected source node to every remote server node through the
+Communication SPI.
+The source node specified by `--node-id` and at least one remote server node must be available.
+The first failed or timed-out request aborts the test.
+
+Optional parameters:
+
+* `--warmup` - warm-up period before samples are recorded, in milliseconds (`0` to `3600000`; default: `5000`).
+* `--duration` - sampling period after warm-up, in milliseconds (`1` to `3600000`; default: `30000`).
+* `--threads` - number of concurrent test threads (`1` to `64`; default: `1`).
+* `--payload-size` - payload in each request and response, in bytes (`0` to `1048576`; default: `0`).
+* `--process-in-nio-thread` - processes requests and responses in network I/O threads instead of the system pool.
+
+[source,shell]
+----
+control.sh --io-test communication --node-id <source-node-id>
+----
+
+=== Discovery SPI Diagnostic
+
+The `discovery` test measures how custom messages traverse the server ring and is executed on the cluster coordinator.
+It requires `TcpDiscoverySpi` on the coordinator and at least two server nodes.
+
+Optional parameters:
+
+* `--samples` - number of full-ring samples (`1` to `100`; default: `10`).
+* `--interval` - delay between samples, in milliseconds (`10` to `60000`; default: `100`).
+* `--payload-size` - discovery message payload, in bytes (`0` to `65536`; default: `100`).
+
+[source,shell]
+----
+control.sh --io-test discovery
+----
+
+WARNING: These tests generate network traffic and consume CPU on participating nodes. High thread counts, long durations, or large
+payloads can affect cluster workloads.
+The `--process-in-nio-thread` communication option handles test messages directly on network I/O threads and should only be used
+during controlled diagnostic runs because it can delay normal network processing.
+
+NOTE: Estimated one-way Communication SPI delivery times and per-hop Discovery SPI delays use the operating system wall clock
+(`System.currentTimeMillis`) on different nodes and therefore require synchronized clocks.
+RTT and full-ring latency use a monotonic clock and do not require clock synchronization.
+
+
 == Working With Persistence Data
 
 [WARNING]

modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java

@@ -70,6 +70,7 @@
 import org.apache.ignite.internal.management.encryption.EncryptionChangeMasterKeyCommand;
 import org.apache.ignite.internal.management.encryption.EncryptionCommand;
 import org.apache.ignite.internal.management.event.EventCommand;
+import org.apache.ignite.internal.management.io.IoTestCommand;
 import org.apache.ignite.internal.management.kill.KillCommand;
 import org.apache.ignite.internal.management.meta.MetaCommand;
 import org.apache.ignite.internal.management.meta.MetaRemoveCommand;
@@ -1214,6 +1215,7 @@ private boolean requireArgs(Class<?> cmd) {
             cmd == PerformanceStatisticsCommand.class ||
             cmd == ConsistencyCommand.class ||
             cmd == CdcCommand.class ||
+            cmd == IoTestCommand.class ||
             cmd == EventCommand.class;
     }
 }

modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite2.java

@@ -33,6 +33,7 @@
 import org.apache.ignite.util.GridCommandHandlerPropertiesTest;
 import org.apache.ignite.util.GridCommandHandlerScheduleIndexRebuildTest;
 import org.apache.ignite.util.IdleVerifyDumpTest;
+import org.apache.ignite.util.IoTestCommandTest;
 import org.apache.ignite.util.MetricCommandTest;
 import org.apache.ignite.util.PerformanceStatisticsCommandTest;
 import org.apache.ignite.util.SystemViewCommandTest;
@@ -61,6 +62,7 @@
 
     SystemViewCommandTest.class,
     MetricCommandTest.class,
+    IoTestCommandTest.class,
     PerformanceStatisticsCommandTest.class,
     CacheMetricsCommandTest.class,
 

modules/control-utility/src/test/java/org/apache/ignite/util/IoTestCommandTest.java

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.util;
+
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.management.io.IoTestCommand;
+import org.junit.Test;
+
+import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_INVALID_ARGUMENTS;
+import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_OK;
+import static org.apache.ignite.util.SystemViewCommandTest.NODE_ID;
+
+/**
+ * Tests for the {@link IoTestCommand}.
+ */
+public class IoTestCommandTest extends GridCommandHandlerAbstractTest {
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        super.beforeTest();
+
+        injectTestSystemOut();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        super.afterTest();
+
+        stopAllGrids();
+    }
+
+    /** */
+    @Test
+    public void testCommunication() throws Exception {
+        IgniteEx srv = startGrids(2);
+
+        String output = executeCommand(
+            EXIT_CODE_OK,
+            "--io-test", "communication",
+            NODE_ID, srv.localNode().id().toString(),
+            "--warmup", "0",
+            "--duration", "100",
+            "--threads", "1",
+            "--payload-size", "32",
+            "--process-in-nio-thread"
+        );
+
+        assertTrue(output, output.contains("Communication SPI test"));
+        assertTrue(output, output.contains("id=" + srv.localNode().id()));
+        assertTrue(output, output.contains("Parameters: warmup=0 ms | duration=100 ms | threads=1 | payload=32 bytes"));
+        assertTrue(output, output.contains("Handling: NIO thread"));
+        assertTrue(output, output.contains(grid(1).localNode().consistentId().toString()));
+        assertFalse(output, output.contains(grid(1).localNode().id().toString()));
+        assertTrue(output, output.contains("RTT:"));
+        assertTrue(output, output.contains("Target"));
+        assertTrue(output, output.contains("Samples"));
+        assertTrue(output, output.contains("Min, ms"));
+        assertTrue(output, output.contains("Avg, ms"));
+        assertTrue(output, output.contains("Max, ms"));
+        assertTrue(output, output.contains("Estimated one-way delivery*:"));
+        assertTrue(output, output.contains("Request"));
+        assertTrue(output, output.contains("Response"));
+        assertTrue(output, output.contains("OS wall-clock time and requires synchronized node clocks"));
+        assertTrue(output, output.contains("RTT uses a monotonic clock"));
+        assertFalse(output, output.contains("System-pool handling includes executor dispatch time"));
+        assertFalse(output, output.contains("RTT histogram"));
+    }
+
+    /** */
+    @Test
+    public void testDiscovery() throws Exception {
+        startGrids(3);
+
+        String output = executeCommand(
+            EXIT_CODE_OK,
+            "--io-test", "discovery",
+            "--samples", "3",
+            "--interval", "10",
+            "--payload-size", "32"
+        );
+
+        assertTrue(output, output.contains("TcpDiscoverySpi ring test"));
+        assertTrue(output, output.contains("id=" + grid(0).localNode().id()));
+        assertTrue(output, output.contains("Parameters: samples=3 | interval=10 ms | payload=32 bytes"));
+        assertFalse(output, output.contains("Server ring path:"));
+        assertTrue(output, output.contains(grid(0).localNode().consistentId().toString()));
+        assertTrue(output, output.contains(grid(1).localNode().consistentId().toString()));
+        assertTrue(output, output.contains(grid(2).localNode().consistentId().toString()));
+        assertFalse(output, output.contains(grid(1).localNode().id().toString()));
+        assertFalse(output, output.contains(grid(2).localNode().id().toString()));
+        assertTrue(output, output.contains("Full-ring latency:"));
+        assertTrue(output, output.contains("Samples"));
+        assertTrue(output, output.contains("Min, ms"));
+        assertTrue(output, output.contains("Avg, ms"));
+        assertTrue(output, output.contains("Max, ms"));
+        assertFalse(output, output.contains("P50"));
+        assertFalse(output, output.contains("P95"));
+        assertTrue(output, output.contains("Estimated per-hop delivery (ring order)*:"));
+        assertTrue(output, output.contains("OS wall-clock time and requires synchronized node clocks"));
+        assertTrue(output, output.contains("Full-ring latency uses a monotonic clock"));
+    }
+
+    /** */
+    @Test
+    public void testInvalidPayloadSize() {
+        executeCommand(
+            EXIT_CODE_INVALID_ARGUMENTS,
+            "--io-test", "discovery",
+            "--payload-size", "65537"
+        );
+    }
+}

modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java

@@ -33,6 +33,8 @@
 import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoMessage;
 import org.apache.ignite.internal.managers.deployment.GridDeploymentRequest;
 import org.apache.ignite.internal.managers.deployment.GridDeploymentResponse;
+import org.apache.ignite.internal.managers.discovery.IoTestDiscoveryAckMessage;
+import org.apache.ignite.internal.managers.discovery.IoTestDiscoveryMessage;
 import org.apache.ignite.internal.managers.encryption.ChangeCacheEncryptionRequest;
 import org.apache.ignite.internal.managers.encryption.EncryptionDataBagItem;
 import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyRequest;
@@ -492,6 +494,8 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) {
         register(CacheJoinNodeDiscoveryData.class);
         register(CacheReconnectInfo.class);
         register(ClusterCacheGroupRecoveryData.class);
+        register(IoTestDiscoveryMessage.class);
+        register(IoTestDiscoveryAckMessage.class);
 
         // [10000 - 10200]: Transaction and lock related messages. Most of them originally comes from Communication.
         msgIdx = 10000;

modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java

@@ -3360,26 +3360,6 @@ public void dumpDebugInfo() {
         }
     }
 
-    /**
-     * @param node Node.
-     * @param payload Message payload.
-     * @param procFromNioThread If {@code true} message is processed from NIO thread.
-     * @return Response future.
-     */
-    public IgniteInternalFuture sendIoTest(ClusterNode node, byte[] payload, boolean procFromNioThread) {
-        return ctx.io().sendIoTest(node, payload, procFromNioThread);
-    }
-
-    /**
-     * @param nodes Nodes.
-     * @param payload Message payload.
-     * @param procFromNioThread If {@code true} message is processed from NIO thread.
-     * @return Response future.
-     */
-    public IgniteInternalFuture sendIoTest(List<ClusterNode> nodes, byte[] payload, boolean procFromNioThread) {
-        return ctx.io().sendIoTest(nodes, payload, procFromNioThread);
-    }
-
     /** Registers configuration system view. */
     private void registerConfigurationSystemView() {
         ctx.systemView().registerInnerCollectionView(

modules/core/src/main/java/org/apache/ignite/internal/IgniteMXBeanImpl.java

@@ -24,8 +24,11 @@
 import javax.management.JMException;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.cluster.ClusterState;
+import org.apache.ignite.internal.management.io.IoTestCommunicationCommand;
+import org.apache.ignite.internal.managers.communication.IoTestResult;
 import org.apache.ignite.internal.util.typedef.internal.A;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.mxbean.IgniteMXBean;
@@ -298,6 +301,7 @@ public IgniteMXBeanImpl(IgniteKernal kernal) {
     }
 
     /** {@inheritDoc} */
+    @Deprecated
     @Override public void runIoTest(
         long warmup,
         long duration,
@@ -307,8 +311,28 @@ public IgniteMXBeanImpl(IgniteKernal kernal) {
         int payLoadSize,
         boolean procFromNioThread
     ) {
-        ctx.io().runIoTest(warmup, duration, threads, maxLatency, rangesCnt, payLoadSize,
-            procFromNioThread, new ArrayList(ctx.cluster().get().forServers().forRemotes().nodes()));
+        IgniteInternalFuture<IoTestResult> fut = ctx.io().ioTest().runIoTest(
+            warmup,
+            duration,
+            threads,
+            payLoadSize,
+            procFromNioThread,
+            new ArrayList<>(ctx.cluster().get().forServers().forRemotes().nodes())
+        );
+
+        fut.listen(f -> {
+            IgniteLogger log = ctx.log(ctx.io().getClass());
+
+            try {
+                IoTestResult res = f.get();
+
+                if (log.isInfoEnabled())
+                    log.info(IoTestCommunicationCommand.formatResult(res));
+            }
+            catch (IgniteCheckedException | RuntimeException e) {
+                U.error(log, "IO test failed.", e);
+            }
+        });
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/management/IgniteCommandRegistry.java

@@ -30,6 +30,7 @@
 import org.apache.ignite.internal.management.diagnostic.DiagnosticCommand;
 import org.apache.ignite.internal.management.encryption.EncryptionCommand;
 import org.apache.ignite.internal.management.event.EventCommand;
+import org.apache.ignite.internal.management.io.IoTestCommand;
 import org.apache.ignite.internal.management.kill.KillCommand;
 import org.apache.ignite.internal.management.meta.MetaCommand;
 import org.apache.ignite.internal.management.metric.MetricCommand;
@@ -75,7 +76,8 @@ public IgniteCommandRegistry() {
             new PerformanceStatisticsCommand(),
             new CdcCommand(),
             new ConsistencyCommand(),
-            new EventCommand()
+            new EventCommand(),
+            new IoTestCommand()
         );
 
         U.loadService(CommandsProvider.class).forEach(p -> p.commands().forEach(this::register));

modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommand.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.management.io;
+
+import org.apache.ignite.internal.management.api.CommandRegistryImpl;
+
+/** */
+public class IoTestCommand extends CommandRegistryImpl {
+    /** */
+    public IoTestCommand() {
+        super(
+            new IoTestCommunicationCommand(),
+            new IoTestDiscoveryCommand()
+        );
+    }
+}