Source code for libcloud.container.drivers.kubernetes

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import hashlib
import datetime
from typing import Any, Dict, List, Union, Optional
from collections import OrderedDict

from libcloud.compute.base import Node, NodeSize, NodeImage
from libcloud.compute.types import NodeState
from libcloud.container.base import Container, ContainerImage, ContainerDriver, ContainerCluster
from libcloud.container.types import ContainerState
from libcloud.common.exceptions import BaseHTTPError
from libcloud.common.kubernetes import (
from libcloud.container.providers import Provider

__all__ = [

ROOT_URL = "/api/"

K8S_UNIT_MAP = OrderedDict(
        "K": 1000,
        "Ki": 1024,
        "M": 1000 * 1000,
        "Mi": 1024 * 1024,
        "G": 1000 * 1000 * 1000,
        "Gi": 1024 * 1024 * 1024,

[docs]def to_n_bytes(memory_str: str) -> int: """Convert memory string to number of bytes (e.g. '1234Mi'-> 1293942784) """ if memory_str.startswith("0"): return 0 if memory_str.isnumeric(): return int(memory_str) for unit, multiplier in K8S_UNIT_MAP.items(): if memory_str.endswith(unit): return int(memory_str.strip(unit)) * multiplier
[docs]def to_memory_str(n_bytes: int, unit: Optional[str] = None) -> str: """Convert number of bytes to k8s memory string (e.g. 1293942784 -> '1234Mi') """ if n_bytes == 0: return "0K" n_bytes = int(n_bytes) memory_str = None if unit is None: for unit, multiplier in reversed(K8S_UNIT_MAP.items()): converted_n_bytes_float = n_bytes / multiplier converted_n_bytes = n_bytes // multiplier memory_str = f"{converted_n_bytes}{unit}" if converted_n_bytes_float % 1 == 0: break elif K8S_UNIT_MAP.get(unit): memory_str = f"{n_bytes // K8S_UNIT_MAP[unit]}{unit}" return memory_str
[docs]def to_cpu_str(n_cpus: Union[int, float]) -> str: """Convert number of cpus to cpu string (e.g. 0.5 -> '500m') """ if n_cpus == 0: return "0" millicores = n_cpus * 1000 if millicores % 1 == 0: return f"{int(millicores)}m" microcores = n_cpus * 1000000 if microcores % 1 == 0: return f"{int(microcores)}u" nanocores = n_cpus * 1000000000 return f"{int(nanocores)}n"
[docs]def to_n_cpus(cpu_str: str) -> Union[int, float]: """Convert cpu string to number of cpus (e.g. '500m' -> 0.5, '2000000000n' -> 2) """ if cpu_str.endswith("n"): return int(cpu_str.strip("n")) / 1000000000 elif cpu_str.endswith("u"): return int(cpu_str.strip("u")) / 1000000 elif cpu_str.endswith("m"): return int(cpu_str.strip("m")) / 1000 elif cpu_str.isnumeric(): return int(cpu_str) else: return 0
def sum_resources(*resource_dicts): total_cpu = 0 total_memory = 0 for rd in resource_dicts: total_cpu += to_n_cpus(rd.get("cpu", "0m")) total_memory += to_n_bytes(rd.get("memory", "0K")) return {"cpu": to_cpu_str(total_cpu), "memory": to_memory_str(total_memory)} class KubernetesDeployment: def __init__( self, id: str, name: str, namespace: str, created_at: str, replicas: int, selector: Dict[str, Any], extra: Optional[Dict[str, Any]] = None, ): = id = name self.namespace = namespace self.created_at = created_at self.replicas = replicas self.selector = selector self.extra = extra or {} def __repr__(self): return "<KubernetesDeployment name={} namespace={} replicas={}>".format(, self.namespace, self.replicas, ) class KubernetesPod: def __init__( self, id: str, name: str, containers: List[Container], namespace: str, state: str, ip_addresses: List[str], created_at: datetime.datetime, node_name: str, extra: Dict[str, Any], ): """ A Kubernetes pod """ = id = name self.containers = containers self.namespace = namespace self.state = state self.ip_addresses = ip_addresses self.created_at = created_at self.node_name = node_name self.extra = extra def __repr__(self): return "<KubernetesPod name={} namespace={} state={}>".format(, self.namespace, self.state, ) class KubernetesNamespace(ContainerCluster): """ A Kubernetes namespace """ def __repr__(self): return "<KubernetesNamespace name={}>".format(
[docs]class KubernetesContainerDriver(KubernetesDriverMixin, ContainerDriver): type = Provider.KUBERNETES name = "Kubernetes" website = "" connectionCls = KubernetesBasicAuthConnection supports_clusters = True
[docs] def list_containers(self, image=None, all=True) -> List[Container]: """ List the deployed container images :param image: Filter to containers with a certain image(unused) :type image: :class:`libcloud.container.base.ContainerImage` :param all: Show all container (unused) :type all: ``bool`` :rtype: ``list`` of :class:`libcloud.container.base.Container` """ try: result = self.connection.request(ROOT_URL + "v1/pods").object except Exception as exc: errno = getattr(exc, "errno", None) if errno == 111: raise KubernetesException( errno, "Make sure kube host is accessible" "and the API port is correct", ) raise pods = [self._to_pod(value) for value in result["items"]] containers = [] for pod in pods: containers.extend(pod.containers) return containers
[docs] def get_container(self, id: str) -> Container: """ Get a container by ID :param id: The ID of the container to get :type id: ``str`` :rtype: :class:`libcloud.container.base.Container` """ containers = self.list_containers() match = [container for container in containers if == id] return match[0]
[docs] def list_namespaces(self) -> List[KubernetesNamespace]: """ Get a list of namespaces that pods can be deployed into :rtype: ``list`` of :class:`.KubernetesNamespace` """ try: result = self.connection.request(ROOT_URL + "v1/namespaces/").object except Exception as exc: errno = getattr(exc, "errno", None) if errno == 111: raise KubernetesException( errno, "Make sure kube host is accessible" "and the API port is correct", ) raise namespaces = [self._to_namespace(value) for value in result["items"]] return namespaces
[docs] def get_namespace(self, id: str) -> KubernetesNamespace: """ Get a namespace by ID :param id: The ID of the namespace to get :type id: ``str`` :rtype: :class:`.KubernetesNamespace` """ result = self.connection.request(ROOT_URL + "v1/namespaces/%s" % id).object return self._to_namespace(result)
[docs] def delete_namespace(self, namespace: KubernetesNamespace) -> bool: """ Delete a namespace :return: ``True`` if the destroy was successful, otherwise ``False``. :rtype: ``bool`` """ self.connection.request( ROOT_URL + "v1/namespaces/%s" %, method="DELETE" ).object return True
[docs] def create_namespace(self, name: str) -> KubernetesNamespace: """ Create a namespace :param name: The name of the namespace :type name: ``str`` :rtype: :class:`.KubernetesNamespace` """ request = {"metadata": {"name": name}} result = self.connection.request( ROOT_URL + "v1/namespaces", method="POST", data=json.dumps(request) ).object return self._to_namespace(result)
[docs] def deploy_container( self, name: str, image: ContainerImage, namespace: KubernetesNamespace = None, parameters: Optional[str] = None, start: Optional[bool] = True, ): """ Deploy an installed container image. In kubernetes this deploys a single container Pod. :param name: The name of the new container :type name: ``str`` :param image: The container image to deploy :type image: :class:`.ContainerImage` :param namespace: The namespace to deploy to, None is default :type namespace: :class:`.KubernetesNamespace` :param parameters: Container Image parameters(unused) :type parameters: ``str`` :param start: Start the container on deployment(unused) :type start: ``bool`` :rtype: :class:`.Container` """ if namespace is None: namespace = "default" else: namespace = request = { "metadata": {"name": name}, "spec": {"containers": [{"name": name, "image":}]}, } result = self.connection.request( ROOT_URL + "v1/namespaces/%s/pods" % namespace, method="POST", data=json.dumps(request), ).object return self._to_namespace(result)
[docs] def destroy_container(self, container: Container) -> bool: """ Destroy a deployed container. Because the containers are single container pods, this will delete the pod. :param container: The container to destroy :type container: :class:`.Container` :rtype: ``bool`` """ return self.ex_destroy_pod(container.extra["namespace"], container.extra["pod"])
[docs] def ex_list_pods(self, fetch_metrics: bool = False) -> List[KubernetesPod]: """ List available Pods :param fetch_metrics: Fetch metrics for pods :type fetch_metrics: ``bool`` :rtype: ``list`` of :class:`.KubernetesPod` """ result = self.connection.request(ROOT_URL + "v1/pods").object metrics = None if fetch_metrics: try: metrics = { ( metric["metadata"]["name"], metric["metadata"]["namespace"], ): metric["containers"] for metric in self.ex_list_pods_metrics() } except BaseHTTPError: # Metrics Server may not be installed pass return [self._to_pod(value, metrics=metrics) for value in result["items"]]
[docs] def ex_destroy_pod(self, namespace: str, pod_name: str) -> bool: """ Delete a pod and the containers within it. :param namespace: The pod's namespace :type namespace: ``str`` :param pod_name: Name of the pod to destroy :type pod_name: ``str`` :rtype: ``bool`` """ self.connection.request( ROOT_URL + "v1/namespaces/{}/pods/{}".format(namespace, pod_name), method="DELETE", ).object return True
[docs] def ex_list_nodes(self) -> List[Node]: """ List available Nodes :rtype: ``list`` of :class:`.Node` """ result = self.connection.request(ROOT_URL + "v1/nodes").object return [self._to_node(node) for node in result["items"]]
[docs] def ex_destroy_node(self, node_name: str) -> bool: """ Destroy a node. :param node_name: Name of the node to destroy :type node_name: ``str`` :rtype: ``bool`` """ self.connection.request(ROOT_URL + f"v1/nodes/{node_name}", method="DELETE").object return True
[docs] def ex_get_version(self) -> str: """Get Kubernetes version :rtype: ``str`` """ return self.connection.request("/version").object["gitVersion"]
[docs] def ex_list_nodes_metrics(self) -> List[Dict[str, Any]]: """Get nodes metrics from Kubernetes Metrics Server :rtype: ``list`` of ``dict`` """ return self.connection.request("/apis/").object["items"]
[docs] def ex_list_pods_metrics(self) -> List[Dict[str, Any]]: """Get pods metrics from Kubernetes Metrics Server :rtype: ``list`` of ``dict`` """ return self.connection.request("/apis/").object["items"]
[docs] def ex_list_services(self) -> List[Dict[str, Any]]: """Get cluster services :rtype: ``list`` of ``dict`` """ return self.connection.request(ROOT_URL + "v1/services").object["items"]
[docs] def ex_list_deployments(self) -> List[KubernetesDeployment]: """Get cluster deployments :rtype: ``list`` of :class:`.KubernetesDeployment` """ items = self.connection.request("/apis/apps/v1/deployments").object["items"] return [self._to_deployment(item) for item in items]
def _to_deployment(self, data): id_ = data["metadata"]["uid"] name = data["metadata"]["name"] namespace = data["metadata"]["namespace"] created_at = data["metadata"]["creationTimestamp"] replicas = data["spec"]["replicas"] selector = data["spec"]["selector"] extra = { "labels": data["metadata"]["labels"], "strategy": data["spec"]["strategy"]["type"], "total_replicas": data["status"]["replicas"], "updated_replicas": data["status"]["updatedReplicas"], "ready_replicas": data["status"]["readyReplicas"], "available_replicas": data["status"]["availableReplicas"], "conditions": data["status"]["conditions"], } return KubernetesDeployment( id=id_, name=name, namespace=namespace, created_at=created_at, replicas=replicas, selector=selector, extra=extra, ) def _to_node(self, data): """ Convert an API node data object to a `Node` object """ ID = data["metadata"]["uid"] name = data["metadata"]["name"] driver = self.connection.driver memory = data["status"].get("capacity", {}).get("memory", "0K") cpu = data["status"].get("capacity", {}).get("cpu", "1") if isinstance(cpu, str) and not cpu.isnumeric(): cpu = to_n_cpus(cpu) image_name = data["status"]["nodeInfo"].get("osImage") image = NodeImage(image_name, image_name, driver) size_name = f"{cpu} vCPUs, {memory} Ram" size_id = hashlib.md5(size_name.encode("utf-8")).hexdigest() extra_size = {"cpus": cpu} size = NodeSize( id=size_id, name=size_name, ram=memory, disk=0, bandwidth=0, price=0, driver=driver, extra=extra_size, ) extra = {"memory": memory, "cpu": cpu} extra["os"] = data["status"]["nodeInfo"].get("operatingSystem") extra["kubeletVersion"] = data["status"]["nodeInfo"]["kubeletVersion"] extra["provider_id"] = data.get("spec", {}).get("providerID") for condition in data["status"]["conditions"]: if condition["type"] == "Ready" and condition["status"] == "True": state = NodeState.RUNNING break else: state = NodeState.UNKNOWN public_ips, private_ips = [], [] for address in data["status"]["addresses"]: if address["type"] == "InternalIP": private_ips.append(address["address"]) elif address["type"] == "ExternalIP": public_ips.append(address["address"]) created_at = datetime.datetime.strptime( data["metadata"]["creationTimestamp"], "%Y-%m-%dT%H:%M:%SZ" ) return Node( id=ID, name=name, state=state, public_ips=public_ips, private_ips=private_ips, driver=driver, image=image, size=size, extra=extra, created_at=created_at, ) def _to_pod(self, data, metrics=None): """ Convert an API response to a Pod object """ id_ = data["metadata"]["uid"] name = data["metadata"]["name"] namespace = data["metadata"]["namespace"] state = data["status"]["phase"].lower() node_name = data["spec"].get("nodeName") container_statuses = data["status"].get("containerStatuses", {}) containers = [] extra = {"resources": {}} if metrics: try: extra["metrics"] = metrics[name, namespace] except KeyError: pass # response contains the status of the containers in a separate field for container in data["spec"]["containers"]: if container_statuses: spec = list(filter(lambda i: i["name"] == container["name"], container_statuses))[0] else: spec = container_statuses container_obj = self._to_container(container, spec, data) # Calculate new resources resources = extra["resources"] container_resources = container_obj.extra.get("resources", {}) resources["limits"] = sum_resources( resources.get("limits", {}), container_resources.get("limits", {}) ) extra["resources"]["requests"] = sum_resources( resources.get("requests", {}), container_resources.get("requests", {}) ) containers.append(container_obj) ip_addresses = [ip_dict["ip"] for ip_dict in data["status"].get("podIPs", [])] created_at = datetime.datetime.strptime( data["metadata"]["creationTimestamp"], "%Y-%m-%dT%H:%M:%SZ" ) return KubernetesPod( id=id_, name=name, namespace=namespace, state=state, ip_addresses=ip_addresses, containers=containers, created_at=created_at, node_name=node_name, extra=extra, ) def _to_container(self, data, container_status, pod_data): """ Convert container in Container instances """ state = container_status.get("state") created_at = None if state: started_at = list(state.values())[0].get("startedAt") if started_at: created_at = datetime.datetime.strptime(started_at, "%Y-%m-%dT%H:%M:%SZ") extra = { "pod": pod_data["metadata"]["name"], "namespace": pod_data["metadata"]["namespace"], } resources = data.get("resources") if resources: extra["resources"] = resources return Container( id=container_status.get("containerID") or data["name"], name=data["name"], image=ContainerImage( id=container_status.get("imageID") or data["image"], name=data["image"], path=None, version=None, driver=self.connection.driver, ), ip_addresses=None, state=(ContainerState.RUNNING if container_status else ContainerState.UNKNOWN), driver=self.connection.driver, created_at=created_at, extra=extra, ) def _to_namespace(self, data): """ Convert an API node data object to a `KubernetesNamespace` object """ return KubernetesNamespace( id=data["metadata"]["name"], name=data["metadata"]["name"], driver=self.connection.driver, extra={"phase": data["status"]["phase"]}, )
def ts_to_str(timestamp): """ Return a timestamp as a nicely formatted datetime string. """ date = datetime.datetime.fromtimestamp(timestamp) date_string = date.strftime("%d/%m/%Y %H:%M %Z") return date_string