X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=juju%2Fmodel.py;h=480813c78a86298e1fde7ccaa704af0b47bf2ee0;hb=2ed7314a9ea1240883655bc521b6e27f149aa485;hp=9d638cbb19a3eece5773cbb039b4d91bc391afbb;hpb=17b45c65afac23a123e66834f93049a1c7e45201;p=osm%2FN2VC.git diff --git a/juju/model.py b/juju/model.py index 9d638cb..480813c 100644 --- a/juju/model.py +++ b/juju/model.py @@ -1,4 +1,162 @@ +import logging + +from .client import client +from .client import watcher +from .delta import get_entity_delta + +log = logging.getLogger(__name__) + + +class ModelObserver(object): + def __call__(self, delta, old, new, model): + if old is None and new is not None: + type_ = 'add' + else: + type_ = delta.type + handler_name = 'on_{}_{}'.format(delta.entity, type_) + method = getattr(self, handler_name, self.on_change) + log.debug( + 'Model changed: %s %s %s', + delta.entity, delta.type, delta.get_id()) + method(delta, old, new, model) + + def on_change(self, delta, old, new, model): + pass + + +class ModelEntity(object): + """An object in the Model tree""" + + def __init__(self, data, model): + """Initialize a new entity + + :param data: dict of data from a watcher delta + :param model: The model instance in whose object tree this + entity resides + + """ + self.data = data + self.model = model + self.connection = model.connection + + def __getattr__(self, name): + return self.data[name] + + class Model(object): + def __init__(self, connection): + """Instantiate a new connected Model. + + :param connection: `juju.client.connection.Connection` instance + + """ + self.connection = connection + self.observers = set() + self.state = dict() + self._watching = False + + @property + def applications(self): + return self.state.get('application', {}) + + @property + def units(self): + return self.state.get('unit', {}) + + def stop_watching(self): + self.connection.cancel() + self._watching = False + + def add_observer(self, callable_): + """Register an "on-model-change" callback + + Once a watch is started (Model.watch() is called), ``callable_`` + will be called each time the model changes. callable_ should + accept the following positional arguments: + + delta - An instance of :class:`juju.delta.EntityDelta` + containing the raw delta data recv'd from the Juju + websocket. + + old_obj - If the delta modifies an existing object in the model, + old_obj will be a copy of that object, as it was before the + delta was applied. Will be None if the delta creates a new + entity in the model. + + new_obj - A copy of the new or updated object, after the delta + is applied. Will be None if the delta removes an entity + from the model. + + model - The :class:`Model` itself. + + """ + self.observers.add(callable_) + + async def watch(self): + """Start an asynchronous watch against this model. + + See :meth:`add_observer` to register an onchange callback. + + """ + self._watching = True + allwatcher = watcher.AllWatcher() + allwatcher.connect(await self.connection.clone()) + while self._watching: + results = await allwatcher.Next() + for delta in results.deltas: + delta = get_entity_delta(delta) + old_obj, new_obj = self._apply_delta(delta) + self._notify_observers(delta, old_obj, new_obj) + + def _apply_delta(self, delta): + """Apply delta to our model state and return the a copy of the + affected object as it was before and after the update, e.g.: + + old_obj, new_obj = self._apply_delta(delta) + + old_obj may be None if the delta is for the creation of a new object, + e.g. a new application or unit is deployed. + + new_obj may be None if no object was created or updated, or if an + object was deleted as a result of the delta being applied. + + """ + old_obj, new_obj = None, None + + if (delta.entity in self.state and + delta.get_id() in self.state[delta.entity]): + old_obj = self.state[delta.entity][delta.get_id()] + if delta.type == 'remove': + del self.state[delta.entity][delta.get_id()] + return old_obj, new_obj + + new_obj = self.state.setdefault(delta.entity, {})[delta.get_id()] = ( + self._create_model_entity(delta)) + + return old_obj, new_obj + + def _create_model_entity(self, delta): + """Return an object instance representing the entity created or + updated by ``delta`` + + """ + entity_class = delta.get_entity_class() + return entity_class(delta.data, self) + + def _notify_observers(self, delta, old_obj, new_obj): + """Call observing callbacks, notifying them of a change in model state + + :param delta: The raw change from the watcher + (:class:`juju.client.overrides.Delta`) + :param old_obj: The object in the model that this delta updates. + May be None. + :param new_obj: The object in the model that is created or updated + by applying this delta. + + """ + for o in self.observers: + o(delta, old_obj, new_obj, self) + def add_machine( self, spec=None, constraints=None, disks=None, series=None, count=1): @@ -144,10 +302,10 @@ class Model(object): """ pass - def deploy( + async def deploy( self, entity_url, service_name=None, bind=None, budget=None, channel=None, config=None, constraints=None, force=False, - num_units=1, plan=None, resource=None, series=None, storage=None, + num_units=1, plan=None, resources=None, series=None, storage=None, to=None): """Deploy a new service or bundle. @@ -164,7 +322,7 @@ class Model(object): an unsupported series :param int num_units: Number of units to deploy :param str plan: Plan under which to deploy charm - :param dict resource: : pairs + :param dict resources: : pairs :param str series: Series on which to deploy :param dict storage: Storage constraints TODO how do these look? :param str to: Placement directive, e.g.:: @@ -175,8 +333,56 @@ class Model(object): If None, a new machine is provisioned. - """ - pass + + TODO:: + + - entity_url must have a revision; look up latest automatically if + not provided by caller + - service_name is required; fill this in automatically if not + provided by caller + - series is required; how do we pick a default? + + """ + if constraints: + constraints = client.Value(**constraints) + + if to: + placement = [ + client.Placement(**p) for p in to + ] + else: + placement = [] + + if storage: + storage = { + k: client.Constraints(**v) + for k, v in storage.items() + } + + app_facade = client.ApplicationFacade() + client_facade = client.ClientFacade() + app_facade.connect(self.connection) + client_facade.connect(self.connection) + + log.debug( + 'Deploying %s', entity_url) + + await client_facade.AddCharm(channel, entity_url) + app = client.ApplicationDeploy( + application=service_name, + channel=channel, + charm_url=entity_url, + config=config, + constraints=constraints, + endpoint_bindings=bind, + num_units=num_units, + placement=placement, + resources=resources, + series=series, + storage=storage, + ) + + return await app_facade.Deploy([app]) def destroy(self): """Terminate all machines and resources for this model.