Metrics view data conversion (census-instrumentation#469)

c24t · web-flow · commit ad778ff092d7 · 2019-01-28T16:18:02.000-08:00
Add a utility function to convert ViewDatas to Metrics.
diff --git a/opencensus/stats/measure_to_view_map.py b/opencensus/stats/measure_to_view_map.py
@@ -55,12 +55,21 @@ def get_view(self, view_name, timestamp):
 
         view_data_list = self._measure_to_view_data_list_map.get(
             view.measure.name)
-        if view_data_list is not None:
-            for view_data in view_data_list:
-                if view_data.view.name == view_name:
-                    view_data_copy = copy.deepcopy(view_data)
-                    view_data_copy.end()
-                    return view_data_copy
+
+        if not view_data_list:
+            return None
+
+        for view_data in view_data_list:
+            if view_data.view.name == view_name:
+                break
+        else:
+            return None
+
+        view_data_copy = copy.copy(view_data)
+        tvdam_copy = copy.deepcopy(view_data.tag_value_aggregation_data_map)
+        view_data_copy._tag_value_aggregation_data_map = tvdam_copy
+        view_data_copy.end()
+        return view_data_copy
 
     def filter_exported_views(self, all_views):
         """returns the subset of the given view that should be exported"""
diff --git a/opencensus/stats/metric_utils.py b/opencensus/stats/metric_utils.py
@@ -15,8 +15,10 @@
 Utilities to convert stats data models to metrics data models.
 """
 
-from opencensus.metrics import label_key
+from opencensus.metrics import label_value
+from opencensus.metrics.export import metric
 from opencensus.metrics.export import metric_descriptor
+from opencensus.metrics.export import time_series
 from opencensus.stats import aggregation as aggregation_module
 from opencensus.stats import measure as measure_module
 
@@ -70,14 +72,58 @@ def get_metric_type(measure, aggregation):
         raise AssertionError  # pragma: NO COVER
 
 
-def view_to_metric_descriptor(view):
-    """Get a MetricDescriptor for given view data.
+def is_gauge(md_type):
+    """Whether a given MetricDescriptorType value is a gauge.
 
-    :type view: (:class: '~opencensus.stats.view.View')
-    :param view: the view data to for which to build a metric descriptor
+    :type md_type: int
+    :param md_type: A MetricDescriptorType enum value.
     """
-    return metric_descriptor.MetricDescriptor(
-        view.name, view.description, view.measure.unit,
-        get_metric_type(view.measure, view.aggregation),
-        # TODO: add label key description
-        [label_key.LabelKey(tk, "") for tk in view.columns])
+    if md_type not in metric_descriptor.MetricDescriptorType:
+        raise ValueError  # pragma: NO COVER
+
+    return md_type in {
+        metric_descriptor.MetricDescriptorType.GAUGE_INT64,
+        metric_descriptor.MetricDescriptorType.GAUGE_DOUBLE,
+        metric_descriptor.MetricDescriptorType.GAUGE_DISTRIBUTION
+    }
+
+
+def get_label_values(tag_values):
+    """Convert an iterable of TagValues into a list of LabelValues.
+
+    :type tag_values: list(:class: `opencensus.tags.tag_value.TagValue`)
+    :param tag_values: An iterable of TagValues to convert.
+
+    :rtype: list(:class: `opencensus.metrics.label_value.LabelValue`)
+    :return: A list of LabelValues, converted from TagValues.
+    """
+    return [label_value.LabelValue(tv) for tv in tag_values]
+
+
+def view_data_to_metric(view_data, timestamp):
+    """Convert a ViewData to a Metric at time `timestamp`.
+
+    :type view_data: :class: `opencensus.stats.view_data.ViewData`
+    :param view_data: The ViewData to convert.
+
+    :type timestamp: :class: `datetime.datetime`
+    :param timestamp: The time to set on the metric's point's aggregation,
+    usually the current time.
+
+    :rtype: :class: `opencensus.metrics.export.metric.Metric`
+    :return: A converted Metric.
+    """
+    md = view_data.view.get_metric_descriptor()
+
+    # TODO: implement gauges
+    if is_gauge(md.type):
+        ts_start = None  # pragma: NO COVER
+    else:
+        ts_start = view_data.start_time
+
+    ts_list = []
+    for tag_vals, agg_data in view_data.tag_value_aggregation_data_map.items():
+        label_values = get_label_values(tag_vals)
+        point = agg_data.to_point(timestamp)
+        ts_list.append(time_series.TimeSeries(label_values, [point], ts_start))
+    return metric.Metric(md, ts_list)
diff --git a/opencensus/stats/view.py b/opencensus/stats/view.py
@@ -13,6 +13,13 @@
 # limitations under the License.
 
 
+import threading
+
+from opencensus.metrics import label_key
+from opencensus.metrics.export import metric_descriptor
+from opencensus.stats import metric_utils
+
+
 class View(object):
     """A view defines a specific aggregation and a set of tag keys
 
@@ -33,13 +40,19 @@ class View(object):
     :param aggregation: the aggregation the view will support
 
     """
+
     def __init__(self, name, description, columns, measure, aggregation):
         self._name = name
         self._description = description
         self._columns = columns
         self._measure = measure
         self._aggregation = aggregation
 
+        # Cache the converted MetricDescriptor here to avoid creating it each
+        # time we convert a ViewData that realizes this View into a Metric.
+        self._md_cache_lock = threading.Lock()
+        self._metric_descriptor = None
+
     @property
     def name(self):
         """the name of the current view"""
@@ -64,3 +77,24 @@ def measure(self):
     def aggregation(self):
         """the aggregation of the current view"""
         return self._aggregation
+
+    def get_metric_descriptor(self):
+        """Get a MetricDescriptor for this view.
+
+        Lazily creates a MetricDescriptor for metrics conversion.
+
+        :rtype: :class:
+                `opencensus.metrics.export.metric_descriptor.MetricDescriptor`
+        :return: A converted Metric.
+        """  # noqa
+        with self._md_cache_lock:
+            if self._metric_descriptor is None:
+                self._metric_descriptor = metric_descriptor.MetricDescriptor(
+                    self.name,
+                    self.description,
+                    self.measure.unit,
+                    metric_utils.get_metric_type(self.measure,
+                                                 self.aggregation),
+                    # TODO: add label key description
+                    [label_key.LabelKey(tk, "") for tk in self.columns])
+        return self._metric_descriptor
diff --git a/tests/unit/stats/test_metric_utils.py b/tests/unit/stats/test_metric_utils.py
@@ -17,13 +17,20 @@
 except ImportError:
     from unittest import mock
 
+import datetime
 import unittest
 
 from opencensus.metrics.export import metric_descriptor
+from opencensus.metrics.export import point
+from opencensus.metrics.export import value
 from opencensus.stats import aggregation
+from opencensus.stats import aggregation_data
 from opencensus.stats import measure
 from opencensus.stats import metric_utils
 from opencensus.stats import view
+from opencensus.stats import view_data
+from opencensus.tags import tag_key
+from opencensus.tags import tag_value
 
 
 class TestMetricUtils(unittest.TestCase):
@@ -83,20 +90,84 @@ def test_get_metric_type_bad_measure(self):
         with self.assertRaises(ValueError):
             metric_utils.get_metric_type(base_measure, agg_lv)
 
-    def test_view_to_metric_descriptor(self):
+    def do_test_view_data_to_metric(self, aggregation_type, aggregation_class,
+                                    value_type, metric_descriptor_type):
+        """Test that ViewDatas are converted correctly into Metrics.
+
+        This test doesn't check that the various aggregation data `to_point`
+        methods handle the point conversion correctly, just that converted
+        Point is included in the Metric, and the metric has the expected
+        structure, descriptor, and labels.
+        """
+        start_time = datetime.datetime(2019, 1, 25, 11, 12, 13)
+        current_time = datetime.datetime(2019, 1, 25, 12, 13, 14)
+
         mock_measure = mock.Mock(spec=measure.MeasureFloat)
-        mock_agg = mock.Mock(spec=aggregation.SumAggregation)
-        mock_agg.aggregation_type = aggregation.Type.SUM
-        test_view = view.View("name", "description", ["tk1", "tk2"],
-                              mock_measure, mock_agg)
-
-        md = metric_utils.view_to_metric_descriptor(test_view)
-        self.assertTrue(isinstance(md, metric_descriptor.MetricDescriptor))
-        self.assertEqual(md.name, test_view.name)
-        self.assertEqual(md.description, test_view.description)
-        self.assertEqual(md.unit, test_view.measure.unit)
-        self.assertEqual(
-            md.type, metric_descriptor.MetricDescriptorType.CUMULATIVE_DOUBLE)
-        self.assertTrue(
-            all(lk.key == col
-                for lk, col in zip(md.label_keys, test_view.columns)))
+        mock_aggregation = mock.Mock(spec=aggregation_class)
+        mock_aggregation.aggregation_type = aggregation_type
+
+        vv = view.View(
+            name=mock.Mock(),
+            description=mock.Mock(),
+            columns=[tag_key.TagKey('k1'), tag_key.TagKey('k2')],
+            measure=mock_measure,
+            aggregation=mock_aggregation)
+
+        vd = mock.Mock(spec=view_data.ViewData)
+        vd.view = vv
+        vd.start_time = start_time
+
+        mock_point = mock.Mock(spec=point.Point)
+        mock_point.value = mock.Mock(spec=value_type)
+
+        mock_agg = mock.Mock(spec=aggregation_data.SumAggregationDataFloat)
+        mock_agg.to_point.return_value = mock_point
+
+        vd.tag_value_aggregation_data_map = {
+            (tag_value.TagValue('v1'), tag_value.TagValue('v2')): mock_agg
+        }
+
+        metric = metric_utils.view_data_to_metric(vd, current_time)
+        mock_agg.to_point.assert_called_once_with(current_time)
+
+        self.assertEqual(metric.descriptor.name, vv.name)
+        self.assertEqual(metric.descriptor.description, vv.description)
+        self.assertEqual(metric.descriptor.unit, vv.measure.unit)
+        self.assertEqual(metric.descriptor.type, metric_descriptor_type)
+        self.assertListEqual(
+            [lk.key for lk in metric.descriptor.label_keys],
+            ['k1', 'k2'])
+
+        self.assertEqual(len(metric.time_series), 1)
+        [ts] = metric.time_series
+        self.assertEqual(ts.start_timestamp, start_time)
+        self.assertListEqual(
+            [lv.value for lv in ts.label_values],
+            ['v1', 'v2'])
+        self.assertEqual(len(ts.points), 1)
+        [pt] = ts.points
+        self.assertEqual(pt, mock_point)
+
+    def test_view_data_to_metric(self):
+        args_list = [
+            [
+                aggregation.Type.SUM,
+                aggregation.SumAggregation,
+                value.ValueDouble,
+                metric_descriptor.MetricDescriptorType.CUMULATIVE_DOUBLE
+            ],
+            [
+                aggregation.Type.COUNT,
+                aggregation.CountAggregation,
+                value.ValueLong,
+                metric_descriptor.MetricDescriptorType.CUMULATIVE_INT64
+            ],
+            [
+                aggregation.Type.DISTRIBUTION,
+                aggregation.DistributionAggregation,
+                value.ValueDistribution,
+                metric_descriptor.MetricDescriptorType.CUMULATIVE_DISTRIBUTION
+            ]
+        ]
+        for args in args_list:
+            self.do_test_view_data_to_metric(*args)
diff --git a/tests/unit/stats/test_view.py b/tests/unit/stats/test_view.py
@@ -14,6 +14,11 @@
 
 import unittest
 import mock
+
+from opencensus.metrics.export import metric_descriptor
+from opencensus.stats import aggregation
+from opencensus.stats import measure
+from opencensus.stats import view
 from opencensus.stats import view as view_module
 
 
@@ -37,3 +42,29 @@ def test_constructor(self):
         self.assertEqual(["testTagKey1", "testTagKey2"], view.columns)
         self.assertEqual(measure, view.measure)
         self.assertEqual(aggregation, view.aggregation)
+
+    def test_view_to_metric_descriptor(self):
+        mock_measure = mock.Mock(spec=measure.MeasureFloat)
+        mock_agg = mock.Mock(spec=aggregation.SumAggregation)
+        mock_agg.aggregation_type = aggregation.Type.SUM
+        test_view = view.View("name", "description", ["tk1", "tk2"],
+                              mock_measure, mock_agg)
+
+        self.assertIsNone(test_view._metric_descriptor)
+        md = test_view.get_metric_descriptor()
+        self.assertTrue(isinstance(md, metric_descriptor.MetricDescriptor))
+        self.assertEqual(md.name, test_view.name)
+        self.assertEqual(md.description, test_view.description)
+        self.assertEqual(md.unit, test_view.measure.unit)
+        self.assertEqual(
+            md.type, metric_descriptor.MetricDescriptorType.CUMULATIVE_DOUBLE)
+        self.assertTrue(
+            all(lk.key == col
+                for lk, col in zip(md.label_keys, test_view.columns)))
+
+        md_path = ('opencensus.metrics.export.metric_descriptor'
+                   '.MetricDescriptor')
+        with mock.patch(md_path) as mock_md_cls:
+            md2 = test_view.get_metric_descriptor()
+            mock_md_cls.assert_not_called()
+        self.assertEqual(md, md2)
