Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

Greptile Overview

Greptile Summary

Implements continuous log fetching for MPIJobs on Kubernetes to prevent log loss when the launcher is removed before logs are collected. The _is_mpijob_running method now accepts the full job object and calls store_logs_for_job during each polling interval.

Key Changes:

• Modified _is_mpijob_running signature to accept KubernetesJob instead of just job name
• Added log fetching during polling in _is_mpijob_running (line 171)
• Added log fetching in kill method before job deletion (line 397)
• Addressed previous review concern about empty conditions by placing log fetch before the empty check

Issues Found:

• I/O overhead: Logs are fetched every second (default monitor_interval=1) and files are overwritten, resulting in ~600 write operations for a 10-minute job
• Redundant log fetching: Logs are fetched in the final _is_mpijob_running call that detects completion, then fetched again in on_job_completion (kubernetes_runner.py:55)

Confidence Score: 3/5

• This PR is safe to merge with moderate risk - the core functionality is sound but has efficiency concerns.
• The implementation successfully addresses the primary goal of preventing log loss when launchers are removed early. However, the continuous log fetching every second introduces significant I/O overhead (file overwrites ~600 times for a 10-minute job) and there's redundant log fetching at job completion. These are performance/efficiency concerns rather than correctness issues. The changes are isolated to MPIJob handling and won't affect other job types.
• Pay close attention to src/cloudai/systems/kubernetes/kubernetes_system.py for potential I/O overhead in production environments with long-running jobs.

Important Files Changed

sequenceDiagram
    participant Runner as BaseRunner
    participant System as KubernetesSystem
    participant K8sAPI as Kubernetes API
    participant FS as File System

    Note over Runner,FS: Job Monitoring Loop (every 1 second)
    
    Runner->>System: is_job_completed(job)
    System->>System: _is_job_running(job)
    System->>System: _is_mpijob_running(job)
    System->>K8sAPI: get_namespaced_custom_object(mpijob)
    K8sAPI-->>System: status, conditions
    
    System->>System: store_logs_for_job(job_name, output_path)
    System->>K8sAPI: list_namespaced_pod() for job
    K8sAPI-->>System: pod list
    loop For each pod
        System->>K8sAPI: read_namespaced_pod_log(pod)
        K8sAPI-->>System: logs
        System->>FS: write logs (overwrite mode)
    end
    
    System-->>Runner: is_running=False (when complete)
    
    Note over Runner,FS: Job Completion Flow
    
    Runner->>Runner: on_job_completion(job)
    Runner->>System: store_logs_for_job(job_name, output_path)
    Note right of System: Redundant log fetch
    System->>K8sAPI: list_namespaced_pod() for job
    loop For each pod
        System->>K8sAPI: read_namespaced_pod_log(pod)
        System->>FS: write logs (overwrite mode)
    end
    
    Runner->>System: delete_job(job_name, kind)
    System->>K8sAPI: delete_namespaced_custom_object(mpijob)