pythonkubernetesmicroservicespytestchaos

Simple way to delete existing pods from Python


I have end-to-end tests written in pytest, running on a Kubernetes cluster in Namespace foo. Now I want to add simple chaos engineering to the tests to check the resilience of my services. For this, I only need to delete specific pods within foo -- since K8s restarts the corresponding service, this simulates the service being temporarily unreachable.

What is a simple way to delete a specific pod in the current namespace with Python?


What I have tried so far:

Since I did not find a suitable example in https://github.com/kubernetes-client/python/tree/master/examples, but the ones using pods looked quite complex, I looked into kubetest, which seems very simple and elegant.

I wanted to use the kube fixture and then do something like this:

pods = kube.get_pods()
for pod in pods:
  if can_be_temporarily_unreachable(pod):
    kube.delete(pod)

I thought calling pytest with parameter --in-cluster would tell kubetest to use the current cluster setup and not create new K8s resources. However, kubetest wants to create a new namespace for each test case that uses the kube fixture, which I do not want. Is there a way to tell kubetest not to create new namespaces but do everything in the current namespace?

Though kubetest looks very simple and elegant otherwise, I am happy to use another solution, too. A simple solution that requires little time and maintenance and does not complicate (reading) the tests would be awesome.


Solution

  • you can use delete_namespaced_pod(from CoreV1Api) to delete specific pods within a namespace.

    Here is an example:

    from kubernetes import client, config
    from kubernetes.client.rest import ApiException
    config.load_incluster_config() # or config.load_kube_config()
    
    configuration = client.Configuration()
    
    with client.ApiClient(configuration) as api_client:
        api_instance = client.CoreV1Api(api_client)
        
        namespace = 'kube-system' # str | see @Max Lobur's answer on how to get this
        name = 'kindnet-mpkvf' # str | Pod name, e.g. via api_instance.list_namespaced_pod(namespace)
        
        try:
            api_response = api_instance.delete_namespaced_pod(name, namespace)
            print(api_response)
        except ApiException as e:
            print("Exception when calling CoreV1Api->delete_namespaced_pod: %s\n" % e)