fix: Stackdriver batch limit for TimeSeries (#88)

* Adding batch limit, unit test, and fixes to unit tests

* Improving previous unit test

* Fix lint

* Fix lint

* Preventing passed-in array mutation

* Updating comment on helper function

Author: TigerHe7

packages/opentelemetry-cloud-monitoring-exporter/src/monitoring.ts

@@ -25,6 +25,11 @@ import { GoogleAuth, JWT } from 'google-auth-library';
 import { google } from 'googleapis';
 import { transformMetricDescriptor, createTimeSeries } from './transform';
 import { TimeSeries } from './types';
+import { partitionList } from './utils';
+
+// Stackdriver Monitoring v3 only accepts up to 200 TimeSeries per
+// CreateTimeSeries call.
+const MAX_BATCH_EXPORT_SIZE = 200;
 
 const OT_USER_AGENT = {
   product: 'opentelemetry-js',
@@ -111,7 +116,13 @@ export class MetricExporter implements IMetricExporter {
         );
       }
     }
-    this._sendTimeSeries(timeSeries);
+
+    for (const batchedTimeSeries of partitionList(
+      timeSeries,
+      MAX_BATCH_EXPORT_SIZE
+    )) {
+      await this._sendTimeSeries(batchedTimeSeries);
+    }
     cb(ExportResult.SUCCESS);
   }
 

packages/opentelemetry-cloud-monitoring-exporter/src/transform.ts

@@ -99,7 +99,7 @@ function transformValueType(valueType: OTValueType): ValueType {
 }
 
 /**
- * Converts metric's timeseries to a list of TimeSeries, so that metric can be
+ * Converts metric's timeseries to a TimeSeries, so that metric can be
  * uploaded to StackDriver.
  */
 export function createTimeSeries(

packages/opentelemetry-cloud-monitoring-exporter/src/utils.ts

@@ -0,0 +1,25 @@
+// Copyright 2020 Google LLC
+//
+// 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 { TimeSeries } from './types';
+
+/** Returns the minimum number of arrays of max size chunkSize, partitioned from the given array. */
+export function partitionList(list: TimeSeries[], chunkSize: number) {
+  const listCopy = [...list];
+  const results = [];
+  while (listCopy.length) {
+    results.push(listCopy.splice(0, chunkSize));
+  }
+  return results;
+}

packages/opentelemetry-cloud-monitoring-exporter/test/monitoring.test.ts

@@ -53,7 +53,9 @@ describe('MetricExporter', () => {
     let exporter: MetricExporter;
     let logger: ConsoleLogger;
     /* tslint:disable no-any */
-    let metricDescriptors: sinon.SinonSpy<[any, any], any>;
+    let metricDescriptors: sinon.SinonSpy<[any, any, any], any>;
+    /* tslint:disable no-any */
+    let timeSeries: sinon.SinonSpy<[any, any, any], any>;
     let debug: sinon.SinonSpy;
     let info: sinon.SinonSpy;
     let warn: sinon.SinonSpy;
@@ -69,7 +71,11 @@ describe('MetricExporter', () => {
 
       metricDescriptors = sinon.spy(
         /* tslint:disable no-any */
-        (request: any, callback: (err: Error | null) => void): any => {
+        (
+          request: any,
+          params: any,
+          callback: (err: Error | null) => void
+        ): any => {
           callback(null);
         }
       );
@@ -81,6 +87,24 @@ describe('MetricExporter', () => {
         metricDescriptors as any
       );
 
+      timeSeries = sinon.spy(
+        /* tslint:disable no-any */
+        (
+          request: any,
+          params: any,
+          callback: (err: Error | null) => void
+        ): any => {
+          callback(null);
+        }
+      );
+
+      sinon.replace(
+        MetricExporter['_monitoring'].projects.timeSeries,
+        'create',
+        /* tslint:disable no-any */
+        timeSeries as any
+      );
+
       sinon.replace(exporter['_auth'], 'getClient', () => {
         if (getClientShouldFail) {
           throw new Error('fail');
@@ -137,6 +161,49 @@ describe('MetricExporter', () => {
         'custom.googleapis.com/opentelemetry/name'
       );
 
+      assert.equal(metricDescriptors.callCount, 1);
+      assert.equal(timeSeries.callCount, 1);
+
+      assert.deepStrictEqual(result, ExportResult.SUCCESS);
+    });
+
+    it('should enforce batch size limit on metrics', async () => {
+      const meter = new MeterProvider().getMeter('test-meter');
+
+      const labels: Labels = { ['keyb']: 'value2', ['keya']: 'value1' };
+      let nMetrics = 401;
+      while (nMetrics > 0) {
+        nMetrics -= 1;
+        const counter = meter.createCounter(`name${nMetrics.toString()}`, {
+          labelKeys: ['keya', 'keyb'],
+        });
+        counter.bind(labels).add(10);
+      }
+      meter.collect();
+      const records = meter.getBatcher().checkPointSet();
+
+      const result = await new Promise((resolve, reject) => {
+        exporter.export(records, result => {
+          resolve(result);
+        });
+      });
+
+      assert.deepStrictEqual(
+        metricDescriptors.getCall(0).args[0].resource.type,
+        'custom.googleapis.com/opentelemetry/name400'
+      );
+      assert.deepStrictEqual(
+        metricDescriptors.getCall(100).args[0].resource.type,
+        'custom.googleapis.com/opentelemetry/name300'
+      );
+      assert.deepStrictEqual(
+        metricDescriptors.getCall(400).args[0].resource.type,
+        'custom.googleapis.com/opentelemetry/name0'
+      );
+
+      assert.equal(metricDescriptors.callCount, 401);
+      assert.equal(timeSeries.callCount, 3);
+
       assert.deepStrictEqual(result, ExportResult.SUCCESS);
     });
   });