51eb4758a3496a3c6ae5e0897a1fdb6568af2528
2 # Copyright 2019 Telefonica Investigacion y Desarrollo, S.A.U.
3 # This file is part of OSM
6 # Licensed under the Apache License, Version 2.0 (the "License");
7 # you may not use this file except in compliance with the License.
8 # You may obtain a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS,
14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
16 # See the License for the specific language governing permissions and
17 # limitations under the License.
19 # For those usages not covered by the Apache License, Version 2.0 please
20 # contact with: nfvlabs@tid.es
24 from typing
import Union
32 from uuid
import uuid4
34 from n2vc
.config
import EnvironConfig
35 from n2vc
.exceptions
import K8sException
36 from n2vc
.k8s_conn
import K8sConnector
39 class K8sHelmBaseConnector(K8sConnector
):
42 ####################################################################################
43 ################################### P U B L I C ####################################
44 ####################################################################################
47 service_account
= "osm"
53 kubectl_command
: str = "/usr/bin/kubectl",
54 helm_command
: str = "/usr/bin/helm",
60 :param fs: file system for kubernetes and helm configuration
61 :param db: database object to write current operation status
62 :param kubectl_command: path to kubectl executable
63 :param helm_command: path to helm executable
65 :param on_update_db: callback called when k8s connector updates database
69 K8sConnector
.__init
__(self
, db
=db
, log
=log
, on_update_db
=on_update_db
)
71 self
.log
.info("Initializing K8S Helm connector")
73 self
.config
= EnvironConfig()
74 # random numbers for release name generation
75 random
.seed(time
.time())
80 # exception if kubectl is not installed
81 self
.kubectl_command
= kubectl_command
82 self
._check
_file
_exists
(filename
=kubectl_command
, exception_if_not_exists
=True)
84 # exception if helm is not installed
85 self
._helm
_command
= helm_command
86 self
._check
_file
_exists
(filename
=helm_command
, exception_if_not_exists
=True)
88 # obtain stable repo url from config or apply default
89 self
._stable
_repo
_url
= self
.config
.get("stablerepourl")
90 if self
._stable
_repo
_url
== "None":
91 self
._stable
_repo
_url
= None
93 # Lock to avoid concurrent execution of helm commands
94 self
.cmd_lock
= asyncio
.Lock()
96 def _get_namespace(self
, cluster_uuid
: str) -> str:
98 Obtains the namespace used by the cluster with the uuid passed by argument
100 param: cluster_uuid: cluster's uuid
103 # first, obtain the cluster corresponding to the uuid passed by argument
104 k8scluster
= self
.db
.get_one(
105 "k8sclusters", q_filter
={"_id": cluster_uuid
}, fail_on_empty
=False
107 return k8scluster
.get("namespace")
112 namespace
: str = "kube-system",
113 reuse_cluster_uuid
=None,
117 It prepares a given K8s cluster environment to run Charts
119 :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid
121 :param namespace: optional namespace to be used for helm. By default,
122 'kube-system' will be used
123 :param reuse_cluster_uuid: existing cluster uuid for reuse
124 :param kwargs: Additional parameters (None yet)
125 :return: uuid of the K8s cluster and True if connector has installed some
126 software in the cluster
127 (on error, an exception will be raised)
130 if reuse_cluster_uuid
:
131 cluster_id
= reuse_cluster_uuid
133 cluster_id
= str(uuid4())
136 "Initializing K8S Cluster {}. namespace: {}".format(cluster_id
, namespace
)
139 paths
, env
= self
._init
_paths
_env
(
140 cluster_name
=cluster_id
, create_if_not_exist
=True
142 mode
= stat
.S_IRUSR | stat
.S_IWUSR
143 with
open(paths
["kube_config"], "w", mode
) as f
:
145 os
.chmod(paths
["kube_config"], 0o600)
147 # Code with initialization specific of helm version
148 n2vc_installed_sw
= await self
._cluster
_init
(cluster_id
, namespace
, paths
, env
)
150 # sync fs with local data
151 self
.fs
.reverse_sync(from_path
=cluster_id
)
153 self
.log
.info("Cluster {} initialized".format(cluster_id
))
155 return cluster_id
, n2vc_installed_sw
162 repo_type
: str = "chart",
165 password
: str = None,
168 "Cluster {}, adding {} repository {}. URL: {}".format(
169 cluster_uuid
, repo_type
, name
, url
174 paths
, env
= self
._init
_paths
_env
(
175 cluster_name
=cluster_uuid
, create_if_not_exist
=True
179 self
.fs
.sync(from_path
=cluster_uuid
)
181 # helm repo add name url
182 command
= ("env KUBECONFIG={} {} repo add {} {}").format(
183 paths
["kube_config"], self
._helm
_command
, name
, url
187 temp_cert_file
= os
.path
.join(
188 self
.fs
.path
, "{}/helmcerts/".format(cluster_uuid
), "temp.crt"
190 os
.makedirs(os
.path
.dirname(temp_cert_file
), exist_ok
=True)
191 with
open(temp_cert_file
, "w") as the_cert
:
193 command
+= " --ca-file {}".format(temp_cert_file
)
196 command
+= " --username={}".format(user
)
199 command
+= " --password={}".format(password
)
201 self
.log
.debug("adding repo: {}".format(command
))
202 await self
._local
_async
_exec
(
203 command
=command
, raise_exception_on_error
=True, env
=env
207 command
= "env KUBECONFIG={} {} repo update {}".format(
208 paths
["kube_config"], self
._helm
_command
, name
210 self
.log
.debug("updating repo: {}".format(command
))
211 await self
._local
_async
_exec
(
212 command
=command
, raise_exception_on_error
=False, env
=env
216 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
218 async def repo_update(self
, cluster_uuid
: str, name
: str, repo_type
: str = "chart"):
220 "Cluster {}, updating {} repository {}".format(
221 cluster_uuid
, repo_type
, name
226 paths
, env
= self
._init
_paths
_env
(
227 cluster_name
=cluster_uuid
, create_if_not_exist
=True
231 self
.fs
.sync(from_path
=cluster_uuid
)
234 command
= "{} repo update {}".format(self
._helm
_command
, name
)
235 self
.log
.debug("updating repo: {}".format(command
))
236 await self
._local
_async
_exec
(
237 command
=command
, raise_exception_on_error
=False, env
=env
241 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
243 async def repo_list(self
, cluster_uuid
: str) -> list:
245 Get the list of registered repositories
247 :return: list of registered repositories: [ (name, url) .... ]
250 self
.log
.debug("list repositories for cluster {}".format(cluster_uuid
))
253 paths
, env
= self
._init
_paths
_env
(
254 cluster_name
=cluster_uuid
, create_if_not_exist
=True
258 self
.fs
.sync(from_path
=cluster_uuid
)
260 command
= "env KUBECONFIG={} {} repo list --output yaml".format(
261 paths
["kube_config"], self
._helm
_command
264 # Set exception to false because if there are no repos just want an empty list
265 output
, _rc
= await self
._local
_async
_exec
(
266 command
=command
, raise_exception_on_error
=False, env
=env
270 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
273 if output
and len(output
) > 0:
274 repos
= yaml
.load(output
, Loader
=yaml
.SafeLoader
)
275 # unify format between helm2 and helm3 setting all keys lowercase
276 return self
._lower
_keys
_list
(repos
)
282 async def repo_remove(self
, cluster_uuid
: str, name
: str):
284 "remove {} repositories for cluster {}".format(name
, cluster_uuid
)
288 paths
, env
= self
._init
_paths
_env
(
289 cluster_name
=cluster_uuid
, create_if_not_exist
=True
293 self
.fs
.sync(from_path
=cluster_uuid
)
295 command
= "env KUBECONFIG={} {} repo remove {}".format(
296 paths
["kube_config"], self
._helm
_command
, name
298 await self
._local
_async
_exec
(
299 command
=command
, raise_exception_on_error
=True, env
=env
303 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
309 uninstall_sw
: bool = False,
314 Resets the Kubernetes cluster by removing the helm deployment that represents it.
316 :param cluster_uuid: The UUID of the cluster to reset
317 :param force: Boolean to force the reset
318 :param uninstall_sw: Boolean to force the reset
319 :param kwargs: Additional parameters (None yet)
320 :return: Returns True if successful or raises an exception.
322 namespace
= self
._get
_namespace
(cluster_uuid
=cluster_uuid
)
324 "Resetting K8s environment. cluster uuid: {} uninstall={}".format(
325 cluster_uuid
, uninstall_sw
330 self
.fs
.sync(from_path
=cluster_uuid
)
332 # uninstall releases if needed.
334 releases
= await self
.instances_list(cluster_uuid
=cluster_uuid
)
335 if len(releases
) > 0:
339 kdu_instance
= r
.get("name")
340 chart
= r
.get("chart")
342 "Uninstalling {} -> {}".format(chart
, kdu_instance
)
344 await self
.uninstall(
345 cluster_uuid
=cluster_uuid
, kdu_instance
=kdu_instance
347 except Exception as e
:
348 # will not raise exception as it was found
349 # that in some cases of previously installed helm releases it
352 "Error uninstalling release {}: {}".format(
358 "Cluster uuid: {} has releases and not force. Leaving K8s helm environment"
359 ).format(cluster_uuid
)
362 False # Allow to remove k8s cluster without removing Tiller
366 await self
._uninstall
_sw
(cluster_id
=cluster_uuid
, namespace
=namespace
)
368 # delete cluster directory
369 self
.log
.debug("Removing directory {}".format(cluster_uuid
))
370 self
.fs
.file_delete(cluster_uuid
, ignore_non_exist
=True)
371 # Remove also local directorio if still exist
372 direct
= self
.fs
.path
+ "/" + cluster_uuid
373 shutil
.rmtree(direct
, ignore_errors
=True)
377 def _is_helm_chart_a_file(self
, chart_name
: str):
378 return chart_name
.count("/") > 1
380 async def _install_impl(
388 timeout
: float = 300,
390 db_dict
: dict = None,
391 kdu_name
: str = None,
392 namespace
: str = None,
395 paths
, env
= self
._init
_paths
_env
(
396 cluster_name
=cluster_id
, create_if_not_exist
=True
400 params_str
, file_to_delete
= self
._params
_to
_file
_option
(
401 cluster_id
=cluster_id
, params
=params
405 kdu_model
, version
= self
._split
_version
(kdu_model
)
407 repo
= self
._split
_repo
(kdu_model
)
409 await self
.repo_update(cluster_id
, repo
)
411 command
= self
._get
_install
_command
(
419 paths
["kube_config"],
422 self
.log
.debug("installing: {}".format(command
))
425 # exec helm in a task
426 exec_task
= asyncio
.ensure_future(
427 coro_or_future
=self
._local
_async
_exec
(
428 command
=command
, raise_exception_on_error
=False, env
=env
432 # write status in another task
433 status_task
= asyncio
.ensure_future(
434 coro_or_future
=self
._store
_status
(
435 cluster_id
=cluster_id
,
436 kdu_instance
=kdu_instance
,
443 # wait for execution task
444 await asyncio
.wait([exec_task
])
449 output
, rc
= exec_task
.result()
453 output
, rc
= await self
._local
_async
_exec
(
454 command
=command
, raise_exception_on_error
=False, env
=env
457 # remove temporal values yaml file
459 os
.remove(file_to_delete
)
462 await self
._store
_status
(
463 cluster_id
=cluster_id
,
464 kdu_instance
=kdu_instance
,
471 msg
= "Error executing command: {}\nOutput: {}".format(command
, output
)
473 raise K8sException(msg
)
479 kdu_model
: str = None,
481 timeout
: float = 300,
483 db_dict
: dict = None,
485 self
.log
.debug("upgrading {} in cluster {}".format(kdu_model
, cluster_uuid
))
488 self
.fs
.sync(from_path
=cluster_uuid
)
490 # look for instance to obtain namespace
491 instance_info
= await self
.get_instance_info(cluster_uuid
, kdu_instance
)
492 if not instance_info
:
493 raise K8sException("kdu_instance {} not found".format(kdu_instance
))
496 paths
, env
= self
._init
_paths
_env
(
497 cluster_name
=cluster_uuid
, create_if_not_exist
=True
501 self
.fs
.sync(from_path
=cluster_uuid
)
504 params_str
, file_to_delete
= self
._params
_to
_file
_option
(
505 cluster_id
=cluster_uuid
, params
=params
509 kdu_model
, version
= self
._split
_version
(kdu_model
)
511 repo
= self
._split
_repo
(kdu_model
)
513 await self
.repo_update(cluster_uuid
, repo
)
515 command
= self
._get
_upgrade
_command
(
518 instance_info
["namespace"],
523 paths
["kube_config"],
526 self
.log
.debug("upgrading: {}".format(command
))
530 # exec helm in a task
531 exec_task
= asyncio
.ensure_future(
532 coro_or_future
=self
._local
_async
_exec
(
533 command
=command
, raise_exception_on_error
=False, env
=env
536 # write status in another task
537 status_task
= asyncio
.ensure_future(
538 coro_or_future
=self
._store
_status
(
539 cluster_id
=cluster_uuid
,
540 kdu_instance
=kdu_instance
,
541 namespace
=instance_info
["namespace"],
547 # wait for execution task
548 await asyncio
.wait([exec_task
])
552 output
, rc
= exec_task
.result()
556 output
, rc
= await self
._local
_async
_exec
(
557 command
=command
, raise_exception_on_error
=False, env
=env
560 # remove temporal values yaml file
562 os
.remove(file_to_delete
)
565 await self
._store
_status
(
566 cluster_id
=cluster_uuid
,
567 kdu_instance
=kdu_instance
,
568 namespace
=instance_info
["namespace"],
574 msg
= "Error executing command: {}\nOutput: {}".format(command
, output
)
576 raise K8sException(msg
)
579 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
581 # return new revision number
582 instance
= await self
.get_instance_info(
583 cluster_uuid
=cluster_uuid
, kdu_instance
=kdu_instance
586 revision
= int(instance
.get("revision"))
587 self
.log
.debug("New revision: {}".format(revision
))
597 total_timeout
: float = 1800,
598 cluster_uuid
: str = None,
599 kdu_model
: str = None,
601 db_dict
: dict = None,
604 """Scale a resource in a Helm Chart.
607 kdu_instance: KDU instance name
608 scale: Scale to which to set the resource
609 resource_name: Resource name
610 total_timeout: The time, in seconds, to wait
611 cluster_uuid: The UUID of the cluster
612 kdu_model: The chart reference
613 atomic: if set, upgrade process rolls back changes made in case of failed upgrade.
614 The --wait flag will be set automatically if --atomic is used
615 db_dict: Dictionary for any additional data
616 kwargs: Additional parameters
619 True if successful, False otherwise
622 debug_mgs
= "scaling {} in cluster {}".format(kdu_model
, cluster_uuid
)
624 debug_mgs
= "scaling resource {} in model {} (cluster {})".format(
625 resource_name
, kdu_model
, cluster_uuid
628 self
.log
.debug(debug_mgs
)
630 # look for instance to obtain namespace
631 # get_instance_info function calls the sync command
632 instance_info
= await self
.get_instance_info(cluster_uuid
, kdu_instance
)
633 if not instance_info
:
634 raise K8sException("kdu_instance {} not found".format(kdu_instance
))
637 paths
, env
= self
._init
_paths
_env
(
638 cluster_name
=cluster_uuid
, create_if_not_exist
=True
642 kdu_model
, version
= self
._split
_version
(kdu_model
)
644 repo_url
= await self
._find
_repo
(kdu_model
, cluster_uuid
)
646 _
, replica_str
= await self
._get
_replica
_count
_url
(
647 kdu_model
, repo_url
, resource_name
650 command
= self
._get
_upgrade
_scale
_command
(
653 instance_info
["namespace"],
660 paths
["kube_config"],
663 self
.log
.debug("scaling: {}".format(command
))
666 # exec helm in a task
667 exec_task
= asyncio
.ensure_future(
668 coro_or_future
=self
._local
_async
_exec
(
669 command
=command
, raise_exception_on_error
=False, env
=env
672 # write status in another task
673 status_task
= asyncio
.ensure_future(
674 coro_or_future
=self
._store
_status
(
675 cluster_id
=cluster_uuid
,
676 kdu_instance
=kdu_instance
,
677 namespace
=instance_info
["namespace"],
683 # wait for execution task
684 await asyncio
.wait([exec_task
])
688 output
, rc
= exec_task
.result()
691 output
, rc
= await self
._local
_async
_exec
(
692 command
=command
, raise_exception_on_error
=False, env
=env
696 await self
._store
_status
(
697 cluster_id
=cluster_uuid
,
698 kdu_instance
=kdu_instance
,
699 namespace
=instance_info
["namespace"],
705 msg
= "Error executing command: {}\nOutput: {}".format(command
, output
)
707 raise K8sException(msg
)
710 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
714 async def get_scale_count(
722 """Get a resource scale count.
725 cluster_uuid: The UUID of the cluster
726 resource_name: Resource name
727 kdu_instance: KDU instance name
728 kdu_model: The name or path of an Helm Chart
729 kwargs: Additional parameters
732 Resource instance count
736 "getting scale count for {} in cluster {}".format(kdu_model
, cluster_uuid
)
739 # look for instance to obtain namespace
740 instance_info
= await self
.get_instance_info(cluster_uuid
, kdu_instance
)
741 if not instance_info
:
742 raise K8sException("kdu_instance {} not found".format(kdu_instance
))
745 paths
, env
= self
._init
_paths
_env
(
746 cluster_name
=cluster_uuid
, create_if_not_exist
=True
749 replicas
= await self
._get
_replica
_count
_instance
(
750 kdu_instance
=kdu_instance
,
751 namespace
=instance_info
["namespace"],
752 kubeconfig
=paths
["kube_config"],
753 resource_name
=resource_name
,
756 # Get default value if scale count is not found from provided values
758 repo_url
= await self
._find
_repo
(
759 kdu_model
=kdu_model
, cluster_uuid
=cluster_uuid
761 replicas
, _
= await self
._get
_replica
_count
_url
(
762 kdu_model
=kdu_model
, repo_url
=repo_url
, resource_name
=resource_name
766 msg
= "Replica count not found. Cannot be scaled"
768 raise K8sException(msg
)
773 self
, cluster_uuid
: str, kdu_instance
: str, revision
=0, db_dict
: dict = None
776 "rollback kdu_instance {} to revision {} from cluster {}".format(
777 kdu_instance
, revision
, cluster_uuid
782 self
.fs
.sync(from_path
=cluster_uuid
)
784 # look for instance to obtain namespace
785 instance_info
= await self
.get_instance_info(cluster_uuid
, kdu_instance
)
786 if not instance_info
:
787 raise K8sException("kdu_instance {} not found".format(kdu_instance
))
790 paths
, env
= self
._init
_paths
_env
(
791 cluster_name
=cluster_uuid
, create_if_not_exist
=True
795 self
.fs
.sync(from_path
=cluster_uuid
)
797 command
= self
._get
_rollback
_command
(
798 kdu_instance
, instance_info
["namespace"], revision
, paths
["kube_config"]
801 self
.log
.debug("rolling_back: {}".format(command
))
803 # exec helm in a task
804 exec_task
= asyncio
.ensure_future(
805 coro_or_future
=self
._local
_async
_exec
(
806 command
=command
, raise_exception_on_error
=False, env
=env
809 # write status in another task
810 status_task
= asyncio
.ensure_future(
811 coro_or_future
=self
._store
_status
(
812 cluster_id
=cluster_uuid
,
813 kdu_instance
=kdu_instance
,
814 namespace
=instance_info
["namespace"],
816 operation
="rollback",
820 # wait for execution task
821 await asyncio
.wait([exec_task
])
826 output
, rc
= exec_task
.result()
829 await self
._store
_status
(
830 cluster_id
=cluster_uuid
,
831 kdu_instance
=kdu_instance
,
832 namespace
=instance_info
["namespace"],
834 operation
="rollback",
838 msg
= "Error executing command: {}\nOutput: {}".format(command
, output
)
840 raise K8sException(msg
)
843 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
845 # return new revision number
846 instance
= await self
.get_instance_info(
847 cluster_uuid
=cluster_uuid
, kdu_instance
=kdu_instance
850 revision
= int(instance
.get("revision"))
851 self
.log
.debug("New revision: {}".format(revision
))
856 async def uninstall(self
, cluster_uuid
: str, kdu_instance
: str, **kwargs
):
858 Removes an existing KDU instance. It would implicitly use the `delete` or 'uninstall' call
859 (this call should happen after all _terminate-config-primitive_ of the VNF
862 :param cluster_uuid: UUID of a K8s cluster known by OSM, or namespace:cluster_id
863 :param kdu_instance: unique name for the KDU instance to be deleted
864 :param kwargs: Additional parameters (None yet)
865 :return: True if successful
869 "uninstall kdu_instance {} from cluster {}".format(
870 kdu_instance
, cluster_uuid
875 self
.fs
.sync(from_path
=cluster_uuid
)
877 # look for instance to obtain namespace
878 instance_info
= await self
.get_instance_info(cluster_uuid
, kdu_instance
)
879 if not instance_info
:
880 self
.log
.warning(("kdu_instance {} not found".format(kdu_instance
)))
883 paths
, env
= self
._init
_paths
_env
(
884 cluster_name
=cluster_uuid
, create_if_not_exist
=True
888 self
.fs
.sync(from_path
=cluster_uuid
)
890 command
= self
._get
_uninstall
_command
(
891 kdu_instance
, instance_info
["namespace"], paths
["kube_config"]
893 output
, _rc
= await self
._local
_async
_exec
(
894 command
=command
, raise_exception_on_error
=True, env
=env
898 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
900 return self
._output
_to
_table
(output
)
902 async def instances_list(self
, cluster_uuid
: str) -> list:
904 returns a list of deployed releases in a cluster
906 :param cluster_uuid: the 'cluster' or 'namespace:cluster'
910 self
.log
.debug("list releases for cluster {}".format(cluster_uuid
))
913 self
.fs
.sync(from_path
=cluster_uuid
)
915 # execute internal command
916 result
= await self
._instances
_list
(cluster_uuid
)
919 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
923 async def get_instance_info(self
, cluster_uuid
: str, kdu_instance
: str):
924 instances
= await self
.instances_list(cluster_uuid
=cluster_uuid
)
925 for instance
in instances
:
926 if instance
.get("name") == kdu_instance
:
928 self
.log
.debug("Instance {} not found".format(kdu_instance
))
931 async def upgrade_charm(
935 charm_id
: str = None,
936 charm_type
: str = None,
937 timeout
: float = None,
939 """This method upgrade charms in VNFs
942 ee_id: Execution environment id
943 path: Local path to the charm
945 charm_type: Charm type can be lxc-proxy-charm, native-charm or k8s-proxy-charm
946 timeout: (Float) Timeout for the ns update operation
949 The output of the update operation if status equals to "completed"
951 raise K8sException("KDUs deployed with Helm do not support charm upgrade")
953 async def exec_primitive(
955 cluster_uuid
: str = None,
956 kdu_instance
: str = None,
957 primitive_name
: str = None,
958 timeout
: float = 300,
960 db_dict
: dict = None,
963 """Exec primitive (Juju action)
965 :param cluster_uuid: The UUID of the cluster or namespace:cluster
966 :param kdu_instance: The unique name of the KDU instance
967 :param primitive_name: Name of action that will be executed
968 :param timeout: Timeout for action execution
969 :param params: Dictionary of all the parameters needed for the action
970 :db_dict: Dictionary for any additional data
971 :param kwargs: Additional parameters (None yet)
973 :return: Returns the output of the action
976 "KDUs deployed with Helm don't support actions "
977 "different from rollback, upgrade and status"
980 async def get_services(
981 self
, cluster_uuid
: str, kdu_instance
: str, namespace
: str
984 Returns a list of services defined for the specified kdu instance.
986 :param cluster_uuid: UUID of a K8s cluster known by OSM
987 :param kdu_instance: unique name for the KDU instance
988 :param namespace: K8s namespace used by the KDU instance
989 :return: If successful, it will return a list of services, Each service
990 can have the following data:
991 - `name` of the service
992 - `type` type of service in the k8 cluster
993 - `ports` List of ports offered by the service, for each port includes at least
995 - `cluster_ip` Internal ip to be used inside k8s cluster
996 - `external_ip` List of external ips (in case they are available)
1000 "get_services: cluster_uuid: {}, kdu_instance: {}".format(
1001 cluster_uuid
, kdu_instance
1006 paths
, env
= self
._init
_paths
_env
(
1007 cluster_name
=cluster_uuid
, create_if_not_exist
=True
1011 self
.fs
.sync(from_path
=cluster_uuid
)
1013 # get list of services names for kdu
1014 service_names
= await self
._get
_services
(
1015 cluster_uuid
, kdu_instance
, namespace
, paths
["kube_config"]
1019 for service
in service_names
:
1020 service
= await self
._get
_service
(cluster_uuid
, service
, namespace
)
1021 service_list
.append(service
)
1024 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
1028 async def get_service(
1029 self
, cluster_uuid
: str, service_name
: str, namespace
: str
1033 "get service, service_name: {}, namespace: {}, cluster_uuid: {}".format(
1034 service_name
, namespace
, cluster_uuid
1039 self
.fs
.sync(from_path
=cluster_uuid
)
1041 service
= await self
._get
_service
(cluster_uuid
, service_name
, namespace
)
1044 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
1048 async def status_kdu(
1049 self
, cluster_uuid
: str, kdu_instance
: str, yaml_format
: str = False, **kwargs
1050 ) -> Union
[str, dict]:
1052 This call would retrieve tha current state of a given KDU instance. It would be
1053 would allow to retrieve the _composition_ (i.e. K8s objects) and _specific
1054 values_ of the configuration parameters applied to a given instance. This call
1055 would be based on the `status` call.
1057 :param cluster_uuid: UUID of a K8s cluster known by OSM
1058 :param kdu_instance: unique name for the KDU instance
1059 :param kwargs: Additional parameters (None yet)
1060 :param yaml_format: if the return shall be returned as an YAML string or as a
1062 :return: If successful, it will return the following vector of arguments:
1063 - K8s `namespace` in the cluster where the KDU lives
1064 - `state` of the KDU instance. It can be:
1071 - List of `resources` (objects) that this release consists of, sorted by kind,
1072 and the status of those resources
1073 - Last `deployment_time`.
1077 "status_kdu: cluster_uuid: {}, kdu_instance: {}".format(
1078 cluster_uuid
, kdu_instance
1083 self
.fs
.sync(from_path
=cluster_uuid
)
1085 # get instance: needed to obtain namespace
1086 instances
= await self
._instances
_list
(cluster_id
=cluster_uuid
)
1087 for instance
in instances
:
1088 if instance
.get("name") == kdu_instance
:
1091 # instance does not exist
1093 "Instance name: {} not found in cluster: {}".format(
1094 kdu_instance
, cluster_uuid
1098 status
= await self
._status
_kdu
(
1099 cluster_id
=cluster_uuid
,
1100 kdu_instance
=kdu_instance
,
1101 namespace
=instance
["namespace"],
1102 yaml_format
=yaml_format
,
1103 show_error_log
=True,
1107 self
.fs
.reverse_sync(from_path
=cluster_uuid
)
1111 async def get_values_kdu(
1112 self
, kdu_instance
: str, namespace
: str, kubeconfig
: str
1115 self
.log
.debug("get kdu_instance values {}".format(kdu_instance
))
1117 return await self
._exec
_get
_command
(
1118 get_command
="values",
1119 kdu_instance
=kdu_instance
,
1120 namespace
=namespace
,
1121 kubeconfig
=kubeconfig
,
1124 async def values_kdu(self
, kdu_model
: str, repo_url
: str = None) -> str:
1125 """Method to obtain the Helm Chart package's values
1128 kdu_model: The name or path of an Helm Chart
1129 repo_url: Helm Chart repository url
1132 str: the values of the Helm Chart package
1136 "inspect kdu_model values {} from (optional) repo: {}".format(
1141 return await self
._exec
_inspect
_command
(
1142 inspect_command
="values", kdu_model
=kdu_model
, repo_url
=repo_url
1145 async def help_kdu(self
, kdu_model
: str, repo_url
: str = None) -> str:
1148 "inspect kdu_model {} readme.md from repo: {}".format(kdu_model
, repo_url
)
1151 return await self
._exec
_inspect
_command
(
1152 inspect_command
="readme", kdu_model
=kdu_model
, repo_url
=repo_url
1155 async def synchronize_repos(self
, cluster_uuid
: str):
1157 self
.log
.debug("synchronize repos for cluster helm-id: {}".format(cluster_uuid
))
1159 db_repo_ids
= self
._get
_helm
_chart
_repos
_ids
(cluster_uuid
)
1160 db_repo_dict
= self
._get
_db
_repos
_dict
(db_repo_ids
)
1162 local_repo_list
= await self
.repo_list(cluster_uuid
)
1163 local_repo_dict
= {repo
["name"]: repo
["url"] for repo
in local_repo_list
}
1165 deleted_repo_list
= []
1166 added_repo_dict
= {}
1168 # iterate over the list of repos in the database that should be
1169 # added if not present
1170 for repo_name
, db_repo
in db_repo_dict
.items():
1172 # check if it is already present
1173 curr_repo_url
= local_repo_dict
.get(db_repo
["name"])
1174 repo_id
= db_repo
.get("_id")
1175 if curr_repo_url
!= db_repo
["url"]:
1178 "repo {} url changed, delete and and again".format(
1182 await self
.repo_remove(cluster_uuid
, db_repo
["name"])
1183 deleted_repo_list
.append(repo_id
)
1186 self
.log
.debug("add repo {}".format(db_repo
["name"]))
1187 if "ca_cert" in db_repo
:
1188 await self
.repo_add(
1192 cert
=db_repo
["ca_cert"],
1195 await self
.repo_add(
1200 added_repo_dict
[repo_id
] = db_repo
["name"]
1201 except Exception as e
:
1203 "Error adding repo id: {}, err_msg: {} ".format(
1208 # Delete repos that are present but not in nbi_list
1209 for repo_name
in local_repo_dict
:
1210 if not db_repo_dict
.get(repo_name
) and repo_name
!= "stable":
1211 self
.log
.debug("delete repo {}".format(repo_name
))
1213 await self
.repo_remove(cluster_uuid
, repo_name
)
1214 deleted_repo_list
.append(repo_name
)
1215 except Exception as e
:
1217 "Error deleting repo, name: {}, err_msg: {}".format(
1222 return deleted_repo_list
, added_repo_dict
1224 except K8sException
:
1226 except Exception as e
:
1227 # Do not raise errors synchronizing repos
1228 self
.log
.error("Error synchronizing repos: {}".format(e
))
1229 raise Exception("Error synchronizing repos: {}".format(e
))
1231 def _get_db_repos_dict(self
, repo_ids
: list):
1233 for repo_id
in repo_ids
:
1234 db_repo
= self
.db
.get_one("k8srepos", {"_id": repo_id
})
1235 db_repos_dict
[db_repo
["name"]] = db_repo
1236 return db_repos_dict
1239 ####################################################################################
1240 ################################### TO BE IMPLEMENTED SUBCLASSES ###################
1241 ####################################################################################
1245 def _init_paths_env(self
, cluster_name
: str, create_if_not_exist
: bool = True):
1247 Creates and returns base cluster and kube dirs and returns them.
1248 Also created helm3 dirs according to new directory specification, paths are
1249 not returned but assigned to helm environment variables
1251 :param cluster_name: cluster_name
1252 :return: Dictionary with config_paths and dictionary with helm environment variables
1256 async def _cluster_init(self
, cluster_id
, namespace
, paths
, env
):
1258 Implements the helm version dependent cluster initialization
1262 async def _instances_list(self
, cluster_id
):
1264 Implements the helm version dependent helm instances list
1268 async def _get_services(self
, cluster_id
, kdu_instance
, namespace
, kubeconfig
):
1270 Implements the helm version dependent method to obtain services from a helm instance
1274 async def _status_kdu(
1278 namespace
: str = None,
1279 yaml_format
: bool = False,
1280 show_error_log
: bool = False,
1281 ) -> Union
[str, dict]:
1283 Implements the helm version dependent method to obtain status of a helm instance
1287 def _get_install_command(
1299 Obtain command to be executed to delete the indicated instance
1303 def _get_upgrade_scale_command(
1316 """Generates the command to scale a Helm Chart release
1319 kdu_model (str): Kdu model name, corresponding to the Helm local location or repository
1320 kdu_instance (str): KDU instance, corresponding to the Helm Chart release in question
1321 namespace (str): Namespace where this KDU instance is deployed
1322 scale (int): Scale count
1323 version (str): Constraint with specific version of the Chart to use
1324 atomic (bool): If set, upgrade process rolls back changes made in case of failed upgrade.
1325 The --wait flag will be set automatically if --atomic is used
1326 replica_str (str): The key under resource_name key where the scale count is stored
1327 timeout (float): The time, in seconds, to wait
1328 resource_name (str): The KDU's resource to scale
1329 kubeconfig (str): Kubeconfig file path
1332 str: command to scale a Helm Chart release
1336 def _get_upgrade_command(
1347 """Generates the command to upgrade a Helm Chart release
1350 kdu_model (str): Kdu model name, corresponding to the Helm local location or repository
1351 kdu_instance (str): KDU instance, corresponding to the Helm Chart release in question
1352 namespace (str): Namespace where this KDU instance is deployed
1353 params_str (str): Params used to upgrade the Helm Chart release
1354 version (str): Constraint with specific version of the Chart to use
1355 atomic (bool): If set, upgrade process rolls back changes made in case of failed upgrade.
1356 The --wait flag will be set automatically if --atomic is used
1357 timeout (float): The time, in seconds, to wait
1358 kubeconfig (str): Kubeconfig file path
1361 str: command to upgrade a Helm Chart release
1365 def _get_rollback_command(
1366 self
, kdu_instance
, namespace
, revision
, kubeconfig
1369 Obtain command to be executed to rollback the indicated instance
1373 def _get_uninstall_command(
1374 self
, kdu_instance
: str, namespace
: str, kubeconfig
: str
1377 Obtain command to be executed to delete the indicated instance
1381 def _get_inspect_command(
1382 self
, show_command
: str, kdu_model
: str, repo_str
: str, version
: str
1384 """Generates the command to obtain the information about an Helm Chart package
1385 (´helm show ...´ command)
1388 show_command: the second part of the command (`helm show <show_command>`)
1389 kdu_model: The name or path of an Helm Chart
1390 repo_url: Helm Chart repository url
1391 version: constraint with specific version of the Chart to use
1394 str: the generated Helm Chart command
1398 def _get_get_command(
1399 self
, get_command
: str, kdu_instance
: str, namespace
: str, kubeconfig
: str
1401 """Obtain command to be executed to get information about the kdu instance."""
1404 async def _uninstall_sw(self
, cluster_id
: str, namespace
: str):
1406 Method call to uninstall cluster software for helm. This method is dependent
1408 For Helm v2 it will be called when Tiller must be uninstalled
1409 For Helm v3 it does nothing and does not need to be callled
1413 def _get_helm_chart_repos_ids(self
, cluster_uuid
) -> list:
1415 Obtains the cluster repos identifiers
1419 ####################################################################################
1420 ################################### P R I V A T E ##################################
1421 ####################################################################################
1425 def _check_file_exists(filename
: str, exception_if_not_exists
: bool = False):
1426 if os
.path
.exists(filename
):
1429 msg
= "File {} does not exist".format(filename
)
1430 if exception_if_not_exists
:
1431 raise K8sException(msg
)
1434 def _remove_multiple_spaces(strobj
):
1435 strobj
= strobj
.strip()
1436 while " " in strobj
:
1437 strobj
= strobj
.replace(" ", " ")
1441 def _output_to_lines(output
: str) -> list:
1442 output_lines
= list()
1443 lines
= output
.splitlines(keepends
=False)
1447 output_lines
.append(line
)
1451 def _output_to_table(output
: str) -> list:
1452 output_table
= list()
1453 lines
= output
.splitlines(keepends
=False)
1455 line
= line
.replace("\t", " ")
1457 output_table
.append(line_list
)
1458 cells
= line
.split(sep
=" ")
1462 line_list
.append(cell
)
1466 def _parse_services(output
: str) -> list:
1467 lines
= output
.splitlines(keepends
=False)
1470 line
= line
.replace("\t", " ")
1471 cells
= line
.split(sep
=" ")
1472 if len(cells
) > 0 and cells
[0].startswith("service/"):
1473 elems
= cells
[0].split(sep
="/")
1475 services
.append(elems
[1])
1479 def _get_deep(dictionary
: dict, members
: tuple):
1484 value
= target
.get(m
)
1493 # find key:value in several lines
1495 def _find_in_lines(p_lines
: list, p_key
: str) -> str:
1496 for line
in p_lines
:
1498 if line
.startswith(p_key
+ ":"):
1499 parts
= line
.split(":")
1500 the_value
= parts
[1].strip()
1508 def _lower_keys_list(input_list
: list):
1510 Transform the keys in a list of dictionaries to lower case and returns a new list
1515 for dictionary
in input_list
:
1516 new_dict
= dict((k
.lower(), v
) for k
, v
in dictionary
.items())
1517 new_list
.append(new_dict
)
1520 async def _local_async_exec(
1523 raise_exception_on_error
: bool = False,
1524 show_error_log
: bool = True,
1525 encode_utf8
: bool = False,
1529 command
= K8sHelmBaseConnector
._remove
_multiple
_spaces
(command
)
1531 "Executing async local command: {}, env: {}".format(command
, env
)
1535 command
= shlex
.split(command
)
1537 environ
= os
.environ
.copy()
1542 async with self
.cmd_lock
:
1543 process
= await asyncio
.create_subprocess_exec(
1545 stdout
=asyncio
.subprocess
.PIPE
,
1546 stderr
=asyncio
.subprocess
.PIPE
,
1550 # wait for command terminate
1551 stdout
, stderr
= await process
.communicate()
1553 return_code
= process
.returncode
1557 output
= stdout
.decode("utf-8").strip()
1558 # output = stdout.decode()
1560 output
= stderr
.decode("utf-8").strip()
1561 # output = stderr.decode()
1563 if return_code
!= 0 and show_error_log
:
1565 "Return code (FAIL): {}\nOutput:\n{}".format(return_code
, output
)
1568 self
.log
.debug("Return code: {}".format(return_code
))
1570 if raise_exception_on_error
and return_code
!= 0:
1571 raise K8sException(output
)
1574 output
= output
.encode("utf-8").strip()
1575 output
= str(output
).replace("\\n", "\n")
1577 return output
, return_code
1579 except asyncio
.CancelledError
:
1581 except K8sException
:
1583 except Exception as e
:
1584 msg
= "Exception executing command: {} -> {}".format(command
, e
)
1586 if raise_exception_on_error
:
1587 raise K8sException(e
) from e
1591 async def _local_async_exec_pipe(
1595 raise_exception_on_error
: bool = True,
1596 show_error_log
: bool = True,
1597 encode_utf8
: bool = False,
1601 command1
= K8sHelmBaseConnector
._remove
_multiple
_spaces
(command1
)
1602 command2
= K8sHelmBaseConnector
._remove
_multiple
_spaces
(command2
)
1603 command
= "{} | {}".format(command1
, command2
)
1605 "Executing async local command: {}, env: {}".format(command
, env
)
1609 command1
= shlex
.split(command1
)
1610 command2
= shlex
.split(command2
)
1612 environ
= os
.environ
.copy()
1617 async with self
.cmd_lock
:
1618 read
, write
= os
.pipe()
1619 await asyncio
.create_subprocess_exec(
1620 *command1
, stdout
=write
, env
=environ
1623 process_2
= await asyncio
.create_subprocess_exec(
1624 *command2
, stdin
=read
, stdout
=asyncio
.subprocess
.PIPE
, env
=environ
1627 stdout
, stderr
= await process_2
.communicate()
1629 return_code
= process_2
.returncode
1633 output
= stdout
.decode("utf-8").strip()
1634 # output = stdout.decode()
1636 output
= stderr
.decode("utf-8").strip()
1637 # output = stderr.decode()
1639 if return_code
!= 0 and show_error_log
:
1641 "Return code (FAIL): {}\nOutput:\n{}".format(return_code
, output
)
1644 self
.log
.debug("Return code: {}".format(return_code
))
1646 if raise_exception_on_error
and return_code
!= 0:
1647 raise K8sException(output
)
1650 output
= output
.encode("utf-8").strip()
1651 output
= str(output
).replace("\\n", "\n")
1653 return output
, return_code
1654 except asyncio
.CancelledError
:
1656 except K8sException
:
1658 except Exception as e
:
1659 msg
= "Exception executing command: {} -> {}".format(command
, e
)
1661 if raise_exception_on_error
:
1662 raise K8sException(e
) from e
1666 async def _get_service(self
, cluster_id
, service_name
, namespace
):
1668 Obtains the data of the specified service in the k8cluster.
1670 :param cluster_id: id of a K8s cluster known by OSM
1671 :param service_name: name of the K8s service in the specified namespace
1672 :param namespace: K8s namespace used by the KDU instance
1673 :return: If successful, it will return a service with the following data:
1674 - `name` of the service
1675 - `type` type of service in the k8 cluster
1676 - `ports` List of ports offered by the service, for each port includes at least
1677 name, port, protocol
1678 - `cluster_ip` Internal ip to be used inside k8s cluster
1679 - `external_ip` List of external ips (in case they are available)
1683 paths
, env
= self
._init
_paths
_env
(
1684 cluster_name
=cluster_id
, create_if_not_exist
=True
1687 command
= "{} --kubeconfig={} --namespace={} get service {} -o=yaml".format(
1688 self
.kubectl_command
, paths
["kube_config"], namespace
, service_name
1691 output
, _rc
= await self
._local
_async
_exec
(
1692 command
=command
, raise_exception_on_error
=True, env
=env
1695 data
= yaml
.load(output
, Loader
=yaml
.SafeLoader
)
1698 "name": service_name
,
1699 "type": self
._get
_deep
(data
, ("spec", "type")),
1700 "ports": self
._get
_deep
(data
, ("spec", "ports")),
1701 "cluster_ip": self
._get
_deep
(data
, ("spec", "clusterIP")),
1703 if service
["type"] == "LoadBalancer":
1704 ip_map_list
= self
._get
_deep
(data
, ("status", "loadBalancer", "ingress"))
1705 ip_list
= [elem
["ip"] for elem
in ip_map_list
]
1706 service
["external_ip"] = ip_list
1710 async def _exec_get_command(
1711 self
, get_command
: str, kdu_instance
: str, namespace
: str, kubeconfig
: str
1713 """Obtains information about the kdu instance."""
1715 full_command
= self
._get
_get
_command
(
1716 get_command
, kdu_instance
, namespace
, kubeconfig
1719 output
, _rc
= await self
._local
_async
_exec
(command
=full_command
)
1723 async def _exec_inspect_command(
1724 self
, inspect_command
: str, kdu_model
: str, repo_url
: str = None
1726 """Obtains information about an Helm Chart package (´helm show´ command)
1729 inspect_command: the Helm sub command (`helm show <inspect_command> ...`)
1730 kdu_model: The name or path of an Helm Chart
1731 repo_url: Helm Chart repository url
1734 str: the requested info about the Helm Chart package
1739 repo_str
= " --repo {}".format(repo_url
)
1741 idx
= kdu_model
.find("/")
1744 kdu_model
= kdu_model
[idx
:]
1746 kdu_model
, version
= self
._split
_version
(kdu_model
)
1748 version_str
= "--version {}".format(version
)
1752 full_command
= self
._get
_inspect
_command
(
1753 inspect_command
, kdu_model
, repo_str
, version_str
1756 output
, _rc
= await self
._local
_async
_exec
(command
=full_command
)
1760 async def _get_replica_count_url(
1763 repo_url
: str = None,
1764 resource_name
: str = None,
1766 """Get the replica count value in the Helm Chart Values.
1769 kdu_model: The name or path of an Helm Chart
1770 repo_url: Helm Chart repository url
1771 resource_name: Resource name
1774 True if replicas, False replicaCount
1777 kdu_values
= yaml
.load(
1778 await self
.values_kdu(kdu_model
=kdu_model
, repo_url
=repo_url
),
1779 Loader
=yaml
.SafeLoader
,
1784 "kdu_values not found for kdu_model {}".format(kdu_model
)
1788 kdu_values
= kdu_values
.get(resource_name
, None)
1791 msg
= "resource {} not found in the values in model {}".format(
1792 resource_name
, kdu_model
1795 raise K8sException(msg
)
1797 duplicate_check
= False
1802 if kdu_values
.get("replicaCount", None):
1803 replicas
= kdu_values
["replicaCount"]
1804 replica_str
= "replicaCount"
1805 elif kdu_values
.get("replicas", None):
1806 duplicate_check
= True
1807 replicas
= kdu_values
["replicas"]
1808 replica_str
= "replicas"
1812 "replicaCount or replicas not found in the resource"
1813 "{} values in model {}. Cannot be scaled".format(
1814 resource_name
, kdu_model
1819 "replicaCount or replicas not found in the values"
1820 "in model {}. Cannot be scaled".format(kdu_model
)
1823 raise K8sException(msg
)
1825 # Control if replicas and replicaCount exists at the same time
1826 msg
= "replicaCount and replicas are exists at the same time"
1828 if "replicaCount" in kdu_values
:
1830 raise K8sException(msg
)
1832 if "replicas" in kdu_values
:
1834 raise K8sException(msg
)
1836 return replicas
, replica_str
1838 async def _get_replica_count_instance(
1843 resource_name
: str = None,
1845 """Get the replica count value in the instance.
1848 kdu_instance: The name of the KDU instance
1849 namespace: KDU instance namespace
1851 resource_name: Resource name
1854 True if replicas, False replicaCount
1857 kdu_values
= yaml
.load(
1858 await self
.get_values_kdu(kdu_instance
, namespace
, kubeconfig
),
1859 Loader
=yaml
.SafeLoader
,
1866 kdu_values
.get(resource_name
, None) if resource_name
else None
1870 resource_values
.get("replicaCount", None)
1871 or resource_values
.get("replicas", None)
1875 kdu_values
.get("replicaCount", None)
1876 or kdu_values
.get("replicas", None)
1882 async def _store_status(
1887 namespace
: str = None,
1888 db_dict
: dict = None,
1891 Obtains the status of the KDU instance based on Helm Charts, and stores it in the database.
1893 :param cluster_id (str): the cluster where the KDU instance is deployed
1894 :param operation (str): The operation related to the status to be updated (for instance, "install" or "upgrade")
1895 :param kdu_instance (str): The KDU instance in relation to which the status is obtained
1896 :param namespace (str): The Kubernetes namespace where the KDU instance was deployed. Defaults to None
1897 :param db_dict (dict): A dictionary with the database necessary information. It shall contain the
1898 values for the keys:
1899 - "collection": The Mongo DB collection to write to
1900 - "filter": The query filter to use in the update process
1901 - "path": The dot separated keys which targets the object to be updated
1906 detailed_status
= await self
._status
_kdu
(
1907 cluster_id
=cluster_id
,
1908 kdu_instance
=kdu_instance
,
1910 namespace
=namespace
,
1913 status
= detailed_status
.get("info").get("description")
1914 self
.log
.debug(f
"Status for KDU {kdu_instance} obtained: {status}.")
1916 # write status to db
1917 result
= await self
.write_app_status_to_db(
1920 detailed_status
=str(detailed_status
),
1921 operation
=operation
,
1925 self
.log
.info("Error writing in database. Task exiting...")
1927 except asyncio
.CancelledError
as e
:
1929 f
"Exception in method {self._store_status.__name__} (task cancelled): {e}"
1931 except Exception as e
:
1932 self
.log
.warning(f
"Exception in method {self._store_status.__name__}: {e}")
1934 # params for use in -f file
1935 # returns values file option and filename (in order to delete it at the end)
1936 def _params_to_file_option(self
, cluster_id
: str, params
: dict) -> (str, str):
1938 if params
and len(params
) > 0:
1939 self
._init
_paths
_env
(cluster_name
=cluster_id
, create_if_not_exist
=True)
1941 def get_random_number():
1942 r
= random
.randrange(start
=1, stop
=99999999)
1950 value
= params
.get(key
)
1951 if "!!yaml" in str(value
):
1952 value
= yaml
.safe_load(value
[7:])
1953 params2
[key
] = value
1955 values_file
= get_random_number() + ".yaml"
1956 with
open(values_file
, "w") as stream
:
1957 yaml
.dump(params2
, stream
, indent
=4, default_flow_style
=False)
1959 return "-f {}".format(values_file
), values_file
1963 # params for use in --set option
1965 def _params_to_set_option(params
: dict) -> str:
1967 if params
and len(params
) > 0:
1970 value
= params
.get(key
, None)
1971 if value
is not None:
1973 params_str
+= "--set "
1977 params_str
+= "{}={}".format(key
, value
)
1981 def generate_kdu_instance_name(**kwargs
):
1982 chart_name
= kwargs
["kdu_model"]
1983 # check embeded chart (file or dir)
1984 if chart_name
.startswith("/"):
1985 # extract file or directory name
1986 chart_name
= chart_name
[chart_name
.rfind("/") + 1 :]
1988 elif "://" in chart_name
:
1989 # extract last portion of URL
1990 chart_name
= chart_name
[chart_name
.rfind("/") + 1 :]
1993 for c
in chart_name
:
1994 if c
.isalpha() or c
.isnumeric():
2001 # if does not start with alpha character, prefix 'a'
2002 if not name
[0].isalpha():
2007 def get_random_number():
2008 r
= random
.randrange(start
=1, stop
=99999999)
2010 s
= s
.rjust(10, "0")
2013 name
= name
+ get_random_number()
2016 def _split_version(self
, kdu_model
: str) -> (str, str):
2018 if not self
._is
_helm
_chart
_a
_file
(kdu_model
) and ":" in kdu_model
:
2019 parts
= kdu_model
.split(sep
=":")
2021 version
= str(parts
[1])
2022 kdu_model
= parts
[0]
2023 return kdu_model
, version
2025 def _split_repo(self
, kdu_model
: str) -> str:
2027 idx
= kdu_model
.find("/")
2029 repo_name
= kdu_model
[:idx
]
2032 async def _find_repo(self
, kdu_model
: str, cluster_uuid
: str) -> str:
2033 """Obtain the Helm repository for an Helm Chart
2036 kdu_model (str): the KDU model associated with the Helm Chart instantiation
2037 cluster_uuid (str): The cluster UUID associated with the Helm Chart instantiation
2040 str: the repository URL; if Helm Chart is a local one, the function returns None
2044 idx
= kdu_model
.find("/")
2046 repo_name
= kdu_model
[:idx
]
2047 # Find repository link
2048 local_repo_list
= await self
.repo_list(cluster_uuid
)
2049 for repo
in local_repo_list
:
2050 repo_url
= repo
["url"] if repo
["name"] == repo_name
else None