Merge "blueprint host-aggregates: host maintenance"

2 files changed, 132 insertions, 140 deletions

diff --git a/nova/compute/__init__.py b/nova/compute/__init__.py
index 72df5ddd5..55d1e2439 100644
--- a/nova/compute/__init__.py
+++ b/nova/compute/__init__.py
@@ -17,6 +17,7 @@
 #    under the License.
 
 from nova.compute.api import AggregateAPI
+from nova.compute.api import HostAPI
 # Importing full names to not pollute the namespace and cause possible
 # collisions with use of 'from nova.compute import <foo>' elsewhere.
 import nova.flags
diff --git a/nova/compute/api.py b/nova/compute/api.py
index b908a6587..be9b8dbb9 100644
--- a/nova/compute/api.py
+++ b/nova/compute/api.py
@@ -773,10 +773,8 @@ class API(base.Base):
         params = {"security_group_id": security_group['id']}
         # NOTE(comstud): No instance_uuid argument to this compute manager
         # call
-        self._cast_compute_message('refresh_security_group_rules',
-                                   context,
-                                   host=instance['host'],
-                                   params=params)
+        _cast_compute_message(self.db, 'refresh_security_group_rules', context,
+                              host=instance['host'], params=params)
 
     @wrap_check_policy
     def remove_security_group(self, context, instance, security_group_name):
@@ -804,10 +802,8 @@ class API(base.Base):
         params = {"security_group_id": security_group['id']}
         # NOTE(comstud): No instance_uuid argument to this compute manager
         # call
-        self._cast_compute_message('refresh_security_group_rules',
-                                   context,
-                                   host=instance['host'],
-                                   params=params)
+        _cast_compute_message(self.db, 'refresh_security_group_rules',
+                              context, host=instance['host'], params=params)
 
     @wrap_check_policy
     def update(self, context, instance, **kwargs):
@@ -846,8 +842,8 @@ class API(base.Base):
                         task_state=task_states.POWERING_OFF,
                         deleted_at=utils.utcnow())
 
-            self._cast_compute_message('power_off_instance', context,
-                                       instance)
+            _cast_compute_message(self.db, 'power_off_instance', context,
+                                  instance)
         else:
             LOG.warning(_('No host for instance, deleting immediately'),
                         instance=instance)
@@ -866,12 +862,12 @@ class API(base.Base):
                             task_state=task_states.DELETING,
                             progress=0)
 
-                self._cast_compute_message('terminate_instance', context,
-                                           instance)
+                _cast_compute_message(self.db, 'terminate_instance', context,
+                                      instance)
             else:
                 self.db.instance_destroy(context, instance['id'])
         except exception.InstanceNotFound:
-            # NOTE(comstud): Race condition.  Instance already gone.
+            # NOTE(comstud): Race condition. Instance already gone.
             pass
 
     # NOTE(jerdfelt): The API implies that only ACTIVE and ERROR are
@@ -905,8 +901,8 @@ class API(base.Base):
             self.update(context,
                         instance,
                         task_state=task_states.POWERING_ON)
-            self._cast_compute_message('power_on_instance', context,
-                                       instance)
+            _cast_compute_message(self.db, 'power_on_instance',
+                                  context, instance)
 
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.SOFT_DELETE])
@@ -931,8 +927,8 @@ class API(base.Base):
                     progress=0)
 
         rpc_method = rpc.cast if do_cast else rpc.call
-        self._cast_or_call_compute_message(rpc_method, 'stop_instance',
-                                           context, instance)
+        _cast_or_call_compute_message(self.db, rpc_method, 'stop_instance',
+                                      context, instance)
 
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.STOPPED, vm_states.SHUTOFF])
@@ -960,7 +956,7 @@ class API(base.Base):
         # TODO(yamahata): injected_files isn't supported right now.
         #                 It is used only for osapi. not for ec2 api.
         #                 availability_zone isn't used by run_instance.
-        self._cast_compute_message('start_instance', context, instance)
+        _cast_compute_message(self.db, 'start_instance', context, instance)
 
     #NOTE(bcwaldon): no policy check here since it should be rolled in to
     # search_opts in get_all
@@ -1085,49 +1081,6 @@ class API(base.Base):
 
         return self.db.instance_get_all_by_filters(context, filters)
 
-    def _cast_or_call_compute_message(self, rpc_method, compute_method,
-            context, instance=None, host=None, params=None):
-        """Generic handler for RPC casts and calls to compute.
-
-        :param rpc_method: RPC method to use (rpc.call or rpc.cast)
-        :param compute_method: Compute manager method to call
-        :param context: RequestContext of caller
-        :param instance: The instance object to use to find host to send to
-                         Can be None to not include instance_uuid in args
-        :param host: Optional host to send to instead of instance['host']
-                     Must be specified if 'instance' is None
-        :param params: Optional dictionary of arguments to be passed to the
-                       compute worker
-
-        :returns: None
-        """
-        if not params:
-            params = {}
-        if not host:
-            if not instance:
-                raise exception.Error(_("No compute host specified"))
-            host = instance['host']
-            if not host:
-                raise exception.Error(_("Unable to find host for "
-                        "Instance %s") % instance['uuid'])
-        queue = self.db.queue_get_for(context, FLAGS.compute_topic, host)
-        if instance:
-            params['instance_uuid'] = instance['uuid']
-        kwargs = {'method': compute_method, 'args': params}
-        return rpc_method(context, queue, kwargs)
-
-    def _cast_compute_message(self, *args, **kwargs):
-        """Generic handler for RPC casts to compute."""
-        self._cast_or_call_compute_message(rpc.cast, *args, **kwargs)
-
-    def _call_compute_message(self, *args, **kwargs):
-        """Generic handler for RPC calls to compute."""
-        return self._cast_or_call_compute_message(rpc.call, *args, **kwargs)
-
-    def _cast_scheduler_message(self, context, args):
-        """Generic handler for RPC calls to the scheduler."""
-        rpc.cast(context, FLAGS.scheduler_topic, args)
-
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.ACTIVE, vm_states.SHUTOFF],
                           task_state=[None, task_states.RESIZE_VERIFY])
@@ -1204,8 +1157,8 @@ class API(base.Base):
         recv_meta = self.image_service.create(context, sent_meta)
         params = {'image_id': recv_meta['id'], 'image_type': image_type,
                   'backup_type': backup_type, 'rotation': rotation}
-        self._cast_compute_message('snapshot_instance', context,
-                                   instance, params=params)
+        _cast_compute_message(self.db, 'snapshot_instance', context,
+                              instance, params=params)
         return recv_meta
 
     def _get_minram_mindisk_params(self, context, instance):
@@ -1239,10 +1192,8 @@ class API(base.Base):
                     instance,
                     vm_state=vm_states.ACTIVE,
                     task_state=state)
-        self._cast_compute_message('reboot_instance',
-                                   context,
-                                   instance,
-                                   params={'reboot_type': reboot_type})
+        _cast_compute_message(self.db, 'reboot_instance', context,
+                              instance, params={'reboot_type': reboot_type})
 
     def _validate_image_href(self, context, image_href):
         """Throws an ImageNotFound exception if image_href does not exist."""
@@ -1277,10 +1228,8 @@ class API(base.Base):
             "injected_files": files_to_inject,
         }
 
-        self._cast_compute_message('rebuild_instance',
-                                   context,
-                                   instance,
-                                   params=rebuild_params)
+        _cast_compute_message(self.db, 'rebuild_instance', context,
+                              instance, params=rebuild_params)
 
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.ACTIVE, vm_states.SHUTOFF],
@@ -1300,10 +1249,9 @@ class API(base.Base):
                     task_state=task_states.RESIZE_REVERTING)
 
         params = {'migration_id': migration_ref['id']}
-        self._cast_compute_message('revert_resize', context,
-                                   instance,
-                                   host=migration_ref['dest_compute'],
-                                   params=params)
+        _cast_compute_message(self.db, 'revert_resize', context, instance,
+                              host=migration_ref['dest_compute'],
+                              params=params)
 
         self.db.migration_update(context, migration_ref['id'],
                                  {'status': 'reverted'})
@@ -1326,10 +1274,9 @@ class API(base.Base):
                     task_state=None)
 
         params = {'migration_id': migration_ref['id']}
-        self._cast_compute_message('confirm_resize', context,
-                                   instance,
-                                   host=migration_ref['source_compute'],
-                                   params=params)
+        _cast_compute_message(self.db, 'confirm_resize', context, instance,
+                              host=migration_ref['source_compute'],
+                              params=params)
 
         self.db.migration_update(context, migration_ref['id'],
                 {'status': 'confirmed'})
@@ -1393,24 +1340,21 @@ class API(base.Base):
             "request_spec": utils.to_primitive(request_spec),
             "filter_properties": filter_properties,
         }
-        self._cast_scheduler_message(context, {"method": "prep_resize",
-                                     "args": args})
+        _cast_scheduler_message(context,
+                    {"method": "prep_resize",
+                     "args": args})
 
     @wrap_check_policy
     def add_fixed_ip(self, context, instance, network_id):
         """Add fixed_ip from specified network to given instance."""
-        self._cast_compute_message('add_fixed_ip_to_instance',
-                                   context,
-                                   instance,
-                                   params=dict(network_id=network_id))
+        _cast_compute_message(self.db, 'add_fixed_ip_to_instance', context,
+                              instance, params=dict(network_id=network_id))
 
     @wrap_check_policy
     def remove_fixed_ip(self, context, instance, address):
         """Remove fixed_ip from specified network to given instance."""
-        self._cast_compute_message('remove_fixed_ip_from_instance',
-                                   context,
-                                   instance,
-                                   params=dict(address=address))
+        _cast_compute_message(self.db, 'remove_fixed_ip_from_instance',
+                              context, instance, params=dict(address=address))
 
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.ACTIVE, vm_states.SHUTOFF,
@@ -1422,7 +1366,7 @@ class API(base.Base):
                     instance,
                     vm_state=vm_states.ACTIVE,
                     task_state=task_states.PAUSING)
-        self._cast_compute_message('pause_instance', context, instance)
+        _cast_compute_message(self.db, 'pause_instance', context, instance)
 
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.PAUSED])
@@ -1432,28 +1376,13 @@ class API(base.Base):
                     instance,
                     vm_state=vm_states.PAUSED,
                     task_state=task_states.UNPAUSING)
-        self._cast_compute_message('unpause_instance', context, instance)
-
-    def set_host_enabled(self, context, host, enabled):
-        """Sets the specified host's ability to accept new instances."""
-        # NOTE(comstud): No instance_uuid argument to this compute manager
-        # call
-        return self._call_compute_message("set_host_enabled", context,
-                host=host, params={"enabled": enabled})
-
-    def host_power_action(self, context, host, action):
-        """Reboots, shuts down or powers up the host."""
-        # NOTE(comstud): No instance_uuid argument to this compute manager
-        # call
-        return self._call_compute_message("host_power_action", context,
-                host=host, params={"action": action})
+        _cast_compute_message(self.db, 'unpause_instance', context, instance)
 
     @wrap_check_policy
     def get_diagnostics(self, context, instance):
         """Retrieve diagnostics for the given instance."""
-        return self._call_compute_message("get_diagnostics",
-                                          context,
-                                          instance)
+        return _call_compute_message(self.db, "get_diagnostics", context,
+                                     instance)
 
     @wrap_check_policy
     def get_actions(self, context, instance):
@@ -1470,7 +1399,7 @@ class API(base.Base):
                     instance,
                     vm_state=vm_states.ACTIVE,
                     task_state=task_states.SUSPENDING)
-        self._cast_compute_message('suspend_instance', context, instance)
+        _cast_compute_message(self.db, 'suspend_instance', context, instance)
 
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.SUSPENDED])
@@ -1480,7 +1409,7 @@ class API(base.Base):
                     instance,
                     vm_state=vm_states.SUSPENDED,
                     task_state=task_states.RESUMING)
-        self._cast_compute_message('resume_instance', context, instance)
+        _cast_compute_message(self.db, 'resume_instance', context, instance)
 
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.ACTIVE, vm_states.SHUTOFF,
@@ -1496,9 +1425,8 @@ class API(base.Base):
         rescue_params = {
             "rescue_password": rescue_password
         }
-        self._cast_compute_message('rescue_instance', context,
-                                   instance,
-                                   params=rescue_params)
+        _cast_compute_message(self.db, 'rescue_instance', context,
+                              instance, params=rescue_params)
 
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.RESCUED])
@@ -1508,8 +1436,7 @@ class API(base.Base):
                     instance,
                     vm_state=vm_states.RESCUED,
                     task_state=task_states.UNRESCUING)
-        self._cast_compute_message('unrescue_instance', context,
-                                   instance)
+        _cast_compute_message(self.db, 'unrescue_instance', context, instance)
 
     @wrap_check_policy
     @check_instance_state(vm_state=[vm_states.ACTIVE])
@@ -1520,24 +1447,22 @@ class API(base.Base):
                     task_state=task_states.UPDATING_PASSWORD)
 
         params = {"new_pass": password}
-        self._cast_compute_message('set_admin_password', context,
-                                   instance,
-                                   params=params)
+        _cast_compute_message(self.db, 'set_admin_password', context,
+                              instance, params=params)
 
     @wrap_check_policy
     def inject_file(self, context, instance, path, file_contents):
         """Write a file to the given instance."""
         params = {'path': path, 'file_contents': file_contents}
-        self._cast_compute_message('inject_file', context,
+        _cast_compute_message(self.db, 'inject_file', context,
                                    instance, params=params)
 
     @wrap_check_policy
     def get_vnc_console(self, context, instance, console_type):
         """Get a url to an instance Console."""
-        connect_info = self._call_compute_message('get_vnc_console',
-                                        context,
-                                        instance,
-                                        params={"console_type": console_type})
+        connect_info = _call_compute_message(self.db, 'get_vnc_console',
+                                    context, instance,
+                                    params={"console_type": console_type})
 
         rpc.call(context, '%s' % FLAGS.consoleauth_topic,
                  {'method': 'authorize_console',
@@ -1554,22 +1479,18 @@ class API(base.Base):
     def get_console_output(self, context, instance, tail_length=None):
         """Get console output for an an instance."""
         params = {'tail_length': tail_length}
-        return self._call_compute_message('get_console_output',
-                                          context,
-                                          instance,
-                                          params=params)
+        return _call_compute_message(self.db, 'get_console_output', context,
+                                     instance, params=params)
 
     @wrap_check_policy
     def lock(self, context, instance):
         """Lock the given instance."""
-        self._cast_compute_message('lock_instance', context, instance)
+        _cast_compute_message(self.db, 'lock_instance', context, instance)
 
     @wrap_check_policy
     def unlock(self, context, instance):
         """Unlock the given instance."""
-        self._cast_compute_message('unlock_instance',
-                                   context,
-                                   instance)
+        _cast_compute_message(self.db, 'unlock_instance', context, instance)
 
     @wrap_check_policy
     def get_lock(self, context, instance):
@@ -1579,13 +1500,13 @@ class API(base.Base):
     @wrap_check_policy
     def reset_network(self, context, instance):
         """Reset networking on the instance."""
-        self._cast_compute_message('reset_network', context, instance)
+        _cast_compute_message(self.db, 'reset_network', context, instance)
 
     @wrap_check_policy
     def inject_network_info(self, context, instance):
         """Inject network info for the instance."""
-        self._cast_compute_message('inject_network_info', context,
-                                   instance)
+        _cast_compute_message(self.db, 'inject_network_info',
+                              context, instance)
 
     @wrap_check_policy
     def attach_volume(self, context, instance, volume_id, device):
@@ -1596,9 +1517,8 @@ class API(base.Base):
         self.volume_api.check_attach(context, volume)
         params = {"volume_id": volume_id,
                   "mountpoint": device}
-        self._cast_compute_message('attach_volume', context,
-                                   instance,
-                                   params=params)
+        _cast_compute_message(self.db, 'attach_volume', context, instance,
+                              params=params)
 
     # FIXME(comstud): I wonder if API should pull in the instance from
     # the volume ID via volume API and pass it and the volume object here
@@ -1614,9 +1534,8 @@ class API(base.Base):
         self.volume_api.check_detach(context, volume)
 
         params = {'volume_id': volume_id}
-        self._cast_compute_message('detach_volume', context,
-                                   instance,
-                                   params=params)
+        _cast_compute_message(self.db, 'detach_volume', context, instance,
+                              params=params)
         return instance
 
     @wrap_check_policy
@@ -1699,6 +1618,31 @@ class API(base.Base):
         return self.db.instance_fault_get_by_instance_uuids(context, uuids)
 
 
+class HostAPI(base.Base):
+    """Sub-set of the Compute Manager API for managing host operations."""
+    def __init__(self, **kwargs):
+        super(HostAPI, self).__init__(**kwargs)
+
+    def set_host_enabled(self, context, host, enabled):
+        """Sets the specified host's ability to accept new instances."""
+        # NOTE(comstud): No instance_uuid argument to this compute manager
+        # call
+        return _call_compute_message(self.db, "set_host_enabled", context,
+                                     host=host, params={"enabled": enabled})
+
+    def host_power_action(self, context, host, action):
+        """Reboots, shuts down or powers up the host."""
+        # NOTE(comstud): No instance_uuid argument to this compute manager
+        # call
+        return _call_compute_message(self.db, "host_power_action", context,
+                                     host=host, params={"action": action})
+
+    def set_host_maintenance(self, context, host, mode):
+        """Start/Stop host maintenance window. On start, it triggers
+        guest VMs evacuation."""
+        raise NotImplementedError()
+
+
 class AggregateAPI(base.Base):
     """Sub-set of the Compute Manager API for managing host aggregates."""
     def __init__(self, **kwargs):
@@ -1832,3 +1776,50 @@ class AggregateAPI(base.Base):
         result["metadata"] = metadata
         result["hosts"] = hosts
         return result
+
+
+def _cast_or_call_compute_message(db, rpc_method, compute_method,
+        context, instance=None, host=None, params=None):
+    """Generic handler for RPC casts and calls to compute.
+
+    :param rpc_method: RPC method to use (rpc.call or rpc.cast)
+    :param compute_method: Compute manager method to call
+    :param context: RequestContext of caller
+    :param instance: The instance object to use to find host to send to
+                     Can be None to not include instance_uuid in args
+    :param host: Optional host to send to instead of instance['host']
+                 Must be specified if 'instance' is None
+    :param params: Optional dictionary of arguments to be passed to the
+                   compute worker
+
+    :returns: None
+    """
+    if not params:
+        params = {}
+    if not host:
+        if not instance:
+            raise exception.Error(_("No compute host specified"))
+        host = instance['host']
+        if not host:
+            raise exception.Error(_("Unable to find host for "
+                                    "Instance %s") % instance['uuid'])
+    queue = db.queue_get_for(context, FLAGS.compute_topic, host)
+    if instance:
+        params['instance_uuid'] = instance['uuid']
+    kwargs = {'method': compute_method, 'args': params}
+    return rpc_method(context, queue, kwargs)
+
+
+def _cast_compute_message(db, *args, **kwargs):
+    """Generic handler for RPC casts to compute."""
+    _cast_or_call_compute_message(db, rpc.cast, *args, **kwargs)
+
+
+def _call_compute_message(db, *args, **kwargs):
+    """Generic handler for RPC calls to compute."""
+    return _cast_or_call_compute_message(db, rpc.call, *args, **kwargs)
+
+
+def _cast_scheduler_message(context, args):
+    """Generic handler for RPC calls to the scheduler."""
+    rpc.cast(context, FLAGS.scheduler_topic, args)