How to launch job, monitor pod logs and wait for completion

[jptissot]

Hi, I am trying to build a web application that interacts with Kubernetes to launch a series of data-science jobs and monitor these jobs. I think the code below will work but wanted your input on if I am handling the error statuses properly?

using k8s;
using k8s.Models;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace Test
{
    internal class TestK8sJobWatcher
    {
        private static async Task Main(string[] args)
        {
            var config = KubernetesClientConfiguration.BuildConfigFromConfigFile();
            IKubernetes client = new Kubernetes(config);
            Console.WriteLine("Starting Request!");

            var jobId = "t" + Guid.NewGuid();

            // create the job
            var jobObject = new V1Job()
            {
                ApiVersion = "batch/v1",
                Kind = V1Job.KubeKind,
                Metadata = new V1ObjectMeta() { Name = jobId },
                Spec = new V1JobSpec()
                {
                    TtlSecondsAfterFinished = 240,
                    Template = new V1PodTemplateSpec()
                    {
                        Spec = new V1PodSpec()
                        {
                            Containers = new List<V1Container>()
                                {
                                         new V1Container()
                                         {
                                             Image = "container:1.0.0",
                                             Name = "some-name",
                                             ImagePullPolicy = "Never" ,
                                             Command = new List<string>() { "/bin/bash", "-c", "./long-running-command.sh" },
                                         },
                                },
                            RestartPolicy = "Never",
      
                        },
                    },
                    BackoffLimit = 0
                },
            };

            var job = await client.BatchV1.CreateNamespacedJobAsync(jobObject, "default");

            var podSelector = $"job-name={jobId},controller-uid={job.Labels()["controller-uid"]}";

            // Wait for the job to be active
            var jobs = client.BatchV1.ListNamespacedJobWithHttpMessagesAsync("default", labelSelector: podSelector, watch: true, timeoutSeconds: 300);
            await foreach (var (type, item) in jobs.WatchAsync<V1Job, V1JobList>())
            {
                if (item.Status.Active > 0)
                {
                    Console.WriteLine("Job is active");
                    break;
                }
                if (item.Status.Conditions?.Any(c => c.Type == "Failed" && c.Status == "True") == true)
                {
                    throw new Exception("The Job failed");
                }
            }


            V1Pod jobPod = null;
            var podlistResp = client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", labelSelector: podSelector, watch: true, timeoutSeconds: 300);
           // wait for container to be running
            await foreach (var (type, item) in podlistResp.WatchAsync<V1Pod, V1PodList>())
            {
               if(item.Status.ContainerStatuses.Any(c=>c.State.Running != null))
                {
                    jobPod = item;
                    Console.WriteLine("The container is running");
                    break;
                }
               if(item.Status.ContainerStatuses.Any(c=>c.State.Terminated != null))
                {
                    throw new Exception("The container is terminated");
                }
            }

            var response = await client.CoreV1.ReadNamespacedPodLogWithHttpMessagesAsync(jobPod.Metadata.Name, "default", follow: true);
            var stream = response.Body;
            // Any way to steam these differently ? 
            stream.CopyTo(Console.OpenStandardOutput());


            var jobCompletionWatcher = client.BatchV1.ListNamespacedJobWithHttpMessagesAsync("default", labelSelector: $"controller-uid={job.Labels()["controller-uid"]}", watch: true, timeoutSeconds: 300);
            await foreach (var (type, item) in jobCompletionWatcher.WatchAsync<V1Job, V1JobList>())
            {
                if(item.Status.Conditions?.Any(c=>c.Type == "Complete" && c.Status == "True") == true)
                {
                    Console.WriteLine("THE JOB Succeeded");
                    break;
                }
                if (item.Status.Conditions?.Any(c => c.Type == "Failed" && c.Status == "True") == true)
                {
                    throw new Exception("The Job failed");
                }
            }
        }
    }
}

[tg123]

any issue?
looks good to me after going through roughly

you do not have to watch, seems for(;;) get is easier in your case

[jptissot]

No issues, but I did not test all edge cases (image pull backoff, etc) that might happen.

What do you mean by for(;;) get

[jptissot]

An update: the sample above worked fine when ran from a console application in the kubernetes-client project. However, when moving to my project, I ran into the following issue: #893

I think it was due to a timing issue / empty result, which caused that exception. So in the end I ended up doing this: which worked.

        // Send the job to the Kubernetes api.
        var job = await client.BatchV1.CreateNamespacedJobAsync(jobObject, request.Namespace);

        var labelSelector = $"job-name={jobName},controller-uid={job.Labels()["controller-uid"]}";

        V1Pod jobPod = null;

        var timeout = DateTime.UtcNow.AddSeconds(request.KubernetesApiTimeout);

        // Attempt to get the pod that is launched by the job.
        while (DateTime.UtcNow < timeout)
        {
            var pods = await client.CoreV1.ListNamespacedPodAsync(request.Namespace, labelSelector: labelSelector);

            if (pods.Items.Count > 0)
            {
                var podWithRunningContainer = pods.Items.FirstOrDefault(p => p.Status?.ContainerStatuses?.Any(c => c.State?.Running != null) == true || p.Status?.Phase == "Succeeded");

                if (podWithRunningContainer is not null)
                {
                    jobPod = podWithRunningContainer;
                    break;
                }

                var failedPod = pods.Items.FirstOrDefault(p => p.Status?.Phase == "Failed");
                if (failedPod is not null)
                {
                    var logs = await client.CoreV1.ReadNamespacedPodLogAsync(failedPod.Metadata.Name, request.Namespace);
                    //todo: log the pod log somewhere.

                    throw new InvalidOperationException("The Job failed. Pod in failed state.");
                }
            }
            // Add a 1s delay to the requets to not flood the kubernetes api.
            await Task.Delay(1000);
        }

        if (jobPod == null)
        {
            throw new InvalidOperationException("Job failed. Could not find a running pod.");
        }

        // Stream container logs to the logger. Follow: true keeps the stream opened until the container completes it's job.
        // By default, this will stream logs from the first container. We currently only support having a single container for a job.
        var response = await client.CoreV1.ReadNamespacedPodLogWithHttpMessagesAsync(jobPod.Metadata.Name, request.Namespace, follow: true);
        var stream = response.Body;

        using (var sr = new StreamReader(stream))
        {
            var s = "";
            while ((s = await sr.ReadLineAsync()) != null)
            {
                _logger.LogInformation(s);
            }
        }

        var jobFailed = true;
        timeout = DateTime.UtcNow.AddSeconds(request.KubernetesApiTimeout); 

        while (DateTime.UtcNow < timeout)
        {
            var finalJob = await client.BatchV1.ReadNamespacedJobAsync(jobName, request.Namespace);

            // Todo: check if there is another way to get the Complete status.
            if (finalJob.Status.Conditions?.Any(c => c.Type == "Complete" && c.Status == "True") == true)
            {
                jobFailed = false;
                _logger.LogInformation("Job {JobName} is successfull.", request.Name);
                break;
            }
            // Add a 1s delay to the requets to not flood the kubernetes api.
            await Task.Delay(1000);
        }

        // Timeout or if the job failed.
        if (jobFailed)
        {
            throw new InvalidOperationException($"Job {jobName} failed");
        }