[AI Generated] perf(ntttcp): run ntttcp/lagscope over internal IPv4 on dual-stack (use_ipv6)

lisa/microsoft/testsuites/performance/common.py

@@ -381,6 +381,11 @@ def perf_ntttcp(  # noqa: C901
         # if it's not filled, assume it's called by case directly.
         test_case_name = inspect.stack()[1][3]
 
+    # ntttcp and lagscope run over IPv4. On dual-stack (use_ipv6) environments
+    # server.internal_address is IPv6, so resolve the server's internal IPv4 to
+    # use as the ntttcp/lagscope target.
+    server_comm_address = server.nics.get_internal_ipv4_address()
+
     if connections is None:
         if udp_mode:
             connections = NTTTCP_UDP_CONCURRENCY
@@ -535,7 +540,7 @@ def perf_ntttcp(  # noqa: C901
                         ip=(
                             lagscope_server_ip
                             if lagscope_server_ip is not None
-                            else server.internal_address
+                            else server_comm_address
                         ),
                         no_debug_log=True,
                     )
@@ -545,9 +550,7 @@ def perf_ntttcp(  # noqa: C901
                     server_result = server_ntttcp.run_as_server_async(
                         server_nic_name,
                         server_ip=(
-                            server.internal_address
-                            if isinstance(server.os, BSD)
-                            else ""
+                            server_comm_address if isinstance(server.os, BSD) else ""
                         ),
                         ports_count=num_threads_p,
                         buffer_size=buffer_size,
@@ -562,7 +565,7 @@ def perf_ntttcp(  # noqa: C901
                     # Start lagscope client to measure latency during the
                     # ntttcp test
                     client_lagscope_process = client_lagscope.run_as_client_async(
-                        server_ip=server.internal_address,
+                        server_ip=server_comm_address,
                         ping_count=0,
                         run_time_seconds=(
                             run_time_seconds
@@ -582,7 +585,7 @@ def perf_ntttcp(  # noqa: C901
                     # Use daemon mode to run in background, then monitor process
                     client_ntttcp_result = client_ntttcp.run_as_client(
                         client_nic_name,
-                        server.internal_address,
+                        server_comm_address,
                         buffer_size=buffer_size,
                         threads_count=num_threads_n,
                         ports_count=num_threads_p,

lisa/nic.py

@@ -8,7 +8,7 @@
 from collections import OrderedDict
 from dataclasses import dataclass
 from pathlib import PurePosixPath
-from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
 
 from assertpy import assert_that
 from retry import retry
@@ -18,6 +18,7 @@
 
 if TYPE_CHECKING:
     from lisa import Node
+    from lisa.node import RemoteNode
 
 
 class NicInfo:
@@ -274,6 +275,34 @@ def get_nic_names(self) -> List[str]:
     def get_primary_nic(self) -> NicInfo:
         return self.get_nic_by_index(0)
 
+    def get_internal_ipv4_address(self) -> str:
+        # Return an IPv4 address usable for intra-VM communication. Prefer the
+        # node's internal_address, but on a dual-stack (use_ipv6) environment
+        # that address is IPv6 -- tools that bind/connect over IPv4 only (e.g.
+        # ntttcp/lagscope) cannot use it. The node still has an internal IPv4
+        # on its primary (SRIOV) NIC, so fall back to that. IPv4-only
+        # deployments are unaffected: internal_address is already IPv4 and is
+        # returned unchanged.
+        address: str = cast("RemoteNode", self._node).internal_address
+        try:
+            is_ipv6 = ipaddress.ip_address(address).version == 6
+        except ValueError:
+            is_ipv6 = False
+        if not is_ipv6:
+            return address
+        ipv4_address: str = self.get_primary_nic().ip_addr
+        if not ipv4_address:
+            raise LisaException(
+                f"internal address {address} is IPv6 but the primary NIC "
+                f"'{self.get_primary_nic().name}' has no IPv4 address to fall "
+                f"back to."
+            )
+        self._node.log.debug(
+            f"internal address {address} is IPv6; using the node's internal "
+            f"IPv4 {ipv4_address} instead."
+        )
+        return ipv4_address
+
     def get_secondary_nic(self) -> NicInfo:
         # get a nic which isn't servicing the SSH connection with lisa.
         # will assert if none is present.