add scripts_execution_transformer

lisa/mixin_modules.py

@@ -78,5 +78,6 @@
 import lisa.transformers.repo_package_installer # noqa: F401
 import lisa.transformers.rpm_kernel_installer # noqa: F401
 import lisa.transformers.script_transformer # noqa: F401
+import lisa.transformers.scripts_execution_transformer # noqa: F401
 import lisa.transformers.to_list # noqa: F401
 import lisa.transformers.upgrade_packages # noqa: F401

lisa/transformers/scripts_execution_transformer.py

@@ -0,0 +1,95 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Dict, List, Optional, Type
+
+from dataclasses_json import dataclass_json
+
+from lisa import schema
+from lisa.transformers.deployment_transformer import (
+ DeploymentTransformer,
+ DeploymentTransformerSchema,
+)
+from lisa.util import LisaException
+
+
+class ScriptInterpreter(str, Enum):
+ BASH = "bash"
+
+
+@dataclass_json()
+@dataclass
+class ScriptEntry:
+ script: str = ""
+ interpreter: ScriptInterpreter = ScriptInterpreter.BASH
+ args: Optional[str] = None
+
+
+@dataclass_json()
+@dataclass
+class ScriptsExecutionTransformerSchema(DeploymentTransformerSchema):
+ dependent_packages: List[str] = field(default_factory=list)
+ scripts: List[ScriptEntry] = field(default_factory=list)
+ reboot: bool = field(default=False)
+ exit_on_error: bool = field(default=False)
+
+
+class ScriptsExecutionTransformer(DeploymentTransformer):
+ """
+ This Transformer is to execute scripts.
+
+ Sample runbook section:
+ - type: scripts_execution
+ phase: expanded
+ connection:
+ address: $(build_vm_address)
+ private_key_file: $(admin_private_key_file)
+ reboot: true
+ exit_on_error: true
+ dependent_packages:
+ - git
+ scripts:
+ - script: "/tmp/waagent.sh"
+ interpreter: bash
+ args: "--flag"
+ """
+
+ __results_name = "results"
+
+ @classmethod
+ def type_name(cls) -> str:
+ return "scripts_execution"
+
+ @classmethod
+ def type_schema(cls) -> Type[schema.TypedSchema]:
+ return ScriptsExecutionTransformerSchema
+
+ @property
+ def _output_names(self) -> List[str]:
+ return [self.__results_name]
+
+ def _internal_run(self) -> Dict[str, Any]:
+ runbook: ScriptsExecutionTransformerSchema = self.runbook
+ self._node.os.install_packages(runbook.dependent_packages) # type: ignore
+
+ results: Dict[str, Any] = {}
+ for item in runbook.scripts:
+ command = f"{item.interpreter} {item.script} {item.args}"
+ execution_result = self._node.execute(command, sudo=True, shell=True)
+ results[item.script] = execution_result
+
+ if runbook.exit_on_error:
+ failed_scripts = []
+ for script, execution_result in results.items():
+ if execution_result.exit_code != 0:
+ failed_scripts.append((script, execution_result.exit_code))
+ if failed_scripts:
+ raise LisaException(
+ f"Execution failed for scripts (name, exit_code): {failed_scripts}"
+ )
+
+ if runbook.reboot:
+ self._node.reboot()
+ return {self.__results_name: results}