Sync with k8s api
[osm/N2VC.git] / n2vc / vnf.py
1 # Copyright 2019 Canonical Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14
15 import asyncio
16 import base64
17 import binascii
18 import logging
19 import os
20 import os.path
21 import re
22 import shlex
23 import ssl
24 import subprocess
25 import sys
26 # import time
27 import n2vc.exceptions
28 from n2vc.provisioner import SSHProvisioner
29
30 # FIXME: this should load the juju inside or modules without having to
31 # explicitly install it. Check why it's not working.
32 # Load our subtree of the juju library
33 # path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
34 # path = os.path.join(path, "modules/libjuju/")
35 # if path not in sys.path:
36 # sys.path.insert(1, path)
37
38 from juju.client import client
39 from juju.controller import Controller
40 from juju.model import ModelObserver
41 from juju.errors import JujuAPIError, JujuError
42
43
44 # We might need this to connect to the websocket securely, but test and verify.
45 try:
46 ssl._create_default_https_context = ssl._create_unverified_context
47 except AttributeError:
48 # Legacy Python doesn't verify by default (see pep-0476)
49 # https://www.python.org/dev/peps/pep-0476/
50 pass
51
52
53 # Custom exceptions
54 # Deprecated. Please use n2vc.exceptions namespace.
55 class JujuCharmNotFound(Exception):
56 """The Charm can't be found or is not readable."""
57
58
59 class JujuApplicationExists(Exception):
60 """The Application already exists."""
61
62
63 class N2VCPrimitiveExecutionFailed(Exception):
64 """Something failed while attempting to execute a primitive."""
65
66
67 class NetworkServiceDoesNotExist(Exception):
68 """The Network Service being acted against does not exist."""
69
70
71 class PrimitiveDoesNotExist(Exception):
72 """The Primitive being executed does not exist."""
73
74
75 # Quiet the debug logging
76 logging.getLogger('websockets.protocol').setLevel(logging.INFO)
77 logging.getLogger('juju.client.connection').setLevel(logging.WARN)
78 logging.getLogger('juju.model').setLevel(logging.WARN)
79 logging.getLogger('juju.machine').setLevel(logging.WARN)
80
81
82 class VCAMonitor(ModelObserver):
83 """Monitor state changes within the Juju Model."""
84 log = None
85
86 def __init__(self, ns_name):
87 self.log = logging.getLogger(__name__)
88
89 self.ns_name = ns_name
90 self.applications = {}
91
92 def AddApplication(self, application_name, callback, *callback_args):
93 if application_name not in self.applications:
94 self.applications[application_name] = {
95 'callback': callback,
96 'callback_args': callback_args
97 }
98
99 def RemoveApplication(self, application_name):
100 if application_name in self.applications:
101 del self.applications[application_name]
102
103 async def on_change(self, delta, old, new, model):
104 """React to changes in the Juju model."""
105
106 if delta.entity == "unit":
107 # Ignore change events from other applications
108 if delta.data['application'] not in self.applications.keys():
109 return
110
111 try:
112
113 application_name = delta.data['application']
114
115 callback = self.applications[application_name]['callback']
116 callback_args = \
117 self.applications[application_name]['callback_args']
118
119 if old and new:
120 # Fire off a callback with the application state
121 if callback:
122 callback(
123 self.ns_name,
124 delta.data['application'],
125 new.workload_status,
126 new.workload_status_message,
127 *callback_args)
128
129 if old and not new:
130 # This is a charm being removed
131 if callback:
132 callback(
133 self.ns_name,
134 delta.data['application'],
135 "removed",
136 "",
137 *callback_args)
138 except Exception as e:
139 self.log.debug("[1] notify_callback exception: {}".format(e))
140
141 elif delta.entity == "action":
142 # TODO: Decide how we want to notify the user of actions
143
144 # uuid = delta.data['id'] # The Action's unique id
145 # msg = delta.data['message'] # The output of the action
146 #
147 # if delta.data['status'] == "pending":
148 # # The action is queued
149 # pass
150 # elif delta.data['status'] == "completed""
151 # # The action was successful
152 # pass
153 # elif delta.data['status'] == "failed":
154 # # The action failed.
155 # pass
156
157 pass
158
159 ########
160 # TODO
161 #
162 # Create unique models per network service
163 # Document all public functions
164
165
166 class N2VC:
167 def __init__(self,
168 log=None,
169 server='127.0.0.1',
170 port=17070,
171 user='admin',
172 secret=None,
173 artifacts=None,
174 loop=None,
175 juju_public_key=None,
176 ca_cert=None,
177 api_proxy=None
178 ):
179 """Initialize N2VC
180
181 Initializes the N2VC object, allowing the caller to interoperate with the VCA.
182
183
184 :param log obj: The logging object to log to
185 :param server str: The IP Address or Hostname of the Juju controller
186 :param port int: The port of the Juju Controller
187 :param user str: The Juju username to authenticate with
188 :param secret str: The Juju password to authenticate with
189 :param artifacts str: The directory where charms required by a vnfd are
190 stored.
191 :param loop obj: The loop to use.
192 :param juju_public_key str: The contents of the Juju public SSH key
193 :param ca_cert str: The CA certificate to use to authenticate
194 :param api_proxy str: The IP of the host machine
195
196 :Example:
197 client = n2vc.vnf.N2VC(
198 log=log,
199 server='10.1.1.28',
200 port=17070,
201 user='admin',
202 secret='admin',
203 artifacts='/app/storage/myvnf/charms',
204 loop=loop,
205 juju_public_key='<contents of the juju public key>',
206 ca_cert='<contents of CA certificate>',
207 api_proxy='192.168.1.155'
208 )
209 """
210
211 # Initialize instance-level variables
212 self.api = None
213 self.log = None
214 self.controller = None
215 self.connecting = False
216 self.authenticated = False
217 self.api_proxy = api_proxy
218
219 if log:
220 self.log = log
221 else:
222 self.log = logging.getLogger(__name__)
223
224 # For debugging
225 self.refcount = {
226 'controller': 0,
227 'model': 0,
228 }
229
230 self.models = {}
231
232 # Model Observers
233 self.monitors = {}
234
235 # VCA config
236 self.hostname = ""
237 self.port = 17070
238 self.username = ""
239 self.secret = ""
240
241 self.juju_public_key = juju_public_key
242 if juju_public_key:
243 self._create_juju_public_key(juju_public_key)
244 else:
245 self.juju_public_key = ''
246
247 # TODO: Verify ca_cert is valid before using. VCA will crash
248 # if the ca_cert isn't formatted correctly.
249 def base64_to_cacert(b64string):
250 """Convert the base64-encoded string containing the VCA CACERT.
251
252 The input string....
253
254 """
255 try:
256 cacert = base64.b64decode(b64string).decode("utf-8")
257
258 cacert = re.sub(
259 r'\\n',
260 r'\n',
261 cacert,
262 )
263 except binascii.Error as e:
264 self.log.debug("Caught binascii.Error: {}".format(e))
265 raise n2vc.exceptions.InvalidCACertificate("Invalid CA Certificate")
266
267 return cacert
268
269 self.ca_cert = None
270 if ca_cert:
271 self.ca_cert = base64_to_cacert(ca_cert)
272
273
274 # Quiet websocket traffic
275 logging.getLogger('websockets.protocol').setLevel(logging.INFO)
276 logging.getLogger('juju.client.connection').setLevel(logging.WARN)
277 logging.getLogger('model').setLevel(logging.WARN)
278 # logging.getLogger('websockets.protocol').setLevel(logging.DEBUG)
279
280 self.log.debug('JujuApi: instantiated')
281
282 self.server = server
283 self.port = port
284
285 self.secret = secret
286 if user.startswith('user-'):
287 self.user = user
288 else:
289 self.user = 'user-{}'.format(user)
290
291 self.endpoint = '%s:%d' % (server, int(port))
292
293 self.artifacts = artifacts
294
295 self.loop = loop or asyncio.get_event_loop()
296
297 def __del__(self):
298 """Close any open connections."""
299 yield self.logout()
300
301 def _create_juju_public_key(self, public_key):
302 """Recreate the Juju public key on disk.
303
304 Certain libjuju commands expect to be run from the same machine as Juju
305 is bootstrapped to. This method will write the public key to disk in
306 that location: ~/.local/share/juju/ssh/juju_id_rsa.pub
307 """
308 # Make sure that we have a public key before writing to disk
309 if public_key is None or len(public_key) == 0:
310 if 'OSM_VCA_PUBKEY' in os.environ:
311 public_key = os.getenv('OSM_VCA_PUBKEY', '')
312 if len(public_key == 0):
313 return
314 else:
315 return
316
317 path = "{}/.local/share/juju/ssh".format(
318 os.path.expanduser('~'),
319 )
320 if not os.path.exists(path):
321 os.makedirs(path)
322
323 with open('{}/juju_id_rsa.pub'.format(path), 'w') as f:
324 f.write(public_key)
325
326 def notify_callback(self, model_name, application_name, status, message,
327 callback=None, *callback_args):
328 try:
329 if callback:
330 callback(
331 model_name,
332 application_name,
333 status, message,
334 *callback_args,
335 )
336 except Exception as e:
337 self.log.error("[0] notify_callback exception {}".format(e))
338 raise e
339 return True
340
341 # Public methods
342 async def Relate(self, model_name, vnfd):
343 """Create a relation between the charm-enabled VDUs in a VNF.
344
345 The Relation mapping has two parts: the id of the vdu owning the endpoint, and the name of the endpoint.
346
347 vdu:
348 ...
349 vca-relationships:
350 relation:
351 - provides: dataVM:db
352 requires: mgmtVM:app
353
354 This tells N2VC that the charm referred to by the dataVM vdu offers a relation named 'db', and the mgmtVM vdu has an 'app' endpoint that should be connected to a database.
355
356 :param str ns_name: The name of the network service.
357 :param dict vnfd: The parsed yaml VNF descriptor.
358 """
359
360 # Currently, the call to Relate() is made automatically after the
361 # deployment of each charm; if the relation depends on a charm that
362 # hasn't been deployed yet, the call will fail silently. This will
363 # prevent an API breakage, with the intent of making this an explicitly
364 # required call in a more object-oriented refactor of the N2VC API.
365
366 configs = []
367 vnf_config = vnfd.get("vnf-configuration")
368 if vnf_config:
369 juju = vnf_config['juju']
370 if juju:
371 configs.append(vnf_config)
372
373 for vdu in vnfd['vdu']:
374 vdu_config = vdu.get('vdu-configuration')
375 if vdu_config:
376 juju = vdu_config['juju']
377 if juju:
378 configs.append(vdu_config)
379
380 def _get_application_name(name):
381 """Get the application name that's mapped to a vnf/vdu."""
382 vnf_member_index = 0
383 vnf_name = vnfd['name']
384
385 for vdu in vnfd.get('vdu'):
386 # Compare the named portion of the relation to the vdu's id
387 if vdu['id'] == name:
388 application_name = self.FormatApplicationName(
389 model_name,
390 vnf_name,
391 str(vnf_member_index),
392 )
393 return application_name
394 else:
395 vnf_member_index += 1
396
397 return None
398
399 # Loop through relations
400 for cfg in configs:
401 if 'juju' in cfg:
402 juju = cfg['juju']
403 if 'vca-relationships' in juju and 'relation' in juju['vca-relationships']:
404 for rel in juju['vca-relationships']['relation']:
405 try:
406
407 # get the application name for the provides
408 (name, endpoint) = rel['provides'].split(':')
409 application_name = _get_application_name(name)
410
411 provides = "{}:{}".format(
412 application_name,
413 endpoint
414 )
415
416 # get the application name for thr requires
417 (name, endpoint) = rel['requires'].split(':')
418 application_name = _get_application_name(name)
419
420 requires = "{}:{}".format(
421 application_name,
422 endpoint
423 )
424 self.log.debug("Relation: {} <-> {}".format(
425 provides,
426 requires
427 ))
428 await self.add_relation(
429 model_name,
430 provides,
431 requires,
432 )
433 except Exception as e:
434 self.log.debug("Exception: {}".format(e))
435
436 return
437
438 async def DeployCharms(self, model_name, application_name, vnfd,
439 charm_path, params={}, machine_spec={},
440 callback=None, *callback_args):
441 """Deploy one or more charms associated with a VNF.
442
443 Deploy the charm(s) referenced in a VNF Descriptor.
444
445 :param str model_name: The name or unique id of the network service.
446 :param str application_name: The name of the application
447 :param dict vnfd: The name of the application
448 :param str charm_path: The path to the Juju charm
449 :param dict params: A dictionary of runtime parameters
450 Examples::
451 {
452 'rw_mgmt_ip': '1.2.3.4',
453 # Pass the initial-config-primitives section of the vnf or vdu
454 'initial-config-primitives': {...}
455 'user_values': dictionary with the day-1 parameters provided at instantiation time. It will replace values
456 inside < >. rw_mgmt_ip will be included here also
457 }
458 :param dict machine_spec: A dictionary describing the machine to
459 install to
460 Examples::
461 {
462 'hostname': '1.2.3.4',
463 'username': 'ubuntu',
464 }
465 :param obj callback: A callback function to receive status changes.
466 :param tuple callback_args: A list of arguments to be passed to the
467 callback
468 """
469
470 ########################################################
471 # Verify the path to the charm exists and is readable. #
472 ########################################################
473 if not os.path.exists(charm_path):
474 self.log.debug("Charm path doesn't exist: {}".format(charm_path))
475 self.notify_callback(
476 model_name,
477 application_name,
478 "failed",
479 callback,
480 *callback_args,
481 )
482 raise JujuCharmNotFound("No artifacts configured.")
483
484 ################################
485 # Login to the Juju controller #
486 ################################
487 if not self.authenticated:
488 self.log.debug("Authenticating with Juju")
489 await self.login()
490
491 ##########################################
492 # Get the model for this network service #
493 ##########################################
494 model = await self.get_model(model_name)
495
496 ########################################
497 # Verify the application doesn't exist #
498 ########################################
499 app = await self.get_application(model, application_name)
500 if app:
501 raise JujuApplicationExists("Can't deploy application \"{}\" to model \"{}\" because it already exists.".format(application_name, model_name))
502
503 ################################################################
504 # Register this application with the model-level event monitor #
505 ################################################################
506 if callback:
507 self.log.debug("JujuApi: Registering callback for {}".format(
508 application_name,
509 ))
510 await self.Subscribe(model_name, application_name, callback, *callback_args)
511
512 #######################################
513 # Get the initial charm configuration #
514 #######################################
515
516 rw_mgmt_ip = None
517 if 'rw_mgmt_ip' in params:
518 rw_mgmt_ip = params['rw_mgmt_ip']
519
520 if 'initial-config-primitive' not in params:
521 params['initial-config-primitive'] = {}
522
523 initial_config = self._get_config_from_dict(
524 params['initial-config-primitive'],
525 {'<rw_mgmt_ip>': rw_mgmt_ip}
526 )
527
528 ########################################################
529 # Check for specific machine placement (native charms) #
530 ########################################################
531 to = ""
532 series = "xenial"
533
534 if machine_spec.keys():
535 if all(k in machine_spec for k in ['hostname', 'username']):
536
537 # Allow series to be derived from the native charm
538 series = None
539
540 self.log.debug("Provisioning manual machine {}@{}".format(
541 machine_spec['username'],
542 machine_spec['hostname'],
543 ))
544
545 """Native Charm support
546
547 Taking a bare VM (assumed to be an Ubuntu cloud image),
548 the provisioning process will:
549 - Create an ubuntu user w/sudo access
550 - Detect hardware
551 - Detect architecture
552 - Download and install Juju agent from controller
553 - Enable Juju agent
554 - Add an iptables rule to route traffic to the API proxy
555 """
556
557 to = await self.provision_machine(
558 model_name=model_name,
559 username=machine_spec['username'],
560 hostname=machine_spec['hostname'],
561 private_key_path=self.GetPrivateKeyPath(),
562 )
563 self.log.debug("Provisioned machine id {}".format(to))
564
565 # TODO: If to is none, raise an exception
566
567 # The native charm won't have the sshproxy layer, typically, but LCM uses the config primitive
568 # to interpret what the values are. That's a gap to fill.
569
570 """
571 The ssh-* config parameters are unique to the sshproxy layer,
572 which most native charms will not be aware of.
573
574 Setting invalid config parameters will cause the deployment to
575 fail.
576
577 For the moment, we will strip the ssh-* parameters from native
578 charms, until the feature gap is addressed in the information
579 model.
580 """
581
582 # Native charms don't include the ssh-* config values, so strip them
583 # from the initial_config, otherwise the deploy will raise an error.
584 # self.log.debug("Removing ssh-* from initial-config")
585 for k in ['ssh-hostname', 'ssh-username', 'ssh-password']:
586 if k in initial_config:
587 self.log.debug("Removing parameter {}".format(k))
588 del initial_config[k]
589
590 self.log.debug("JujuApi: Deploying charm ({}/{}) from {} to {}".format(
591 model_name,
592 application_name,
593 charm_path,
594 to,
595 ))
596
597 ########################################################
598 # Deploy the charm and apply the initial configuration #
599 ########################################################
600 app = await model.deploy(
601 # We expect charm_path to be either the path to the charm on disk
602 # or in the format of cs:series/name
603 charm_path,
604 # This is the formatted, unique name for this charm
605 application_name=application_name,
606 # Proxy charms should use the current LTS. This will need to be
607 # changed for native charms.
608 series=series,
609 # Apply the initial 'config' primitive during deployment
610 config=initial_config,
611 # Where to deploy the charm to.
612 to=to,
613 )
614
615 #############################
616 # Map the vdu id<->app name #
617 #############################
618 try:
619 await self.Relate(model_name, vnfd)
620 except KeyError as ex:
621 # We don't currently support relations between NS and VNF/VDU charms
622 self.log.warn("[N2VC] Relations not supported: {}".format(ex))
623 except Exception as ex:
624 # This may happen if not all of the charms needed by the relation
625 # are ready. We can safely ignore this, because Relate will be
626 # retried when the endpoint of the relation is deployed.
627 self.log.warn("[N2VC] Relations not ready")
628
629 # #######################################
630 # # Execute initial config primitive(s) #
631 # #######################################
632 uuids = await self.ExecuteInitialPrimitives(
633 model_name,
634 application_name,
635 params,
636 )
637 return uuids
638
639 # primitives = {}
640 #
641 # # Build a sequential list of the primitives to execute
642 # for primitive in params['initial-config-primitive']:
643 # try:
644 # if primitive['name'] == 'config':
645 # # This is applied when the Application is deployed
646 # pass
647 # else:
648 # seq = primitive['seq']
649 #
650 # params = {}
651 # if 'parameter' in primitive:
652 # params = primitive['parameter']
653 #
654 # primitives[seq] = {
655 # 'name': primitive['name'],
656 # 'parameters': self._map_primitive_parameters(
657 # params,
658 # {'<rw_mgmt_ip>': rw_mgmt_ip}
659 # ),
660 # }
661 #
662 # for primitive in sorted(primitives):
663 # await self.ExecutePrimitive(
664 # model_name,
665 # application_name,
666 # primitives[primitive]['name'],
667 # callback,
668 # callback_args,
669 # **primitives[primitive]['parameters'],
670 # )
671 # except N2VCPrimitiveExecutionFailed as e:
672 # self.log.debug(
673 # "[N2VC] Exception executing primitive: {}".format(e)
674 # )
675 # raise
676
677 async def GetPrimitiveStatus(self, model_name, uuid):
678 """Get the status of an executed Primitive.
679
680 The status of an executed Primitive will be one of three values:
681 - completed
682 - failed
683 - running
684 """
685 status = None
686 try:
687 if not self.authenticated:
688 await self.login()
689
690 model = await self.get_model(model_name)
691
692 results = await model.get_action_status(uuid)
693
694 if uuid in results:
695 status = results[uuid]
696
697 except Exception as e:
698 self.log.debug(
699 "Caught exception while getting primitive status: {}".format(e)
700 )
701 raise N2VCPrimitiveExecutionFailed(e)
702
703 return status
704
705 async def GetPrimitiveOutput(self, model_name, uuid):
706 """Get the output of an executed Primitive.
707
708 Note: this only returns output for a successfully executed primitive.
709 """
710 results = None
711 try:
712 if not self.authenticated:
713 await self.login()
714
715 model = await self.get_model(model_name)
716 results = await model.get_action_output(uuid, 60)
717 except Exception as e:
718 self.log.debug(
719 "Caught exception while getting primitive status: {}".format(e)
720 )
721 raise N2VCPrimitiveExecutionFailed(e)
722
723 return results
724
725 # async def ProvisionMachine(self, model_name, hostname, username):
726 # """Provision machine for usage with Juju.
727 #
728 # Provisions a previously instantiated machine for use with Juju.
729 # """
730 # try:
731 # if not self.authenticated:
732 # await self.login()
733 #
734 # # FIXME: This is hard-coded until model-per-ns is added
735 # model_name = 'default'
736 #
737 # model = await self.get_model(model_name)
738 # model.add_machine(spec={})
739 #
740 # machine = await model.add_machine(spec='ssh:{}@{}:{}'.format(
741 # "ubuntu",
742 # host['address'],
743 # private_key_path,
744 # ))
745 # return machine.id
746 #
747 # except Exception as e:
748 # self.log.debug(
749 # "Caught exception while getting primitive status: {}".format(e)
750 # )
751 # raise N2VCPrimitiveExecutionFailed(e)
752
753 def GetPrivateKeyPath(self):
754 homedir = os.environ['HOME']
755 sshdir = "{}/.ssh".format(homedir)
756 private_key_path = "{}/id_n2vc_rsa".format(sshdir)
757 return private_key_path
758
759 async def GetPublicKey(self):
760 """Get the N2VC SSH public key.abs
761
762 Returns the SSH public key, to be injected into virtual machines to
763 be managed by the VCA.
764
765 The first time this is run, a ssh keypair will be created. The public
766 key is injected into a VM so that we can provision the machine with
767 Juju, after which Juju will communicate with the VM directly via the
768 juju agent.
769 """
770 public_key = ""
771
772 # Find the path to where we expect our key to live.
773 homedir = os.environ['HOME']
774 sshdir = "{}/.ssh".format(homedir)
775 if not os.path.exists(sshdir):
776 os.mkdir(sshdir)
777
778 private_key_path = "{}/id_n2vc_rsa".format(sshdir)
779 public_key_path = "{}.pub".format(private_key_path)
780
781 # If we don't have a key generated, generate it.
782 if not os.path.exists(private_key_path):
783 cmd = "ssh-keygen -t {} -b {} -N '' -f {}".format(
784 "rsa",
785 "4096",
786 private_key_path
787 )
788 subprocess.check_output(shlex.split(cmd))
789
790 # Read the public key
791 with open(public_key_path, "r") as f:
792 public_key = f.readline()
793
794 return public_key
795
796 async def ExecuteInitialPrimitives(self, model_name, application_name,
797 params, callback=None, *callback_args):
798 """Execute multiple primitives.
799
800 Execute multiple primitives as declared in initial-config-primitive.
801 This is useful in cases where the primitives initially failed -- for
802 example, if the charm is a proxy but the proxy hasn't been configured
803 yet.
804 """
805 uuids = []
806 primitives = {}
807
808 # Build a sequential list of the primitives to execute
809 for primitive in params['initial-config-primitive']:
810 try:
811 if primitive['name'] == 'config':
812 pass
813 else:
814 seq = primitive['seq']
815
816 params_ = {}
817 if 'parameter' in primitive:
818 params_ = primitive['parameter']
819
820 user_values = params.get("user_values", {})
821 if 'rw_mgmt_ip' not in user_values:
822 user_values['rw_mgmt_ip'] = None
823 # just for backward compatibility, because it will be provided always by modern version of LCM
824
825 primitives[seq] = {
826 'name': primitive['name'],
827 'parameters': self._map_primitive_parameters(
828 params_,
829 user_values
830 ),
831 }
832
833 for primitive in sorted(primitives):
834 try:
835 # self.log.debug("Queuing action {}".format(primitives[primitive]['name']))
836 uuids.append(
837 await self.ExecutePrimitive(
838 model_name,
839 application_name,
840 primitives[primitive]['name'],
841 callback,
842 callback_args,
843 **primitives[primitive]['parameters'],
844 )
845 )
846 except PrimitiveDoesNotExist as e:
847 self.log.debug("Ignoring exception PrimitiveDoesNotExist: {}".format(e))
848 pass
849 except Exception as e:
850 self.log.debug("XXXXXXXXXXXXXXXXXXXXXXXXX Unexpected exception: {}".format(e))
851 raise e
852
853 except N2VCPrimitiveExecutionFailed as e:
854 self.log.debug(
855 "[N2VC] Exception executing primitive: {}".format(e)
856 )
857 raise
858 return uuids
859
860 async def ExecutePrimitive(self, model_name, application_name, primitive,
861 callback, *callback_args, **params):
862 """Execute a primitive of a charm for Day 1 or Day 2 configuration.
863
864 Execute a primitive defined in the VNF descriptor.
865
866 :param str model_name: The name or unique id of the network service.
867 :param str application_name: The name of the application
868 :param str primitive: The name of the primitive to execute.
869 :param obj callback: A callback function to receive status changes.
870 :param tuple callback_args: A list of arguments to be passed to the
871 callback function.
872 :param dict params: A dictionary of key=value pairs representing the
873 primitive's parameters
874 Examples::
875 {
876 'rw_mgmt_ip': '1.2.3.4',
877 # Pass the initial-config-primitives section of the vnf or vdu
878 'initial-config-primitives': {...}
879 }
880 """
881 self.log.debug("Executing primitive={} params={}".format(primitive, params))
882 uuid = None
883 try:
884 if not self.authenticated:
885 await self.login()
886
887 model = await self.get_model(model_name)
888
889 if primitive == 'config':
890 # config is special, and expecting params to be a dictionary
891 await self.set_config(
892 model,
893 application_name,
894 params['params'],
895 )
896 else:
897 app = await self.get_application(model, application_name)
898 if app:
899 # Does this primitive exist?
900 actions = await app.get_actions()
901
902 if primitive not in actions.keys():
903 raise PrimitiveDoesNotExist("Primitive {} does not exist".format(primitive))
904
905 # Run against the first (and probably only) unit in the app
906 unit = app.units[0]
907 if unit:
908 action = await unit.run_action(primitive, **params)
909 uuid = action.id
910 except PrimitiveDoesNotExist as e:
911 # Catch and raise this exception if it's thrown from the inner block
912 raise e
913 except Exception as e:
914 # An unexpected exception was caught
915 self.log.debug(
916 "Caught exception while executing primitive: {}".format(e)
917 )
918 raise N2VCPrimitiveExecutionFailed(e)
919 return uuid
920
921 async def RemoveCharms(self, model_name, application_name, callback=None,
922 *callback_args):
923 """Remove a charm from the VCA.
924
925 Remove a charm referenced in a VNF Descriptor.
926
927 :param str model_name: The name of the network service.
928 :param str application_name: The name of the application
929 :param obj callback: A callback function to receive status changes.
930 :param tuple callback_args: A list of arguments to be passed to the
931 callback function.
932 """
933 try:
934 if not self.authenticated:
935 await self.login()
936
937 model = await self.get_model(model_name)
938 app = await self.get_application(model, application_name)
939 if app:
940 # Remove this application from event monitoring
941 await self.Unsubscribe(model_name, application_name)
942
943 # self.notify_callback(model_name, application_name, "removing", callback, *callback_args)
944 self.log.debug(
945 "Removing the application {}".format(application_name)
946 )
947 await app.remove()
948
949 # await self.disconnect_model(self.monitors[model_name])
950
951 self.notify_callback(
952 model_name,
953 application_name,
954 "removed",
955 "Removing charm {}".format(application_name),
956 callback,
957 *callback_args,
958 )
959
960 except Exception as e:
961 print("Caught exception: {}".format(e))
962 self.log.debug(e)
963 raise e
964
965 async def CreateNetworkService(self, ns_uuid):
966 """Create a new Juju model for the Network Service.
967
968 Creates a new Model in the Juju Controller.
969
970 :param str ns_uuid: A unique id representing an instaance of a
971 Network Service.
972
973 :returns: True if the model was created. Raises JujuError on failure.
974 """
975 if not self.authenticated:
976 await self.login()
977
978 models = await self.controller.list_models()
979 if ns_uuid not in models:
980 # Get the new model
981 await self.get_model(ns_uuid)
982
983 return True
984
985 async def DestroyNetworkService(self, ns_uuid):
986 """Destroy a Network Service.
987
988 Destroy the Network Service and any deployed charms.
989
990 :param ns_uuid The unique id of the Network Service
991
992 :returns: True if the model was created. Raises JujuError on failure.
993 """
994
995 # Do not delete the default model. The default model was used by all
996 # Network Services, prior to the implementation of a model per NS.
997 if ns_uuid.lower() == "default":
998 return False
999
1000 if not self.authenticated:
1001 await self.login()
1002
1003 models = await self.controller.list_models()
1004 if ns_uuid in models:
1005 model = await self.controller.get_model(ns_uuid)
1006
1007 for application in model.applications:
1008 app = model.applications[application]
1009
1010 await self.RemoveCharms(ns_uuid, application)
1011
1012 self.log.debug("Unsubscribing Watcher for {}".format(application))
1013 await self.Unsubscribe(ns_uuid, application)
1014
1015 self.log.debug("Waiting for application to terminate")
1016 timeout = 30
1017 try:
1018 await model.block_until(
1019 lambda: all(
1020 unit.workload_status in ['terminated'] for unit in app.units
1021 ),
1022 timeout=timeout
1023 )
1024 except Exception as e:
1025 self.log.debug("Timed out waiting for {} to terminate.".format(application))
1026
1027 for machine in model.machines:
1028 try:
1029 self.log.debug("Destroying machine {}".format(machine))
1030 await model.machines[machine].destroy(force=True)
1031 except JujuAPIError as e:
1032 if 'does not exist' in str(e):
1033 # Our cached model may be stale, because the machine
1034 # has already been removed. It's safe to continue.
1035 continue
1036 else:
1037 self.log.debug("Caught exception: {}".format(e))
1038 raise e
1039
1040 # Disconnect from the Model
1041 if ns_uuid in self.models:
1042 self.log.debug("Disconnecting model {}".format(ns_uuid))
1043 # await self.disconnect_model(self.models[ns_uuid])
1044 await self.disconnect_model(ns_uuid)
1045
1046 try:
1047 self.log.debug("Destroying model {}".format(ns_uuid))
1048 await self.controller.destroy_models(ns_uuid)
1049 except JujuError:
1050 raise NetworkServiceDoesNotExist(
1051 "The Network Service '{}' does not exist".format(ns_uuid)
1052 )
1053
1054 return True
1055
1056 async def GetMetrics(self, model_name, application_name):
1057 """Get the metrics collected by the VCA.
1058
1059 :param model_name The name or unique id of the network service
1060 :param application_name The name of the application
1061 """
1062 metrics = {}
1063 model = await self.get_model(model_name)
1064 app = await self.get_application(model, application_name)
1065 if app:
1066 metrics = await app.get_metrics()
1067
1068 return metrics
1069
1070 async def HasApplication(self, model_name, application_name):
1071 model = await self.get_model(model_name)
1072 app = await self.get_application(model, application_name)
1073 if app:
1074 return True
1075 return False
1076
1077 async def Subscribe(self, ns_name, application_name, callback, *callback_args):
1078 """Subscribe to callbacks for an application.
1079
1080 :param ns_name str: The name of the Network Service
1081 :param application_name str: The name of the application
1082 :param callback obj: The callback method
1083 :param callback_args list: The list of arguments to append to calls to
1084 the callback method
1085 """
1086 self.monitors[ns_name].AddApplication(
1087 application_name,
1088 callback,
1089 *callback_args
1090 )
1091
1092 async def Unsubscribe(self, ns_name, application_name):
1093 """Unsubscribe to callbacks for an application.
1094
1095 Unsubscribes the caller from notifications from a deployed application.
1096
1097 :param ns_name str: The name of the Network Service
1098 :param application_name str: The name of the application
1099 """
1100 self.monitors[ns_name].RemoveApplication(
1101 application_name,
1102 )
1103
1104 # Non-public methods
1105 async def add_relation(self, model_name, relation1, relation2):
1106 """
1107 Add a relation between two application endpoints.
1108
1109 :param str model_name: The name or unique id of the network service
1110 :param str relation1: '<application>[:<relation_name>]'
1111 :param str relation2: '<application>[:<relation_name>]'
1112 """
1113
1114 if not self.authenticated:
1115 await self.login()
1116
1117 m = await self.get_model(model_name)
1118 try:
1119 await m.add_relation(relation1, relation2)
1120 except JujuAPIError as e:
1121 # If one of the applications in the relationship doesn't exist,
1122 # or the relation has already been added, let the operation fail
1123 # silently.
1124 if 'not found' in e.message:
1125 return
1126 if 'already exists' in e.message:
1127 return
1128
1129 raise e
1130
1131 # async def apply_config(self, config, application):
1132 # """Apply a configuration to the application."""
1133 # print("JujuApi: Applying configuration to {}.".format(
1134 # application
1135 # ))
1136 # return await self.set_config(application=application, config=config)
1137
1138 def _get_config_from_dict(self, config_primitive, values):
1139 """Transform the yang config primitive to dict.
1140
1141 Expected result:
1142
1143 config = {
1144 'config':
1145 }
1146 """
1147 config = {}
1148 for primitive in config_primitive:
1149 if primitive['name'] == 'config':
1150 # config = self._map_primitive_parameters()
1151 for parameter in primitive['parameter']:
1152 param = str(parameter['name'])
1153 if parameter['value'] == "<rw_mgmt_ip>":
1154 config[param] = str(values[parameter['value']])
1155 else:
1156 config[param] = str(parameter['value'])
1157
1158 return config
1159
1160 def _map_primitive_parameters(self, parameters, user_values):
1161 params = {}
1162 for parameter in parameters:
1163 param = str(parameter['name'])
1164 value = parameter.get('value')
1165
1166 # map parameters inside a < >; e.g. <rw_mgmt_ip>. with the provided user_values.
1167 # Must exist at user_values except if there is a default value
1168 if isinstance(value, str) and value.startswith("<") and value.endswith(">"):
1169 if parameter['value'][1:-1] in user_values:
1170 value = user_values[parameter['value'][1:-1]]
1171 elif 'default-value' in parameter:
1172 value = parameter['default-value']
1173 else:
1174 raise KeyError("parameter {}='{}' not supplied ".format(param, value))
1175
1176 # If there's no value, use the default-value (if set)
1177 if value is None and 'default-value' in parameter:
1178 value = parameter['default-value']
1179
1180 # Typecast parameter value, if present
1181 paramtype = "string"
1182 try:
1183 if 'data-type' in parameter:
1184 paramtype = str(parameter['data-type']).lower()
1185
1186 if paramtype == "integer":
1187 value = int(value)
1188 elif paramtype == "boolean":
1189 value = bool(value)
1190 else:
1191 value = str(value)
1192 else:
1193 # If there's no data-type, assume the value is a string
1194 value = str(value)
1195 except ValueError:
1196 raise ValueError("parameter {}='{}' cannot be converted to type {}".format(param, value, paramtype))
1197
1198 params[param] = value
1199 return params
1200
1201 def _get_config_from_yang(self, config_primitive, values):
1202 """Transform the yang config primitive to dict."""
1203 config = {}
1204 for primitive in config_primitive.values():
1205 if primitive['name'] == 'config':
1206 for parameter in primitive['parameter'].values():
1207 param = str(parameter['name'])
1208 if parameter['value'] == "<rw_mgmt_ip>":
1209 config[param] = str(values[parameter['value']])
1210 else:
1211 config[param] = str(parameter['value'])
1212
1213 return config
1214
1215 def FormatApplicationName(self, *args):
1216 """
1217 Generate a Juju-compatible Application name
1218
1219 :param args tuple: Positional arguments to be used to construct the
1220 application name.
1221
1222 Limitations::
1223 - Only accepts characters a-z and non-consequitive dashes (-)
1224 - Application name should not exceed 50 characters
1225
1226 Examples::
1227
1228 FormatApplicationName("ping_pong_ns", "ping_vnf", "a")
1229 """
1230 appname = ""
1231 for c in "-".join(list(args)):
1232 if c.isdigit():
1233 c = chr(97 + int(c))
1234 elif not c.isalpha():
1235 c = "-"
1236 appname += c
1237 return re.sub('-+', '-', appname.lower())
1238
1239 # def format_application_name(self, nsd_name, vnfr_name, member_vnf_index=0):
1240 # """Format the name of the application
1241 #
1242 # Limitations:
1243 # - Only accepts characters a-z and non-consequitive dashes (-)
1244 # - Application name should not exceed 50 characters
1245 # """
1246 # name = "{}-{}-{}".format(nsd_name, vnfr_name, member_vnf_index)
1247 # new_name = ''
1248 # for c in name:
1249 # if c.isdigit():
1250 # c = chr(97 + int(c))
1251 # elif not c.isalpha():
1252 # c = "-"
1253 # new_name += c
1254 # return re.sub('\-+', '-', new_name.lower())
1255
1256 def format_model_name(self, name):
1257 """Format the name of model.
1258
1259 Model names may only contain lowercase letters, digits and hyphens
1260 """
1261
1262 return name.replace('_', '-').lower()
1263
1264 async def get_application(self, model, application):
1265 """Get the deployed application."""
1266 if not self.authenticated:
1267 await self.login()
1268
1269 app = None
1270 if application and model:
1271 if model.applications:
1272 if application in model.applications:
1273 app = model.applications[application]
1274
1275 return app
1276
1277 async def get_model(self, model_name):
1278 """Get a model from the Juju Controller.
1279
1280 Note: Model objects returned must call disconnected() before it goes
1281 out of scope."""
1282 if not self.authenticated:
1283 await self.login()
1284
1285 if model_name not in self.models:
1286 # Get the models in the controller
1287 models = await self.controller.list_models()
1288
1289 if model_name not in models:
1290 try:
1291 self.models[model_name] = await self.controller.add_model(
1292 model_name,
1293 config={'authorized-keys': self.juju_public_key}
1294
1295 )
1296 except JujuError as e:
1297 if "already exists" not in e.message:
1298 raise e
1299 else:
1300 self.models[model_name] = await self.controller.get_model(
1301 model_name
1302 )
1303
1304 self.refcount['model'] += 1
1305
1306 # Create an observer for this model
1307 await self.create_model_monitor(model_name)
1308
1309 return self.models[model_name]
1310
1311 async def create_model_monitor(self, model_name):
1312 """Create a monitor for the model, if none exists."""
1313 if not self.authenticated:
1314 await self.login()
1315
1316 if model_name not in self.monitors:
1317 self.monitors[model_name] = VCAMonitor(model_name)
1318 self.models[model_name].add_observer(self.monitors[model_name])
1319
1320 return True
1321
1322 async def login(self):
1323 """Login to the Juju controller."""
1324
1325 if self.authenticated:
1326 return
1327
1328 self.connecting = True
1329
1330 self.log.debug("JujuApi: Logging into controller")
1331
1332 self.controller = Controller(loop=self.loop)
1333
1334 if self.secret:
1335 self.log.debug(
1336 "Connecting to controller... ws://{} as {}/{}".format(
1337 self.endpoint,
1338 self.user,
1339 self.secret,
1340 )
1341 )
1342 try:
1343 await self.controller.connect(
1344 endpoint=self.endpoint,
1345 username=self.user,
1346 password=self.secret,
1347 cacert=self.ca_cert,
1348 )
1349 self.refcount['controller'] += 1
1350 self.authenticated = True
1351 self.log.debug("JujuApi: Logged into controller")
1352 except Exception as ex:
1353 self.log.debug("Caught exception: {}".format(ex))
1354 else:
1355 # current_controller no longer exists
1356 # self.log.debug("Connecting to current controller...")
1357 # await self.controller.connect_current()
1358 # await self.controller.connect(
1359 # endpoint=self.endpoint,
1360 # username=self.user,
1361 # cacert=cacert,
1362 # )
1363 self.log.fatal("VCA credentials not configured.")
1364 self.authenticated = False
1365
1366
1367 async def logout(self):
1368 """Logout of the Juju controller."""
1369 if not self.authenticated:
1370 return False
1371
1372 try:
1373 for model in self.models:
1374 await self.disconnect_model(model)
1375
1376 if self.controller:
1377 self.log.debug("Disconnecting controller {}".format(
1378 self.controller
1379 ))
1380 await self.controller.disconnect()
1381 self.refcount['controller'] -= 1
1382 self.controller = None
1383
1384 self.authenticated = False
1385
1386 self.log.debug(self.refcount)
1387
1388 except Exception as e:
1389 self.log.fatal(
1390 "Fatal error logging out of Juju Controller: {}".format(e)
1391 )
1392 raise e
1393 return True
1394
1395 async def disconnect_model(self, model):
1396 self.log.debug("Disconnecting model {}".format(model))
1397 if model in self.models:
1398 try:
1399 await self.models[model].disconnect()
1400 self.refcount['model'] -= 1
1401 self.models[model] = None
1402 except Exception as e:
1403 self.log.debug("Caught exception: {}".format(e))
1404
1405 async def provision_machine(self, model_name: str,
1406 hostname: str, username: str,
1407 private_key_path: str) -> int:
1408 """Provision a machine.
1409
1410 This executes the SSH provisioner, which will log in to a machine via
1411 SSH and prepare it for use with the Juju model
1412
1413 :param model_name str: The name of the model
1414 :param hostname str: The IP or hostname of the target VM
1415 :param user str: The username to login to
1416 :param private_key_path str: The path to the private key that's been injected to the VM via cloud-init
1417 :return machine_id int: Returns the id of the machine or None if provisioning fails
1418 """
1419 if not self.authenticated:
1420 await self.login()
1421
1422 machine_id = None
1423
1424 if self.api_proxy:
1425 self.log.debug("Instantiating SSH Provisioner for {}@{} ({})".format(
1426 username,
1427 hostname,
1428 private_key_path
1429 ))
1430 provisioner = SSHProvisioner(
1431 host=hostname,
1432 user=username,
1433 private_key_path=private_key_path,
1434 log=self.log,
1435 )
1436
1437 params = None
1438 try:
1439 params = provisioner.provision_machine()
1440 except Exception as ex:
1441 self.log.debug("caught exception from provision_machine: {}".format(ex))
1442 return None
1443
1444 if params:
1445 params.jobs = ['JobHostUnits']
1446
1447 model = await self.get_model(model_name)
1448
1449 connection = model.connection()
1450
1451 # Submit the request.
1452 self.log.debug("Adding machine to model")
1453 client_facade = client.ClientFacade.from_connection(connection)
1454 results = await client_facade.AddMachines(params=[params])
1455 error = results.machines[0].error
1456 if error:
1457 raise ValueError("Error adding machine: %s" % error.message)
1458
1459 machine_id = results.machines[0].machine
1460
1461 # Need to run this after AddMachines has been called,
1462 # as we need the machine_id
1463 self.log.debug("Installing Juju agent")
1464 await provisioner.install_agent(
1465 connection,
1466 params.nonce,
1467 machine_id,
1468 self.api_proxy,
1469 )
1470 else:
1471 self.log.debug("Missing API Proxy")
1472 return machine_id
1473
1474 # async def remove_application(self, name):
1475 # """Remove the application."""
1476 # if not self.authenticated:
1477 # await self.login()
1478 #
1479 # app = await self.get_application(name)
1480 # if app:
1481 # self.log.debug("JujuApi: Destroying application {}".format(
1482 # name,
1483 # ))
1484 #
1485 # await app.destroy()
1486
1487 async def remove_relation(self, a, b):
1488 """
1489 Remove a relation between two application endpoints
1490
1491 :param a An application endpoint
1492 :param b An application endpoint
1493 """
1494 if not self.authenticated:
1495 await self.login()
1496
1497 m = await self.get_model()
1498 try:
1499 m.remove_relation(a, b)
1500 finally:
1501 await m.disconnect()
1502
1503 async def resolve_error(self, model_name, application=None):
1504 """Resolve units in error state."""
1505 if not self.authenticated:
1506 await self.login()
1507
1508 model = await self.get_model(model_name)
1509
1510 app = await self.get_application(model, application)
1511 if app:
1512 self.log.debug(
1513 "JujuApi: Resolving errors for application {}".format(
1514 application,
1515 )
1516 )
1517
1518 for unit in app.units:
1519 app.resolved(retry=True)
1520
1521 async def run_action(self, model_name, application, action_name, **params):
1522 """Execute an action and return an Action object."""
1523 if not self.authenticated:
1524 await self.login()
1525 result = {
1526 'status': '',
1527 'action': {
1528 'tag': None,
1529 'results': None,
1530 }
1531 }
1532
1533 model = await self.get_model(model_name)
1534
1535 app = await self.get_application(model, application)
1536 if app:
1537 # We currently only have one unit per application
1538 # so use the first unit available.
1539 unit = app.units[0]
1540
1541 self.log.debug(
1542 "JujuApi: Running Action {} against Application {}".format(
1543 action_name,
1544 application,
1545 )
1546 )
1547
1548 action = await unit.run_action(action_name, **params)
1549
1550 # Wait for the action to complete
1551 await action.wait()
1552
1553 result['status'] = action.status
1554 result['action']['tag'] = action.data['id']
1555 result['action']['results'] = action.results
1556
1557 return result
1558
1559 async def set_config(self, model_name, application, config):
1560 """Apply a configuration to the application."""
1561 if not self.authenticated:
1562 await self.login()
1563
1564 app = await self.get_application(model_name, application)
1565 if app:
1566 self.log.debug("JujuApi: Setting config for Application {}".format(
1567 application,
1568 ))
1569 await app.set_config(config)
1570
1571 # Verify the config is set
1572 newconf = await app.get_config()
1573 for key in config:
1574 if config[key] != newconf[key]['value']:
1575 self.log.debug("JujuApi: Config not set! Key {} Value {} doesn't match {}".format(key, config[key], newconf[key]))
1576
1577 # async def set_parameter(self, parameter, value, application=None):
1578 # """Set a config parameter for a service."""
1579 # if not self.authenticated:
1580 # await self.login()
1581 #
1582 # self.log.debug("JujuApi: Setting {}={} for Application {}".format(
1583 # parameter,
1584 # value,
1585 # application,
1586 # ))
1587 # return await self.apply_config(
1588 # {parameter: value},
1589 # application=application,
1590 # )
1591
1592 async def wait_for_application(self, model_name, application_name,
1593 timeout=300):
1594 """Wait for an application to become active."""
1595 if not self.authenticated:
1596 await self.login()
1597
1598 model = await self.get_model(model_name)
1599
1600 app = await self.get_application(model, application_name)
1601 self.log.debug("Application: {}".format(app))
1602 if app:
1603 self.log.debug(
1604 "JujuApi: Waiting {} seconds for Application {}".format(
1605 timeout,
1606 application_name,
1607 )
1608 )
1609
1610 await model.block_until(
1611 lambda: all(
1612 unit.agent_status == 'idle' and unit.workload_status in
1613 ['active', 'unknown'] for unit in app.units
1614 ),
1615 timeout=timeout
1616 )