Merge "gnocchi: reduce the number of patch to gnocchi API"

ceilometer/dispatcher/gnocchi.py

@@ -320,89 +320,79 @@ class GnocchiDispatcher(dispatcher.Base):
             data, key=operator.itemgetter('resource_id'))
 
         for resource_id, samples_of_resource in resource_grouped_samples:
-            resource_need_to_be_updated = True
-
             metric_grouped_samples = itertools.groupby(
                 list(samples_of_resource),
                 key=operator.itemgetter('counter_name'))
-            for metric_name, samples in metric_grouped_samples:
-                samples = list(samples)
-                rd = self._get_resource_definition(metric_name)
-                if rd:
-                    self._process_samples(rd, resource_id, metric_name,
-                                          samples,
-                                          resource_need_to_be_updated)
-                else:
-                    LOG.warn("metric %s is not handled by gnocchi" %
-                             metric_name)
 
-                # FIXME(sileht): Does it reasonable to skip the resource
+            self._process_resource(resource_id, metric_grouped_samples)
-                # update here ? Does differents kind of counter_name
-                # can have different metadata set ?
-                # (ie: one have only flavor_id, and an other one have only
-                # image_ref ?)
-                #
-                # resource_need_to_be_updated = False
 
     @log_and_ignore_unexpected_workflow_error
-    def _process_samples(self, resource_def, resource_id, metric_name, samples,
+    def _process_resource(self, resource_id, metric_grouped_samples):
-                         resource_need_to_be_updated):
+        # TODO(sileht): Any HTTP 50X/401 error is just logged and this method
-        resource_type = resource_def.cfg['resource_type']
+        # stop, perhaps we can be smarter and retry later in case of 50X and
-        measure_attributes = [{'timestamp': sample['timestamp'],
+        # directly in case of 401. A gnocchiclient would help a lot for the
-                               'value': sample['counter_volume']}
+        # latest.
-                              for sample in samples]
+
+        resource_extra = {}
+        for metric_name, samples in metric_grouped_samples:
+            samples = list(samples)
+            rd = self._get_resource_definition(metric_name)
+            if rd is None:
+                LOG.warn("metric %s is not handled by gnocchi" %
+                         metric_name)
+                continue
+
+            resource_type = rd.cfg['resource_type']
+            resource = {
+                "id": resource_id,
+                "user_id": samples[0]['user_id'],
+                "project_id": samples[0]['project_id'],
+                "metrics": rd.metrics(),
+            }
+            measures = []
+
+            for sample in samples:
+                resource_extra.update(rd.attributes(sample))
+                measures.append({'timestamp': sample['timestamp'],
+                                 'value': sample['counter_volume']})
+
+            resource.update(resource_extra)
 
-        try:
-            self._post_measure(resource_type, resource_id, metric_name,
-                               measure_attributes)
-        except NoSuchMetric:
-            # NOTE(sileht): we try first to create the resource, because
-            # they more chance that the resource doesn't exists than the metric
-            # is missing, the should be reduce the number of resource API call
-            resource_attributes = self._get_resource_attributes(
-                resource_def, resource_id, metric_name, samples)
             try:
-                self._create_resource(resource_type, resource_id,
+                self._post_measure(resource_type, resource_id, metric_name,
-                                      resource_attributes)
+                                   measures)
-            except ResourceAlreadyExists:
+            except NoSuchMetric:
+                # TODO(sileht): Make gnocchi smarter to be able to detect 404
+                # for 'resource doesn't exist' and for 'metric doesn't exist'
+                # https://bugs.launchpad.net/gnocchi/+bug/1476186
+                self._ensure_resource_and_metric(resource_type, resource,
+                                                 metric_name)
+
                 try:
-                    archive_policy = (resource_def.metrics()[metric_name])
+                    self._post_measure(resource_type, resource_id,
-                    self._create_metric(resource_type, resource_id,
+                                       metric_name, measures)
-                                        metric_name, archive_policy)
+                except NoSuchMetric:
-                except MetricAlreadyExists:
+                    LOG.error(_LE("Fail to post measures for "
-                    # NOTE(sileht): Just ignore the metric have been created in
+                                  "%(resource_id)s/%(metric_name)s") %
-                    # the meantime.
+                              dict(resource_id=resource_id,
-                    pass
+                                   metric_name=metric_name))
-            else:
-                # No need to update it we just created it
-                # with everything we need
-                resource_need_to_be_updated = False
 
-            # NOTE(sileht): we retry to post the measure but if it fail we
+        if resource_extra:
-            # don't catch the exception to just log it and continue to process
+            self._update_resource(resource_type, resource_id, resource_extra)
-            # other samples
-            self._post_measure(resource_type, resource_id, metric_name,
-                               measure_attributes)
 
-        if resource_need_to_be_updated:
+    def _ensure_resource_and_metric(self, resource_type, resource,
-            resource_attributes = self._get_resource_attributes(
+                                    metric_name):
-                resource_def, resource_id, metric_name, samples,
+        try:
-                for_update=True)
+            self._create_resource(resource_type, resource)
-            if resource_attributes:
+        except ResourceAlreadyExists:
-                self._update_resource(resource_type, resource_id,
+            try:
-                                      resource_attributes)
+                archive_policy = resource['metrics'][metric_name]
-
+                self._create_metric(resource_type, resource['id'],
-    def _get_resource_attributes(self, resource_def, resource_id, metric_name,
+                                    metric_name, archive_policy)
-                                 samples, for_update=False):
+            except MetricAlreadyExists:
-        # FIXME(sileht): Should I merge attibutes of all samples ?
+                # NOTE(sileht): Just ignore the metric have been
-        # Or keep only the last one is sufficient ?
+                # created in the meantime.
-        attributes = resource_def.attributes(samples[-1])
+                pass
-        if not for_update:
-            attributes["id"] = resource_id
-            attributes["user_id"] = samples[-1]['user_id']
-            attributes["project_id"] = samples[-1]['project_id']
-            attributes["metrics"] = resource_def.metrics()
-        return attributes
 
     def _post_measure(self, resource_type, resource_id, metric_name,
                       measure_attributes):
@@ -432,33 +422,32 @@ class GnocchiDispatcher(dispatcher.Base):
             LOG.debug("Measure posted on metric %s of resource %s",
                       metric_name, resource_id)
 
-    def _create_resource(self, resource_type, resource_id,
+    def _create_resource(self, resource_type, resource):
-                         resource_attributes):
         r = self.gnocchi_api.post("%s/v1/resource/%s"
                                   % (self.gnocchi_url, resource_type),
                                   headers=self._get_headers(),
-                                  data=json.dumps(resource_attributes))
+                                  data=json.dumps(resource))
         if r.status_code == 409:
-            LOG.debug("Resource %s already exists", resource_id)
+            LOG.debug("Resource %s already exists", resource['id'])
             raise ResourceAlreadyExists
 
         elif int(r.status_code / 100) != 2:
             raise UnexpectedWorkflowError(
                 _("Resource %(resource_id)s creation failed with "
                   "status: %(status_code)d: %(msg)s") %
-                {'resource_id': resource_id,
+                {'resource_id': resource['id'],
                  'status_code': r.status_code,
                  'msg': r.text})
         else:
-            LOG.debug("Resource %s created", resource_id)
+            LOG.debug("Resource %s created", resource['id'])
 
     def _update_resource(self, resource_type, resource_id,
-                         resource_attributes):
+                         resource_extra):
         r = self.gnocchi_api.patch(
             "%s/v1/resource/%s/%s"
             % (self.gnocchi_url, resource_type, resource_id),
             headers=self._get_headers(),
-            data=json.dumps(resource_attributes))
+            data=json.dumps(resource_extra))
 
         if int(r.status_code / 100) != 2:
             raise UnexpectedWorkflowError(

ceilometer/tests/dispatcher/test_gnocchi.py

@@ -139,15 +139,23 @@ class DispatcherTest(base.BaseTestCase):
                               gnocchi.GnocchiDispatcher, self.conf.conf)
 
     @mock.patch('ceilometer.dispatcher.gnocchi.GnocchiDispatcher'
-                '._process_samples')
+                '._process_resource')
-    def _do_test_activity_filter(self, expected_samples, fake_process_samples):
+    def _do_test_activity_filter(self, expected_samples,
+                                 fake_process_resource):
+
+        def assert_samples(resource_id, metric_grouped_samples):
+            samples = []
+            for metric_name, s in metric_grouped_samples:
+                samples.extend(list(s))
+            self.assertEqual(expected_samples, samples)
+
+        fake_process_resource.side_effect = assert_samples
+
         d = gnocchi.GnocchiDispatcher(self.conf.conf)
         d.record_metering_data(self.samples)
 
-        fake_process_samples.assert_called_with(
+        fake_process_resource.assert_called_with(self.resource_id,
-            mock.ANY, self.resource_id, 'disk.root.size',
+                                                 mock.ANY)
-            expected_samples, True,
-        )
 
     def test_archive_policy_map_config(self):
         archive_policy_map = yaml.dump({
@@ -284,7 +292,7 @@ class DispatcherWorkflowTest(base.BaseTestCase,
         ('normal_workflow', dict(measure=204, post_resource=None, metric=None,
                                  measure_retry=None, patch_resource=204)),
         ('new_resource', dict(measure=404, post_resource=204, metric=None,
-                              measure_retry=204, patch_resource=None)),
+                              measure_retry=204, patch_resource=204)),
         ('new_resource_fail', dict(measure=404, post_resource=500, metric=None,
                                    measure_retry=None, patch_resource=None)),
         ('resource_update_fail', dict(measure=204, post_resource=None,