Simple way to delete existing pods from Python

Question

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.

Answer 1

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)