Fix: StackDriver exporter fails with "One or more Timeseries could not be written" (#374)

Fix multiple bugs in the StackdriverStatsExporter, including changing its
behavior so that it emits multiple timeseries each with a single point instead
of a single timeseries with multiple points.

Author: mayurkale22

examples/stats/exporter/stackdriver.py

@@ -21,22 +21,15 @@
 from opencensus.stats import measure as measure_module
 from opencensus.stats import stats as stats_module
 from opencensus.stats import view as view_module
-from opencensus.stats.exporters import stackdriver_exporter as stackdriver
-from opencensus.tags import tag_key as tag_key_module
+from opencensus.stats.exporters import stackdriver_exporter
 from opencensus.tags import tag_map as tag_map_module
-from opencensus.tags import tag_value as tag_value_module
 
-MiB = 1 << 20
-FRONTEND_KEY = tag_key_module.TagKey("my.org/keys/frontend")
-VIDEO_SIZE_MEASURE = measure_module.MeasureInt(
-    "my.org/measure/video_size_test2", "size of processed videos", "By")
-VIDEO_SIZE_VIEW_NAME = "my.org/views/video_size_test2"
-VIDEO_SIZE_DISTRIBUTION = aggregation_module.DistributionAggregation(
-    [0.0, 16.0 * MiB, 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)
+# Create the measures
+# The latency in milliseconds
+m_latency_ms = measure_module.MeasureFloat(
+    "task_latency", "The task latency in milliseconds", "ms")
 
+# The stats recorder
 stats = stats_module.Stats()
 view_manager = stats.view_manager
 stats_recorder = stats.stats_recorder
@@ -47,22 +40,35 @@
     raise ValueError("Couldn't find Google Cloud credentials, set the "
                      "project ID with 'gcloud set project'")
 
-exporter = stackdriver.new_stats_exporter(
-    stackdriver.Options(project_id=project_id))
-view_manager.register_exporter(exporter)
+latency_view = view_module.View(
+    "task_latency_distribution",
+    "The distribution of the task latencies",
+    [],
+    m_latency_ms,
+    # Latency in buckets: [>=0ms, >=100ms, >=200ms, >=400ms, >=1s, >=2s, >=4s]
+    aggregation_module.DistributionAggregation(
+        [100.0, 200.0, 400.0, 1000.0, 2000.0, 4000.0]))
 
-# Register view.
-view_manager.register_view(VIDEO_SIZE_VIEW)
 
-# Sleep for [0, 10] milliseconds to fake work.
-time.sleep(random.randint(1, 10) / 1000.0)
+def main():
+    # Enable metrics
+    exporter = stackdriver_exporter.new_stats_exporter(
+        stackdriver_exporter.Options(project_id=project_id))
+    view_manager.register_exporter(exporter)
 
-# Process video.
-# Record the processed video size.
-tag_value = tag_value_module.TagValue(str(1200))
-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)
+    view_manager.register_view(latency_view)
+    mmap = stats_recorder.new_measurement_map()
+    tmap = tag_map_module.TagMap()
 
-measure_map.record(tag_map)
+    for i in range(100):
+        ms = random.random() * 5 * 1000
+        print("Latency {0}:{1}".format(i, ms))
+        mmap.measure_float_put(m_latency_ms, ms)
+        mmap.record(tmap)
+        time.sleep(1)
+
+    print("Done recording metrics")
+
+
+if __name__ == "__main__":
+    main()

opencensus/trace/utils.py opencensus/common/utils.pyopencensus/trace/utils.py renamed to opencensus/common/utils.py

@@ -1,4 +1,4 @@
-# Copyright 2017, OpenCensus Authors
+# Copyright 2018, OpenCensus Authors
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -72,3 +72,17 @@ def timestamp_to_microseconds(timestamp):
     epoch_time_secs = calendar.timegm(timestamp_str.timetuple())
     epoch_time_mus = epoch_time_secs * 1e6 + timestamp_str.microsecond
     return epoch_time_mus
+
+
+def iuniq(ible):
+    """Get an iterator over unique items of `ible`."""
+    items = set()
+    for item in ible:
+        if item not in items:
+            items.add(item)
+            yield item
+
+
+def uniq(ible):
+    """Get a list of unique items of `ible`."""
+    return list(iuniq(ible))

opencensus/stats/exporters/stackdriver_exporter.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
 import os
 import platform
 import re
@@ -21,6 +22,7 @@
 from google.cloud import monitoring_v3
 
 from opencensus.__version__ import __version__
+from opencensus.common import utils
 from opencensus.common.monitored_resource_util.monitored_resource_util \
     import MonitoredResourceUtil
 from opencensus.common.transports import async_
@@ -29,6 +31,7 @@
 from opencensus.stats.exporters import base
 
 MAX_TIME_SERIES_PER_UPLOAD = 200
+OPENCENSUS_TASK = "opencensus_task"
 OPENCENSUS_TASK_DESCRIPTION = "Opencensus task identifier"
 DEFAULT_DISPLAY_NAME_PREFIX = "OpenCensus"
 ERROR_BLANK_PROJECT_ID = "expecting a non-blank ProjectID"
@@ -107,8 +110,8 @@ def default_monitoring_labels(self):
 
 
 class StackdriverStatsExporter(base.StatsExporter):
-    """ StackdriverStatsExporter exports stats
-    to the Stackdriver Monitoring."""
+    """Stats exporter for the Stackdriver Monitoring backend."""
+
     def __init__(self,
                  options=Options(),
                  client=None,
@@ -164,7 +167,8 @@ def upload_stats(self, view_data):
         """ It receives an array of view_data object
             and create time series for each value
         """
-        requests = self.make_request(view_data, MAX_TIME_SERIES_PER_UPLOAD)
+        view_data_set = utils.uniq(view_data)
+        requests = self.make_request(view_data_set, MAX_TIME_SERIES_PER_UPLOAD)
         for request in requests:
             self.client.create_time_series(request[CONS_NAME],
                                            request[CONS_TIME_SERIES])
@@ -181,7 +185,7 @@ def make_request(self, view_data, limit):
         for v_data in view_data:
             series = self.create_time_series_list(v_data, resource,
                                                   metric_prefix)
-            time_series.append(series)
+            time_series.extend(series)
 
             project_id = self.options.project_id
             request = {}
@@ -197,16 +201,19 @@ def create_time_series_list(self, v_data, option_resource_type,
                                 metric_prefix):
         """ Create the TimeSeries object based on the view data
         """
-        series = monitoring_v3.types.TimeSeries()
-        series.metric.type = namespaced_view_name(v_data.view.name,
-                                                  metric_prefix)
-        set_monitored_resource(series, option_resource_type)
-
+        time_series_list = []
         tag_agg = v_data.tag_value_aggregation_data_map
         for tag_value, agg in tag_agg.items():
+            series = monitoring_v3.types.TimeSeries()
+            series.metric.type = namespaced_view_name(v_data.view.name,
+                                                      metric_prefix)
+            set_metric_labels(series, v_data.view, tag_value)
+            set_monitored_resource(series, option_resource_type)
+
             point = series.points.add()
-            if type(agg) is \
-                    aggregation.aggregation_data.DistributionAggregationData:
+            if isinstance(
+                    agg,
+                    aggregation.aggregation_data.DistributionAggregationData):
                 agg_data = tag_agg.get(tag_value)
                 dist_value = point.value.distribution_value
                 dist_value.count = agg_data.count_data
@@ -228,12 +235,21 @@ def create_time_series_list(self, v_data, option_resource_type,
                 buckets.extend([0])
                 bounds.extend(list(map(float, agg_data.bounds)))
                 buckets.extend(list(map(int, agg_data.counts_per_bucket)))
+            elif isinstance(agg,
+                            aggregation.aggregation_data.CountAggregationData):
+                point.value.int64_value = agg.count_data
+            elif isinstance(
+                    agg, aggregation.aggregation_data.SumAggregationDataFloat):
+                point.value.double_value = agg.sum_data
+            elif not isinstance(
+                    agg, aggregation.aggregation_data
+                    .LastValueAggregationData):  # pragma: NO COVER
+                if isinstance(v_data.view.measure, measure.MeasureInt):
+                    point.value.int64_value = int(agg.value)
+                elif isinstance(v_data.view.measure, measure.MeasureFloat):
+                    point.value.double_value = float(agg.value)
             else:
-                convFloat, isFloat = as_float(tag_value[0])
-                if isFloat:  # pragma: NO COVER
-                    point.value.double_value = convFloat
-                else:  # pragma: NO COVER
-                    point.value.string_value = str(tag_value[0])
+                point.value.string_value = str(tag_value[0])
 
             start = datetime.strptime(v_data.start_time, EPOCH_PATTERN)
             end = datetime.strptime(v_data.end_time, EPOCH_PATTERN)
@@ -244,7 +260,7 @@ def create_time_series_list(self, v_data, option_resource_type,
             point.interval.end_time.seconds = int(timestamp_end)
 
             secs = point.interval.end_time.seconds
-            point.interval.end_time.nanos = int((timestamp_end-secs)*10**9)
+            point.interval.end_time.nanos = int((timestamp_end - secs) * 10**9)
 
             if type(agg) is not aggregation.aggregation_data.\
                     LastValueAggregationData:  # pragma: NO COVER
@@ -256,7 +272,10 @@ def create_time_series_list(self, v_data, option_resource_type,
             start_time.seconds = int(timestamp_start)
             start_secs = start_time.seconds
             start_time.nanos = int((timestamp_start - start_secs) * 1e9)
-        return series
+
+            time_series_list.append(series)
+
+        return time_series_list
 
     def create_metric_descriptor(self, view):
         """ it creates a MetricDescriptor
@@ -300,8 +319,8 @@ def create_metric_descriptor(self, view):
             if isinstance(view_measure, measure.MeasureFloat):
                 value_type = metric_desc.ValueType.DOUBLE
         else:
-            raise Exception("unsupported aggregation type: %s"
-                            % type(view_aggregation))
+            raise Exception(
+                "unsupported aggregation type: %s" % type(view_aggregation))
 
         display_name_prefix = DEFAULT_DISPLAY_NAME_PREFIX
         if self.options.metric_prefix != "":
@@ -406,11 +425,6 @@ def new_stats_exporter(options):
 
     if options.default_monitoring_labels is not None:
         exporter.set_default_labels(options.default_monitoring_labels)
-    else:
-        label = {}
-        key = remove_non_alphanumeric(get_task_value())
-        label[key] = OPENCENSUS_TASK_DESCRIPTION
-        exporter.set_default_labels(label)
     return exporter
 
 
@@ -446,9 +460,24 @@ def new_label_descriptors(defaults, keys):
         label = {}
         label["key"] = remove_non_alphanumeric(tag_key)
         label_descriptors.append(label)
+    label_descriptors.append({"key": OPENCENSUS_TASK,
+                              "description": OPENCENSUS_TASK_DESCRIPTION})
     return label_descriptors
 
 
+def set_metric_labels(series, view, tag_values):
+    if len(view.columns) != len(tag_values):
+        logging.warning(
+            "TagKeys and TagValues don't have same size."
+        )  # pragma: NO COVER
+
+    for ii, tag_value in enumerate(tag_values):
+        if tag_value is not None:
+            metric_label = remove_non_alphanumeric(view.columns[ii])
+            series.metric.labels[metric_label] = tag_value
+    series.metric.labels[OPENCENSUS_TASK] = get_task_value()
+
+
 def remove_non_alphanumeric(text):
     """ Remove characters not accepted in labels key
     """

opencensus/trace/attributes.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 import six
 
-from opencensus.trace import utils
+from opencensus.common import utils
 
 
 def _format_attribute_value(value):

opencensus/trace/exporters/jaeger_exporter.py

@@ -20,11 +20,11 @@
 from thrift.protocol import TBinaryProtocol, TCompactProtocol
 from thrift.transport import THttpClient, TTransport
 
+from opencensus.common.utils import timestamp_to_microseconds
 from opencensus.trace import link as link_module
 from opencensus.trace.exporters import base
 from opencensus.trace.exporters.gen.jaeger import agent, jaeger
 from opencensus.trace.exporters.transports import sync
-from opencensus.trace.utils import timestamp_to_microseconds
 
 DEFAULT_HOST_NAME = 'localhost'
 DEFAULT_AGENT_PORT = 6831

opencensus/trace/exporters/zipkin_exporter.py

@@ -19,9 +19,10 @@
 
 import requests
 
+from opencensus.common.utils import check_str_length
+from opencensus.common.utils import timestamp_to_microseconds
 from opencensus.trace.exporters import base
 from opencensus.trace.exporters.transports import sync
-from opencensus.trace.utils import check_str_length, timestamp_to_microseconds
 
 DEFAULT_ENDPOINT = '/api/v2/spans'
 DEFAULT_HOST_NAME = 'localhost'

opencensus/trace/span.py

@@ -15,6 +15,7 @@
 from datetime import datetime
 from itertools import chain
 
+from opencensus.common.utils import _get_truncatable_str
 from opencensus.trace import attributes
 from opencensus.trace import base_span
 from opencensus.trace import link as link_module
@@ -23,7 +24,6 @@
 from opencensus.trace import time_event as time_event_module
 from opencensus.trace.span_context import generate_span_id
 from opencensus.trace.tracers import base
-from opencensus.trace.utils import _get_truncatable_str
 
 
 class SpanKind(object):

opencensus/trace/span_data.py

@@ -13,7 +13,8 @@
 # limitations under the License.
 
 import collections
-from opencensus.trace import utils
+
+from opencensus.common import utils
 from opencensus.trace import attributes
 
 _SpanData = collections.namedtuple(

opencensus/trace/stack_trace.py

@@ -17,7 +17,7 @@
 import random
 import traceback
 
-from opencensus.trace.utils import _get_truncatable_str
+from opencensus.common.utils import _get_truncatable_str
 
 MAX_FRAMES = 128
 

opencensus/trace/time_event.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from opencensus.trace.utils import _get_truncatable_str
+from opencensus.common.utils import _get_truncatable_str
 
 
 class Type(object):