End to End Testing in Kubernetes

By Samir Behara on June 25, 2018 • ( 3 )

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.

End to End Tests

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()

view raw

Client Setup for E2E Tests.py

hosted with ❤ by GitHub

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))

view raw

Test K8s Master Components are Healthy.py

hosted with ❤ by GitHub

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

view raw

Test K8s Master Controller in Healthy.py

hosted with ❤ by GitHub

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

view raw

Test K8s Node Status is Ready.py

hosted with ❤ by GitHub

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" )

view raw

Test K8s Pod Status is Running.py

hosted with ❤ by GitHub

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.