Using the Kubernetes Python SDK to Access CCI

This section describes how to use cci-iam-authenticator (the CCI authentication tool) and kubernetes-client/python to call the APIs.

Installing cci-iam-authenticator

Download, install, and set cci-iam-authenticator by referring to Using Native kubectl (Recommended).

Installing kubernetes-client/python

Download and install kubernetes-client/python by referring to Installation.

The Kubernetes API version of CCI is Kubernetes 1.19. According to Compatibility, the recommended SDK version is client 11.y.z.

Using the Python SDK

Run generate-kubeconfig to generate the kubeconfig file. For details, see cci-iam-authenticator Usage Reference.

The validity period of a cached token is 24 hours. In the following example, the token is periodically refreshed to prevent token expiration. You can also add the operation of retry upon failure to improve availability.
Periodically refreshing the token is not applicable after the account permission is changed. For example, if the permissions of a subaccount are changed by the main account, the token obtained before the change becomes invalid and needs to be obtained again.

# -*- 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)

    # The validity period of a token is 24 hours. Therefore, set a scheduled task for obtaining a new token every 12 hours.
    # Note: If the account permissions are changed (for example, the permissions of a subaccount are changed by the main account), the token obtained before the change becomes invalid and needs to be obtained again.
    # You can also add the operation of retry upon failure to improve availability.
    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()