Merge branch 'main' into m-bridge

Author: srivatsankrishnan

src/cloudai/_core/json_gen_strategy.py

@@ -1,5 +1,5 @@
 # SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
-# Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -55,8 +55,13 @@ def sanitize_k8s_job_name(self, job_name: str) -> str:
         sanitized_name = job_name.lower()
         sanitized_name = re.sub(r"[^a-z0-9-]", "-", sanitized_name)
         sanitized_name = re.sub(r"^[^a-z0-9]+", "", sanitized_name)
+        sanitized_name = sanitized_name[:253]
         sanitized_name = re.sub(r"[^a-z0-9]+$", "", sanitized_name)
-        return sanitized_name[:253]
+
+        if not sanitized_name:
+            raise ValueError(f"'{job_name}' cannot be sanitized to a valid Kubernetes job name.")
+
+        return sanitized_name
 
     def store_test_run(self) -> None:
         from cloudai.models.scenario import TestRunDetails

src/cloudai/systems/kubernetes/kubernetes_installer.py

@@ -1,5 +1,5 @@
 # SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
-# Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -118,6 +118,13 @@ def is_installed_one(self, item: Installable) -> InstallStatusResult:
             return InstallStatusResult(False, f"Git repository {item.url} not cloned")
         elif isinstance(item, PythonExecutable):
             return self._is_python_executable_installed(item)
+        elif isinstance(item, File):
+            if (self.system.install_path / item.src.name).exists() and (
+                self.system.install_path / item.src.name
+            ).read_text() == item.src.read_text():
+                item.installed_path = self.system.install_path / item.src.name
+                return InstallStatusResult(True)
+            return InstallStatusResult(False, f"File {self.system.install_path / item.src.name} does not exist")
         elif isinstance(item, HFModel):
             return self.hf_model_manager.is_model_downloaded(item)
         return InstallStatusResult(False, f"Unsupported item type: {type(item)}")
@@ -132,6 +139,9 @@ def mark_as_installed_one(self, item: Installable) -> InstallStatusResult:
             item.git_repo.installed_path = self.system.install_path / item.git_repo.repo_name
             item.venv_path = self.system.install_path / item.venv_name
             return InstallStatusResult(True)
+        elif isinstance(item, File):
+            item.installed_path = self.system.install_path / item.src.name
+            return InstallStatusResult(True)
         elif isinstance(item, HFModel):
             item.installed_path = self.system.hf_home_path  # fake path is OK here as the whole HF home will be mounted
             return InstallStatusResult(True)

src/cloudai/systems/kubernetes/kubernetes_system.py

@@ -1,5 +1,5 @@
 # SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
-# Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -431,13 +431,25 @@ def _delete_mpi_job(self, job_name: str) -> None:
 
     def _delete_batch_job(self, job_name: str) -> None:
         logging.debug(f"Deleting batch job '{job_name}'")
-        api_response = self.batch_v1.delete_namespaced_job(
-            name=job_name,
-            namespace=self.default_namespace,
-            body=lazy.k8s.client.V1DeleteOptions(propagation_policy="Foreground", grace_period_seconds=5),
-        )
-        api_response = cast("k8s.client.V1Job", api_response)
+        try:
+            api_response = self.batch_v1.delete_namespaced_job(
+                name=job_name,
+                namespace=self.default_namespace,
+                body=lazy.k8s.client.V1DeleteOptions(propagation_policy="Foreground", grace_period_seconds=5),
+            )
+        except lazy.k8s.client.ApiException as e:
+            if e.status == 404:
+                logging.debug(f"Batch job '{job_name}' not found. It may have already been deleted.")
+                return
+
+            logging.error(
+                f"An error occurred while attempting to delete batch job '{job_name}'. "
+                f"Error code: {e.status}. Message: {e.reason}. "
+                "Please verify the job name and Kubernetes API server."
+            )
+            raise
 
+        api_response = cast("k8s.client.V1Status", api_response)
         logging.debug(f"Batch job '{job_name}' deleted with status: {api_response.status}")
 
     def _delete_dynamo_graph_deployment(self, job_name: str) -> None:
@@ -662,7 +674,7 @@ def store_logs_for_job(self, job_name: str, output_dir: Path) -> None:
         """
         pod_names = self.get_pod_names_for_job(job_name)
         if not pod_names:
-            logging.warning(f"No pods found for job '{job_name}'")
+            logging.debug(f"No pods found for job '{job_name}'")
             return
 
         output_dir.mkdir(parents=True, exist_ok=True)

src/cloudai/workloads/sleep/kubernetes_json_gen_strategy.py

@@ -1,5 +1,5 @@
 # SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
-# Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -33,7 +33,10 @@ def gen_json(self) -> Dict[Any, Any]:
         job_spec = {
             "apiVersion": "batch/v1",
             "kind": "Job",
-            "metadata": {"name": self.test_run.name, "namespace": kubernetes_system.default_namespace},
+            "metadata": {
+                "name": self.sanitize_k8s_job_name(self.test_run.name),
+                "namespace": kubernetes_system.default_namespace,
+            },
             "spec": {
                 "ttlSecondsAfterFinished": 0,
                 "template": {

tests/json_gen_strategy/test_common_kubernetes.py

@@ -0,0 +1,52 @@
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import pytest
+
+from cloudai.core import JsonGenStrategy, TestRun
+from cloudai.systems.kubernetes import KubernetesSystem
+
+
+class MyJsonGenStrategy(JsonGenStrategy):
+    def gen_json(self) -> dict:
+        return {}
+
+
+@pytest.mark.parametrize(
+    "tname,expected",
+    [
+        ("simple-name", "simple-name"),
+        ("name_with_underscores", "name-with-underscores"),
+        ("name.with.dots", "name-with-dots"),
+        ("name@with#special$chars", "name-with-special-chars"),
+        ("NameWithUpperCase", "namewithuppercase"),
+        ("a" * 260, "a" * 253),
+        ("---leading-and-trailing---", "leading-and-trailing"),
+        ("a" * 250 + "-" * 3 + "b" * 10, "a" * 250),  # ensure no trailing hyphens on truncation
+    ],
+)
+def test_job_name_sanitization(k8s_system: KubernetesSystem, base_tr: TestRun, tname: str, expected: str) -> None:
+    base_tr.name = tname
+    json_gen = MyJsonGenStrategy(k8s_system, base_tr)
+    assert json_gen.sanitize_k8s_job_name(base_tr.name) == expected
+
+
+def test_job_name_sanitization_raises(k8s_system: KubernetesSystem, base_tr: TestRun) -> None:
+    base_tr.name = "!@#$%^&*()"
+    json_gen = MyJsonGenStrategy(k8s_system, base_tr)
+    with pytest.raises(ValueError):
+        json_gen.sanitize_k8s_job_name(base_tr.name)

tests/json_gen_strategy/test_sleep_kubernetes_json_gen_strategy.py

@@ -0,0 +1,28 @@
+# SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from cloudai.core import TestRun
+from cloudai.systems.kubernetes import KubernetesSystem
+from cloudai.workloads.sleep import SleepCmdArgs, SleepKubernetesJsonGenStrategy, SleepTestDefinition
+
+
+def test_job_name_sanitization(k8s_system: KubernetesSystem) -> None:
+    tdef = SleepTestDefinition(name="name", description="desc", test_template_name="tt", cmd_args=SleepCmdArgs())
+    tr = TestRun(name="t!e@st#-n$am%e^", test=tdef, nodes=["node1"], num_nodes=1)
+    json_gen = SleepKubernetesJsonGenStrategy(k8s_system, tr)
+
+    assert json_gen.gen_json()["metadata"]["name"] == json_gen.sanitize_k8s_job_name(tr.name)

tests/test_base_installer.py

@@ -1,5 +1,5 @@
 # SPDX-FileCopyrightText: NVIDIA CORPORATION & AFFILIATES
-# Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -331,9 +331,7 @@ def test_check_supported(installer: KubernetesInstaller | SlurmInstaller):
     installer.hf_model_manager = Mock()
 
     git = GitRepo(url="./git_url", commit="commit_hash")
-    items = [DockerImage("fake_url/img"), PythonExecutable(git), HFModel("model_name")]
-    if isinstance(installer, SlurmInstaller):
-        items.append(File(Path(__file__)))
+    items = [DockerImage("fake_url/img"), PythonExecutable(git), HFModel("model_name"), File(Path(__file__))]
     for item in items:
         res = installer.install_one(item)
         assert res.success, f"Failed to install {item} for {installer.__class__.__name__=} {res.message=}"