Fix Prometheus Stats Exporter. (#440)

* Rename a dict. Remove desribe.

* Fix the creation of metric samples.

* Fix tests.

* Remove view signature. Fix build failures.

* Minor fix

* Fix system test.

* More fixes.

* Replace dict kwarg default

* Sort imports

* Nitfix for dict iterator

Author: songy23

examples/stats/exporter/prometheus.py

@@ -28,10 +28,10 @@
 from pprint import pprint
 
 MiB = 1 << 20
-FRONTEND_KEY = tag_key_module.TagKey("my.org/keys/frontend")
+FRONTEND_KEY = tag_key_module.TagKey("myorg_keys_frontend")
 VIDEO_SIZE_MEASURE = measure_module.MeasureInt(
-    "my.org/measures/video_size", "size of processed videos", "By")
-VIDEO_SIZE_VIEW_NAME = "my.org/views/video_size"
+    "myorg_measures_video_size", "size of processed videos", "By")
+VIDEO_SIZE_VIEW_NAME = "myorg_views_video_size"
 VIDEO_SIZE_DISTRIBUTION = aggregation_module.DistributionAggregation(
     [0.0, 16.0 * MiB, 256.0 * MiB])
 VIDEO_SIZE_VIEW = view_module.View(
@@ -63,17 +63,18 @@ def main():
     measure_map.measure_int_put(VIDEO_SIZE_MEASURE, 25 * MiB)
     measure_map.record(tag_map)
 
-    # Use the line below to see the data on prometheus
-    # while True:
-    #     pass
-
     # Get aggregated stats and print it to console.
     view_data = view_manager.get_view(VIDEO_SIZE_VIEW_NAME)
     pprint(vars(view_data))
-    for k, v in view_data._tag_value_aggregation_data_map.items():
+    for k, v in view_data.tag_value_aggregation_data_map.items():
         pprint(k)
         pprint(vars(v))
 
+    # Prevent main from exiting to see the data on prometheus
+    # localhost:8000/metrics
+    while True:
+        pass
+
 
 if __name__ == '__main__':
     main()

opencensus/stats/exporters/prometheus_exporter.py

@@ -14,14 +14,15 @@
 
 from prometheus_client import start_http_server
 from prometheus_client.core import CollectorRegistry
-from prometheus_client.core import GaugeMetricFamily
 from prometheus_client.core import CounterMetricFamily
-from prometheus_client.core import UntypedMetricFamily
+from prometheus_client.core import GaugeMetricFamily
 from prometheus_client.core import HistogramMetricFamily
 from prometheus_client.core import REGISTRY
-from opencensus.stats.exporters import base
-from opencensus.stats import aggregation_data as aggregation_data_module
+from prometheus_client.core import UntypedMetricFamily
+
 from opencensus.common.transports import sync
+from opencensus.stats import aggregation_data as aggregation_data_module
+from opencensus.stats.exporters import base
 
 
 class Options(object):
@@ -82,10 +83,12 @@ def address(self):
 class Collector(object):
     """ Collector represents the Prometheus Collector object
     """
-    def __init__(self, options=Options(), view_data={}):
+    def __init__(self, options=Options(), view_name_to_data_map=None):
+        if view_name_to_data_map is None:
+            view_name_to_data_map = {}
         self._options = options
         self._registry = options.registry
-        self._view_data = view_data
+        self._view_name_to_data_map = view_name_to_data_map
         self._registered_views = {}
 
     @property
@@ -101,11 +104,11 @@ def registry(self):
         return self._registry
 
     @property
-    def view_data(self):
+    def view_name_to_data_map(self):
         """ Map with all view data objects
         that will be sent to Prometheus
         """
-        return self._view_data
+        return self._view_name_to_data_map
 
     @property
     def registered_views(self):
@@ -117,51 +120,58 @@ def register_view(self, view):
         """ register_view will create the needed structure
         in order to be able to sent all data to Prometheus
         """
-        count = 0
-
-        signature = view_signature(self.options.namespace, view)
+        v_name = get_view_name(self.options.namespace, view)
 
-        if signature not in self.registered_views:
-            desc = {'name': view_name(self.options.namespace, view),
+        if v_name not in self.registered_views:
+            desc = {'name': v_name,
                     'documentation': view.description,
-                    'labels': tag_keys_to_labels(view.columns)}
-            self.registered_views[signature] = desc
-            count += 1
+                    'labels': list(view.columns)}
+            self.registered_views[v_name] = desc
             self.registry.register(self)
 
     def add_view_data(self, view_data):
         """ Add view data object to be sent to server
         """
         self.register_view(view_data.view)
-        signature = view_signature(self.options.namespace, view_data.view)
-        self.view_data[signature] = view_data
+        v_name = get_view_name(self.options.namespace, view_data.view)
+        self.view_name_to_data_map[v_name] = view_data
 
-    def to_metric(self, desc, view):
+    # TODO: add start and end timestamp
+    def to_metric(self, desc, tag_values, agg_data):
         """ to_metric translate the data that OpenCensus create
         to Prometheus format, using Prometheus Metric object
 
-        :type desc: str
-        :param desc: The view descriptor
+        :type desc: dict
+        :param desc: The map that describes view definition
 
-        :type view: object of :class:
-            `~opencensus.stats.view.View`
-        :param object of opencensus.stats.view.View view:
-            View object to translate
+        :type tag_values: tuple of :class:
+            `~opencensus.tags.tag_value.TagValue`
+        :param object of opencensus.tags.tag_value.TagValue:
+            TagValue object used as label values
+
+        :type agg_data: object of :class:
+            `~opencensus.stats.aggregation_data.AggregationData`
+        :param object of opencensus.stats.aggregation_data.AggregationData:
+            Aggregated data that needs to be converted as Prometheus samples
 
         :rtype: :class:`~prometheus_client.core.CounterMetricFamily` or
                 :class:`~prometheus_client.core.HistogramMetricFamily` or
                 :class:`~prometheus_client.core.UntypedMetricFamily` or
                 :class:`~prometheus_client.core.GaugeMetricFamily`
         :returns: A Prometheus metric object
         """
-        agg_data = view.aggregation.aggregation_data
+        metric_name = desc['name']
+        metric_description = desc['documentation']
+        label_keys = desc['labels']
 
         if isinstance(agg_data, aggregation_data_module.CountAggregationData):
-            labels = desc['labels'] if agg_data.count_data is None else None
-            return CounterMetricFamily(name=desc['name'],
-                                       documentation=desc['documentation'],
-                                       value=float(agg_data.count_data),
-                                       labels=labels)
+            metric = CounterMetricFamily(name=metric_name,
+                                         documentation=metric_description,
+                                         labels=label_keys)
+            metric.add_metric(labels=list(tag_values),
+                              value=agg_data.count_data)
+            return metric
+
         elif isinstance(agg_data,
                         aggregation_data_module.DistributionAggregationData):
 
@@ -171,28 +181,31 @@ def to_metric(self, desc, view):
             for ii, bound in enumerate(agg_data.bounds):
                 cum_count += agg_data.counts_per_bucket[ii]
                 points[str(bound)] = cum_count
-            labels = desc['labels'] if points is None else None
-            return HistogramMetricFamily(name=desc['name'],
-                                         documentation=desc['documentation'],
-                                         buckets=list(points.items()),
-                                         sum_value=agg_data.sum,
-                                         labels=labels)
+            metric = HistogramMetricFamily(name=metric_name,
+                                           documentation=metric_description,
+                                           labels=label_keys)
+            metric.add_metric(labels=list(tag_values),
+                              buckets=list(points.items()),
+                              sum_value=agg_data.sum,)
+            return metric
 
         elif isinstance(agg_data,
                         aggregation_data_module.SumAggregationDataFloat):
-            labels = desc['labels'] if agg_data.sum_data is None else None
-            return UntypedMetricFamily(name=desc['name'],
-                                       documentation=desc['documentation'],
-                                       value=agg_data.sum_data,
-                                       labels=labels)
+            metric = UntypedMetricFamily(name=metric_name,
+                                         documentation=metric_description,
+                                         labels=label_keys)
+            metric.add_metric(labels=list(tag_values),
+                              value=agg_data.sum_data)
+            return metric
 
         elif isinstance(agg_data,
                         aggregation_data_module.LastValueAggregationData):
-            labels = desc['labels'] if agg_data.value is None else None
-            return GaugeMetricFamily(name=desc['name'],
-                                     documentation=desc['documentation'],
-                                     value=agg_data.value,
-                                     labels=labels)
+            metric = GaugeMetricFamily(name=metric_name,
+                                       documentation=metric_description,
+                                       labels=label_keys)
+            metric.add_metric(labels=list(tag_values),
+                              value=agg_data.value)
+            return metric
 
         else:
             raise ValueError("unsupported aggregation type %s"
@@ -201,30 +214,16 @@ def to_metric(self, desc, view):
     def collect(self):  # pragma: NO COVER
         """Collect fetches the statistics from OpenCensus
         and delivers them as Prometheus Metrics.
-        Collect is invoked everytime a prometheus.Gatherer is run
+        Collect is invoked every time a prometheus.Gatherer is run
         for example when the HTTP endpoint is invoked by Prometheus.
         """
-        for v_data in list(self.view_data):
-            signature = view_signature(self.options.namespace,
-                                       self.view_data[v_data].view)
-            desc = self.registered_views[signature]
-            metric = self.to_metric(desc,
-                                    self.view_data[v_data].view)
-            yield metric
-
-    def describe(self):
-        """ describe will be used by Prometheus Client
-        to retrieve all registered views.
-        """
-        registered = {}
-        for sign in self.registered_views:
-            registered[sign] = self.registered_views[sign]
-        for v_data in list(self.view_data):  # pragma: NO COVER
-            if not isinstance(v_data, str):
-                signature = view_signature(self.options.namespace, v_data.view)
-                desc = self.registered_views[signature]
-                metric = self.to_metric(desc,
-                                        self.view_data[v_data].view)
+        for v_name, view_data in self.view_name_to_data_map.items():
+            if v_name not in self.registered_views:
+                continue
+            desc = self.registered_views[v_name]
+            for tag_values in view_data.tag_value_aggregation_data_map:
+                agg_data = view_data.tag_value_aggregation_data_map[tag_values]
+                metric = self.to_metric(desc, tag_values, agg_data)
                 yield metric
 
 
@@ -332,15 +331,6 @@ def new_stats_exporter(option):
     return exporter
 
 
-def tag_keys_to_labels(tag_keys):
-    """ Translate Tag keys to labels
-    """
-    labels = []
-    for key in tag_keys:
-        labels.append(key)
-    return labels
-
-
 def new_collector(options):
     """ new_collector should be used
     to create instance of Collector class in order to
@@ -349,19 +339,10 @@ def new_collector(options):
     return Collector(options=options)
 
 
-def view_name(namespace, view):
+def get_view_name(namespace, view):
     """ create the name for the view
     """
     name = ""
     if namespace != "":
         name = namespace + "_"
     return name + view.name
-
-
-def view_signature(namespace, view):
-    """ create the signature for the view
-    """
-    sign = view_name(namespace, view)
-    for key in view.columns:
-        sign += "-" + key
-    return sign

tests/system/stats/prometheus/prometheus_stats_test.py

@@ -30,16 +30,15 @@
 class TestPrometheusStats(unittest.TestCase):
     def test_prometheus_stats(self):
 
-        MiB = 1 << 20
-        FRONTEND_KEY = tag_key_module.TagKey("my.org/keys/frontend")
-        VIDEO_SIZE_MEASURE = measure_module.MeasureInt(
-            "my.org/measures/video_size", "size of processed videos", "By")
-        VIDEO_SIZE_VIEW_NAME = "my.org/views/video_size"
-        VIDEO_SIZE_DISTRIBUTION = aggregation_module.CountAggregation(
-            256.0 * MiB)
-        VIDEO_SIZE_VIEW = view_module.View(
-            VIDEO_SIZE_VIEW_NAME, "processed video size over time",
-            [FRONTEND_KEY], VIDEO_SIZE_MEASURE, VIDEO_SIZE_DISTRIBUTION)
+        method_key = tag_key_module.TagKey("method")
+        request_count_measure = measure_module.MeasureInt(
+            "request_count", "number of requests", "1")
+        request_count_view_name = "request_count_view"
+        count_agg = aggregation_module.CountAggregation()
+        request_count_view = view_module.View(
+            request_count_view_name,
+            "number of requests broken down by methods",
+            [method_key], request_count_measure, count_agg)
         stats = stats_module.Stats()
         view_manager = stats.view_manager
         stats_recorder = stats.stats_recorder
@@ -48,16 +47,24 @@ def test_prometheus_stats(self):
             prometheus.Options(namespace="opencensus", port=9303))
         view_manager.register_exporter(exporter)
 
-        view_manager.register_view(VIDEO_SIZE_VIEW)
+        view_manager.register_view(request_count_view)
 
         time.sleep(random.randint(1, 10) / 1000.0)
 
-        tag_value = tag_value_module.TagValue(str(random.randint(1, 10000)))
-        tag_map = tag_map_module.TagMap()
-        tag_map.insert(FRONTEND_KEY, tag_value)
-        measure_map = stats_recorder.new_measurement_map()
-        measure_map.measure_int_put(VIDEO_SIZE_MEASURE, 25 * MiB)
-        measure_map.record(tag_map)
+        method_value_1 = tag_value_module.TagValue("some method")
+        tag_map_1 = tag_map_module.TagMap()
+        tag_map_1.insert(method_key, method_value_1)
+        measure_map_1 = stats_recorder.new_measurement_map()
+        measure_map_1.measure_int_put(request_count_measure, 1)
+        measure_map_1.record(tag_map_1)
+
+        method_value_2 = tag_value_module.TagValue("some other method")
+        tag_map_2 = tag_map_module.TagMap()
+        tag_map_2.insert(method_key, method_value_2)
+        measure_map_2 = stats_recorder.new_measurement_map()
+        measure_map_2.measure_int_put(request_count_measure, 1)
+        measure_map_2.record(tag_map_2)
+        measure_map_2.record(tag_map_2)
 
         if sys.version_info > (3, 0):
             import urllib.request
@@ -67,7 +74,11 @@ def test_prometheus_stats(self):
             import urllib2
             contents = urllib2.urlopen("http://localhost:9303/metrics").read()
 
-        self.assertIn(b'# TYPE opencensus_my.org/views/video_size counter',
+        self.assertIn(b'# TYPE opencensus_request_count_view counter',
+                      contents)
+        self.assertIn(b'opencensus_request_count_view'
+                      b'{method="some method"} 1.0',
                       contents)
-        self.assertIn(b'opencensus_my.org/views/video_size 268435456.0',
+        self.assertIn(b'opencensus_request_count_view'
+                      b'{method="some other method"} 2.0',
                       contents)