import logging
from typing import Dict
import typing
+import uuid
+from distutils.version import LooseVersion
from kubernetes import client, config
+from kubernetes.client.api import VersionApi
from kubernetes.client.models import (
V1ClusterRole,
V1ObjectMeta,
V1ClusterRoleBinding,
V1RoleRef,
V1Subject,
+ V1Secret,
+ V1SecretReference,
)
from kubernetes.client.rest import ApiException
from retrying_async import retry
"""
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,
:param: namespace: Kubernetes namespace for service account metadata
Default: kube-system
"""
- service_accounts = self.clients[CORE_CLIENT].list_namespaced_service_account(
+ 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:
)
metadata = V1ObjectMeta(name=name, labels=labels, namespace=namespace)
- service_account = V1ServiceAccount(metadata=metadata)
- self.clients[CORE_CLIENT].create_namespaced_service_account(
- namespace, service_account
- )
+ 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_service_account(self, name: str, namespace: str = "kube-system"):
"""
delay=1,
fallback=Exception("Failed getting the secret from service account"),
)
- async def get_secret_data(self, name: str, namespace: str = "kube-system") -> (str, str):
+ 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]