Add tftp deployment on BareMetal platform

lisa/sut_orchestrator/baremetal/bootconfig.py

@@ -0,0 +1,170 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+from dataclasses import dataclass, field
+from typing import Optional, Type
+
+from dataclasses_json import dataclass_json
+from marshmallow import validate
+
+from lisa import schema
+from lisa.node import quick_connect
+from lisa.util import InitializableMixin, LisaException, field_metadata, subclasses
+from lisa.util.logger import get_logger
+
+from .schema import BootConfigSchema
+
+OV_CP = "cp"
+OV_GP = "gp"
+
+OV_DEV = "dev"
+OV_PIPELINE = "pipeline"
+
+
+class BootConfig(subclasses.BaseClassWithRunbookMixin, InitializableMixin):
+    def __init__(
+        self,
+        runbook: BootConfigSchema,
+    ) -> None:
+        super().__init__(runbook=runbook)
+        self.boot_config_runbook: BootConfigSchema = self.runbook
+        self._log = get_logger("boot_config", self.__class__.__name__)
+
+    @classmethod
+    def type_schema(cls) -> Type[schema.TypedSchema]:
+        return BootConfigSchema
+
+    def set_boot_config(self) -> None:
+        raise NotImplementedError()
+
+
+@dataclass_json()
+@dataclass
+class PxeBootSchema(BootConfigSchema):
+    connection: Optional[schema.RemoteNode] = field(
+        default=None, metadata=field_metadata(required=True)
+    )
+    node_setup: str = field(
+        default=OV_DEV,
+        metadata=field_metadata(
+            validate=validate.OneOf(
+                [
+                    OV_DEV,
+                    OV_PIPELINE,
+                ]
+            )
+        ),
+    )
+    node_type: str = field(
+        default=OV_CP,
+        metadata=field_metadata(
+            validate=validate.OneOf(
+                [
+                    OV_CP,
+                    OV_GP,
+                ]
+            )
+        ),
+    )
+    node_number: int = field(
+        default=1, metadata=field_metadata(validate=validate.Range(min=1, max=20))
+    )
+    image_source: str = field(default="", metadata=field_metadata(required=True))
+    kernel_append: Optional[str] = field(default="")
+
+
+class PxeBoot(BootConfig):
+    def __init__(
+        self,
+        runbook: PxeBootSchema,
+    ) -> None:
+        super().__init__(runbook=runbook)
+        self.pxe_runbook: PxeBootSchema = self.runbook
+        self._log = get_logger("pxe_boot", self.__class__.__name__)
+        self._boot_label = "label lisav3"
+        self._boot_dir = "/var/lib/tftpboot/"
+        self._boot_config = self._boot_dir + "pxelinux.cfg/"
+
+    @classmethod
+    def type_name(cls) -> str:
+        return "pxe_boot"
+
+    @classmethod
+    def type_schema(cls) -> Type[schema.TypedSchema]:
+        return PxeBootSchema
+
+    def set_boot_config(self) -> None:
+        boot_type = f"{self.pxe_runbook.node_setup}-{self.pxe_runbook.node_type}"
+        boot_name = f"{boot_type}{self.pxe_runbook.node_number}"
+        host_ref = f"host {boot_name}"
+
+        self._connect_to_pxe_server()
+        pxe_boot_config_path = self._get_dhcp_info(boot_name, host_ref)
+        if not pxe_boot_config_path:
+            raise LisaException(f"Failed to find DHCP entry for {boot_name}")
+
+        boot_image = self.pxe_runbook.image_source.replace(
+            self._boot_dir,
+            "",
+        ).strip("/")
+        boot_entry = f"kernel {boot_image}"
+
+        # Delete it if one is there, so to always insert at the top
+        self.pxe_server.execute(
+            cmd=f"sed -i '/^{self._boot_label}$/,/^$/d' " f"{pxe_boot_config_path}",
+            expected_exit_code_failure_message=(
+                f"Failed to delete previous boot entry on {boot_name}"
+            ),
+            expected_exit_code=0,
+            sudo=False,
+        )
+
+        self._log.debug(
+            "Adding boot entry for LISA at "
+            f"{pxe_boot_config_path}, "
+            f"pointing to {boot_entry}"
+        )
+
+        entry = self.pxe_runbook.kernel_append
+        append = f"\\n    append {entry}" if entry else ""
+        label = self._boot_label
+        self.pxe_server.execute(
+            cmd=f"sed -i '/^menu.*/a \\\\n{label}\\n    {boot_entry}{append}' "
+            f"{pxe_boot_config_path}",
+            expected_exit_code_failure_message=(
+                f"Failed to create new boot entry on {boot_name}"
+            ),
+            expected_exit_code=0,
+            sudo=False,
+        )
+
+        self._log.debug(
+            "Added boot entry for LISA at "
+            f"{pxe_boot_config_path}, "
+            f"pointing to {boot_entry}"
+        )
+
+    def _connect_to_pxe_server(self) -> None:
+        assert self.pxe_runbook.connection, "connection is required for pxe_server"
+        self.pxe_runbook.connection.name = "pxe_server"
+        self.pxe_server = quick_connect(
+            self.pxe_runbook.connection, logger_name="pxe_server"
+        )
+
+    def _get_dhcp_info(self, node_name: str, host_ref: str) -> str:
+        output_dhcp_info = self.pxe_server.execute(
+            cmd="cat /etc/dhcp/dhcpd.conf",
+            expected_exit_code_failure_message=(
+                f"failed to obtain {node_name}'s configuration information\n"
+            ),
+            expected_exit_code=0,
+            sudo=False,
+        ).stdout
+
+        board_info = output_dhcp_info[output_dhcp_info.index(host_ref) :]
+        config_info = board_info[board_info.index("{") : board_info.index("}") + 1]
+        start = config_info.index("ethernet") + len("ethernet ")
+        end = config_info.index(";")
+        address = "01-" + config_info[start:end].replace(":", "-").strip()
+        address_path = self._boot_config + address
+        return address_path

lisa/sut_orchestrator/baremetal/build.py

@@ -9,10 +9,35 @@
 from smb.SMBConnection import SMBConnection  # type: ignore
 
 from lisa import schema
+from lisa.node import quick_connect
+from lisa.tools import Ls
 from lisa.util import ContextMixin, InitializableMixin, subclasses
 from lisa.util.logger import get_logger
 
-from .schema import BuildSchema, FileSchema, SMBBuildSchema
+from .schema import BuildSchema, FileSchema, SMBBuildSchema, TftpBuildSchema
+
+
+def find_matched_files(
+    sources_path: List[Path],
+    files_map: List[FileSchema],
+) -> Dict[str, FileSchema]:
+    all_files = []
+    match_files: Dict[str, FileSchema] = {}
+    for source_path in sources_path:
+        for root, _, files in os.walk(source_path):
+            for file in files:
+                all_files.append(os.path.join(root, file))
+
+        for file_map in files_map:
+            file_path = rf"{source_path}\{file_map.source}".replace("\\", "\\\\")
+            pattern = re.compile(
+                file_path,
+                re.I | re.M,
+            )
+            for file in all_files:
+                if pattern.match(file):
+                    match_files[file] = file_map
+    return match_files
 
 
 class Build(subclasses.BaseClassWithRunbookMixin, ContextMixin, InitializableMixin):
@@ -55,8 +80,9 @@ def copy(self, sources_path: List[Path], files_map: List[FileSchema]) -> None:
         ) as conn:
             conn.connect(server_name)
 
-            for file, file_map in self._find_matched_files(
-                sources_path, files_map
+            for file, file_map in find_matched_files(
+                sources_path,
+                files_map,
             ).items():
                 with open(file, "rb") as f:
                     if file_map.destination:
@@ -79,23 +105,51 @@ def copy(self, sources_path: List[Path], files_map: List[FileSchema]) -> None:
                         )
                 self._log.debug(f"copy file {file} to {share_name}\\{file_name}")
 
-    def _find_matched_files(
-        self, sources_path: List[Path], files_map: List[FileSchema]
-    ) -> Dict[str, FileSchema]:
-        all_files = []
-        match_files: Dict[str, FileSchema] = {}
-        for source_path in sources_path:
-            for root, _, files in os.walk(source_path):
-                for file in files:
-                    all_files.append(os.path.join(root, file))
-
-            for file_map in files_map:
-                file_path = rf"{source_path}\{file_map.source}".replace("\\", "\\\\")
-                pattern = re.compile(
-                    file_path,
-                    re.I | re.M,
+
+class TftpBuild(Build):
+    def __init__(self, runbook: TftpBuildSchema) -> None:
+        super().__init__(runbook)
+        self.pxe_runbook: TftpBuildSchema = self.runbook
+
+    @classmethod
+    def type_name(cls) -> str:
+        return "tftp"
+
+    @classmethod
+    def type_schema(cls) -> Type[schema.TypedSchema]:
+        return TftpBuildSchema
+
+    def copy(self, sources_path: List[Path], files_map: List[FileSchema]) -> None:
+        assert self.pxe_runbook.connection, "The build server is not specified"
+
+        try:
+            build_server = quick_connect(
+                self.pxe_runbook.connection,
+                logger_name="build_server",
+            )
+        except Exception as ex:
+            self._log.debug(
+                f"failed to connect {self.pxe_runbook.connection}",
+                exc_info=ex,
+            )
+
+        ls = build_server.tools[Ls]
+        for file, file_map in find_matched_files(
+            sources_path,
+            files_map,
+        ).items():
+            if file_map.destination:
+                if ls.is_file(Path(file_map.destination)):
+                    file_name = file_map.destination
+                else:
+                    file_name = file_map.destination + "/" + file.rsplit("\\")[-1]
+            else:
+                file_name = file.rsplit("\\")[-1]
+
+            try:
+                build_server.shell.copy(Path(file), Path(file_name))
+            except Exception as ex:
+                self._log.debug(
+                    f"failed to copy {file} to {file_name}",
+                    exc_info=ex,
                 )
-                for file in all_files:
-                    if pattern.match(file):
-                        match_files[file] = file_map
-        return match_files