Create CPU/Memory/FileIO perf test using sysbench

This commit has changes to add testcases for CPU/Memory/FileIO
using sysbench tool of LISA.

Signed-off-by: Smit Gardhariya <[email protected]>

microsoft/testsuites/sysbench/sysbenchperf.py

@@ -0,0 +1,187 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+import json
+from pathlib import Path
+from typing import Any, Dict, List
+
+from assertpy.assertpy import assert_that, fail
+
+from lisa import (
+    Environment,
+    Logger,
+    Node,
+    TestCaseMetadata,
+    TestSuite,
+    TestSuiteMetadata,
+    notifier,
+)
+from lisa.messages import TestStatus, send_sub_test_result_message
+from lisa.testsuite import TestResult
+from lisa.tools import Sysbench
+from lisa.tools.sysbench import SysbenchTestResult
+
+
+@TestSuiteMetadata(
+    area="sysbench",
+    category="performance",
+    description="""
+    This test suite is for executing the sysbench tests
+    """,
+)
+class SysbenchTestSuite(TestSuite):
+    @TestCaseMetadata(
+        description="""
+            Runs Sysbench test for cpu
+        """,
+        priority=3,
+    )
+    def perf_sysbench_cpu(
+        self,
+        log: Logger,
+        node: Node,
+        environment: Environment,
+        log_path: Path,
+        result: TestResult,
+        variables: Dict[str, Any],
+    ) -> None:
+        subtest_name: str = "default_cpu_perf_check"
+        message = ""
+        status = TestStatus.QUEUED
+        output: SysbenchTestResult = node.tools[Sysbench].run_cpu_perf()
+
+        if output.exit_code != 0:
+            message = f"{subtest_name} failed with exit code {output.exit_code}"
+            status = TestStatus.FAILED
+        else:
+            message = json.dumps(output.result)
+            status = TestStatus.PASSED
+
+        self._send_subtest_msg(
+            result,
+            subtest_name,
+            status,
+            message,
+        )
+
+        msg = f"Sysbench-cpu test failed with exit code {output.exit_code}"
+        assert_that(status).described_as(msg).is_equal_to(TestStatus.PASSED)
+
+    @TestCaseMetadata(
+        description="""
+            Runs Sysbench test for fileio
+        """,
+        priority=3,
+    )
+    def perf_sysbench_fileio(
+        self,
+        log: Logger,
+        node: Node,
+        environment: Environment,
+        log_path: Path,
+        result: TestResult,
+        variables: Dict[str, Any],
+    ) -> None:
+        io_mode_ops = {
+            "seqwr": "write",
+            "seqrd": "read",
+            "rndrd": "read",
+            "rndwr": "write",
+            "seqrewr": "all",
+            "rndrw": "all",
+        }
+        sysbench = node.tools[Sysbench]
+        failure = []
+        for mode in io_mode_ops.keys():
+            subtest_name: str = f"{mode}_default_fileio_perf_check"
+            message = ""
+            status = TestStatus.QUEUED
+            output: SysbenchTestResult = sysbench.run_fileio_perf(
+                test_mode=mode,
+                total_file=1,
+                ops=io_mode_ops[mode],
+            )
+            if output.exit_code != 0:
+                failure.append(subtest_name)
+                message = (
+                    f"{subtest_name} failed with exit code {output.exit_code}"
+                )
+                status = TestStatus.FAILED
+            else:
+                message = json.dumps(output.result)
+                status = TestStatus.PASSED
+
+            self._send_subtest_msg(
+                result,
+                subtest_name,
+                status,
+                message,
+            )
+
+        if len(failure) > 0:
+            fail(f"Sysbench-fileio test failed: {failure}")
+
+    @TestCaseMetadata(
+        description="""
+            Runs Sysbench test for memory
+        """,
+        priority=3,
+    )
+    def perf_sysbench_memory(
+        self,
+        log: Logger,
+        node: Node,
+        environment: Environment,
+        log_path: Path,
+        result: TestResult,
+        variables: Dict[str, Any],
+    ) -> None:
+        memory_operation: List[str] = ["read", "write"]
+        memory_access_mode: List[str] = ["seq", "rnd"]
+        failure = []
+        sysbench = node.tools[Sysbench]
+
+        for op in memory_operation:
+            for access_mode in memory_access_mode:
+                message = ""
+                status = TestStatus.QUEUED
+                subtest_name: str = f"{access_mode}_{op}_default_memory_perf_check"
+                output: SysbenchTestResult = sysbench.run_memory_perf(
+                    memory_access_mode=access_mode,
+                    memory_oper=op,
+                )
+
+                if output.exit_code != 0:
+                    failure.append(subtest_name)
+                    message = (
+                        f"{subtest_name} failed with exit code {output.exit_code}"
+                    )
+                    status = TestStatus.FAILED
+                else:
+                    message = json.dumps(output.result)
+                    status = TestStatus.PASSED
+
+                self._send_subtest_msg(
+                    result,
+                    subtest_name,
+                    status,
+                    message,
+                )
+
+        if len(failure) > 0:
+            fail(f"Sysbench-memory test failed: {failure}")
+
+    def _send_subtest_msg(
+        self,
+        test_result: TestResult,
+        test_name: str,
+        test_status: TestStatus,
+        test_message: str = "",
+    ) -> None:
+        subtest_msg = send_sub_test_result_message(
+            test_result=test_result,
+            test_case_name=test_name,
+            test_status=test_status,
+            test_message=test_message,
+        )
+
+        notifier.notify(subtest_msg)