13 from concurrent
.futures
import CancelledError
14 from functools
import partial
15 from pathlib
import Path
18 import theblues
.charmstore
19 import theblues
.errors
21 from . import tag
, utils
22 from .client
import client
23 from .client
import connection
24 from .constraints
import parse
as parse_constraints
, normalize_key
25 from .delta
import get_entity_delta
26 from .delta
import get_entity_class
27 from .exceptions
import DeadEntityException
28 from .errors
import JujuError
, JujuAPIError
29 from .placement
import parse
as parse_placement
31 log
= logging
.getLogger(__name__
)
34 class _Observer(object):
35 """Wrapper around an observer callable.
37 This wrapper allows filter criteria to be associated with the
38 callable so that it's only called for changes that meet the criteria.
41 def __init__(self
, callable_
, entity_type
, action
, entity_id
, predicate
):
42 self
.callable_
= callable_
43 self
.entity_type
= entity_type
45 self
.entity_id
= entity_id
46 self
.predicate
= predicate
48 self
.entity_id
= str(self
.entity_id
)
49 if not self
.entity_id
.startswith('^'):
50 self
.entity_id
= '^' + self
.entity_id
51 if not self
.entity_id
.endswith('$'):
54 async def __call__(self
, delta
, old
, new
, model
):
55 await self
.callable_(delta
, old
, new
, model
)
57 def cares_about(self
, delta
):
58 """Return True if this observer "cares about" (i.e. wants to be
59 called) for a this delta.
62 if (self
.entity_id
and delta
.get_id() and
63 not re
.match(self
.entity_id
, str(delta
.get_id()))):
66 if self
.entity_type
and self
.entity_type
!= delta
.entity
:
69 if self
.action
and self
.action
!= delta
.type:
72 if self
.predicate
and not self
.predicate(delta
):
78 class ModelObserver(object):
80 Base class for creating observers that react to changes in a model.
82 async def __call__(self
, delta
, old
, new
, model
):
83 handler_name
= 'on_{}_{}'.format(delta
.entity
, delta
.type)
84 method
= getattr(self
, handler_name
, self
.on_change
)
85 await method(delta
, old
, new
, model
)
87 async def on_change(self
, delta
, old
, new
, model
):
88 """Generic model-change handler.
90 This should be overridden in a subclass.
92 :param delta: :class:`juju.client.overrides.Delta`
93 :param old: :class:`juju.model.ModelEntity`
94 :param new: :class:`juju.model.ModelEntity`
95 :param model: :class:`juju.model.Model`
101 class ModelState(object):
102 """Holds the state of the model, including the delta history of all
103 entities in the model.
106 def __init__(self
, model
):
110 def _live_entity_map(self
, entity_type
):
111 """Return an id:Entity map of all the living entities of
112 type ``entity_type``.
116 entity_id
: self
.get_entity(entity_type
, entity_id
)
117 for entity_id
, history
in self
.state
.get(entity_type
, {}).items()
118 if history
[-1] is not None
122 def applications(self
):
123 """Return a map of application-name:Application for all applications
124 currently in the model.
127 return self
._live
_entity
_map
('application')
131 """Return a map of machine-id:Machine for all machines currently in
135 return self
._live
_entity
_map
('machine')
139 """Return a map of unit-id:Unit for all units currently in
143 return self
._live
_entity
_map
('unit')
145 def entity_history(self
, entity_type
, entity_id
):
146 """Return the history deque for an entity.
149 return self
.state
[entity_type
][entity_id
]
151 def entity_data(self
, entity_type
, entity_id
, history_index
):
152 """Return the data dict for an entity at a specific index of its
156 return self
.entity_history(entity_type
, entity_id
)[history_index
]
158 def apply_delta(self
, delta
):
159 """Apply delta to our state and return a copy of the
160 affected object as it was before and after the update, e.g.:
162 old_obj, new_obj = self.apply_delta(delta)
164 old_obj may be None if the delta is for the creation of a new object,
165 e.g. a new application or unit is deployed.
167 new_obj will never be None, but may be dead (new_obj.dead == True)
168 if the object was deleted as a result of the delta being applied.
173 .setdefault(delta
.entity
, {})
174 .setdefault(delta
.get_id(), collections
.deque())
177 history
.append(delta
.data
)
178 if delta
.type == 'remove':
181 entity
= self
.get_entity(delta
.entity
, delta
.get_id())
182 return entity
.previous(), entity
185 self
, entity_type
, entity_id
, history_index
=-1, connected
=True):
186 """Return an object instance for the given entity_type and id.
188 By default the object state matches the most recent state from
189 Juju. To get an instance of the object in an older state, pass
190 history_index, an index into the history deque for the entity.
194 if history_index
< 0 and history_index
!= -1:
195 history_index
+= len(self
.entity_history(entity_type
, entity_id
))
196 if history_index
< 0:
200 self
.entity_data(entity_type
, entity_id
, history_index
)
204 entity_class
= get_entity_class(entity_type
)
206 entity_id
, self
.model
, history_index
=history_index
,
210 class ModelEntity(object):
211 """An object in the Model tree"""
213 def __init__(self
, entity_id
, model
, history_index
=-1, connected
=True):
214 """Initialize a new entity
216 :param entity_id str: The unique id of the object in the model
217 :param model: The model instance in whose object tree this
219 :history_index int: The index of this object's state in the model's
220 history deque for this entity
221 :connected bool: Flag indicating whether this object gets live updates
225 self
.entity_id
= entity_id
227 self
._history
_index
= history_index
228 self
.connected
= connected
229 self
.connection
= model
.connection
232 return '<{} entity_id="{}">'.format(type(self
).__name
__,
235 def __getattr__(self
, name
):
236 """Fetch object attributes from the underlying data dict held in the
241 return self
.safe_data
[name
]
243 name
= name
.replace('_', '-')
244 if name
in self
.safe_data
:
245 return self
.safe_data
[name
]
250 return bool(self
.data
)
252 def on_change(self
, callable_
):
253 """Add a change observer to this entity.
256 self
.model
.add_observer(
257 callable_
, self
.entity_type
, 'change', self
.entity_id
)
259 def on_remove(self
, callable_
):
260 """Add a remove observer to this entity.
263 self
.model
.add_observer(
264 callable_
, self
.entity_type
, 'remove', self
.entity_id
)
267 def entity_type(self
):
268 """A string identifying the entity type of this object, e.g.
269 'application' or 'unit', etc.
272 return self
.__class
__.__name
__.lower()
276 """Return True if this object represents the current state of the
277 entity in the underlying model.
279 This will be True except when the object represents an entity at a
280 non-latest state in history, e.g. if the object was obtained by calling
281 .previous() on another object.
284 return self
._history
_index
== -1
288 """Returns True if this entity no longer exists in the underlying
294 self
.model
.state
.entity_data(
295 self
.entity_type
, self
.entity_id
, -1) is None
300 """Returns True if this entity still exists in the underlying
308 """The data dictionary for this entity.
311 return self
.model
.state
.entity_data(
312 self
.entity_type
, self
.entity_id
, self
._history
_index
)
316 """The data dictionary for this entity.
318 If this `ModelEntity` points to the dead state, it will
319 raise `DeadEntityException`.
322 if self
.data
is None:
323 raise DeadEntityException(
324 "Entity {}:{} is dead - its attributes can no longer be "
325 "accessed. Use the .previous() method on this object to get "
326 "a copy of the object at its previous state.".format(
327 self
.entity_type
, self
.entity_id
))
331 """Return a copy of this object as was at its previous state in
334 Returns None if this object is new (and therefore has no history).
336 The returned object is always "disconnected", i.e. does not receive
340 return self
.model
.state
.get_entity(
341 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
345 """Return a copy of this object at its next state in
348 Returns None if this object is already the latest.
350 The returned object is "disconnected", i.e. does not receive
351 live updates, unless it is current (latest).
354 if self
._history
_index
== -1:
357 new_index
= self
._history
_index
+ 1
359 new_index
== len(self
.model
.state
.entity_history(
360 self
.entity_type
, self
.entity_id
)) - 1
362 return self
.model
.state
.get_entity(
363 self
.entity_type
, self
.entity_id
, self
._history
_index
- 1,
367 """Return a copy of this object at its current state in the model.
369 Returns self if this object is already the latest.
371 The returned object is always "connected", i.e. receives
372 live updates from the model.
375 if self
._history
_index
== -1:
378 return self
.model
.state
.get_entity(self
.entity_type
, self
.entity_id
)
383 The main API for interacting with a Juju model.
385 def __init__(self
, loop
=None):
386 """Instantiate a new connected Model.
388 :param loop: an asyncio event loop
391 self
.loop
= loop
or asyncio
.get_event_loop()
392 self
.connection
= None
393 self
.observers
= weakref
.WeakValueDictionary()
394 self
.state
= ModelState(self
)
396 self
._watch
_stopping
= asyncio
.Event(loop
=self
.loop
)
397 self
._watch
_stopped
= asyncio
.Event(loop
=self
.loop
)
398 self
._watch
_received
= asyncio
.Event(loop
=self
.loop
)
399 self
._charmstore
= CharmStore(self
.loop
)
401 async def __aenter__(self
):
402 await self
.connect_current()
405 async def __aexit__(self
, exc_type
, exc
, tb
):
406 await self
.disconnect()
408 if exc_type
is not None:
411 async def connect(self
, *args
, **kw
):
412 """Connect to an arbitrary Juju model.
414 args and kw are passed through to Connection.connect()
418 kw
['loop'] = self
.loop
419 self
.connection
= await connection
.Connection
.connect(*args
, **kw
)
420 await self
._after
_connect
()
422 async def connect_current(self
):
423 """Connect to the current Juju model.
426 self
.connection
= await connection
.Connection
.connect_current(
428 await self
._after
_connect
()
430 async def connect_model(self
, model_name
):
431 """Connect to a specific Juju model by name.
433 :param model_name: Format [controller:][user/]model
436 self
.connection
= await connection
.Connection
.connect_model(model_name
,
438 await self
._after
_connect
()
440 async def _after_connect(self
):
441 """Run initialization steps after connecting to websocket.
445 await self
._watch
_received
.wait()
446 await self
.get_info()
448 async def disconnect(self
):
449 """Shut down the watcher task and close websockets.
452 if self
.connection
and self
.connection
.is_open
:
453 log
.debug('Stopping watcher task')
454 self
._watch
_stopping
.set()
455 await self
._watch
_stopped
.wait()
456 log
.debug('Closing model connection')
457 await self
.connection
.close()
458 self
.connection
= None
460 async def add_local_charm_dir(self
, charm_dir
, series
):
461 """Upload a local charm to the model.
463 This will automatically generate an archive from
466 :param charm_dir: Path to the charm directory
467 :param series: Charm series
470 fh
= tempfile
.NamedTemporaryFile()
471 CharmArchiveGenerator(charm_dir
).make_archive(fh
.name
)
474 self
.add_local_charm
, fh
, series
, os
.stat(fh
.name
).st_size
)
475 charm_url
= await self
.loop
.run_in_executor(None, func
)
477 log
.debug('Uploaded local charm: %s -> %s', charm_dir
, charm_url
)
480 def add_local_charm(self
, charm_file
, series
, size
=None):
481 """Upload a local charm archive to the model.
483 Returns the 'local:...' url that should be used to deploy the charm.
485 :param charm_file: Path to charm zip archive
486 :param series: Charm series
487 :param size: Size of the archive, in bytes
488 :return str: 'local:...' url for deploying the charm
489 :raises: :class:`JujuError` if the upload fails
491 Uses an https endpoint at the same host:port as the wss.
492 Supports large file uploads.
496 This method will block. Consider using :meth:`add_local_charm_dir`
500 conn
, headers
, path_prefix
= self
.connection
.https_connection()
501 path
= "%s/charms?series=%s" % (path_prefix
, series
)
502 headers
['Content-Type'] = 'application/zip'
504 headers
['Content-Length'] = size
505 conn
.request("POST", path
, charm_file
, headers
)
506 response
= conn
.getresponse()
507 result
= response
.read().decode()
508 if not response
.status
== 200:
509 raise JujuError(result
)
510 result
= json
.loads(result
)
511 return result
['charm-url']
513 def all_units_idle(self
):
514 """Return True if all units are idle.
517 for unit
in self
.units
.values():
518 unit_status
= unit
.data
['agent-status']['current']
519 if unit_status
!= 'idle':
523 async def reset(self
, force
=False):
524 """Reset the model to a clean state.
526 :param bool force: Force-terminate machines.
528 This returns only after the model has reached a clean state. "Clean"
529 means no applications or machines exist in the model.
532 log
.debug('Resetting model')
533 for app
in self
.applications
.values():
535 for machine
in self
.machines
.values():
536 await machine
.destroy(force
=force
)
537 await self
.block_until(
538 lambda: len(self
.machines
) == 0
541 async def block_until(self
, *conditions
, timeout
=None, wait_period
=0.5):
542 """Return only after all conditions are true.
546 while not all(c() for c
in conditions
):
547 await asyncio
.sleep(wait_period
, loop
=self
.loop
)
548 await asyncio
.wait_for(_block(), timeout
, loop
=self
.loop
)
551 def applications(self
):
552 """Return a map of application-name:Application for all applications
553 currently in the model.
556 return self
.state
.applications
560 """Return a map of machine-id:Machine for all machines currently in
564 return self
.state
.machines
568 """Return a map of unit-id:Unit for all units currently in
572 return self
.state
.units
574 async def get_info(self
):
575 """Return a client.ModelInfo object for this Model.
577 Retrieves latest info for this Model from the api server. The
578 return value is cached on the Model.info attribute so that the
579 valued may be accessed again without another api call, if
582 This method is called automatically when the Model is connected,
583 resulting in Model.info being initialized without requiring an
584 explicit call to this method.
587 facade
= client
.ClientFacade
.from_connection(self
.connection
)
589 self
.info
= await facade
.ModelInfo()
590 log
.debug('Got ModelInfo: %s', vars(self
.info
))
595 self
, callable_
, entity_type
=None, action
=None, entity_id
=None,
597 """Register an "on-model-change" callback
599 Once the model is connected, ``callable_``
600 will be called each time the model changes. ``callable_`` should
601 be Awaitable and accept the following positional arguments:
603 delta - An instance of :class:`juju.delta.EntityDelta`
604 containing the raw delta data recv'd from the Juju
607 old_obj - If the delta modifies an existing object in the model,
608 old_obj will be a copy of that object, as it was before the
609 delta was applied. Will be None if the delta creates a new
612 new_obj - A copy of the new or updated object, after the delta
613 is applied. Will be None if the delta removes an entity
616 model - The :class:`Model` itself.
618 Events for which ``callable_`` is called can be specified by passing
619 entity_type, action, and/or entitiy_id filter criteria, e.g.::
623 entity_type='application', action='add', entity_id='ubuntu')
625 For more complex filtering conditions, pass a predicate function. It
626 will be called with a delta as its only argument. If the predicate
627 function returns True, the ``callable_`` will be called.
630 observer
= _Observer(
631 callable_
, entity_type
, action
, entity_id
, predicate
)
632 self
.observers
[observer
] = callable_
635 """Start an asynchronous watch against this model.
637 See :meth:`add_observer` to register an onchange callback.
640 async def _start_watch():
642 allwatcher
= client
.AllWatcherFacade
.from_connection(
644 while not self
._watch
_stopping
.is_set():
645 results
= await utils
.run_with_interrupt(
647 self
._watch
_stopping
,
649 if self
._watch
_stopping
.is_set():
651 for delta
in results
.deltas
:
652 delta
= get_entity_delta(delta
)
653 old_obj
, new_obj
= self
.state
.apply_delta(delta
)
654 await self
._notify
_observers
(delta
, old_obj
, new_obj
)
655 self
._watch
_received
.set()
656 except CancelledError
:
659 log
.exception('Error in watcher')
662 self
._watch
_stopped
.set()
664 log
.debug('Starting watcher task')
665 self
._watch
_received
.clear()
666 self
._watch
_stopping
.clear()
667 self
._watch
_stopped
.clear()
668 self
.loop
.create_task(_start_watch())
670 async def _notify_observers(self
, delta
, old_obj
, new_obj
):
671 """Call observing callbacks, notifying them of a change in model state
673 :param delta: The raw change from the watcher
674 (:class:`juju.client.overrides.Delta`)
675 :param old_obj: The object in the model that this delta updates.
677 :param new_obj: The object in the model that is created or updated
678 by applying this delta.
681 if new_obj
and not old_obj
:
685 'Model changed: %s %s %s',
686 delta
.entity
, delta
.type, delta
.get_id())
688 for o
in self
.observers
:
689 if o
.cares_about(delta
):
690 asyncio
.ensure_future(o(delta
, old_obj
, new_obj
, self
),
693 async def _wait(self
, entity_type
, entity_id
, action
, predicate
=None):
695 Block the calling routine until a given action has happened to the
698 :param entity_type: The entity's type.
699 :param entity_id: The entity's id.
700 :param action: the type of action (e.g., 'add', 'change', or 'remove')
701 :param predicate: optional callable that must take as an
702 argument a delta, and must return a boolean, indicating
703 whether the delta contains the specific action we're looking
704 for. For example, you might check to see whether a 'change'
705 has a 'completed' status. See the _Observer class for details.
708 q
= asyncio
.Queue(loop
=self
.loop
)
710 async def callback(delta
, old
, new
, model
):
711 await q
.put(delta
.get_id())
713 self
.add_observer(callback
, entity_type
, action
, entity_id
, predicate
)
714 entity_id
= await q
.get()
715 # object might not be in the entity_map if we were waiting for a
717 return self
.state
._live
_entity
_map
(entity_type
).get(entity_id
)
719 async def _wait_for_new(self
, entity_type
, entity_id
=None, predicate
=None):
720 """Wait for a new object to appear in the Model and return it.
722 Waits for an object of type ``entity_type`` with id ``entity_id``.
723 If ``entity_id`` is ``None``, it will wait for the first new entity
726 This coroutine blocks until the new object appears in the model.
729 # if the entity is already in the model, just return it
730 if entity_id
in self
.state
._live
_entity
_map
(entity_type
):
731 return self
.state
._live
_entity
_map
(entity_type
)[entity_id
]
732 # if we know the entity_id, we can trigger on any action that puts
733 # the enitty into the model; otherwise, we have to watch for the
734 # next "add" action on that entity_type
735 action
= 'add' if entity_id
is None else None
736 return await self
._wait
(entity_type
, entity_id
, action
, predicate
)
738 async def wait_for_action(self
, action_id
):
739 """Given an action, wait for it to complete."""
741 if action_id
.startswith("action-"):
742 # if we've been passed action.tag, transform it into the
743 # id that the api deltas will use.
744 action_id
= action_id
[7:]
746 def predicate(delta
):
747 return delta
.data
['status'] in ('completed', 'failed')
749 return await self
._wait
('action', action_id
, 'change', predicate
)
751 async def add_machine(
752 self
, spec
=None, constraints
=None, disks
=None, series
=None):
753 """Start a new, empty machine and optionally a container, or add a
754 container to a machine.
756 :param str spec: Machine specification
759 (None) - starts a new machine
760 'lxd' - starts a new machine with one lxd container
761 'lxd:4' - starts a new lxd container on machine 4
762 'ssh:user@10.10.0.3' - manually provisions a machine with ssh
763 'zone=us-east-1a' - starts a machine in zone us-east-1s on AWS
764 'maas2.name' - acquire machine maas2.name on MAAS
766 :param dict constraints: Machine constraints, which can contain the
787 :param list disks: List of disk constraint dictionaries, which can
788 contain the following keys::
802 :param str series: Series, e.g. 'xenial'
804 Supported container types are: lxd, kvm
806 When deploying a container to an existing machine, constraints cannot
810 params
= client
.AddMachineParams()
811 params
.jobs
= ['JobHostUnits']
814 placement
= parse_placement(spec
)
816 params
.placement
= placement
[0]
819 params
.constraints
= client
.Value
.from_json(constraints
)
823 client
.Constraints
.from_json(o
) for o
in disks
]
826 params
.series
= series
828 # Submit the request.
829 client_facade
= client
.ClientFacade
.from_connection(self
.connection
)
830 results
= await client_facade
.AddMachines([params
])
831 error
= results
.machines
[0].error
833 raise ValueError("Error adding machine: %s" % error
.message
)
834 machine_id
= results
.machines
[0].machine
835 log
.debug('Added new machine %s', machine_id
)
836 return await self
._wait
_for
_new
('machine', machine_id
)
838 async def add_relation(self
, relation1
, relation2
):
839 """Add a relation between two applications.
841 :param str relation1: '<application>[:<relation_name>]'
842 :param str relation2: '<application>[:<relation_name>]'
845 app_facade
= client
.ApplicationFacade
.from_connection(self
.connection
)
848 'Adding relation %s <-> %s', relation1
, relation2
)
851 result
= await app_facade
.AddRelation([relation1
, relation2
])
852 except JujuAPIError
as e
:
853 if 'relation already exists' not in e
.message
:
856 'Relation %s <-> %s already exists', relation1
, relation2
)
857 # TODO: if relation already exists we should return the
858 # Relation ModelEntity here
861 def predicate(delta
):
863 for endpoint
in delta
.data
['endpoints']:
864 endpoints
[endpoint
['application-name']] = endpoint
['relation']
865 return endpoints
== result
.endpoints
867 return await self
._wait
_for
_new
('relation', None, predicate
)
869 def add_space(self
, name
, *cidrs
):
870 """Add a new network space.
872 Adds a new space with the given name and associates the given
873 (optional) list of existing subnet CIDRs with it.
875 :param str name: Name of the space
876 :param \*cidrs: Optional list of existing subnet CIDRs
879 raise NotImplementedError()
881 async def add_ssh_key(self
, user
, key
):
882 """Add a public SSH key to this model.
884 :param str user: The username of the user
885 :param str key: The public ssh key
888 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection
)
889 return await key_facade
.AddKeys([key
], user
)
890 add_ssh_keys
= add_ssh_key
892 def add_subnet(self
, cidr_or_id
, space
, *zones
):
893 """Add an existing subnet to this model.
895 :param str cidr_or_id: CIDR or provider ID of the existing subnet
896 :param str space: Network space with which to associate
897 :param str \*zones: Zone(s) in which the subnet resides
900 raise NotImplementedError()
902 def get_backups(self
):
903 """Retrieve metadata for backups in this model.
906 raise NotImplementedError()
908 def block(self
, *commands
):
909 """Add a new block to this model.
911 :param str \*commands: The commands to block. Valid values are
912 'all-changes', 'destroy-model', 'remove-object'
915 raise NotImplementedError()
917 def get_blocks(self
):
918 """List blocks for this model.
921 raise NotImplementedError()
923 def get_cached_images(self
, arch
=None, kind
=None, series
=None):
924 """Return a list of cached OS images.
926 :param str arch: Filter by image architecture
927 :param str kind: Filter by image kind, e.g. 'lxd'
928 :param str series: Filter by image series, e.g. 'xenial'
931 raise NotImplementedError()
933 def create_backup(self
, note
=None, no_download
=False):
934 """Create a backup of this model.
936 :param str note: A note to store with the backup
937 :param bool no_download: Do not download the backup archive
938 :return str: Path to downloaded archive
941 raise NotImplementedError()
943 def create_storage_pool(self
, name
, provider_type
, **pool_config
):
944 """Create or define a storage pool.
946 :param str name: Name to give the storage pool
947 :param str provider_type: Pool provider type
948 :param \*\*pool_config: key/value pool configuration pairs
951 raise NotImplementedError()
954 self
, no_tail
=False, exclude_module
=None, include_module
=None,
955 include
=None, level
=None, limit
=0, lines
=10, replay
=False,
957 """Get log messages for this model.
959 :param bool no_tail: Stop after returning existing log messages
960 :param list exclude_module: Do not show log messages for these logging
962 :param list include_module: Only show log messages for these logging
964 :param list include: Only show log messages for these entities
965 :param str level: Log level to show, valid options are 'TRACE',
966 'DEBUG', 'INFO', 'WARNING', 'ERROR,
967 :param int limit: Return this many of the most recent (possibly
968 filtered) lines are shown
969 :param int lines: Yield this many of the most recent lines, and keep
971 :param bool replay: Yield the entire log, and keep yielding
972 :param list exclude: Do not show log messages for these entities
975 raise NotImplementedError()
977 def _get_series(self
, entity_url
, entity
):
978 # try to get the series from the provided charm URL
979 if entity_url
.startswith('cs:'):
980 parts
= entity_url
[3:].split('/')
982 parts
= entity_url
.split('/')
983 if parts
[0].startswith('~'):
986 # series was specified in the URL
988 # series was not supplied at all, so use the newest
989 # supported series according to the charm store
990 ss
= entity
['Meta']['supported-series']
991 return ss
['SupportedSeries'][0]
994 self
, entity_url
, application_name
=None, bind
=None, budget
=None,
995 channel
=None, config
=None, constraints
=None, force
=False,
996 num_units
=1, plan
=None, resources
=None, series
=None, storage
=None,
998 """Deploy a new service or bundle.
1000 :param str entity_url: Charm or bundle url
1001 :param str application_name: Name to give the service
1002 :param dict bind: <charm endpoint>:<network space> pairs
1003 :param dict budget: <budget name>:<limit> pairs
1004 :param str channel: Charm store channel from which to retrieve
1005 the charm or bundle, e.g. 'development'
1006 :param dict config: Charm configuration dictionary
1007 :param constraints: Service constraints
1008 :type constraints: :class:`juju.Constraints`
1009 :param bool force: Allow charm to be deployed to a machine running
1010 an unsupported series
1011 :param int num_units: Number of units to deploy
1012 :param str plan: Plan under which to deploy charm
1013 :param dict resources: <resource name>:<file path> pairs
1014 :param str series: Series on which to deploy
1015 :param dict storage: Storage constraints TODO how do these look?
1016 :param to: Placement directive as a string. For example:
1018 '23' - place on machine 23
1019 'lxd:7' - place in new lxd container on machine 7
1020 '24/lxd/3' - place in container 3 on machine 24
1022 If None, a new machine is provisioned.
1027 - support local resources
1032 k
: client
.Constraints(**v
)
1033 for k
, v
in storage
.items()
1037 entity_url
.startswith('local:') or
1038 os
.path
.isdir(entity_url
)
1041 entity_id
= entity_url
.replace('local:', '')
1043 entity
= await self
.charmstore
.entity(entity_url
)
1044 entity_id
= entity
['Id']
1046 client_facade
= client
.ClientFacade
.from_connection(self
.connection
)
1048 is_bundle
= ((is_local
and
1049 (Path(entity_id
) / 'bundle.yaml').exists()) or
1050 (not is_local
and 'bundle/' in entity_id
))
1053 handler
= BundleHandler(self
)
1054 await handler
.fetch_plan(entity_id
)
1055 await handler
.execute_plan()
1056 extant_apps
= {app
for app
in self
.applications
}
1057 pending_apps
= set(handler
.applications
) - extant_apps
1059 # new apps will usually be in the model by now, but if some
1060 # haven't made it yet we'll need to wait on them to be added
1061 await asyncio
.gather(*[
1062 asyncio
.ensure_future(
1063 self
._wait
_for
_new
('application', app_name
),
1065 for app_name
in pending_apps
1067 return [app
for name
, app
in self
.applications
.items()
1068 if name
in handler
.applications
]
1071 if not application_name
:
1072 application_name
= entity
['Meta']['charm-metadata']['Name']
1074 series
= self
._get
_series
(entity_url
, entity
)
1077 await client_facade
.AddCharm(channel
, entity_id
)
1078 # XXX: we're dropping local resources here, but we don't
1079 # actually support them yet anyway
1080 resources
= await self
._add
_store
_resources
(application_name
,
1084 # We have a local charm dir that needs to be uploaded
1085 charm_dir
= os
.path
.abspath(
1086 os
.path
.expanduser(entity_id
))
1087 series
= series
or get_charm_series(charm_dir
)
1090 "Couldn't determine series for charm at {}. "
1091 "Pass a 'series' kwarg to Model.deploy().".format(
1093 entity_id
= await self
.add_local_charm_dir(charm_dir
, series
)
1094 return await self
._deploy
(
1095 charm_url
=entity_id
,
1096 application
=application_name
,
1098 config
=config
or {},
1099 constraints
=constraints
,
1100 endpoint_bindings
=bind
,
1101 resources
=resources
,
1104 num_units
=num_units
,
1105 placement
=parse_placement(to
)
1108 async def _add_store_resources(self
, application
, entity_url
, entity
=None):
1110 # avoid extra charm store call if one was already made
1111 entity
= await self
.charmstore
.entity(entity_url
)
1114 'description': resource
['Description'],
1115 'fingerprint': resource
['Fingerprint'],
1116 'name': resource
['Name'],
1117 'path': resource
['Path'],
1118 'revision': resource
['Revision'],
1119 'size': resource
['Size'],
1120 'type_': resource
['Type'],
1122 } for resource
in entity
['Meta']['resources']
1128 resources_facade
= client
.ResourcesFacade
.from_connection(
1130 response
= await resources_facade
.AddPendingResources(
1131 tag
.application(application
),
1133 [client
.CharmResource(**resource
) for resource
in resources
])
1134 resource_map
= {resource
['name']: pid
1136 in zip(resources
, response
.pending_ids
)}
1139 async def _deploy(self
, charm_url
, application
, series
, config
,
1140 constraints
, endpoint_bindings
, resources
, storage
,
1141 channel
=None, num_units
=None, placement
=None):
1142 """Logic shared between `Model.deploy` and `BundleHandler.deploy`.
1144 log
.info('Deploying %s', charm_url
)
1146 # stringify all config values for API, and convert to YAML
1147 config
= {k
: str(v
) for k
, v
in config
.items()}
1148 config
= yaml
.dump({application
: config
},
1149 default_flow_style
=False)
1151 app_facade
= client
.ApplicationFacade
.from_connection(
1154 app
= client
.ApplicationDeploy(
1155 charm_url
=charm_url
,
1156 application
=application
,
1160 constraints
=parse_constraints(constraints
),
1161 endpoint_bindings
=endpoint_bindings
,
1162 num_units
=num_units
,
1163 resources
=resources
,
1168 result
= await app_facade
.Deploy([app
])
1169 errors
= [r
.error
.message
for r
in result
.results
if r
.error
]
1171 raise JujuError('\n'.join(errors
))
1172 return await self
._wait
_for
_new
('application', application
)
1174 async def destroy(self
):
1175 """Terminate all machines and resources for this model.
1176 Is already implemented in controller.py.
1178 raise NotImplementedError()
1180 async def destroy_unit(self
, *unit_names
):
1181 """Destroy units by name.
1184 app_facade
= client
.ApplicationFacade
.from_connection(self
.connection
)
1187 'Destroying unit%s %s',
1188 's' if len(unit_names
) == 1 else '',
1189 ' '.join(unit_names
))
1191 return await app_facade
.DestroyUnits(list(unit_names
))
1192 destroy_units
= destroy_unit
1194 def get_backup(self
, archive_id
):
1195 """Download a backup archive file.
1197 :param str archive_id: The id of the archive to download
1198 :return str: Path to the archive file
1201 raise NotImplementedError()
1204 self
, num_controllers
=0, constraints
=None, series
=None, to
=None):
1205 """Ensure sufficient controllers exist to provide redundancy.
1207 :param int num_controllers: Number of controllers to make available
1208 :param constraints: Constraints to apply to the controller machines
1209 :type constraints: :class:`juju.Constraints`
1210 :param str series: Series of the controller machines
1211 :param list to: Placement directives for controller machines, e.g.::
1214 'lxc:7' - new lxc container on machine 7
1215 '24/lxc/3' - lxc container 3 or machine 24
1217 If None, a new machine is provisioned.
1220 raise NotImplementedError()
1222 def get_config(self
):
1223 """Return the configuration settings for this model.
1226 raise NotImplementedError()
1228 def get_constraints(self
):
1229 """Return the machine constraints for this model.
1232 raise NotImplementedError()
1234 async def grant(self
, username
, acl
='read'):
1235 """Grant a user access to this model.
1237 :param str username: Username
1238 :param str acl: Access control ('read' or 'write')
1241 controller_conn
= await self
.connection
.controller()
1242 model_facade
= client
.ModelManagerFacade
.from_connection(
1244 user
= tag
.user(username
)
1245 model
= tag
.model(self
.info
.uuid
)
1246 changes
= client
.ModifyModelAccess(acl
, 'grant', model
, user
)
1247 await self
.revoke(username
)
1248 return await model_facade
.ModifyModelAccess([changes
])
1250 def import_ssh_key(self
, identity
):
1251 """Add a public SSH key from a trusted indentity source to this model.
1253 :param str identity: User identity in the form <lp|gh>:<username>
1256 raise NotImplementedError()
1257 import_ssh_keys
= import_ssh_key
1259 async def get_machines(self
):
1260 """Return list of machines in this model.
1263 return list(self
.state
.machines
.keys())
1265 def get_shares(self
):
1266 """Return list of all users with access to this model.
1269 raise NotImplementedError()
1271 def get_spaces(self
):
1272 """Return list of all known spaces, including associated subnets.
1275 raise NotImplementedError()
1277 async def get_ssh_key(self
, raw_ssh
=False):
1278 """Return known SSH keys for this model.
1279 :param bool raw_ssh: if True, returns the raw ssh key,
1280 else it's fingerprint
1283 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection
)
1284 entity
= {'tag': tag
.model(self
.info
.uuid
)}
1285 entities
= client
.Entities([entity
])
1286 return await key_facade
.ListKeys(entities
, raw_ssh
)
1287 get_ssh_keys
= get_ssh_key
1289 def get_storage(self
, filesystem
=False, volume
=False):
1290 """Return details of storage instances.
1292 :param bool filesystem: Include filesystem storage
1293 :param bool volume: Include volume storage
1296 raise NotImplementedError()
1298 def get_storage_pools(self
, names
=None, providers
=None):
1299 """Return list of storage pools.
1301 :param list names: Only include pools with these names
1302 :param list providers: Only include pools for these providers
1305 raise NotImplementedError()
1307 def get_subnets(self
, space
=None, zone
=None):
1308 """Return list of known subnets.
1310 :param str space: Only include subnets in this space
1311 :param str zone: Only include subnets in this zone
1314 raise NotImplementedError()
1316 def remove_blocks(self
):
1317 """Remove all blocks from this model.
1320 raise NotImplementedError()
1322 def remove_backup(self
, backup_id
):
1325 :param str backup_id: The id of the backup to remove
1328 raise NotImplementedError()
1330 def remove_cached_images(self
, arch
=None, kind
=None, series
=None):
1331 """Remove cached OS images.
1333 :param str arch: Architecture of the images to remove
1334 :param str kind: Image kind to remove, e.g. 'lxd'
1335 :param str series: Image series to remove, e.g. 'xenial'
1338 raise NotImplementedError()
1340 def remove_machine(self
, *machine_ids
):
1341 """Remove a machine from this model.
1343 :param str \*machine_ids: Ids of the machines to remove
1346 raise NotImplementedError()
1347 remove_machines
= remove_machine
1349 async def remove_ssh_key(self
, user
, key
):
1350 """Remove a public SSH key(s) from this model.
1352 :param str key: Full ssh key
1353 :param str user: Juju user to which the key is registered
1356 key_facade
= client
.KeyManagerFacade
.from_connection(self
.connection
)
1357 key
= base64
.b64decode(bytes(key
.strip().split()[1].encode('ascii')))
1358 key
= hashlib
.md5(key
).hexdigest()
1359 key
= ':'.join(a
+b
for a
, b
in zip(key
[::2], key
[1::2]))
1360 await key_facade
.DeleteKeys([key
], user
)
1361 remove_ssh_keys
= remove_ssh_key
1364 self
, bootstrap
=False, constraints
=None, archive
=None,
1365 backup_id
=None, upload_tools
=False):
1366 """Restore a backup archive to a new controller.
1368 :param bool bootstrap: Bootstrap a new state machine
1369 :param constraints: Model constraints
1370 :type constraints: :class:`juju.Constraints`
1371 :param str archive: Path to backup archive to restore
1372 :param str backup_id: Id of backup to restore
1373 :param bool upload_tools: Upload tools if bootstrapping a new machine
1376 raise NotImplementedError()
1378 def retry_provisioning(self
):
1379 """Retry provisioning for failed machines.
1382 raise NotImplementedError()
1384 async def revoke(self
, username
):
1385 """Revoke a user's access to this model.
1387 :param str username: Username to revoke
1390 controller_conn
= await self
.connection
.controller()
1391 model_facade
= client
.ModelManagerFacade
.from_connection(
1393 user
= tag
.user(username
)
1394 model
= tag
.model(self
.info
.uuid
)
1395 changes
= client
.ModifyModelAccess('read', 'revoke', model
, user
)
1396 return await model_facade
.ModifyModelAccess([changes
])
1398 def run(self
, command
, timeout
=None):
1399 """Run command on all machines in this model.
1401 :param str command: The command to run
1402 :param int timeout: Time to wait before command is considered failed
1405 raise NotImplementedError()
1407 def set_config(self
, **config
):
1408 """Set configuration keys on this model.
1410 :param \*\*config: Config key/values
1413 raise NotImplementedError()
1415 def set_constraints(self
, constraints
):
1416 """Set machine constraints on this model.
1418 :param :class:`juju.Constraints` constraints: Machine constraints
1421 raise NotImplementedError()
1423 def get_action_output(self
, action_uuid
, wait
=-1):
1424 """Get the results of an action by ID.
1426 :param str action_uuid: Id of the action
1427 :param int wait: Time in seconds to wait for action to complete
1430 raise NotImplementedError()
1432 def get_action_status(self
, uuid_or_prefix
=None, name
=None):
1433 """Get the status of all actions, filtered by ID, ID prefix, or action name.
1435 :param str uuid_or_prefix: Filter by action uuid or prefix
1436 :param str name: Filter by action name
1439 raise NotImplementedError()
1441 def get_budget(self
, budget_name
):
1442 """Get budget usage info.
1444 :param str budget_name: Name of budget
1447 raise NotImplementedError()
1449 async def get_status(self
, filters
=None, utc
=False):
1450 """Return the status of the model.
1452 :param str filters: Optional list of applications, units, or machines
1453 to include, which can use wildcards ('*').
1454 :param bool utc: Display time as UTC in RFC3339 format
1457 client_facade
= client
.ClientFacade
.from_connection(self
.connection
)
1458 return await client_facade
.FullStatus(filters
)
1461 self
, all_
=False, destination
=None, dry_run
=False, public
=False,
1462 source
=None, stream
=None, version
=None):
1463 """Copy Juju tools into this model.
1465 :param bool all_: Copy all versions, not just the latest
1466 :param str destination: Path to local destination directory
1467 :param bool dry_run: Don't do the actual copy
1468 :param bool public: Tools are for a public cloud, so generate mirrors
1470 :param str source: Path to local source directory
1471 :param str stream: Simplestreams stream for which to sync metadata
1472 :param str version: Copy a specific major.minor version
1475 raise NotImplementedError()
1477 def unblock(self
, *commands
):
1478 """Unblock an operation that would alter this model.
1480 :param str \*commands: The commands to unblock. Valid values are
1481 'all-changes', 'destroy-model', 'remove-object'
1484 raise NotImplementedError()
1486 def unset_config(self
, *keys
):
1487 """Unset configuration on this model.
1489 :param str \*keys: The keys to unset
1492 raise NotImplementedError()
1494 def upgrade_gui(self
):
1495 """Upgrade the Juju GUI for this model.
1498 raise NotImplementedError()
1501 self
, dry_run
=False, reset_previous_upgrade
=False,
1502 upload_tools
=False, version
=None):
1503 """Upgrade Juju on all machines in a model.
1505 :param bool dry_run: Don't do the actual upgrade
1506 :param bool reset_previous_upgrade: Clear the previous (incomplete)
1508 :param bool upload_tools: Upload local version of tools
1509 :param str version: Upgrade to a specific version
1512 raise NotImplementedError()
1514 def upload_backup(self
, archive_path
):
1515 """Store a backup archive remotely in Juju.
1517 :param str archive_path: Path to local archive
1520 raise NotImplementedError()
1523 def charmstore(self
):
1524 return self
._charmstore
1526 async def get_metrics(self
, *tags
):
1527 """Retrieve metrics.
1529 :param str \*tags: Tags of entities from which to retrieve metrics.
1530 No tags retrieves the metrics of all units in the model.
1531 :return: Dictionary of unit_name:metrics
1534 log
.debug("Retrieving metrics for %s",
1535 ', '.join(tags
) if tags
else "all units")
1537 metrics_facade
= client
.MetricsDebugFacade
.from_connection(
1540 entities
= [client
.Entity(tag
) for tag
in tags
]
1541 metrics_result
= await metrics_facade
.GetMetrics(entities
)
1543 metrics
= collections
.defaultdict(list)
1545 for entity_metrics
in metrics_result
.results
:
1546 error
= entity_metrics
.error
1548 if "is not a valid tag" in error
:
1549 raise ValueError(error
.message
)
1551 raise Exception(error
.message
)
1553 for metric
in entity_metrics
.metrics
:
1554 metrics
[metric
.unit
].append(vars(metric
))
1559 def get_charm_series(path
):
1560 """Inspects the charm directory at ``path`` and returns a default
1561 series from its metadata.yaml (the first item in the 'series' list).
1563 Returns None if no series can be determined.
1566 md
= Path(path
) / "metadata.yaml"
1569 data
= yaml
.load(md
.open())
1570 series
= data
.get('series')
1571 return series
[0] if series
else None
1574 class BundleHandler(object):
1576 Handle bundles by using the API to translate bundle YAML into a plan of
1577 steps and then dispatching each of those using the API.
1579 def __init__(self
, model
):
1581 self
.charmstore
= model
.charmstore
1583 self
.references
= {}
1584 self
._units
_by
_app
= {}
1585 for unit_name
, unit
in model
.units
.items():
1586 app_units
= self
._units
_by
_app
.setdefault(unit
.application
, [])
1587 app_units
.append(unit_name
)
1588 self
.client_facade
= client
.ClientFacade
.from_connection(
1590 self
.app_facade
= client
.ApplicationFacade
.from_connection(
1592 self
.ann_facade
= client
.AnnotationsFacade
.from_connection(
1595 async def _handle_local_charms(self
, bundle
):
1596 """Search for references to local charms (i.e. filesystem paths)
1597 in the bundle. Upload the local charms to the model, and replace
1598 the filesystem paths with appropriate 'local:' paths in the bundle.
1600 Return the modified bundle.
1602 :param dict bundle: Bundle dictionary
1603 :return: Modified bundle dictionary
1608 default_series
= bundle
.get('series')
1609 for app_name
in self
.applications
:
1610 app_dict
= bundle
['services'][app_name
]
1611 charm_dir
= os
.path
.abspath(os
.path
.expanduser(app_dict
['charm']))
1612 if not os
.path
.isdir(charm_dir
):
1615 app_dict
.get('series') or
1617 get_charm_series(charm_dir
)
1621 "Couldn't determine series for charm at {}. "
1622 "Add a 'series' key to the bundle.".format(charm_dir
))
1624 # Keep track of what we need to update. We keep a list of apps
1625 # that need to be updated, and a corresponding list of args
1626 # needed to update those apps.
1627 apps
.append(app_name
)
1628 args
.append((charm_dir
, series
))
1631 # If we have apps to update, spawn all the coroutines concurrently
1632 # and wait for them to finish.
1633 charm_urls
= await asyncio
.gather(*[
1634 self
.model
.add_local_charm_dir(*params
)
1636 ], loop
=self
.model
.loop
)
1637 # Update the 'charm:' entry for each app with the new 'local:' url.
1638 for app_name
, charm_url
in zip(apps
, charm_urls
):
1639 bundle
['services'][app_name
]['charm'] = charm_url
1643 async def fetch_plan(self
, entity_id
):
1644 is_local
= not entity_id
.startswith('cs:') and os
.path
.isdir(entity_id
)
1646 bundle_yaml
= (Path(entity_id
) / "bundle.yaml").read_text()
1648 bundle_yaml
= await self
.charmstore
.files(entity_id
,
1649 filename
='bundle.yaml',
1651 self
.bundle
= yaml
.safe_load(bundle_yaml
)
1652 self
.bundle
= await self
._handle
_local
_charms
(self
.bundle
)
1654 self
.plan
= await self
.client_facade
.GetBundleChanges(
1655 yaml
.dump(self
.bundle
))
1657 async def execute_plan(self
):
1658 for step
in self
.plan
.changes
:
1659 method
= getattr(self
, step
.method
)
1660 result
= await method(*step
.args
)
1661 self
.references
[step
.id_
] = result
1664 def applications(self
):
1665 return list(self
.bundle
['services'].keys())
1667 def resolve(self
, reference
):
1668 if reference
and reference
.startswith('$'):
1669 reference
= self
.references
[reference
[1:]]
1672 async def addCharm(self
, charm
, series
):
1674 :param charm string:
1675 Charm holds the URL of the charm to be added.
1677 :param series string:
1678 Series holds the series of the charm to be added
1679 if the charm default is not sufficient.
1681 # We don't add local charms because they've already been added
1682 # by self._handle_local_charms
1683 if charm
.startswith('local:'):
1686 entity_id
= await self
.charmstore
.entityId(charm
)
1687 log
.debug('Adding %s', entity_id
)
1688 await self
.client_facade
.AddCharm(None, entity_id
)
1691 async def addMachines(self
, params
=None):
1694 Dictionary specifying the machine to add. All keys are optional.
1697 series: string specifying the machine OS series.
1699 constraints: string holding machine constraints, if any. We'll
1700 parse this into the json friendly dict that the juju api
1703 container_type: string holding the type of the container (for
1704 instance ""lxd" or kvm"). It is not specified for top level
1707 parent_id: string holding a placeholder pointing to another
1708 machine change or to a unit change. This value is only
1709 specified in the case this machine is a container, in
1710 which case also ContainerType is set.
1713 params
= params
or {}
1716 params
= {normalize_key(k
): params
[k
] for k
in params
.keys()}
1718 # Fix up values, as necessary.
1719 if 'parent_id' in params
:
1720 params
['parent_id'] = self
.resolve(params
['parent_id'])
1722 params
['constraints'] = parse_constraints(
1723 params
.get('constraints'))
1724 params
['jobs'] = params
.get('jobs', ['JobHostUnits'])
1726 if params
.get('container_type') == 'lxc':
1727 log
.warning('Juju 2.0 does not support lxc containers. '
1728 'Converting containers to lxd.')
1729 params
['container_type'] = 'lxd'
1731 # Submit the request.
1732 params
= client
.AddMachineParams(**params
)
1733 results
= await self
.client_facade
.AddMachines([params
])
1734 error
= results
.machines
[0].error
1736 raise ValueError("Error adding machine: %s" % error
.message
)
1737 machine
= results
.machines
[0].machine
1738 log
.debug('Added new machine %s', machine
)
1741 async def addRelation(self
, endpoint1
, endpoint2
):
1743 :param endpoint1 string:
1744 :param endpoint2 string:
1745 Endpoint1 and Endpoint2 hold relation endpoints in the
1746 "application:interface" form, where the application is always a
1747 placeholder pointing to an application change, and the interface is
1748 optional. Examples are "$deploy-42:web" or just "$deploy-42".
1750 endpoints
= [endpoint1
, endpoint2
]
1751 # resolve indirect references
1752 for i
in range(len(endpoints
)):
1753 parts
= endpoints
[i
].split(':')
1754 parts
[0] = self
.resolve(parts
[0])
1755 endpoints
[i
] = ':'.join(parts
)
1757 log
.info('Relating %s <-> %s', *endpoints
)
1758 return await self
.model
.add_relation(*endpoints
)
1760 async def deploy(self
, charm
, series
, application
, options
, constraints
,
1761 storage
, endpoint_bindings
, resources
):
1763 :param charm string:
1764 Charm holds the URL of the charm to be used to deploy this
1767 :param series string:
1768 Series holds the series of the application to be deployed
1769 if the charm default is not sufficient.
1771 :param application string:
1772 Application holds the application name.
1774 :param options map[string]interface{}:
1775 Options holds application options.
1777 :param constraints string:
1778 Constraints holds the optional application constraints.
1780 :param storage map[string]string:
1781 Storage holds the optional storage constraints.
1783 :param endpoint_bindings map[string]string:
1784 EndpointBindings holds the optional endpoint bindings
1786 :param resources map[string]int:
1787 Resources identifies the revision to use for each resource
1788 of the application's charm.
1790 # resolve indirect references
1791 charm
= self
.resolve(charm
)
1792 # the bundle plan doesn't actually do anything with resources, even
1793 # though it ostensibly gives us something (None) for that param
1794 if not charm
.startswith('local:'):
1795 resources
= await self
.model
._add
_store
_resources
(application
,
1797 await self
.model
._deploy
(
1799 application
=application
,
1802 constraints
=constraints
,
1803 endpoint_bindings
=endpoint_bindings
,
1804 resources
=resources
,
1809 async def addUnit(self
, application
, to
):
1811 :param application string:
1812 Application holds the application placeholder name for which a unit
1816 To holds the optional location where to add the unit, as a
1817 placeholder pointing to another unit change or to a machine change.
1819 application
= self
.resolve(application
)
1820 placement
= self
.resolve(to
)
1821 if self
._units
_by
_app
.get(application
):
1822 # enough units for this application already exist;
1823 # claim one, and carry on
1824 # NB: this should probably honor placement, but the juju client
1825 # doesn't, so we're not bothering, either
1826 unit_name
= self
._units
_by
_app
[application
].pop()
1827 log
.debug('Reusing unit %s for %s', unit_name
, application
)
1828 return self
.model
.units
[unit_name
]
1830 log
.debug('Adding new unit for %s%s', application
,
1831 ' to %s' % placement
if placement
else '')
1832 return await self
.model
.applications
[application
].add_unit(
1837 async def expose(self
, application
):
1839 :param application string:
1840 Application holds the placeholder name of the application that must
1843 application
= self
.resolve(application
)
1844 log
.info('Exposing %s', application
)
1845 return await self
.model
.applications
[application
].expose()
1847 async def setAnnotations(self
, id_
, entity_type
, annotations
):
1850 Id is the placeholder for the application or machine change
1851 corresponding to the entity to be annotated.
1853 :param entity_type EntityType:
1854 EntityType holds the type of the entity, "application" or
1857 :param annotations map[string]string:
1858 Annotations holds the annotations as key/value pairs.
1860 entity_id
= self
.resolve(id_
)
1862 entity
= self
.model
.state
.get_entity(entity_type
, entity_id
)
1864 entity
= await self
.model
._wait
_for
_new
(entity_type
, entity_id
)
1865 return await entity
.set_annotations(annotations
)
1868 class CharmStore(object):
1870 Async wrapper around theblues.charmstore.CharmStore
1872 def __init__(self
, loop
):
1874 self
._cs
= theblues
.charmstore
.CharmStore(timeout
=5)
1876 def __getattr__(self
, name
):
1878 Wrap method calls in coroutines that use run_in_executor to make them
1881 attr
= getattr(self
._cs
, name
)
1882 if not callable(attr
):
1883 wrapper
= partial(getattr, self
._cs
, name
)
1884 setattr(self
, name
, wrapper
)
1886 async def coro(*args
, **kwargs
):
1887 method
= partial(attr
, *args
, **kwargs
)
1888 for attempt
in range(1, 4):
1890 return await self
.loop
.run_in_executor(None, method
)
1891 except theblues
.errors
.ServerError
:
1894 await asyncio
.sleep(1, loop
=self
.loop
)
1895 setattr(self
, name
, coro
)
1900 class CharmArchiveGenerator(object):
1902 Create a Zip archive of a local charm directory for upload to a controller.
1904 This is used automatically by
1905 `Model.add_local_charm_dir <#juju.model.Model.add_local_charm_dir>`_.
1907 def __init__(self
, path
):
1908 self
.path
= os
.path
.abspath(os
.path
.expanduser(path
))
1910 def make_archive(self
, path
):
1911 """Create archive of directory and write to ``path``.
1913 :param path: Path to archive
1917 * build/\* - This is used for packing the charm itself and any
1919 * \*/.\* - Hidden files are all ignored for now. This will most
1920 likely be changed into a specific ignore list
1924 zf
= zipfile
.ZipFile(path
, 'w', zipfile
.ZIP_DEFLATED
)
1925 for dirpath
, dirnames
, filenames
in os
.walk(self
.path
):
1926 relative_path
= dirpath
[len(self
.path
) + 1:]
1927 if relative_path
and not self
._ignore
(relative_path
):
1928 zf
.write(dirpath
, relative_path
)
1929 for name
in filenames
:
1930 archive_name
= os
.path
.join(relative_path
, name
)
1931 if not self
._ignore
(archive_name
):
1932 real_path
= os
.path
.join(dirpath
, name
)
1933 self
._check
_type
(real_path
)
1934 if os
.path
.islink(real_path
):
1935 self
._check
_link
(real_path
)
1936 self
._write
_symlink
(
1937 zf
, os
.readlink(real_path
), archive_name
)
1939 zf
.write(real_path
, archive_name
)
1943 def _check_type(self
, path
):
1947 if stat
.S_ISDIR(s
.st_mode
) or stat
.S_ISREG(s
.st_mode
):
1949 raise ValueError("Invalid Charm at % %s" % (
1950 path
, "Invalid file type for a charm"))
1952 def _check_link(self
, path
):
1953 link_path
= os
.readlink(path
)
1954 if link_path
[0] == "/":
1956 "Invalid Charm at %s: %s" % (
1957 path
, "Absolute links are invalid"))
1958 path_dir
= os
.path
.dirname(path
)
1959 link_path
= os
.path
.join(path_dir
, link_path
)
1960 if not link_path
.startswith(os
.path
.abspath(self
.path
)):
1962 "Invalid charm at %s %s" % (
1963 path
, "Only internal symlinks are allowed"))
1965 def _write_symlink(self
, zf
, link_target
, link_path
):
1966 """Package symlinks with appropriate zipfile metadata."""
1967 info
= zipfile
.ZipInfo()
1968 info
.filename
= link_path
1969 info
.create_system
= 3
1970 # Magic code for symlinks / py2/3 compat
1971 # 27166663808 = (stat.S_IFLNK | 0755) << 16
1972 info
.external_attr
= 2716663808
1973 zf
.writestr(info
, link_target
)
1975 def _ignore(self
, path
):
1976 if path
== "build" or path
.startswith("build/"):
1978 if path
.startswith('.'):