feat: add stats recorder implementation (#85)

* feat: add stats recorder implementation

* refactor: change distribution data attributes

- Change bucket.max to bucket.highBoundary
- Change bucket.min to bucket.lowBoundary
- Remove bucketBoundaries

* refactor: removes unecessary depencency

* refactor(fix): changes to address review comments

* fix: recorder truncates int64 values

* refactor(fix): changes to address review comments

packages/opencensus-core/package.json

@@ -54,7 +54,6 @@
   },
   "dependencies": {
     "continuation-local-storage": "^3.2.1",
-    "hdr-histogram-js": "^1.1.4",
     "log-driver": "^1.2.7",
     "semver": "^5.5.0",
     "shimmer": "^1.2.0",

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

@@ -14,12 +14,80 @@
  * limitations under the License.
  */
 
-import {AggregationData, Measurement} from './types';
+import {AggregationData, AggregationType, CountData, DistributionData, LastValueData, Measurement, MeasureType, SumData} from './types';
 
 export class Recorder {
   static addMeasurement(
       aggregationData: AggregationData,
       measurement: Measurement): AggregationData {
-    throw new Error('Not Implemented');
+    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);
+
+      case AggregationType.SUM:
+        return this.addToSum(aggregationData, value);
+
+      case AggregationType.COUNT:
+        return this.addToCount(aggregationData, value);
+
+      default:
+        return this.addToLastValue(aggregationData, value);
+    }
+  }
+
+  private static addToDistribution(
+      distributionData: DistributionData, value: number): DistributionData {
+    distributionData.count += 1;
+
+    const inletBucket = distributionData.buckets.find((bucket) => {
+      return bucket.lowBoundary <= value && value < bucket.highBoundary;
+    });
+    inletBucket.count += 1;
+
+    if (value > distributionData.max) {
+      distributionData.max = value;
+    }
+
+    if (value < distributionData.min) {
+      distributionData.min = value;
+    }
+
+    if (distributionData.count === 1) {
+      distributionData.mean = value;
+    }
+
+    distributionData.sum += value;
+
+    const oldMean = distributionData.mean;
+    distributionData.mean = distributionData.mean +
+        (value - distributionData.mean) / distributionData.count;
+    distributionData.sumSquaredDeviations =
+        distributionData.sumSquaredDeviations +
+        (value - oldMean) * (value - distributionData.mean);
+    distributionData.stdDeviation = Math.sqrt(
+        distributionData.sumSquaredDeviations / distributionData.count);
+
+    return distributionData;
+  }
+
+  private static addToSum(sumData: SumData, value: number): SumData {
+    sumData.value += value;
+    return sumData;
+  }
+
+  private static addToCount(countData: CountData, value: number): CountData {
+    countData.value += 1;
+    return countData;
+  }
+
+  private static addToLastValue(lastValueData: LastValueData, value: number):
+      LastValueData {
+    lastValueData.value = value;
+    return lastValueData;
   }
 }

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

@@ -100,8 +100,6 @@ export interface View {
   registered: boolean;
   /**  Returns a snapshot of an AggregationData for that tags/labels values */
   getSnapshot(tags: Tags): AggregationData;
-  /** Returns a list of all AggregationData in the view */
-  getSnapshots(): AggregationData[];
 }
 
 /**
@@ -116,7 +114,7 @@ export const enum AggregationType {
 }
 
 /** Defines how data is collected and aggregated */
-export interface AggregationData {
+export interface AggregationMetadata {
   /** The aggregation type of the aggregation data */
   readonly type: AggregationType;
   /** The tags/labels that this AggregationData collects and aggregates */
@@ -126,32 +124,37 @@ export interface AggregationData {
 }
 
 /**
- * Data collected and aggregated with this AggregationData will be summed up.
+ * Data collected and aggregated with this AggregationData will be summed
+ * up.
  */
-export interface SumData extends AggregationData {
+export interface SumData extends AggregationMetadata {
+  type: AggregationType.SUM;
   /** The current accumulated value */
   value: number;
 }
 
 /**
  * This AggregationData counts the number of measurements recorded.
  */
-export interface CountData extends AggregationData {
+export interface CountData extends AggregationMetadata {
+  type: AggregationType.COUNT;
   /** The current counted value */
   value: number;
 }
 
 /**
- * This AggregationData represents the last recorded value. This is useful when
- * giving support to Gauges.
+ * This AggregationData represents the last recorded value. This is useful
+ * when giving support to Gauges.
  */
-export interface LastValueData extends AggregationData {
+export interface LastValueData extends AggregationMetadata {
+  type: AggregationType.LAST_VALUE;
   /** The last recorded value */
   value: number;
 }
 
 /** This AggregationData contains a histogram of the collected values. */
-export interface DistributionData extends AggregationData {
+export interface DistributionData extends AggregationMetadata {
+  type: AggregationType.DISTRIBUTION;
   /** The first timestamp a datapoint was added */
   readonly startTime: number;
   /** Get the total count of all recorded values in the histogram */
@@ -176,16 +179,16 @@ export interface DistributionData extends AggregationData {
   sumSquaredDeviations: number;
   /** Bucket distribution of the histogram */
   buckets: Bucket[];
-  /** The bucket boundaries for a histogram */
-  readonly bucketsBoundaries: number[];
 }
 
 /** A simple histogram bucket interface. */
 export interface Bucket {
   /** Number of occurrences in the domain */
   count: number;
-  /** The maximum bucket limit in domain */
-  readonly max: number;
-  /** The minimum bucket limit in domain */
-  readonly min: number;
+  /** The maximum possible value for a data point to fall in this bucket */
+  readonly highBoundary: number;
+  /** The minimum possible value for a data point to fall in this bucket */
+  readonly lowBoundary: number;
 }
+
+export type AggregationData = SumData|CountData|LastValueData|DistributionData;

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

@@ -60,9 +60,4 @@ export class BaseView implements View {
   getSnapshot(tags: Tags): AggregationData {
     throw new Error('Not Implemented');
   }
-
-  /** Returns a list of all AggregationData in the view */
-  getSnapshots(): AggregationData[] {
-    throw new Error('Not Implemented');
-  }
 }

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

@@ -0,0 +1,200 @@
+/**
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as assert from 'assert';
+import * as mocha from 'mocha';
+
+import {Recorder} from '../src';
+import {AggregationType, CountData, DistributionData, LastValueData, Measure, Measurement, MeasureType, MeasureUnit, SumData, Tags} from '../src/stats/types';
+
+/** The order of how close values must be to be considerated almost equal */
+const EPSILON = 6;
+
+interface RecorderTestCase {
+  values: number[];
+  description: string;
+}
+
+function isAlmostEqual(
+    actual: number, expected: number, epsilon: number): boolean {
+  return Math.abs(actual - expected) < Math.pow(10, -epsilon);
+}
+
+function assertDistributionData(
+    distributionData: DistributionData, values: number[]) {
+  const valuesSum = values.reduce((acc, cur) => acc + cur);
+
+  assert.strictEqual(distributionData.max, Math.max(...values));
+  assert.strictEqual(distributionData.min, Math.min(...values));
+  assert.strictEqual(distributionData.count, values.length);
+  assert.strictEqual(distributionData.sum, valuesSum);
+
+  for (const bucket of distributionData.buckets) {
+    const expectedBucketCount = values
+                                    .filter(
+                                        value => bucket.lowBoundary <= value &&
+                                            value < bucket.highBoundary)
+                                    .length;
+    assert.strictEqual(bucket.count, expectedBucketCount);
+  }
+
+  const expectedMean = valuesSum / values.length;
+  assert.ok(isAlmostEqual(distributionData.mean, expectedMean, EPSILON));
+
+  const expectedSumSquaredDeviations =
+      values.map(value => Math.pow(value - expectedMean, 2))
+          .reduce((acc, curr) => acc + curr);
+  assert.ok(isAlmostEqual(
+      distributionData.sumSquaredDeviations, expectedSumSquaredDeviations,
+      EPSILON));
+
+  const expectedStdDeviation =
+      Math.sqrt(expectedSumSquaredDeviations / values.length);
+  assert.ok(isAlmostEqual(
+      distributionData.stdDeviation, expectedStdDeviation, EPSILON));
+}
+
+describe('Recorder', () => {
+  const measures: Measure[] = [
+    {name: 'Test Measure 1', type: MeasureType.DOUBLE, unit: MeasureUnit.UNIT},
+    {name: 'Test Measure 2', type: MeasureType.INT64, unit: MeasureUnit.UNIT}
+  ];
+  const tags: Tags = {testKey: 'testValue'};
+  const testCases: RecorderTestCase[] = [
+    {values: [1.1, 2.5, 3.2, 4.7, 5.2], description: 'with positive values'}, {
+      values: [-1.5, -2.3, -3.7, -4.3, -5.9],
+      description: 'with negative values'
+    },
+    {values: [0, 0, 0, 0], description: 'with zeros'},
+    {values: [1.1, -2.3, 3.2, -4.3, 5.2], description: 'with mixed values'}
+  ];
+
+  for (const measure of measures) {
+    describe(`for count aggregation data of ${measure.type} values`, () => {
+      for (const testCase of testCases) {
+        it(`should record measurements ${testCase.description} correctly`,
+           () => {
+             const countData: CountData = {
+               type: AggregationType.COUNT,
+               tags,
+               timestamp: Date.now(),
+               value: 0
+             };
+             let count = 0;
+             for (const value of testCase.values) {
+               count++;
+               const measurement: Measurement = {measure, tags, value};
+               const updatedAggregationData =
+                   Recorder.addMeasurement(countData, measurement) as CountData;
+
+               assert.strictEqual(updatedAggregationData.value, count);
+             }
+           });
+      }
+    });
+
+    describe(
+        `for last value aggregation data of ${measure.type} values`, () => {
+          for (const testCase of testCases) {
+            it(`should record measurements ${testCase.description} correctly`,
+               () => {
+                 const lastValueData: LastValueData = {
+                   type: AggregationType.LAST_VALUE,
+                   tags,
+                   timestamp: Date.now(),
+                   value: undefined
+                 };
+                 for (const value of testCase.values) {
+                   const measurement: Measurement = {measure, tags, value};
+                   const lastValue = measure.type === MeasureType.DOUBLE ?
+                       value :
+                       Math.trunc(value);
+
+                   const updatedAggregationData =
+                       Recorder.addMeasurement(lastValueData, measurement) as
+                       LastValueData;
+                   assert.strictEqual(updatedAggregationData.value, lastValue);
+                 }
+               });
+          }
+        });
+
+    describe(`for sum aggregation data of ${measure.type} values`, () => {
+      for (const testCase of testCases) {
+        it(`should record measurements ${testCase.description} correctly`,
+           () => {
+             const sumData: SumData = {
+               type: AggregationType.SUM,
+               tags,
+               timestamp: Date.now(),
+               value: 0
+             };
+             let acc = 0;
+             for (const value of testCase.values) {
+               acc += measure.type === MeasureType.DOUBLE ? value :
+                                                            Math.trunc(value);
+               const measurement: Measurement = {measure, tags, value};
+               const updatedAggregationData =
+                   Recorder.addMeasurement(sumData, measurement) as SumData;
+
+               assert.strictEqual(updatedAggregationData.value, acc);
+             }
+           });
+      }
+    });
+
+    describe(
+        `for distribution aggregation data of ${measure.type} values`, () => {
+          for (const testCase of testCases) {
+            it(`should record measurements ${testCase.description} correctly`,
+               () => {
+                 const distributionData: DistributionData = {
+                   type: AggregationType.DISTRIBUTION,
+                   tags,
+                   timestamp: Date.now(),
+                   startTime: Date.now(),
+                   count: 0,
+                   sum: 0,
+                   max: Number.MIN_SAFE_INTEGER,
+                   min: Number.MAX_SAFE_INTEGER,
+                   mean: 0,
+                   stdDeviation: 0,
+                   sumSquaredDeviations: 0,
+                   buckets: [
+                     {highBoundary: 0, lowBoundary: -Infinity, count: 0},
+                     {highBoundary: 2, lowBoundary: 0, count: 0},
+                     {highBoundary: 4, lowBoundary: 2, count: 0},
+                     {highBoundary: 6, lowBoundary: 4, count: 0},
+                     {highBoundary: Infinity, lowBoundary: 6, count: 0}
+                   ]
+                 };
+                 const sentValues = [];
+                 for (const value of testCase.values) {
+                   sentValues.push(
+                       measure.type === MeasureType.DOUBLE ? value :
+                                                             Math.trunc(value));
+                   const measurement: Measurement = {measure, tags, value};
+                   const updatedAggregationData =
+                       Recorder.addMeasurement(distributionData, measurement) as
+                       DistributionData;
+
+                   assertDistributionData(distributionData, sentValues);
+                 }
+               });
+          }
+        });
+  }
+});