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 .constraints
import parse
as parse_constraints
, normalize_key
26 from .delta
import get_entity_delta
27 from .delta
import get_entity_class
28 from .exceptions
import DeadEntityException
29 from .errors
import JujuError
, JujuAPIError
30 from .placement
import parse
as parse_placement
32 log
= logging
.getLogger(__name__
)
35 class _Observer(object):
36 """Wrapper around an observer callable.
38 This wrapper allows filter criteria to be associated with the
39 callable so that it's only called for changes that meet the criteria.
42 def __init__(self
, callable_
, entity_type
, action
, entity_id
, predicate
):
43 self
.callable_
= callable_
44 self
.entity_type
= entity_type
46 self
.entity_id
= entity_id
47 self
.predicate
= predicate
49 self
.entity_id
= str(self
.entity_id
)
50 if not self
.entity_id
.startswith('^'):
51 self
.entity_id
= '^' + self
.entity_id
52 if not self
.entity_id
.endswith('$'):
55 async def __call__(self
, delta
, old
, new
, model
):
56 await self
.callable_(delta
, old
, new
, model
)
58 def cares_about(self
, delta
):
59 """Return True if this observer "cares about" (i.e. wants to be
60 called) for a this delta.
63 if (self
.entity_id
and delta
.get_id() and
64 not re
.match(self
.entity_id
, str(delta
.get_id()))):
67 if self
.entity_type
and self
.entity_type
!= delta
.entity
:
70 if self
.action
and self
.action
!= delta
.type:
73 if self
.predicate
and not self
.predicate(delta
):
79 class ModelObserver(object):
81 Base class for creating observers that react to changes in a model.
83 async def __call__(self
, delta
, old
, new
, model
):
84 handler_name
= 'on_{}_{}'.format(delta
.entity
, delta
.type)
85 method
= getattr(self
, handler_name
, self
.on_change
)
86 await method(delta
, old
, new
, model
)
88 async def on_change(self
, delta
, old
, new
, model
):
89 """Generic model-change handler.
91 This should be overridden in a subclass.
93 :param delta: :class:`juju.client.overrides.Delta`
94 :param old: :class:`juju.model.ModelEntity`
95 :param new: :class:`juju.model.ModelEntity`
96 :param model: :class:`juju.model.Model`
102 class ModelState(object):
103 """Holds the state of the model, including the delta history of all
104 entities in the model.
107 def __init__(self
, model
):
111 def _live_entity_map(self
, entity_type
):
112 """Return an id:Entity map of all the living entities of
113 type ``entity_type``.
117 entity_id
: self
.get_entity(entity_type
, entity_id
)
118 for entity_id
, history
in self
.state
.get(entity_type
, {}).items()
119 if history
[-1] is not None
123 def applications(self
):
124 """Return a map of application-name:Application for all applications
125 currently in the model.
128 return self
._live
_entity
_map
('application')
132 """Return a map of machine-id:Machine for all machines currently in
136 return self
._live
_entity
_map
('machine')
140 """Return a map of unit-id:Unit for all units currently in
144 return self
._live
_entity
_map
('unit')
146 def entity_history(self
, entity_type
, entity_id
):
147 """Return the history deque for an entity.
150 return self
.state
[entity_type
][entity_id
]
152 def entity_data(self
, entity_type
, entity_id
, history_index
):
153 """Return the data dict for an entity at a specific index of its
157 return self
.entity_history(entity_type
, entity_id
)[history_index
]
159 def apply_delta(self
, delta
):
160 """Apply delta to our state and return a copy of the
161 affected object as it was before and after the update, e.g.:
163 old_obj, new_obj = self.apply_delta(delta)
165 old_obj may be None if the delta is for the creation of a new object,
166 e.g. a new application or unit is deployed.
168 new_obj will never be None, but may be dead (new_obj.dead == True)
169 if the object was deleted as a result of the delta being applied.
174 .setdefault(delta
.entity
, {})
175 .setdefault(delta
.get_id(), collections
.deque())
178 history
.append(delta
.data
)
179 if delta
.type == 'remove':
182 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
183 return entity
.previous(), entity
186 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
187 """Return an object instance for the given entity_type and id.
189 By default the object state matches the most recent state from
190 Juju. To get an instance of the object in an older state, pass
191 history_index, an index into the history deque for the entity.
195 if history_index
< 0 and history_index
!= -1:
196 history_index
+= len(self
.entity_history(entity_type
, entity_id
))
197 if history_index
< 0:
201 self
.entity_data(entity_type
, entity_id
, history_index
)
205 entity_class
= get_entity_class(entity_type
)
207 entity_id
, self
.model
, history_index
=history_index
,
211 class ModelEntity(object):
212 """An object in the Model tree"""
214 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
215 """Initialize a new entity
217 :param entity_id str: The unique id of the object in the model
218 :param model: The model instance in whose object tree this
220 :history_index int: The index of this object's state in the model's
221 history deque for this entity
222 :connected bool: Flag indicating whether this object gets live updates
226 self
.entity_id
= entity_id
228 self
._history
_index
= history_index
229 self
.connected
= connected
230 self
.connection
= model
.connection
233 return '<{} entity_id="{}">'.format(type(self
).__name
__,
236 def __getattr__(self
, name
):
237 """Fetch object attributes from the underlying data dict held in the
242 return self
.safe_data
[name
]
244 name
= name
.replace('_', '-')
245 if name
in self
.safe_data
:
246 return self
.safe_data
[name
]
251 return bool(self
.data
)
253 def on_change(self
, callable_
):
254 """Add a change observer to this entity.
257 self
.model
.add_observer(
258 callable_
, self
.entity_type
, 'change', self
.entity_id
)
260 def on_remove(self
, callable_
):
261 """Add a remove observer to this entity.
264 self
.model
.add_observer(
265 callable_
, self
.entity_type
, 'remove', self
.entity_id
)
268 def entity_type(self
):
269 """A string identifying the entity type of this object, e.g.
270 'application' or 'unit', etc.
273 return self
.__class
__.__name
__.lower()
277 """Return True if this object represents the current state of the
278 entity in the underlying model.
280 This will be True except when the object represents an entity at a
281 non-latest state in history, e.g. if the object was obtained by calling
282 .previous() on another object.
285 return self
._history
_index
== -1
289 """Returns True if this entity no longer exists in the underlying
295 self
.model
.state
.entity_data(
296 self
.entity_type
, self
.entity_id
, -1) is None
301 """Returns True if this entity still exists in the underlying
309 """The data dictionary for this entity.
312 return self
.model
.state
.entity_data(
313 self
.entity_type
, self
.entity_id
, self
._history
_index
)
317 """The data dictionary for this entity.
319 If this `ModelEntity` points to the dead state, it will
320 raise `DeadEntityException`.
323 if self
.data
is None:
324 raise DeadEntityException(
325 "Entity {}:{} is dead - its attributes can no longer be "
326 "accessed. Use the .previous() method on this object to get "
327 "a copy of the object at its previous state.".format(
328 self
.entity_type
, self
.entity_id
))
332 """Return a copy of this object as was at its previous state in
335 Returns None if this object is new (and therefore has no history).
337 The returned object is always "disconnected", i.e. does not receive
341 return self
.model
.state
.get_entity(
342 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
346 """Return a copy of this object at its next state in
349 Returns None if this object is already the latest.
351 The returned object is "disconnected", i.e. does not receive
352 live updates, unless it is current (latest).
355 if self
._history
_index
== -1:
358 new_index
= self
._history
_index
+ 1
360 new_index
== len(self
.model
.state
.entity_history(
361 self
.entity_type
, self
.entity_id
)) - 1
363 return self
.model
.state
.get_entity(
364 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
368 """Return a copy of this object at its current state in the model.
370 Returns self if this object is already the latest.
372 The returned object is always "connected", i.e. receives
373 live updates from the model.
376 if self
._history
_index
== -1:
379 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
384 The main API for interacting with a Juju model.
386 def __init__(self
, loop
=None,
387 max_frame_size
=connection
.Connection
.DEFAULT_FRAME_SIZE
):
388 """Instantiate a new connected Model.
390 :param loop: an asyncio event loop
391 :param max_frame_size: See
392 `juju.client.connection.Connection.MAX_FRAME_SIZE`
395 self
.loop
= loop
or asyncio
.get_event_loop()
396 self
.max_frame_size
= max_frame_size
397 self
.connection
= None
398 self
.observers
= weakref
.WeakValueDictionary()
399 self
.state
= ModelState(self
)
401 self
._watch
_stopping
= asyncio
.Event(loop
=self
.loop
)
402 self
._watch
_stopped
= asyncio
.Event(loop
=self
.loop
)
403 self
._watch
_received
= asyncio
.Event(loop
=self
.loop
)
404 self
._charmstore
= CharmStore(self
.loop
)
406 async def __aenter__(self
):
407 await self
.connect_current()
410 async def __aexit__(self
, exc_type
, exc
, tb
):
411 await self
.disconnect()
413 if exc_type
is not None:
416 async def connect(self
, *args
, **kw
):
417 """Connect to an arbitrary Juju model.
419 args and kw are passed through to Connection.connect()
423 kw
['loop'] = self
.loop
424 if 'max_frame_size' not in kw
:
425 kw
['max_frame_size'] = self
.max_frame_size
426 self
.connection
= await connection
.Connection
.connect(*args
, **kw
)
427 await self
._after
_connect
()
429 async def connect_current(self
):
430 """Connect to the current Juju model.
433 self
.connection
= await connection
.Connection
.connect_current(
434 self
.loop
, max_frame_size
=self
.max_frame_size
)
435 await self
._after
_connect
()
437 async def connect_model(self
, model_name
):
438 """Connect to a specific Juju model by name.
440 :param model_name: Format [controller:][user/]model
443 self
.connection
= await connection
.Connection
.connect_model(
444 model_name
, self
.loop
, self
.max_frame_size
)
445 await self
._after
_connect
()
447 async def _after_connect(self
):
448 """Run initialization steps after connecting to websocket.
452 await self
._watch
_received
.wait()
453 await self
.get_info()
455 async def disconnect(self
):
456 """Shut down the watcher task and close websockets.
459 if self
.connection
and self
.connection
.is_open
:
460 log
.debug('Stopping watcher task')
461 self
._watch
_stopping
.set()
462 await self
._watch
_stopped
.wait()
463 log
.debug('Closing model connection')
464 await self
.connection
.close()
465 self
.connection
= None
467 async def add_local_charm_dir(self
, charm_dir
, series
):
468 """Upload a local charm to the model.
470 This will automatically generate an archive from
473 :param charm_dir: Path to the charm directory
474 :param series: Charm series
477 fh
= tempfile
.NamedTemporaryFile()
478 CharmArchiveGenerator(charm_dir
).make_archive(fh
.name
)
481 self
.add_local_charm
, fh
, series
, os
.stat(fh
.name
).st_size
)
482 charm_url
= await self
.loop
.run_in_executor(None, func
)
484 log
.debug('Uploaded local charm: %s -> %s', charm_dir
, charm_url
)
487 def add_local_charm(self
, charm_file
, series
, size
=None):
488 """Upload a local charm archive to the model.
490 Returns the 'local:...' url that should be used to deploy the charm.
492 :param charm_file: Path to charm zip archive
493 :param series: Charm series
494 :param size: Size of the archive, in bytes
495 :return str: 'local:...' url for deploying the charm
496 :raises: :class:`JujuError` if the upload fails
498 Uses an https endpoint at the same host:port as the wss.
499 Supports large file uploads.
503 This method will block. Consider using :meth:`add_local_charm_dir`
507 conn
, headers
, path_prefix
= self
.connection
.https_connection()
508 path
= "%s/charms?series=%s" % (path_prefix
, series
)
509 headers
['Content-Type'] = 'application/zip'
511 headers
['Content-Length'] = size
512 conn
.request("POST", path
, charm_file
, headers
)
513 response
= conn
.getresponse()
514 result
= response
.read().decode()
515 if not response
.status
== 200:
516 raise JujuError(result
)
517 result
= json
.loads(result
)
518 return result
['charm-url']
520 def all_units_idle(self
):
521 """Return True if all units are idle.
524 for unit
in self
.units
.values():
525 unit_status
= unit
.data
['agent-status']['current']
526 if unit_status
!= 'idle':
530 async def reset(self
, force
=False):
531 """Reset the model to a clean state.
533 :param bool force: Force-terminate machines.
535 This returns only after the model has reached a clean state. "Clean"
536 means no applications or machines exist in the model.
539 log
.debug('Resetting model')
540 for app
in self
.applications
.values():
542 for machine
in self
.machines
.values():
543 await machine
.destroy(force
=force
)
544 await self
.block_until(
545 lambda: len(self
.machines
) == 0
548 async def block_until(self
, *conditions
, timeout
=None, wait_period
=0.5):
549 """Return only after all conditions are true.
553 while not all(c() for c
in conditions
):
554 if not (self
.connection
and self
.connection
.is_open
):
555 raise websockets
.ConnectionClosed(1006, 'no reason')
556 await asyncio
.sleep(wait_period
, loop
=self
.loop
)
557 await asyncio
.wait_for(_block(), timeout
, loop
=self
.loop
)
560 def applications(self
):
561 """Return a map of application-name:Application for all applications
562 currently in the model.
565 return self
.state
.applications
569 """Return a map of machine-id:Machine for all machines currently in
573 return self
.state
.machines
577 """Return a map of unit-id:Unit for all units currently in
581 return self
.state
.units
583 async def get_info(self
):
584 """Return a client.ModelInfo object for this Model.
586 Retrieves latest info for this Model from the api server. The
587 return value is cached on the Model.info attribute so that the
588 valued may be accessed again without another api call, if
591 This method is called automatically when the Model is connected,
592 resulting in Model.info being initialized without requiring an
593 explicit call to this method.
596 facade
= client
.ClientFacade
.from_connection(self
.connection
)
598 self
.info
= await facade
.ModelInfo()
599 log
.debug('Got ModelInfo: %s', vars(self
.info
))
604 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
606 """Register an "on-model-change" callback
608 Once the model is connected, ``callable_``
609 will be called each time the model changes. ``callable_`` should
610 be Awaitable and accept the following positional arguments:
612 delta - An instance of :class:`juju.delta.EntityDelta`
613 containing the raw delta data recv'd from the Juju
616 old_obj - If the delta modifies an existing object in the model,
617 old_obj will be a copy of that object, as it was before the
618 delta was applied. Will be None if the delta creates a new
621 new_obj - A copy of the new or updated object, after the delta
622 is applied. Will be None if the delta removes an entity
625 model - The :class:`Model` itself.
627 Events for which ``callable_`` is called can be specified by passing
628 entity_type, action, and/or entitiy_id filter criteria, e.g.::
632 entity_type='application', action='add', entity_id='ubuntu')
634 For more complex filtering conditions, pass a predicate function. It
635 will be called with a delta as its only argument. If the predicate
636 function returns True, the ``callable_`` will be called.
639 observer
= _Observer(
640 callable_
, entity_type
, action
, entity_id
, predicate
)
641 self
.observers
[observer
] = callable_
644 """Start an asynchronous watch against this model.
646 See :meth:`add_observer` to register an onchange callback.
649 async def _all_watcher():
651 allwatcher
= client
.AllWatcherFacade
.from_connection(
653 while not self
._watch
_stopping
.is_set():
655 results
= await utils
.run_with_interrupt(
657 self
._watch
_stopping
,
659 except JujuAPIError
as e
:
660 if 'watcher was stopped' not in str(e
):
662 if self
._watch
_stopping
.is_set():
663 # this shouldn't ever actually happen, because
664 # the event should trigger before the controller
665 # has a chance to tell us the watcher is stopped
666 # but handle it gracefully, just in case
668 # controller stopped our watcher for some reason
669 # but we're not actually stopping, so just restart it
671 'Watcher: watcher stopped, restarting')
674 except websockets
.ConnectionClosed
:
675 monitor
= self
.connection
.monitor
676 if monitor
.status
== monitor
.ERROR
:
677 # closed unexpectedly, try to reopen
679 'Watcher: connection closed, reopening')
680 await self
.connection
.reconnect()
684 # closed on request, go ahead and shutdown
686 if self
._watch
_stopping
.is_set():
687 await allwatcher
.Stop()
689 for delta
in results
.deltas
:
690 delta
= get_entity_delta(delta
)
691 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
692 await self
._notify
_observers
(delta
, old_obj
, new_obj
)
693 self
._watch
_received
.set()
694 except CancelledError
:
697 log
.exception('Error in watcher')
700 self
._watch
_stopped
.set()
702 log
.debug('Starting watcher task')
703 self
._watch
_received
.clear()
704 self
._watch
_stopping
.clear()
705 self
._watch
_stopped
.clear()
706 self
.loop
.create_task(_all_watcher())
708 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
709 """Call observing callbacks, notifying them of a change in model state
711 :param delta: The raw change from the watcher
712 (:class:`juju.client.overrides.Delta`)
713 :param old_obj: The object in the model that this delta updates.
715 :param new_obj: The object in the model that is created or updated
716 by applying this delta.
719 if new_obj
and not old_obj
:
723 'Model changed: %s %s %s',
724 delta
.entity
, delta
.type, delta
.get_id())
726 for o
in self
.observers
:
727 if o
.cares_about(delta
):
728 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
),
731 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
733 Block the calling routine until a given action has happened to the
736 :param entity_type: The entity's type.
737 :param entity_id: The entity's id.
738 :param action: the type of action (e.g., 'add', 'change', or 'remove')
739 :param predicate: optional callable that must take as an
740 argument a delta, and must return a boolean, indicating
741 whether the delta contains the specific action we're looking
742 for. For example, you might check to see whether a 'change'
743 has a 'completed' status. See the _Observer class for details.
746 q
= asyncio
.Queue(loop
=self
.loop
)
748 async def callback(delta
, old
, new
, model
):
749 await q
.put(delta
.get_id())
751 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
752 entity_id
= await q
.get()
753 # object might not be in the entity_map if we were waiting for a
755 return self
.state
._live
_entity
_map
(entity_type
).get(entity_id
)
757 async def _wait_for_new(self
, entity_type
, entity_id
=None, predicate
=None):
758 """Wait for a new object to appear in the Model and return it.
760 Waits for an object of type ``entity_type`` with id ``entity_id``.
761 If ``entity_id`` is ``None``, it will wait for the first new entity
764 This coroutine blocks until the new object appears in the model.
767 # if the entity is already in the model, just return it
768 if entity_id
in self
.state
._live
_entity
_map
(entity_type
):
769 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
770 # if we know the entity_id, we can trigger on any action that puts
771 # the enitty into the model; otherwise, we have to watch for the
772 # next "add" action on that entity_type
773 action
= 'add' if entity_id
is None else None
774 return await self
._wait
(entity_type
, entity_id
, action
, predicate
)
776 async def wait_for_action(self
, action_id
):
777 """Given an action, wait for it to complete."""
779 if action_id
.startswith("action-"):
780 # if we've been passed action.tag, transform it into the
781 # id that the api deltas will use.
782 action_id
= action_id
[7:]
784 def predicate(delta
):
785 return delta
.data
['status'] in ('completed', 'failed')
787 return await self
._wait
('action', action_id
, 'change', predicate
)
789 async def add_machine(
790 self
, spec
=None, constraints
=None, disks
=None, series
=None):
791 """Start a new, empty machine and optionally a container, or add a
792 container to a machine.
794 :param str spec: Machine specification
797 (None) - starts a new machine
798 'lxd' - starts a new machine with one lxd container
799 'lxd:4' - starts a new lxd container on machine 4
800 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
801 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
802 'maas2.name' - acquire machine maas2.name on MAAS
804 :param dict constraints: Machine constraints, which can contain the
825 :param list disks: List of disk constraint dictionaries, which can
826 contain the following keys::
840 :param str series: Series, e.g. 'xenial'
842 Supported container types are: lxd, kvm
844 When deploying a container to an existing machine, constraints cannot
848 params
= client
.AddMachineParams()
849 params
.jobs
= ['JobHostUnits']
852 placement
= parse_placement(spec
)
854 params
.placement
= placement
[0]
857 params
.constraints
= client
.Value
.from_json(constraints
)
861 client
.Constraints
.from_json(o
) for o
in disks
]
864 params
.series
= series
866 # Submit the request.
867 client_facade
= client
.ClientFacade
.from_connection(self
.connection
)
868 results
= await client_facade
.AddMachines([params
])
869 error
= results
.machines
[0].error
871 raise ValueError("Error adding machine: %s" % error
.message
)
872 machine_id
= results
.machines
[0].machine
873 log
.debug('Added new machine %s', machine_id
)
874 return await self
._wait
_for
_new
('machine', machine_id
)
876 async def add_relation(self
, relation1
, relation2
):
877 """Add a relation between two applications.
879 :param str relation1: '<application>[:<relation_name>]'
880 :param str relation2: '<application>[:<relation_name>]'
883 app_facade
= client
.ApplicationFacade
.from_connection(self
.connection
)
886 'Adding relation %s <-> %s', relation1
, relation2
)
889 result
= await app_facade
.AddRelation([relation1
, relation2
])
890 except JujuAPIError
as e
:
891 if 'relation already exists' not in e
.message
:
894 'Relation %s <-> %s already exists', relation1
, relation2
)
895 # TODO: if relation already exists we should return the
896 # Relation ModelEntity here
899 def predicate(delta
):
901 for endpoint
in delta
.data
['endpoints']:
902 endpoints
[endpoint
['application-name']] = endpoint
['relation']
903 return endpoints
== result
.endpoints
905 return await self
._wait
_for
_new
('relation', None, predicate
)
907 def add_space(self
, name
, *cidrs
):
908 """Add a new network space.
910 Adds a new space with the given name and associates the given
911 (optional) list of existing subnet CIDRs with it.
913 :param str name: Name of the space
914 :param \*cidrs: Optional list of existing subnet CIDRs
917 raise NotImplementedError()
919 async def add_ssh_key(self
, user
, key
):
920 """Add a public SSH key to this model.
922 :param str user: The username of the user
923 :param str key: The public ssh key
926 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection
)
927 return await key_facade
.AddKeys([key
], user
)
928 add_ssh_keys
= add_ssh_key
930 def add_subnet(self
, cidr_or_id
, space
, *zones
):
931 """Add an existing subnet to this model.
933 :param str cidr_or_id: CIDR or provider ID of the existing subnet
934 :param str space: Network space with which to associate
935 :param str \*zones: Zone(s) in which the subnet resides
938 raise NotImplementedError()
940 def get_backups(self
):
941 """Retrieve metadata for backups in this model.
944 raise NotImplementedError()
946 def block(self
, *commands
):
947 """Add a new block to this model.
949 :param str \*commands: The commands to block. Valid values are
950 'all-changes', 'destroy-model', 'remove-object'
953 raise NotImplementedError()
955 def get_blocks(self
):
956 """List blocks for this model.
959 raise NotImplementedError()
961 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
962 """Return a list of cached OS images.
964 :param str arch: Filter by image architecture
965 :param str kind: Filter by image kind, e.g. 'lxd'
966 :param str series: Filter by image series, e.g. 'xenial'
969 raise NotImplementedError()
971 def create_backup(self
, note
=None, no_download
=False):
972 """Create a backup of this model.
974 :param str note: A note to store with the backup
975 :param bool no_download: Do not download the backup archive
976 :return str: Path to downloaded archive
979 raise NotImplementedError()
981 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
982 """Create or define a storage pool.
984 :param str name: Name to give the storage pool
985 :param str provider_type: Pool provider type
986 :param \*\*pool_config: key/value pool configuration pairs
989 raise NotImplementedError()
992 self
, no_tail
=False, exclude_module
=None, include_module
=None,
993 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
995 """Get log messages for this model.
997 :param bool no_tail: Stop after returning existing log messages
998 :param list exclude_module: Do not show log messages for these logging
1000 :param list include_module: Only show log messages for these logging
1002 :param list include: Only show log messages for these entities
1003 :param str level: Log level to show, valid options are 'TRACE',
1004 'DEBUG', 'INFO', 'WARNING', 'ERROR,
1005 :param int limit: Return this many of the most recent (possibly
1006 filtered) lines are shown
1007 :param int lines: Yield this many of the most recent lines, and keep
1009 :param bool replay: Yield the entire log, and keep yielding
1010 :param list exclude: Do not show log messages for these entities
1013 raise NotImplementedError()
1015 def _get_series(self
, entity_url
, entity
):
1016 # try to get the series from the provided charm URL
1017 if entity_url
.startswith('cs:'):
1018 parts
= entity_url
[3:].split('/')
1020 parts
= entity_url
.split('/')
1021 if parts
[0].startswith('~'):
1024 # series was specified in the URL
1026 # series was not supplied at all, so use the newest
1027 # supported series according to the charm store
1028 ss
= entity
['Meta']['supported-series']
1029 return ss
['SupportedSeries'][0]
1032 self
, entity_url
, application_name
=None, bind
=None, budget
=None,
1033 channel
=None, config
=None, constraints
=None, force
=False,
1034 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
1036 """Deploy a new service or bundle.
1038 :param str entity_url: Charm or bundle url
1039 :param str application_name: Name to give the service
1040 :param dict bind: <charm endpoint>:<network space> pairs
1041 :param dict budget: <budget name>:<limit> pairs
1042 :param str channel: Charm store channel from which to retrieve
1043 the charm or bundle, e.g. 'edge'
1044 :param dict config: Charm configuration dictionary
1045 :param constraints: Service constraints
1046 :type constraints: :class:`juju.Constraints`
1047 :param bool force: Allow charm to be deployed to a machine running
1048 an unsupported series
1049 :param int num_units: Number of units to deploy
1050 :param str plan: Plan under which to deploy charm
1051 :param dict resources: <resource name>:<file path> pairs
1052 :param str series: Series on which to deploy
1053 :param dict storage: Storage constraints TODO how do these look?
1054 :param to: Placement directive as a string. For example:
1056 '23' - place on machine 23
1057 'lxd:7' - place in new lxd container on machine 7
1058 '24/lxd/3' - place in container 3 on machine 24
1060 If None, a new machine is provisioned.
1065 - support local resources
1070 k
: client
.Constraints(**v
)
1071 for k
, v
in storage
.items()
1075 entity_url
.startswith('local:') or
1076 os
.path
.isdir(entity_url
)
1079 entity_id
= entity_url
.replace('local:', '')
1081 entity
= await self
.charmstore
.entity(entity_url
, channel
=channel
)
1082 entity_id
= entity
['Id']
1084 client_facade
= client
.ClientFacade
.from_connection(self
.connection
)
1086 is_bundle
= ((is_local
and
1087 (Path(entity_id
) / 'bundle.yaml').exists()) or
1088 (not is_local
and 'bundle/' in entity_id
))
1091 handler
= BundleHandler(self
)
1092 await handler
.fetch_plan(entity_id
)
1093 await handler
.execute_plan()
1094 extant_apps
= {app
for app
in self
.applications
}
1095 pending_apps
= set(handler
.applications
) - extant_apps
1097 # new apps will usually be in the model by now, but if some
1098 # haven't made it yet we'll need to wait on them to be added
1099 await asyncio
.gather(*[
1100 asyncio
.ensure_future(
1101 self
._wait
_for
_new
('application', app_name
),
1103 for app_name
in pending_apps
1105 return [app
for name
, app
in self
.applications
.items()
1106 if name
in handler
.applications
]
1109 if not application_name
:
1110 application_name
= entity
['Meta']['charm-metadata']['Name']
1112 series
= self
._get
_series
(entity_url
, entity
)
1113 await client_facade
.AddCharm(channel
, entity_id
)
1114 # XXX: we're dropping local resources here, but we don't
1115 # actually support them yet anyway
1116 resources
= await self
._add
_store
_resources
(application_name
,
1120 # We have a local charm dir that needs to be uploaded
1121 charm_dir
= os
.path
.abspath(
1122 os
.path
.expanduser(entity_id
))
1123 series
= series
or get_charm_series(charm_dir
)
1126 "Couldn't determine series for charm at {}. "
1127 "Pass a 'series' kwarg to Model.deploy().".format(
1129 entity_id
= await self
.add_local_charm_dir(charm_dir
, series
)
1130 return await self
._deploy
(
1131 charm_url
=entity_id
,
1132 application
=application_name
,
1134 config
=config
or {},
1135 constraints
=constraints
,
1136 endpoint_bindings
=bind
,
1137 resources
=resources
,
1140 num_units
=num_units
,
1141 placement
=parse_placement(to
)
1144 async def _add_store_resources(self
, application
, entity_url
, entity
=None):
1146 # avoid extra charm store call if one was already made
1147 entity
= await self
.charmstore
.entity(entity_url
)
1150 'description': resource
['Description'],
1151 'fingerprint': resource
['Fingerprint'],
1152 'name': resource
['Name'],
1153 'path': resource
['Path'],
1154 'revision': resource
['Revision'],
1155 'size': resource
['Size'],
1156 'type_': resource
['Type'],
1158 } for resource
in entity
['Meta']['resources']
1164 resources_facade
= client
.ResourcesFacade
.from_connection(
1166 response
= await resources_facade
.AddPendingResources(
1167 tag
.application(application
),
1169 [client
.CharmResource(**resource
) for resource
in resources
])
1170 resource_map
= {resource
['name']: pid
1172 in zip(resources
, response
.pending_ids
)}
1175 async def _deploy(self
, charm_url
, application
, series
, config
,
1176 constraints
, endpoint_bindings
, resources
, storage
,
1177 channel
=None, num_units
=None, placement
=None):
1178 """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
1180 log
.info('Deploying %s', charm_url
)
1182 # stringify all config values for API, and convert to YAML
1183 config
= {k
: str(v
) for k
, v
in config
.items()}
1184 config
= yaml
.dump({application
: config
},
1185 default_flow_style
=False)
1187 app_facade
= client
.ApplicationFacade
.from_connection(
1190 app
= client
.ApplicationDeploy(
1191 charm_url
=charm_url
,
1192 application
=application
,
1196 constraints
=parse_constraints(constraints
),
1197 endpoint_bindings
=endpoint_bindings
,
1198 num_units
=num_units
,
1199 resources
=resources
,
1204 result
= await app_facade
.Deploy([app
])
1205 errors
= [r
.error
.message
for r
in result
.results
if r
.error
]
1207 raise JujuError('\n'.join(errors
))
1208 return await self
._wait
_for
_new
('application', application
)
1210 async def destroy(self
):
1211 """Terminate all machines and resources for this model.
1212 Is already implemented in controller.py.
1214 raise NotImplementedError()
1216 async def destroy_unit(self
, *unit_names
):
1217 """Destroy units by name.
1220 app_facade
= client
.ApplicationFacade
.from_connection(self
.connection
)
1223 'Destroying unit%s %s',
1224 's' if len(unit_names
) == 1 else '',
1225 ' '.join(unit_names
))
1227 return await app_facade
.DestroyUnits(list(unit_names
))
1228 destroy_units
= destroy_unit
1230 def get_backup(self
, archive_id
):
1231 """Download a backup archive file.
1233 :param str archive_id: The id of the archive to download
1234 :return str: Path to the archive file
1237 raise NotImplementedError()
1240 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
1241 """Ensure sufficient controllers exist to provide redundancy.
1243 :param int num_controllers: Number of controllers to make available
1244 :param constraints: Constraints to apply to the controller machines
1245 :type constraints: :class:`juju.Constraints`
1246 :param str series: Series of the controller machines
1247 :param list to: Placement directives for controller machines, e.g.::
1250 'lxc:7' - new lxc container on machine 7
1251 '24/lxc/3' - lxc container 3 or machine 24
1253 If None, a new machine is provisioned.
1256 raise NotImplementedError()
1258 def get_config(self
):
1259 """Return the configuration settings for this model.
1262 raise NotImplementedError()
1264 def get_constraints(self
):
1265 """Return the machine constraints for this model.
1268 raise NotImplementedError()
1270 async def grant(self
, username
, acl
='read'):
1271 """Grant a user access to this model.
1273 :param str username: Username
1274 :param str acl: Access control ('read' or 'write')
1277 controller_conn
= await self
.connection
.controller()
1278 model_facade
= client
.ModelManagerFacade
.from_connection(
1280 user
= tag
.user(username
)
1281 model
= tag
.model(self
.info
.uuid
)
1282 changes
= client
.ModifyModelAccess(acl
, 'grant', model
, user
)
1283 await self
.revoke(username
)
1284 return await model_facade
.ModifyModelAccess([changes
])
1286 def import_ssh_key(self
, identity
):
1287 """Add a public SSH key from a trusted indentity source to this model.
1289 :param str identity: User identity in the form <lp|gh>:<username>
1292 raise NotImplementedError()
1293 import_ssh_keys
= import_ssh_key
1295 async def get_machines(self
):
1296 """Return list of machines in this model.
1299 return list(self
.state
.machines
.keys())
1301 def get_shares(self
):
1302 """Return list of all users with access to this model.
1305 raise NotImplementedError()
1307 def get_spaces(self
):
1308 """Return list of all known spaces, including associated subnets.
1311 raise NotImplementedError()
1313 async def get_ssh_key(self
, raw_ssh
=False):
1314 """Return known SSH keys for this model.
1315 :param bool raw_ssh: if True, returns the raw ssh key,
1316 else it's fingerprint
1319 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection
)
1320 entity
= {'tag': tag
.model(self
.info
.uuid
)}
1321 entities
= client
.Entities([entity
])
1322 return await key_facade
.ListKeys(entities
, raw_ssh
)
1323 get_ssh_keys
= get_ssh_key
1325 def get_storage(self
, filesystem
=False, volume
=False):
1326 """Return details of storage instances.
1328 :param bool filesystem: Include filesystem storage
1329 :param bool volume: Include volume storage
1332 raise NotImplementedError()
1334 def get_storage_pools(self
, names
=None, providers
=None):
1335 """Return list of storage pools.
1337 :param list names: Only include pools with these names
1338 :param list providers: Only include pools for these providers
1341 raise NotImplementedError()
1343 def get_subnets(self
, space
=None, zone
=None):
1344 """Return list of known subnets.
1346 :param str space: Only include subnets in this space
1347 :param str zone: Only include subnets in this zone
1350 raise NotImplementedError()
1352 def remove_blocks(self
):
1353 """Remove all blocks from this model.
1356 raise NotImplementedError()
1358 def remove_backup(self
, backup_id
):
1361 :param str backup_id: The id of the backup to remove
1364 raise NotImplementedError()
1366 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1367 """Remove cached OS images.
1369 :param str arch: Architecture of the images to remove
1370 :param str kind: Image kind to remove, e.g. 'lxd'
1371 :param str series: Image series to remove, e.g. 'xenial'
1374 raise NotImplementedError()
1376 def remove_machine(self
, *machine_ids
):
1377 """Remove a machine from this model.
1379 :param str \*machine_ids: Ids of the machines to remove
1382 raise NotImplementedError()
1383 remove_machines
= remove_machine
1385 async def remove_ssh_key(self
, user
, key
):
1386 """Remove a public SSH key(s) from this model.
1388 :param str key: Full ssh key
1389 :param str user: Juju user to which the key is registered
1392 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection
)
1393 key
= base64
.b64decode(bytes(key
.strip().split()[1].encode('ascii')))
1394 key
= hashlib
.md5(key
).hexdigest()
1395 key
= ':'.join(a
+b
for a
, b
in zip(key
[::2], key
[1::2]))
1396 await key_facade
.DeleteKeys([key
], user
)
1397 remove_ssh_keys
= remove_ssh_key
1400 self
, bootstrap
=False, constraints
=None, archive
=None,
1401 backup_id
=None, upload_tools
=False):
1402 """Restore a backup archive to a new controller.
1404 :param bool bootstrap: Bootstrap a new state machine
1405 :param constraints: Model constraints
1406 :type constraints: :class:`juju.Constraints`
1407 :param str archive: Path to backup archive to restore
1408 :param str backup_id: Id of backup to restore
1409 :param bool upload_tools: Upload tools if bootstrapping a new machine
1412 raise NotImplementedError()
1414 def retry_provisioning(self
):
1415 """Retry provisioning for failed machines.
1418 raise NotImplementedError()
1420 async def revoke(self
, username
):
1421 """Revoke a user's access to this model.
1423 :param str username: Username to revoke
1426 controller_conn
= await self
.connection
.controller()
1427 model_facade
= client
.ModelManagerFacade
.from_connection(
1429 user
= tag
.user(username
)
1430 model
= tag
.model(self
.info
.uuid
)
1431 changes
= client
.ModifyModelAccess('read', 'revoke', model
, user
)
1432 return await model_facade
.ModifyModelAccess([changes
])
1434 def run(self
, command
, timeout
=None):
1435 """Run command on all machines in this model.
1437 :param str command: The command to run
1438 :param int timeout: Time to wait before command is considered failed
1441 raise NotImplementedError()
1443 def set_config(self
, **config
):
1444 """Set configuration keys on this model.
1446 :param \*\*config: Config key/values
1449 raise NotImplementedError()
1451 def set_constraints(self
, constraints
):
1452 """Set machine constraints on this model.
1454 :param :class:`juju.Constraints` constraints: Machine constraints
1457 raise NotImplementedError()
1459 def get_action_output(self
, action_uuid
, wait
=-1):
1460 """Get the results of an action by ID.
1462 :param str action_uuid: Id of the action
1463 :param int wait: Time in seconds to wait for action to complete
1466 raise NotImplementedError()
1468 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1469 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1471 :param str uuid_or_prefix: Filter by action uuid or prefix
1472 :param str name: Filter by action name
1475 raise NotImplementedError()
1477 def get_budget(self
, budget_name
):
1478 """Get budget usage info.
1480 :param str budget_name: Name of budget
1483 raise NotImplementedError()
1485 async def get_status(self
, filters
=None, utc
=False):
1486 """Return the status of the model.
1488 :param str filters: Optional list of applications, units, or machines
1489 to include, which can use wildcards ('*').
1490 :param bool utc: Display time as UTC in RFC3339 format
1493 client_facade
= client
.ClientFacade
.from_connection(self
.connection
)
1494 return await client_facade
.FullStatus(filters
)
1497 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1498 source
=None, stream
=None, version
=None):
1499 """Copy Juju tools into this model.
1501 :param bool all_: Copy all versions, not just the latest
1502 :param str destination: Path to local destination directory
1503 :param bool dry_run: Don't do the actual copy
1504 :param bool public: Tools are for a public cloud, so generate mirrors
1506 :param str source: Path to local source directory
1507 :param str stream: Simplestreams stream for which to sync metadata
1508 :param str version: Copy a specific major.minor version
1511 raise NotImplementedError()
1513 def unblock(self
, *commands
):
1514 """Unblock an operation that would alter this model.
1516 :param str \*commands: The commands to unblock. Valid values are
1517 'all-changes', 'destroy-model', 'remove-object'
1520 raise NotImplementedError()
1522 def unset_config(self
, *keys
):
1523 """Unset configuration on this model.
1525 :param str \*keys: The keys to unset
1528 raise NotImplementedError()
1530 def upgrade_gui(self
):
1531 """Upgrade the Juju GUI for this model.
1534 raise NotImplementedError()
1537 self
, dry_run
=False, reset_previous_upgrade
=False,
1538 upload_tools
=False, version
=None):
1539 """Upgrade Juju on all machines in a model.
1541 :param bool dry_run: Don't do the actual upgrade
1542 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1544 :param bool upload_tools: Upload local version of tools
1545 :param str version: Upgrade to a specific version
1548 raise NotImplementedError()
1550 def upload_backup(self
, archive_path
):
1551 """Store a backup archive remotely in Juju.
1553 :param str archive_path: Path to local archive
1556 raise NotImplementedError()
1559 def charmstore(self
):
1560 return self
._charmstore
1562 async def get_metrics(self
, *tags
):
1563 """Retrieve metrics.
1565 :param str \*tags: Tags of entities from which to retrieve metrics.
1566 No tags retrieves the metrics of all units in the model.
1567 :return: Dictionary of unit_name:metrics
1570 log
.debug("Retrieving metrics for %s",
1571 ', '.join(tags
) if tags
else "all units")
1573 metrics_facade
= client
.MetricsDebugFacade
.from_connection(
1576 entities
= [client
.Entity(tag
) for tag
in tags
]
1577 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1579 metrics
= collections
.defaultdict(list)
1581 for entity_metrics
in metrics_result
.results
:
1582 error
= entity_metrics
.error
1584 if "is not a valid tag" in error
:
1585 raise ValueError(error
.message
)
1587 raise Exception(error
.message
)
1589 for metric
in entity_metrics
.metrics
:
1590 metrics
[metric
.unit
].append(vars(metric
))
1595 def get_charm_series(path
):
1596 """Inspects the charm directory at ``path`` and returns a default
1597 series from its metadata.yaml (the first item in the 'series' list).
1599 Returns None if no series can be determined.
1602 md
= Path(path
) / "metadata.yaml"
1605 data
= yaml
.load(md
.open())
1606 series
= data
.get('series')
1607 return series
[0] if series
else None
1610 class BundleHandler(object):
1612 Handle bundles by using the API to translate bundle YAML into a plan of
1613 steps and then dispatching each of those using the API.
1615 def __init__(self
, model
):
1617 self
.charmstore
= model
.charmstore
1619 self
.references
= {}
1620 self
._units
_by
_app
= {}
1621 for unit_name
, unit
in model
.units
.items():
1622 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1623 app_units
.append(unit_name
)
1624 self
.client_facade
= client
.ClientFacade
.from_connection(
1626 self
.app_facade
= client
.ApplicationFacade
.from_connection(
1628 self
.ann_facade
= client
.AnnotationsFacade
.from_connection(
1631 async def _handle_local_charms(self
, bundle
):
1632 """Search for references to local charms (i.e. filesystem paths)
1633 in the bundle. Upload the local charms to the model, and replace
1634 the filesystem paths with appropriate 'local:' paths in the bundle.
1636 Return the modified bundle.
1638 :param dict bundle: Bundle dictionary
1639 :return: Modified bundle dictionary
1644 default_series
= bundle
.get('series')
1645 for app_name
in self
.applications
:
1646 app_dict
= bundle
['services'][app_name
]
1647 charm_dir
= os
.path
.abspath(os
.path
.expanduser(app_dict
['charm']))
1648 if not os
.path
.isdir(charm_dir
):
1651 app_dict
.get('series') or
1653 get_charm_series(charm_dir
)
1657 "Couldn't determine series for charm at {}. "
1658 "Add a 'series' key to the bundle.".format(charm_dir
))
1660 # Keep track of what we need to update. We keep a list of apps
1661 # that need to be updated, and a corresponding list of args
1662 # needed to update those apps.
1663 apps
.append(app_name
)
1664 args
.append((charm_dir
, series
))
1667 # If we have apps to update, spawn all the coroutines concurrently
1668 # and wait for them to finish.
1669 charm_urls
= await asyncio
.gather(*[
1670 self
.model
.add_local_charm_dir(*params
)
1672 ], loop
=self
.model
.loop
)
1673 # Update the 'charm:' entry for each app with the new 'local:' url.
1674 for app_name
, charm_url
in zip(apps
, charm_urls
):
1675 bundle
['services'][app_name
]['charm'] = charm_url
1679 async def fetch_plan(self
, entity_id
):
1680 is_local
= not entity_id
.startswith('cs:') and os
.path
.isdir(entity_id
)
1682 bundle_yaml
= (Path(entity_id
) / "bundle.yaml").read_text()
1684 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1685 filename
='bundle.yaml',
1687 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1688 self
.bundle
= await self
._handle
_local
_charms
(self
.bundle
)
1690 self
.plan
= await self
.client_facade
.GetBundleChanges(
1691 yaml
.dump(self
.bundle
))
1693 async def execute_plan(self
):
1694 for step
in self
.plan
.changes
:
1695 method
= getattr(self
, step
.method
)
1696 result
= await method(*step
.args
)
1697 self
.references
[step
.id_
] = result
1700 def applications(self
):
1701 return list(self
.bundle
['services'].keys())
1703 def resolve(self
, reference
):
1704 if reference
and reference
.startswith('$'):
1705 reference
= self
.references
[reference
[1:]]
1708 async def addCharm(self
, charm
, series
):
1710 :param charm string:
1711 Charm holds the URL of the charm to be added.
1713 :param series string:
1714 Series holds the series of the charm to be added
1715 if the charm default is not sufficient.
1717 # We don't add local charms because they've already been added
1718 # by self._handle_local_charms
1719 if charm
.startswith('local:'):
1722 entity_id
= await self
.charmstore
.entityId(charm
)
1723 log
.debug('Adding %s', entity_id
)
1724 await self
.client_facade
.AddCharm(None, entity_id
)
1727 async def addMachines(self
, params
=None):
1730 Dictionary specifying the machine to add. All keys are optional.
1733 series: string specifying the machine OS series.
1735 constraints: string holding machine constraints, if any. We'll
1736 parse this into the json friendly dict that the juju api
1739 container_type: string holding the type of the container (for
1740 instance ""lxd" or kvm"). It is not specified for top level
1743 parent_id: string holding a placeholder pointing to another
1744 machine change or to a unit change. This value is only
1745 specified in the case this machine is a container, in
1746 which case also ContainerType is set.
1749 params
= params
or {}
1752 params
= {normalize_key(k
): params
[k
] for k
in params
.keys()}
1754 # Fix up values, as necessary.
1755 if 'parent_id' in params
:
1756 params
['parent_id'] = self
.resolve(params
['parent_id'])
1758 params
['constraints'] = parse_constraints(
1759 params
.get('constraints'))
1760 params
['jobs'] = params
.get('jobs', ['JobHostUnits'])
1762 if params
.get('container_type') == 'lxc':
1763 log
.warning('Juju 2.0 does not support lxc containers. '
1764 'Converting containers to lxd.')
1765 params
['container_type'] = 'lxd'
1767 # Submit the request.
1768 params
= client
.AddMachineParams(**params
)
1769 results
= await self
.client_facade
.AddMachines([params
])
1770 error
= results
.machines
[0].error
1772 raise ValueError("Error adding machine: %s" % error
.message
)
1773 machine
= results
.machines
[0].machine
1774 log
.debug('Added new machine %s', machine
)
1777 async def addRelation(self
, endpoint1
, endpoint2
):
1779 :param endpoint1 string:
1780 :param endpoint2 string:
1781 Endpoint1 and Endpoint2 hold relation endpoints in the
1782 "application:interface" form, where the application is always a
1783 placeholder pointing to an application change, and the interface is
1784 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1786 endpoints
= [endpoint1
, endpoint2
]
1787 # resolve indirect references
1788 for i
in range(len(endpoints
)):
1789 parts
= endpoints
[i
].split(':')
1790 parts
[0] = self
.resolve(parts
[0])
1791 endpoints
[i
] = ':'.join(parts
)
1793 log
.info('Relating %s <-> %s', *endpoints
)
1794 return await self
.model
.add_relation(*endpoints
)
1796 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1797 storage
, endpoint_bindings
, resources
):
1799 :param charm string:
1800 Charm holds the URL of the charm to be used to deploy this
1803 :param series string:
1804 Series holds the series of the application to be deployed
1805 if the charm default is not sufficient.
1807 :param application string:
1808 Application holds the application name.
1810 :param options map[string]interface{}:
1811 Options holds application options.
1813 :param constraints string:
1814 Constraints holds the optional application constraints.
1816 :param storage map[string]string:
1817 Storage holds the optional storage constraints.
1819 :param endpoint_bindings map[string]string:
1820 EndpointBindings holds the optional endpoint bindings
1822 :param resources map[string]int:
1823 Resources identifies the revision to use for each resource
1824 of the application's charm.
1826 # resolve indirect references
1827 charm
= self
.resolve(charm
)
1828 # the bundle plan doesn't actually do anything with resources, even
1829 # though it ostensibly gives us something (None) for that param
1830 if not charm
.startswith('local:'):
1831 resources
= await self
.model
._add
_store
_resources
(application
,
1833 await self
.model
._deploy
(
1835 application
=application
,
1838 constraints
=constraints
,
1839 endpoint_bindings
=endpoint_bindings
,
1840 resources
=resources
,
1845 async def addUnit(self
, application
, to
):
1847 :param application string:
1848 Application holds the application placeholder name for which a unit
1852 To holds the optional location where to add the unit, as a
1853 placeholder pointing to another unit change or to a machine change.
1855 application
= self
.resolve(application
)
1856 placement
= self
.resolve(to
)
1857 if self
._units
_by
_app
.get(application
):
1858 # enough units for this application already exist;
1859 # claim one, and carry on
1860 # NB: this should probably honor placement, but the juju client
1861 # doesn't, so we're not bothering, either
1862 unit_name
= self
._units
_by
_app
[application
].pop()
1863 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1864 return self
.model
.units
[unit_name
]
1866 log
.debug('Adding new unit for %s%s', application
,
1867 ' to %s' % placement
if placement
else '')
1868 return await self
.model
.applications
[application
].add_unit(
1873 async def expose(self
, application
):
1875 :param application string:
1876 Application holds the placeholder name of the application that must
1879 application
= self
.resolve(application
)
1880 log
.info('Exposing %s', application
)
1881 return await self
.model
.applications
[application
].expose()
1883 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1886 Id is the placeholder for the application or machine change
1887 corresponding to the entity to be annotated.
1889 :param entity_type EntityType:
1890 EntityType holds the type of the entity, "application" or
1893 :param annotations map[string]string:
1894 Annotations holds the annotations as key/value pairs.
1896 entity_id
= self
.resolve(id_
)
1898 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
1900 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
1901 return await entity
.set_annotations(annotations
)
1904 class CharmStore(object):
1906 Async wrapper around theblues.charmstore.CharmStore
1908 def __init__(self
, loop
):
1910 self
._cs
= theblues
.charmstore
.CharmStore(timeout
=5)
1912 def __getattr__(self
, name
):
1914 Wrap method calls in coroutines that use run_in_executor to make them
1917 attr
= getattr(self
._cs
, name
)
1918 if not callable(attr
):
1919 wrapper
= partial(getattr, self
._cs
, name
)
1920 setattr(self
, name
, wrapper
)
1922 async def coro(*args
, **kwargs
):
1923 method
= partial(attr
, *args
, **kwargs
)
1924 for attempt
in range(1, 4):
1926 return await self
.loop
.run_in_executor(None, method
)
1927 except theblues
.errors
.ServerError
:
1930 await asyncio
.sleep(1, loop
=self
.loop
)
1931 setattr(self
, name
, coro
)
1936 class CharmArchiveGenerator(object):
1938 Create a Zip archive of a local charm directory for upload to a controller.
1940 This is used automatically by
1941 `Model.add_local_charm_dir <#juju.model.Model.add_local_charm_dir>`_.
1943 def __init__(self
, path
):
1944 self
.path
= os
.path
.abspath(os
.path
.expanduser(path
))
1946 def make_archive(self
, path
):
1947 """Create archive of directory and write to ``path``.
1949 :param path: Path to archive
1953 * build/\* - This is used for packing the charm itself and any
1955 * \*/.\* - Hidden files are all ignored for now. This will most
1956 likely be changed into a specific ignore list
1960 zf
= zipfile
.ZipFile(path
, 'w', zipfile
.ZIP_DEFLATED
)
1961 for dirpath
, dirnames
, filenames
in os
.walk(self
.path
):
1962 relative_path
= dirpath
[len(self
.path
) + 1:]
1963 if relative_path
and not self
._ignore
(relative_path
):
1964 zf
.write(dirpath
, relative_path
)
1965 for name
in filenames
:
1966 archive_name
= os
.path
.join(relative_path
, name
)
1967 if not self
._ignore
(archive_name
):
1968 real_path
= os
.path
.join(dirpath
, name
)
1969 self
._check
_type
(real_path
)
1970 if os
.path
.islink(real_path
):
1971 self
._check
_link
(real_path
)
1972 self
._write
_symlink
(
1973 zf
, os
.readlink(real_path
), archive_name
)
1975 zf
.write(real_path
, archive_name
)
1979 def _check_type(self
, path
):
1983 if stat
.S_ISDIR(s
.st_mode
) or stat
.S_ISREG(s
.st_mode
):
1985 raise ValueError("Invalid Charm at % %s" % (
1986 path
, "Invalid file type for a charm"))
1988 def _check_link(self
, path
):
1989 link_path
= os
.readlink(path
)
1990 if link_path
[0] == "/":
1992 "Invalid Charm at %s: %s" % (
1993 path
, "Absolute links are invalid"))
1994 path_dir
= os
.path
.dirname(path
)
1995 link_path
= os
.path
.join(path_dir
, link_path
)
1996 if not link_path
.startswith(os
.path
.abspath(self
.path
)):
1998 "Invalid charm at %s %s" % (
1999 path
, "Only internal symlinks are allowed"))
2001 def _write_symlink(self
, zf
, link_target
, link_path
):
2002 """Package symlinks with appropriate zipfile metadata."""
2003 info
= zipfile
.ZipInfo()
2004 info
.filename
= link_path
2005 info
.create_system
= 3
2006 # Magic code for symlinks / py2/3 compat
2007 # 27166663808 = (stat.S_IFLNK | 0755) << 16
2008 info
.external_attr
= 2716663808
2009 zf
.writestr(info
, link_target
)
2011 def _ignore(self
, path
):
2012 if path
== "build" or path
.startswith("build/"):
2014 if path
.startswith('.'):