运维首页 > 云运维 > 实现Kubernetes Operator的新方式：Python

实现Kubernetes Operator的新方式：Python

2023年 7月 10日 17.0k 0

本文为翻译文章，点击查看原文。

编者按

云原生领域，Go几乎成了垄断编程语言。本文作者团队另辟蹊径，向读者们展示了如何使用最流行的编程语言之一Python创建一个可靠的Kubernetes operator。

前言

目前，人们创建Kubernetes operator时，Go编程语言几乎成了唯一选择。他们的偏好来自如下客观原因:

  • 有一个强大的框架支持基于Go开发operator - Operator SDK。
  • 许多基于Go的应用程序，如Docker和Kubernetes，已经成为游戏的主导者。用Go编写operator可以让你用同一种语言与生态系统对话。
  • 基于Go的应用程序的高性能以及开箱即用的并发机制。

    • 但是，如果缺乏时间或者仅仅是没有动力去学习Go语言呢？在本文中，我们将向您展示如何使用几乎所有DevOps工程师都熟悉的最流行的编程语言之一Python创建一个可靠的operator。

    欢迎Copyrator — the copy operator

    为了使事情变得简单实用，让我们创建一个简单的operator：当出现一个新的namespace，或ConfigMap与Secret之一更改其状态时，复制ConfigMap。从实用的角度来看，我们的新operator可以用于批量更新应用程序的配置（通过更新ConfigMap）或重置Secret，例如用于Docker注册中心的键（当一个Secret添加到namespace时）。

    那么，一个好的Kubernetes operator 应该具备哪些特征呢？让我们列举出来:

  • 与operator的交互通过Custom Resource Definitions（以下简称CRD）进行。
  • Operator是可配置的。我们可以使用命令行参数和环境变量来设置它。
  • Docker image和Helm chart在创建时考虑到了简单性，因此用户可以毫不费力地将其安装到Kubernetes集群中（基本上只需一个命令）。

    • CRD

    为了让operator知道要查找哪些资源和在哪里查找，我们需要设置一些规则。每个规则都将表示为一个特定的CRD对象。那么，这个CRD对象应该有哪些字段呢？

  • 我们感兴趣的资源的类型 (ConfigMap or Secret）。
  • 储存资源的namespace列表。
  • 帮助我们搜索特定namespace中的资源的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

    … 然后立即添加一个简单规则来选择ConfigMaps，要求在默认namespace中使用与copyrator: "true"匹配的标签：

    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设置的时候了。配置应用程序有两种主要方法:

    • 通过命令行参数，
    • 通过环境变量。

    您可以通过命令行参数获取设置，其具有更大的灵活性，并支持/验证数据类型。我们将使用标准Python库中的*argparser*模块，使用的详细信息和示例可以在Python文档中找到。

    下面是一个配置命令行参数检索的例子，适用于我们的情况:

    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运行的namespace的信息:

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

    然后必须从API服务器接收事件。我们将以以下方式实现该功能：

    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_infofrom 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='[email protected]',
    include_package_data=True,
    install_requires=[
        'kubernetes==9.0.0',
    ],
    entry_points={
        'console_scripts': [
            '{0} = {0}.cli:main'.format(_NAME),
        ]
    }
)

    注意: 用于Kubernetes的Python客户端库有自己的版本控制系统。客户端和Kubernetes版本的兼容性概述在这个matrix中。

    目前，我们的项目有如下结构：

    copyrator
├── copyrator
│ ├── cli.py # Command line operating logic
│ ├── constant.py # Constants that we described above
│ ├── load_crd.py # CRD loading logic
│ └── operator.pyк # Basic logic of the operator
└── setup.py # Package description

    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。当然，它仍然有改进的空间:您可以通过使其处理多个规则、监视其CRDs中的更改、从并发功能中获益等手段来丰富它……

    所有代码都可以在我们的公共仓库中找到，以便于您去熟悉它。如果您对基于python的其他operator示例感兴趣，我们建议您可以查看部署mongodb的两个operator(链接1和链接2)。

    P.S.如果你不喜欢处理Kubernetes事件，或者你只是更习惯于使用Bash，你也可以享受我们易于使用的被称为shell-operator (我们有在4月宣布)的解决方案。

    P.P.S.还有一种使用Python编写K8s的替代方法—一个名为kopf(Kubernetes Operator Pythonic Framework)的特定框架。如果您想将Python代码量最小化，这个框架是非常有用的。查看kopf文档。

    张二河
    云运维
    0 0

    相关文章

    KubeSphere 部署向量数据库 Milvus 实战指南
    探索 Kubernetes 持久化存储之 Longhorn 初窥门径
    征服 Docker 镜像访问限制！KubeSphere v3.4.1 成功部署全攻略
    那些年在 Terraform 上吃到的糖和踩过的坑
    无需 Kubernetes 测试 Kubernetes 网络实现
    Kubernetes v1.31 中的移除和主要变更

    发布评论

    张二河

    这个人很懒，什么都没有留下～

    文章8551 评论0