Basic node

packages/opentelemetry-exporter-collector/src/platform/node/CollectorMetricExporter.ts

@@ -0,0 +1,137 @@
+/*
+ * Copyright The OpenTelemetry 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
+ *
+ *      https://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 { MetricRecord } from '@opentelemetry/metrics';
+import * as collectorTypes from '../../types';
+import { CollectorExporterConfigNode } from './types';
+import { GRPCMetricQueueItem, ServiceClient } from './types';
+import { removeProtocol } from './util';
+import * as path from 'path';
+import * as protoLoader from '@grpc/proto-loader';
+import * as grpc from 'grpc';
+import { toCollectorExportMetricServiceRequest } from '../../transformMetrics';
+import { CollectorMetricExporterBase } from '../../CollectorMetricExporterBase';
+import { parseHeaders } from '../../util';
+
+const DEFAULT_COLLECTOR_URL = 'localhost:55678';
+
+/**
+ * Collector Metric Exporter for Node
+ */
+export class CollectorMetricExporter extends CollectorMetricExporterBase<
+  CollectorExporterConfigNode
+> {
+  DEFAULT_HEADERS: Record<string, string> = {
+    [collectorTypes.OT_REQUEST_HEADER]: '1',
+  };
+  isShutDown: boolean = false;
+  grpcMetricsQueue: GRPCMetricQueueItem[] = [];
+  metricServiceClient?: ServiceClient = undefined;
+  credentials: grpc.ChannelCredentials;
+  metadata?: grpc.Metadata;
+  headers: Record<string, string>;
+
+  constructor(config: CollectorExporterConfigNode = {}) {
+    super(config);
+    this.grpcMetricsQueue = [];
+    this.credentials = config.credentials || grpc.credentials.createInsecure();
+    this.metadata = config.metadata;
+    this.headers =
+      parseHeaders(config.headers, this.logger) || this.DEFAULT_HEADERS;
+  }
+
+  onShutdown(): void {
+    this.isShutDown = true;
+    if (this.metricServiceClient) {
+      this.metricServiceClient.close();
+    }
+  }
+
+  onInit(): void {
+    this.isShutDown = false;
+    const serverAddress = removeProtocol(this.url);
+    const metricServiceProtoPath =
+      'opentelemetry/proto/collector/metrics/v1/metrics_service.proto';
+    const includeDirs = [path.resolve(__dirname, 'protos')];
+    protoLoader
+      .load(metricServiceProtoPath, {
+        keepCase: false,
+        longs: String,
+        enums: String,
+        defaults: true,
+        oneofs: true,
+        includeDirs,
+      })
+      .then(packageDefinition => {
+        const packageObject: any = grpc.loadPackageDefinition(
+          packageDefinition
+        );
+        this.metricServiceClient = new packageObject.opentelemetry.proto.collector.metrics.v1.MetricsService(
+          serverAddress,
+          this.credentials
+        );
+        if (this.grpcMetricsQueue.length > 0) {
+          const queue = this.grpcMetricsQueue.splice(0);
+          queue.forEach((item: GRPCMetricQueueItem) => {
+            this.sendMetrics(item.metrics, item.onSuccess, item.onError);
+          });
+        }
+      });
+  }
+
+  sendMetrics(
+    metrics: MetricRecord[],
+    onSuccess: () => void,
+    onError: (error: collectorTypes.CollectorExporterError) => void
+  ): void {
+    if (this.isShutDown) {
+      this.logger.debug('Shutdown already started. Cannot send metrics');
+      return;
+    }
+    if (this.metricServiceClient) {
+      const exportMetricServiceRequest = toCollectorExportMetricServiceRequest(
+        metrics,
+        this._startTime,
+        this
+      );
+      this.metricServiceClient.export(
+        exportMetricServiceRequest,
+        this.metadata,
+        (err: collectorTypes.ExportServiceError) => {
+          if (err) {
+            this.logger.error(
+              'exportMetricServiceRequest',
+              exportMetricServiceRequest
+            );
+            onError(err);
+          } else {
+            onSuccess();
+          }
+        }
+      );
+    } else {
+      this.grpcMetricsQueue.push({
+        metrics,
+        onSuccess,
+        onError,
+      });
+    }
+  }
+
+  getDefaultUrl(url: string | undefined): string {
+    return url || DEFAULT_COLLECTOR_URL;
+  }
+}

packages/opentelemetry-exporter-collector/src/platform/node/index.ts

@@ -15,3 +15,4 @@
  */
 
 export * from './CollectorTraceExporter';
+export * from './CollectorMetricExporter';

packages/opentelemetry-exporter-collector/src/platform/node/types.ts

@@ -17,6 +17,7 @@
 import * as grpc from 'grpc';
 import { ReadableSpan } from '@opentelemetry/tracing';
 import { CollectorProtocolNode } from '../../enums';
+import { MetricRecord } from '@opentelemetry/metrics';
 import {
   CollectorExporterError,
   CollectorExporterConfigBase,
@@ -32,6 +33,15 @@ export interface GRPCSpanQueueItem {
   onError: (error: CollectorExporterError) => void;
 }
 
+/**
+ * Queue item to be used to save temporary metrics
+ */
+export interface GRPCMetricQueueItem {
+  metrics: MetricRecord[];
+  onSuccess: () => void;
+  onError: (error: CollectorExporterError) => void;
+}
+
 /**
  * Service Client for sending spans or metrics
  */

packages/opentelemetry-exporter-collector/src/platform/node/utilWithJson.ts

@@ -23,6 +23,10 @@ import * as collectorTypes from '../../types';
 import { toCollectorExportTraceServiceRequest } from '../../transform';
 import { CollectorTraceExporter } from './CollectorTraceExporter';
 import { CollectorExporterConfigNode } from './types';
+import { CollectorMetricExporter } from './CollectorMetricExporter';
+import { MetricRecord } from '@opentelemetry/metrics';
+import { toCollectorExportMetricServiceRequest } from '../../transformMetrics';
+import { Logger } from '@opentelemetry/api';
 
 export const DEFAULT_COLLECTOR_URL_JSON = 'http://localhost:55680/v1/trace';
 
@@ -33,6 +37,30 @@ export function onInitWithJson(
   // nothing to be done for json yet
 }
 
+export function sendMetricsUsingJson(
+  collector: CollectorMetricExporter,
+  metrics: MetricRecord[],
+  startTime: number,
+  onSuccess: () => void,
+  onError: (error: collectorTypes.CollectorExporterError) => void
+): void {
+  const exportMetricServiceRequest = toCollectorExportMetricServiceRequest(
+    metrics,
+    startTime,
+    collector
+  );
+
+  const body = JSON.stringify(exportMetricServiceRequest);
+  _sendWithJson(
+    body,
+    collector.url,
+    collector.headers,
+    collector.logger,
+    onSuccess,
+    onError
+  );
+}
+
 export function sendSpansUsingJson(
   collector: CollectorTraceExporter,
   spans: ReadableSpan[],
@@ -45,7 +73,25 @@ export function sendSpansUsingJson(
   );
 
   const body = JSON.stringify(exportTraceServiceRequest);
-  const parsedUrl = new url.URL(collector.url);
+  _sendWithJson(
+    body,
+    collector.url,
+    collector.headers,
+    collector.logger,
+    onSuccess,
+    onError
+  );
+}
+
+function _sendWithJson(
+  body: string,
+  collectorUrl: string,
+  collectorHeaders: Partial<Record<string, unknown>>,
+  logger: Logger,
+  onSuccess: () => void,
+  onError: (error: collectorTypes.CollectorExporterError) => void
+): void {
+  const parsedUrl = new url.URL(collectorUrl);
 
   const options = {
     hostname: parsedUrl.hostname,
@@ -55,17 +101,17 @@ export function sendSpansUsingJson(
     headers: {
       'Content-Length': Buffer.byteLength(body),
       'Content-Type': 'application/json',
-      ...collector.headers,
+      ...collectorHeaders,
     },
   };
 
   const request = parsedUrl.protocol === 'http:' ? http.request : https.request;
   const req = request(options, (res: http.IncomingMessage) => {
     if (res.statusCode && res.statusCode < 299) {
-      collector.logger.debug(`statusCode: ${res.statusCode}`);
+      logger.debug(`statusCode: ${res.statusCode}`);
       onSuccess();
     } else {
-      collector.logger.error(`statusCode: ${res.statusCode}`);
+      logger.error(`statusCode: ${res.statusCode}`);
       onError({
         code: res.statusCode,
         message: res.statusMessage,
@@ -74,7 +120,7 @@ export function sendSpansUsingJson(
   });
 
   req.on('error', (error: Error) => {
-    collector.logger.error('error', error.message);
+    logger.error('error', error.message);
     onError({
       message: error.message,
     });

packages/opentelemetry-exporter-collector/test/common/transformMetrics.test.ts

@@ -33,26 +33,40 @@ import { HistogramAggregator } from '@opentelemetry/metrics';
 import { hrTimeToNanoseconds } from '@opentelemetry/core';
 describe('transformMetrics', () => {
   describe('toCollectorMetric', () => {
-    it('should convert metric', () => {
+    beforeEach(() => {
+      // Counter
       mockCounter.aggregator.update(1);
+
+      // Observer
+      mockObserver.aggregator.update(10);
+
+      // Histogram
+      mockHistogram.aggregator.update(7);
+      mockHistogram.aggregator.update(14);
+      (mockHistogram.aggregator as HistogramAggregator).reset();
+
+      // ValueRecorder
+      mockValueRecorder.aggregator.update(5);
+    });
+
+    afterEach(() => {
+      mockCounter.aggregator.update(-1); // Reset counter
+      (mockHistogram.aggregator as HistogramAggregator).reset();
+    });
+    it('should convert metric', () => {
       ensureCounterIsCorrect(
         transform.toCollectorMetric(mockCounter, 1592602232694000000),
         hrTimeToNanoseconds(mockCounter.aggregator.toPoint().timestamp)
       );
-      mockObserver.aggregator.update(10);
       ensureObserverIsCorrect(
         transform.toCollectorMetric(mockObserver, 1592602232694000000),
         hrTimeToNanoseconds(mockObserver.aggregator.toPoint().timestamp)
       );
-      mockHistogram.aggregator.update(7);
-      mockHistogram.aggregator.update(14);
-      (mockHistogram.aggregator as HistogramAggregator).reset();
       ensureHistogramIsCorrect(
         transform.toCollectorMetric(mockHistogram, 1592602232694000000),
         hrTimeToNanoseconds(mockHistogram.aggregator.toPoint().timestamp)
       );
 
-      mockValueRecorder.aggregator.update(5);
       ensureValueRecorderIsCorrect(
         transform.toCollectorMetric(mockValueRecorder, 1592602232694000000),
         hrTimeToNanoseconds(mockValueRecorder.aggregator.toPoint().timestamp)

packages/opentelemetry-exporter-collector/test/helper.ts

@@ -757,6 +757,51 @@ export function ensureHistogramIsCorrect(
   });
 }
 
+export function ensureExportedCounterIsCorrect(
+  metric: collectorTypes.opentelemetryProto.metrics.v1.Metric
+) {
+  assert.deepStrictEqual(metric.metricDescriptor, {
+    name: 'test-counter',
+    description: 'sample counter description',
+    unit: '1',
+    type: 'MONOTONIC_INT64',
+    temporality: 'CUMULATIVE',
+  });
+  assert.deepStrictEqual(metric.doubleDataPoints, []);
+  assert.deepStrictEqual(metric.summaryDataPoints, []);
+  assert.deepStrictEqual(metric.histogramDataPoints, []);
+  assert.ok(metric.int64DataPoints);
+  assert.deepStrictEqual(metric.int64DataPoints[0].labels, []);
+  assert.deepStrictEqual(metric.int64DataPoints[0].value, '1');
+  assert.deepStrictEqual(
+    metric.int64DataPoints[0].startTimeUnixNano,
+    '1592602232694000128'
+  );
+}
+
+export function ensureExportedObserverIsCorrect(
+  metric: collectorTypes.opentelemetryProto.metrics.v1.Metric
+) {
+  assert.deepStrictEqual(metric.metricDescriptor, {
+    name: 'test-observer',
+    description: 'sample observer description',
+    unit: '2',
+    type: 'DOUBLE',
+    temporality: 'DELTA',
+  });
+
+  assert.deepStrictEqual(metric.int64DataPoints, []);
+  assert.deepStrictEqual(metric.summaryDataPoints, []);
+  assert.deepStrictEqual(metric.histogramDataPoints, []);
+  assert.ok(metric.doubleDataPoints);
+  assert.deepStrictEqual(metric.doubleDataPoints[0].labels, []);
+  assert.deepStrictEqual(metric.doubleDataPoints[0].value, 10);
+  assert.deepStrictEqual(
+    metric.doubleDataPoints[0].startTimeUnixNano,
+    '1592602232694000128'
+  );
+}
+
 export function ensureResourceIsCorrect(
   resource: collectorTypes.opentelemetryProto.resource.v1.Resource
 ) {