13 from concurrent
.futures
import CancelledError
14 from functools
import partial
15 from pathlib
import Path
18 import theblues
.charmstore
19 import theblues
.errors
22 from .client
import client
23 from .client
import watcher
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):
80 async def __call__(self
, delta
, old
, new
, model
):
81 handler_name
= 'on_{}_{}'.format(delta
.entity
, delta
.type)
82 method
= getattr(self
, handler_name
, self
.on_change
)
83 await method(delta
, old
, new
, model
)
85 async def on_change(self
, delta
, old
, new
, model
):
86 """Generic model-change handler.
88 :param delta: :class:`juju.client.overrides.Delta`
89 :param old: :class:`juju.model.ModelEntity`
90 :param new: :class:`juju.model.ModelEntity`
91 :param model: :class:`juju.model.Model`
97 class ModelState(object):
98 """Holds the state of the model, including the delta history of all
99 entities in the model.
102 def __init__(self
, model
):
106 def _live_entity_map(self
, entity_type
):
107 """Return an id:Entity map of all the living entities of
108 type ``entity_type``.
112 entity_id
: self
.get_entity(entity_type
, entity_id
)
113 for entity_id
, history
in self
.state
.get(entity_type
, {}).items()
114 if history
[-1] is not None
118 def applications(self
):
119 """Return a map of application-name:Application for all applications
120 currently in the model.
123 return self
._live
_entity
_map
('application')
127 """Return a map of machine-id:Machine for all machines currently in
131 return self
._live
_entity
_map
('machine')
135 """Return a map of unit-id:Unit for all units currently in
139 return self
._live
_entity
_map
('unit')
141 def entity_history(self
, entity_type
, entity_id
):
142 """Return the history deque for an entity.
145 return self
.state
[entity_type
][entity_id
]
147 def entity_data(self
, entity_type
, entity_id
, history_index
):
148 """Return the data dict for an entity at a specific index of its
152 return self
.entity_history(entity_type
, entity_id
)[history_index
]
154 def apply_delta(self
, delta
):
155 """Apply delta to our state and return a copy of the
156 affected object as it was before and after the update, e.g.:
158 old_obj, new_obj = self.apply_delta(delta)
160 old_obj may be None if the delta is for the creation of a new object,
161 e.g. a new application or unit is deployed.
163 new_obj will never be None, but may be dead (new_obj.dead == True)
164 if the object was deleted as a result of the delta being applied.
169 .setdefault(delta
.entity
, {})
170 .setdefault(delta
.get_id(), collections
.deque())
173 history
.append(delta
.data
)
174 if delta
.type == 'remove':
177 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
178 return entity
.previous(), entity
181 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
182 """Return an object instance for the given entity_type and id.
184 By default the object state matches the most recent state from
185 Juju. To get an instance of the object in an older state, pass
186 history_index, an index into the history deque for the entity.
190 if history_index
< 0 and history_index
!= -1:
191 history_index
+= len(self
.entity_history(entity_type
, entity_id
))
192 if history_index
< 0:
196 self
.entity_data(entity_type
, entity_id
, history_index
)
200 entity_class
= get_entity_class(entity_type
)
202 entity_id
, self
.model
, history_index
=history_index
,
206 class ModelEntity(object):
207 """An object in the Model tree"""
209 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
210 """Initialize a new entity
212 :param entity_id str: The unique id of the object in the model
213 :param model: The model instance in whose object tree this
215 :history_index int: The index of this object's state in the model's
216 history deque for this entity
217 :connected bool: Flag indicating whether this object gets live updates
221 self
.entity_id
= entity_id
223 self
._history
_index
= history_index
224 self
.connected
= connected
225 self
.connection
= model
.connection
228 return '<{} entity_id="{}">'.format(type(self
).__name
__,
231 def __getattr__(self
, name
):
232 """Fetch object attributes from the underlying data dict held in the
237 return self
.safe_data
[name
]
239 name
= name
.replace('_', '-')
240 if name
in self
.safe_data
:
241 return self
.safe_data
[name
]
246 return bool(self
.data
)
248 def on_change(self
, callable_
):
249 """Add a change observer to this entity.
252 self
.model
.add_observer(
253 callable_
, self
.entity_type
, 'change', self
.entity_id
)
255 def on_remove(self
, callable_
):
256 """Add a remove observer to this entity.
259 self
.model
.add_observer(
260 callable_
, self
.entity_type
, 'remove', self
.entity_id
)
263 def entity_type(self
):
264 """A string identifying the entity type of this object, e.g.
265 'application' or 'unit', etc.
268 return self
.__class
__.__name
__.lower()
272 """Return True if this object represents the current state of the
273 entity in the underlying model.
275 This will be True except when the object represents an entity at a
276 non-latest state in history, e.g. if the object was obtained by calling
277 .previous() on another object.
280 return self
._history
_index
== -1
284 """Returns True if this entity no longer exists in the underlying
290 self
.model
.state
.entity_data(
291 self
.entity_type
, self
.entity_id
, -1) is None
296 """Returns True if this entity still exists in the underlying
304 """The data dictionary for this entity.
307 return self
.model
.state
.entity_data(
308 self
.entity_type
, self
.entity_id
, self
._history
_index
)
312 """The data dictionary for this entity.
314 If this `ModelEntity` points to the dead state, it will
315 raise `DeadEntityException`.
318 if self
.data
is None:
319 raise DeadEntityException(
320 "Entity {}:{} is dead - its attributes can no longer be "
321 "accessed. Use the .previous() method on this object to get "
322 "a copy of the object at its previous state.".format(
323 self
.entity_type
, self
.entity_id
))
327 """Return a copy of this object as was at its previous state in
330 Returns None if this object is new (and therefore has no history).
332 The returned object is always "disconnected", i.e. does not receive
336 return self
.model
.state
.get_entity(
337 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
341 """Return a copy of this object at its next state in
344 Returns None if this object is already the latest.
346 The returned object is "disconnected", i.e. does not receive
347 live updates, unless it is current (latest).
350 if self
._history
_index
== -1:
353 new_index
= self
._history
_index
+ 1
355 new_index
== len(self
.model
.state
.entity_history(
356 self
.entity_type
, self
.entity_id
)) - 1
358 return self
.model
.state
.get_entity(
359 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
363 """Return a copy of this object at its current state in the model.
365 Returns self if this object is already the latest.
367 The returned object is always "connected", i.e. receives
368 live updates from the model.
371 if self
._history
_index
== -1:
374 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
378 def __init__(self
, loop
=None):
379 """Instantiate a new connected Model.
381 :param loop: an asyncio event loop
384 self
.loop
= loop
or asyncio
.get_event_loop()
385 self
.connection
= None
386 self
.observers
= weakref
.WeakValueDictionary()
387 self
.state
= ModelState(self
)
389 self
._watcher
_task
= None
390 self
._watch
_shutdown
= asyncio
.Event(loop
=self
.loop
)
391 self
._watch
_received
= asyncio
.Event(loop
=self
.loop
)
392 self
._charmstore
= CharmStore(self
.loop
)
394 async def connect(self
, *args
, **kw
):
395 """Connect to an arbitrary Juju model.
397 args and kw are passed through to Connection.connect()
401 kw
['loop'] = self
.loop
402 self
.connection
= await connection
.Connection
.connect(*args
, **kw
)
403 await self
._after
_connect
()
405 async def connect_current(self
):
406 """Connect to the current Juju model.
409 self
.connection
= await connection
.Connection
.connect_current(
411 await self
._after
_connect
()
413 async def connect_model(self
, model_name
):
414 """Connect to a specific Juju model by name.
416 :param model_name: Format [controller:][user/]model
419 self
.connection
= await connection
.Connection
.connect_model(model_name
,
421 await self
._after
_connect
()
423 async def _after_connect(self
):
424 """Run initialization steps after connecting to websocket.
428 await self
._watch
_received
.wait()
429 await self
.get_info()
431 async def disconnect(self
):
432 """Shut down the watcher task and close websockets.
435 self
._stop
_watching
()
436 if self
.connection
and self
.connection
.is_open
:
437 await self
._watch
_shutdown
.wait()
438 log
.debug('Closing model connection')
439 await self
.connection
.close()
440 self
.connection
= None
442 async def add_local_charm_dir(self
, charm_dir
, series
):
443 """Upload a local charm to the model.
445 This will automatically generate an archive from
448 :param charm_dir: Path to the charm directory
449 :param series: Charm series
452 fh
= tempfile
.NamedTemporaryFile()
453 CharmArchiveGenerator(charm_dir
).make_archive(fh
.name
)
456 self
.add_local_charm
, fh
, series
, os
.stat(fh
.name
).st_size
)
457 charm_url
= await self
.loop
.run_in_executor(None, func
)
459 log
.debug('Uploaded local charm: %s -> %s', charm_dir
, charm_url
)
462 def add_local_charm(self
, charm_file
, series
, size
=None):
463 """Upload a local charm archive to the model.
465 Returns the 'local:...' url that should be used to deploy the charm.
467 :param charm_file: Path to charm zip archive
468 :param series: Charm series
469 :param size: Size of the archive, in bytes
470 :return str: 'local:...' url for deploying the charm
471 :raises: :class:`JujuError` if the upload fails
473 Uses an https endpoint at the same host:port as the wss.
474 Supports large file uploads.
478 This method will block. Consider using :meth:`add_local_charm_dir`
482 conn
, headers
, path_prefix
= self
.connection
.https_connection()
483 path
= "%s/charms?series=%s" % (path_prefix
, series
)
484 headers
['Content-Type'] = 'application/zip'
486 headers
['Content-Length'] = size
487 conn
.request("POST", path
, charm_file
, headers
)
488 response
= conn
.getresponse()
489 result
= response
.read().decode()
490 if not response
.status
== 200:
491 raise JujuError(result
)
492 result
= json
.loads(result
)
493 return result
['charm-url']
495 def all_units_idle(self
):
496 """Return True if all units are idle.
499 for unit
in self
.units
.values():
500 unit_status
= unit
.data
['agent-status']['current']
501 if unit_status
!= 'idle':
505 async def reset(self
, force
=False):
506 """Reset the model to a clean state.
508 :param bool force: Force-terminate machines.
510 This returns only after the model has reached a clean state. "Clean"
511 means no applications or machines exist in the model.
514 log
.debug('Resetting model')
515 for app
in self
.applications
.values():
517 for machine
in self
.machines
.values():
518 await machine
.destroy(force
=force
)
519 await self
.block_until(
520 lambda: len(self
.machines
) == 0
523 async def block_until(self
, *conditions
, timeout
=None, wait_period
=0.5):
524 """Return only after all conditions are true.
528 while not all(c() for c
in conditions
):
529 await asyncio
.sleep(wait_period
, loop
=self
.loop
)
530 await asyncio
.wait_for(_block(), timeout
, loop
=self
.loop
)
533 def applications(self
):
534 """Return a map of application-name:Application for all applications
535 currently in the model.
538 return self
.state
.applications
542 """Return a map of machine-id:Machine for all machines currently in
546 return self
.state
.machines
550 """Return a map of unit-id:Unit for all units currently in
554 return self
.state
.units
556 async def get_info(self
):
557 """Return a client.ModelInfo object for this Model.
559 Retrieves latest info for this Model from the api server. The
560 return value is cached on the Model.info attribute so that the
561 valued may be accessed again without another api call, if
564 This method is called automatically when the Model is connected,
565 resulting in Model.info being initialized without requiring an
566 explicit call to this method.
569 facade
= client
.ClientFacade()
570 facade
.connect(self
.connection
)
572 self
.info
= await facade
.ModelInfo()
573 log
.debug('Got ModelInfo: %s', vars(self
.info
))
578 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
580 """Register an "on-model-change" callback
582 Once the model is connected, ``callable_``
583 will be called each time the model changes. ``callable_`` should
584 be Awaitable and accept the following positional arguments:
586 delta - An instance of :class:`juju.delta.EntityDelta`
587 containing the raw delta data recv'd from the Juju
590 old_obj - If the delta modifies an existing object in the model,
591 old_obj will be a copy of that object, as it was before the
592 delta was applied. Will be None if the delta creates a new
595 new_obj - A copy of the new or updated object, after the delta
596 is applied. Will be None if the delta removes an entity
599 model - The :class:`Model` itself.
601 Events for which ``callable_`` is called can be specified by passing
602 entity_type, action, and/or entitiy_id filter criteria, e.g.::
606 entity_type='application', action='add', entity_id='ubuntu')
608 For more complex filtering conditions, pass a predicate function. It
609 will be called with a delta as its only argument. If the predicate
610 function returns True, the ``callable_`` will be called.
613 observer
= _Observer(
614 callable_
, entity_type
, action
, entity_id
, predicate
)
615 self
.observers
[observer
] = callable_
618 """Start an asynchronous watch against this model.
620 See :meth:`add_observer` to register an onchange callback.
623 async def _start_watch():
624 self
._watch
_shutdown
.clear()
626 allwatcher
= watcher
.AllWatcher()
627 self
._watch
_conn
= await self
.connection
.clone()
628 allwatcher
.connect(self
._watch
_conn
)
630 results
= await allwatcher
.Next()
631 for delta
in results
.deltas
:
632 delta
= get_entity_delta(delta
)
633 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
634 # XXX: Might not want to shield at this level
635 # We are shielding because when the watcher is
636 # canceled (on disconnect()), we don't want all of
637 # its children (every observer callback) to be
638 # canceled with it. So we shield them. But this means
639 # they can *never* be canceled.
640 await asyncio
.shield(
641 self
._notify
_observers
(delta
, old_obj
, new_obj
),
643 self
._watch
_received
.set()
644 except CancelledError
:
645 log
.debug('Closing watcher connection')
646 await self
._watch
_conn
.close()
647 self
._watch
_shutdown
.set()
648 self
._watch
_conn
= None
650 log
.debug('Starting watcher task')
651 self
._watcher
_task
= self
.loop
.create_task(_start_watch())
653 def _stop_watching(self
):
654 """Stop the asynchronous watch against this model.
657 log
.debug('Stopping watcher task')
658 if self
._watcher
_task
:
659 self
._watcher
_task
.cancel()
661 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
662 """Call observing callbacks, notifying them of a change in model state
664 :param delta: The raw change from the watcher
665 (:class:`juju.client.overrides.Delta`)
666 :param old_obj: The object in the model that this delta updates.
668 :param new_obj: The object in the model that is created or updated
669 by applying this delta.
672 if new_obj
and not old_obj
:
676 'Model changed: %s %s %s',
677 delta
.entity
, delta
.type, delta
.get_id())
679 for o
in self
.observers
:
680 if o
.cares_about(delta
):
681 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
),
684 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
686 Block the calling routine until a given action has happened to the
689 :param entity_type: The entity's type.
690 :param entity_id: The entity's id.
691 :param action: the type of action (e.g., 'add', 'change', or 'remove')
692 :param predicate: optional callable that must take as an
693 argument a delta, and must return a boolean, indicating
694 whether the delta contains the specific action we're looking
695 for. For example, you might check to see whether a 'change'
696 has a 'completed' status. See the _Observer class for details.
699 q
= asyncio
.Queue(loop
=self
.loop
)
701 async def callback(delta
, old
, new
, model
):
702 await q
.put(delta
.get_id())
704 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
705 entity_id
= await q
.get()
706 # object might not be in the entity_map if we were waiting for a
708 return self
.state
._live
_entity
_map
(entity_type
).get(entity_id
)
710 async def _wait_for_new(self
, entity_type
, entity_id
=None, predicate
=None):
711 """Wait for a new object to appear in the Model and return it.
713 Waits for an object of type ``entity_type`` with id ``entity_id``.
714 If ``entity_id`` is ``None``, it will wait for the first new entity
717 This coroutine blocks until the new object appears in the model.
720 # if the entity is already in the model, just return it
721 if entity_id
in self
.state
._live
_entity
_map
(entity_type
):
722 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
723 # if we know the entity_id, we can trigger on any action that puts
724 # the enitty into the model; otherwise, we have to watch for the
725 # next "add" action on that entity_type
726 action
= 'add' if entity_id
is None else None
727 return await self
._wait
(entity_type
, entity_id
, action
, predicate
)
729 async def wait_for_action(self
, action_id
):
730 """Given an action, wait for it to complete."""
732 if action_id
.startswith("action-"):
733 # if we've been passed action.tag, transform it into the
734 # id that the api deltas will use.
735 action_id
= action_id
[7:]
737 def predicate(delta
):
738 return delta
.data
['status'] in ('completed', 'failed')
740 return await self
._wait
('action', action_id
, 'change', predicate
)
742 async def add_machine(
743 self
, spec
=None, constraints
=None, disks
=None, series
=None):
744 """Start a new, empty machine and optionally a container, or add a
745 container to a machine.
747 :param str spec: Machine specification
750 (None) - starts a new machine
751 'lxd' - starts a new machine with one lxd container
752 'lxd:4' - starts a new lxd container on machine 4
753 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
754 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
755 'maas2.name' - acquire machine maas2.name on MAAS
757 :param dict constraints: Machine constraints
764 :param list disks: List of disk constraint dictionaries
773 :param str series: Series, e.g. 'xenial'
775 Supported container types are: lxd, kvm
777 When deploying a container to an existing machine, constraints cannot
781 params
= client
.AddMachineParams()
782 params
.jobs
= ['JobHostUnits']
785 placement
= parse_placement(spec
)
787 params
.placement
= placement
[0]
790 params
.constraints
= client
.Value
.from_json(constraints
)
794 client
.Constraints
.from_json(o
) for o
in disks
]
797 params
.series
= series
799 # Submit the request.
800 client_facade
= client
.ClientFacade()
801 client_facade
.connect(self
.connection
)
802 results
= await client_facade
.AddMachines([params
])
803 error
= results
.machines
[0].error
805 raise ValueError("Error adding machine: %s" % error
.message
)
806 machine_id
= results
.machines
[0].machine
807 log
.debug('Added new machine %s', machine_id
)
808 return await self
._wait
_for
_new
('machine', machine_id
)
810 async def add_relation(self
, relation1
, relation2
):
811 """Add a relation between two applications.
813 :param str relation1: '<application>[:<relation_name>]'
814 :param str relation2: '<application>[:<relation_name>]'
817 app_facade
= client
.ApplicationFacade()
818 app_facade
.connect(self
.connection
)
821 'Adding relation %s <-> %s', relation1
, relation2
)
824 result
= await app_facade
.AddRelation([relation1
, relation2
])
825 except JujuAPIError
as e
:
826 if 'relation already exists' not in e
.message
:
829 'Relation %s <-> %s already exists', relation1
, relation2
)
830 # TODO: if relation already exists we should return the
831 # Relation ModelEntity here
834 def predicate(delta
):
836 for endpoint
in delta
.data
['endpoints']:
837 endpoints
[endpoint
['application-name']] = endpoint
['relation']
838 return endpoints
== result
.endpoints
840 return await self
._wait
_for
_new
('relation', None, predicate
)
842 def add_space(self
, name
, *cidrs
):
843 """Add a new network space.
845 Adds a new space with the given name and associates the given
846 (optional) list of existing subnet CIDRs with it.
848 :param str name: Name of the space
849 :param \*cidrs: Optional list of existing subnet CIDRs
852 raise NotImplementedError()
854 async def add_ssh_key(self
, user
, key
):
855 """Add a public SSH key to this model.
857 :param str user: The username of the user
858 :param str key: The public ssh key
861 key_facade
= client
.KeyManagerFacade()
862 key_facade
.connect(self
.connection
)
863 return await key_facade
.AddKeys([key
], user
)
864 add_ssh_keys
= add_ssh_key
866 def add_subnet(self
, cidr_or_id
, space
, *zones
):
867 """Add an existing subnet to this model.
869 :param str cidr_or_id: CIDR or provider ID of the existing subnet
870 :param str space: Network space with which to associate
871 :param str \*zones: Zone(s) in which the subnet resides
874 raise NotImplementedError()
876 def get_backups(self
):
877 """Retrieve metadata for backups in this model.
880 raise NotImplementedError()
882 def block(self
, *commands
):
883 """Add a new block to this model.
885 :param str \*commands: The commands to block. Valid values are
886 'all-changes', 'destroy-model', 'remove-object'
889 raise NotImplementedError()
891 def get_blocks(self
):
892 """List blocks for this model.
895 raise NotImplementedError()
897 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
898 """Return a list of cached OS images.
900 :param str arch: Filter by image architecture
901 :param str kind: Filter by image kind, e.g. 'lxd'
902 :param str series: Filter by image series, e.g. 'xenial'
905 raise NotImplementedError()
907 def create_backup(self
, note
=None, no_download
=False):
908 """Create a backup of this model.
910 :param str note: A note to store with the backup
911 :param bool no_download: Do not download the backup archive
912 :return str: Path to downloaded archive
915 raise NotImplementedError()
917 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
918 """Create or define a storage pool.
920 :param str name: Name to give the storage pool
921 :param str provider_type: Pool provider type
922 :param \*\*pool_config: key/value pool configuration pairs
925 raise NotImplementedError()
928 self
, no_tail
=False, exclude_module
=None, include_module
=None,
929 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
931 """Get log messages for this model.
933 :param bool no_tail: Stop after returning existing log messages
934 :param list exclude_module: Do not show log messages for these logging
936 :param list include_module: Only show log messages for these logging
938 :param list include: Only show log messages for these entities
939 :param str level: Log level to show, valid options are 'TRACE',
940 'DEBUG', 'INFO', 'WARNING', 'ERROR,
941 :param int limit: Return this many of the most recent (possibly
942 filtered) lines are shown
943 :param int lines: Yield this many of the most recent lines, and keep
945 :param bool replay: Yield the entire log, and keep yielding
946 :param list exclude: Do not show log messages for these entities
949 raise NotImplementedError()
951 def _get_series(self
, entity_url
, entity
):
952 # try to get the series from the provided charm URL
953 if entity_url
.startswith('cs:'):
954 parts
= entity_url
[3:].split('/')
956 parts
= entity_url
.split('/')
957 if parts
[0].startswith('~'):
960 # series was specified in the URL
962 # series was not supplied at all, so use the newest
963 # supported series according to the charm store
964 ss
= entity
['Meta']['supported-series']
965 return ss
['SupportedSeries'][0]
968 self
, entity_url
, application_name
=None, bind
=None, budget
=None,
969 channel
=None, config
=None, constraints
=None, force
=False,
970 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
972 """Deploy a new service or bundle.
974 :param str entity_url: Charm or bundle url
975 :param str application_name: Name to give the service
976 :param dict bind: <charm endpoint>:<network space> pairs
977 :param dict budget: <budget name>:<limit> pairs
978 :param str channel: Charm store channel from which to retrieve
979 the charm or bundle, e.g. 'development'
980 :param dict config: Charm configuration dictionary
981 :param constraints: Service constraints
982 :type constraints: :class:`juju.Constraints`
983 :param bool force: Allow charm to be deployed to a machine running
984 an unsupported series
985 :param int num_units: Number of units to deploy
986 :param str plan: Plan under which to deploy charm
987 :param dict resources: <resource name>:<file path> pairs
988 :param str series: Series on which to deploy
989 :param dict storage: Storage constraints TODO how do these look?
990 :param to: Placement directive as a string. For example:
992 '23' - place on machine 23
993 'lxd:7' - place in new lxd container on machine 7
994 '24/lxd/3' - place in container 3 on machine 24
996 If None, a new machine is provisioned.
1001 - support local resources
1006 k
: client
.Constraints(**v
)
1007 for k
, v
in storage
.items()
1011 entity_url
.startswith('local:') or
1012 os
.path
.isdir(entity_url
)
1015 entity_id
= entity_url
1017 entity
= await self
.charmstore
.entity(entity_url
)
1018 entity_id
= entity
['Id']
1020 client_facade
= client
.ClientFacade()
1021 client_facade
.connect(self
.connection
)
1023 is_bundle
= ((is_local
and
1024 (Path(entity_id
) / 'bundle.yaml').exists()) or
1025 (not is_local
and 'bundle/' in entity_id
))
1028 handler
= BundleHandler(self
)
1029 await handler
.fetch_plan(entity_id
)
1030 await handler
.execute_plan()
1031 extant_apps
= {app
for app
in self
.applications
}
1032 pending_apps
= set(handler
.applications
) - extant_apps
1034 # new apps will usually be in the model by now, but if some
1035 # haven't made it yet we'll need to wait on them to be added
1036 await asyncio
.gather(*[
1037 asyncio
.ensure_future(
1038 self
._wait
_for
_new
('application', app_name
),
1040 for app_name
in pending_apps
1042 return [app
for name
, app
in self
.applications
.items()
1043 if name
in handler
.applications
]
1046 if not application_name
:
1047 application_name
= entity
['Meta']['charm-metadata']['Name']
1049 series
= self
._get
_series
(entity_url
, entity
)
1052 await client_facade
.AddCharm(channel
, entity_id
)
1053 # XXX: we're dropping local resources here, but we don't
1054 # actually support them yet anyway
1055 resources
= await self
._add
_store
_resources
(application_name
,
1059 # We have a local charm dir that needs to be uploaded
1060 charm_dir
= os
.path
.abspath(
1061 os
.path
.expanduser(entity_id
))
1062 series
= series
or get_charm_series(charm_dir
)
1065 "Couldn't determine series for charm at {}. "
1066 "Pass a 'series' kwarg to Model.deploy().".format(
1068 entity_id
= await self
.add_local_charm_dir(charm_dir
, series
)
1069 return await self
._deploy
(
1070 charm_url
=entity_id
,
1071 application
=application_name
,
1073 config
=config
or {},
1074 constraints
=constraints
,
1075 endpoint_bindings
=bind
,
1076 resources
=resources
,
1079 num_units
=num_units
,
1080 placement
=parse_placement(to
)
1083 async def _add_store_resources(self
, application
, entity_url
, entity
=None):
1085 # avoid extra charm store call if one was already made
1086 entity
= await self
.charmstore
.entity(entity_url
)
1089 'description': resource
['Description'],
1090 'fingerprint': resource
['Fingerprint'],
1091 'name': resource
['Name'],
1092 'path': resource
['Path'],
1093 'revision': resource
['Revision'],
1094 'size': resource
['Size'],
1095 'type_': resource
['Type'],
1097 } for resource
in entity
['Meta']['resources']
1103 resources_facade
= client
.ResourcesFacade()
1104 resources_facade
.connect(self
.connection
)
1105 response
= await resources_facade
.AddPendingResources(
1106 tag
.application(application
),
1108 [client
.CharmResource(**resource
) for resource
in resources
])
1109 resource_map
= {resource
['name']: pid
1111 in zip(resources
, response
.pending_ids
)}
1114 async def _deploy(self
, charm_url
, application
, series
, config
,
1115 constraints
, endpoint_bindings
, resources
, storage
,
1116 channel
=None, num_units
=None, placement
=None):
1117 """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
1119 log
.info('Deploying %s', charm_url
)
1121 # stringify all config values for API, and convert to YAML
1122 config
= {k
: str(v
) for k
, v
in config
.items()}
1123 config
= yaml
.dump({application
: config
},
1124 default_flow_style
=False)
1126 app_facade
= client
.ApplicationFacade()
1127 app_facade
.connect(self
.connection
)
1129 app
= client
.ApplicationDeploy(
1130 charm_url
=charm_url
,
1131 application
=application
,
1135 constraints
=parse_constraints(constraints
),
1136 endpoint_bindings
=endpoint_bindings
,
1137 num_units
=num_units
,
1138 resources
=resources
,
1143 result
= await app_facade
.Deploy([app
])
1144 errors
= [r
.error
.message
for r
in result
.results
if r
.error
]
1146 raise JujuError('\n'.join(errors
))
1147 return await self
._wait
_for
_new
('application', application
)
1149 async def destroy(self
):
1150 """Terminate all machines and resources for this model.
1151 Is already implemented in controller.py.
1153 raise NotImplementedError()
1155 async def destroy_unit(self
, *unit_names
):
1156 """Destroy units by name.
1159 app_facade
= client
.ApplicationFacade()
1160 app_facade
.connect(self
.connection
)
1163 'Destroying unit%s %s',
1164 's' if len(unit_names
) == 1 else '',
1165 ' '.join(unit_names
))
1167 return await app_facade
.DestroyUnits(list(unit_names
))
1168 destroy_units
= destroy_unit
1170 def get_backup(self
, archive_id
):
1171 """Download a backup archive file.
1173 :param str archive_id: The id of the archive to download
1174 :return str: Path to the archive file
1177 raise NotImplementedError()
1180 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
1181 """Ensure sufficient controllers exist to provide redundancy.
1183 :param int num_controllers: Number of controllers to make available
1184 :param constraints: Constraints to apply to the controller machines
1185 :type constraints: :class:`juju.Constraints`
1186 :param str series: Series of the controller machines
1187 :param list to: Placement directives for controller machines, e.g.::
1190 'lxc:7' - new lxc container on machine 7
1191 '24/lxc/3' - lxc container 3 or machine 24
1193 If None, a new machine is provisioned.
1196 raise NotImplementedError()
1198 def get_config(self
):
1199 """Return the configuration settings for this model.
1202 raise NotImplementedError()
1204 def get_constraints(self
):
1205 """Return the machine constraints for this model.
1208 raise NotImplementedError()
1210 async def grant(self
, username
, acl
='read'):
1211 """Grant a user access to this model.
1213 :param str username: Username
1214 :param str acl: Access control ('read' or 'write')
1217 model_facade
= client
.ModelManagerFacade()
1218 controller_conn
= await self
.connection
.controller()
1219 model_facade
.connect(controller_conn
)
1220 user
= tag
.user(username
)
1221 model
= tag
.model(self
.info
.uuid
)
1222 changes
= client
.ModifyModelAccess(acl
, 'grant', model
, user
)
1223 await self
.revoke(username
)
1224 return await model_facade
.ModifyModelAccess([changes
])
1226 def import_ssh_key(self
, identity
):
1227 """Add a public SSH key from a trusted indentity source to this model.
1229 :param str identity: User identity in the form <lp|gh>:<username>
1232 raise NotImplementedError()
1233 import_ssh_keys
= import_ssh_key
1235 async def get_machines(self
):
1236 """Return list of machines in this model.
1239 return list(self
.state
.machines
.keys())
1241 def get_shares(self
):
1242 """Return list of all users with access to this model.
1245 raise NotImplementedError()
1247 def get_spaces(self
):
1248 """Return list of all known spaces, including associated subnets.
1251 raise NotImplementedError()
1253 async def get_ssh_key(self
, raw_ssh
=False):
1254 """Return known SSH keys for this model.
1255 :param bool raw_ssh: if True, returns the raw ssh key,
1256 else it's fingerprint
1259 key_facade
= client
.KeyManagerFacade()
1260 key_facade
.connect(self
.connection
)
1261 entity
= {'tag': tag
.model(self
.info
.uuid
)}
1262 entities
= client
.Entities([entity
])
1263 return await key_facade
.ListKeys(entities
, raw_ssh
)
1264 get_ssh_keys
= get_ssh_key
1266 def get_storage(self
, filesystem
=False, volume
=False):
1267 """Return details of storage instances.
1269 :param bool filesystem: Include filesystem storage
1270 :param bool volume: Include volume storage
1273 raise NotImplementedError()
1275 def get_storage_pools(self
, names
=None, providers
=None):
1276 """Return list of storage pools.
1278 :param list names: Only include pools with these names
1279 :param list providers: Only include pools for these providers
1282 raise NotImplementedError()
1284 def get_subnets(self
, space
=None, zone
=None):
1285 """Return list of known subnets.
1287 :param str space: Only include subnets in this space
1288 :param str zone: Only include subnets in this zone
1291 raise NotImplementedError()
1293 def remove_blocks(self
):
1294 """Remove all blocks from this model.
1297 raise NotImplementedError()
1299 def remove_backup(self
, backup_id
):
1302 :param str backup_id: The id of the backup to remove
1305 raise NotImplementedError()
1307 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1308 """Remove cached OS images.
1310 :param str arch: Architecture of the images to remove
1311 :param str kind: Image kind to remove, e.g. 'lxd'
1312 :param str series: Image series to remove, e.g. 'xenial'
1315 raise NotImplementedError()
1317 def remove_machine(self
, *machine_ids
):
1318 """Remove a machine from this model.
1320 :param str \*machine_ids: Ids of the machines to remove
1323 raise NotImplementedError()
1324 remove_machines
= remove_machine
1326 async def remove_ssh_key(self
, user
, key
):
1327 """Remove a public SSH key(s) from this model.
1329 :param str key: Full ssh key
1330 :param str user: Juju user to which the key is registered
1333 key_facade
= client
.KeyManagerFacade()
1334 key_facade
.connect(self
.connection
)
1335 key
= base64
.b64decode(bytes(key
.strip().split()[1].encode('ascii')))
1336 key
= hashlib
.md5(key
).hexdigest()
1337 key
= ':'.join(a
+b
for a
, b
in zip(key
[::2], key
[1::2]))
1338 await key_facade
.DeleteKeys([key
], user
)
1339 remove_ssh_keys
= remove_ssh_key
1342 self
, bootstrap
=False, constraints
=None, archive
=None,
1343 backup_id
=None, upload_tools
=False):
1344 """Restore a backup archive to a new controller.
1346 :param bool bootstrap: Bootstrap a new state machine
1347 :param constraints: Model constraints
1348 :type constraints: :class:`juju.Constraints`
1349 :param str archive: Path to backup archive to restore
1350 :param str backup_id: Id of backup to restore
1351 :param bool upload_tools: Upload tools if bootstrapping a new machine
1354 raise NotImplementedError()
1356 def retry_provisioning(self
):
1357 """Retry provisioning for failed machines.
1360 raise NotImplementedError()
1362 async def revoke(self
, username
):
1363 """Revoke a user's access to this model.
1365 :param str username: Username to revoke
1368 model_facade
= client
.ModelManagerFacade()
1369 controller_conn
= await self
.connection
.controller()
1370 model_facade
.connect(controller_conn
)
1371 user
= tag
.user(username
)
1372 model
= tag
.model(self
.info
.uuid
)
1373 changes
= client
.ModifyModelAccess('read', 'revoke', model
, user
)
1374 return await model_facade
.ModifyModelAccess([changes
])
1376 def run(self
, command
, timeout
=None):
1377 """Run command on all machines in this model.
1379 :param str command: The command to run
1380 :param int timeout: Time to wait before command is considered failed
1383 raise NotImplementedError()
1385 def set_config(self
, **config
):
1386 """Set configuration keys on this model.
1388 :param \*\*config: Config key/values
1391 raise NotImplementedError()
1393 def set_constraints(self
, constraints
):
1394 """Set machine constraints on this model.
1396 :param :class:`juju.Constraints` constraints: Machine constraints
1399 raise NotImplementedError()
1401 def get_action_output(self
, action_uuid
, wait
=-1):
1402 """Get the results of an action by ID.
1404 :param str action_uuid: Id of the action
1405 :param int wait: Time in seconds to wait for action to complete
1408 raise NotImplementedError()
1410 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1411 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1413 :param str uuid_or_prefix: Filter by action uuid or prefix
1414 :param str name: Filter by action name
1417 raise NotImplementedError()
1419 def get_budget(self
, budget_name
):
1420 """Get budget usage info.
1422 :param str budget_name: Name of budget
1425 raise NotImplementedError()
1427 async def get_status(self
, filters
=None, utc
=False):
1428 """Return the status of the model.
1430 :param str filters: Optional list of applications, units, or machines
1431 to include, which can use wildcards ('*').
1432 :param bool utc: Display time as UTC in RFC3339 format
1435 client_facade
= client
.ClientFacade()
1436 client_facade
.connect(self
.connection
)
1437 return await client_facade
.FullStatus(filters
)
1440 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1441 source
=None, stream
=None, version
=None):
1442 """Copy Juju tools into this model.
1444 :param bool all_: Copy all versions, not just the latest
1445 :param str destination: Path to local destination directory
1446 :param bool dry_run: Don't do the actual copy
1447 :param bool public: Tools are for a public cloud, so generate mirrors
1449 :param str source: Path to local source directory
1450 :param str stream: Simplestreams stream for which to sync metadata
1451 :param str version: Copy a specific major.minor version
1454 raise NotImplementedError()
1456 def unblock(self
, *commands
):
1457 """Unblock an operation that would alter this model.
1459 :param str \*commands: The commands to unblock. Valid values are
1460 'all-changes', 'destroy-model', 'remove-object'
1463 raise NotImplementedError()
1465 def unset_config(self
, *keys
):
1466 """Unset configuration on this model.
1468 :param str \*keys: The keys to unset
1471 raise NotImplementedError()
1473 def upgrade_gui(self
):
1474 """Upgrade the Juju GUI for this model.
1477 raise NotImplementedError()
1480 self
, dry_run
=False, reset_previous_upgrade
=False,
1481 upload_tools
=False, version
=None):
1482 """Upgrade Juju on all machines in a model.
1484 :param bool dry_run: Don't do the actual upgrade
1485 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1487 :param bool upload_tools: Upload local version of tools
1488 :param str version: Upgrade to a specific version
1491 raise NotImplementedError()
1493 def upload_backup(self
, archive_path
):
1494 """Store a backup archive remotely in Juju.
1496 :param str archive_path: Path to local archive
1499 raise NotImplementedError()
1502 def charmstore(self
):
1503 return self
._charmstore
1505 async def get_metrics(self
, *tags
):
1506 """Retrieve metrics.
1508 :param str \*tags: Tags of entities from which to retrieve metrics.
1509 No tags retrieves the metrics of all units in the model.
1510 :return: Dictionary of unit_name:metrics
1513 log
.debug("Retrieving metrics for %s",
1514 ', '.join(tags
) if tags
else "all units")
1516 metrics_facade
= client
.MetricsDebugFacade()
1517 metrics_facade
.connect(self
.connection
)
1519 entities
= [client
.Entity(tag
) for tag
in tags
]
1520 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1522 metrics
= collections
.defaultdict(list)
1524 for entity_metrics
in metrics_result
.results
:
1525 error
= entity_metrics
.error
1527 if "is not a valid tag" in error
:
1528 raise ValueError(error
.message
)
1530 raise Exception(error
.message
)
1532 for metric
in entity_metrics
.metrics
:
1533 metrics
[metric
.unit
].append(vars(metric
))
1538 def get_charm_series(path
):
1539 """Inspects the charm directory at ``path`` and returns a default
1540 series from its metadata.yaml (the first item in the 'series' list).
1542 Returns None if no series can be determined.
1545 md
= Path(path
) / "metadata.yaml"
1548 data
= yaml
.load(md
.open())
1549 series
= data
.get('series')
1550 return series
[0] if series
else None
1553 class BundleHandler(object):
1555 Handle bundles by using the API to translate bundle YAML into a plan of
1556 steps and then dispatching each of those using the API.
1558 def __init__(self
, model
):
1560 self
.charmstore
= model
.charmstore
1562 self
.references
= {}
1563 self
._units
_by
_app
= {}
1564 for unit_name
, unit
in model
.units
.items():
1565 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1566 app_units
.append(unit_name
)
1567 self
.client_facade
= client
.ClientFacade()
1568 self
.client_facade
.connect(model
.connection
)
1569 self
.app_facade
= client
.ApplicationFacade()
1570 self
.app_facade
.connect(model
.connection
)
1571 self
.ann_facade
= client
.AnnotationsFacade()
1572 self
.ann_facade
.connect(model
.connection
)
1574 async def _handle_local_charms(self
, bundle
):
1575 """Search for references to local charms (i.e. filesystem paths)
1576 in the bundle. Upload the local charms to the model, and replace
1577 the filesystem paths with appropriate 'local:' paths in the bundle.
1579 Return the modified bundle.
1581 :param dict bundle: Bundle dictionary
1582 :return: Modified bundle dictionary
1587 default_series
= bundle
.get('series')
1588 for app_name
in self
.applications
:
1589 app_dict
= bundle
['services'][app_name
]
1590 charm_dir
= os
.path
.abspath(os
.path
.expanduser(app_dict
['charm']))
1591 if not os
.path
.isdir(charm_dir
):
1594 app_dict
.get('series') or
1596 get_charm_series(charm_dir
)
1600 "Couldn't determine series for charm at {}. "
1601 "Add a 'series' key to the bundle.".format(charm_dir
))
1603 # Keep track of what we need to update. We keep a list of apps
1604 # that need to be updated, and a corresponding list of args
1605 # needed to update those apps.
1606 apps
.append(app_name
)
1607 args
.append((charm_dir
, series
))
1610 # If we have apps to update, spawn all the coroutines concurrently
1611 # and wait for them to finish.
1612 charm_urls
= await asyncio
.gather(*[
1613 self
.model
.add_local_charm_dir(*params
)
1615 ], loop
=self
.model
.loop
)
1616 # Update the 'charm:' entry for each app with the new 'local:' url.
1617 for app_name
, charm_url
in zip(apps
, charm_urls
):
1618 bundle
['services'][app_name
]['charm'] = charm_url
1622 async def fetch_plan(self
, entity_id
):
1623 is_local
= not entity_id
.startswith('cs:') and os
.path
.isdir(entity_id
)
1625 bundle_yaml
= (Path(entity_id
) / "bundle.yaml").read_text()
1627 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1628 filename
='bundle.yaml',
1630 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1631 self
.bundle
= await self
._handle
_local
_charms
(self
.bundle
)
1633 self
.plan
= await self
.client_facade
.GetBundleChanges(
1634 yaml
.dump(self
.bundle
))
1636 async def execute_plan(self
):
1637 for step
in self
.plan
.changes
:
1638 method
= getattr(self
, step
.method
)
1639 result
= await method(*step
.args
)
1640 self
.references
[step
.id_
] = result
1643 def applications(self
):
1644 return list(self
.bundle
['services'].keys())
1646 def resolve(self
, reference
):
1647 if reference
and reference
.startswith('$'):
1648 reference
= self
.references
[reference
[1:]]
1651 async def addCharm(self
, charm
, series
):
1653 :param charm string:
1654 Charm holds the URL of the charm to be added.
1656 :param series string:
1657 Series holds the series of the charm to be added
1658 if the charm default is not sufficient.
1660 # We don't add local charms because they've already been added
1661 # by self._handle_local_charms
1662 if charm
.startswith('local:'):
1665 entity_id
= await self
.charmstore
.entityId(charm
)
1666 log
.debug('Adding %s', entity_id
)
1667 await self
.client_facade
.AddCharm(None, entity_id
)
1670 async def addMachines(self
, params
=None):
1673 Dictionary specifying the machine to add. All keys are optional.
1676 series: string specifying the machine OS series.
1678 constraints: string holding machine constraints, if any. We'll
1679 parse this into the json friendly dict that the juju api
1682 container_type: string holding the type of the container (for
1683 instance ""lxd" or kvm"). It is not specified for top level
1686 parent_id: string holding a placeholder pointing to another
1687 machine change or to a unit change. This value is only
1688 specified in the case this machine is a container, in
1689 which case also ContainerType is set.
1692 params
= params
or {}
1695 params
= {normalize_key(k
): params
[k
] for k
in params
.keys()}
1697 # Fix up values, as necessary.
1698 if 'parent_id' in params
:
1699 params
['parent_id'] = self
.resolve(params
['parent_id'])
1701 params
['constraints'] = parse_constraints(
1702 params
.get('constraints'))
1703 params
['jobs'] = params
.get('jobs', ['JobHostUnits'])
1705 if params
.get('container_type') == 'lxc':
1706 log
.warning('Juju 2.0 does not support lxc containers. '
1707 'Converting containers to lxd.')
1708 params
['container_type'] = 'lxd'
1710 # Submit the request.
1711 params
= client
.AddMachineParams(**params
)
1712 results
= await self
.client_facade
.AddMachines([params
])
1713 error
= results
.machines
[0].error
1715 raise ValueError("Error adding machine: %s" % error
.message
)
1716 machine
= results
.machines
[0].machine
1717 log
.debug('Added new machine %s', machine
)
1720 async def addRelation(self
, endpoint1
, endpoint2
):
1722 :param endpoint1 string:
1723 :param endpoint2 string:
1724 Endpoint1 and Endpoint2 hold relation endpoints in the
1725 "application:interface" form, where the application is always a
1726 placeholder pointing to an application change, and the interface is
1727 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1729 endpoints
= [endpoint1
, endpoint2
]
1730 # resolve indirect references
1731 for i
in range(len(endpoints
)):
1732 parts
= endpoints
[i
].split(':')
1733 parts
[0] = self
.resolve(parts
[0])
1734 endpoints
[i
] = ':'.join(parts
)
1736 log
.info('Relating %s <-> %s', *endpoints
)
1737 return await self
.model
.add_relation(*endpoints
)
1739 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1740 storage
, endpoint_bindings
, resources
):
1742 :param charm string:
1743 Charm holds the URL of the charm to be used to deploy this
1746 :param series string:
1747 Series holds the series of the application to be deployed
1748 if the charm default is not sufficient.
1750 :param application string:
1751 Application holds the application name.
1753 :param options map[string]interface{}:
1754 Options holds application options.
1756 :param constraints string:
1757 Constraints holds the optional application constraints.
1759 :param storage map[string]string:
1760 Storage holds the optional storage constraints.
1762 :param endpoint_bindings map[string]string:
1763 EndpointBindings holds the optional endpoint bindings
1765 :param resources map[string]int:
1766 Resources identifies the revision to use for each resource
1767 of the application's charm.
1769 # resolve indirect references
1770 charm
= self
.resolve(charm
)
1771 # the bundle plan doesn't actually do anything with resources, even
1772 # though it ostensibly gives us something (None) for that param
1773 if not charm
.startswith('local:'):
1774 resources
= await self
.model
._add
_store
_resources
(application
,
1776 await self
.model
._deploy
(
1778 application
=application
,
1781 constraints
=constraints
,
1782 endpoint_bindings
=endpoint_bindings
,
1783 resources
=resources
,
1788 async def addUnit(self
, application
, to
):
1790 :param application string:
1791 Application holds the application placeholder name for which a unit
1795 To holds the optional location where to add the unit, as a
1796 placeholder pointing to another unit change or to a machine change.
1798 application
= self
.resolve(application
)
1799 placement
= self
.resolve(to
)
1800 if self
._units
_by
_app
.get(application
):
1801 # enough units for this application already exist;
1802 # claim one, and carry on
1803 # NB: this should probably honor placement, but the juju client
1804 # doesn't, so we're not bothering, either
1805 unit_name
= self
._units
_by
_app
[application
].pop()
1806 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1807 return self
.model
.units
[unit_name
]
1809 log
.debug('Adding new unit for %s%s', application
,
1810 ' to %s' % placement
if placement
else '')
1811 return await self
.model
.applications
[application
].add_unit(
1816 async def expose(self
, application
):
1818 :param application string:
1819 Application holds the placeholder name of the application that must
1822 application
= self
.resolve(application
)
1823 log
.info('Exposing %s', application
)
1824 return await self
.model
.applications
[application
].expose()
1826 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1829 Id is the placeholder for the application or machine change
1830 corresponding to the entity to be annotated.
1832 :param entity_type EntityType:
1833 EntityType holds the type of the entity, "application" or
1836 :param annotations map[string]string:
1837 Annotations holds the annotations as key/value pairs.
1839 entity_id
= self
.resolve(id_
)
1841 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
1843 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
1844 return await entity
.set_annotations(annotations
)
1847 class CharmStore(object):
1849 Async wrapper around theblues.charmstore.CharmStore
1851 def __init__(self
, loop
):
1853 self
._cs
= theblues
.charmstore
.CharmStore(timeout
=5)
1855 def __getattr__(self
, name
):
1857 Wrap method calls in coroutines that use run_in_executor to make them
1860 attr
= getattr(self
._cs
, name
)
1861 if not callable(attr
):
1862 wrapper
= partial(getattr, self
._cs
, name
)
1863 setattr(self
, name
, wrapper
)
1865 async def coro(*args
, **kwargs
):
1866 method
= partial(attr
, *args
, **kwargs
)
1867 for attempt
in range(1, 4):
1869 return await self
.loop
.run_in_executor(None, method
)
1870 except theblues
.errors
.ServerError
:
1873 await asyncio
.sleep(1, loop
=self
.loop
)
1874 setattr(self
, name
, coro
)
1879 class CharmArchiveGenerator(object):
1880 def __init__(self
, path
):
1881 self
.path
= os
.path
.abspath(os
.path
.expanduser(path
))
1883 def make_archive(self
, path
):
1884 """Create archive of directory and write to ``path``.
1886 :param path: Path to archive
1890 * build/\* - This is used for packing the charm itself and any
1892 * \*/.\* - Hidden files are all ignored for now. This will most
1893 likely be changed into a specific ignore list
1897 zf
= zipfile
.ZipFile(path
, 'w', zipfile
.ZIP_DEFLATED
)
1898 for dirpath
, dirnames
, filenames
in os
.walk(self
.path
):
1899 relative_path
= dirpath
[len(self
.path
) + 1:]
1900 if relative_path
and not self
._ignore
(relative_path
):
1901 zf
.write(dirpath
, relative_path
)
1902 for name
in filenames
:
1903 archive_name
= os
.path
.join(relative_path
, name
)
1904 if not self
._ignore
(archive_name
):
1905 real_path
= os
.path
.join(dirpath
, name
)
1906 self
._check
_type
(real_path
)
1907 if os
.path
.islink(real_path
):
1908 self
._check
_link
(real_path
)
1909 self
._write
_symlink
(
1910 zf
, os
.readlink(real_path
), archive_name
)
1912 zf
.write(real_path
, archive_name
)
1916 def _check_type(self
, path
):
1920 if stat
.S_ISDIR(s
.st_mode
) or stat
.S_ISREG(s
.st_mode
):
1922 raise ValueError("Invalid Charm at % %s" % (
1923 path
, "Invalid file type for a charm"))
1925 def _check_link(self
, path
):
1926 link_path
= os
.readlink(path
)
1927 if link_path
[0] == "/":
1929 "Invalid Charm at %s: %s" % (
1930 path
, "Absolute links are invalid"))
1931 path_dir
= os
.path
.dirname(path
)
1932 link_path
= os
.path
.join(path_dir
, link_path
)
1933 if not link_path
.startswith(os
.path
.abspath(self
.path
)):
1935 "Invalid charm at %s %s" % (
1936 path
, "Only internal symlinks are allowed"))
1938 def _write_symlink(self
, zf
, link_target
, link_path
):
1939 """Package symlinks with appropriate zipfile metadata."""
1940 info
= zipfile
.ZipInfo()
1941 info
.filename
= link_path
1942 info
.create_system
= 3
1943 # Magic code for symlinks / py2/3 compat
1944 # 27166663808 = (stat.S_IFLNK | 0755) << 16
1945 info
.external_attr
= 2716663808
1946 zf
.writestr(info
, link_target
)
1948 def _ignore(self
, path
):
1949 if path
== "build" or path
.startswith("build/"):
1951 if path
.startswith('.'):