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.
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)A continuation to an existing answer by Tibebes. M
How do I find out the namespace that the code itself is running in?
There's two ways to do this:
env:
- name: CURRENT_NAMESPACE
value: "{{ .Values.namespace }}"