Add support for recording exemplars (#405)

* Add exemplar support

* Fix review comments

Fix typos (THe -> The).
Use const and a ternary expression instead of let.
Remove unwanted template backticks.
Add separate function to create/get metric Bucket (getMetricBucket).

Author: mayurkale22

CHANGELOG.md

@@ -20,6 +20,7 @@ All notable changes to this project will be documented in this file.
 - Add an API `globalStats.unregisterExporter()`.
 - Add support for overriding sampling for a span.
 - Enforce `--strictNullChecks` and `--noUnusedLocals` Compiler Options on [opencensus-exporter-jaeger] packages.
+- Add support for recording Exemplars.
 
 ## 0.0.9 - 2019-02-12
 - Add Metrics API.

packages/opencensus-core/src/stats/recorder.ts

@@ -21,16 +21,16 @@ const UNKNOWN_TAG_VALUE: TagValue = null;
 
 export class Recorder {
   static addMeasurement(
-      aggregationData: AggregationData,
-      measurement: Measurement): AggregationData {
+      aggregationData: AggregationData, measurement: Measurement,
+      attachments?: {[key: string]: string}): AggregationData {
     aggregationData.timestamp = Date.now();
     const value = measurement.measure.type === MeasureType.DOUBLE ?
         measurement.value :
         Math.trunc(measurement.value);
 
     switch (aggregationData.type) {
       case AggregationType.DISTRIBUTION:
-        return this.addToDistribution(aggregationData, value);
+        return this.addToDistribution(aggregationData, value, attachments);
 
       case AggregationType.SUM:
         return this.addToSum(aggregationData, value);
@@ -53,7 +53,8 @@ export class Recorder {
   }
 
   private static addToDistribution(
-      distributionData: DistributionData, value: number): DistributionData {
+      distributionData: DistributionData, value: number,
+      attachments?: {[key: string]: string}): DistributionData {
     distributionData.count += 1;
 
     let bucketIndex =
@@ -79,6 +80,15 @@ export class Recorder {
     distributionData.stdDeviation = Math.sqrt(
         distributionData.sumOfSquaredDeviation / distributionData.count);
 
+    // No implicit recording for exemplars - if there are no attachments
+    // (contextual information), don't record exemplars.
+    if (attachments) {
+      distributionData.exemplars[bucketIndex] = {
+        value,
+        timestamp: distributionData.timestamp,
+        attachments
+      };
+    }
     return distributionData;
   }
 

packages/opencensus-core/src/stats/stats.ts

@@ -172,10 +172,15 @@ export class BaseStats implements Stats {
    * Updates all views with the new measurements.
    * @param measurements A list of measurements to record
    * @param tags optional The tags to which the value is applied.
-   *  tags could either be explicitly passed to the method, or implicitly
-   *  read from current execution context.
+   *     tags could either be explicitly passed to the method, or implicitly
+   *     read from current execution context.
+   * @param attachments optional The contextual information associated with an
+   *     example value. THe contextual information is represented as key - value
+   *     string pairs.
    */
-  record(measurements: Measurement[], tags?: TagMap): void {
+  record(
+      measurements: Measurement[], tags?: TagMap,
+      attachments?: {[key: string]: string}): void {
     if (this.hasNegativeValue(measurements)) {
       this.logger.warn(`Dropping measurments ${measurements}, value to record
           must be non-negative.`);
@@ -194,7 +199,7 @@ export class BaseStats implements Stats {
       }
       // Updates all views
       for (const view of views) {
-        view.recordMeasurement(measurement, tags);
+        view.recordMeasurement(measurement, tags, attachments);
       }
 
       // Notifies all exporters

packages/opencensus-core/src/stats/types.ts

@@ -65,10 +65,15 @@ export interface Stats {
    * Updates all views with the new measurements.
    * @param measurements A list of measurements to record
    * @param tags optional The tags to which the value is applied.
-   *  tags could either be explicitly passed to the method, or implicitly
-   *  read from current execution context.
+   *     tags could either be explicitly passed to the method, or implicitly
+   *     read from current execution context.
+   * @param attachments optional The contextual information associated with an
+   *     example value. The contextual information is represented as key - value
+   *     string pairs.
    */
-  record(measurements: Measurement[], tags?: TagMap): void;
+  record(
+      measurements: Measurement[], tags?: TagMap,
+      attachments?: {[key: string]: string}): void;
 
   /**
    * Remove all registered Views and exporters from the stats.
@@ -181,8 +186,13 @@ export interface View {
    * Measurements with measurement type INT64 will have its value truncated.
    * @param measurement The measurement to record
    * @param tags The tags to which the value is applied
+   * @param attachments optional The contextual information associated with an
+   *     example value. THe contextual information is represented as key - value
+   *     string pairs.
    */
-  recordMeasurement(measurement: Measurement, tags: TagMap): void;
+  recordMeasurement(
+      measurement: Measurement, tags: TagMap,
+      attachments?: {[key: string]: string}): void;
   /**
    * Returns a snapshot of an AggregationData for that tags/labels values.
    * @param tagValues The desired data's tag values.
@@ -269,6 +279,25 @@ export interface DistributionData extends AggregationMetadata {
   buckets: Bucket[];
   /** Buckets count */
   bucketCounts?: number[];
+  /** If the distribution does not have a histogram, then omit this field. */
+  exemplars?: StatsExemplar[];
+}
+
+/**
+ * Exemplars are example points that may be used to annotate aggregated
+ * Distribution values. They are metadata that gives information about a
+ * particular value added to a Distribution bucket.
+ */
+export interface StatsExemplar {
+  /**
+   * Value of the exemplar point. It determines which bucket the exemplar
+   * belongs to.
+   */
+  readonly value: number;
+  /** The observation (sampling) time of the above value. */
+  readonly timestamp: number;
+  /** Contextual information about the example value. */
+  readonly attachments: {[key: string]: string};
 }
 
 export type Bucket = number;

packages/opencensus-core/src/stats/view.ts

@@ -17,14 +17,15 @@
 import * as defaultLogger from '../common/console-logger';
 import {getTimestampWithProcessHRTime, timestampFromMillis} from '../common/time-util';
 import * as loggerTypes from '../common/types';
-import {DistributionValue, LabelValue, Metric, MetricDescriptor, MetricDescriptorType, Point, TimeSeries, Timestamp} from '../metrics/export/types';
+import {Bucket as metricBucket, DistributionValue, LabelValue, Metric, MetricDescriptor, MetricDescriptorType, Point, TimeSeries, Timestamp} from '../metrics/export/types';
 import {TagMap} from '../tags/tag-map';
 import {TagKey, TagValue} from '../tags/types';
 import {isValidTagKey} from '../tags/validation';
+
 import {BucketBoundaries} from './bucket-boundaries';
 import {MetricUtils} from './metric-utils';
 import {Recorder} from './recorder';
-import {AggregationData, AggregationType, Measure, Measurement, View} from './types';
+import {AggregationData, AggregationType, Measure, Measurement, StatsExemplar, View} from './types';
 
 const RECORD_SEPARATOR = String.fromCharCode(30);
 
@@ -114,8 +115,13 @@ export class BaseView implements View {
    * Measurements with measurement type INT64 will have its value truncated.
    * @param measurement The measurement to record
    * @param tags The tags to which the value is applied
+   * @param attachments optional The contextual information associated with an
+   *     example value. The contextual information is represented as key - value
+   *     string pairs.
    */
-  recordMeasurement(measurement: Measurement, tags: TagMap) {
+  recordMeasurement(
+      measurement: Measurement, tags: TagMap,
+      attachments?: {[key: string]: string}) {
     const tagValues = Recorder.getTagValues(tags.tags, this.columns);
     const encodedTags = this.encodeTagValues(tagValues);
     if (!this.tagValueAggregationMap[encodedTags]) {
@@ -124,7 +130,7 @@ export class BaseView implements View {
     }
 
     Recorder.addMeasurement(
-        this.tagValueAggregationMap[encodedTags], measurement);
+        this.tagValueAggregationMap[encodedTags], measurement, attachments);
   }
 
   /**
@@ -143,12 +149,14 @@ export class BaseView implements View {
    */
   private createAggregationData(tagValues: TagValue[]): AggregationData {
     const aggregationMetadata = {tagValues, timestamp: Date.now()};
-    const {buckets, bucketCounts} = this.bucketBoundaries;
-    const bucketsCopy = Object.assign([], buckets);
-    const bucketCountsCopy = Object.assign([], bucketCounts);
 
     switch (this.aggregation) {
       case AggregationType.DISTRIBUTION:
+        const {buckets, bucketCounts} = this.bucketBoundaries;
+        const bucketsCopy = Object.assign([], buckets);
+        const bucketCountsCopy = Object.assign([], bucketCounts);
+        const exemplars = new Array(bucketCounts.length);
+
         return {
           ...aggregationMetadata,
           type: AggregationType.DISTRIBUTION,
@@ -159,7 +167,8 @@ export class BaseView implements View {
           stdDeviation: null as number,
           sumOfSquaredDeviation: null as number,
           buckets: bucketsCopy,
-          bucketCounts: bucketCountsCopy
+          bucketCounts: bucketCountsCopy,
+          exemplars
         };
       case AggregationType.SUM:
         return {...aggregationMetadata, type: AggregationType.SUM, value: 0};
@@ -221,18 +230,21 @@ export class BaseView implements View {
    */
   private toPoint(timestamp: Timestamp, data: AggregationData): Point {
     let value;
-
     if (data.type === AggregationType.DISTRIBUTION) {
-      // TODO: Add examplar transition
-      const {count, sum, sumOfSquaredDeviation} = data;
+      const {count, sum, sumOfSquaredDeviation, exemplars} = data;
+      const buckets = [];
+      for (let bucket = 0; bucket < data.bucketCounts.length; bucket++) {
+        const bucketCount = data.bucketCounts[bucket];
+        const statsExemplar = exemplars ? exemplars[bucket] : undefined;
+        buckets.push(this.getMetricBucket(statsExemplar, bucketCount));
+      }
+
       value = {
         count,
         sum,
         sumOfSquaredDeviation,
+        buckets,
         bucketOptions: {explicit: {bounds: data.buckets}},
-        // Bucket without an Exemplar.
-        buckets:
-            data.bucketCounts.map(bucketCount => ({count: bucketCount}))
       } as DistributionValue;
     } else {
       value = data.value as number;
@@ -249,6 +261,24 @@ export class BaseView implements View {
     return this.tagValueAggregationMap[this.encodeTagValues(tagValues)];
   }
 
+  /** Returns a Bucket with count and examplar (if present) */
+  private getMetricBucket(statsExemplar: StatsExemplar, bucketCount: number):
+      metricBucket {
+    if (statsExemplar) {
+      // Bucket with an Exemplar.
+      return {
+        count: bucketCount,
+        exemplar: {
+          value: statsExemplar.value,
+          timestamp: timestampFromMillis(statsExemplar.timestamp),
+          attachments: statsExemplar.attachments
+        }
+      };
+    }
+    // Bucket with no Exemplar.
+    return {count: bucketCount};
+  }
+
   /** Determines whether the given TagKeys are valid. */
   private validateTagKeys(tagKeys: TagKey[]): TagKey[] {
     const tagKeysCopy = Object.assign([], tagKeys);

packages/opencensus-core/test/test-recorder.ts

@@ -176,6 +176,41 @@ describe('Recorder', () => {
           }
         });
 
+    describe('for distribution aggregation data with attachments', () => {
+      const attachments = {'k1': 'v1', 'k2': 'v2', 'k3': 'v3'};
+      it('should record measurements and attachments correctly', () => {
+        const distributionData: DistributionData = {
+          type: AggregationType.DISTRIBUTION,
+          tagValues,
+          timestamp: Date.now(),
+          startTime: Date.now(),
+          count: 0,
+          sum: 0,
+          mean: 0,
+          stdDeviation: 0,
+          sumOfSquaredDeviation: 0,
+          buckets: [2, 4, 6],
+          bucketCounts: [0, 0, 0, 0],
+          exemplars: [undefined, undefined, undefined, undefined]
+        };
+        const value = 5;
+        const measurement: Measurement = {measure, value};
+        const aggregationData =
+            Recorder.addMeasurement(
+                distributionData, measurement, attachments) as DistributionData;
+
+        assert.equal(aggregationData.sum, 5);
+        assert.equal(aggregationData.mean, 5);
+        assert.deepStrictEqual(aggregationData.buckets, [2, 4, 6]);
+        assert.deepStrictEqual(aggregationData.bucketCounts, [0, 0, 1, 0]);
+        assert.deepStrictEqual(aggregationData.exemplars, [
+          undefined, undefined,
+          {value: 5, timestamp: aggregationData.timestamp, attachments},
+          undefined
+        ]);
+      });
+    });
+
     describe('getTagValues()', () => {
       const CALLER = {name: 'caller'};
       const METHOD = {name: 'method'};