Merge "Added resources support in pollster's interface"

ceilometer/agent.py

@@ -17,6 +17,7 @@
 # under the License.
 
 import abc
+import collections
 import itertools
 
 from ceilometer.openstack.common import context
@@ -37,11 +38,17 @@ class PollingTask(object):
     def __init__(self, agent_manager):
         self.manager = agent_manager
         self.pollsters = set()
+        # Resource definitions are indexed by the pollster
+        # Use dict of set here to remove the duplicated resource definitions
+        # for each pollster.
+        self.resources = collections.defaultdict(set)
         self.publish_context = pipeline.PublishContext(
             agent_manager.context)
 
     def add(self, pollster, pipelines):
         self.publish_context.add_pipelines(pipelines)
+        for pipeline in pipelines:
+            self.resources[pollster.name].update(pipeline.resources)
         self.pollsters.update([pollster])
 
     @abc.abstractmethod

ceilometer/central/manager.py

@@ -41,9 +41,11 @@ class PollingTask(agent.PollingTask):
             for pollster in self.pollsters:
                 try:
                     LOG.info(_("Polling pollster %s"), pollster.name)
+                    resources = list(self.resources[pollster.name])
                     samples = list(pollster.obj.get_samples(
                         self.manager,
                         cache,
+                        resources=resources,
                     ))
                     publisher(samples)
                 except Exception as err:

ceilometer/energy/kwapi.py

@@ -82,7 +82,7 @@ class _Base(plugin.CentralPollster):
 class EnergyPollster(_Base):
     """Measures energy consumption."""
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         """Returns all samples."""
         for probe in self._iter_probes(manager.keystone, cache):
             yield sample.Sample(
@@ -102,7 +102,7 @@ class EnergyPollster(_Base):
 class PowerPollster(_Base):
     """Measures power consumption."""
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         """Returns all samples."""
         for probe in self._iter_probes(manager.keystone, cache):
             yield sample.Sample(

ceilometer/image/glance.py

@@ -101,7 +101,7 @@ class _Base(plugin.PollsterBase):
 
 class ImagePollster(_Base):
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         for image in self._iter_images(manager.keystone, cache):
             yield sample.Sample(
                 name='image',
@@ -118,7 +118,7 @@ class ImagePollster(_Base):
 
 class ImageSizePollster(_Base):
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         for image in self._iter_images(manager.keystone, cache):
             yield sample.Sample(
                 name='image.size',

ceilometer/network/floatingip.py

@@ -40,7 +40,7 @@ class FloatingIPPollster(plugin.CentralPollster):
             cache['floating_ips'] = list(self._get_floating_ips())
         return iter(cache['floating_ips'])
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         for ip in self._iter_floating_ips(cache):
             self.LOG.info(_("FLOATING IP USAGE: %s") % ip.ip)
             # FIXME (flwang) Now Nova API /os-floating-ips can't provide those

ceilometer/objectstore/swift.py

@@ -88,7 +88,7 @@ class ObjectsPollster(_Base):
     """Iterate over all accounts, using keystone.
     """
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         for tenant, account in self._iter_accounts(manager.keystone, cache):
             yield sample.Sample(
                 name='storage.objects',
@@ -107,7 +107,7 @@ class ObjectsSizePollster(_Base):
     """Iterate over all accounts, using keystone.
     """
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         for tenant, account in self._iter_accounts(manager.keystone, cache):
             yield sample.Sample(
                 name='storage.objects.size',
@@ -126,7 +126,7 @@ class ObjectsContainersPollster(_Base):
     """Iterate over all accounts, using keystone.
     """
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         for tenant, account in self._iter_accounts(manager.keystone, cache):
             yield sample.Sample(
                 name='storage.objects.containers',
@@ -147,7 +147,7 @@ class ContainersObjectsPollster(_Base):
 
     METHOD = 'get'
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         for project, account in self._iter_accounts(manager.keystone, cache):
             containers_info = account[1]
             for container in containers_info:
@@ -170,7 +170,7 @@ class ContainersSizePollster(_Base):
 
     METHOD = 'get'
 
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         for project, account in self._iter_accounts(manager.keystone, cache):
             containers_info = account[1]
             for container in containers_info:

ceilometer/plugin.py

@@ -21,6 +21,7 @@
 import abc
 import collections
 import fnmatch
+
 from oslo.config import cfg
 import six
 
@@ -90,7 +91,7 @@ class PollsterBase(PluginBase):
     """Base class for plugins that support the polling API."""
 
     @abc.abstractmethod
-    def get_samples(self, manager, cache):
+    def get_samples(self, manager, cache, resources=[]):
         """Return a sequence of Counter instances from polling the resources.
 
         :param manager: The service manager class invoking the plugin.
@@ -98,5 +99,8 @@ class PollsterBase(PluginBase):
                       between themselves when recomputing it would be
                       expensive (e.g., asking another service for a
                       list of objects).
+        :param resources: A list of the endpoints the pollster will get data
+                          from. It's up to the specific pollster to decide
+                          how to use it.
 
         """

ceilometer/tests/agentbase.py

@@ -29,6 +29,7 @@ from stevedore.tests import manager as extension_tests
 from ceilometer import agent
 from ceilometer.openstack.common.fixture import config
 from ceilometer import pipeline
+from ceilometer import plugin
 from ceilometer import sample
 from ceilometer.tests import base
 from ceilometer import transformer
@@ -47,21 +48,23 @@ default_test_data = sample.Sample(
 )
 
 
-class TestPollster:
+class TestPollster(plugin.PollsterBase):
     test_data = default_test_data
 
-    def get_samples(self, manager, cache, instance=None):
+    def get_samples(self, manager, cache, instance=None, resources=[]):
         self.samples.append((manager, instance))
+        self.resources.extend(resources)
         return [self.test_data]
 
 
 class TestPollsterException(TestPollster):
-    def get_samples(self, manager, cache, instance=None):
+    def get_samples(self, manager, cache, instance=None, resources=[]):
         # Put an instance parameter here so that it can be used
         # by both central manager and compute manager
         # In future, we possibly don't need such hack if we
         # combine the get_samples() function again
         self.samples.append((manager, instance))
+        self.resources.extend(resources)
         raise Exception()
 
 
@@ -69,10 +72,12 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
 
     class Pollster(TestPollster):
         samples = []
+        resources = []
         test_data = default_test_data
 
     class PollsterAnother(TestPollster):
         samples = []
+        resources = []
         test_data = sample.Sample(
             name='testanother',
             type=default_test_data.type,
@@ -86,6 +91,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
 
     class PollsterException(TestPollsterException):
         samples = []
+        resources = []
         test_data = sample.Sample(
             name='testexception',
             type=default_test_data.type,
@@ -99,6 +105,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
 
     class PollsterExceptionAnother(TestPollsterException):
         samples = []
+        resources = []
         test_data = sample.Sample(
             name='testexceptionanother',
             type=default_test_data.type,
@@ -159,6 +166,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
             'name': "test_pipeline",
             'interval': 60,
             'counters': ['test'],
+            'resources': ['test://'],
             'transformers': [],
             'publishers': ["test"],
         }, ]
@@ -174,12 +182,18 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
         self.PollsterAnother.samples = []
         self.PollsterException.samples = []
         self.PollsterExceptionAnother.samples = []
+        self.Pollster.resources = []
+        self.PollsterAnother.resources = []
+        self.PollsterException.resources = []
+        self.PollsterExceptionAnother.resources = []
         super(BaseAgentManagerTestCase, self).tearDown()
 
     def test_setup_polling_tasks(self):
         polling_tasks = self.mgr.setup_polling_tasks()
         self.assertEqual(len(polling_tasks), 1)
         self.assertTrue(60 in polling_tasks.keys())
+        self.assertEqual(len(polling_tasks[60].resources), 1)
+        self.assertEqual(len(polling_tasks[60].resources['test']), 1)
         self.mgr.interval_task(polling_tasks.values()[0])
         pub = self.mgr.pipeline_manager.pipelines[0].publishers[0]
         self.assertEqual(pub.samples[0], self.Pollster.test_data)
@@ -189,6 +203,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
             'name': "test_pipeline",
             'interval': 10,
             'counters': ['test'],
+            'resources': ['test://'],
             'transformers': [],
             'publishers': ["test"],
         })
@@ -204,6 +219,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
                 'name': "test_pipeline_1",
                 'interval': 10,
                 'counters': ['test_invalid'],
+                'resources': ['invalid://'],
                 'transformers': [],
                 'publishers': ["test"],
             })
@@ -216,6 +232,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
             'name': "test_pipeline",
             'interval': 60,
             'counters': ['testanother'],
+            'resources': ['testanother://'],
             'transformers': [],
             'publishers': ["test"],
         })
@@ -224,6 +241,9 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
         self.assertEqual(len(polling_tasks), 1)
         pollsters = polling_tasks.get(60).pollsters
         self.assertEqual(len(pollsters), 2)
+        self.assertEqual(len(polling_tasks[60].resources), 2)
+        self.assertEqual(len(polling_tasks[60].resources['test']), 1)
+        self.assertEqual(len(polling_tasks[60].resources['testanother']), 1)
 
     def test_interval_exception_isolation(self):
         self.pipeline_cfg = [
@@ -231,6 +251,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
                 'name': "test_pipeline_1",
                 'interval': 10,
                 'counters': ['testexceptionanother'],
+                'resources': ['test://'],
                 'transformers': [],
                 'publishers': ["test"],
             },
@@ -238,6 +259,7 @@ class BaseAgentManagerTestCase(base.BaseTestCase):
                 'name': "test_pipeline_2",
                 'interval': 10,
                 'counters': ['testexception'],
+                'resources': ['test://'],
                 'transformers': [],
                 'publishers': ["test"],
             },

ceilometer/tests/central/test_manager.py

@@ -44,3 +44,8 @@ class TestRunTasks(agentbase.BaseAgentManagerTestCase):
         self.useFixture(mockpatch.Patch(
             'keystoneclient.v2_0.client.Client',
             return_value=None))
+
+    def test_get_sample_resources(self):
+        polling_tasks = self.mgr.setup_polling_tasks()
+        self.mgr.interval_task(polling_tasks.values()[0])
+        self.assertTrue(self.Pollster.resources)