Skip to content

Continuously fetch logs for NCCL over k8s #788

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
amaslenn merged 3 commits into from
Jan 27, 2026
Merged

Continuously fetch logs for NCCL over k8s #788

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
eed8e1a
Continously fetch logs from k8s for nccl
amaslenn Jan 26, 2026
163c8fc
Improve continous logs fetching logic
amaslenn Jan 26, 2026
35b1f7d
Address review comments
amaslenn Jan 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions src/cloudai/systems/kubernetes/kubernetes_system.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ def _is_job_running(self, job: KubernetesJob) -> bool:
logging.debug(f"Checking for job '{job.name}' of kind '{job.kind}' to determine if it is running.")

if "mpijob" in job.kind.lower():
return self._is_mpijob_running(job.name)
return self._is_mpijob_running(job)
elif "job" in job.kind.lower():
return self._is_batch_job_running(job.name)
elif "dynamographdeployment" in job.kind.lower():
Expand All @@ -153,20 +153,22 @@ def _is_job_running(self, job: KubernetesJob) -> bool:
logging.error(error_message)
raise ValueError(error_message)

def _is_mpijob_running(self, job_name: str) -> bool:
def _is_mpijob_running(self, job: KubernetesJob) -> bool:
try:
mpijob = self.custom_objects_api.get_namespaced_custom_object(
group="kubeflow.org",
version="v2beta1",
namespace=self.default_namespace,
plural="mpijobs",
name=job_name,
name=job.name,
)

assert isinstance(mpijob, dict)
status: dict = cast(dict, mpijob.get("status", {}))
conditions = status.get("conditions", [])
logging.debug(f"MPIJob '{job_name}': {conditions=} {status=}")
logging.debug(f"MPIJob '{job.name}': {conditions=} {status=}")

self.store_logs_for_job(job.name, job.test_run.output_path)

# Consider an empty conditions list as running
if not conditions:
Expand All @@ -183,11 +185,11 @@ def _is_mpijob_running(self, job_name: str) -> bool:

except lazy.k8s.client.ApiException as e:
if e.status == 404:
logging.debug(f"MPIJob '{job_name}' not found. It may have completed and been removed from the system.")
logging.debug(f"MPIJob '{job.name}' not found. It may have completed and been removed from the system.")
return False
else:
error_message = (
f"Error occurred while retrieving status for MPIJob '{job_name}' "
f"Error occurred while retrieving status for MPIJob '{job.name}' "
f"Error code: {e.status}. Message: {e.reason}. Please check the job name, namespace, and "
"Kubernetes API server."
)
Expand Down Expand Up @@ -392,6 +394,7 @@ def kill(self, job: BaseJob) -> None:
job (BaseJob): The job to be terminated.
"""
k_job: KubernetesJob = cast(KubernetesJob, job)
self.store_logs_for_job(k_job.name, k_job.test_run.output_path)
self.delete_job(k_job.name, k_job.kind)

def delete_job(self, job_name: str, job_kind: str) -> None:
Expand Down