使用kubernetes官方Python SDK访问CCI

安装cci-iam-authenticator

下载安装及设置cci-iam-authenticator。

安装kubernetes-client/python

您可参考Installation，下载安装kubernetes-client/python。

当前CCI的kubernetes API对应社区版本为1.19，根据Compatibility，推荐使用的SDK版本为client 11.y.z。

使用Python SDK

首先需要先生成kubeconfig配置文件。

• 这里的示例代码采用了定期刷新token的方式来防止token过期（缓存值token有效期为24小时），您可以增加获取失败重试的操作，以提升可用性。
• 定期刷新token的方式不适用于该账号权限发生变更的情形，如果账号权限发生变更（如主账号变更子账号权限，导致子账号权限发生变更），变更前获取的token会失效，需要重新获取

# -*- coding: utf-8 -*-
import logging
import time
import threading

from kubernetes import client, config

NAMESPACE = "test-k8s-client-namespace"

logging.basicConfig(
    level=logging.INFO,
    datefmt="%Y-%m-%d %H:%M:%S",
    format="%(asctime)s %(levelname)s %(message)s",
)


def create_namespace():
    flavor = "general-computing"
    pool_size = "10"

    namespace = client.V1Namespace(
        metadata=client.V1ObjectMeta(
            name=NAMESPACE,
            annotations={
                "namespace.kubernetes.io/flavor": flavor,
                "network.cci.io/warm-pool-size": pool_size,
            },
            labels={
                "rbac.authorization.cci.io/enable-k8s-rbac": "false",
            }
        )
    )

    logging.info("start to create namespace %s", NAMESPACE)
    client.CoreV1Api().create_namespace(namespace)
    logging.info("namespace created")


def create_network():
    name = "test-k8s-client-namespace-cn-north-7-default-network"
    project_id = "{project_id}"
    domain_id = "{domain_id}"
    security_group_id = "{security_group_id}"
    available_zone = "{available_zone}"
    vpc_id = "{vpc_id}"
    cidr = "{cidr}"
    network_id = "{network_id}"
    subnet_id = "{subnet_id}"

    body = {
        "apiVersion": "networking.cci.io/v1beta1",
        "kind": "Network",
        "metadata": {
            "annotations": {
                "network.alpha.kubernetes.io/default-security-group": security_group_id,
                "network.alpha.kubernetes.io/domain-id": domain_id,
                "network.alpha.kubernetes.io/project-id": project_id,
            },
            "name": name,
        },
        "spec": {
            "availableZone": available_zone,
            "cidr": cidr,
            "attachedVPC": vpc_id,
            "networkID": network_id,
            "networkType": "underlay_neutron",
            "subnetID": subnet_id,
        }
    }

    api = client.CustomObjectsApi()
    logging.info("start to create network")
    api.create_namespaced_custom_object(
        group="networking.cci.io",
        version="v1beta1",
        namespace=NAMESPACE,
        plural="networks",
        body=body,
    )
    logging.info("network created")


def create_deployment():
    app = "test-k8s-client-deployment"
    image = "library/nginx:stable-alpine-perl"

    body = client.V1Deployment(
        api_version="apps/v1",
        kind="Deployment",
        metadata=client.V1ObjectMeta(name=app),
        spec=client.V1DeploymentSpec(
            replicas=2,
            selector={"matchLabels": {"app": app}},
            template=client.V1PodTemplateSpec(
                metadata=client.V1ObjectMeta(labels={"app": app}),
                spec=client.V1PodSpec(
                    containers=[
                        client.V1Container(
                            name="container-0",
                            image=image,
                            resources=client.V1ResourceRequirements(
                                requests={"cpu": "500m", "memory": "1024Mi"},
                                limits={"cpu": "500m", "memory": "1024Mi"},
                            ),
                        )
                    ],
                    image_pull_secrets=[
                        client.V1LocalObjectReference(name="imagepull-secret")],
                    priority=0),
            ),
        )
    )
    logging.info("start to create deployment %s/%s", NAMESPACE, app)
    client.AppsV1Api().create_namespaced_deployment(NAMESPACE, body)
    logging.info("deployment created")


def get_deployment():
    app = "test-k8s-client-deployment"
    resp = client.AppsV1Api().read_namespaced_deployment(app, NAMESPACE)
    logging.info("deployment detail: %s", resp)


def delete_deployment():
    app = "test-k8s-client-deployment"
    logging.info("start to delete deployment")
    client.AppsV1Api().delete_namespaced_deployment(app, NAMESPACE)
    logging.info("deployment deleted")


def delete_namespace():
    logging.info("start to delete namespace: %s", NAMESPACE)
    client.CoreV1Api().delete_namespace(NAMESPACE)


def main():
    # Configs can be set in Configuration class directly or using helper
    # utility. If no argument provided, the config will be loaded from
    # default location.
    path = '{path to kubeconfig}'
    config.load_kube_config(path)

    # 因为token有效期为24小时，所以这里设置了一个每12小时获取新的token的定时任务
    # 注意：如果账号权限发生变更（如主账号变更子账号权限，导致子账号权限发生变更），变更前获取的token会失效，需要重新获取。
    # 另外，您可以增加获取失败重试的操作，以提升可用性
    def _refresh():
        while True:
            time.sleep(12 * 3600)
            try:
                config.load_kube_config(path)
            except Exception as e:
                print("load_kube_config error: %s" % e)
    t = threading.Thread(target=_refresh)
    t.daemon = True
    t.start()

    create_namespace()
    create_network()
    # wait for namespace and network to be active
    logging.info("waiting for namespace and network to be active")
    time.sleep(30)
    create_deployment()
    get_deployment()
    delete_deployment()
    delete_namespace()


if __name__ == '__main__':
    main()

FAQ

问：以上示例是否适用于其他版本的kubernetes-client/python？

答：上述示例已通过测试，测试环境python3.7.4，测试版本包括：

1. 9.0.1
2. 10.1.0
3. 11.0.0
4. 12.0.1
5. 17.17.0
6. 18.17.0a1
7. 19.15.0