Move remaining aggregate operations to conductor

This patch moves the following aggregate operations from compute
to conductor:

  aggregate_get_by_host()
  aggregate_metadata_add()
  aggregate_metadata_delete()

In order to do that, it also changes the signature of the corresponding
VirtAPI methods to take the aggregate object instead of just the id,
so that we don't re-introduce any db messaging behavior.

I debated about using the set_delete mechanism in the metadata_add
API to avoid adding a separate conductor API for delete, but decided
that it wasn't worth the extra round trips, no matter how much I may
wish it were.

Change-Id: Ic0117879c8ce576cdc8e6c5af018bb918a15d4c0

8 files changed, 116 insertions, 23 deletions

diff --git a/nova/compute/manager.py b/nova/compute/manager.py
index 2f2d2fd95..616083079 100644
--- a/nova/compute/manager.py
+++ b/nova/compute/manager.py
@@ -266,17 +266,17 @@ class ComputeVirtAPI(virtapi.VirtAPI):
             context, host)
 
     def aggregate_get_by_host(self, context, host, key=None):
-        return self._compute.db.aggregate_get_by_host(context, host, key=key)
+        return self._compute.conductor_api.aggregate_get_by_host(context,
+                                                                 host, key=key)
 
-    def aggregate_metadata_add(self, context, aggregate_id, metadata,
+    def aggregate_metadata_add(self, context, aggregate, metadata,
                                set_delete=False):
-        return self._compute.db.aggregate_metadata_add(context, aggregate_id,
-                                                       metadata,
-                                                       set_delete=set_delete)
+        return self._compute.conductor_api.aggregate_metadata_add(
+            context, aggregate, metadata, set_delete=set_delete)
 
-    def aggregate_metadata_delete(self, context, aggregate_id, key):
-        return self._compute.db.aggregate_metadata_delete(context,
-                                                          aggregate_id, key)
+    def aggregate_metadata_delete(self, context, aggregate, key):
+        return self._compute.conductor_api.aggregate_metadata_delete(
+            context, aggregate, key)
 
     def security_group_get_by_instance(self, context, instance_uuid):
         return self._compute.db.security_group_get_by_instance(context,
diff --git a/nova/conductor/api.py b/nova/conductor/api.py
index b9013c4c6..7a5e0fce9 100644
--- a/nova/conductor/api.py
+++ b/nova/conductor/api.py
@@ -66,6 +66,20 @@ class LocalAPI(object):
     def aggregate_host_delete(self, context, aggregate, host):
         return self._manager.aggregate_host_delete(context, aggregate, host)
 
+    def aggregate_get_by_host(self, context, host, key=None):
+        return self._manager.aggregate_get_by_host(context, host, key)
+
+    def aggregate_metadata_add(self, context, aggregate, metadata,
+                               set_delete=False):
+        return self._manager.aggregate_metadata_add(context, aggregate,
+                                                    metadata,
+                                                    set_delete)
+
+    def aggregate_metadata_delete(self, context, aggregate, key):
+        return self._manager.aggregate_metadata_delete(context,
+                                                       aggregate,
+                                                       key)
+
     def bw_usage_get(self, context, uuid, start_period, mac):
         return self._manager.bw_usage_update(context, uuid, mac, start_period)
 
@@ -114,6 +128,20 @@ class API(object):
         return self.conductor_rpcapi.aggregate_host_delete(context, aggregate,
                                                            host)
 
+    def aggregate_get_by_host(self, context, host, key=None):
+        return self.conductor_rpcapi.aggregate_get_by_host(context, host, key)
+
+    def aggregate_metadata_add(self, context, aggregate, metadata,
+                               set_delete=False):
+        return self.conductor_rpcapi.aggregate_metadata_add(context, aggregate,
+                                                            metadata,
+                                                            set_delete)
+
+    def aggregate_metadata_delete(self, context, aggregate, key):
+        return self.conductor_rpcapi.aggregate_metadata_delete(context,
+                                                               aggregate,
+                                                               key)
+
     def bw_usage_get(self, context, uuid, start_period, mac):
         return self.conductor_rpcapi.bw_usage_update(context, uuid, mac,
                                                      start_period)
diff --git a/nova/conductor/manager.py b/nova/conductor/manager.py
index 796b99360..84d148792 100644
--- a/nova/conductor/manager.py
+++ b/nova/conductor/manager.py
@@ -41,7 +41,7 @@ datetime_fields = ['launched_at', 'terminated_at']
 class ConductorManager(manager.SchedulerDependentManager):
     """Mission: TBD"""
 
-    RPC_API_VERSION = '1.6'
+    RPC_API_VERSION = '1.7'
 
     def __init__(self, *args, **kwargs):
         super(ConductorManager, self).__init__(service_name='conductor',
@@ -90,6 +90,22 @@ class ConductorManager(manager.SchedulerDependentManager):
         self.db.aggregate_host_delete(context.elevated(),
                 aggregate['id'], host)
 
+    def aggregate_get_by_host(self, context, host, key=None):
+        aggregates = self.db.aggregate_get_by_host(context.elevated(),
+                                                   host, key)
+        return jsonutils.to_primitive(aggregates)
+
+    def aggregate_metadata_add(self, context, aggregate, metadata,
+                               set_delete=False):
+        new_metadata = self.db.aggregate_metadata_add(context.elevated(),
+                                                      aggregate['id'],
+                                                      metadata, set_delete)
+        return jsonutils.to_primitive(new_metadata)
+
+    def aggregate_metadata_delete(self, context, aggregate, key):
+        self.db.aggregate_metadata_delete(context.elevated(),
+                                          aggregate['id'], key)
+
     def bw_usage_update(self, context, uuid, mac, start_period,
                         bw_in=None, bw_out=None,
                         last_ctr_in=None, last_ctr_out=None,
diff --git a/nova/conductor/rpcapi.py b/nova/conductor/rpcapi.py
index 1f38b0ea5..04bff83c7 100644
--- a/nova/conductor/rpcapi.py
+++ b/nova/conductor/rpcapi.py
@@ -33,6 +33,8 @@ class ConductorAPI(nova.openstack.common.rpc.proxy.RpcProxy):
     1.4 - Added migration_get
     1.5 - Added bw_usage_update
     1.6 - Added get_backdoor_port()
+    1.7 - Added aggregate_get_by_host, aggregate_metadata_add,
+          and aggregate_metadata_delete
     """
 
     BASE_RPC_API_VERSION = '1.0'
@@ -80,6 +82,24 @@ class ConductorAPI(nova.openstack.common.rpc.proxy.RpcProxy):
                             host=host)
         return self.call(context, msg, version='1.3')
 
+    def aggregate_get_by_host(self, context, host, key=None):
+        msg = self.make_msg('aggregate_get_by_host', host=host, key=key)
+        return self.call(context, msg, version='1.7')
+
+    def aggregate_metadata_add(self, context, aggregate, metadata,
+                               set_delete=False):
+        aggregate_p = jsonutils.to_primitive(aggregate)
+        msg = self.make_msg('aggregate_metadata_add', aggregate=aggregate_p,
+                            metadata=metadata,
+                            set_delete=set_delete)
+        return self.call(context, msg, version='1.7')
+
+    def aggregate_metadata_delete(self, context, aggregate, key):
+        aggregate_p = jsonutils.to_primitive(aggregate)
+        msg = self.make_msg('aggregate_metadata_delete', aggregate=aggregate_p,
+                            key=key)
+        return self.call(context, msg, version='1.7')
+
     def bw_usage_update(self, context, uuid, mac, start_period,
                         bw_in=None, bw_out=None,
                         last_ctr_in=None, last_ctr_out=None,
diff --git a/nova/tests/conductor/test_conductor.py b/nova/tests/conductor/test_conductor.py
index cace21984..04cf95d5e 100644
--- a/nova/tests/conductor/test_conductor.py
+++ b/nova/tests/conductor/test_conductor.py
@@ -14,6 +14,8 @@
 
 """Tests for the conductor service"""
 
+import mox
+
 from nova.compute import instance_types
 from nova.compute import vm_states
 from nova import conductor
@@ -155,6 +157,33 @@ class ConductorTestCase(BaseTestCase):
 
         db.aggregate_delete(self.context.elevated(), aggregate_ref['id'])
 
+    def test_aggregate_get_by_host(self):
+        self._setup_aggregate_with_host()
+        aggregates = self.conductor.aggregate_get_by_host(self.context, 'bar')
+        self.assertEqual(aggregates[0]['availability_zone'], 'foo')
+
+    def test_aggregate_metadata_add(self):
+        aggregate = {'name': 'fake aggregate', 'id': 'fake-id'}
+        metadata = {'foo': 'bar'}
+        self.mox.StubOutWithMock(db, 'aggregate_metadata_add')
+        db.aggregate_metadata_add(
+            mox.IgnoreArg(), aggregate['id'], metadata, False).AndReturn(
+                metadata)
+        self.mox.ReplayAll()
+        result = self.conductor.aggregate_metadata_add(self.context,
+                                                       aggregate,
+                                                       metadata)
+        self.assertEqual(result, metadata)
+
+    def test_aggregate_metadata_delete(self):
+        aggregate = {'name': 'fake aggregate', 'id': 'fake-id'}
+        self.mox.StubOutWithMock(db, 'aggregate_metadata_delete')
+        db.aggregate_metadata_delete(mox.IgnoreArg(), aggregate['id'], 'fake')
+        self.mox.ReplayAll()
+        result = self.conductor.aggregate_metadata_delete(self.context,
+                                                       aggregate,
+                                                       'fake')
+
     def test_bw_usage_update(self):
         self.mox.StubOutWithMock(db, 'bw_usage_update')
         self.mox.StubOutWithMock(db, 'bw_usage_get')
diff --git a/nova/virt/fake.py b/nova/virt/fake.py
index c354b34d8..6f95256be 100644
--- a/nova/virt/fake.py
+++ b/nova/virt/fake.py
@@ -412,13 +412,13 @@ class FakeVirtAPI(virtapi.VirtAPI):
     def aggregate_get_by_host(self, context, host, key=None):
         return db.aggregate_get_by_host(context, host, key)
 
-    def aggregate_metadata_add(self, context, aggregate_id, metadata,
+    def aggregate_metadata_add(self, context, aggregate, metadata,
                                set_delete=False):
-        return db.aggregate_metadata_add(context, aggregate_id, metadata,
+        return db.aggregate_metadata_add(context, aggregate['id'], metadata,
                                          set_delete)
 
-    def aggregate_metadata_delete(self, context, aggregate_id, key):
-        return db.aggregate_metadata_delete(context, aggregate_id, key)
+    def aggregate_metadata_delete(self, context, aggregate, key):
+        return db.aggregate_metadata_delete(context, aggregate['id'], key)
 
     def security_group_get_by_instance(self, context, instance_uuid):
         return db.security_group_get_by_instance(context, instance_uuid)
diff --git a/nova/virt/virtapi.py b/nova/virt/virtapi.py
index 59a4006a6..24a66f53b 100644
--- a/nova/virt/virtapi.py
+++ b/nova/virt/virtapi.py
@@ -48,20 +48,20 @@ class VirtAPI(object):
         """
         raise NotImplementedError()
 
-    def aggregate_metadata_add(self, context, aggregate_id, metadata,
+    def aggregate_metadata_add(self, context, aggregate, metadata,
                                set_delete=False):
         """Add/update metadata for specified aggregate
         :param context: security context
-        :param aggregate_id: id of aggregate on which to update metadata
+        :param aggregate: aggregate on which to update metadata
         :param metadata: dict of metadata to add/update
         :param set_delete: if True, only add
         """
         raise NotImplementedError()
 
-    def aggregate_metadata_delete(self, context, aggregate_id, key):
+    def aggregate_metadata_delete(self, context, aggregate, key):
         """Delete the given metadata key from specified aggregate
         :param context: security context
-        :param aggregate_id: id of aggregate from which to delete metadata
+        :param aggregate: aggregate from which to delete metadata
         :param key: metadata key to delete
         """
         raise NotImplementedError()
diff --git a/nova/virt/xenapi/pool.py b/nova/virt/xenapi/pool.py
index 8d0e0a91c..5e6a8195c 100644
--- a/nova/virt/xenapi/pool.py
+++ b/nova/virt/xenapi/pool.py
@@ -63,7 +63,7 @@ class ResourcePool(object):
         try:
             if set_error:
                 metadata = {pool_states.KEY: pool_states.ERROR}
-                self._virtapi.aggregate_metadata_add(context, aggregate['id'],
+                self._virtapi.aggregate_metadata_add(context, aggregate,
                                                      metadata)
             op(context, aggregate, host)
         except Exception:
@@ -87,7 +87,7 @@ class ResourcePool(object):
                     reason=aggregate['metadetails'][pool_states.KEY])
 
         if (aggregate['metadetails'][pool_states.KEY] == pool_states.CREATED):
-            self._virtapi.aggregate_metadata_add(context, aggregate['id'],
+            self._virtapi.aggregate_metadata_add(context, aggregate,
                                                  {pool_states.KEY:
                                                       pool_states.CHANGING})
         if len(aggregate['hosts']) == 1:
@@ -97,7 +97,7 @@ class ResourcePool(object):
             metadata = {'master_compute': host,
                         host: self._host_uuid,
                         pool_states.KEY: pool_states.ACTIVE}
-            self._virtapi.aggregate_metadata_add(context, aggregate['id'],
+            self._virtapi.aggregate_metadata_add(context, aggregate,
                                                  metadata)
         else:
             # the pool is already up and running, we need to figure out
@@ -112,7 +112,7 @@ class ResourcePool(object):
                                  slave_info.get('url'), slave_info.get('user'),
                                  slave_info.get('passwd'))
                 metadata = {host: slave_info.get('xenhost_uuid'), }
-                self._virtapi.aggregate_metadata_add(context, aggregate['id'],
+                self._virtapi.aggregate_metadata_add(context, aggregate,
                                                      metadata)
             elif master_compute and master_compute != host:
                 # send rpc cast to master, asking to add the following
@@ -143,7 +143,7 @@ class ResourcePool(object):
             host_uuid = aggregate['metadetails'][host]
             self._eject_slave(aggregate['id'],
                               slave_info.get('compute_uuid'), host_uuid)
-            self._virtapi.aggregate_metadata_delete(context, aggregate['id'],
+            self._virtapi.aggregate_metadata_delete(context, aggregate,
                                                     host)
         elif master_compute == host:
             # Remove master from its own pool -> destroy pool only if the
@@ -161,7 +161,7 @@ class ResourcePool(object):
             self._clear_pool(aggregate['id'])
             for key in ['master_compute', host]:
                 self._virtapi.aggregate_metadata_delete(context,
-                        aggregate['id'], key)
+                        aggregate, key)
         elif master_compute and master_compute != host:
             # A master exists -> forward pool-eject request to master
             slave_info = self._create_slave_info()