rollout status deployment : wait for a deployment rollout

[Smana]

Hey guys,

I would like to know how would it be possible to wait for a deployment rollout and get the output similar to what does the command :

kubectl rollout status deployment foo

Sorry but didn't find the answer in the documentation,
Thanks for your help

[tomplus]

Maybe it would help: read_namespaced_deployment_status

[Smana]

@tomplus thank you for your answer.
Actually, I could definitely write something with that but I was just wondering if there's a complete example somewhere. A waiting loop that prints the status until the rollout is finished.
Anyway, I'll close this issue in a few days if no one has an example

[kamilglod]

@Smana Following code from https://github.com/kubernetes/kubernetes/blob/74bcefc8b2bf88a2f5816336999b524cc48cf6c0/pkg/controller/deployment/util/deployment_util.go#L745 you can check deployment status like so:

def _wait_for_deployment_complete(deployment_name, timeout=60):
    api = _get_kube_api(client.ExtensionsV1beta1Api)
    start = time.time()
    while time.time() - start < timeout:
        time.sleep(2)
        response = api.read_namespaced_deployment_status(deployment_name, K8S_API_NAMESPACE)
        s = response.status
        if (s.updated_replicas == response.spec.replicas and
                s.replicas == response.spec.replicas and
                s.available_replicas == response.spec.replicas and
                s.observed_generation >= response.metadata.generation):
            return True
        else:
            print(f'[updated_replicas:{s.updated_replicas},replicas:{s.replicas}'
                  ',available_replicas:{s.available_replicas},observed_generation:{s.observed_generation}] waiting...')

    raise RuntimeError(f'Waiting timeout for deployment {deployment_name}')

[Smana]

Awesome! Thank you @kamilglod

[PrajwalShenoy]

You can find more docs here
This is my implementation

from kubernetes import client, config
from kubeconfig import KubeConfig
import time

conf = KubeConfig()
namespace = conf.view()['contexts'][0]['context']['namespace']
config.load_kube_config()
v1=client.AppsV1Api()
dep = v1.read_namespaced_deployment_status(deployment_name, namespace)
while dep.status.updated_replicas != dep.spec.replicas:
    time.sleep(10)

[jaketf]

kind of an edge case, but stumbled upon this while implementing a test suite of autoscaling to zero. TIL that if the deployment is scaled to zero dep.spec.replicas will be None not 0.

Perhaps this will be useful to another person who also finds this thread.