Kubernetes Operator in Python without frameworks and SDK

Go is currently a monopolist among programming languages that people choose to write operators for Kubernetes. There are such objective reasons as:

There is a powerful framework for developing operators on the Go - Operator SDK .
Go has written applications that turn the game around like Docker and Kubernetes. To write your operator on Go - to speak with the ecosystem in the same language.
High performance applications on Go and simple tools for working with concurrency out of the box.

NB : By the way, how to write your operator on Go, we already described in one of our translations of foreign authors.

')

But what if studying Go prevents you from lack of time or, trivially, motivation? The article provides an example of how to write a good-quality operator using one of the most popular languages that almost every DevOps engineer knows - Python .

Meet: Copywriter - copy operator!

For example, consider the development of a simple operator designed to copy ConfigMap, either when a new namespace appears or when one of two entities changes: ConfigMap and Secret. From the point of view of practical application, the operator can be useful for mass updating application configurations (by updating ConfigMap) or for updating sensitive data — for example, keys for working with Docker Registry (when adding Secret to the namespace).

So, what a good operator should have :

Interaction with the operator is carried out using Custom Resource Definitions (hereinafter - CRD).
The operator can be customized. For this we will use the command line flags and environment variables.
The assembly of the Docker-container and the Helm-chart are worked out so that users can easily (literally with one team) install the operator into their Kubernetes cluster.

CRD

In order for the operator to know what resources and where to look for him, we need to set a rule for him. Each rule will be represented as a single CRD object. What fields should this CRD have?

The type of resource we will be looking for (ConfigMap or Secret).
A list of namespaces in which resources should be located.
Selector , according to which we will search for resources in the namespace.

We describe the CRD:

apiVersion: apiextensions.k8s.io/v1beta1 kind: CustomResourceDefinition metadata: name: copyrator.flant.com spec: group: flant.com versions: - name: v1 served: true storage: true scope: Namespaced names: plural: copyrators singular: copyrator kind: CopyratorRule shortNames: - copyr validation: openAPIV3Schema: type: object properties: ruleType: type: string namespaces: type: array items: type: string selector: type: string

And immediately we will create a simple rule - to search in the namespace with the name of all ConfigMap's default labels with the type of copyrator: "true" :

 apiVersion: flant.com/v1 kind: CopyratorRule metadata: name: main-rule labels: module: copyrator ruleType: configmap selector: copyrator: "true" namespace: default

Done! Now we need to somehow get information about our rule. Immediately, I’ll make a reservation that we will not write requests to the API Server cluster. To do this, we use the ready kubernetes-client Python library:

 import kubernetes from contextlib import suppress CRD_GROUP = 'flant.com' CRD_VERSION = 'v1' CRD_PLURAL = 'copyrators' def load_crd(namespace, name): client = kubernetes.client.ApiClient() custom_api = kubernetes.client.CustomObjectsApi(client) with suppress(kubernetes.client.api_client.ApiException): crd = custom_api.get_namespaced_custom_object( CRD_GROUP, CRD_VERSION, namespace, CRD_PLURAL, name, ) return {x: crd[x] for x in ('ruleType', 'selector', 'namespace')}

As a result of this code, we get the following:

 {'ruleType': 'configmap', 'selector': {'copyrator': 'true'}, 'namespace': ['default']}

Great: we managed to get a rule for the operator. And most importantly - we did it, as they say, the Kubernetes way.

Variables or flags? Take everything!

We proceed to the basic configuration of the operator. There are two basic approaches to configuring applications:

use command line parameters;
use environment variables.

Command line parameters allow you to read settings more flexibly, with support and validation of data types. In the standard Python library, there is an argparser module, which we will use. Details and examples of its capabilities are available in the official documentation .

Here is how for our case the example of setting up the reading of command line flags will look like:

  parser = ArgumentParser( description='Copyrator - copy operator.', prog='copyrator' ) parser.add_argument( '--namespace', type=str, default=getenv('NAMESPACE', 'default'), help='Operator Namespace' ) parser.add_argument( '--rule-name', type=str, default=getenv('RULE_NAME', 'main-rule'), help='CRD Name' ) args = parser.parse_args()

On the other hand, using environment variables in Kubernetes, you can easily transfer the pod service information inside the container. For example, information about the namespace, in which the pod is running, we can get the following construction:

 env: - name: NAMESPACE valueFrom: fieldRef: fieldPath: metadata.namespace

Operator Logic

To understand how to separate the methods for working with ConfigMap and Secret, let's use special maps. Then we will be able to understand what methods we need to track and create an object:

 LIST_TYPES_MAP = { 'configmap': 'list_namespaced_config_map', 'secret': 'list_namespaced_secret', } CREATE_TYPES_MAP = { 'configmap': 'create_namespaced_config_map', 'secret': 'create_namespaced_secret', }

Next, you need to receive events from the server API. We implement it as follows:

 def handle(specs): kubernetes.config.load_incluster_config() v1 = kubernetes.client.CoreV1Api() #       method = getattr(v1, LIST_TYPES_MAP[specs['ruleType']]) func = partial(method, specs['namespace']) w = kubernetes.watch.Watch() for event in w.stream(func, _request_timeout=60): handle_event(v1, specs, event)

After receiving the event, go to the main logic of its processing:

 #  ,     ALLOWED_EVENT_TYPES = {'ADDED', 'UPDATED'} def handle_event(v1, specs, event): if event['type'] not in ALLOWED_EVENT_TYPES: return object_ = event['object'] labels = object_['metadata'].get('labels', {}) #    selector' for key, value in specs['selector'].items(): if labels.get(key) != value: return #   namespace' namespaces = map( lambda x: x.metadata.name, filter( lambda x: x.status.phase == 'Active', v1.list_namespace().items ) ) for namespace in namespaces: #  ,  namespace object_['metadata'] = { 'labels': object_['metadata']['labels'], 'namespace': namespace, 'name': object_['metadata']['name'], } #   /  methodcaller( CREATE_TYPES_MAP[specs['ruleType']], namespace, object_ )(v1)

The main logic is ready! Now you need to pack it all in one Python package. We make the setup.py , we write there meta information about the project:

 from sys import version_info from setuptools import find_packages, setup if version_info[:2] < (3, 5): raise RuntimeError( 'Unsupported python version %s.' % '.'.join(version_info) ) _NAME = 'copyrator' setup( name=_NAME, version='0.0.1', packages=find_packages(), classifiers=[ 'Development Status :: 3 - Alpha', 'Programming Language :: Python', 'Programming Language :: Python :: 3', 'Programming Language :: Python :: 3.5', 'Programming Language :: Python :: 3.6', 'Programming Language :: Python :: 3.7', ], author='Flant', author_email='maksim.nabokikh@flant.com', include_package_data=True, install_requires=[ 'kubernetes==9.0.0', ], entry_points={ 'console_scripts': [ '{0} = {0}.cli:main'.format(_NAME), ] } )

NB : The Python kubernetes client has its own versioning. More information about the compatibility of client versions and Kubernetes versions can be found in the compatibility matrix .

Now our project looks like this:

 copyrator ├── copyrator │ ├── cli.py #      │ ├── constant.py # ,     │ ├── load_crd.py #   CRD │ └── operator.py #     └── setup.py #

Docker and Helm

Dockerfile will be outrageously simple: take a basic python-alpine image and install our package. We postpone its optimization until better times:

 FROM python:3.7.3-alpine3.9 ADD . /app RUN pip3 install /app ENTRYPOINT ["copyrator"]

Deployment for the operator is also very simple:

 apiVersion: apps/v1 kind: Deployment metadata: name: {{ .Chart.Name }} spec: selector: matchLabels: name: {{ .Chart.Name }} template: metadata: labels: name: {{ .Chart.Name }} spec: containers: - name: {{ .Chart.Name }} image: privaterepo.yourcompany.com/copyrator:latest imagePullPolicy: Always args: ["--rule-type", "main-rule"] env: - name: NAMESPACE valueFrom: fieldRef: fieldPath: metadata.namespace serviceAccountName: {{ .Chart.Name }}-acc

Finally, it is necessary to create an appropriate role for the operator with the necessary rights:

 apiVersion: v1 kind: ServiceAccount metadata: name: {{ .Chart.Name }}-acc --- apiVersion: rbac.authorization.k8s.io/v1beta1 kind: ClusterRole metadata: name: {{ .Chart.Name }} rules: - apiGroups: [""] resources: ["namespaces"] verbs: ["get", "watch", "list"] - apiGroups: [""] resources: ["secrets", "configmaps"] verbs: ["*"] --- apiVersion: rbac.authorization.k8s.io/v1beta1 kind: ClusterRoleBinding metadata: name: {{ .Chart.Name }} roleRef: apiGroup: rbac.authorization.k8s.io kind: ClusterRole name: {{ .Chart.Name }} subjects: - kind: ServiceAccount name: {{ .Chart.Name }}

Total

So, without fear, reproach and learning Go, we were able to assemble our own operator for Kubernetes in Python. Of course, he still has room to grow: in the future he will be able to process several rules, work in several streams, independently monitor changes in his CRDs ...

So that we could get acquainted with the code, we put it in a public repository . If you want examples of more serious operators implemented using Python, you can turn your attention to two operators for deploying mongodb (the first and second ).

PS And if you are too lazy to deal with the events of Kubernetes, or you simply get used to using Bash, our colleagues prepared a ready-made solution in the form of a shell-operator (we announced it in April).