版本说明

本示例基于K8S社区python版SDK开发，示例代码已通过测试，已测试SDK版本包括：

• kubernetes==9.0.1
• kubernetes==10.1.0
• kubernetes==11.0.0
• kubernetes==12.0.1
• kubernetes==17.17.0
• kubernetes==18.17.0a1
• kubernetes==19.15.0

功能介绍

云容器实例（Cloud Container Instance，CCI）服务提供Serverless Container（无服务器容器）引擎，让您无需创建和管理服务器集群即可直接运行容器。更多信息参考产品介绍。

该示例展示了如何通过K8S社区python版SDK执行以下典型场景：

1. 创建Namespace
2. 创建Namespace对应的Network
3. 在该Namespace下创建Deployment
4. 查询创建的Deployment对象
5. 删除该Deployment对象
6. 删除Namespace

前置条件

获取AK/SK或者username/password等

开发者在使用前需先获取账号的AK/SK或者username/password、endpoint、projectId。

您需要拥有华为云账号以及该账号对应的 Access Key（AK）和 Secret Access Key（SK）。请在华为云控制台“我的凭证-访问密钥”页面上创建和查看您的 AK/SK。更多信息请查看访问密钥。

或者您也可以通过帐号（username）和密码（password）完成认证。

endpoint 华为云各服务应用区域和各服务的终端节点，详情请查看地区和终端节点。

projectId 云服务所在项目 ID ，根据你想操作的项目所属区域选择对应的项目 ID ，请在华为云控制台“我的凭证-API凭证”页面上查看。

运行环境

支持python2/3，SDK版本与python版本是否匹配具体请参考PyPI

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

您可参考使用kubectl，下载、安装及设置cci-iam-authenticator。

SDK获取和安装

从源码安装：

                    
                    Copied!
                
git clone --recursive https://github.com/kubernetes-client/python.git
cd python
python setup.py install

pip安装PyPI:

                    
                    Copied!
                
pip install kubernetes

示例代码

                    
                    Copied!
                
# -*- 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()

参考

• 云容器实例CCI使用前必读
• 使用kubernetes官方Python SDK访问CCI
• kubernetes-client/python

修订记录