无需框架和SDK！使用Python来写一个Kubernetes Operator

目前，Go在人们创建Kubernetes Operator时选用的编程语言中成为了事实上的垄断者。他们的偏好源于以下客观原因：

Operator SDK[1]这个强大的框架可用于使用Go来开发Operator
许多基于Go的应用程序，例如Docker和Kubernetes，已成为改变游戏规则的角色。使用Go来编写Operator允许你使用同种语言与这些生态对话。
基于Go的应用程序的高性能以及开箱即用的简单机制。

但是如果你缺少时间或仅是积极性阻碍了你学习Go呢？在此文中，我们将向你展示如何使用几乎所有DevOps工程师熟悉的、最流行的编程语言之一即Python来创建一个可靠的Operator。

欢迎Copyrator！

为了简单实用，我们将创建一个简单的Operator，用于当新的命名空间出现或当ConfigMap或Secret两者之一更改其状态时复制ConfigMap。从实用角度来看，我们新的Operator可用于批量更新应用程序配置（通过更新ConfigMap）或者重设secrets。例如用于Docker Registry的密钥（当Secret添加到命名空间时）。

那么一个优秀的Kubernetes Operator需具备什么功能呢？让我们罗列一下：

与Operator的交互是通过Custom Resource Definitions[2]（以下简称CRD）
该Operator是可配置的，我们能使用命令行参数或者是环境变量来配置它。
Docker镜像和Helm图表在创建时考虑了易用性，所以用户可以毫不费力地安装它（基本上只需一个命令）到他们的Kubernetes集群。

CRD

为了让Operator知道哪些资源以及从哪里查找，我们需要配置一些规则。每个规则将被表示为指定的CRD对象。那这个CRD对象中需要有哪些字段呢？

我们所感兴趣的资源的类型（ConfigMap或者是Secret）
存储资源的命名空间列表
Selector用于帮助我们在特定的命名空间中查找资源。

让我们来定义我们的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

并立即添加一个简单的规则来选择匹配在default命名空间中带有 copyrator:"true"标签的ConfigMap。

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



    
namespace: default

现在我们必须以某种方式获取有关我们规则的信息。我们将不使用手动方式制作集群API请求。所以我们将使用名为kubernetes-client的Python库：

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')}

执行以上代码之后，我们将能看到以下结果：

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

非常好！现在我们已经有一个针对Operator的规则。更重要的是，我们已经可以使用所谓的Kubernetes的方式来做到这一点。

环境变量还是标志呢？我全都要！

现在是时候进行基本的Operator设置了。配置应用程序有两种主要的方法：

通过命令行参数
通过环境变量

你可以通过具备更多灵活性以及支持数据类型验证的命令行参数检索配置。我们将使用 *argparser*标准Python库中的模块。Python文档中[3]提供了其使用的详细信息和示例。

以下是适配我们需求的用于配置命令行标志检索的示例：

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()

另一方面，你可以通过Kubernetes中的环境变量轻松地将有关Pod的服务信息传递到容器中。例如，你可以通过以下结构获取有关运行Pod的命名空间的信息：

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

Operator的操作逻辑

让我们使用指定的字典来划分使用ConfigMap和Secret的方法。它们将使我们能够找出跟踪和创建对象所需的方法：

LIST_TYPES_MAP = {
    'configmap': 'list_namespaced_config_map',
    'secret': 'list_namespaced_secret',
}

CREATE_TYPES_MAP = {
    'configmap': 'create_namespaced_config_map',
    'secret': 'create_namespaced_secret',
}

然后你需要从APIserver获取事件。我们将以下面的方式来实现该功能：

def handle(specs):
    kubernetes.config.load_incluster_config()
    v1 = kubernetes.client.CoreV1Api()
    # Get the method for tracking objects
    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)

收到事件后，我们继续处理它的基本逻辑：

# Types of events to which we will respond
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', {})
    # Look for the matches using selector



    
    for key, value in specs['selector'].items():
        if labels.get(key) != value:
            return
    # Get active namespaces
    namespaces = map(
        lambda x: x.metadata.name,
        filter(
            lambda


    
 x: x.status.phase == 'Active',
            v1.list_namespace().items
        )
    )
    for namespace in namespaces:
        # Clear the metadata, set the namespace
        object_['metadata'] = {
            'labels': object_['metadata']['labels'],


    
            'namespace': namespace,
            'name': object_['metadata']['name'],
        }
        # Call the method for creating/updating an object
        methodcaller(
            CREATE_TYPES_MAP[specs['ruleType']],
            namespace,
            object_
        )(v1)

完成基本逻辑之后，现在我们需要将它打包到单个Python包中。我们将创建 setup.py并添加有关项目的元数据：

from sys import version_info
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),
        ]
    }
)

注意：Kubernetes的Python客户端库有自己的版本控制系统。此矩阵[4]中概述了客户端和Kubernetes版本的兼容性。

目前，我们的项目具备以下结构：

copyrator
├── copyrator
│ ├── cli.py # 命令行操作逻辑
│ ├──


    
 constant.py # 上面定义的常量
│ ├── load_crd.py # CRD加载逻辑
│ └── operator.py # operator的集成逻辑
└── setup.py # 包描述

Docker和Helm

生成的Dockerfile非常简单：我们将采用基本的python-alpine基础镜像并安装我们的软件包（我们先忽略掉优化相关部分）：

FROM python:3.7.3-alpine3.9
ADD . /


    
app
RUN pip3 install /app
ENTRYPOINT ["copyrator"]

Copyrator的部署也非常简单。

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

最后，我们必须为Operator创建一个具有必要权限的相关角色：

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 }}

结论

在本文中，我们展示了如何为Kubernetes创建自己的基于Python的Operator。当然它还有增长的空间，例如你可以通过处理多个规则的能力来丰富它，通过自身来监控CRD的变化，从并发能力中受益等等。

所有代码都可以在我们的公共存储库[5]中找到，以便你了解它。如果你对基于Python的Operator的其他示例感兴趣，我们建议你关注两个用来部署MongoDB的Operator，（[6]和[7]）。

PS. 如果你不想处理Kubernetes事件，或者你更喜欢使用Bash，那么你可能也会喜欢我们易于使用的称为shell-operator的解决方案（我们已在4月份宣布）。

再次PS，有一种使用Python编写Kubernetes的替代方案——通过称为kopf[8]（Kubernetes Operator Pythonic Framework）的特定框架。如果你想最小化你的Python代码，它会很有用。点击这里查看kopf文档[9]。