Build a simple operator

In this guide we will build a controller that periodically reconciles all Deployments and adds a label to any with a certain annotation.

Warning

While you can build operators with kr8s we would recommend folks look at using kopf for building anything more complex than the below example.

Controller

First we need to create a Python script called controller.py containing the controller code that uses kr8s.

This script runs a reconciliation loop that periodically lists all Deployments with kr8s.get(). It then checks the Deployment.annotations property and if it has the annotation pykube-test-operator it adds the label foo=bar using Deployment.label().

Sync

# controller.py
import time
import kr8s

def run():
    while True:
        for deploy in kr8s.get("deployments", namespace=kr8s.ALL):
            if 'pykube-test-operator' in deploy.annotations:
                deploy.label(foo="bar")
        time.sleep(15)

if __name__ == "__main__":
    run()

Async

# controller.py
import asyncio
import kr8s

async def run():
    while True:
        for deploy in await kr8s.asyncio.get("deployments", namespace=kr8s.ALL):
            if 'pykube-test-operator' in deploy.annotations:
                await deploy.label(foo="bar")
        await asyncio.sleep(15)

if __name__ == "__main__":
    asyncio.run(run())

Packaging

Now we can package our controller code in a container image.

# Dockerfile
FROM python:3.11

WORKDIR /usr/local/src

RUN pip install kr8s

COPY controller.py /usr/local/src/

CMD ["python3", "/usr/local/src/controller.py"]

$ docker build -t foo/labelling-operator:latest .
$ docker push foo/labelling-operator:latest

Deploying

Then to deploy our operator we need to create a ServiceAccount with a ClusterRole for our controller to use to communicate with the Kubernetes API.

# rbac.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: labelling-operator
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: labelling-operator
rules:
- apiGroups:
  - apps
  resources:
  - deployments
  verbs:
  - get
  - watch
  - list
  - update
  - patch
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: labelling-operator
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: labelling-operator
subjects:
- kind: ServiceAccount
  name: labelling-operator
  namespace: default

$ kubectl apply -f rbac.yaml

Then create a deployment to run our controller container.

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: labelling-operator
spec:
  replicas: 1
  selector:
    matchLabels:
      app: labelling-operator
  template:
    metadata:
      labels:
        app: labelling-operator
    spec:
      serviceAccountName: labelling-operator
      containers:
      - name: operator
        image: foo/labelling-operator:latest

$ kubectl apply -f deployment.yaml