How to Create Cronjobs with Kubernetes Client Python?

Reading Time: 3 minutes

Hello Readers! In this blog, we will see How to Create Cronjobs with Kubernetes Client Python.

As we all know that generally, we use kubectl commands for creating, listing, updating, and

deleting the Kubernetes resources. But In this blog, we will see how we can use python for

doing these things with resources. 

Installation:

From source:

git clone --recursive https://github.com/kubernetes-client/python.git
cd python
python setup.py install

From PyPI directly:

pip install kubernetes
How to Create Cronjobs with Kubernetes Client Python?

Now, we have installed the python-Kubernetes package installed.

Creating the Cronjobs:

My create.py file contains the following code for creating the cronjobs using

Kubernetes Python Client. Now that we have the python-Kubernetes package installed,

we can import it as:

from kubernetes import client, config

In Python, we instantiate BatchV1beta1Api class from client module:

client_api = client.BatchV1beta1Api()

Authenticating to the Kubernetes API server:

But what if you want to list all the automated cronjobs of a GKE Cluster, you just need to

authenticate the configuration

I’ve used Bearer Token which enables requests to authenticate using an access key.

configuration.api_key = {"authorization": "Bearer" + bearer_token}
from time import sleep

import kubernetes.client
from kubernetes.client.rest import ApiException
from kubernetes import client, config


JOB_NAME = "pi"

def __get_kubernetes_client(bearer_token,api_server_endpoint):
    try:
        configuration = kubernetes.client.Configuration()
        configuration.host = api_server_endpoint
        configuration.verify_ssl = False
        configuration.api_key = {"authorization": "Bearer " + bearer_token}
        client.Configuration.set_default(configuration)
        with kubernetes.client.ApiClient(configuration) as api_client:
    # Create an instance of the API class
            api_instance1 = kubernetes.client.BatchV1beta1Api(api_client)
        return api_instance1
    except Exception as e:
        print("Error getting kubernetes client \n{}".format(e))
        return None

def create_cron_job(api_instance,cluster_details,namespace):
    container = client.V1Container(
    name="pi",
    image="perl",
    command=["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"])
    # Create and configure a spec section
    template = client.V1PodTemplateSpec(
        metadata=client.V1ObjectMeta(labels={"app": "pi"}),
        spec=client.V1PodSpec(restart_policy="Never", containers=[container]))
    # Create the specification of deployment
    spec = client.V1JobSpec(
        template=template,
        backoff_limit=4)  #specify the number of retries before considering a Job as failed
    # Instantiate the job object
    job = client.V1Job(
        api_version="batch/v1",
        kind="Job",
        metadata=client.V1ObjectMeta(name=JOB_NAME),
        spec=spec)


    client_api= __get_kubernetes_client(
            bearer_token=cluster_details["bearer_token"],
            api_server_endpoint=cluster_details["api_server_endpoint"],
        )

    api_response = client_api.create_namespaced_cron_job(
        body=client.V1beta1CronJob(
            api_version='batch/v1beta1',
            kind='CronJob',
            metadata=client.V1ObjectMeta(name='cronjob8'),
            spec = client.V1CronJobSpec(

                schedule="25 17 8 6 2", #     25(minute)   5(pm)   8(date)   june(month) tuesday(day of the week)
                job_template=job
            )
            
        ),
        namespace=namespace)
    print("cronJob created. status='%s'" % str(api_response.status))
    
   


if __name__ == '__main__':
    batch_v1 = client.BatchV1Api()
    cluster_details={
        "bearer_token":"GKE-Bearer-Token",
        "api_server_endpoint":"Ip-k8s-crontrol-plane"
    }

    create_cron_job(batch_v1,cluster_details,"default")

save the code file name e.g create-cronjobs.py and run the below command, but before

that, you’ve to provide the GKE Cluster Bearer token and also the IP of the Kubernetes control plane.

python3 create-cronjobs.py
How to Create Cronjobs with Kubernetes Client Python?

As you can see here my cronjob is successfully created.

Get the cronjobs:

To get the list of created and existing cronjobs use the below methods and call the function getcronjobs like how I’ve called the create_cron_job() in the above example.

def __get_kubernetes_client(bearer_token,api_server_endpoint):
    try:
        configuration = client.Configuration()
        configuration.host = api_server_endpoint
        configuration.verify_ssl = False
        configuration.api_key = {"authorization": "Bearer " + bearer_token} # for authentication to the GKE Cluster
        client.Configuration.set_default(configuration)
        client_api = client.BatchV1beta1Api() #BatchV1beta1Api for interacting with cronjobs
        return client_api
    except Exception as e:
        print("Error getting kubernetes client \n{}".format(e))
        return None

def getcronjobs(cluster_details,namespace="default",all_namespaces=True):
        client_api= __get_kubernetes_client(
            bearer_token=cluster_details["bearer_token"],
            api_server_endpoint=cluster_details["api_server_endpoint"],
        )
        if all_namespaces is True: #list all the cronjobs in all namespaces
            
            ret =client_api.list_cron_job_for_all_namespaces(watch=False)
            temp_dict_obj={}
            temp_list_obj=[]
            for i in ret.items:
                temp_dict_obj={
                    "name": i.metadata.name,
                    "namespace": i.metadata.namespace
                }
                temp_list_obj.append(temp_dict_obj)
            print("cronjob under all namespaces: {}".format(temp_list_obj))
            return temp_list_obj
        else:
            cronjob_list = client_api.list_namespaced_cron_job("{}".format(namespace)) #list the cronjobs of default namespace
            print("cronjob under default namespaces:{}".format(cronjob_list))
            return cronjob_list

After that, I will run the following command:

python3 get-cronjobs.py
How to Create Cronjobs with Kubernetes Client Python?

As you can see here that my cronjobs list under all namespaces and Similarly you can list in default and any other

namespace by doing if all_namespaces is False:

So, now you have an idea of the Python Kubernetes Client of the GKE cluster for more information check out the examples. and Deomonset.

Conclusion:

Therefore, In this blog, we have seen how easily we can create and get the list of cronjobs using the Kubernetes Python client. If you like this blog, please do show your appreciation by giving thumbs-ups and sharing it with your friends.

Written by 

A curious DevOps Intern , love to learn and working on technical skills/tools, know the basics of Linux, Docker, Ansible , Kubernetes and more..