13 from concurrent
.futures
import CancelledError
14 from functools
import partial
15 from pathlib
import Path
19 import theblues
.charmstore
20 import theblues
.errors
22 from . import tag
, utils
23 from .client
import client
24 from .client
import connection
25 from .client
.client
import ConfigValue
26 from .constraints
import parse
as parse_constraints
, normalize_key
27 from .delta
import get_entity_delta
28 from .delta
import get_entity_class
29 from .exceptions
import DeadEntityException
30 from .errors
import JujuError
, JujuAPIError
31 from .placement
import parse
as parse_placement
33 log
= logging
.getLogger(__name__
)
36 class _Observer(object):
37 """Wrapper around an observer callable.
39 This wrapper allows filter criteria to be associated with the
40 callable so that it's only called for changes that meet the criteria.
43 def __init__(self
, callable_
, entity_type
, action
, entity_id
, predicate
):
44 self
.callable_
= callable_
45 self
.entity_type
= entity_type
47 self
.entity_id
= entity_id
48 self
.predicate
= predicate
50 self
.entity_id
= str(self
.entity_id
)
51 if not self
.entity_id
.startswith('^'):
52 self
.entity_id
= '^' + self
.entity_id
53 if not self
.entity_id
.endswith('$'):
56 async def __call__(self
, delta
, old
, new
, model
):
57 await self
.callable_(delta
, old
, new
, model
)
59 def cares_about(self
, delta
):
60 """Return True if this observer "cares about" (i.e. wants to be
61 called) for a this delta.
64 if (self
.entity_id
and delta
.get_id() and
65 not re
.match(self
.entity_id
, str(delta
.get_id()))):
68 if self
.entity_type
and self
.entity_type
!= delta
.entity
:
71 if self
.action
and self
.action
!= delta
.type:
74 if self
.predicate
and not self
.predicate(delta
):
80 class ModelObserver(object):
82 Base class for creating observers that react to changes in a model.
84 async def __call__(self
, delta
, old
, new
, model
):
85 handler_name
= 'on_{}_{}'.format(delta
.entity
, delta
.type)
86 method
= getattr(self
, handler_name
, self
.on_change
)
87 await method(delta
, old
, new
, model
)
89 async def on_change(self
, delta
, old
, new
, model
):
90 """Generic model-change handler.
92 This should be overridden in a subclass.
94 :param delta: :class:`juju.client.overrides.Delta`
95 :param old: :class:`juju.model.ModelEntity`
96 :param new: :class:`juju.model.ModelEntity`
97 :param model: :class:`juju.model.Model`
103 class ModelState(object):
104 """Holds the state of the model, including the delta history of all
105 entities in the model.
108 def __init__(self
, model
):
112 def _live_entity_map(self
, entity_type
):
113 """Return an id:Entity map of all the living entities of
114 type ``entity_type``.
118 entity_id
: self
.get_entity(entity_type
, entity_id
)
119 for entity_id
, history
in self
.state
.get(entity_type
, {}).items()
120 if history
[-1] is not None
124 def applications(self
):
125 """Return a map of application-name:Application for all applications
126 currently in the model.
129 return self
._live
_entity
_map
('application')
133 """Return a map of machine-id:Machine for all machines currently in
137 return self
._live
_entity
_map
('machine')
141 """Return a map of unit-id:Unit for all units currently in
145 return self
._live
_entity
_map
('unit')
147 def entity_history(self
, entity_type
, entity_id
):
148 """Return the history deque for an entity.
151 return self
.state
[entity_type
][entity_id
]
153 def entity_data(self
, entity_type
, entity_id
, history_index
):
154 """Return the data dict for an entity at a specific index of its
158 return self
.entity_history(entity_type
, entity_id
)[history_index
]
160 def apply_delta(self
, delta
):
161 """Apply delta to our state and return a copy of the
162 affected object as it was before and after the update, e.g.:
164 old_obj, new_obj = self.apply_delta(delta)
166 old_obj may be None if the delta is for the creation of a new object,
167 e.g. a new application or unit is deployed.
169 new_obj will never be None, but may be dead (new_obj.dead == True)
170 if the object was deleted as a result of the delta being applied.
175 .setdefault(delta
.entity
, {})
176 .setdefault(delta
.get_id(), collections
.deque())
179 history
.append(delta
.data
)
180 if delta
.type == 'remove':
183 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
184 return entity
.previous(), entity
187 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
188 """Return an object instance for the given entity_type and id.
190 By default the object state matches the most recent state from
191 Juju. To get an instance of the object in an older state, pass
192 history_index, an index into the history deque for the entity.
196 if history_index
< 0 and history_index
!= -1:
197 history_index
+= len(self
.entity_history(entity_type
, entity_id
))
198 if history_index
< 0:
202 self
.entity_data(entity_type
, entity_id
, history_index
)
206 entity_class
= get_entity_class(entity_type
)
208 entity_id
, self
.model
, history_index
=history_index
,
212 class ModelEntity(object):
213 """An object in the Model tree"""
215 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
216 """Initialize a new entity
218 :param entity_id str: The unique id of the object in the model
219 :param model: The model instance in whose object tree this
221 :history_index int: The index of this object's state in the model's
222 history deque for this entity
223 :connected bool: Flag indicating whether this object gets live updates
227 self
.entity_id
= entity_id
229 self
._history
_index
= history_index
230 self
.connected
= connected
231 self
.connection
= model
.connection
234 return '<{} entity_id="{}">'.format(type(self
).__name
__,
237 def __getattr__(self
, name
):
238 """Fetch object attributes from the underlying data dict held in the
243 return self
.safe_data
[name
]
245 name
= name
.replace('_', '-')
246 if name
in self
.safe_data
:
247 return self
.safe_data
[name
]
252 return bool(self
.data
)
254 def on_change(self
, callable_
):
255 """Add a change observer to this entity.
258 self
.model
.add_observer(
259 callable_
, self
.entity_type
, 'change', self
.entity_id
)
261 def on_remove(self
, callable_
):
262 """Add a remove observer to this entity.
265 self
.model
.add_observer(
266 callable_
, self
.entity_type
, 'remove', self
.entity_id
)
269 def entity_type(self
):
270 """A string identifying the entity type of this object, e.g.
271 'application' or 'unit', etc.
274 return self
.__class
__.__name
__.lower()
278 """Return True if this object represents the current state of the
279 entity in the underlying model.
281 This will be True except when the object represents an entity at a
282 non-latest state in history, e.g. if the object was obtained by calling
283 .previous() on another object.
286 return self
._history
_index
== -1
290 """Returns True if this entity no longer exists in the underlying
296 self
.model
.state
.entity_data(
297 self
.entity_type
, self
.entity_id
, -1) is None
302 """Returns True if this entity still exists in the underlying
310 """The data dictionary for this entity.
313 return self
.model
.state
.entity_data(
314 self
.entity_type
, self
.entity_id
, self
._history
_index
)
318 """The data dictionary for this entity.
320 If this `ModelEntity` points to the dead state, it will
321 raise `DeadEntityException`.
324 if self
.data
is None:
325 raise DeadEntityException(
326 "Entity {}:{} is dead - its attributes can no longer be "
327 "accessed. Use the .previous() method on this object to get "
328 "a copy of the object at its previous state.".format(
329 self
.entity_type
, self
.entity_id
))
333 """Return a copy of this object as was at its previous state in
336 Returns None if this object is new (and therefore has no history).
338 The returned object is always "disconnected", i.e. does not receive
342 return self
.model
.state
.get_entity(
343 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
347 """Return a copy of this object at its next state in
350 Returns None if this object is already the latest.
352 The returned object is "disconnected", i.e. does not receive
353 live updates, unless it is current (latest).
356 if self
._history
_index
== -1:
359 new_index
= self
._history
_index
+ 1
361 new_index
== len(self
.model
.state
.entity_history(
362 self
.entity_type
, self
.entity_id
)) - 1
364 return self
.model
.state
.get_entity(
365 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
369 """Return a copy of this object at its current state in the model.
371 Returns self if this object is already the latest.
373 The returned object is always "connected", i.e. receives
374 live updates from the model.
377 if self
._history
_index
== -1:
380 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
385 The main API for interacting with a Juju model.
387 def __init__(self
, loop
=None,
388 max_frame_size
=connection
.Connection
.DEFAULT_FRAME_SIZE
):
389 """Instantiate a new connected Model.
391 :param loop: an asyncio event loop
392 :param max_frame_size: See
393 `juju.client.connection.Connection.MAX_FRAME_SIZE`
396 self
.loop
= loop
or asyncio
.get_event_loop()
397 self
.max_frame_size
= max_frame_size
398 self
.connection
= None
399 self
.observers
= weakref
.WeakValueDictionary()
400 self
.state
= ModelState(self
)
402 self
._watch
_stopping
= asyncio
.Event(loop
=self
.loop
)
403 self
._watch
_stopped
= asyncio
.Event(loop
=self
.loop
)
404 self
._watch
_received
= asyncio
.Event(loop
=self
.loop
)
405 self
._charmstore
= CharmStore(self
.loop
)
407 async def __aenter__(self
):
408 await self
.connect_current()
411 async def __aexit__(self
, exc_type
, exc
, tb
):
412 await self
.disconnect()
414 if exc_type
is not None:
417 async def connect(self
, *args
, **kw
):
418 """Connect to an arbitrary Juju model.
420 args and kw are passed through to Connection.connect()
424 kw
['loop'] = self
.loop
425 if 'max_frame_size' not in kw
:
426 kw
['max_frame_size'] = self
.max_frame_size
427 self
.connection
= await connection
.Connection
.connect(*args
, **kw
)
428 await self
._after
_connect
()
430 async def connect_current(self
):
431 """Connect to the current Juju model.
434 self
.connection
= await connection
.Connection
.connect_current(
435 self
.loop
, max_frame_size
=self
.max_frame_size
)
436 await self
._after
_connect
()
438 async def connect_model(self
, model_name
):
439 """Connect to a specific Juju model by name.
441 :param model_name: Format [controller:][user/]model
444 self
.connection
= await connection
.Connection
.connect_model(
445 model_name
, self
.loop
, self
.max_frame_size
)
446 await self
._after
_connect
()
448 async def _after_connect(self
):
449 """Run initialization steps after connecting to websocket.
453 await self
._watch
_received
.wait()
454 await self
.get_info()
456 async def disconnect(self
):
457 """Shut down the watcher task and close websockets.
460 if self
.connection
and self
.connection
.is_open
:
461 log
.debug('Stopping watcher task')
462 self
._watch
_stopping
.set()
463 await self
._watch
_stopped
.wait()
464 log
.debug('Closing model connection')
465 await self
.connection
.close()
466 self
.connection
= None
468 async def add_local_charm_dir(self
, charm_dir
, series
):
469 """Upload a local charm to the model.
471 This will automatically generate an archive from
474 :param charm_dir: Path to the charm directory
475 :param series: Charm series
478 fh
= tempfile
.NamedTemporaryFile()
479 CharmArchiveGenerator(charm_dir
).make_archive(fh
.name
)
482 self
.add_local_charm
, fh
, series
, os
.stat(fh
.name
).st_size
)
483 charm_url
= await self
.loop
.run_in_executor(None, func
)
485 log
.debug('Uploaded local charm: %s -> %s', charm_dir
, charm_url
)
488 def add_local_charm(self
, charm_file
, series
, size
=None):
489 """Upload a local charm archive to the model.
491 Returns the 'local:...' url that should be used to deploy the charm.
493 :param charm_file: Path to charm zip archive
494 :param series: Charm series
495 :param size: Size of the archive, in bytes
496 :return str: 'local:...' url for deploying the charm
497 :raises: :class:`JujuError` if the upload fails
499 Uses an https endpoint at the same host:port as the wss.
500 Supports large file uploads.
504 This method will block. Consider using :meth:`add_local_charm_dir`
508 conn
, headers
, path_prefix
= self
.connection
.https_connection()
509 path
= "%s/charms?series=%s" % (path_prefix
, series
)
510 headers
['Content-Type'] = 'application/zip'
512 headers
['Content-Length'] = size
513 conn
.request("POST", path
, charm_file
, headers
)
514 response
= conn
.getresponse()
515 result
= response
.read().decode()
516 if not response
.status
== 200:
517 raise JujuError(result
)
518 result
= json
.loads(result
)
519 return result
['charm-url']
521 def all_units_idle(self
):
522 """Return True if all units are idle.
525 for unit
in self
.units
.values():
526 unit_status
= unit
.data
['agent-status']['current']
527 if unit_status
!= 'idle':
531 async def reset(self
, force
=False):
532 """Reset the model to a clean state.
534 :param bool force: Force-terminate machines.
536 This returns only after the model has reached a clean state. "Clean"
537 means no applications or machines exist in the model.
540 log
.debug('Resetting model')
541 for app
in self
.applications
.values():
543 for machine
in self
.machines
.values():
544 await machine
.destroy(force
=force
)
545 await self
.block_until(
546 lambda: len(self
.machines
) == 0
549 async def block_until(self
, *conditions
, timeout
=None, wait_period
=0.5):
550 """Return only after all conditions are true.
554 while not all(c() for c
in conditions
):
555 if not (self
.connection
and self
.connection
.is_open
):
556 raise websockets
.ConnectionClosed(1006, 'no reason')
557 await asyncio
.sleep(wait_period
, loop
=self
.loop
)
558 await asyncio
.wait_for(_block(), timeout
, loop
=self
.loop
)
561 def applications(self
):
562 """Return a map of application-name:Application for all applications
563 currently in the model.
566 return self
.state
.applications
570 """Return a map of machine-id:Machine for all machines currently in
574 return self
.state
.machines
578 """Return a map of unit-id:Unit for all units currently in
582 return self
.state
.units
584 async def get_info(self
):
585 """Return a client.ModelInfo object for this Model.
587 Retrieves latest info for this Model from the api server. The
588 return value is cached on the Model.info attribute so that the
589 valued may be accessed again without another api call, if
592 This method is called automatically when the Model is connected,
593 resulting in Model.info being initialized without requiring an
594 explicit call to this method.
597 facade
= client
.ClientFacade
.from_connection(self
.connection
)
599 self
.info
= await facade
.ModelInfo()
600 log
.debug('Got ModelInfo: %s', vars(self
.info
))
605 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
607 """Register an "on-model-change" callback
609 Once the model is connected, ``callable_``
610 will be called each time the model changes. ``callable_`` should
611 be Awaitable and accept the following positional arguments:
613 delta - An instance of :class:`juju.delta.EntityDelta`
614 containing the raw delta data recv'd from the Juju
617 old_obj - If the delta modifies an existing object in the model,
618 old_obj will be a copy of that object, as it was before the
619 delta was applied. Will be None if the delta creates a new
622 new_obj - A copy of the new or updated object, after the delta
623 is applied. Will be None if the delta removes an entity
626 model - The :class:`Model` itself.
628 Events for which ``callable_`` is called can be specified by passing
629 entity_type, action, and/or entitiy_id filter criteria, e.g.::
633 entity_type='application', action='add', entity_id='ubuntu')
635 For more complex filtering conditions, pass a predicate function. It
636 will be called with a delta as its only argument. If the predicate
637 function returns True, the ``callable_`` will be called.
640 observer
= _Observer(
641 callable_
, entity_type
, action
, entity_id
, predicate
)
642 self
.observers
[observer
] = callable_
645 """Start an asynchronous watch against this model.
647 See :meth:`add_observer` to register an onchange callback.
650 async def _all_watcher():
652 allwatcher
= client
.AllWatcherFacade
.from_connection(
654 while not self
._watch
_stopping
.is_set():
656 results
= await utils
.run_with_interrupt(
658 self
._watch
_stopping
,
660 except JujuAPIError
as e
:
661 if 'watcher was stopped' not in str(e
):
663 if self
._watch
_stopping
.is_set():
664 # this shouldn't ever actually happen, because
665 # the event should trigger before the controller
666 # has a chance to tell us the watcher is stopped
667 # but handle it gracefully, just in case
669 # controller stopped our watcher for some reason
670 # but we're not actually stopping, so just restart it
672 'Watcher: watcher stopped, restarting')
675 except websockets
.ConnectionClosed
:
676 monitor
= self
.connection
.monitor
677 if monitor
.status
== monitor
.ERROR
:
678 # closed unexpectedly, try to reopen
680 'Watcher: connection closed, reopening')
681 await self
.connection
.reconnect()
685 # closed on request, go ahead and shutdown
687 if self
._watch
_stopping
.is_set():
688 await allwatcher
.Stop()
690 for delta
in results
.deltas
:
691 delta
= get_entity_delta(delta
)
692 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
693 await self
._notify
_observers
(delta
, old_obj
, new_obj
)
694 self
._watch
_received
.set()
695 except CancelledError
:
698 log
.exception('Error in watcher')
701 self
._watch
_stopped
.set()
703 log
.debug('Starting watcher task')
704 self
._watch
_received
.clear()
705 self
._watch
_stopping
.clear()
706 self
._watch
_stopped
.clear()
707 self
.loop
.create_task(_all_watcher())
709 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
710 """Call observing callbacks, notifying them of a change in model state
712 :param delta: The raw change from the watcher
713 (:class:`juju.client.overrides.Delta`)
714 :param old_obj: The object in the model that this delta updates.
716 :param new_obj: The object in the model that is created or updated
717 by applying this delta.
720 if new_obj
and not old_obj
:
724 'Model changed: %s %s %s',
725 delta
.entity
, delta
.type, delta
.get_id())
727 for o
in self
.observers
:
728 if o
.cares_about(delta
):
729 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
),
732 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
734 Block the calling routine until a given action has happened to the
737 :param entity_type: The entity's type.
738 :param entity_id: The entity's id.
739 :param action: the type of action (e.g., 'add', 'change', or 'remove')
740 :param predicate: optional callable that must take as an
741 argument a delta, and must return a boolean, indicating
742 whether the delta contains the specific action we're looking
743 for. For example, you might check to see whether a 'change'
744 has a 'completed' status. See the _Observer class for details.
747 q
= asyncio
.Queue(loop
=self
.loop
)
749 async def callback(delta
, old
, new
, model
):
750 await q
.put(delta
.get_id())
752 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
753 entity_id
= await q
.get()
754 # object might not be in the entity_map if we were waiting for a
756 return self
.state
._live
_entity
_map
(entity_type
).get(entity_id
)
758 async def _wait_for_new(self
, entity_type
, entity_id
=None, predicate
=None):
759 """Wait for a new object to appear in the Model and return it.
761 Waits for an object of type ``entity_type`` with id ``entity_id``.
762 If ``entity_id`` is ``None``, it will wait for the first new entity
765 This coroutine blocks until the new object appears in the model.
768 # if the entity is already in the model, just return it
769 if entity_id
in self
.state
._live
_entity
_map
(entity_type
):
770 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
771 # if we know the entity_id, we can trigger on any action that puts
772 # the enitty into the model; otherwise, we have to watch for the
773 # next "add" action on that entity_type
774 action
= 'add' if entity_id
is None else None
775 return await self
._wait
(entity_type
, entity_id
, action
, predicate
)
777 async def wait_for_action(self
, action_id
):
778 """Given an action, wait for it to complete."""
780 if action_id
.startswith("action-"):
781 # if we've been passed action.tag, transform it into the
782 # id that the api deltas will use.
783 action_id
= action_id
[7:]
785 def predicate(delta
):
786 return delta
.data
['status'] in ('completed', 'failed')
788 return await self
._wait
('action', action_id
, 'change', predicate
)
790 async def add_machine(
791 self
, spec
=None, constraints
=None, disks
=None, series
=None):
792 """Start a new, empty machine and optionally a container, or add a
793 container to a machine.
795 :param str spec: Machine specification
798 (None) - starts a new machine
799 'lxd' - starts a new machine with one lxd container
800 'lxd:4' - starts a new lxd container on machine 4
801 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
802 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
803 'maas2.name' - acquire machine maas2.name on MAAS
805 :param dict constraints: Machine constraints, which can contain the
826 :param list disks: List of disk constraint dictionaries, which can
827 contain the following keys::
841 :param str series: Series, e.g. 'xenial'
843 Supported container types are: lxd, kvm
845 When deploying a container to an existing machine, constraints cannot
849 params
= client
.AddMachineParams()
850 params
.jobs
= ['JobHostUnits']
853 placement
= parse_placement(spec
)
855 params
.placement
= placement
[0]
858 params
.constraints
= client
.Value
.from_json(constraints
)
862 client
.Constraints
.from_json(o
) for o
in disks
]
865 params
.series
= series
867 # Submit the request.
868 client_facade
= client
.ClientFacade
.from_connection(self
.connection
)
869 results
= await client_facade
.AddMachines([params
])
870 error
= results
.machines
[0].error
872 raise ValueError("Error adding machine: %s" % error
.message
)
873 machine_id
= results
.machines
[0].machine
874 log
.debug('Added new machine %s', machine_id
)
875 return await self
._wait
_for
_new
('machine', machine_id
)
877 async def add_relation(self
, relation1
, relation2
):
878 """Add a relation between two applications.
880 :param str relation1: '<application>[:<relation_name>]'
881 :param str relation2: '<application>[:<relation_name>]'
884 app_facade
= client
.ApplicationFacade
.from_connection(self
.connection
)
887 'Adding relation %s <-> %s', relation1
, relation2
)
890 result
= await app_facade
.AddRelation([relation1
, relation2
])
891 except JujuAPIError
as e
:
892 if 'relation already exists' not in e
.message
:
895 'Relation %s <-> %s already exists', relation1
, relation2
)
896 # TODO: if relation already exists we should return the
897 # Relation ModelEntity here
900 def predicate(delta
):
902 for endpoint
in delta
.data
['endpoints']:
903 endpoints
[endpoint
['application-name']] = endpoint
['relation']
904 return endpoints
== result
.endpoints
906 return await self
._wait
_for
_new
('relation', None, predicate
)
908 def add_space(self
, name
, *cidrs
):
909 """Add a new network space.
911 Adds a new space with the given name and associates the given
912 (optional) list of existing subnet CIDRs with it.
914 :param str name: Name of the space
915 :param \*cidrs: Optional list of existing subnet CIDRs
918 raise NotImplementedError()
920 async def add_ssh_key(self
, user
, key
):
921 """Add a public SSH key to this model.
923 :param str user: The username of the user
924 :param str key: The public ssh key
927 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection
)
928 return await key_facade
.AddKeys([key
], user
)
929 add_ssh_keys
= add_ssh_key
931 def add_subnet(self
, cidr_or_id
, space
, *zones
):
932 """Add an existing subnet to this model.
934 :param str cidr_or_id: CIDR or provider ID of the existing subnet
935 :param str space: Network space with which to associate
936 :param str \*zones: Zone(s) in which the subnet resides
939 raise NotImplementedError()
941 def get_backups(self
):
942 """Retrieve metadata for backups in this model.
945 raise NotImplementedError()
947 def block(self
, *commands
):
948 """Add a new block to this model.
950 :param str \*commands: The commands to block. Valid values are
951 'all-changes', 'destroy-model', 'remove-object'
954 raise NotImplementedError()
956 def get_blocks(self
):
957 """List blocks for this model.
960 raise NotImplementedError()
962 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
963 """Return a list of cached OS images.
965 :param str arch: Filter by image architecture
966 :param str kind: Filter by image kind, e.g. 'lxd'
967 :param str series: Filter by image series, e.g. 'xenial'
970 raise NotImplementedError()
972 def create_backup(self
, note
=None, no_download
=False):
973 """Create a backup of this model.
975 :param str note: A note to store with the backup
976 :param bool no_download: Do not download the backup archive
977 :return str: Path to downloaded archive
980 raise NotImplementedError()
982 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
983 """Create or define a storage pool.
985 :param str name: Name to give the storage pool
986 :param str provider_type: Pool provider type
987 :param \*\*pool_config: key/value pool configuration pairs
990 raise NotImplementedError()
993 self
, no_tail
=False, exclude_module
=None, include_module
=None,
994 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
996 """Get log messages for this model.
998 :param bool no_tail: Stop after returning existing log messages
999 :param list exclude_module: Do not show log messages for these logging
1001 :param list include_module: Only show log messages for these logging
1003 :param list include: Only show log messages for these entities
1004 :param str level: Log level to show, valid options are 'TRACE',
1005 'DEBUG', 'INFO', 'WARNING', 'ERROR,
1006 :param int limit: Return this many of the most recent (possibly
1007 filtered) lines are shown
1008 :param int lines: Yield this many of the most recent lines, and keep
1010 :param bool replay: Yield the entire log, and keep yielding
1011 :param list exclude: Do not show log messages for these entities
1014 raise NotImplementedError()
1016 def _get_series(self
, entity_url
, entity
):
1017 # try to get the series from the provided charm URL
1018 if entity_url
.startswith('cs:'):
1019 parts
= entity_url
[3:].split('/')
1021 parts
= entity_url
.split('/')
1022 if parts
[0].startswith('~'):
1025 # series was specified in the URL
1027 # series was not supplied at all, so use the newest
1028 # supported series according to the charm store
1029 ss
= entity
['Meta']['supported-series']
1030 return ss
['SupportedSeries'][0]
1033 self
, entity_url
, application_name
=None, bind
=None, budget
=None,
1034 channel
=None, config
=None, constraints
=None, force
=False,
1035 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
1037 """Deploy a new service or bundle.
1039 :param str entity_url: Charm or bundle url
1040 :param str application_name: Name to give the service
1041 :param dict bind: <charm endpoint>:<network space> pairs
1042 :param dict budget: <budget name>:<limit> pairs
1043 :param str channel: Charm store channel from which to retrieve
1044 the charm or bundle, e.g. 'edge'
1045 :param dict config: Charm configuration dictionary
1046 :param constraints: Service constraints
1047 :type constraints: :class:`juju.Constraints`
1048 :param bool force: Allow charm to be deployed to a machine running
1049 an unsupported series
1050 :param int num_units: Number of units to deploy
1051 :param str plan: Plan under which to deploy charm
1052 :param dict resources: <resource name>:<file path> pairs
1053 :param str series: Series on which to deploy
1054 :param dict storage: Storage constraints TODO how do these look?
1055 :param to: Placement directive as a string. For example:
1057 '23' - place on machine 23
1058 'lxd:7' - place in new lxd container on machine 7
1059 '24/lxd/3' - place in container 3 on machine 24
1061 If None, a new machine is provisioned.
1066 - support local resources
1071 k
: client
.Constraints(**v
)
1072 for k
, v
in storage
.items()
1076 entity_url
.startswith('local:') or
1077 os
.path
.isdir(entity_url
)
1080 entity_id
= entity_url
.replace('local:', '')
1082 entity
= await self
.charmstore
.entity(entity_url
, channel
=channel
)
1083 entity_id
= entity
['Id']
1085 client_facade
= client
.ClientFacade
.from_connection(self
.connection
)
1087 is_bundle
= ((is_local
and
1088 (Path(entity_id
) / 'bundle.yaml').exists()) or
1089 (not is_local
and 'bundle/' in entity_id
))
1092 handler
= BundleHandler(self
)
1093 await handler
.fetch_plan(entity_id
)
1094 await handler
.execute_plan()
1095 extant_apps
= {app
for app
in self
.applications
}
1096 pending_apps
= set(handler
.applications
) - extant_apps
1098 # new apps will usually be in the model by now, but if some
1099 # haven't made it yet we'll need to wait on them to be added
1100 await asyncio
.gather(*[
1101 asyncio
.ensure_future(
1102 self
._wait
_for
_new
('application', app_name
),
1104 for app_name
in pending_apps
1106 return [app
for name
, app
in self
.applications
.items()
1107 if name
in handler
.applications
]
1110 if not application_name
:
1111 application_name
= entity
['Meta']['charm-metadata']['Name']
1113 series
= self
._get
_series
(entity_url
, entity
)
1114 await client_facade
.AddCharm(channel
, entity_id
)
1115 # XXX: we're dropping local resources here, but we don't
1116 # actually support them yet anyway
1117 resources
= await self
._add
_store
_resources
(application_name
,
1121 # We have a local charm dir that needs to be uploaded
1122 charm_dir
= os
.path
.abspath(
1123 os
.path
.expanduser(entity_id
))
1124 series
= series
or get_charm_series(charm_dir
)
1127 "Couldn't determine series for charm at {}. "
1128 "Pass a 'series' kwarg to Model.deploy().".format(
1130 entity_id
= await self
.add_local_charm_dir(charm_dir
, series
)
1131 return await self
._deploy
(
1132 charm_url
=entity_id
,
1133 application
=application_name
,
1135 config
=config
or {},
1136 constraints
=constraints
,
1137 endpoint_bindings
=bind
,
1138 resources
=resources
,
1141 num_units
=num_units
,
1142 placement
=parse_placement(to
)
1145 async def _add_store_resources(self
, application
, entity_url
, entity
=None):
1147 # avoid extra charm store call if one was already made
1148 entity
= await self
.charmstore
.entity(entity_url
)
1151 'description': resource
['Description'],
1152 'fingerprint': resource
['Fingerprint'],
1153 'name': resource
['Name'],
1154 'path': resource
['Path'],
1155 'revision': resource
['Revision'],
1156 'size': resource
['Size'],
1157 'type_': resource
['Type'],
1159 } for resource
in entity
['Meta']['resources']
1165 resources_facade
= client
.ResourcesFacade
.from_connection(
1167 response
= await resources_facade
.AddPendingResources(
1168 tag
.application(application
),
1170 [client
.CharmResource(**resource
) for resource
in resources
])
1171 resource_map
= {resource
['name']: pid
1173 in zip(resources
, response
.pending_ids
)}
1176 async def _deploy(self
, charm_url
, application
, series
, config
,
1177 constraints
, endpoint_bindings
, resources
, storage
,
1178 channel
=None, num_units
=None, placement
=None):
1179 """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
1181 log
.info('Deploying %s', charm_url
)
1183 # stringify all config values for API, and convert to YAML
1184 config
= {k
: str(v
) for k
, v
in config
.items()}
1185 config
= yaml
.dump({application
: config
},
1186 default_flow_style
=False)
1188 app_facade
= client
.ApplicationFacade
.from_connection(
1191 app
= client
.ApplicationDeploy(
1192 charm_url
=charm_url
,
1193 application
=application
,
1197 constraints
=parse_constraints(constraints
),
1198 endpoint_bindings
=endpoint_bindings
,
1199 num_units
=num_units
,
1200 resources
=resources
,
1205 result
= await app_facade
.Deploy([app
])
1206 errors
= [r
.error
.message
for r
in result
.results
if r
.error
]
1208 raise JujuError('\n'.join(errors
))
1209 return await self
._wait
_for
_new
('application', application
)
1211 async def destroy(self
):
1212 """Terminate all machines and resources for this model.
1213 Is already implemented in controller.py.
1215 raise NotImplementedError()
1217 async def destroy_unit(self
, *unit_names
):
1218 """Destroy units by name.
1221 app_facade
= client
.ApplicationFacade
.from_connection(self
.connection
)
1224 'Destroying unit%s %s',
1225 's' if len(unit_names
) == 1 else '',
1226 ' '.join(unit_names
))
1228 return await app_facade
.DestroyUnits(list(unit_names
))
1229 destroy_units
= destroy_unit
1231 def get_backup(self
, archive_id
):
1232 """Download a backup archive file.
1234 :param str archive_id: The id of the archive to download
1235 :return str: Path to the archive file
1238 raise NotImplementedError()
1241 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
1242 """Ensure sufficient controllers exist to provide redundancy.
1244 :param int num_controllers: Number of controllers to make available
1245 :param constraints: Constraints to apply to the controller machines
1246 :type constraints: :class:`juju.Constraints`
1247 :param str series: Series of the controller machines
1248 :param list to: Placement directives for controller machines, e.g.::
1251 'lxc:7' - new lxc container on machine 7
1252 '24/lxc/3' - lxc container 3 or machine 24
1254 If None, a new machine is provisioned.
1257 raise NotImplementedError()
1259 async def get_config(self
):
1260 """Return the configuration settings for this model.
1262 :returns: A ``dict`` mapping keys to `ConfigValue` instances,
1263 which have `source` and `value` attributes.
1265 config_facade
= client
.ModelConfigFacade
.from_connection(
1268 result
= await config_facade
.ModelGet()
1269 config
= result
.config
1270 for key
, value
in config
.items():
1271 config
[key
] = ConfigValue
.from_json(value
)
1274 def get_constraints(self
):
1275 """Return the machine constraints for this model.
1278 raise NotImplementedError()
1280 async def grant(self
, username
, acl
='read'):
1281 """Grant a user access to this model.
1283 :param str username: Username
1284 :param str acl: Access control ('read' or 'write')
1287 controller_conn
= await self
.connection
.controller()
1288 model_facade
= client
.ModelManagerFacade
.from_connection(
1290 user
= tag
.user(username
)
1291 model
= tag
.model(self
.info
.uuid
)
1292 changes
= client
.ModifyModelAccess(acl
, 'grant', model
, user
)
1293 await self
.revoke(username
)
1294 return await model_facade
.ModifyModelAccess([changes
])
1296 def import_ssh_key(self
, identity
):
1297 """Add a public SSH key from a trusted indentity source to this model.
1299 :param str identity: User identity in the form <lp|gh>:<username>
1302 raise NotImplementedError()
1303 import_ssh_keys
= import_ssh_key
1305 async def get_machines(self
):
1306 """Return list of machines in this model.
1309 return list(self
.state
.machines
.keys())
1311 def get_shares(self
):
1312 """Return list of all users with access to this model.
1315 raise NotImplementedError()
1317 def get_spaces(self
):
1318 """Return list of all known spaces, including associated subnets.
1321 raise NotImplementedError()
1323 async def get_ssh_key(self
, raw_ssh
=False):
1324 """Return known SSH keys for this model.
1325 :param bool raw_ssh: if True, returns the raw ssh key,
1326 else it's fingerprint
1329 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection
)
1330 entity
= {'tag': tag
.model(self
.info
.uuid
)}
1331 entities
= client
.Entities([entity
])
1332 return await key_facade
.ListKeys(entities
, raw_ssh
)
1333 get_ssh_keys
= get_ssh_key
1335 def get_storage(self
, filesystem
=False, volume
=False):
1336 """Return details of storage instances.
1338 :param bool filesystem: Include filesystem storage
1339 :param bool volume: Include volume storage
1342 raise NotImplementedError()
1344 def get_storage_pools(self
, names
=None, providers
=None):
1345 """Return list of storage pools.
1347 :param list names: Only include pools with these names
1348 :param list providers: Only include pools for these providers
1351 raise NotImplementedError()
1353 def get_subnets(self
, space
=None, zone
=None):
1354 """Return list of known subnets.
1356 :param str space: Only include subnets in this space
1357 :param str zone: Only include subnets in this zone
1360 raise NotImplementedError()
1362 def remove_blocks(self
):
1363 """Remove all blocks from this model.
1366 raise NotImplementedError()
1368 def remove_backup(self
, backup_id
):
1371 :param str backup_id: The id of the backup to remove
1374 raise NotImplementedError()
1376 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1377 """Remove cached OS images.
1379 :param str arch: Architecture of the images to remove
1380 :param str kind: Image kind to remove, e.g. 'lxd'
1381 :param str series: Image series to remove, e.g. 'xenial'
1384 raise NotImplementedError()
1386 def remove_machine(self
, *machine_ids
):
1387 """Remove a machine from this model.
1389 :param str \*machine_ids: Ids of the machines to remove
1392 raise NotImplementedError()
1393 remove_machines
= remove_machine
1395 async def remove_ssh_key(self
, user
, key
):
1396 """Remove a public SSH key(s) from this model.
1398 :param str key: Full ssh key
1399 :param str user: Juju user to which the key is registered
1402 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection
)
1403 key
= base64
.b64decode(bytes(key
.strip().split()[1].encode('ascii')))
1404 key
= hashlib
.md5(key
).hexdigest()
1405 key
= ':'.join(a
+ b
for a
, b
in zip(key
[::2], key
[1::2]))
1406 await key_facade
.DeleteKeys([key
], user
)
1407 remove_ssh_keys
= remove_ssh_key
1410 self
, bootstrap
=False, constraints
=None, archive
=None,
1411 backup_id
=None, upload_tools
=False):
1412 """Restore a backup archive to a new controller.
1414 :param bool bootstrap: Bootstrap a new state machine
1415 :param constraints: Model constraints
1416 :type constraints: :class:`juju.Constraints`
1417 :param str archive: Path to backup archive to restore
1418 :param str backup_id: Id of backup to restore
1419 :param bool upload_tools: Upload tools if bootstrapping a new machine
1422 raise NotImplementedError()
1424 def retry_provisioning(self
):
1425 """Retry provisioning for failed machines.
1428 raise NotImplementedError()
1430 async def revoke(self
, username
):
1431 """Revoke a user's access to this model.
1433 :param str username: Username to revoke
1436 controller_conn
= await self
.connection
.controller()
1437 model_facade
= client
.ModelManagerFacade
.from_connection(
1439 user
= tag
.user(username
)
1440 model
= tag
.model(self
.info
.uuid
)
1441 changes
= client
.ModifyModelAccess('read', 'revoke', model
, user
)
1442 return await model_facade
.ModifyModelAccess([changes
])
1444 def run(self
, command
, timeout
=None):
1445 """Run command on all machines in this model.
1447 :param str command: The command to run
1448 :param int timeout: Time to wait before command is considered failed
1451 raise NotImplementedError()
1453 async def set_config(self
, config
):
1454 """Set configuration keys on this model.
1456 :param dict config: Mapping of config keys to either string values or
1457 `ConfigValue` instances, as returned by `get_config`.
1459 config_facade
= client
.ModelConfigFacade
.from_connection(
1462 for key
, value
in config
.items():
1463 if isinstance(value
, ConfigValue
):
1464 config
[key
] = value
.value
1465 await config_facade
.ModelSet(config
)
1467 def set_constraints(self
, constraints
):
1468 """Set machine constraints on this model.
1470 :param :class:`juju.Constraints` constraints: Machine constraints
1473 raise NotImplementedError()
1475 def get_action_output(self
, action_uuid
, wait
=-1):
1476 """Get the results of an action by ID.
1478 :param str action_uuid: Id of the action
1479 :param int wait: Time in seconds to wait for action to complete
1482 raise NotImplementedError()
1484 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1485 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1487 :param str uuid_or_prefix: Filter by action uuid or prefix
1488 :param str name: Filter by action name
1491 raise NotImplementedError()
1493 def get_budget(self
, budget_name
):
1494 """Get budget usage info.
1496 :param str budget_name: Name of budget
1499 raise NotImplementedError()
1501 async def get_status(self
, filters
=None, utc
=False):
1502 """Return the status of the model.
1504 :param str filters: Optional list of applications, units, or machines
1505 to include, which can use wildcards ('*').
1506 :param bool utc: Display time as UTC in RFC3339 format
1509 client_facade
= client
.ClientFacade
.from_connection(self
.connection
)
1510 return await client_facade
.FullStatus(filters
)
1513 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1514 source
=None, stream
=None, version
=None):
1515 """Copy Juju tools into this model.
1517 :param bool all_: Copy all versions, not just the latest
1518 :param str destination: Path to local destination directory
1519 :param bool dry_run: Don't do the actual copy
1520 :param bool public: Tools are for a public cloud, so generate mirrors
1522 :param str source: Path to local source directory
1523 :param str stream: Simplestreams stream for which to sync metadata
1524 :param str version: Copy a specific major.minor version
1527 raise NotImplementedError()
1529 def unblock(self
, *commands
):
1530 """Unblock an operation that would alter this model.
1532 :param str \*commands: The commands to unblock. Valid values are
1533 'all-changes', 'destroy-model', 'remove-object'
1536 raise NotImplementedError()
1538 def unset_config(self
, *keys
):
1539 """Unset configuration on this model.
1541 :param str \*keys: The keys to unset
1544 raise NotImplementedError()
1546 def upgrade_gui(self
):
1547 """Upgrade the Juju GUI for this model.
1550 raise NotImplementedError()
1553 self
, dry_run
=False, reset_previous_upgrade
=False,
1554 upload_tools
=False, version
=None):
1555 """Upgrade Juju on all machines in a model.
1557 :param bool dry_run: Don't do the actual upgrade
1558 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1560 :param bool upload_tools: Upload local version of tools
1561 :param str version: Upgrade to a specific version
1564 raise NotImplementedError()
1566 def upload_backup(self
, archive_path
):
1567 """Store a backup archive remotely in Juju.
1569 :param str archive_path: Path to local archive
1572 raise NotImplementedError()
1575 def charmstore(self
):
1576 return self
._charmstore
1578 async def get_metrics(self
, *tags
):
1579 """Retrieve metrics.
1581 :param str \*tags: Tags of entities from which to retrieve metrics.
1582 No tags retrieves the metrics of all units in the model.
1583 :return: Dictionary of unit_name:metrics
1586 log
.debug("Retrieving metrics for %s",
1587 ', '.join(tags
) if tags
else "all units")
1589 metrics_facade
= client
.MetricsDebugFacade
.from_connection(
1592 entities
= [client
.Entity(tag
) for tag
in tags
]
1593 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1595 metrics
= collections
.defaultdict(list)
1597 for entity_metrics
in metrics_result
.results
:
1598 error
= entity_metrics
.error
1600 if "is not a valid tag" in error
:
1601 raise ValueError(error
.message
)
1603 raise Exception(error
.message
)
1605 for metric
in entity_metrics
.metrics
:
1606 metrics
[metric
.unit
].append(vars(metric
))
1611 def get_charm_series(path
):
1612 """Inspects the charm directory at ``path`` and returns a default
1613 series from its metadata.yaml (the first item in the 'series' list).
1615 Returns None if no series can be determined.
1618 md
= Path(path
) / "metadata.yaml"
1621 data
= yaml
.load(md
.open())
1622 series
= data
.get('series')
1623 return series
[0] if series
else None
1626 class BundleHandler(object):
1628 Handle bundles by using the API to translate bundle YAML into a plan of
1629 steps and then dispatching each of those using the API.
1631 def __init__(self
, model
):
1633 self
.charmstore
= model
.charmstore
1635 self
.references
= {}
1636 self
._units
_by
_app
= {}
1637 for unit_name
, unit
in model
.units
.items():
1638 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1639 app_units
.append(unit_name
)
1640 self
.client_facade
= client
.ClientFacade
.from_connection(
1642 self
.app_facade
= client
.ApplicationFacade
.from_connection(
1644 self
.ann_facade
= client
.AnnotationsFacade
.from_connection(
1647 async def _handle_local_charms(self
, bundle
):
1648 """Search for references to local charms (i.e. filesystem paths)
1649 in the bundle. Upload the local charms to the model, and replace
1650 the filesystem paths with appropriate 'local:' paths in the bundle.
1652 Return the modified bundle.
1654 :param dict bundle: Bundle dictionary
1655 :return: Modified bundle dictionary
1660 default_series
= bundle
.get('series')
1661 apps_dict
= bundle
.get('applications', bundle
.get('services', {}))
1662 for app_name
in self
.applications
:
1663 app_dict
= apps_dict
[app_name
]
1664 charm_dir
= os
.path
.abspath(os
.path
.expanduser(app_dict
['charm']))
1665 if not os
.path
.isdir(charm_dir
):
1668 app_dict
.get('series') or
1670 get_charm_series(charm_dir
)
1674 "Couldn't determine series for charm at {}. "
1675 "Add a 'series' key to the bundle.".format(charm_dir
))
1677 # Keep track of what we need to update. We keep a list of apps
1678 # that need to be updated, and a corresponding list of args
1679 # needed to update those apps.
1680 apps
.append(app_name
)
1681 args
.append((charm_dir
, series
))
1684 # If we have apps to update, spawn all the coroutines concurrently
1685 # and wait for them to finish.
1686 charm_urls
= await asyncio
.gather(*[
1687 self
.model
.add_local_charm_dir(*params
)
1689 ], loop
=self
.model
.loop
)
1690 # Update the 'charm:' entry for each app with the new 'local:' url.
1691 for app_name
, charm_url
in zip(apps
, charm_urls
):
1692 apps_dict
[app_name
]['charm'] = charm_url
1696 async def fetch_plan(self
, entity_id
):
1697 is_local
= not entity_id
.startswith('cs:') and os
.path
.isdir(entity_id
)
1699 bundle_yaml
= (Path(entity_id
) / "bundle.yaml").read_text()
1701 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1702 filename
='bundle.yaml',
1704 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1705 self
.bundle
= await self
._handle
_local
_charms
(self
.bundle
)
1707 self
.plan
= await self
.client_facade
.GetBundleChanges(
1708 yaml
.dump(self
.bundle
))
1710 async def execute_plan(self
):
1711 for step
in self
.plan
.changes
:
1712 method
= getattr(self
, step
.method
)
1713 result
= await method(*step
.args
)
1714 self
.references
[step
.id_
] = result
1717 def applications(self
):
1718 apps_dict
= self
.bundle
.get('applications',
1719 self
.bundle
.get('services', {}))
1720 return list(apps_dict
.keys())
1722 def resolve(self
, reference
):
1723 if reference
and reference
.startswith('$'):
1724 reference
= self
.references
[reference
[1:]]
1727 async def addCharm(self
, charm
, series
):
1729 :param charm string:
1730 Charm holds the URL of the charm to be added.
1732 :param series string:
1733 Series holds the series of the charm to be added
1734 if the charm default is not sufficient.
1736 # We don't add local charms because they've already been added
1737 # by self._handle_local_charms
1738 if charm
.startswith('local:'):
1741 entity_id
= await self
.charmstore
.entityId(charm
)
1742 log
.debug('Adding %s', entity_id
)
1743 await self
.client_facade
.AddCharm(None, entity_id
)
1746 async def addMachines(self
, params
=None):
1749 Dictionary specifying the machine to add. All keys are optional.
1752 series: string specifying the machine OS series.
1754 constraints: string holding machine constraints, if any. We'll
1755 parse this into the json friendly dict that the juju api
1758 container_type: string holding the type of the container (for
1759 instance ""lxd" or kvm"). It is not specified for top level
1762 parent_id: string holding a placeholder pointing to another
1763 machine change or to a unit change. This value is only
1764 specified in the case this machine is a container, in
1765 which case also ContainerType is set.
1768 params
= params
or {}
1771 params
= {normalize_key(k
): params
[k
] for k
in params
.keys()}
1773 # Fix up values, as necessary.
1774 if 'parent_id' in params
:
1775 params
['parent_id'] = self
.resolve(params
['parent_id'])
1777 params
['constraints'] = parse_constraints(
1778 params
.get('constraints'))
1779 params
['jobs'] = params
.get('jobs', ['JobHostUnits'])
1781 if params
.get('container_type') == 'lxc':
1782 log
.warning('Juju 2.0 does not support lxc containers. '
1783 'Converting containers to lxd.')
1784 params
['container_type'] = 'lxd'
1786 # Submit the request.
1787 params
= client
.AddMachineParams(**params
)
1788 results
= await self
.client_facade
.AddMachines([params
])
1789 error
= results
.machines
[0].error
1791 raise ValueError("Error adding machine: %s" % error
.message
)
1792 machine
= results
.machines
[0].machine
1793 log
.debug('Added new machine %s', machine
)
1796 async def addRelation(self
, endpoint1
, endpoint2
):
1798 :param endpoint1 string:
1799 :param endpoint2 string:
1800 Endpoint1 and Endpoint2 hold relation endpoints in the
1801 "application:interface" form, where the application is always a
1802 placeholder pointing to an application change, and the interface is
1803 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1805 endpoints
= [endpoint1
, endpoint2
]
1806 # resolve indirect references
1807 for i
in range(len(endpoints
)):
1808 parts
= endpoints
[i
].split(':')
1809 parts
[0] = self
.resolve(parts
[0])
1810 endpoints
[i
] = ':'.join(parts
)
1812 log
.info('Relating %s <-> %s', *endpoints
)
1813 return await self
.model
.add_relation(*endpoints
)
1815 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1816 storage
, endpoint_bindings
, resources
):
1818 :param charm string:
1819 Charm holds the URL of the charm to be used to deploy this
1822 :param series string:
1823 Series holds the series of the application to be deployed
1824 if the charm default is not sufficient.
1826 :param application string:
1827 Application holds the application name.
1829 :param options map[string]interface{}:
1830 Options holds application options.
1832 :param constraints string:
1833 Constraints holds the optional application constraints.
1835 :param storage map[string]string:
1836 Storage holds the optional storage constraints.
1838 :param endpoint_bindings map[string]string:
1839 EndpointBindings holds the optional endpoint bindings
1841 :param resources map[string]int:
1842 Resources identifies the revision to use for each resource
1843 of the application's charm.
1845 # resolve indirect references
1846 charm
= self
.resolve(charm
)
1847 # the bundle plan doesn't actually do anything with resources, even
1848 # though it ostensibly gives us something (None) for that param
1849 if not charm
.startswith('local:'):
1850 resources
= await self
.model
._add
_store
_resources
(application
,
1852 await self
.model
._deploy
(
1854 application
=application
,
1857 constraints
=constraints
,
1858 endpoint_bindings
=endpoint_bindings
,
1859 resources
=resources
,
1864 async def addUnit(self
, application
, to
):
1866 :param application string:
1867 Application holds the application placeholder name for which a unit
1871 To holds the optional location where to add the unit, as a
1872 placeholder pointing to another unit change or to a machine change.
1874 application
= self
.resolve(application
)
1875 placement
= self
.resolve(to
)
1876 if self
._units
_by
_app
.get(application
):
1877 # enough units for this application already exist;
1878 # claim one, and carry on
1879 # NB: this should probably honor placement, but the juju client
1880 # doesn't, so we're not bothering, either
1881 unit_name
= self
._units
_by
_app
[application
].pop()
1882 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1883 return self
.model
.units
[unit_name
]
1885 log
.debug('Adding new unit for %s%s', application
,
1886 ' to %s' % placement
if placement
else '')
1887 return await self
.model
.applications
[application
].add_unit(
1892 async def expose(self
, application
):
1894 :param application string:
1895 Application holds the placeholder name of the application that must
1898 application
= self
.resolve(application
)
1899 log
.info('Exposing %s', application
)
1900 return await self
.model
.applications
[application
].expose()
1902 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1905 Id is the placeholder for the application or machine change
1906 corresponding to the entity to be annotated.
1908 :param entity_type EntityType:
1909 EntityType holds the type of the entity, "application" or
1912 :param annotations map[string]string:
1913 Annotations holds the annotations as key/value pairs.
1915 entity_id
= self
.resolve(id_
)
1917 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
1919 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
1920 return await entity
.set_annotations(annotations
)
1923 class CharmStore(object):
1925 Async wrapper around theblues.charmstore.CharmStore
1927 def __init__(self
, loop
):
1929 self
._cs
= theblues
.charmstore
.CharmStore(timeout
=5)
1931 def __getattr__(self
, name
):
1933 Wrap method calls in coroutines that use run_in_executor to make them
1936 attr
= getattr(self
._cs
, name
)
1937 if not callable(attr
):
1938 wrapper
= partial(getattr, self
._cs
, name
)
1939 setattr(self
, name
, wrapper
)
1941 async def coro(*args
, **kwargs
):
1942 method
= partial(attr
, *args
, **kwargs
)
1943 for attempt
in range(1, 4):
1945 return await self
.loop
.run_in_executor(None, method
)
1946 except theblues
.errors
.ServerError
:
1949 await asyncio
.sleep(1, loop
=self
.loop
)
1950 setattr(self
, name
, coro
)
1955 class CharmArchiveGenerator(object):
1957 Create a Zip archive of a local charm directory for upload to a controller.
1959 This is used automatically by
1960 `Model.add_local_charm_dir <#juju.model.Model.add_local_charm_dir>`_.
1962 def __init__(self
, path
):
1963 self
.path
= os
.path
.abspath(os
.path
.expanduser(path
))
1965 def make_archive(self
, path
):
1966 """Create archive of directory and write to ``path``.
1968 :param path: Path to archive
1972 * build/\* - This is used for packing the charm itself and any
1974 * \*/.\* - Hidden files are all ignored for now. This will most
1975 likely be changed into a specific ignore list
1979 zf
= zipfile
.ZipFile(path
, 'w', zipfile
.ZIP_DEFLATED
)
1980 for dirpath
, dirnames
, filenames
in os
.walk(self
.path
):
1981 relative_path
= dirpath
[len(self
.path
) + 1:]
1982 if relative_path
and not self
._ignore
(relative_path
):
1983 zf
.write(dirpath
, relative_path
)
1984 for name
in filenames
:
1985 archive_name
= os
.path
.join(relative_path
, name
)
1986 if not self
._ignore
(archive_name
):
1987 real_path
= os
.path
.join(dirpath
, name
)
1988 self
._check
_type
(real_path
)
1989 if os
.path
.islink(real_path
):
1990 self
._check
_link
(real_path
)
1991 self
._write
_symlink
(
1992 zf
, os
.readlink(real_path
), archive_name
)
1994 zf
.write(real_path
, archive_name
)
1998 def _check_type(self
, path
):
2002 if stat
.S_ISDIR(s
.st_mode
) or stat
.S_ISREG(s
.st_mode
):
2004 raise ValueError("Invalid Charm at % %s" % (
2005 path
, "Invalid file type for a charm"))
2007 def _check_link(self
, path
):
2008 link_path
= os
.readlink(path
)
2009 if link_path
[0] == "/":
2011 "Invalid Charm at %s: %s" % (
2012 path
, "Absolute links are invalid"))
2013 path_dir
= os
.path
.dirname(path
)
2014 link_path
= os
.path
.join(path_dir
, link_path
)
2015 if not link_path
.startswith(os
.path
.abspath(self
.path
)):
2017 "Invalid charm at %s %s" % (
2018 path
, "Only internal symlinks are allowed"))
2020 def _write_symlink(self
, zf
, link_target
, link_path
):
2021 """Package symlinks with appropriate zipfile metadata."""
2022 info
= zipfile
.ZipInfo()
2023 info
.filename
= link_path
2024 info
.create_system
= 3
2025 # Magic code for symlinks / py2/3 compat
2026 # 27166663808 = (stat.S_IFLNK | 0755) << 16
2027 info
.external_attr
= 2716663808
2028 zf
.writestr(info
, link_target
)
2030 def _ignore(self
, path
):
2031 if path
== "build" or path
.startswith("build/"):
2033 if path
.startswith('.'):