In this blog post, I will show you how to write a series of end to end tests which can validate the Kubernetes cluster health and provide immediate feedback. These tests provides a good level of abstraction and will run fine irrespective of where you are deploying your Kubernetes cluster — Azure or AWS.
Recently I have been working on a Python Library to orchestrate the deployment of MicroServices to a Kubernetes/Istio Service Mesh hosted in Cloud.
As part of this effort, we wanted to have an end to end testing strategy against our Kubernetes/Istio cluster and validate that the system is functioning as expected.
We have been writing unit tests for all our python code, but in a distributed system like Kubernetes there is every chance that a code change may pass all unit tests, but create unexpected issues at the system level. This is where having a solid End to End Testing strategy saves the day.
As the first step, we wanted to have a set of reliable end to end tests, which we can execute against our Kubernetes cluster and determine if the K8s cluster is in a good state.
Before we start with our deployment, we wanted to determine the overall health of the K8s cluster –
- We want to check if all the pods are running
- We want to detect the existence of any Crash Loops
- We want to look at the health of the various Master Components (Scheduler, Controller and Cluster Store – etcd) and Nodes (Kubelet, Kube-Proxy, Pods )
Now let’s move to the code and I will show you the E2E tests I have written. I am using the Python API Client for Kubernetes to interact with the cluster in Cloud. I am primarily using the CoreV1Api class and the method it provides for the tests below.
The first step is to setup the client for the End to End Tests. The Authentication and Kubernetes cluster information are stored in the kube-config file.
| from kubernetes import client, config | |
| # Setup the client for E2E Tests | |
| def k8s_client(): | |
| # Configuration for accessing the K8s Cluster | |
| config.load_kube_config("./config") | |
| # Create and return an instance of the CoreV1Api class | |
| return client.CoreV1Api() |
The below test iterates through the K8 Master Components and verifies if they are Healthy —
| def test_kubernetes_master_components_are_healthy(k8s_client): | |
| ret = k8s_client.list_component_status() | |
| # Iterate through the K8s Master components in the cluster and verify if the components are reporting healthy | |
| for item in ret.items: | |
| assert item.conditions[0].type == "Healthy" # Verify status of Scheduler, Controller and Etcd | |
| print("%s\t%s\t" % (item.metadata.name, item.conditions[0].type)) | |
The below test verifies if the status of the Master Controller is Healthy —
| def test_kubernetes_master_controller_status_is_healthy(k8s_client): | |
| ret = k8s_client.read_component_status('controller-manager') | |
| assert(ret.conditions[0].type == "Healthy" ) # Verify status of Master Controller | |
The below test iterates through all the nodes and verifies if they are Ready. Incase there are any issues with the Kubelet it will also identify the exact issue —
| def test_kubernetes_node_status_is_ready(k8s_client): | |
| nodelist = k8s_client.list_node() | |
| # Iterate through all the nodes and verify that they are Active | |
| for item in nodelist.items: | |
| node = k8s_client.read_node_status(name=item.metadata.name) | |
| print("%s\t" % item.metadata.name) | |
| assert(node.status.conditions[0].status == "False" ) # Verify if kubelet is OutOfDisk | |
| assert(node.status.conditions[1].status == "False" ) # Verify if kubelet has MemoryPressure | |
| assert(node.status.conditions[2].status == "False" ) # Verify if kubelet has DiskPressure | |
| assert(node.status.conditions[3].type == "Ready" ) # Verify kubelet is posting ready status | |
The below test iterates through all the pods in the ‘default’ namespace and verifies if they are in a Running state —
| def test_kubernetes_pod_status_is_running(k8s_client): | |
| podlist = k8s_client.list_namespaced_pod("default") | |
| # Iterate through all the pods in the default namespace and verify that they are Running | |
| for item in podlist.items: | |
| pod = k8s_client.read_namespaced_pod_status(namespace='default', name=item.metadata.name) | |
| print("%s\t%s\t" % (item.metadata.name, item.metadata.namespace)) | |
| print(pod.status.phase) | |
| assert(pod.status.phase == "Running" ) | |
Official Python client library for Kubernetes –
https://github.com/kubernetes-client/python
Documentation for Kubernetes API Endpoints –
https://github.com/kubernetes-client/python/tree/master/kubernetes#documentation-for-api-endpoints
In my next blogpost, I will show you how to write additional tests to verify other objects in the K8s cluster like pods, services and namespaces. Stay Tuned!
Also please let me know if you have any questions and I would be happy to discuss.
Categories: Kubernetes
CNCF Cloud Native Landscape and Trail Map 
Microservices on Kubernetes 
Kubernetes Architecture 
Istio 1.0 – The Production Ready Service Mesh 
Leave a Reply