kubecrd 0.5.0

在K8s集群中创建CRD

还可以使用Kubernetes客户端对象在集群中安装CRD

from from kubernetes import client, config
config.load_kube_config()
k8s_client = client.ApiClient()
Resource.install(k8s_client)

然后您可以在集群中找到该资源

» kubectl get crds/resources.example.com
NAME                    CREATED AT
resources.example.com   2022-03-20T03:58:25Z

$ kubectl api-resources | grep example.com
resources     example.com/v1alpha1                  true         Resource

资源安装是幂等的，因此如果传入exist_ok=True，则重新安装已安装的资源不会引发任何异常

Resource.install(k8s_client, exist_ok=True)

序列化

您可以将资源序列化为适合POST到K8s的形式

>>> example = Resource(name='myResource', tags=['tag1', 'tag2'])
>>> import json
>>> print(json.dumps(example.serialize(), sort_keys=True, indent=4))
{
    "apiVersion": "example.com/v1alpha1",
    "kind": "Resource",
    "metadata": {
        "name": "..."
    },
    "spec": {
        "name": "myResource",
        "tags": [
            "tag1",
            "tag2"
        ]
    }
}

对象也可以直接在K8s中序列化和保存

example.save(k8s_client)

在上面的例子中，“client”是Kubernetes客户端对象。您还可以使用kubernetes_asyncio客户端与asyncio一起使用，并执行以下操作

await example.async_save(k8s_async_client)

反序列化

您可以将Kubernetes API的JSON反序列化为Python CR对象。

$ cat -p testdata/cr.json
{
 "apiVersion": "example.com/v1alpha1",
 "kind": "Resource",
 "metadata": {
     "generation": 1,
     "name": "myresource1",
     "namespace": "default",
     "resourceVersion": "105572812",
     "uid": "02102eb3-968b-418a-8023-75df383daa3c"
 },
 "spec": {
     "name": "bestID",
     "tags": [
         "tag1",
         "tag2"
     ]
 }
 }

通过使用资源的from_json类方法

>>> import json
>>> with open('testdata/cr.json') as fd:
...     json_schema = json.load(fd)
>>> res = Resource.from_json(json_schema)
>>> print(res.name)
bestID
>>> print(res.tags)
['tag1', 'tag2']

这还加载了Kubernetes的V1ObjectMeta并将其设置为CR的.metadata属性

>>> print(res.metadata.namespace)
default
>>> print(res.metadata.name)
myresource1
>>> print(res.metadata.resource_version)
105572812

监控

可以使用 Kubernetes 的标准 Watch API 监视自定义资源的更改。例如，监视所有资源的更改

async for happened, resource in Resource.async_watch(k8s_async_client):
    print(f'Resource {resource.metadata.name} was {happened}')

或者您可以使用 watch 的阻塞同步 API

for happened, resource in Resource.watch(k8s_client):
    print(f'Resource {resource.metadata.name} was {happened}')

安装

可以使用 pip 或您喜欢的工具从 PyPI 安装 Kube CRD

$ pip install kubecrd