X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Fkubectl.py;h=8b8008efa1bf18ef8ea723113efcf8ce9fe19593;hp=a56b6cdc2f98e39e0b7674e9d586a475ef956625;hb=6343d434fa3cec28d8b9b470054d3a13ada8865a;hpb=015abee87f591b8e28f6b982ae4fb9c67e8791bb

diff --git a/n2vc/kubectl.py b/n2vc/kubectl.py
index a56b6cd..8b8008e 100644
--- a/n2vc/kubectl.py
+++ b/n2vc/kubectl.py
@@ -16,9 +16,12 @@ import base64
 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,
@@ -27,6 +30,8 @@ from kubernetes.client.models import (
     V1ClusterRoleBinding,
     V1RoleRef,
     V1Subject,
+    V1Secret,
+    V1SecretReference,
 )
 from kubernetes.client.rest import ApiException
 from retrying_async import retry
@@ -178,6 +183,58 @@ class Kubectl:
         """
         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,
@@ -192,7 +249,8 @@ class Kubectl:
         :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:
@@ -201,11 +259,16 @@ class Kubectl:
             )
 
         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"):
         """