云容器实例 CCI-使用kubernetes官方Python SDK访问CCI:使用Python SDK

时间:2025-02-12 14:55:06

使用Python SDK

您可以前往开发体验馆 Codelabs / Namespace生命周期代码示例(Python)下载相关代码,并在线调试。

首先需要先生成kubeconfig配置文件,参考cci-iam-authenticator使用参考,使用子命令generate-kubeconfig生成kubeconfig配置文件。

  • 这里的示例代码采用了定期刷新token的方式来防止token过期(缓存值token有效期为24小时),您可以增加获取失败重试的操作,以提升可用性。
  • 定期刷新token的方式不适用于该账号权限发生变更的情形,如果账号权限发生变更(如主账号变更子账号权限,导致子账号权限发生变更),变更前获取的token会失效,需要重新获取
# -*- coding: utf-8 -*-import loggingimport timeimport threadingfrom kubernetes import client, configNAMESPACE = "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()
support.huaweicloud.com/sdkreference-cci/cci_09_0003.html