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 installFrom PyPI directly:
pip install kubernetes
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, configIn 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
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
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.
