feat: support multi-node rackmanager serial testing

lisa/features/serial_console.py

@@ -380,6 +380,9 @@ def check_initramfs(
     def read(self) -> str:
         raise NotImplementedError
 
+    def wait_for_ready(self, timeout: int) -> None:
+        pass
+
     def write(self, data: str) -> None:
         raise NotImplementedError
 

lisa/microsoft/testsuites/ltp/ltp.py

@@ -444,6 +444,7 @@ def _install_dependencies(self) -> None:
                     "kernel-headers",
                     "binutils",
                     "diffutils",
+                    "gawk",
                     "glibc-devel",
                     "zlib-devel",
                 ]

lisa/mixin_modules.py

@@ -40,7 +40,6 @@
 try:
     import lisa.sut_orchestrator.baremetal.build  # noqa: F401
     import lisa.sut_orchestrator.baremetal.cluster.cluster  # noqa: F401
-    import lisa.sut_orchestrator.baremetal.cluster.idrac  # noqa: F401
     import lisa.sut_orchestrator.baremetal.cluster.pxe  # noqa: F401
     import lisa.sut_orchestrator.baremetal.cluster.rackmanager  # noqa: F401
     import lisa.sut_orchestrator.baremetal.ip_getter  # noqa: F401
@@ -50,6 +49,11 @@
 except ModuleNotFoundError as e:
     print(f"baremetal package is not installed. [{e}]")
 
+try:
+    import lisa.sut_orchestrator.baremetal.cluster.idrac  # noqa: F401
+except ModuleNotFoundError as e:
+    print(f"baremetal idrac module is not installed. [{e}]")
+
 # Aws modules
 try:
     import lisa.sut_orchestrator.aws.platform_  # noqa: F401

lisa/node.py

@@ -784,6 +784,24 @@ def set_connection_info(
         self.internal_address = address
         self.internal_port = port
 
+    def update_connection_address(
+        self,
+        address: str,
+        public_address: str = "",
+        use_public_address: bool = True,
+    ) -> None:
+        if not address:
+            raise LisaException("address cannot be empty")
+        if not public_address:
+            public_address = address
+
+        self.close()
+        self._connection_info.address = (
+            public_address if use_public_address else address
+        )
+        self.public_address = public_address
+        self.internal_address = address
+
     def _initialize(self, *args: Any, **kwargs: Any) -> None:
         assert self._connection_info, "call setConnectionInfo before use remote node"
         super()._initialize(*args, **kwargs)

lisa/sut_orchestrator/baremetal/build.py

@@ -6,12 +6,10 @@
 from pathlib import Path, PurePath
 from typing import Dict, List, Type
 
-from smb.SMBConnection import SMBConnection
-
 from lisa import schema
 from lisa.node import quick_connect
 from lisa.tools import Ls, RemoteCopy
-from lisa.util import ContextMixin, InitializableMixin, subclasses
+from lisa.util import ContextMixin, InitializableMixin, LisaException, subclasses
 from lisa.util.logger import get_logger
 
 from .schema import BuildSchema, FileSchema, SMBBuildSchema, TftpBuildSchema
@@ -80,6 +78,14 @@ def copy(self, sources_path: List[Path], files_map: List[FileSchema]) -> None:
         server_name = self.smb_runbook.server_name
         share_name = self.smb_runbook.share
 
+        try:
+            from smb.SMBConnection import SMBConnection
+        except ModuleNotFoundError as e:
+            raise LisaException(
+                "SMB build requires 'pysmb'. Install baremetal extras or "
+                "run 'python -m pip install pysmb'."
+            ) from e
+
         with SMBConnection(
             username=username,
             password=password,

lisa/sut_orchestrator/baremetal/cluster/cluster.py

@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
 
-from typing import Any, Type
+from typing import Any, List, Type
 
 from lisa import features, schema, search_space
 from lisa.environment import Environment
@@ -55,20 +55,22 @@ def cleanup(self) -> None:
         pass
 
     def prepare_clients(self) -> None:
-        client_runbook = self.runbook.client[0]
-        client_capability = self.get_client_capability(client_runbook)
+        self.clients: List[schema.Capability] = []
+        for client_runbook in self.runbook.client:
+            client_capability = self.get_client_capability(client_runbook)
 
-        # to compatible with previous schema, use the whole client as extended
-        # runbook.
-        schema_type = self.runbook.type
-        extended_schema = client_runbook.to_dict()
+            # to compatible with previous schema, use the whole client as extended
+            # runbook.
+            schema_type = self.runbook.type
+            extended_schema = client_runbook.to_dict()
 
-        if client_capability.extended_schemas is None:
-            client_capability.extended_schemas = {}
-        client_capability.extended_schemas[schema_type] = extended_schema
-        self._fill_capability(client_capability)
+            if client_capability.extended_schemas is None:
+                client_capability.extended_schemas = {}
+            client_capability.extended_schemas[schema_type] = extended_schema
+            self._fill_capability(client_capability)
+            self.clients.append(client_capability)
 
-        self.client = client_capability
+        self.client = self.clients[0]
 
     def _initialize(self, *args: Any, **kwargs: Any) -> None:
         self.prepare_clients()
@@ -94,6 +96,7 @@ def _fill_capability(self, node_capability: schema.NodeSpace) -> None:
             items=[
                 schema.FeatureSettings.create(features.SerialConsole.name()),
                 schema.FeatureSettings.create(features.StartStop.name()),
+                features.NvmeSettings(),
                 SecurityProfile.create_setting(),
             ],
         )

lisa/sut_orchestrator/baremetal/cluster/rackmanager.py

@@ -1,15 +1,192 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
-from typing import Any, Type
+import re
+import time
+from pathlib import Path
+from typing import Any, Optional, Type, cast
 
 from lisa import features, schema
 from lisa.environment import Environment
-from lisa.node import quick_connect
+from lisa.node import Node, quick_connect
+from lisa.platform_ import Platform
+from lisa.schema import FeatureSettings
+from lisa.util import LisaException
 
+from ..context import get_node_context
 from ..platform_ import BareMetalPlatform
-from ..schema import RackManagerSchema
+from ..schema import RackManagerClientSchema, RackManagerSchema
 from .cluster import Cluster
 
+SERIAL_LOGIN_TIMEOUT = 300
+SERIAL_PASSWORD_TIMEOUT = 10
+SERIAL_LOGIN_RETRY_TIMEOUT = 3
+SERIAL_PROMPT_WAKE_TIMEOUT = 5
+SERIAL_PROMPT_WAKE_ATTEMPTS = 3
+
+
+class RackManagerSerialConsole(features.SerialConsole):
+    panic_ignorable_patterns = features.SerialConsole.panic_ignorable_patterns + [
+        re.compile(r"^(.*firmware bug.*)$", re.MULTILINE | re.IGNORECASE),
+    ]
+
+    def __init__(
+        self, settings: FeatureSettings, node: Node, platform: Platform
+    ) -> None:
+        super().__init__(settings, node, platform)
+        self._process: Any = None
+        self._username = ""
+        self._password = ""
+        self._cluster: Optional["RackManager"] = None
+        self._management_port: Optional[int] = None
+
+    def read(self) -> str:
+        return self._get_console_log(saved_path=None).decode("utf-8", errors="ignore")
+
+    def wait_for_ready(self, timeout: int) -> None:
+        self._login(timeout)
+
+    def write(self, data: str) -> None:
+        self._login()
+        self._input(f"{data}\n")
+
+    def close(self) -> None:
+        if self._process:
+            self._process.kill()
+
+    def _get_console_log(self, saved_path: Optional[Path]) -> bytes:
+        self._process.wait_output("", timeout=1, error_on_missing=False, interval=0.1)
+        output = self._process.log_buffer.getvalue()
+        if isinstance(output, bytes):
+            return output
+        return str(output).encode("utf-8")
+
+    def _initialize(self, *args: Any, **kwargs: Any) -> None:
+        super()._initialize(*args, **kwargs)
+        platform = cast(BareMetalPlatform, self._platform)
+        cluster = cast(RackManager, platform.cluster)
+        cluster.connect_to_rack_manager()
+        self._cluster = cluster
+
+        context = get_node_context(self._node)
+        client = cast(RackManagerClientSchema, context.client)
+        assert (
+            client.management_port is not None and client.management_port >= 0
+        ), "management_port is required for rackmanager serial console"
+        assert client.connection, "client connection is required for serial login"
+
+        self._management_port = client.management_port
+        self._username = client.connection.username
+        self._password = client.connection.password
+        self._start_serial_session()
+
+    def _start_serial_session(self) -> None:
+        assert self._cluster, "rackmanager cluster is required for serial console"
+        assert self._management_port is not None, (
+            "management port is required for serial console"
+        )
+        self._process = self._cluster.rm_node.execute_async(
+            f"start serial session -i {self._management_port}"
+        )
+
+    def _ensure_serial_session(self) -> None:
+        if self._process and self._process.is_running():
+            return
+        self._log.info("reconnecting Rack Manager serial session")
+        self._start_serial_session()
+
+    def _get_prompt_state(self, output_offset: int = 0) -> str:
+        output = self._get_console_log(saved_path=None).decode("utf-8", errors="ignore")
+        output = output[output_offset:]
+        prompt = output.rstrip()
+        if prompt.endswith(("$", "#")):
+            return "shell"
+        if prompt.lower().endswith("login:"):
+            return "login"
+        if prompt.lower().endswith("password:"):
+            return "password"
+        return "unknown" if prompt else "empty"
+
+    def _wait_for_prompt_state(self, timeout: int, output_offset: int = 0) -> str:
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            state = self._get_prompt_state(output_offset)
+            if state in ("shell", "login", "password"):
+                return state
+        return self._get_prompt_state(output_offset)
+
+    def _input(self, content: str, is_log_input: bool = True) -> None:
+        if not self._process.is_running():
+            self._raise_serial_process_exit()
+        try:
+            self._process.input(content, is_log_input=is_log_input)
+        except OSError as error:
+            self._raise_serial_process_exit(error)
+
+    def _raise_serial_process_exit(self, cause: Optional[OSError] = None) -> None:
+        result = self._process.wait_result()
+        message = (
+            "Rack Manager serial session exited before input could be sent. "
+            f"Exit code: {result.exit_code}. "
+            f"stdout: {result.stdout!r}. stderr: {result.stderr!r}."
+        )
+        if cause:
+            raise LisaException(message) from cause
+        raise LisaException(message)
+
+    def _login(self, timeout: int = SERIAL_LOGIN_TIMEOUT) -> None:
+        self._ensure_serial_session()
+        prompt_state = self._get_prompt_state()
+        if prompt_state == "shell":
+            return
+
+        for _ in range(SERIAL_PROMPT_WAKE_ATTEMPTS):
+            if prompt_state != "empty" and prompt_state != "unknown":
+                break
+            self._input("\n")
+            prompt_state = self._wait_for_prompt_state(SERIAL_PROMPT_WAKE_TIMEOUT)
+
+        if prompt_state == "shell":
+            return
+        if prompt_state in ("empty", "unknown"):
+            prompt_state = self._wait_for_prompt_state(timeout)
+        if prompt_state in ("empty", "unknown"):
+            raise LisaException(
+                "serial console produced no recognizable shell, login, or password "
+                "prompt after wake attempts"
+            )
+
+        output_offset = len(
+            self._get_console_log(saved_path=None).decode("utf-8", errors="ignore")
+        )
+        if prompt_state == "login":
+            self._input(f"{self._username}\n")
+            password_found = self._process.wait_output(
+                "Password:",
+                timeout=SERIAL_PASSWORD_TIMEOUT,
+                error_on_missing=False,
+                interval=0.5,
+                delta_only=True,
+            )
+            if not password_found:
+                raise LisaException("serial console password prompt was not found")
+            prompt_state = "password"
+        if prompt_state == "password":
+            output_offset = len(
+                self._get_console_log(saved_path=None).decode("utf-8", errors="ignore")
+            )
+            self._input(f"{self._password}\n", is_log_input=False)
+
+        deadline = time.time() + SERIAL_LOGIN_RETRY_TIMEOUT
+        while time.time() < deadline:
+            output = self._get_console_log(saved_path=None).decode(
+                "utf-8", errors="ignore"
+            )
+            if re.search(r"(?im)^.*login:\s*$", output[output_offset:]):
+                raise LisaException(
+                    "serial console login failed and returned to login prompt"
+                )
+            time.sleep(0.5)
+
 
 class RackManagerStartStop(features.StartStop):
     def init_rack_manager(self) -> None:
@@ -51,16 +228,47 @@ def type_schema(cls) -> Type[schema.TypedSchema]:
     def get_start_stop(self) -> Type[features.StartStop]:
         return RackManagerStartStop
 
+    def get_serial_console(self) -> Type[features.SerialConsole]:
+        return RackManagerSerialConsole
+
     def connect_to_rack_manager(self) -> None:
         assert self.rm_runbook.connection, "connection is required for rackmanager"
         self.rm_runbook.connection.name = "rackmanager"
         self.rm_node = quick_connect(
             self.rm_runbook.connection, logger_name="rackmanager"
         )
 
+    def _set_boot_from_remote_drive(self, management_port: int) -> None:
+        self.rm_node.execute(
+            f"set system boot -b 0 -m 1 -p 0 -i {management_port} -t 5"
+        ).assert_exit_code()
+
+    def _mount_remote_drive(self, management_port: int, iso_name: str) -> None:
+        self.rm_node.execute(
+            "set system remotedrive mount -b 0 -m 2 "
+            f"-i {management_port} -n {iso_name}"
+        ).assert_exit_code()
+
+    def _reset_client(self, management_port: int) -> None:
+        self.rm_node.execute(f"set sys reset -i {management_port}").assert_exit_code()
+
     def deploy(self, environment: Environment) -> Any:
-        self.reset("off")
-        self.reset("on")
+        clients = self.rm_runbook.client
+        assert clients, "client is required for rackmanager"
+
+        iso_clients = [client for client in clients if client.iso_name]
+        if not iso_clients:
+            return
+
+        self.connect_to_rack_manager()
+        for client in iso_clients:
+            management_port = client.management_port
+            assert (
+                management_port
+            ), "management_port is required when rackmanager iso_name is set"
+            self._set_boot_from_remote_drive(management_port)
+            self._mount_remote_drive(management_port, client.iso_name)
+            self._reset_client(management_port)
 
     def reset(self, operation: str) -> None:
         self.connect_to_rack_manager()