Automate All the Boring Kubernetes Operations with Python
Tags: python, kubernetes, devops
Published: 2022-05-18 10:58:24


Kubernetes became a de-facto standard in recent years and many of us - both DevOps engineers and developers alike - use it on daily basis. Many of the task that we perform are however, same, boring and easy to automate. Oftentimes it's simple enough to whip up a quick shell script with a bunch of `kubectl` commands, but for more complicated automation tasks bash just isn't good enough, and you need the power of proper language, such as Python.

So, in this article we will look at how you can leverage Kubernetes Python Client library to automate whatever annoying Kubernetes task you might be dealing with!

## Playground

Before we start playing with the Kubernetes client, we first need to create a playground cluster where we can safely test things out. We will use _KinD (Kubernetes in Docker)_, which you can install from [here](https://kind.sigs.k8s.io/#installation-and-usage).

We will use the following cluster configuration:

```yaml
# kind.yaml
# https://kind.sigs.k8s.io/docs/user/configuration/
apiVersion: kind.x-k8s.io/v1alpha4
kind: Cluster
name: api-playground
nodes:
- role: control-plane
- role: worker
- role: worker
- role: worker
```

To create cluster from above configuration, you can run:

```bash
kind create cluster --image kindest/node:v1.23.5 --config=kind.yaml

kubectl cluster-info --context kind-api-playground
# Kubernetes control plane is running at https://127.0.0.1:36599
# CoreDNS is running at https://127.0.0.1:36599/api/v1/namespaces/kube-system/services/kube-dns:dns/proxy

kubectl get nodes
# NAME                           STATUS     ROLES                  AGE   VERSION
# api-playground-control-plane   Ready      control-plane,master   58s   v1.23.5
# api-playground-worker          Ready      <none>                 27s   v1.23.5
# api-playground-worker2         NotReady   <none>                 27s   v1.23.5
# api-playground-worker3         NotReady   <none>                 27s   v1.23.5
```

With cluster up-and-running, we also need to install the client library (optionally, inside virtual environment):

```bash
python3 -m venv venv
source venv/bin/activate
pip install kubernetes
```

## Authentication

To perform any action inside our Kubernetes cluster we first need to authenticate.

We will use long-lived tokens so that we don't need to go through the authentication flow repeatedly. Long-lived tokens can be created by creating a _ServiceAccount_:

```bash
kubectl create sa playground
kubectl describe sa playground

Name:                playground
Namespace:           default
Labels:              <none>
Annotations:         <none>
Image pull secrets:  <none>
Mountable secrets:   playground-token-v8bq7
Tokens:              playground-token-v8bq7
Events:              <none>

export KIND_TOKEN=$(kubectl get secret playground-token-v8bq7 -o json | jq -r .data.token | base64 --decode)
```

Using a service account also has the benefit, that it's not tied to any single person, which is always preferable for automation purposes.

Token from the output above can be then used in requests:

```bash
curl -k -X GET -H "Authorization: Bearer $KIND_TOKEN" https://127.0.0.1:36599/apis
```

We're now authenticated, but not authorized to do much of anything. Therefore, next we need to create a _Role_ and bind it to the ServiceAccount so that we can perform actions on resources:

```bash
kubectl create clusterrole manage-pods \
    --verb=get --verb=list --verb=watch --verb=create --verb=update --verb=patch --verb=delete \
    --resource=pods
    
kubectl -n default create rolebinding sa-manage-pods \
    --clusterrole=manage-pods \
    --serviceaccount=default:playground
```

The above gives our service account permission to perform any action on pods, limited to `default` namespace.

You should always keep your roles very narrow and specific, but playing around in KinD, it makes sense to apply cluster-wide admin role:

```bash
kubectl create clusterrolebinding sa-cluster-admin \
  --clusterrole=cluster-admin \
  --serviceaccount=default:playground
```

## Raw Requests

To get a better understanding of what is `kubectl` and also the client doing under the hood, we will start with raw HTTP requests using `curl`.

The easiest way to find out what requests are being made under the hood, is to run the desired `kubectl` command with `-v 10` which will output complete `curl` commands:

```bash
kubectl get pods -v 10
# <snip>
curl -k -v -XGET  -H "Accept: application/json;as=Table;v=v1;g=meta.k8s.io,application/json..." \
    'https://127.0.0.1:36599/api/v1/namespaces/default/pods?limit=500'
# <snip>
```

The output with loglevel 10 will be very verbose, but somewhere it there, you will find the above `curl` command.

Add a Bearer token header in the above `curl` command with your long-lived token and you should be able to perform same actions as `kubectl`, such as:

```bash
curl -s -k -XGET -H "Authorization: Bearer $KIND_TOKEN" -H "Accept: application/json, */*" -H "Content-Type: application/json" \
    -H "kubernetes/$Format" 'https://127.0.0.1:36599/api/v1/namespaces/default/pods/example' | jq .status.phase
# "Running"
```

In case there's request body needed, look up which fields need to be included in the request. For example when creating a Pod, we can use API described [here](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.23/#create-pod-v1-core), which results in following request:

```bash
curl -k -XPOST -H "Authorization: Bearer $KIND_TOKEN" -H "Accept: application/json, */*" -H "Content-Type: application/json" \
    -H "kubernetes/$Format" https://127.0.0.1:36599/api/v1/namespaces/default/pods -d@pod.json

# To confirm
kubectl get pods
NAME      READY   STATUS    RESTARTS   AGE
example   0/1     Running   0          7s
```

Refer to the [Kubernetes API reference](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.23) for object attributes. Additionally, you can also view OpenAPI definition with:

```bash
curl -k -X GET -H "Authorization: Bearer $KIND_TOKEN" https://127.0.0.1:36599/apis
```

Interacting with Kubernetes directly using REST API might be a bit clunky, but there are situations where it might make sense to use it. That includes interacting with APIs that have no equivalent `kubectl` command or for example in case you're using different distribution of Kubernetes - such as OpenShift - which exposes additional APIs not covered by either `kubectl` or client SDK.

## Python Client

Moving onto the Python client itself now. We need to go through the same step as with `kubectl` or `curl`. First being, authentication:

```python
from kubernetes import client
import os

configuration = client.Configuration()
configuration.api_key_prefix["authorization"] = "Bearer"
configuration.host = "https://127.0.0.1:36599"
configuration.api_key["authorization"] = os.getenv("KIND_TOKEN", None)
configuration.verify_ssl = False  # Only for testing with KinD!
api_client = client.ApiClient(configuration)
v1 = client.CoreV1Api(api_client)

ret = v1.list_namespaced_pod(namespace="default", watch=False)
for pod in ret.items:
    print(f"Name: {pod.metadata.name}, Namespace: {pod.metadata.namespace} IP: {pod.status.pod_ip}")
    # Name: example, Namespace: default IP: 10.244.2.2
```

First we define configuration object which tells the client that we will authenticate using Bearer token. Considering that our KinD cluster doesn't use SSL, we disable it, in real cluster however, you should never do that.

To test out the configuration, we use `list_namespaced_pod` method of API client to get all pods in the `default` namespace, and we print out their name, namespace and IP.

Now, for a more realistic task, let's create a Deployment:

```python
deployment_name = "my-deploy"
deployment_manifest = {
    "apiVersion": "apps/v1",
    "kind": "Deployment",
    "metadata": {"name": deployment_name, "namespace": "default"},
    "spec": {"replicas": 3,
             "selector": {
                "matchLabels": {
                    "app": "nginx"
                }},
        "template": {"metadata": {"labels": {"app": "nginx"}},
            "spec": {"containers": [
                {"name": "nginx", "image": "nginx:1.21.6", "ports": [{"containerPort": 80}]}]
            }
        },
    }
}

import time
from kubernetes.client.rest import ApiException

v1 = client.AppsV1Api(api_client)

response = v1.create_namespaced_deployment(body=deployment_manifest, namespace="default")
while True:
    try:
        response = v1.read_namespaced_deployment_status(name=deployment_name, namespace="default")
        if response.status.available_replicas != 3:
            print("Waiting for Deployment to become ready...")
            time.sleep(5)
        else:
            break
    except ApiException as e:
        print(f"Exception when calling AppsV1Api -> read_namespaced_deployment_status: {e}\n")
```

In addition to creating the Deployment, we also wait for its pods to become available. We do that by querying Deployment status and checking number of available replicas.

Also, notice the pattern in function names, such as `create_namespaced_deployment`. To make it more obvious let's look at couple more:

- `replace_namespaced_cron_job`
- `patch_namespaced_stateful_set`
- `list_namespaced_horizontal_pod_autoscaler`
- `read_namespaced_daemon_set`
- `read_custom_resource_definition`

All of these are in format `operation_namespaced_resource` or just `operation_resource` for global resources. They can be additionally suffixed with `_status` or `_scale` for methods that perform operations on resource status such as `read_namespaced_deployment_status` or resource scale such as `patch_namespaced_stateful_set_scale`.

Another thing to highlight is that in the above example we performed the actions using `client.AppsV1Api` which allows us to work with all the resources that belong to `apiVersion: apps/v1`. If we - for example - wanted to use _CronJob_ we would instead choose `BatchV1Api` (which is `apiVersion: batch/v1` in YAML format) or for PVCs we would choose `CoreV1Api` because of `apiVersion: v1` - you get the gist.

As you can imagine, that's a lot of functions to choose from, luckily all of them are listed in [docs](https://github.com/kubernetes-client/python/blob/master/kubernetes/README.md) and you can click on any one of them to get an example of its usage.

Beyond basic CRUD operations, it's also possible to continuously watch objects for changes. Obvious choice is to watch _Events_:

```python
from kubernetes import client, watch

v1 = client.CoreV1Api(api_client)
count = 10
w = watch.Watch()
for event in w.stream(partial(v1.list_namespaced_event, namespace="default"), timeout_seconds=10):
    print(f"Event - Message: {event['object']['message']} at {event['object']['metadata']['creationTimestamp']}")
    count -= 1
    if not count:
        w.stop()
print("Finished namespace stream.")

# Event - Message: Successfully assigned default/my-deploy-cb69f686c-2dspd to api-playground-worker2 at 2022-04-19T11:18:25Z
# Event - Message: Container image "nginx:1.21.6" already present on machine at 2022-04-19T11:18:26Z
# Event - Message: Created container nginx at 2022-04-19T11:18:26Z
# Event - Message: Started container nginx at 2022-04-19T11:18:26Z
```

Here we chose to watch events in `default` namespace. We take the first 10 events and then close the stream. If we wanted to continuously monitor the resources we would just remove the `timeout_seconds` and the `w.stop()` call.

In the first example you saw that we used plain Python `dict` to define the Deployment object which we passed to the client. Alternatively though, we can use a more OOP style by using API Models (classes) provided by the library:

```python
v1 = client.AppsV1Api(api_client)

deployment_manifest = client.V1Deployment(
    api_version="apps/v1",
    kind="Deployment",
    metadata=client.V1ObjectMeta(name=deployment_name),
    spec=client.V1DeploymentSpec(
        replicas=3,
        selector=client.V1LabelSelector(match_labels={
         "app": "nginx"
        }),
        template=client.V1PodTemplateSpec(
            metadata=client.V1ObjectMeta(labels={"app": "nginx"}),
            spec=client.V1PodSpec(
                containers=[client.V1Container(name="nginx",
                                               image="nginx:1.21.6",
                                               ports=[client.V1ContainerPort(container_port=80)]
                                               )]))
        )
)

response = v1.create_namespaced_deployment(body=deployment_manifest, namespace="default")
```

Trying to figure out which model you should use for each argument is a losing battle, tough. When creating resources like shown above, you should always use [documentation for models](https://github.com/kubernetes-client/python/blob/master/kubernetes/README.md#documentation-for-models) and traverse the links as you create the individual sub-objects to figure out what values/types are expected in each field.

## Handy Examples

You should now have a basic idea about how the client works, so let's take a look at some handy examples and snippets that might help you automate daily Kubernetes operations.

A very common thing you might want to perform is a Deployment rollout - usually done with `kubectl rollout restart`. There's however no API to do this. [The way `kubectl` does it](https://github.com/kubernetes/kubectl/blob/release-1.23/pkg/polymorphichelpers/objectrestarter.go#L32) is by updating Deployment Annotations, more specifically, setting `kubectl.kubernetes.io/restartedAt` to current time. This works because any change made to Pod spec causes a restart.

If we want to perform a restart using Python client we need to do the same:

```python
from kubernetes import dynamic
from kubernetes.client import api_client  # Careful - different import - not the same as previous client!
import datetime

client = dynamic.DynamicClient(api_client.ApiClient(configuration=configuration))

api = client.resources.get(api_version="apps/v1", kind="Deployment")

# Even though the Deployment manifest was previously created with class model, it still behaves as dictionary:
deployment_manifest["spec"]["template"]["metadata"]["annotations"] = {
    "kubectl.kubernetes.io/restartedAt": datetime.datetime.utcnow().isoformat()
}

deployment_patched = api.patch(body=deployment_manifest, name=deployment_name, namespace="default")
```

Another common operation is scaling a Deployment, this one fortunately has an API function we can use:

```python
from kubernetes import client

api_client = client.ApiClient(configuration)
apps_v1 = client.AppsV1Api(api_client)

# The body can be of different patch types - https://github.com/kubernetes-client/python/issues/1206#issuecomment-668118057
api_response = apps_v1.patch_namespaced_deployment_scale(deployment_name, "default", {"spec": {"replicas": 5}})
```

For troubleshooting purposes, it often makes sense to `exec` into a Pod and take a look around, possibly grab environment variable to verify correct configuration:

```python
from kubernetes.stream import stream

def pod_exec(name, namespace, command, api_instance):
    exec_command = ["/bin/sh", "-c", command]

    resp = stream(api_instance.connect_get_namespaced_pod_exec,
                  name,
                  namespace,
                  command=exec_command,
                  stderr=True, stdin=False,
                  stdout=True, tty=False,
                  _preload_content=False)

    while resp.is_open():
        resp.update(timeout=1)
        if resp.peek_stdout():
            print(f"STDOUT: \n{resp.read_stdout()}")
        if resp.peek_stderr():
            print(f"STDERR: \n{resp.read_stderr()}")

    resp.close()

    if resp.returncode != 0:
        raise Exception("Script failed")

pod = "example"
api_client = client.ApiClient(configuration)
v1 = client.CoreV1Api(api_client)

pod_exec(pod, "default", "env", v1)

# STDOUT: 
# KUBERNETES_SERVICE_PORT=443
# KUBERNETES_PORT=tcp://10.96.0.1:443
# HOSTNAME=example
# HOME=/root
# ...
```

The snippet above also allows you to run whole shell scripts if needs be.

Moving onto more cluster administration-oriented tasks - let's say you want to apply a _Taint_ onto a node that has some issue. Well, once again there's no direct API for Node Taints, but we can find a way:

```python
from kubernetes import client

api_client = client.ApiClient(configuration)
v1 = client.CoreV1Api(api_client)

# kubectl taint nodes api-playground-worker some-taint=1:NoSchedule
v1.patch_node("api-playground-worker", {"spec": {"taints": [{"effect": "NoSchedule", "key": "some-taint", "value": "1"}]}})

# kubectl get nodes -o custom-columns=NAME:.metadata.name,TAINTS:.spec.taints --no-headers
# api-playground-control-plane   [map[effect:NoSchedule key:node-role.kubernetes.io/master]]
# api-playground-worker          [map[effect:NoSchedule key:some-taint value:1]]
# api-playground-worker2         <none>
# api-playground-worker3         <none>

```

You might also want to monitor a cluster resource utilization to possibly automate cluster scaling. Usually, you'd use `kubectl top` to get the information interactively, with the client library you can do:

```python
# https://github.com/kubernetes-sigs/kind/issues/398
# kubectl apply -f https://github.com/kubernetes-sigs/metrics-server/releases/download/v0.5.0/components.yaml
# kubectl patch -n kube-system deployment metrics-server --type=json \
#   -p '[{"op":"add","path":"/spec/template/spec/containers/0/args/-","value":"--kubelet-insecure-tls"}]'

from kubernetes import client

api_client = client.ApiClient(configuration)
custom_api = client.CustomObjectsApi(api_client)

response = custom_api.list_cluster_custom_object("metrics.k8s.io", "v1beta1", "nodes")  # also works with "pods" instead of "nodes"  

for node in response["items"]:
    print(f"{node['metadata']['name']: <30} CPU: {node['usage']['cpu']: <10} Memory: {node['usage']['memory']}")

# api-playground-control-plane   CPU: 148318488n Memory: 2363504Ki
# api-playground-worker          CPU: 91635913n  Memory: 1858680Ki
# api-playground-worker2         CPU: 75473747n  Memory: 1880860Ki
# api-playground-worker3         CPU: 105692650n Memory: 1881560Ki
```

The above example assumes that you have `metrics-server` installed in your cluster. You can run `kubectl top` to verify that. Use the comment in the snippet to install it if you're working with KinD.

Last but not least - you might already have a bunch of YAML or JSON files that you want to use to deploy or modify objects in your cluster, or you might want to export and backup what you've created with the client. Here's how you can convert from YAML/JSON files to Kubernetes object and back to files again:

```python
# pip install kopf  # (Python 3.7+)
import kopf

api_client = client.ApiClient(configuration)
v1 = client.CoreV1Api(api_client)

pods = []

# https://stackoverflow.com/questions/59977058/clone-kubernetes-objects-programmatically-using-the-python-api/59977059#59977059
ret = v1.list_namespaced_pod(namespace="default")
for pod in ret.items:
    # Simple conversion to Dict/JSON
    print(api_client.sanitize_for_serialization(pod))

    # Conversion with fields clean-up
    pods.append(kopf.AnnotationsDiffBaseStorage()
                .build(body=kopf.Body(api_client.sanitize_for_serialization(pod))))


# Conversion from Dict back to Client object
class FakeKubeResponse:
    def __init__(self, obj):
        import json
        self.data = json.dumps(obj)


for pod in pods:
    pod_manifest = api_client.deserialize(FakeKubeResponse(pod), "V1Pod")
    ...
```

First way to convert existing object into Python dictionary (JSON) is to use `sanitize_for_serialization` which produces raw output with all the generated/default fields. Better option is to use utility methods of `kopf` library which will remove all the unnecessary fields. From there it's simple enough to convert dictionary into proper YAML or JSON file.

For the reverse - that is if we want to go from dictionary to Client Object Model - we can use `deserialize` method of API Client. This method however expects its argument to have a `data` attribute, so we pass it a container class instance with such attribute.

If you already have YAML files which you'd like to use with the Python client, then you can use the utility function `kubernetes.utils.create_from_yaml`.

To get complete overview of all the features of the library, I recommend you take a look at the [examples directory](https://github.com/kubernetes-client/python/tree/master/examples) in the repository.

I'd also encourage you to look through the issues in the library repository, as it has a lot of great examples of client usage, such as [processing events in parallel](https://github.com/kubernetes-client/python/issues/803) or [watching ConfigMaps for updates](https://github.com/kubernetes-client/python/issues/1387).

## Conclusion

The Python client library contains literally hundreds of function, so it's difficult to cover every little feature or use-case there is. Most of them however, follow a common pattern which should make the library's usage pretty natural after couple minutes.

If you're looking for more examples beyond what was shown and referenced above, I recommend exploring other popular tools that make use Python Kubernetes client, such [`kopf`](https://github.com/nolar/kopf) - the library for creating Kubernetes operators. I also find it very useful to take a look at tests of the library itself, as it showcases its intended usage such this [client test suite](https://github.com/kubernetes-client/python/blob/master/kubernetes/e2e_test/test_client.py).