Replace check_cuda.sh with an appropriate Python script (New)

contrib/checkbox-dss-validation/checkbox-provider-dss/bin/check_cuda_with_mk8s.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+#
+# This file is part of Checkbox.
+#
+# Copyright 2022 Canonical Ltd.
+#
+# Authors:
+#     Abdullah (@motjuste) <abdullah.abdullah@canonical.com>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
+#
+"""Check enabling CUDA with microk8s"""
+
+import time
+import typing as t
+
+from common import create_parser_with_checks_as_commands, run_command
+
+
+def parse_args(args: t.List[str] | None = None) -> dict[str, t.Any]:
+    parser = create_parser_with_checks_as_commands(
+        [
+            can_be_enabled_with_operator_version,
+            has_all_validations_successful,
+        ],
+        description="Check enabling CUDA with microk8s",
+    )
+    return dict(parser.parse_args(args).__dict__)
+
+
+def can_be_enabled_with_operator_version(operator_version: str) -> None:
+    """Verify enabling CUDA addon in microk8s of given operator_version"""
+    result = run_command(
+        "sudo",
+        "microk8s",
+        "enable",
+        "gpu",
+        "--driver=operator",
+        f"--version={operator_version}",
+    )
+    was_enabled_mk8s_in_1_28 = "NVIDIA is enabled" in result
+    was_enabled_mk8s_in_1_31 = "Deployed NVIDIA GPU operator" in result
+    nvidia_was_enabled = was_enabled_mk8s_in_1_28 or was_enabled_mk8s_in_1_31
+    gpu_was_already_enabled = "gpu is already enabled" in result
+    if not (nvidia_was_enabled or gpu_was_already_enabled):
+        raise AssertionError("Couldn't verify if NVIDIA GPU was or is enabled")
+    verify_all_rollouts()
+
+
+def has_all_validations_successful() -> None:
+    """Verify that all NVIDIA operator validations were successful"""
+    result = run_command(
+        "kubectl",
+        "logs",
+        "-n",
+        "gpu-operator-resources",
+        "-lapp=nvidia-operator-validator",
+        "-c",
+        "nvidia-operator-validator",
+    )
+    if "all validations are successful" not in result:
+        raise AssertionError("Couldn't verify that GPU validations succeeded")
+
+
+def verify_all_rollouts():
+    verify_node_feature_discovery_rollout()
+    verify_plugin_rollout()
+    verify_validator_rollout()
+
+
+def verify_node_feature_discovery_rollout():
+    verify_rollout_of_daemonset("gpu-operator-node-feature-discovery-worker")
+
+
+def verify_plugin_rollout():
+    verify_rollout_of_daemonset("nvidia-device-plugin-daemonset")
+
+
+def verify_validator_rollout():
+    verify_rollout_of_daemonset("nvidia-operator-validator")
+
+
+def verify_rollout_of_daemonset(daemonset: str):
+    print(f"sleeping for 10 seconds before checking rollout of {daemonset}")
+    time.sleep(10)
+    result = run_command(
+        "kubectl",
+        "-n",
+        "gpu-operator-resources",
+        "rollout",
+        "status",
+        f"ds/{daemonset}",
+    )
+    expected = f'daemon set "{daemonset}" successfully rolled out'
+    if expected not in result:
+        raise AssertionError(f"Couldn't verify rollout of {daemonset}")
+
+
+def main(args: t.List[str] | None = None) -> None:
+    parsed = parse_args(args)
+    parsed.pop("func")(**parsed)
+
+
+if __name__ == "__main__":
+    main()

contrib/checkbox-dss-validation/checkbox-provider-dss/bin/common.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+#
+# This file is part of Checkbox.
+#
+# Copyright 2022 Canonical Ltd.
+#
+# Authors:
+#     Abdullah (@motjuste) <abdullah.abdullah@canonical.com>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
+#
+"""Common utilities for the checkbox-provider-dss"""
+
+import argparse
+import inspect
+import subprocess
+import typing as t
+
+
+def create_parser_with_checks_as_commands(
+    checks: t.List[t.Callable], **kwargs
+) -> argparse.ArgumentParser:
+    if len(checks) <= 0:
+        raise AssertionError("must provide at least one check")
+
+    if len(checks) != len(set(checks)):
+        # NOTE:@motjuste: Python 3.10 does not de-duplicate sub parser commands
+        raise AssertionError("duplicate checks are not allowed")
+
+    parser = argparse.ArgumentParser(
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter, **kwargs
+    )
+    sub_parsers = parser.add_subparsers(required=True)
+
+    for check in checks:
+        command_parser = sub_parsers.add_parser(
+            check.__name__, description=check.__doc__, help=check.__doc__
+        )
+        command_parser.set_defaults(func=check)
+        for name, param in inspect.signature(check).parameters.items():
+            assert param.kind == inspect.Parameter.POSITIONAL_OR_KEYWORD
+            assert param.annotation is not bool
+            assert param.default is inspect._empty
+
+            command_parser.add_argument(name, type=param.annotation)
+
+    return parser
+
+
+def run_command(*command: str, **kwargs) -> str:
+    """Run a shell command and return its output"""
+    try:
+        result = subprocess.check_output(
+            command,
+            stderr=subprocess.STDOUT,  # We capture stdout and stderr in stdout
+            universal_newlines=True,
+            **kwargs,
+        )
+        print(result)
+        result = result.strip()
+        return result
+    except subprocess.CalledProcessError as err:
+        print(err.stdout)
+        print(err.stderr)
+        raise