Source code for libcloud.compute.drivers.kubevirt

# 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
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# NOTE: Re-enable once we add mypy annotations for the base container API
# type: ignore
"""
kubevirt driver with support for nodes (vms)
"""
import json
import time
import hashlib
from datetime import datetime

from libcloud.common.types import LibcloudError
from libcloud.compute.base import (
    Node,
    NodeSize,
    NodeImage,
    NodeDriver,
    NodeLocation,
    StorageVolume,
)
from libcloud.compute.types import Provider, NodeState
from libcloud.common.kubernetes import (
    VALID_RESPONSE_CODES,
    KubernetesDriverMixin,
    KubernetesBasicAuthConnection,
)

__all__ = ["KubeVirtNodeDriver"]
ROOT_URL = "/api/v1/"
KUBEVIRT_URL = "/apis/kubevirt.io/v1alpha3/"


[docs]class KubeVirtNodeDriver(KubernetesDriverMixin, NodeDriver): type = Provider.KUBEVIRT name = "kubevirt" website = "https://www.kubevirt.io" connectionCls = KubernetesBasicAuthConnection NODE_STATE_MAP = { "pending": NodeState.PENDING, "running": NodeState.RUNNING, "stopped": NodeState.STOPPED, }
[docs] def list_nodes(self, location=None): namespaces = [] if location is not None: namespaces.append(location.name) else: for ns in self.list_locations(): namespaces.append(ns.name) dormant = [] live = [] for ns in namespaces: req = KUBEVIRT_URL + "namespaces/" + ns + "/virtualmachines" result = self.connection.request(req) if result.status != 200: continue result = result.object for item in result["items"]: if not item["spec"]["running"]: dormant.append(item) else: live.append(item) vms = [] for vm in dormant: vms.append(self._to_node(vm, is_stopped=True)) for vm in live: vms.append(self._to_node(vm, is_stopped=False)) return vms
[docs] def get_node(self, id=None, name=None): "get a vm by name or id" if not id and not name: raise ValueError("This method needs id or name to be specified") nodes = self.list_nodes() if id: node_gen = filter(lambda x: x.id == id, nodes) if name: node_gen = filter(lambda x: x.name == name, nodes) try: return next(node_gen) except StopIteration: raise ValueError("Node does not exist")
[docs] def start_node(self, node): # make sure it is stopped if node.state is NodeState.RUNNING: return True name = node.name namespace = node.extra["namespace"] req = KUBEVIRT_URL + "namespaces/" + namespace + "/virtualmachines/" + name data = {"spec": {"running": True}} headers = {"Content-Type": "application/merge-patch+json"} try: result = self.connection.request( req, method="PATCH", data=json.dumps(data), headers=headers ) return result.status in VALID_RESPONSE_CODES except Exception: raise
[docs] def stop_node(self, node): # check if running if node.state is NodeState.STOPPED: return True name = node.name namespace = node.extra["namespace"] req = KUBEVIRT_URL + "namespaces/" + namespace + "/virtualmachines/" + name headers = {"Content-Type": "application/merge-patch+json"} data = {"spec": {"running": False}} try: result = self.connection.request( req, method="PATCH", data=json.dumps(data), headers=headers ) return result.status in VALID_RESPONSE_CODES except Exception: raise
[docs] def reboot_node(self, node): """ Rebooting a node. """ namespace = node.extra["namespace"] name = node.name method = "DELETE" try: result = self.connection.request( KUBEVIRT_URL + "namespaces/" + namespace + "/virtualmachineinstances/" + name, method=method, ) return result.status in VALID_RESPONSE_CODES except Exception: raise return
[docs] def destroy_node(self, node): """ Terminating a VMI and deleting the VM resource backing it """ namespace = node.extra["namespace"] name = node.name # find and delete services for this VM only services = self.ex_list_services(namespace=namespace, node_name=name) for service in services: service_name = service["metadata"]["name"] self.ex_delete_service(namespace=namespace, service_name=service_name) # stop the vmi self.stop_node(node) try: result = self.connection.request( KUBEVIRT_URL + "namespaces/" + namespace + "/virtualmachines/" + name, method="DELETE", ) return result.status in VALID_RESPONSE_CODES except Exception: raise
# only has container disk support atm with no persistency
[docs] def create_node( self, name, image, location=None, ex_memory=128, ex_cpu=1, ex_disks=None, ex_network=None, ex_termination_grace_period=0, ports=None, ): """ Creating a VM with a containerDisk. :param name: A name to give the VM. The VM will be identified by this name and atm it cannot be changed after it is set. :type name: ``str`` :param image: Either a libcloud NodeImage or a string. In both cases it must point to a Docker image with an embedded disk. May be a URI like `kubevirt/cirros-registry-disk-demo`, kubevirt will automatically pull it from https://hub.docker.com/u/URI. For more info visit: https://kubevirt.io/user-guide/docs/latest/creating-virtual-machines/disks-and-volumes.html#containerdisk :type image: `str` :param location: The namespace where the VM will live. (default is 'default') :type location: ``str`` :param ex_memory: The RAM in MB to be allocated to the VM :type ex_memory: ``int`` :param ex_cpu: The amount of cpu to be allocated in miliCPUs ie: 400 will mean 0.4 of a core, 1000 will mean 1 core and 3000 will mean 3 cores. :type ex_cpu: ``int`` :param ex_disks: A list containing disk dictionaries. Each dictionaries should have the following optional keys: -bus: can be "virtio", "sata", or "scsi" -device: can be "lun" or "disk" The following are required keys: -disk_type: atm only "persistentVolumeClaim" is supported -name: The name of the disk configuration -claim_name: the name of the Persistent Volume Claim If you wish a new Persistent Volume Claim can be created by providing the following: required: -size: the desired size (implied in GB) -storage_class_name: the name of the storage class to # NOQA be used for the creation of the Persistent Volume Claim. Make sure it allows for dymamic provisioning. optional: -access_mode: default is ReadWriteOnce -volume_mode: default is `Filesystem`, it can also be `Block` :type ex_disks: `list` of `dict`. For each `dict` the types for its keys are: -bus: `str` -device: `str` -disk_type: `str` -name: `str` -claim_name: `str` (for creating a claim:) -size: `int` -storage_class_name: `str` -volume_mode: `str` -access_mode: `str` :param ex_network: Only the pod type is supported, and in the configuration masquerade or bridge are the accepted values. The parameter must be a tuple or list with (network_type, interface, name) :type ex_network: `iterable` (tuple or list) [network_type, interface, name] network_type: `str` | only "pod" is accepted atm interface: `str` | "masquerade" or "bridge" name: `str` :param ports: A dictionary with keys: 'ports_tcp' and 'ports_udp' 'ports_tcp' value is a list of ints that indicate the ports to be exposed with TCP protocol, and 'ports_udp' is a list of ints that indicate the ports to be exposed with UDP protocol. :type ports: `dict` with keys 'ports_tcp`: `list` of `int` 'ports_udp`: `list` of `int` """ # all valid disk types for which support will be added in the future DISK_TYPES = { "containerDisk", "ephemeral", "configMap", "dataVolume", "cloudInitNoCloud", "persistentVolumeClaim", "emptyDisk", "cloudInitConfigDrive", "hostDisk", } if location is not None: namespace = location.name else: namespace = "default" # vm template to be populated vm = { "apiVersion": "kubevirt.io/v1alpha3", "kind": "VirtualMachine", "metadata": {"labels": {"kubevirt.io/vm": name}, "name": name}, "spec": { "running": False, "template": { "metadata": {"labels": {"kubevirt.io/vm": name}}, "spec": { "domain": { "devices": { "disks": [], "interfaces": [], "networkInterfaceMultiqueue": False, }, "machine": {"type": ""}, "resources": {"requests": {}, "limits": {}}, }, "networks": [], "terminationGracePeriodSeconds": ex_termination_grace_period, # NOQA "volumes": [], }, }, }, } memory = str(ex_memory) + "Mi" vm["spec"]["template"]["spec"]["domain"]["resources"]["requests"]["memory"] = memory vm["spec"]["template"]["spec"]["domain"]["resources"]["limits"]["memory"] = memory if ex_cpu < 10: cpu = int(ex_cpu) vm["spec"]["template"]["spec"]["domain"]["resources"]["requests"]["cpu"] = cpu vm["spec"]["template"]["spec"]["domain"]["resources"]["limits"]["cpu"] = cpu else: cpu = str(ex_cpu) + "m" vm["spec"]["template"]["spec"]["domain"]["resources"]["requests"]["cpu"] = cpu vm["spec"]["template"]["spec"]["domain"]["resources"]["limits"]["cpu"] = cpu i = 0 for disk in ex_disks: disk_type = disk.get("disk_type") bus = disk.get("bus", "virtio") disk_name = disk.get("name", "disk{}".format(i)) i += 1 device = disk.get("device", "disk") if disk_type not in DISK_TYPES: raise ValueError("The possible values for this " "parameter are: ", DISK_TYPES) # depending on disk_type, in the future, # when more will be supported, # additional elif should be added if disk_type == "containerDisk": try: image = disk["image"] except KeyError: raise KeyError("A container disk needs a " "containerized image") volumes_dict = {"containerDisk": {"image": image}, "name": disk_name} if disk_type == "persistentVolumeClaim": if "claim_name" in disk: claimName = disk["claim_name"] if claimName not in self.ex_list_persistent_volume_claims(namespace=namespace): if "size" not in disk or "storage_class_name" not in disk: msg = ( "disk['size'] and " "disk['storage_class_name'] " "are both required to create " "a new claim." ) raise KeyError(msg) size = disk["size"] storage_class = disk["storage_class_name"] volume_mode = disk.get("volume_mode", "Filesystem") access_mode = disk.get("access_mode", "ReadWriteOnce") self.create_volume( size=size, name=claimName, location=location, ex_storage_class_name=storage_class, ex_volume_mode=volume_mode, ex_access_mode=access_mode, ) else: msg = ( "You must provide either a claim_name of an " "existing claim or if you want one to be " "created you must additionally provide size " "and the storage_class_name of the " "cluster, which allows dynamic provisioning, " "so a Persistent Volume Claim can be created. " "In the latter case please provide the desired " "size as well." ) raise KeyError(msg) volumes_dict = { "persistentVolumeClaim": {"claimName": claimName}, "name": disk_name, } disk_dict = {device: {"bus": bus}, "name": disk_name} vm["spec"]["template"]["spec"]["domain"]["devices"]["disks"].append(disk_dict) vm["spec"]["template"]["spec"]["volumes"].append(volumes_dict) # adding image in a container Disk if isinstance(image, NodeImage): image = image.name volumes_dict = {"containerDisk": {"image": image}, "name": "boot-disk"} disk_dict = {"disk": {"bus": "virtio"}, "name": "boot-disk"} vm["spec"]["template"]["spec"]["domain"]["devices"]["disks"].append(disk_dict) vm["spec"]["template"]["spec"]["volumes"].append(volumes_dict) # network if ex_network: interface = ex_network[1] network_name = ex_network[2] network_type = ex_network[0] # add a default network else: interface = "masquerade" network_name = "netw1" network_type = "pod" network_dict = {network_type: {}, "name": network_name} interface_dict = {interface: {}, "name": network_name} ports = ports or {} if ports.get("ports_tcp"): ports_to_expose = [] for port in ports["ports_tcp"]: ports_to_expose.append({"port": port, "protocol": "TCP"}) interface_dict[interface]["ports"] = ports_to_expose if ports.get("ports_udp"): ports_to_expose = interface_dict[interface].get("ports", []) for port in ports.get("ports_udp"): ports_to_expose.append({"port": port, "protocol": "UDP"}) interface_dict[interface]["ports"] = ports_to_expose vm["spec"]["template"]["spec"]["networks"].append(network_dict) vm["spec"]["template"]["spec"]["domain"]["devices"]["interfaces"].append(interface_dict) method = "POST" data = json.dumps(vm) req = KUBEVIRT_URL + "namespaces/" + namespace + "/virtualmachines/" try: self.connection.request(req, method=method, data=data) except Exception: raise # check if new node is present nodes = self.list_nodes() for node in nodes: if node.name == name: self.start_node(node) return node
[docs] def list_images(self, location=None): """ If location (namespace) is provided only the images in that location will be provided. Otherwise all of them. """ nodes = self.list_nodes() if location: namespace = location.name nodes = list(filter(lambda x: x["extra"]["namespace"] == namespace, nodes)) name_set = set() images = [] for node in nodes: if node.image.name in name_set: continue name_set.add(node.image.name) images.append(node.image) return images
[docs] def list_locations(self): """ By locations here it is meant namespaces. """ req = ROOT_URL + "namespaces" namespaces = [] result = self.connection.request(req).object for item in result["items"]: name = item["metadata"]["name"] ID = item["metadata"]["uid"] namespaces.append( NodeLocation(id=ID, name=name, country="", driver=self.connection.driver) ) return namespaces
[docs] def list_sizes(self, location=None): namespace = "" if location: namespace = location.name nodes = self.list_nodes() sizes = [] for node in nodes: if not namespace: sizes.append(node.size) elif namespace == node.extra["namespace"]: sizes.append(node.size) return sizes
[docs] def create_volume( self, size, name, location=None, ex_storage_class_name="", ex_volume_mode="Filesystem", ex_access_mode="ReadWriteOnce", ex_dynamic=True, ex_reclaim_policy="Recycle", ex_volume_type=None, ex_volume_params=None, ): """ :param size: The size in Gigabytes :type size: `int` :param volume_type: This is the type of volume to be created that is dependent on the underlying cloud where Kubernetes is deployed. K8s is supporting the following types: -gcePersistentDisk -awsElasticBlockStore -azureFile -azureDisk -csi -fc (Fibre Channel) -flexVolume -flocker -nfs -iSCSI -rbd (Ceph Block Device) -cephFS -cinder (OpenStack block storage) -glusterfs -vsphereVolume -quobyte Volumes -hostPath (Single node testing only – local storage is not supported in any way and WILL NOT WORK in a multi-node cluster) # NOQA -portworx Volumes -scaleIO Volumes -storageOS This parameter is a dict in the form {type: {key1:value1, key2:value2,...}}, where type is one of the above and key1, key2... are type specific keys and their corresponding values. eg: {nsf: {server: "172.0.0.0", path: "/tmp"}} {awsElasticBlockStore: {fsType: 'ext4', volumeID: "1234"}} :type volume_type: `str` :param volume_params: A dict with the key:value that the volume_type needs. This parameter is a dict in the form {key1:value1, key2:value2,...}, where type is one of the above and key1, key2... are type specific keys and their corresponding values. eg: for nsf volume_type {server: "172.0.0.0", path: "/tmp"} for awsElasticBlockStore volume_type {fsType: 'ext4', volumeID: "1234"} """ if ex_dynamic: if location is None: msg = "Please provide a namespace for the PVC." raise ValueError(msg) vol = self._create_volume_dynamic( size=size, name=name, storage_class_name=ex_storage_class_name, namespace=location.name, volume_mode=ex_volume_mode, access_mode=ex_access_mode, ) return vol else: if ex_volume_type is None or ex_volume_params is None: msg = ( "An ex_volume_type must be provided from the list " "of supported clouds, as well as the ex_volume_params " "necessary for your volume type choice." ) raise ValueError(msg) pv = { "apiVersion": "v1", "kind": "PersistentVolume", "metadata": {"name": name}, "spec": { "capacity": {"storage": str(size) + "Gi"}, "volumeMode": ex_volume_mode, "accessModes": [ex_access_mode], "persistentVolumeReclaimPolicy": ex_reclaim_policy, "storageClassName": ex_storage_class_name, "mountOptions": [], # beta, to add in the future ex_volume_type: ex_volume_params, }, } req = ROOT_URL + "persistentvolumes/" method = "POST" data = json.dumps(pv) try: self.connection.request(req, method=method, data=data) except Exception: raise # make sure that the volume was created volumes = self.list_volumes() for volume in volumes: if volume.name == name: return volume
def _create_volume_dynamic( self, size, name, storage_class_name, volume_mode="Filesystem", namespace="default", access_mode="ReadWriteOnce", ): """ Method to create a Persistent Volume Claim for storage, thus storage is required in the arguments. This method assumes dynamic provisioning of the Persistent Volume so the storage_class given should allow for it (by default it usually is), or already have unbounded Persistent Volumes created by an admin. :param name: The name of the pvc an arbitrary string of lower letters :type name: `str` :param size: An int of the amount of gigabytes desired :type size: `int` :param namespace: The namespace where the claim will live :type namespace: `str` :param storage_class_name: If you want the pvc to be bound to a particular class of PVs specified here. :type storage_class_name: `str` :param access_mode: The desired access mode, ie "ReadOnlyMany" :type access_mode: `str` :param matchLabels: A dictionary with the labels, ie: {'release': 'stable,} :type matchLabels: `dict` with keys `str` and values `str` """ pvc = { "apiVersion": "v1", "kind": "PersistentVolumeClaim", "metadata": {"name": name}, "spec": { "accessModes": [], "volumeMode": volume_mode, "resources": {"requests": {"storage": ""}}, }, } pvc["spec"]["accessModes"].append(access_mode) if storage_class_name is not None: pvc["spec"]["storageClassName"] = storage_class_name else: raise ValueError( "The storage class name must be provided of a" "storage class which allows for dynamic " "provisioning" ) pvc["spec"]["resources"]["requests"]["storage"] = str(size) + "Gi" method = "POST" req = ROOT_URL + "namespaces/" + namespace + "/persistentvolumeclaims" data = json.dumps(pvc) try: result = self.connection.request(req, method=method, data=data) except Exception: raise if result.object["status"]["phase"] != "Bound": for _ in range(3): req = ROOT_URL + "namespaces/" + namespace + "/persistentvolumeclaims/" + name try: result = self.connection.request(req).object except Exception: raise if result["status"]["phase"] == "Bound": break time.sleep(3) # check that the pv was created and bound volumes = self.list_volumes() for volume in volumes: if volume.extra["pvc"]["name"] == name: return volume def _bind_volume(self, volume, namespace="default"): """ This method is for unbound volumes that were statically made. It will bind them to a pvc so they can be used by a kubernetes resource. """ if volume.extra["is_bound"]: return # volume already bound storage_class = volume.extra["storage_class_name"] size = volume.size name = volume.name + "-pvc" volume_mode = volume.extra["volume_mode"] access_mode = volume.extra["access_modes"][0] vol = self._create_volume_dynamic( size=size, name=name, storage_class_name=storage_class, volume_mode=volume_mode, namespace=namespace, access_mode=access_mode, ) return vol
[docs] def destroy_volume(self, volume): # first delete the pvc method = "DELETE" if volume.extra["is_bound"]: pvc = volume.extra["pvc"]["name"] namespace = volume.extra["pvc"]["namespace"] req = ROOT_URL + "namespaces/" + namespace + "/persistentvolumeclaims/" + pvc try: result = self.connection.request(req, method=method) except Exception: raise pv = volume.name req = ROOT_URL + "persistentvolumes/" + pv try: result = self.connection.request(req, method=method) return result.status except Exception: raise
[docs] def attach_volume(self, node, volume, device="disk", ex_bus="virtio", ex_name=None): """ params: bus, name , device (disk or lun) """ # volume must be bound to a claim if not volume.extra["is_bound"]: volume = self._bind_volume(volume, node.extra["namespace"]) if volume is None: raise LibcloudError( "Selected Volume (PV) could not be bound " "(to a PVC), please select another volume", driver=self, ) claimName = volume.extra["pvc"]["name"] if ex_name is None: name = claimName else: name = ex_name namespace = volume.extra["pvc"]["namespace"] # check if vm is stopped self.stop_node(node) # check if it is the same namespace if node.extra["namespace"] != namespace: msg = "The PVC and the VM must be in the same namespace" raise ValueError(msg) vm = node.name req = KUBEVIRT_URL + "namespaces/" + namespace + "/virtualmachines/" + vm disk_dict = {device: {"bus": ex_bus}, "name": name} volumes_dict = {"persistentVolumeClaim": {"claimName": claimName}, "name": name} # Get all the volumes of the vm try: result = self.connection.request(req).object except Exception: raise disks = result["spec"]["template"]["spec"]["domain"]["devices"]["disks"] volumes = result["spec"]["template"]["spec"]["volumes"] disks.append(disk_dict) volumes.append(volumes_dict) # now patch the new volumes and disks lists into the resource headers = {"Content-Type": "application/merge-patch+json"} data = { "spec": { "template": { "spec": { "volumes": volumes, "domain": {"devices": {"disks": disks}}, } } } } try: result = self.connection.request( req, method="PATCH", data=json.dumps(data), headers=headers ) if "pvcs" in node.extra: node.extra["pvcs"].append(claimName) else: node.extra["pvcs"] = [claimName] return result in VALID_RESPONSE_CODES except Exception: raise
[docs] def detach_volume(self, volume, ex_node): """ Detaches a volume from a node but the node must be given since a PVC can have more than one VMI's pointing to it """ # vmi must be stopped self.stop_node(ex_node) claimName = volume.extra["pvc"]["name"] name = ex_node.name namespace = ex_node.extra["namespace"] req = KUBEVIRT_URL + "namespaces/" + namespace + "/virtualmachines/" + name headers = {"Content-Type": "application/merge-patch+json"} # Get all the volumes of the vm try: result = self.connection.request(req).object except Exception: raise disks = result["spec"]["template"]["spec"]["domain"]["devices"]["disks"] volumes = result["spec"]["template"]["spec"]["volumes"] to_delete = None for volume in volumes: if "persistentVolumeClaim" in volume: if volume["persistentVolumeClaim"]["claimName"] == claimName: to_delete = volume["name"] volumes.remove(volume) break if not to_delete: msg = "The given volume is not attached to the given VM" raise ValueError(msg) for disk in disks: if disk["name"] == to_delete: disks.remove(disk) break # now patch the new volumes and disks lists into the resource data = { "spec": { "template": { "spec": { "volumes": volumes, "domain": {"devices": {"disks": disks}}, } } } } try: result = self.connection.request( req, method="PATCH", data=json.dumps(data), headers=headers ) ex_node.extra["pvcs"].remove(claimName) return result in VALID_RESPONSE_CODES except Exception: raise
[docs] def ex_list_persistent_volume_claims(self, namespace="default"): pvc_req = ROOT_URL + "namespaces/" + namespace + "/persistentvolumeclaims" try: result = self.connection.request(pvc_req).object except Exception: raise pvcs = [item["metadata"]["name"] for item in result["items"]] return pvcs
[docs] def ex_list_storage_classes(self): # sc = storage class sc_req = "/apis/storage.k8s.io/v1/storageclasses" try: result = self.connection.request(sc_req).object except Exception: raise scs = [item["metadata"]["name"] for item in result["items"]] return scs
[docs] def list_volumes(self): """ Location is a namespace of the cluster. """ volumes = [] pv_rec = ROOT_URL + "/persistentvolumes/" try: result = self.connection.request(pv_rec).object except Exception: raise for item in result["items"]: if item["status"]["phase"] not in {"Available", "Bound"}: continue ID = item["metadata"]["uid"] size = item["spec"]["capacity"]["storage"] size = int(size.rstrip("Gi")) extra = {"pvc": {}} extra["storage_class_name"] = item["spec"]["storageClassName"] extra["is_bound"] = item["status"]["phase"] == "Bound" extra["access_modes"] = item["spec"]["accessModes"] extra["volume_mode"] = item["spec"]["volumeMode"] if extra["is_bound"]: extra["pvc"]["name"] = item["spec"]["claimRef"]["name"] extra["pvc"]["namespace"] = item["spec"]["claimRef"]["namespace"] extra["pvc"]["uid"] = item["spec"]["claimRef"]["uid"] name = extra["pvc"]["name"] else: name = item["metadata"]["name"] volume = StorageVolume( id=ID, name=name, size=size, driver=self.connection.driver, extra=extra ) volumes.append(volume) return volumes
def _ex_connection_class_kwargs(self): kwargs = {} if hasattr(self, "key_file"): kwargs["key_file"] = self.key_file if hasattr(self, "cert_file"): kwargs["cert_file"] = self.cert_file return kwargs def _to_node(self, vm, is_stopped=False): """ """ ID = vm["metadata"]["uid"] name = vm["metadata"]["name"] driver = self.connection.driver extra = {"namespace": vm["metadata"]["namespace"]} extra["pvcs"] = [] memory = 0 if "limits" in vm["spec"]["template"]["spec"]["domain"]["resources"]: if "memory" in vm["spec"]["template"]["spec"]["domain"]["resources"]["limits"]: memory = vm["spec"]["template"]["spec"]["domain"]["resources"]["limits"]["memory"] elif vm["spec"]["template"]["spec"]["domain"]["resources"].get("requests", None): if vm["spec"]["template"]["spec"]["domain"]["resources"]["requests"].get( "memory", None ): memory = vm["spec"]["template"]["spec"]["domain"]["resources"]["requests"]["memory"] if not isinstance(memory, int): if "M" in memory or "Mi" in memory: memory = memory.rstrip("M") memory = memory.rstrip("Mi") memory = int(memory) elif "G" in memory: memory = memory.rstrip("G") memory = int(memory) // 1000 elif "Gi" in memory: memory = memory.rstrip("Gi") memory = int(memory) // 1024 cpu = 1 if vm["spec"]["template"]["spec"]["domain"]["resources"].get("limits", None): if vm["spec"]["template"]["spec"]["domain"]["resources"]["limits"].get("cpu", None): cpu = vm["spec"]["template"]["spec"]["domain"]["resources"]["limits"]["cpu"] elif vm["spec"]["template"]["spec"]["domain"]["resources"].get("requests", None) and vm[ "spec" ]["template"]["spec"]["domain"]["resources"]["requests"].get("cpu", None): cpu = vm["spec"]["template"]["spec"]["domain"]["resources"]["requests"]["cpu"] elif vm["spec"]["template"]["spec"]["domain"].get("cpu", None): cpu = vm["spec"]["template"]["spec"]["domain"]["cpu"].get("cores", 1) if not isinstance(cpu, int): cpu = int(cpu.rstrip("m")) extra_size = {"cpus": cpu} size_name = "{} vCPUs, {}MB Ram".format(str(cpu), str(memory)) size_id = hashlib.md5(size_name.encode("utf-8")).hexdigest() size = NodeSize( id=size_id, name=size_name, ram=memory, disk=0, bandwidth=0, price=0, driver=driver, extra=extra_size, ) extra["memory"] = memory extra["cpu"] = cpu image_name = "undefined" for volume in vm["spec"]["template"]["spec"]["volumes"]: for k, v in volume.items(): if type(v) is dict: if "image" in v: image_name = v["image"] image = NodeImage(image_name, image_name, driver) if "volumes" in vm["spec"]["template"]["spec"]: for volume in vm["spec"]["template"]["spec"]["volumes"]: if "persistentVolumeClaim" in volume: extra["pvcs"].append(volume["persistentVolumeClaim"]["claimName"]) port_forwards = [] services = self.ex_list_services(namespace=extra["namespace"], node_name=name) for service in services: service_type = service["spec"].get("type") for port_pair in service["spec"]["ports"]: protocol = port_pair.get("protocol") public_port = port_pair.get("port") local_port = port_pair.get("targetPort") try: int(local_port) except ValueError: local_port = public_port port_forwards.append( { "local_port": local_port, "public_port": public_port, "protocol": protocol, "service_type": service_type, } ) extra["port_forwards"] = port_forwards if is_stopped: state = NodeState.STOPPED public_ips = None private_ips = None return Node( id=ID, name=name, state=state, public_ips=public_ips, private_ips=private_ips, driver=driver, size=size, image=image, extra=extra, ) # getting image and image_ID from the container req = ROOT_URL + "namespaces/" + extra["namespace"] + "/pods" result = self.connection.request(req).object pod = None for pd in result["items"]: if "metadata" in pd and "ownerReferences" in pd["metadata"]: if pd["metadata"]["ownerReferences"][0]["name"] == name: pod = pd if pod is None or "containerStatuses" not in pod["status"]: state = NodeState.PENDING public_ips = None private_ips = None return Node( id=ID, name=name, state=state, public_ips=public_ips, private_ips=private_ips, driver=driver, size=size, image=image, extra=extra, ) extra["pod"] = {"name": pod["metadata"]["name"]} for cont_status in pod["status"]["containerStatuses"]: # only 2 containers are present the launcher and the vmi if cont_status["name"] != "compute": image = NodeImage(ID, cont_status["image"], driver) state = ( NodeState.RUNNING if "running" in cont_status["state"] else NodeState.PENDING ) public_ips = None created_at = datetime.strptime(vm["metadata"]["creationTimestamp"], "%Y-%m-%dT%H:%M:%SZ") if "podIPs" in pod["status"]: private_ips = [ip["ip"] for ip in pod["status"]["podIPs"]] else: private_ips = [] return Node( id=ID, name=name, state=state, public_ips=public_ips, private_ips=private_ips, driver=driver, size=size, image=image, extra=extra, created_at=created_at, )
[docs] def ex_list_services(self, namespace="default", node_name=None, service_name=None): """ If node_name is given then the services returned will be those that concern the node """ params = None if service_name is not None: params = {"fieldSelector": "metadata.name={}".format(service_name)} req = ROOT_URL + "/namespaces/{}/services".format(namespace) result = self.connection.request(req, params=params).object["items"] if node_name: res = [] for service in result: if node_name in service["metadata"].get("name", ""): res.append(service) return res return result
[docs] def ex_create_service( self, node, ports, service_type="NodePort", cluster_ip=None, load_balancer_ip=None, override_existing_ports=False, ): """ Each node has a single service of one type on which the exposed ports are described. If a service exists then the port declared will be exposed alongside the existing ones, set override_existing_ports=True to delete existing exposed ports and expose just the ones in the port variable. param node: the libcloud node for which the ports will be exposed type node: libcloud `Node` class param ports: a list of dictionaries with keys --> values: 'port' --> port to be exposed on the service 'target_port' --> port on the pod/node, optional if empty then it gets the same value as 'port' value 'protocol' ---> either 'UDP' or 'TCP', defaults to TCP 'name' --> A name for the service If ports is an empty `list` and a service exists of this type then the service will be deleted. type ports: `list` of `dict` where each `dict` has keys --> values: 'port' --> `int` 'target_port' --> `int` 'protocol' --> `str` 'name' --> `str` param service_type: Valid types are ClusterIP, NodePort, LoadBalancer type service_type: `str` param cluster_ip: This can be set with an IP string value if you want manually set the service's internal IP. If the value is not correct the method will fail, this value can't be updated. type cluster_ip: `str` param override_existing_ports: Set to True if you want to delete the existing ports exposed by the service and keep just the ones declared in the present ports argument. By default it is false and if the service already exists the ports will be added to the existing ones. type override_existing_ports: `boolean` """ # check if service exists first namespace = node.extra.get("namespace", "default") service_name = "service-{}-{}".format(service_type.lower(), node.name) service_list = self.ex_list_services(namespace=namespace, service_name=service_name) ports_to_expose = [] # if ports has a falsey value like None or 0 if not ports: ports = [] for port_group in ports: if not port_group.get("target_port", None): port_group["target_port"] = port_group["port"] if not port_group.get("name", ""): port_group["name"] = "port-{}".format(port_group["port"]) ports_to_expose.append( { "protocol": port_group.get("protocol", "TCP"), "port": int(port_group["port"]), "targetPort": int(port_group["target_port"]), "name": port_group["name"], } ) headers = None data = None if len(service_list) > 0: if not ports: result = True for service in service_list: service_name = service["metadata"]["name"] result = result and self.ex_delete_service( namespace=namespace, service_name=service_name ) return result else: method = "PATCH" spec = {"ports": ports_to_expose} if not override_existing_ports: existing_ports = service_list[0]["spec"]["ports"] spec = {"ports": existing_ports.extend(ports_to_expose)} data = json.dumps({"spec": spec}) headers = {"Content-Type": "application/merge-patch+json"} req = "{}/namespaces/{}/services/{}".format(ROOT_URL, namespace, service_name) else: if not ports: raise ValueError( "Argument ports is empty but there is no " "service of {} type to be deleted".format(service_type) ) method = "POST" service = { "kind": "Service", "apiVersion": "v1", "metadata": { "name": service_name, "labels": {"service": "kubevirt.io"}, }, "spec": { "type": "", "selector": {"kubevirt.io/vm": node.name}, "ports": [], }, } service["spec"]["ports"] = ports_to_expose service["spec"]["type"] = service_type if cluster_ip is not None: service["spec"]["clusterIP"] = cluster_ip if service_type == "LoadBalancer" and load_balancer_ip is not None: service["spec"]["loadBalancerIP"] = load_balancer_ip data = json.dumps(service) req = "{}/namespaces/{}/services".format(ROOT_URL, namespace) try: result = self.connection.request(req, method=method, data=data, headers=headers) except Exception: raise return result.status in VALID_RESPONSE_CODES
[docs] def ex_delete_service(self, namespace, service_name): req = "{}/namespaces/{}/services/{}".format(ROOT_URL, namespace, service_name) headers = {"Content-Type": "application/yaml"} try: result = self.connection.request(req, method="DELETE", headers=headers) except Exception: raise return result.status in VALID_RESPONSE_CODES