Merge "Avoid from storing samples with empty or not numerical volumes" into stable/kilo

2 files changed, 99 insertions, 1 deletions

diff --git a/ceilometer/pipeline.py b/ceilometer/pipeline.py
index 9c31bb1c..ea49e1ea 100644
--- a/ceilometer/pipeline.py
+++ b/ceilometer/pipeline.py
@@ -492,10 +492,42 @@ class SamplePipeline(Pipeline):
     def support_meter(self, meter_name):
         return self.source.support_meter(meter_name)
 
+    def _validate_volume(self, s):
+        volume = s.volume
+        if volume is None:
+            LOG.warning(_(
+                'metering data %(counter_name)s for %(resource_id)s '
+                '@ %(timestamp)s has no volume (volume: None), the sample '
+                'will be dropped')
+                % {'counter_name': s.name,
+                   'resource_id': s.resource_id,
+                   'timestamp': s.timestamp if s.timestamp else
+                   'NO TIMESTAMP'}
+            )
+            return False
+        if not isinstance(volume, (int, float)):
+            try:
+                volume = float(volume)
+            except ValueError:
+                LOG.warning(_(
+                    'metering data %(counter_name)s for %(resource_id)s '
+                    '@ %(timestamp)s has volume which is not a number '
+                    '(volume: %(counter_volume)s), the sample will be '
+                    'dropped')
+                    % {'counter_name': s.name,
+                       'resource_id': s.resource_id,
+                       'timestamp': (
+                           s.timestamp if s.timestamp else 'NO TIMESTAMP'),
+                       'counter_volume': volume}
+                )
+                return False
+        return True
+
     def publish_data(self, ctxt, samples):
         if not isinstance(samples, list):
             samples = [samples]
-        supported = [s for s in samples if self.source.support_meter(s.name)]
+        supported = [s for s in samples if self.source.support_meter(s.name)
+                     and self._validate_volume(s)]
         self.sink.publish_samples(ctxt, supported)
 
 
diff --git a/ceilometer/tests/pipeline_base.py b/ceilometer/tests/pipeline_base.py
index fd747017..6686319a 100644
--- a/ceilometer/tests/pipeline_base.py
+++ b/ceilometer/tests/pipeline_base.py
@@ -305,6 +305,72 @@ class BasePipelineTestCase(base.BaseTestCase):
         self.assertEqual('a_update', getattr(publisher.samples[0], "name"))
         self.assertEqual('b_update', getattr(publisher.samples[1], "name"))
 
+    @mock.patch('ceilometer.pipeline.LOG')
+    def test_none_volume_counter(self, log):
+        self._set_pipeline_cfg('counters', ['empty_volume'])
+        pipeline_manager = pipeline.PipelineManager(self.pipeline_cfg,
+                                                    self.transformer_manager)
+        publisher = pipeline_manager.pipelines[0].publishers[0]
+
+        test_s = sample.Sample(
+            name='empty_volume',
+            type=self.test_counter.type,
+            volume=None,
+            unit=self.test_counter.unit,
+            user_id=self.test_counter.user_id,
+            project_id=self.test_counter.project_id,
+            resource_id=self.test_counter.resource_id,
+            timestamp=self.test_counter.timestamp,
+            resource_metadata=self.test_counter.resource_metadata,
+        )
+
+        with pipeline_manager.publisher(None) as p:
+            p([test_s])
+
+        log.warning.assert_called_with(
+            'metering data %(counter_name)s for %(resource_id)s '
+            '@ %(timestamp)s has no volume (volume: %(counter_volume)s), the '
+            'sample will be dropped'
+            % {'counter_name': test_s.name,
+               'resource_id': test_s.resource_id,
+               'timestamp': test_s.timestamp,
+               'counter_volume': test_s.volume})
+
+        self.assertEqual(0, len(publisher.samples))
+
+    @mock.patch('ceilometer.pipeline.LOG')
+    def test_fake_volume_counter(self, log):
+        self._set_pipeline_cfg('counters', ['fake_volume'])
+        pipeline_manager = pipeline.PipelineManager(self.pipeline_cfg,
+                                                    self.transformer_manager)
+        publisher = pipeline_manager.pipelines[0].publishers[0]
+
+        test_s = sample.Sample(
+            name='fake_volume',
+            type=self.test_counter.type,
+            volume='fake_value',
+            unit=self.test_counter.unit,
+            user_id=self.test_counter.user_id,
+            project_id=self.test_counter.project_id,
+            resource_id=self.test_counter.resource_id,
+            timestamp=self.test_counter.timestamp,
+            resource_metadata=self.test_counter.resource_metadata,
+        )
+
+        with pipeline_manager.publisher(None) as p:
+            p([test_s])
+
+        log.warning.assert_called_with(
+            'metering data %(counter_name)s for %(resource_id)s '
+            '@ %(timestamp)s has volume which is not a number '
+            '(volume: %(counter_volume)s), the sample will be dropped'
+            % {'counter_name': test_s.name,
+               'resource_id': test_s.resource_id,
+               'timestamp': test_s.timestamp,
+               'counter_volume': test_s.volume})
+
+        self.assertEqual(0, len(publisher.samples))
+
     def test_counter_dont_match(self):
         counter_cfg = ['nomatch']
         self._set_pipeline_cfg('counters', counter_cfg)