juju/machine.py

   1 import asyncio
   2 import logging
   3 import os
   4
   5 from dateutil.parser import parse as parse_date
   6
   7 from . import model, utils
   8 from .client import client
   9 from .errors import JujuError
  10
  11 log = logging.getLogger(__name__)
  12
  13
  14 class Machine(model.ModelEntity):
  15     def __init__(self, *args, **kwargs):
  16         super().__init__(*args, **kwargs)
  17         self.on_change(self._workaround_1695335)
  18
  19     async def _workaround_1695335(self, delta, old, new, model):
  20         """
  21         This is a (hacky) temporary work around for a bug in Juju where the
  22         instance status and agent version fields don't get updated properly
  23         by the AllWatcher.
  24
  25         Deltas never contain a value for `data['agent-status']['version']`,
  26         and once the `instance-status` reaches `pending`, we no longer get
  27         any updates for it (the deltas come in, but the `instance-status`
  28         data is always the same after that).
  29
  30         To work around this, whenever a delta comes in for this machine, we
  31         query FullStatus and use the data from there if and only if it's newer.
  32         Luckily, the timestamps on the `since` field does seem to be accurate.
  33
  34         See https://bugs.launchpad.net/juju/+bug/1695335
  35         """
  36         if delta.data.get('synthetic', False):
  37             # prevent infinite loops re-processing already processed deltas
  38             return
  39
  40         full_status = await utils.run_with_interrupt(model.get_status(),
  41                                                      model._watch_stopping,
  42                                                      model.loop)
  43         if model._watch_stopping.is_set():
  44             return
  45
  46         if self.id not in full_status.machines:
  47             return
  48
  49         if not full_status.machines[self.id]['instance-status']['since']:
  50             return
  51
  52         machine = full_status.machines[self.id]
  53
  54         change_log = []
  55         key_map = {
  56             'status': 'current',
  57             'info': 'message',
  58             'since': 'since',
  59         }
  60
  61         # handle agent version specially, because it's never set in
  62         # deltas, and we don't want even a newer delta to clear it
  63         agent_version = machine['agent-status']['version']
  64         if agent_version:
  65             delta.data['agent-status']['version'] = agent_version
  66             change_log.append(('agent-version', '', agent_version))
  67
  68         # only update (other) delta fields if status data is newer
  69         status_since = parse_date(machine['instance-status']['since'])
  70         delta_since = parse_date(delta.data['instance-status']['since'])
  71         if status_since > delta_since:
  72             for status_key in ('status', 'info', 'since'):
  73                 delta_key = key_map[status_key]
  74                 status_value = machine['instance-status'][status_key]
  75                 delta_value = delta.data['instance-status'][delta_key]
  76                 change_log.append((delta_key, delta_value, status_value))
  77                 delta.data['instance-status'][delta_key] = status_value
  78
  79         if change_log:
  80             log.debug('Overriding machine delta with FullStatus data')
  81             for log_item in change_log:
  82                 log.debug('    {}: {} -> {}'.format(*log_item))
  83             delta.data['synthetic'] = True
  84             old_obj, new_obj = self.model.state.apply_delta(delta)
  85             await model._notify_observers(delta, old_obj, new_obj)
  86
  87     async def destroy(self, force=False):
  88         """Remove this machine from the model.
  89
  90         Blocks until the machine is actually removed.
  91
  92         """
  93         facade = client.ClientFacade.from_connection(self.connection)
  94
  95         log.debug(
  96             'Destroying machine %s', self.id)
  97
  98         await facade.DestroyMachines(force, [self.id])
  99         return await self.model._wait(
 100             'machine', self.id, 'remove')
 101     remove = destroy
 102
 103     def run(self, command, timeout=None):
 104         """Run command on this machine.
 105
 106         :param str command: The command to run
 107         :param int timeout: Time to wait before command is considered failed
 108
 109         """
 110         raise NotImplementedError()
 111
 112     async def set_annotations(self, annotations):
 113         """Set annotations on this machine.
 114
 115         :param annotations map[string]string: the annotations as key/value
 116             pairs.
 117
 118         """
 119         log.debug('Updating annotations on machine %s', self.id)
 120
 121         self.ann_facade = client.AnnotationsFacade.from_connection(
 122             self.connection)
 123
 124         ann = client.EntityAnnotations(
 125             entity=self.id,
 126             annotations=annotations,
 127         )
 128         return await self.ann_facade.Set([ann])
 129
 130     async def scp_to(self, source, destination, user='ubuntu', proxy=False,
 131                      scp_opts=''):
 132         """Transfer files to this machine.
 133
 134         :param str source: Local path of file(s) to transfer
 135         :param str destination: Remote destination of transferred files
 136         :param str user: Remote username
 137         :param bool proxy: Proxy through the Juju API server
 138         :param str scp_opts: Additional options to the `scp` command
 139         """
 140         if proxy:
 141             raise NotImplementedError('proxy option is not implemented')
 142
 143         address = self.dns_name
 144         destination = '%s@%s:%s' % (user, address, destination)
 145         await self._scp(source, destination, scp_opts)
 146
 147     async def scp_from(self, source, destination, user='ubuntu', proxy=False,
 148                        scp_opts=''):
 149         """Transfer files from this machine.
 150
 151         :param str source: Remote path of file(s) to transfer
 152         :param str destination: Local destination of transferred files
 153         :param str user: Remote username
 154         :param bool proxy: Proxy through the Juju API server
 155         :param str scp_opts: Additional options to the `scp` command
 156         """
 157         if proxy:
 158             raise NotImplementedError('proxy option is not implemented')
 159
 160         address = self.dns_name
 161         source = '%s@%s:%s' % (user, address, source)
 162         await self._scp(source, destination, scp_opts)
 163
 164     async def _scp(self, source, destination, scp_opts):
 165         """ Execute an scp command. Requires a fully qualified source and
 166         destination.
 167         """
 168         cmd = [
 169             'scp',
 170             '-i', os.path.expanduser('~/.local/share/juju/ssh/juju_id_rsa'),
 171             '-o', 'StrictHostKeyChecking=no',
 172             source, destination
 173         ]
 174         cmd += scp_opts.split()
 175         loop = self.model.loop
 176         process = await asyncio.create_subprocess_exec(*cmd, loop=loop)
 177         await process.wait()
 178         if process.returncode != 0:
 179             raise JujuError("command failed: %s" % cmd)
 180
 181     def ssh(
 182             self, command, user=None, proxy=False, ssh_opts=None):
 183         """Execute a command over SSH on this machine.
 184
 185         :param str command: Command to execute
 186         :param str user: Remote username
 187         :param bool proxy: Proxy through the Juju API server
 188         :param str ssh_opts: Additional options to the `ssh` command
 189
 190         """
 191         raise NotImplementedError()
 192
 193     def status_history(self, num=20, utc=False):
 194         """Get status history for this machine.
 195
 196         :param int num: Size of history backlog
 197         :param bool utc: Display time as UTC in RFC3339 format
 198
 199         """
 200         raise NotImplementedError()
 201
 202     @property
 203     def agent_status(self):
 204         """Returns the current Juju agent status string.
 205
 206         """
 207         return self.safe_data['agent-status']['current']
 208
 209     @property
 210     def agent_status_since(self):
 211         """Get the time when the `agent_status` was last updated.
 212
 213         """
 214         return parse_date(self.safe_data['agent-status']['since'])
 215
 216     @property
 217     def agent_version(self):
 218         """Get the version of the Juju machine agent.
 219
 220         May return None if the agent is not yet available.
 221         """
 222         version = self.safe_data['agent-status']['version']
 223         if version:
 224             return client.Number.from_json(version)
 225         else:
 226             return None
 227
 228     @property
 229     def status(self):
 230         """Returns the current machine provisioning status string.
 231
 232         """
 233         return self.safe_data['instance-status']['current']
 234
 235     @property
 236     def status_message(self):
 237         """Returns the current machine provisioning status message.
 238
 239         """
 240         return self.safe_data['instance-status']['message']
 241
 242     @property
 243     def status_since(self):
 244         """Get the time when the `status` was last updated.
 245
 246         """
 247         return parse_date(self.safe_data['instance-status']['since'])
 248
 249     @property
 250     def dns_name(self):
 251         """Get the DNS name for this machine. This is a best guess based on the
 252         addresses available in current data.
 253
 254         May return None if no suitable address is found.
 255         """
 256         for scope in ['public', 'local-cloud']:
 257             addresses = self.safe_data['addresses'] or []
 258             addresses = [address for address in addresses
 259                          if address['scope'] == scope]
 260             if addresses:
 261                 return addresses[0]['value']
 262         return None