Stats exemplars (#287)

Author: vcasadei

opencensus/stats/aggregation_data.py

@@ -41,7 +41,7 @@ def __init__(self, sum_data):
         super(SumAggregationDataFloat, self).__init__(sum_data)
         self._sum_data = sum_data
 
-    def add_sample(self, value):
+    def add_sample(self, value, timestamp=None, attachments=None):
         """Allows the user to add a sample to the Sum Aggregation Data
         The value of the sample is then added to the current sum data
         """
@@ -64,7 +64,7 @@ def __init__(self, count_data):
         super(CountAggregationData, self).__init__(count_data)
         self._count_data = count_data
 
-    def add_sample(self, value):
+    def add_sample(self, value, timestamp=None, attachments=None):
         """Adds a sample to the current Count Aggregation Data and adds 1 to
         the count data"""
         self._count_data = self._count_data + 1
@@ -97,6 +97,9 @@ class DistributionAggregationData(BaseAggregationData):
     :type counts_per_bucket: list(int)
     :param counts_per_bucket: the number of occurrences per bucket
 
+    :type exemplars: list(Exemplar)
+    :param: exemplars: the exemplars associated with histogram buckets.
+
     :type bounds: list(float)
     :param bounds: the histogram distribution of the values
 
@@ -108,7 +111,8 @@ def __init__(self,
                  max_,
                  sum_of_sqd_deviations,
                  counts_per_bucket=None,
-                 bounds=None):
+                 bounds=None,
+                 exemplars=None):
         super(DistributionAggregationData, self).__init__(mean_data)
         self._mean_data = mean_data
         self._count_data = count_data
@@ -126,6 +130,13 @@ def __init__(self,
         self._counts_per_bucket = counts_per_bucket
         self._bounds = bucket_boundaries.BucketBoundaries(
                                             boundaries=bounds).boundaries
+        bucket = 0
+        for _ in self.bounds:
+            bucket = bucket + 1
+
+        # If there is no histogram, do not record an exemplar
+        self._exemplars = \
+            {bucket: exemplars} if len(self._bounds) > 0 else None
 
     @property
     def mean_data(self):
@@ -157,6 +168,11 @@ def counts_per_bucket(self):
         """The current counts per bucket for the distribution"""
         return self._counts_per_bucket
 
+    @property
+    def exemplars(self):
+        """The current counts per bucket for the distribution"""
+        return self._exemplars
+
     @property
     def bounds(self):
         """The current bounds for the distribution"""
@@ -174,15 +190,17 @@ def variance(self):
             return 0
         return self.sum_of_sqd_deviations / (self._count_data - 1)
 
-    def add_sample(self, value):
+    def add_sample(self, value, timestamp, attachments):
         """Adding a sample to Distribution Aggregation Data"""
         if value < self.min:
             self._min = value
         if value > self.max:
             self._max = value
         self._count_data += 1
-        self.increment_bucket_count(value)
+        bucket = self.increment_bucket_count(value)
 
+        if attachments is not None and self.exemplars is not None:
+            self.exemplars[bucket] = Exemplar(value, timestamp, attachments)
         if self.count_data == 1:
             self._mean_data = value
             return
@@ -196,22 +214,29 @@ def add_sample(self, value):
 
     def increment_bucket_count(self, value):
         """Increment the bucket count based on a given value from the user"""
-        if len(self._bounds) == 0:
-            self._counts_per_bucket[0] += 1
-            return
-
         i = 0
+        incremented = False
         for b in self._bounds:
-            if value < b:
+            if value < b and not incremented:
                 self._counts_per_bucket[i] += 1
-                return
+                incremented = True
             i += 1
 
+        if incremented:
+            return i
+
+        if len(self._bounds) == 0:
+            self._counts_per_bucket[0] += 1
+            return i
+
         self._counts_per_bucket[(len(self._bounds))-1] += 1
 
+        return i
+
 
 class LastValueAggregationData(BaseAggregationData):
-    """LastValue Aggregation Data is the value of aggregated data
+    """
+    LastValue Aggregation Data is the value of aggregated data
 
     :type value: long
     :param value: represents the current value
@@ -221,15 +246,63 @@ def __init__(self, value):
         super(LastValueAggregationData, self).__init__(value)
         self._value = value
 
-    def add_sample(self, value):
+    def add_sample(self, value, timestamp=None, attachments=None):
         """Adds a sample to the current
-           LastValue Aggregation Data and overwrite
-           the current recorded value
-        """
+        LastValue Aggregation Data and overwrite
+        the current recorded value"""
         self._value = value
 
     @property
     def value(self):
-        """The current value recorded
-        """
+        """The current value recorded"""
         return self._value
+
+
+class Exemplar(object):
+    """ Exemplar represents an example point that may be used to annotate
+        aggregated distribution values, associated with a histogram bucket.
+
+        :type value: double
+        :param value: value of the Exemplar point.
+
+        :type timestamp: time
+        :param timestamp: the time that this Exemplar's value was recorded.
+
+        :type attachments: dict
+        :param attachments: the contextual information about the example value.
+    """
+
+    def __init__(self,
+                 value,
+                 timestamp,
+                 attachments):
+        self._value = value
+
+        self._timestamp = timestamp
+
+        if attachments is None:
+            raise TypeError('attachments should not be empty')
+
+        for key, value in attachments.items():
+            if key is None or not isinstance(key, str):
+                raise TypeError('attachment key should not be '
+                                'empty and should be a string')
+            if value is None or not isinstance(value, str):
+                raise TypeError('attachment value should not be '
+                                'empty and should be a string')
+        self._attachments = attachments
+
+    @property
+    def value(self):
+        """The current value of the Exemplar point"""
+        return self._value
+
+    @property
+    def timestamp(self):
+        """The time that this Exemplar's value was recorded"""
+        return self._timestamp
+
+    @property
+    def attachments(self):
+        """The contextual information about the example value"""
+        return self._attachments

opencensus/stats/measure_to_view_map.py

@@ -93,7 +93,7 @@ def register_view(self, view, timestamp):
         self._measure_to_view_data_list_map[view.measure.name].append(
             ViewData(view=view, start_time=timestamp, end_time=timestamp))
 
-    def record(self, tags, measurement_map, timestamp):
+    def record(self, tags, measurement_map, timestamp, attachments=None):
         """records stats with a set of tags"""
         for measure, value in measurement_map.items():
             if measure != self._registered_measures.get(measure.name):
@@ -105,7 +105,8 @@ def record(self, tags, measurement_map, timestamp):
                     view_datas.extend(view_data_list)
             for view_data in view_datas:
                 view_data.record(
-                    context=tags, value=value, timestamp=timestamp)
+                    context=tags, value=value, timestamp=timestamp,
+                    attachment=attachments)
             self.export(view_datas)
 
     def export(self, view_datas):

opencensus/stats/measurement_map.py

@@ -25,10 +25,14 @@ class MeasurementMap(object):
     :param measure_to_view_map: the measure to view map that will store the
                                 recorded stats with tags
 
+    :type: attachments: dict
+    :param attachments: the contextual information about the attachment value.
+
     """
-    def __init__(self, measure_to_view_map):
+    def __init__(self, measure_to_view_map, attachments=None):
         self._measurement_map = {}
         self._measure_to_view_map = measure_to_view_map
+        self._attachments = attachments
 
     @property
     def measurement_map(self):
@@ -40,6 +44,11 @@ def measure_to_view_map(self):
         """the current measure to view map for the measurement map"""
         return self._measure_to_view_map
 
+    @property
+    def attachments(self):
+        """the current contextual information about the attachment value."""
+        return self._attachments
+
     def measure_int_put(self, measure, value):
         """associates the measure of type Int with the given value"""
         self._measurement_map[measure] = value
@@ -48,6 +57,24 @@ def measure_float_put(self, measure, value):
         """associates the measure of type Float with the given value"""
         self._measurement_map[measure] = value
 
+    def measure_put_attachment(self, key, value):
+        """Associate the contextual information of an Exemplar to this MeasureMap
+            Contextual information is represented as key - value string pairs.
+            If this method is called multiple times with the same key,
+            only the last value will be kept.
+        """
+        if self._attachments is None:
+            self._attachments = dict()
+
+        if key is None or not isinstance(key, str):
+            raise TypeError('attachment key should not be '
+                            'empty and should be a string')
+        if value is None or not isinstance(value, str):
+            raise TypeError('attachment value should not be '
+                            'empty and should be a string')
+
+        self._attachments[key] = value
+
     def record(self, tag_map_tags=execution_context.get_current_tag_map()):
         """records all the measures at the same time with a tag_map.
         tag_map could either be explicitly passed to the method, or implicitly
@@ -56,5 +83,6 @@ def record(self, tag_map_tags=execution_context.get_current_tag_map()):
         self.measure_to_view_map.record(
                 tags=tag_map_tags,
                 measurement_map=self.measurement_map,
-                timestamp=datetime.utcnow().isoformat() + 'Z'
+                timestamp=datetime.utcnow().isoformat() + 'Z',
+                attachments=self.attachments
         )

opencensus/stats/view_data.py

@@ -79,12 +79,13 @@ def get_tag_values(self, tags, columns):
             i += 1
         return tag_values
 
-    def record(self, context, value, timestamp):
+    def record(self, context, value, timestamp, attachments=None):
         """records the view data against context"""
         tag_values = self.get_tag_values(tags=context.map,
                                          columns=self.view.columns)
         tuple_vals = tuple(tag_values)
         if tuple_vals not in self.tag_value_aggregation_data_map:
             self.tag_value_aggregation_data_map[tuple_vals] = copy.deepcopy(
                 self.view.aggregation.aggregation_data)
-        self.tag_value_aggregation_data_map.get(tuple_vals).add_sample(value)
+        self.tag_value_aggregation_data_map.get(tuple_vals).\
+            add_sample(value, timestamp, attachments)