2 # Copyright 2020 Telefonica Investigacion y Desarrollo, S.A.U.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
25 from grpclib
.client
import Channel
27 from osm_lcm
.frontend_pb2
import PrimitiveRequest
28 from osm_lcm
.frontend_pb2
import SshKeyRequest
, SshKeyReply
29 from osm_lcm
.frontend_grpc
import FrontendExecutorStub
30 from osm_lcm
.lcm_utils
import LcmBase
32 from osm_lcm
.data_utils
.database
.database
import Database
33 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
35 from n2vc
.n2vc_conn
import N2VCConnector
36 from n2vc
.k8s_helm_conn
import K8sHelmConnector
37 from n2vc
.k8s_helm3_conn
import K8sHelm3Connector
38 from n2vc
.exceptions
import (
39 N2VCBadArgumentsException
,
41 N2VCExecutionException
,
44 from osm_lcm
.lcm_utils
import deep_get
47 def retryer(max_wait_time_var
="_initial_retry_time", delay_time_var
="_retry_delay"):
49 retry_exceptions
= ConnectionRefusedError
51 @functools.wraps(func
)
52 async def wrapped(*args
, **kwargs
):
53 # default values for wait time and delay_time
57 # obtain arguments from variable names
59 if self
.__dict
__.get(max_wait_time_var
):
60 max_wait_time
= self
.__dict
__.get(max_wait_time_var
)
61 if self
.__dict
__.get(delay_time_var
):
62 delay_time
= self
.__dict
__.get(delay_time_var
)
64 wait_time
= max_wait_time
67 return await func(*args
, **kwargs
)
68 except retry_exceptions
:
69 wait_time
= wait_time
- delay_time
70 await asyncio
.sleep(delay_time
)
73 return ConnectionRefusedError
80 class LCMHelmConn(N2VCConnector
, LcmBase
):
81 _KUBECTL_OSM_NAMESPACE
= "osm"
82 _KUBECTL_OSM_CLUSTER_NAME
= "_system-osm-k8s"
83 _EE_SERVICE_PORT
= 50050
85 # Initial max retry time
86 _MAX_INITIAL_RETRY_TIME
= 600
87 # Max retry time for normal operations
89 # Time beetween retries, retry time after a connection error is raised
96 vca_config
: dict = None,
100 Initialize EE helm connector.
103 self
.db
= Database().instance
.db
104 self
.fs
= Filesystem().instance
.fs
106 # parent class constructor
107 N2VCConnector
.__init
__(
108 self
, log
=log
, loop
=loop
, on_update_db
=on_update_db
, db
=self
.db
, fs
=self
.fs
111 self
.vca_config
= vca_config
112 self
.log
.debug("Initialize helm N2VC connector")
113 self
.log
.debug("initial vca_config: {}".format(vca_config
))
115 # TODO - Obtain data from configuration
116 self
._ee
_service
_port
= self
._EE
_SERVICE
_PORT
118 self
._retry
_delay
= self
._EE
_RETRY
_DELAY
120 if self
.vca_config
and self
.vca_config
.get("eegrpcinittimeout"):
121 self
._initial
_retry
_time
= self
.vca_config
.get("eegrpcinittimeout")
122 self
.log
.debug("Initial retry time: {}".format(self
._initial
_retry
_time
))
124 self
._initial
_retry
_time
= self
._MAX
_INITIAL
_RETRY
_TIME
126 "Applied default retry time: {}".format(self
._initial
_retry
_time
)
129 if self
.vca_config
and self
.vca_config
.get("eegrpctimeout"):
130 self
._max
_retry
_time
= self
.vca_config
.get("eegrpctimeout")
131 self
.log
.debug("Retry time: {}".format(self
._max
_retry
_time
))
133 self
._max
_retry
_time
= self
._MAX
_RETRY
_TIME
135 "Applied default retry time: {}".format(self
._max
_retry
_time
)
138 # initialize helm connector for helmv2 and helmv3
139 self
._k
8sclusterhelm
2 = K8sHelmConnector(
140 kubectl_command
=self
.vca_config
.get("kubectlpath"),
141 helm_command
=self
.vca_config
.get("helmpath"),
148 self
._k
8sclusterhelm
3 = K8sHelm3Connector(
149 kubectl_command
=self
.vca_config
.get("kubectlpath"),
150 helm_command
=self
.vca_config
.get("helm3path"),
157 self
._system
_cluster
_id
= None
158 self
.log
.info("Helm N2VC connector initialized")
160 # TODO - ¿reuse_ee_id?
161 async def create_execution_environment(
165 reuse_ee_id
: str = None,
166 progress_timeout
: float = None,
167 total_timeout
: float = None,
169 artifact_path
: str = None,
170 chart_model
: str = None,
171 vca_type
: str = None,
176 Creates a new helm execution environment deploying the helm-chat indicated in the
178 :param str namespace: This param is not used, all helm charts are deployed in the osm
180 :param dict db_dict: where to write to database when the status changes.
181 It contains a dictionary with {collection: str, filter: {}, path: str},
182 e.g. {collection: "nsrs", filter: {_id: <nsd-id>, path:
183 "_admin.deployed.VCA.3"}
184 :param str reuse_ee_id: ee id from an older execution. TODO - right now this param is not used
185 :param float progress_timeout:
186 :param float total_timeout:
187 :param dict config: General variables to instantiate KDU
188 :param str artifact_path: path of package content
189 :param str chart_model: helm chart/reference (string), which can be either
191 - a name of chart available via the repos known by OSM
192 (e.g. stable/openldap, stable/openldap:1.2.4)
193 - a path to a packaged chart (e.g. mychart.tgz)
194 - a path to an unpacked chart directory or a URL (e.g. mychart)
195 :param str vca_type: Type of vca, must be type helm or helm-v3
196 :returns str, dict: id of the new execution environment including namespace.helm_id
197 and credentials object set to None as all credentials should be osm kubernetes .kubeconfig
201 "create_execution_environment: namespace: {}, artifact_path: {}, "
202 "chart_model: {}, db_dict: {}, reuse_ee_id: {}".format(
203 namespace
, artifact_path
, db_dict
, chart_model
, reuse_ee_id
207 # Validate artifact-path is provided
208 if artifact_path
is None or len(artifact_path
) == 0:
209 raise N2VCBadArgumentsException(
210 message
="artifact_path is mandatory", bad_args
=["artifact_path"]
213 # Validate artifact-path exists and sync path
214 from_path
= os
.path
.split(artifact_path
)[0]
215 self
.fs
.sync(from_path
)
217 # remove / in charm path
218 while artifact_path
.find("//") >= 0:
219 artifact_path
= artifact_path
.replace("//", "/")
222 if self
.fs
.file_exists(artifact_path
):
223 helm_chart_path
= artifact_path
225 msg
= "artifact path does not exist: {}".format(artifact_path
)
226 raise N2VCBadArgumentsException(message
=msg
, bad_args
=["artifact_path"])
228 if artifact_path
.startswith("/"):
229 full_path
= self
.fs
.path
+ helm_chart_path
231 full_path
= self
.fs
.path
+ "/" + helm_chart_path
233 while full_path
.find("//") >= 0:
234 full_path
= full_path
.replace("//", "/")
236 # By default, the KDU is expected to be a file
237 kdu_model
= full_path
238 # If the chart_model includes a "/", then it is a reference:
239 # e.g. (stable/openldap; stable/openldap:1.2.4)
240 if chart_model
.find("/") >= 0:
241 kdu_model
= chart_model
244 # Call helm conn install
245 # Obtain system cluster id from database
246 system_cluster_uuid
= await self
._get
_system
_cluster
_id
()
247 # Add parameter osm if exist to global
248 if config
and config
.get("osm"):
249 if not config
.get("global"):
250 config
["global"] = {}
251 config
["global"]["osm"] = config
.get("osm")
253 self
.log
.debug("install helm chart: {}".format(full_path
))
254 if vca_type
== "helm":
255 helm_id
= self
._k
8sclusterhelm
2.generate_kdu_instance_name(
259 await self
._k
8sclusterhelm
2.install(
262 kdu_instance
=helm_id
,
263 namespace
=self
._KUBECTL
_OSM
_NAMESPACE
,
266 timeout
=progress_timeout
,
269 helm_id
= self
._k
8sclusterhelm
2.generate_kdu_instance_name(
273 await self
._k
8sclusterhelm
3.install(
276 kdu_instance
=helm_id
,
277 namespace
=self
._KUBECTL
_OSM
_NAMESPACE
,
280 timeout
=progress_timeout
,
283 ee_id
= "{}:{}.{}".format(vca_type
, self
._KUBECTL
_OSM
_NAMESPACE
, helm_id
)
285 except N2VCException
:
287 except Exception as e
:
288 self
.log
.error("Error deploying chart ee: {}".format(e
), exc_info
=True)
289 raise N2VCException("Error deploying chart ee: {}".format(e
))
291 async def register_execution_environment(
296 progress_timeout
: float = None,
297 total_timeout
: float = None,
304 async def install_configuration_sw(self
, *args
, **kwargs
):
308 async def add_relation(self
, *args
, **kwargs
):
312 async def remove_relation(self
):
316 async def get_status(self
, *args
, **kwargs
):
317 # not used for this connector
320 async def get_ee_ssh_public__key(
324 progress_timeout
: float = None,
325 total_timeout
: float = None,
329 Obtains ssh-public key from ee executing GetSShKey method from the ee.
331 :param str ee_id: the id of the execution environment returned by
332 create_execution_environment or register_execution_environment
334 :param float progress_timeout:
335 :param float total_timeout:
336 :returns: public key of the execution environment
340 "get_ee_ssh_public_key: ee_id: {}, db_dict: {}".format(ee_id
, db_dict
)
344 if ee_id
is None or len(ee_id
) == 0:
345 raise N2VCBadArgumentsException(
346 message
="ee_id is mandatory", bad_args
=["ee_id"]
350 # Obtain ip_addr for the ee service, it is resolved by dns from the ee name by kubernetes
351 version
, namespace
, helm_id
= self
._get
_ee
_id
_parts
(ee_id
)
352 ip_addr
= socket
.gethostbyname(helm_id
)
354 # Obtain ssh_key from the ee, this method will implement retries to allow the ee
355 # install libraries and start successfully
356 ssh_key
= await self
._get
_ssh
_key
(ip_addr
)
358 except Exception as e
:
359 self
.log
.error("Error obtaining ee ssh_key: {}".format(e
), exc_info
=True)
360 raise N2VCException("Error obtaining ee ssh_ke: {}".format(e
))
362 async def upgrade_charm(
366 charm_id
: str = None,
367 charm_type
: str = None,
368 timeout
: float = None,
370 """This method upgrade charms in VNFs
372 This method does not support KDU's deployed with Helm.
375 ee_id: Execution environment id
376 path: Local path to the charm
378 charm_type: Charm type can be lxc-proxy-charm, native-charm or k8s-proxy-charm
379 timeout: (Float) Timeout for the ns update operation
382 the output of the update operation if status equals to "completed"
385 raise N2VCException("KDUs deployed with Helm do not support charm upgrade")
387 async def exec_primitive(
392 db_dict
: dict = None,
393 progress_timeout
: float = None,
394 total_timeout
: float = None,
398 Execute a primitive in the execution environment
400 :param str ee_id: the one returned by create_execution_environment or
401 register_execution_environment with the format namespace.helm_id
402 :param str primitive_name: must be one defined in the software. There is one
403 called 'config', where, for the proxy case, the 'credentials' of VM are
405 :param dict params_dict: parameters of the action
406 :param dict db_dict: where to write into database when the status changes.
407 It contains a dict with
408 {collection: <str>, filter: {}, path: <str>},
409 e.g. {collection: "nslcmops", filter:
410 {_id: <nslcmop_id>, path: "_admin.VCA"}
411 It will be used to store information about intermediate notifications
412 :param float progress_timeout:
413 :param float total_timeout:
414 :returns str: primitive result, if ok. It raises exceptions in case of fail
418 "exec primitive for ee_id : {}, primitive_name: {}, params_dict: {}, db_dict: {}".format(
419 ee_id
, primitive_name
, params_dict
, db_dict
424 if ee_id
is None or len(ee_id
) == 0:
425 raise N2VCBadArgumentsException(
426 message
="ee_id is mandatory", bad_args
=["ee_id"]
428 if primitive_name
is None or len(primitive_name
) == 0:
429 raise N2VCBadArgumentsException(
430 message
="action_name is mandatory", bad_args
=["action_name"]
432 if params_dict
is None:
436 version
, namespace
, helm_id
= self
._get
_ee
_id
_parts
(ee_id
)
437 ip_addr
= socket
.gethostbyname(helm_id
)
438 except Exception as e
:
439 self
.log
.error("Error getting ee ip ee: {}".format(e
))
440 raise N2VCException("Error getting ee ip ee: {}".format(e
))
442 if primitive_name
== "config":
444 # Execute config primitive, higher timeout to check the case ee is starting
445 status
, detailed_message
= await self
._execute
_config
_primitive
(
446 ip_addr
, params_dict
, db_dict
=db_dict
449 "Executed config primitive ee_id_ {}, status: {}, message: {}".format(
450 ee_id
, status
, detailed_message
455 "Error configuring helm ee, status: {}, message: {}".format(
456 status
, detailed_message
459 raise N2VCExecutionException(
460 message
="Error configuring helm ee_id: {}, status: {}, message: {}: ".format(
461 ee_id
, status
, detailed_message
463 primitive_name
=primitive_name
,
465 except Exception as e
:
466 self
.log
.error("Error configuring helm ee: {}".format(e
))
467 raise N2VCExecutionException(
468 message
="Error configuring helm ee_id: {}, {}".format(ee_id
, e
),
469 primitive_name
=primitive_name
,
475 status
, detailed_message
= await self
._execute
_primitive
(
476 ip_addr
, primitive_name
, params_dict
, db_dict
=db_dict
479 "Executed primitive {} ee_id_ {}, status: {}, message: {}".format(
480 primitive_name
, ee_id
, status
, detailed_message
483 if status
!= "OK" and status
!= "PROCESSING":
485 "Execute primitive {} returned not ok status: {}, message: {}".format(
486 primitive_name
, status
, detailed_message
489 raise N2VCExecutionException(
490 message
="Execute primitive {} returned not ok status: {}, message: {}".format(
491 primitive_name
, status
, detailed_message
493 primitive_name
=primitive_name
,
495 except Exception as e
:
497 "Error executing primitive {}: {}".format(primitive_name
, e
)
499 raise N2VCExecutionException(
500 message
="Error executing primitive {} into ee={} : {}".format(
501 primitive_name
, ee_id
, e
503 primitive_name
=primitive_name
,
505 return detailed_message
507 async def deregister_execution_environments(self
):
511 async def delete_execution_environment(
514 db_dict
: dict = None,
515 total_timeout
: float = None,
519 Delete an execution environment
520 :param str ee_id: id of the execution environment to delete, included namespace.helm_id
521 :param dict db_dict: where to write into database when the status changes.
522 It contains a dict with
523 {collection: <str>, filter: {}, path: <str>},
524 e.g. {collection: "nsrs", filter:
525 {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
526 :param float total_timeout:
529 self
.log
.info("ee_id: {}".format(ee_id
))
533 raise N2VCBadArgumentsException(
534 message
="ee_id is mandatory", bad_args
=["ee_id"]
539 # Obtain cluster_uuid
540 system_cluster_uuid
= await self
._get
_system
_cluster
_id
()
543 version
, namespace
, helm_id
= self
._get
_ee
_id
_parts
(ee_id
)
545 # Uninstall chart, for backward compatibility we must assume that if there is no
546 # version it is helm-v2
547 if version
== "helm-v3":
548 await self
._k
8sclusterhelm
3.uninstall(system_cluster_uuid
, helm_id
)
550 await self
._k
8sclusterhelm
2.uninstall(system_cluster_uuid
, helm_id
)
551 self
.log
.info("ee_id: {} deleted".format(ee_id
))
552 except N2VCException
:
554 except Exception as e
:
556 "Error deleting ee id: {}: {}".format(ee_id
, e
), exc_info
=True
558 raise N2VCException("Error deleting ee id {}: {}".format(ee_id
, e
))
560 async def delete_namespace(
561 self
, namespace
: str, db_dict
: dict = None, total_timeout
: float = None
563 # method not implemented for this connector, execution environments must be deleted individually
566 async def install_k8s_proxy_charm(
572 progress_timeout
: float = None,
573 total_timeout
: float = None,
580 @retryer(max_wait_time_var
="_initial_retry_time", delay_time_var
="_retry_delay")
581 async def _get_ssh_key(self
, ip_addr
):
582 channel
= Channel(ip_addr
, self
._ee
_service
_port
)
584 stub
= FrontendExecutorStub(channel
)
585 self
.log
.debug("get ssh key, ip_addr: {}".format(ip_addr
))
586 reply
: SshKeyReply
= await stub
.GetSshKey(SshKeyRequest())
591 @retryer(max_wait_time_var
="_initial_retry_time", delay_time_var
="_retry_delay")
592 async def _execute_config_primitive(self
, ip_addr
, params
, db_dict
=None):
593 return await self
._execute
_primitive
_internal
(
594 ip_addr
, "config", params
, db_dict
=db_dict
597 @retryer(max_wait_time_var
="_max_retry_time", delay_time_var
="_retry_delay")
598 async def _execute_primitive(self
, ip_addr
, primitive_name
, params
, db_dict
=None):
599 return await self
._execute
_primitive
_internal
(
600 ip_addr
, primitive_name
, params
, db_dict
=db_dict
603 async def _execute_primitive_internal(
604 self
, ip_addr
, primitive_name
, params
, db_dict
=None
607 channel
= Channel(ip_addr
, self
._ee
_service
_port
)
609 stub
= FrontendExecutorStub(channel
)
610 async with stub
.RunPrimitive
.open() as stream
:
611 primitive_id
= str(uuid
.uuid1())
614 "Execute primitive internal: id:{}, name:{}, params: {}".format(
615 primitive_id
, primitive_name
, params
618 await stream
.send_message(
620 id=primitive_id
, name
=primitive_name
, params
=yaml
.dump(params
)
624 async for reply
in stream
:
625 self
.log
.debug("Received reply: {}".format(reply
))
627 # If db_dict provided write notifs in database
629 self
._write
_op
_detailed
_status
(
630 db_dict
, reply
.status
, reply
.detailed_message
633 return reply
.status
, reply
.detailed_message
635 return "ERROR", "No result received"
639 def _write_op_detailed_status(self
, db_dict
, status
, detailed_message
):
641 # write ee_id to database: _admin.deployed.VCA.x
643 the_table
= db_dict
["collection"]
644 the_filter
= db_dict
["filter"]
645 update_dict
= {"detailed-status": "{}: {}".format(status
, detailed_message
)}
646 # self.log.debug('Writing ee_id to database: {}'.format(the_path))
650 update_dict
=update_dict
,
653 except asyncio
.CancelledError
:
655 except Exception as e
:
656 self
.log
.error("Error writing detailedStatus to database: {}".format(e
))
658 async def _get_system_cluster_id(self
):
659 if not self
._system
_cluster
_id
:
660 db_k8cluster
= self
.db
.get_one(
661 "k8sclusters", {"name": self
._KUBECTL
_OSM
_CLUSTER
_NAME
}
663 k8s_hc_id
= deep_get(db_k8cluster
, ("_admin", "helm-chart-v3", "id"))
666 # backward compatibility for existing clusters that have not been initialized for helm v3
667 cluster_id
= db_k8cluster
.get("_id")
668 k8s_credentials
= yaml
.safe_dump(db_k8cluster
.get("credentials"))
669 k8s_hc_id
, uninstall_sw
= await self
._k
8sclusterhelm
3.init_env(
670 k8s_credentials
, reuse_cluster_uuid
=cluster_id
672 db_k8scluster_update
= {
673 "_admin.helm-chart-v3.error_msg": None,
674 "_admin.helm-chart-v3.id": k8s_hc_id
,
675 "_admin.helm-chart-v3}.created": uninstall_sw
,
676 "_admin.helm-chart-v3.operationalState": "ENABLED",
678 self
.update_db_2("k8sclusters", cluster_id
, db_k8scluster_update
)
679 except Exception as e
:
681 "error initializing helm-v3 cluster: {}".format(str(e
))
684 "K8s system cluster '{}' has not been initialized for helm-chart-v3".format(
688 self
._system
_cluster
_id
= k8s_hc_id
689 return self
._system
_cluster
_id
691 def _get_ee_id_parts(self
, ee_id
):
693 Parses ee_id stored at database that can be either 'version:namespace.helm_id' or only
694 namespace.helm_id for backward compatibility
695 If exists helm version can be helm-v3 or helm (helm-v2 old version)
697 version
, _
, part_id
= ee_id
.rpartition(":")
698 namespace
, _
, helm_id
= part_id
.rpartition(".")
699 return version
, namespace
, helm_id