How to write an Operator

Pykube can be used to implement Kubernetes Operators. Here is how to write a very simple operator which adds a label foo with value bar to every deployment object which has the pykube-test-operator annotation:

# simplified example script, no error handling!
import pykube, time

while True:
    # loads in-cluster auth or local ~/.kube/config for testing
    config = pykube.KubeConfig.from_env()
    api = pykube.HTTPClient(config)
    for deploy in pykube.Deployment.objects(api, namespace=pykube.all):
        if 'pykube-test-operator' in deploy.annotations:
            print(f'Updating deployment {deploy.namespace}/{deploy.name}..')
            deploy.labels['foo'] = 'bar'
            deploy.update()
    time.sleep(15)

Save the above Python script as main.py.

Testing

You can now test the script locally with Pipenv and Minikube (run minikube start first):

pipenv install pykube-ng
pipenv run python3 main.py

See the operator in action by creating a deployment with the right annotation:

kubectl run nginx --image=nginx
kubectl annotate deploy nginx pykube-test-operator=true

The operator should should now assign the foo label to the nginx deployment.

Building the Docker image

Create a Dockerfile in the same directory as main.py:

FROM python:3.7-alpine3.10

WORKDIR /

RUN pip3 install pykube-ng

COPY main.py /

ENTRYPOINT ["python3", "main.py"]

Now build it:

docker build -t pykube-test-operator .

You need to push the Docker image to some Docker registry before you can deploy it.

Deployment

Now deploy the Docker image to your Kubernetes cluster using a service account with the necessary permissions (in this case to list and update deployments). To create such an service account with the necessary RBAC rights create rbac.yaml with these contents:

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

Apply the RBAC role via kubectl apply -f rbac.yaml.

Finally, the deployment of the operator would then look like (deployment.yaml):

apiVersion: apps/v1
kind: Deployment
metadata:
  name: pykube-test-operator
spec:
  replicas: 1
  selector:
    matchLabels:
      app: pykube-test-operator
  template:
    metadata:
      labels:
        app: pykube-test-operator
    spec:
      serviceAccountName: pykube-test-operator
      containers:
      - name: operator
        # this image needs have been pushed to some Docker registry!
        image: pykube-test-operator
        resources:
          limits:
            memory: 50Mi
          requests:
            cpu: 5m
            memory: 50Mi
        securityContext:
          readOnlyRootFilesystem: true
          runAsNonRoot: true
          runAsUser: 1000

Create the deployment via kubectl apply -f deployment.yaml.

You should now have a working operator deployment in your cluster.