+++ /dev/null
-#######################################################################################
-# Copyright 2020 Canonical Ltd.
-# Copyright ETSI Contributors and Others.
-#
-# Licensed 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.
-#######################################################################################
-
-import base64
-import logging
-from typing import Dict
-import typing
-import uuid
-import json
-
-from distutils.version import LooseVersion
-
-from kubernetes import client, config
-from kubernetes.client.api import VersionApi
-from kubernetes.client.models import (
- V1ClusterRole,
- V1Role,
- V1ObjectMeta,
- V1PolicyRule,
- V1ServiceAccount,
- V1ClusterRoleBinding,
- V1RoleBinding,
- V1RoleRef,
- RbacV1Subject,
- V1Secret,
- V1SecretReference,
- V1Namespace,
-)
-from kubernetes.client.rest import ApiException
-from n2vc.libjuju import retry_callback
-from retrying_async import retry
-
-
-SERVICE_ACCOUNT_TOKEN_KEY = "token"
-SERVICE_ACCOUNT_ROOT_CA_KEY = "ca.crt"
-# clients
-CORE_CLIENT = "core_v1"
-RBAC_CLIENT = "rbac_v1"
-STORAGE_CLIENT = "storage_v1"
-CUSTOM_OBJECT_CLIENT = "custom_object"
-
-
-class Kubectl:
- def __init__(self, config_file=None):
- config.load_kube_config(config_file=config_file)
- self._clients = {
- CORE_CLIENT: client.CoreV1Api(),
- RBAC_CLIENT: client.RbacAuthorizationV1Api(),
- STORAGE_CLIENT: client.StorageV1Api(),
- CUSTOM_OBJECT_CLIENT: client.CustomObjectsApi(),
- }
- self._configuration = config.kube_config.Configuration.get_default_copy()
- self.logger = logging.getLogger("lcm.odu")
-
- @property
- def configuration(self):
- return self._configuration
-
- @property
- def clients(self):
- return self._clients
-
- def get_services(
- self,
- field_selector: str = None,
- label_selector: str = None,
- ) -> typing.List[typing.Dict]:
- """
- Get Service list from a namespace
-
- :param: field_selector: Kubernetes field selector for the namespace
- :param: label_selector: Kubernetes label selector for the namespace
-
- :return: List of the services matching the selectors specified
- """
- kwargs = {}
- if field_selector:
- kwargs["field_selector"] = field_selector
- if label_selector:
- kwargs["label_selector"] = label_selector
- try:
- result = self.clients[CORE_CLIENT].list_service_for_all_namespaces(**kwargs)
- return [
- {
- "name": i.metadata.name,
- "cluster_ip": i.spec.cluster_ip,
- "type": i.spec.type,
- "ports": (
- [
- {
- "name": p.name,
- "node_port": p.node_port,
- "port": p.port,
- "protocol": p.protocol,
- "target_port": p.target_port,
- }
- for p in i.spec.ports
- ]
- if i.spec.ports
- else []
- ),
- "external_ip": [i.ip for i in i.status.load_balancer.ingress]
- if i.status.load_balancer.ingress
- else None,
- }
- for i in result.items
- ]
- except ApiException as e:
- self.logger.error("Error calling get services: {}".format(e))
- raise e
-
- def get_default_storage_class(self) -> str:
- """
- Default storage class
-
- :return: Returns the default storage class name, if exists.
- If not, it returns the first storage class.
- If there are not storage classes, returns None
- """
- storage_classes = self.clients[STORAGE_CLIENT].list_storage_class()
- selected_sc = None
- default_sc_annotations = {
- "storageclass.kubernetes.io/is-default-class": "true",
- # Older clusters still use the beta annotation.
- "storageclass.beta.kubernetes.io/is-default-class": "true",
- }
- for sc in storage_classes.items:
- if not selected_sc:
- # Select the first storage class in case there is no a default-class
- selected_sc = sc.metadata.name
- annotations = sc.metadata.annotations or {}
- if any(
- k in annotations and annotations[k] == v
- for k, v in default_sc_annotations.items()
- ):
- # Default storage
- selected_sc = sc.metadata.name
- break
- return selected_sc
-
- def create_cluster_role(
- self,
- name: str,
- labels: Dict[str, str],
- namespace: str = "kube-system",
- ):
- """
- Create a cluster role
-
- :param: name: Name of the cluster role
- :param: labels: Labels for cluster role metadata
- :param: namespace: Kubernetes namespace for cluster role metadata
- Default: kube-system
- """
- cluster_roles = self.clients[RBAC_CLIENT].list_cluster_role(
- field_selector="metadata.name={}".format(name)
- )
-
- if len(cluster_roles.items) > 0:
- raise Exception("Role with metadata.name={} already exists".format(name))
-
- metadata = V1ObjectMeta(name=name, labels=labels, namespace=namespace)
- # Cluster role
- cluster_role = V1ClusterRole(
- metadata=metadata,
- rules=[
- V1PolicyRule(api_groups=["*"], resources=["*"], verbs=["*"]),
- V1PolicyRule(non_resource_ur_ls=["*"], verbs=["*"]),
- ],
- )
-
- self.clients[RBAC_CLIENT].create_cluster_role(cluster_role)
-
- async def create_role(
- self,
- name: str,
- labels: Dict[str, str],
- api_groups: list,
- resources: list,
- verbs: list,
- namespace: str,
- ):
- """
- Create a role with one PolicyRule
-
- :param: name: Name of the namespaced Role
- :param: labels: Labels for namespaced Role metadata
- :param: api_groups: List with api-groups allowed in the policy rule
- :param: resources: List with resources allowed in the policy rule
- :param: verbs: List with verbs allowed in the policy rule
- :param: namespace: Kubernetes namespace for Role metadata
-
- :return: None
- """
-
- roles = self.clients[RBAC_CLIENT].list_namespaced_role(
- namespace, field_selector="metadata.name={}".format(name)
- )
-
- if len(roles.items) > 0:
- raise Exception("Role with metadata.name={} already exists".format(name))
-
- metadata = V1ObjectMeta(name=name, labels=labels, namespace=namespace)
-
- role = V1Role(
- metadata=metadata,
- rules=[
- V1PolicyRule(api_groups=api_groups, resources=resources, verbs=verbs),
- ],
- )
-
- self.clients[RBAC_CLIENT].create_namespaced_role(namespace, role)
-
- def delete_cluster_role(self, name: str):
- """
- Delete a cluster role
-
- :param: name: Name of the cluster role
- """
- self.clients[RBAC_CLIENT].delete_cluster_role(name)
-
- def _get_kubectl_version(self):
- version = VersionApi().get_code()
- return "{}.{}".format(version.major, version.minor)
-
- def _need_to_create_new_secret(self):
- min_k8s_version = "1.24"
- current_k8s_version = self._get_kubectl_version()
- return LooseVersion(min_k8s_version) <= LooseVersion(current_k8s_version)
-
- def _get_secret_name(self, service_account_name: str):
- random_alphanum = str(uuid.uuid4())[:5]
- return "{}-token-{}".format(service_account_name, random_alphanum)
-
- def _create_service_account_secret(
- self,
- service_account_name: str,
- namespace: str,
- secret_name: str,
- ):
- """
- Create a secret for the service account. K8s version >= 1.24
-
- :param: service_account_name: Name of the service account
- :param: namespace: Kubernetes namespace for service account metadata
- :param: secret_name: Name of the secret
- """
- v1_core = self.clients[CORE_CLIENT]
- secrets = v1_core.list_namespaced_secret(
- namespace, field_selector="metadata.name={}".format(secret_name)
- ).items
-
- if len(secrets) > 0:
- raise Exception(
- "Secret with metadata.name={} already exists".format(secret_name)
- )
-
- annotations = {"kubernetes.io/service-account.name": service_account_name}
- metadata = V1ObjectMeta(
- name=secret_name, namespace=namespace, annotations=annotations
- )
- type = "kubernetes.io/service-account-token"
- secret = V1Secret(metadata=metadata, type=type)
- v1_core.create_namespaced_secret(namespace, secret)
-
- def _get_secret_reference_list(self, namespace: str, secret_name: str):
- """
- Return a secret reference list with one secret.
- K8s version >= 1.24
-
- :param: namespace: Kubernetes namespace for service account metadata
- :param: secret_name: Name of the secret
- :rtype: list[V1SecretReference]
- """
- return [V1SecretReference(name=secret_name, namespace=namespace)]
-
- def create_service_account(
- self,
- name: str,
- labels: Dict[str, str],
- namespace: str = "kube-system",
- ):
- """
- Create a service account
-
- :param: name: Name of the service account
- :param: labels: Labels for service account metadata
- :param: namespace: Kubernetes namespace for service account metadata
- Default: kube-system
- """
- v1_core = self.clients[CORE_CLIENT]
- service_accounts = v1_core.list_namespaced_service_account(
- namespace, field_selector="metadata.name={}".format(name)
- )
- if len(service_accounts.items) > 0:
- raise Exception(
- "Service account with metadata.name={} already exists".format(name)
- )
-
- metadata = V1ObjectMeta(name=name, labels=labels, namespace=namespace)
-
- if self._need_to_create_new_secret():
- secret_name = self._get_secret_name(name)
- secrets = self._get_secret_reference_list(namespace, secret_name)
- service_account = V1ServiceAccount(metadata=metadata, secrets=secrets)
- v1_core.create_namespaced_service_account(namespace, service_account)
- self._create_service_account_secret(name, namespace, secret_name)
- else:
- service_account = V1ServiceAccount(metadata=metadata)
- v1_core.create_namespaced_service_account(namespace, service_account)
-
- def delete_secret(self, name: str, namespace: str = "kube-system"):
- """
- Delete a secret
-
- :param: name: Name of the secret
- :param: namespace: Kubernetes namespace
- Default: kube-system
- """
- self.clients[CORE_CLIENT].delete_namespaced_secret(name, namespace)
-
- def delete_service_account(self, name: str, namespace: str = "kube-system"):
- """
- Delete a service account
-
- :param: name: Name of the service account
- :param: namespace: Kubernetes namespace for service account metadata
- Default: kube-system
- """
- self.clients[CORE_CLIENT].delete_namespaced_service_account(name, namespace)
-
- def create_cluster_role_binding(
- self, name: str, labels: Dict[str, str], namespace: str = "kube-system"
- ):
- """
- Create a cluster role binding
-
- :param: name: Name of the cluster role
- :param: labels: Labels for cluster role binding metadata
- :param: namespace: Kubernetes namespace for cluster role binding metadata
- Default: kube-system
- """
- role_bindings = self.clients[RBAC_CLIENT].list_cluster_role_binding(
- field_selector="metadata.name={}".format(name)
- )
- if len(role_bindings.items) > 0:
- raise Exception("Generated rbac id already exists")
-
- role_binding = V1ClusterRoleBinding(
- metadata=V1ObjectMeta(name=name, labels=labels),
- role_ref=V1RoleRef(kind="ClusterRole", name=name, api_group=""),
- subjects=[
- RbacV1Subject(kind="ServiceAccount", name=name, namespace=namespace)
- ],
- )
- self.clients[RBAC_CLIENT].create_cluster_role_binding(role_binding)
-
- async def create_role_binding(
- self,
- name: str,
- role_name: str,
- sa_name: str,
- labels: Dict[str, str],
- namespace: str,
- ):
- """
- Create a cluster role binding
-
- :param: name: Name of the namespaced Role Binding
- :param: role_name: Name of the namespaced Role to be bound
- :param: sa_name: Name of the Service Account to be bound
- :param: labels: Labels for Role Binding metadata
- :param: namespace: Kubernetes namespace for Role Binding metadata
-
- :return: None
- """
- role_bindings = self.clients[RBAC_CLIENT].list_namespaced_role_binding(
- namespace, field_selector="metadata.name={}".format(name)
- )
- if len(role_bindings.items) > 0:
- raise Exception(
- "Role Binding with metadata.name={} already exists".format(name)
- )
-
- role_binding = V1RoleBinding(
- metadata=V1ObjectMeta(name=name, labels=labels),
- role_ref=V1RoleRef(kind="Role", name=role_name, api_group=""),
- subjects=[
- RbacV1Subject(kind="ServiceAccount", name=sa_name, namespace=namespace)
- ],
- )
- self.clients[RBAC_CLIENT].create_namespaced_role_binding(
- namespace, role_binding
- )
-
- def delete_cluster_role_binding(self, name: str):
- """
- Delete a cluster role binding
-
- :param: name: Name of the cluster role binding
- """
- self.clients[RBAC_CLIENT].delete_cluster_role_binding(name)
-
- @retry(
- attempts=10,
- delay=1,
- fallback=Exception("Failed getting the secret from service account"),
- callback=retry_callback,
- )
- async def get_secret_data(
- self, name: str, namespace: str = "kube-system"
- ) -> (str, str):
- """
- Get secret data
-
- :param: name: Name of the secret data
- :param: namespace: Name of the namespace where the secret is stored
-
- :return: Tuple with the token and client certificate
- """
- v1_core = self.clients[CORE_CLIENT]
-
- secret_name = None
-
- service_accounts = v1_core.list_namespaced_service_account(
- namespace, field_selector="metadata.name={}".format(name)
- )
- if len(service_accounts.items) == 0:
- raise Exception(
- "Service account not found with metadata.name={}".format(name)
- )
- service_account = service_accounts.items[0]
- if service_account.secrets and len(service_account.secrets) > 0:
- secret_name = service_account.secrets[0].name
- if not secret_name:
- raise Exception(
- "Failed getting the secret from service account {}".format(name)
- )
- # TODO: refactor to use get_secret_content
- secret = v1_core.list_namespaced_secret(
- namespace, field_selector="metadata.name={}".format(secret_name)
- ).items[0]
-
- token = secret.data[SERVICE_ACCOUNT_TOKEN_KEY]
- client_certificate_data = secret.data[SERVICE_ACCOUNT_ROOT_CA_KEY]
-
- return (
- base64.b64decode(token).decode("utf-8"),
- base64.b64decode(client_certificate_data).decode("utf-8"),
- )
-
- @retry(
- attempts=10,
- delay=1,
- fallback=Exception("Failed getting data from the secret"),
- )
- async def get_secret_content(
- self,
- name: str,
- namespace: str,
- ) -> dict:
- """
- Get secret data
-
- :param: name: Name of the secret
- :param: namespace: Name of the namespace where the secret is stored
-
- :return: Dictionary with secret's data
- """
- v1_core = self.clients[CORE_CLIENT]
-
- secret = v1_core.read_namespaced_secret(name, namespace)
-
- return secret.data
-
- @retry(
- attempts=10,
- delay=1,
- fallback=Exception("Failed creating the secret"),
- )
- async def create_secret(
- self, name: str, data: dict, namespace: str, secret_type: str
- ):
- """
- Create secret with data
-
- :param: name: Name of the secret
- :param: data: Dict with data content. Values must be already base64 encoded
- :param: namespace: Name of the namespace where the secret will be stored
- :param: secret_type: Type of the secret, e.g., Opaque, kubernetes.io/service-account-token, kubernetes.io/tls
-
- :return: None
- """
- self.logger.info("Enter create_secret function")
- v1_core = self.clients[CORE_CLIENT]
- self.logger.info(f"v1_core: {v1_core}")
- metadata = V1ObjectMeta(name=name, namespace=namespace)
- self.logger.info(f"metadata: {metadata}")
- secret = V1Secret(metadata=metadata, data=data, type=secret_type)
- self.logger.info(f"secret: {secret}")
- v1_core.create_namespaced_secret(namespace, secret)
- self.logger.info("Namespaced secret was created")
-
- async def create_certificate(
- self,
- namespace: str,
- name: str,
- dns_prefix: str,
- secret_name: str,
- usages: list,
- issuer_name: str,
- ):
- """
- Creates cert-manager certificate object
-
- :param: namespace: Name of the namespace where the certificate and secret is stored
- :param: name: Name of the certificate object
- :param: dns_prefix: Prefix for the dnsNames. They will be prefixed to the common k8s svc suffixes
- :param: secret_name: Name of the secret created by cert-manager
- :param: usages: List of X.509 key usages
- :param: issuer_name: Name of the cert-manager's Issuer or ClusterIssuer object
-
- """
- certificate_body = {
- "apiVersion": "cert-manager.io/v1",
- "kind": "Certificate",
- "metadata": {"name": name, "namespace": namespace},
- "spec": {
- "secretName": secret_name,
- "privateKey": {
- "rotationPolicy": "Always",
- "algorithm": "ECDSA",
- "size": 256,
- },
- "duration": "8760h", # 1 Year
- "renewBefore": "2208h", # 9 months
- "subject": {"organizations": ["osm"]},
- "commonName": "osm",
- "isCA": False,
- "usages": usages,
- "dnsNames": [
- "{}.{}".format(dns_prefix, namespace),
- "{}.{}.svc".format(dns_prefix, namespace),
- "{}.{}.svc.cluster".format(dns_prefix, namespace),
- "{}.{}.svc.cluster.local".format(dns_prefix, namespace),
- ],
- "issuerRef": {"name": issuer_name, "kind": "ClusterIssuer"},
- },
- }
- client = self.clients[CUSTOM_OBJECT_CLIENT]
- try:
- client.create_namespaced_custom_object(
- group="cert-manager.io",
- plural="certificates",
- version="v1",
- body=certificate_body,
- namespace=namespace,
- )
- except ApiException as e:
- info = json.loads(e.body)
- if info.get("reason").lower() == "alreadyexists":
- self.logger.warning("Certificate already exists: {}".format(e))
- else:
- raise e
-
- async def delete_certificate(self, namespace, object_name):
- client = self.clients[CUSTOM_OBJECT_CLIENT]
- try:
- client.delete_namespaced_custom_object(
- group="cert-manager.io",
- plural="certificates",
- version="v1",
- name=object_name,
- namespace=namespace,
- )
- except ApiException as e:
- info = json.loads(e.body)
- if info.get("reason").lower() == "notfound":
- self.logger.warning("Certificate already deleted: {}".format(e))
- else:
- raise e
-
- @retry(
- attempts=10,
- delay=1,
- fallback=Exception("Failed creating the namespace"),
- )
- async def create_namespace(self, name: str, labels: dict = None):
- """
- Create a namespace
-
- :param: name: Name of the namespace to be created
- :param: labels: Dictionary with labels for the new namespace
-
- """
- v1_core = self.clients[CORE_CLIENT]
- metadata = V1ObjectMeta(name=name, labels=labels)
- namespace = V1Namespace(
- metadata=metadata,
- )
-
- try:
- v1_core.create_namespace(namespace)
- self.logger.debug("Namespace created: {}".format(name))
- except ApiException as e:
- info = json.loads(e.body)
- if info.get("reason").lower() == "alreadyexists":
- self.logger.warning("Namespace already exists: {}".format(e))
- else:
- raise e
-
- @retry(
- attempts=10,
- delay=1,
- fallback=Exception("Failed deleting the namespace"),
- )
- async def delete_namespace(self, name: str):
- """
- Delete a namespace
-
- :param: name: Name of the namespace to be deleted
-
- """
- try:
- self.clients[CORE_CLIENT].delete_namespace(name)
- except ApiException as e:
- if e.reason == "Not Found":
- self.logger.warning("Namespace already deleted: {}".format(e))
-
- def get_secrets(
- self,
- namespace: str,
- field_selector: str = None,
- ) -> typing.List[typing.Dict]:
- """
- Get Secret list from a namespace
-
- :param: namespace: Kubernetes namespace
- :param: field_selector: Kubernetes field selector
-
- :return: List of the secrets matching the selectors specified
- """
- try:
- v1_core = self.clients[CORE_CLIENT]
- secrets = v1_core.list_namespaced_secret(
- namespace=namespace,
- field_selector=field_selector,
- ).items
- return secrets
- except ApiException as e:
- self.logger.error("Error calling get secrets: {}".format(e))
- raise e
-
- def create_generic_object(
- self,
- api_group: str,
- api_plural: str,
- api_version: str,
- namespace: str,
- manifest_dict: dict,
- ):
- """
- Creates generic object
-
- :param: api_group: API Group
- :param: api_plural: API Plural
- :param: api_version: API Version
- :param: namespace: Namespace
- :param: manifest_dict: Dictionary with the content of the Kubernetes manifest
-
- """
- client = self.clients[CUSTOM_OBJECT_CLIENT]
- try:
- client.create_namespaced_custom_object(
- group=api_group,
- plural=api_plural,
- version=api_version,
- body=manifest_dict,
- namespace=namespace,
- )
- except ApiException as e:
- info = json.loads(e.body)
- if info.get("reason").lower() == "alreadyexists":
- self.logger.warning("Object already exists: {}".format(e))
- else:
- raise e
-
- def delete_generic_object(
- self,
- api_group: str,
- api_plural: str,
- api_version: str,
- namespace: str,
- name: str,
- ):
- """
- Deletes generic object
-
- :param: api_group: API Group
- :param: api_plural: API Plural
- :param: api_version: API Version
- :param: namespace: Namespace
- :param: name: Name of the object
-
- """
- client = self.clients[CUSTOM_OBJECT_CLIENT]
- try:
- client.delete_namespaced_custom_object(
- group=api_group,
- plural=api_plural,
- version=api_version,
- name=name,
- namespace=namespace,
- )
- except ApiException as e:
- info = json.loads(e.body)
- if info.get("reason").lower() == "notfound":
- self.logger.warning("Object already deleted: {}".format(e))
- else:
- raise e
-
- async def get_generic_object(
- self,
- api_group: str,
- api_plural: str,
- api_version: str,
- namespace: str,
- name: str,
- ):
- """
- Gets generic object
-
- :param: api_group: API Group
- :param: api_plural: API Plural
- :param: api_version: API Version
- :param: namespace: Namespace
- :param: name: Name of the object
-
- """
- client = self.clients[CUSTOM_OBJECT_CLIENT]
- try:
- object_dict = client.list_namespaced_custom_object(
- group=api_group,
- plural=api_plural,
- version=api_version,
- namespace=namespace,
- field_selector=f"metadata.name={name}",
- )
- if len(object_dict.get("items")) == 0:
- return None
- return object_dict.get("items")[0]
- except ApiException as e:
- info = json.loads(e.body)
- if info.get("reason").lower() == "notfound":
- self.logger.warning("Cannot get custom object: {}".format(e))
- else:
- raise e
-
- async def list_generic_object(
- self,
- api_group: str,
- api_plural: str,
- api_version: str,
- namespace: str,
- ):
- """
- Lists all generic objects of the requested API group
-
- :param: api_group: API Group
- :param: api_plural: API Plural
- :param: api_version: API Version
- :param: namespace: Namespace
-
- """
- client = self.clients[CUSTOM_OBJECT_CLIENT]
- try:
- object_dict = client.list_namespaced_custom_object(
- group=api_group,
- plural=api_plural,
- version=api_version,
- namespace=namespace,
- )
- self.logger.debug(f"Object-list: {object_dict.get('items')}")
- return object_dict.get("items")
- except ApiException as e:
- info = json.loads(e.body)
- if info.get("reason").lower() == "notfound":
- self.logger.warning(
- "Cannot retrieve list of custom objects: {}".format(e)
- )
- else:
- raise e
-
- @retry(
- attempts=10,
- delay=1,
- fallback=Exception("Failed creating the secret"),
- )
- async def create_secret_string(
- self, name: str, string_data: str, namespace: str, secret_type: str
- ):
- """
- Create secret with data
-
- :param: name: Name of the secret
- :param: string_data: String with data content
- :param: namespace: Name of the namespace where the secret will be stored
- :param: secret_type: Type of the secret, e.g., Opaque, kubernetes.io/service-account-token, kubernetes.io/tls
-
- :return: None
- """
- v1_core = self.clients[CORE_CLIENT]
- metadata = V1ObjectMeta(name=name, namespace=namespace)
- secret = V1Secret(metadata=metadata, string_data=string_data, type=secret_type)
- v1_core.create_namespaced_secret(namespace, secret)
projects / osm / LCM.git / commitdiff