Remove unsecure and unused function
[osm/N2VC.git] / n2vc / k8s_helm3_conn.py
1 ##
2 # Copyright 2019 Telefonica Investigacion y Desarrollo, S.A.U.
3 # This file is part of OSM
4 # All Rights Reserved.
5 #
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
9 #
10 # http://www.apache.org/licenses/LICENSE-2.0
11 #
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
15 # implied.
16 # See the License for the specific language governing permissions and
17 # limitations under the License.
18 #
19 # For those usages not covered by the Apache License, Version 2.0 please
20 # contact with: nfvlabs@tid.es
21 ##
22 import os
23 import yaml
24
25 from n2vc.k8s_helm_base_conn import K8sHelmBaseConnector
26 from n2vc.exceptions import K8sException
27
28
29 class K8sHelm3Connector(K8sHelmBaseConnector):
30
31 """
32 ####################################################################################
33 ################################### P U B L I C ####################################
34 ####################################################################################
35 """
36
37 def __init__(
38 self,
39 fs: object,
40 db: object,
41 kubectl_command: str = "/usr/bin/kubectl",
42 helm_command: str = "/usr/bin/helm3",
43 log: object = None,
44 on_update_db=None,
45 ):
46 """
47 Initializes helm connector for helm v3
48
49 :param fs: file system for kubernetes and helm configuration
50 :param db: database object to write current operation status
51 :param kubectl_command: path to kubectl executable
52 :param helm_command: path to helm executable
53 :param log: logger
54 :param on_update_db: callback called when k8s connector updates database
55 """
56
57 # parent class
58 K8sHelmBaseConnector.__init__(
59 self,
60 db=db,
61 log=log,
62 fs=fs,
63 kubectl_command=kubectl_command,
64 helm_command=helm_command,
65 on_update_db=on_update_db,
66 )
67
68 self.log.info("K8S Helm3 connector initialized")
69
70 async def install(
71 self,
72 cluster_uuid: str,
73 kdu_model: str,
74 kdu_instance: str,
75 atomic: bool = True,
76 timeout: float = 300,
77 params: dict = None,
78 db_dict: dict = None,
79 kdu_name: str = None,
80 namespace: str = None,
81 **kwargs,
82 ):
83 """Install a helm chart
84
85 :param cluster_uuid str: The UUID of the cluster to install to
86 :param kdu_model str: The name or path of a bundle to install
87 :param kdu_instance: Kdu instance name
88 :param atomic bool: If set, waits until the model is active and resets
89 the cluster on failure.
90 :param timeout int: The time, in seconds, to wait for the install
91 to finish
92 :param params dict: Key-value pairs of instantiation parameters
93 :param kdu_name: Name of the KDU instance to be installed
94 :param namespace: K8s namespace to use for the KDU instance
95
96 :param kwargs: Additional parameters (None yet)
97
98 :return: True if successful
99 """
100 _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
101 self.log.debug("installing {} in cluster {}".format(kdu_model, cluster_id))
102
103 # sync local dir
104 self.fs.sync(from_path=cluster_id)
105
106 # init env, paths
107 paths, env = self._init_paths_env(
108 cluster_name=cluster_id, create_if_not_exist=True
109 )
110
111 # for helm3 if namespace does not exist must create it
112 if namespace and namespace != "kube-system":
113 if not await self._namespace_exists(cluster_id, namespace):
114 try:
115 await self._create_namespace(cluster_id, namespace)
116 except Exception as e:
117 if not await self._namespace_exists(cluster_id, namespace):
118 err_msg = (
119 "namespace {} does not exist in cluster_id {} "
120 "error message: ".format(namespace, e)
121 )
122 self.log.error(err_msg)
123 raise K8sException(err_msg)
124
125 await self._install_impl(
126 cluster_id,
127 kdu_model,
128 paths,
129 env,
130 kdu_instance,
131 atomic=atomic,
132 timeout=timeout,
133 params=params,
134 db_dict=db_dict,
135 kdu_name=kdu_name,
136 namespace=namespace,
137 )
138
139 # sync fs
140 self.fs.reverse_sync(from_path=cluster_id)
141
142 self.log.debug("Returning kdu_instance {}".format(kdu_instance))
143 return True
144
145 async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
146
147 self.log.debug(
148 "inspect kdu_model {} from (optional) repo: {}".format(kdu_model, repo_url)
149 )
150
151 return await self._exec_inspect_comand(
152 inspect_command="all", kdu_model=kdu_model, repo_url=repo_url
153 )
154
155 """
156 ####################################################################################
157 ################################### P R I V A T E ##################################
158 ####################################################################################
159 """
160
161 def _init_paths_env(self, cluster_name: str, create_if_not_exist: bool = True):
162 """
163 Creates and returns base cluster and kube dirs and returns them.
164 Also created helm3 dirs according to new directory specification, paths are
165 returned and also environment variables that must be provided to execute commands
166
167 Helm 3 directory specification uses XDG categories for variable support:
168 - Cache: $XDG_CACHE_HOME, for example, ${HOME}/.cache/helm/
169 - Configuration: $XDG_CONFIG_HOME, for example, ${HOME}/.config/helm/
170 - Data: $XDG_DATA_HOME, for example ${HOME}/.local/share/helm
171
172 The variables assigned for this paths are:
173 (In the documentation the variables names are $HELM_PATH_CACHE, $HELM_PATH_CONFIG,
174 $HELM_PATH_DATA but looking and helm env the variable names are different)
175 - Cache: $HELM_CACHE_HOME
176 - Config: $HELM_CONFIG_HOME
177 - Data: $HELM_DATA_HOME
178 - helm kubeconfig: $KUBECONFIG
179
180 :param cluster_name: cluster_name
181 :return: Dictionary with config_paths and dictionary with helm environment variables
182 """
183
184 base = self.fs.path
185 if base.endswith("/") or base.endswith("\\"):
186 base = base[:-1]
187
188 # base dir for cluster
189 cluster_dir = base + "/" + cluster_name
190
191 # kube dir
192 kube_dir = cluster_dir + "/" + ".kube"
193 if create_if_not_exist and not os.path.exists(kube_dir):
194 self.log.debug("Creating dir {}".format(kube_dir))
195 os.makedirs(kube_dir)
196
197 helm_path_cache = cluster_dir + "/.cache/helm"
198 if create_if_not_exist and not os.path.exists(helm_path_cache):
199 self.log.debug("Creating dir {}".format(helm_path_cache))
200 os.makedirs(helm_path_cache)
201
202 helm_path_config = cluster_dir + "/.config/helm"
203 if create_if_not_exist and not os.path.exists(helm_path_config):
204 self.log.debug("Creating dir {}".format(helm_path_config))
205 os.makedirs(helm_path_config)
206
207 helm_path_data = cluster_dir + "/.local/share/helm"
208 if create_if_not_exist and not os.path.exists(helm_path_data):
209 self.log.debug("Creating dir {}".format(helm_path_data))
210 os.makedirs(helm_path_data)
211
212 config_filename = kube_dir + "/config"
213
214 # 2 - Prepare dictionary with paths
215 paths = {
216 "kube_dir": kube_dir,
217 "kube_config": config_filename,
218 "cluster_dir": cluster_dir,
219 }
220
221 # 3 - Prepare environment variables
222 env = {
223 "HELM_CACHE_HOME": helm_path_cache,
224 "HELM_CONFIG_HOME": helm_path_config,
225 "HELM_DATA_HOME": helm_path_data,
226 "KUBECONFIG": config_filename,
227 }
228
229 for file_name, file in paths.items():
230 if "dir" in file_name and not os.path.exists(file):
231 err_msg = "{} dir does not exist".format(file)
232 self.log.error(err_msg)
233 raise K8sException(err_msg)
234
235 return paths, env
236
237 async def _namespace_exists(self, cluster_id, namespace) -> bool:
238 self.log.debug(
239 "checking if namespace {} exists cluster_id {}".format(
240 namespace, cluster_id
241 )
242 )
243 namespaces = await self._get_namespaces(cluster_id)
244 return namespace in namespaces if namespaces else False
245
246 async def _get_namespaces(self, cluster_id: str):
247
248 self.log.debug("get namespaces cluster_id {}".format(cluster_id))
249
250 # init config, env
251 paths, env = self._init_paths_env(
252 cluster_name=cluster_id, create_if_not_exist=True
253 )
254
255 command = "{} --kubeconfig={} get namespaces -o=yaml".format(
256 self.kubectl_command, paths["kube_config"]
257 )
258 output, _rc = await self._local_async_exec(
259 command=command, raise_exception_on_error=True, env=env
260 )
261
262 data = yaml.load(output, Loader=yaml.SafeLoader)
263 namespaces = [item["metadata"]["name"] for item in data["items"]]
264 self.log.debug(f"namespaces {namespaces}")
265
266 return namespaces
267
268 async def _create_namespace(self, cluster_id: str, namespace: str):
269
270 self.log.debug(f"create namespace: {cluster_id} for cluster_id: {namespace}")
271
272 # init config, env
273 paths, env = self._init_paths_env(
274 cluster_name=cluster_id, create_if_not_exist=True
275 )
276
277 command = "{} --kubeconfig={} create namespace {}".format(
278 self.kubectl_command, paths["kube_config"], namespace
279 )
280 _, _rc = await self._local_async_exec(
281 command=command, raise_exception_on_error=True, env=env
282 )
283 self.log.debug(f"namespace {namespace} created")
284
285 return _rc
286
287 async def _get_services(self, cluster_id: str, kdu_instance: str, namespace: str):
288
289 # init config, env
290 paths, env = self._init_paths_env(
291 cluster_name=cluster_id, create_if_not_exist=True
292 )
293
294 command1 = "{} get manifest {} --namespace={}".format(
295 self._helm_command, kdu_instance, namespace
296 )
297 command2 = "{} get --namespace={} -f -".format(self.kubectl_command, namespace)
298 output, _rc = await self._local_async_exec_pipe(
299 command1, command2, env=env, raise_exception_on_error=True
300 )
301 services = self._parse_services(output)
302
303 return services
304
305 async def _cluster_init(self, cluster_id, namespace, paths, env):
306 """
307 Implements the helm version dependent cluster initialization:
308 For helm3 it creates the namespace if it is not created
309 """
310 if namespace != "kube-system":
311 namespaces = await self._get_namespaces(cluster_id)
312 if namespace not in namespaces:
313 await self._create_namespace(cluster_id, namespace)
314
315 # If default repo is not included add
316 cluster_uuid = "{}:{}".format(namespace, cluster_id)
317 repo_list = await self.repo_list(cluster_uuid)
318 stable_repo = [repo for repo in repo_list if repo["name"] == "stable"]
319 if not stable_repo and self._stable_repo_url:
320 await self.repo_add(cluster_uuid, "stable", self._stable_repo_url)
321
322 # Returns False as no software needs to be uninstalled
323 return False
324
325 async def _uninstall_sw(self, cluster_id: str, namespace: str):
326 # nothing to do to uninstall sw
327 pass
328
329 async def _instances_list(self, cluster_id: str):
330
331 # init paths, env
332 paths, env = self._init_paths_env(
333 cluster_name=cluster_id, create_if_not_exist=True
334 )
335
336 command = "{} list --all-namespaces --output yaml".format(self._helm_command)
337 output, _rc = await self._local_async_exec(
338 command=command, raise_exception_on_error=True, env=env
339 )
340
341 if output and len(output) > 0:
342 self.log.debug("instances list output: {}".format(output))
343 return yaml.load(output, Loader=yaml.SafeLoader)
344 else:
345 return []
346
347 def _get_inspect_command(
348 self, inspect_command: str, kdu_model: str, repo_str: str, version: str
349 ):
350 inspect_command = "{} show {} {}{} {}".format(
351 self._helm_command, inspect_command, kdu_model, repo_str, version
352 )
353 return inspect_command
354
355 async def _status_kdu(
356 self,
357 cluster_id: str,
358 kdu_instance: str,
359 namespace: str = None,
360 show_error_log: bool = False,
361 return_text: bool = False,
362 ):
363
364 self.log.debug(
365 "status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
366 )
367
368 if not namespace:
369 namespace = "kube-system"
370
371 # init config, env
372 paths, env = self._init_paths_env(
373 cluster_name=cluster_id, create_if_not_exist=True
374 )
375 command = "{} status {} --namespace={} --output yaml".format(
376 self._helm_command, kdu_instance, namespace
377 )
378
379 output, rc = await self._local_async_exec(
380 command=command,
381 raise_exception_on_error=True,
382 show_error_log=show_error_log,
383 env=env,
384 )
385
386 if return_text:
387 return str(output)
388
389 if rc != 0:
390 return None
391
392 data = yaml.load(output, Loader=yaml.SafeLoader)
393
394 # remove field 'notes' and manifest
395 try:
396 del data.get("info")["notes"]
397 del data["manifest"]
398 except KeyError:
399 pass
400
401 # unable to parse 'resources' as currently it is not included in helm3
402 return data
403
404 def _get_install_command(
405 self,
406 kdu_model: str,
407 kdu_instance: str,
408 namespace: str,
409 params_str: str,
410 version: str,
411 atomic: bool,
412 timeout: float,
413 ) -> str:
414
415 timeout_str = ""
416 if timeout:
417 timeout_str = "--timeout {}s".format(timeout)
418
419 # atomic
420 atomic_str = ""
421 if atomic:
422 atomic_str = "--atomic"
423 # namespace
424 namespace_str = ""
425 if namespace:
426 namespace_str = "--namespace {}".format(namespace)
427
428 # version
429 version_str = ""
430 if version:
431 version_str = "--version {}".format(version)
432
433 command = (
434 "{helm} install {name} {atomic} --output yaml "
435 "{params} {timeout} {ns} {model} {ver}".format(
436 helm=self._helm_command,
437 name=kdu_instance,
438 atomic=atomic_str,
439 params=params_str,
440 timeout=timeout_str,
441 ns=namespace_str,
442 model=kdu_model,
443 ver=version_str,
444 )
445 )
446 return command
447
448 def _get_upgrade_command(
449 self,
450 kdu_model: str,
451 kdu_instance: str,
452 namespace: str,
453 params_str: str,
454 version: str,
455 atomic: bool,
456 timeout: float,
457 ) -> str:
458
459 timeout_str = ""
460 if timeout:
461 timeout_str = "--timeout {}s".format(timeout)
462
463 # atomic
464 atomic_str = ""
465 if atomic:
466 atomic_str = "--atomic"
467
468 # version
469 version_str = ""
470 if version:
471 version_str = "--version {}".format(version)
472
473 # namespace
474 namespace_str = ""
475 if namespace:
476 namespace_str = "--namespace {}".format(namespace)
477
478 command = (
479 "{helm} upgrade {name} {model} {namespace} {atomic} --output yaml {params} "
480 "{timeout} {ver}".format(
481 helm=self._helm_command,
482 name=kdu_instance,
483 namespace=namespace_str,
484 atomic=atomic_str,
485 params=params_str,
486 timeout=timeout_str,
487 model=kdu_model,
488 ver=version_str,
489 )
490 )
491 return command
492
493 def _get_rollback_command(
494 self, kdu_instance: str, namespace: str, revision: float
495 ) -> str:
496 return "{} rollback {} {} --namespace={} --wait".format(
497 self._helm_command, kdu_instance, revision, namespace
498 )
499
500 def _get_uninstall_command(self, kdu_instance: str, namespace: str) -> str:
501
502 return "{} uninstall {} --namespace={}".format(
503 self._helm_command, kdu_instance, namespace
504 )
505
506 def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
507 repo_ids = []
508 cluster_filter = {"_admin.helm-chart-v3.id": cluster_uuid}
509 cluster = self.db.get_one("k8sclusters", cluster_filter)
510 if cluster:
511 repo_ids = cluster.get("_admin").get("helm_chart_repos") or []
512 return repo_ids
513 else:
514 raise K8sException(
515 "k8cluster with helm-id : {} not found".format(cluster_uuid)
516 )