feat: adding json over http for collector exporter

examples/collector-exporter-node/package.json

@@ -27,6 +27,7 @@
   },
   "dependencies": {
     "@opentelemetry/api": "^0.9.0",
+    "@opentelemetry/core": "^0.9.0",
     "@opentelemetry/exporter-collector": "^0.9.0",
     "@opentelemetry/tracing": "^0.9.0"
   },

examples/collector-exporter-node/start.js

@@ -2,12 +2,13 @@
 
 const opentelemetry = require('@opentelemetry/api');
 const { BasicTracerProvider, SimpleSpanProcessor } = require('@opentelemetry/tracing');
+const { ConsoleLogger, LogLevel } = require('@opentelemetry/core');
 const { CollectorExporter } = require('@opentelemetry/exporter-collector');
 
-const address = '127.0.0.1:55678';
 const exporter = new CollectorExporter({
+  logger: new ConsoleLogger(LogLevel.DEBUG),
   serviceName: 'basic-service',
-  url: address,
+  // useJson: true,
 });
 
 const provider = new BasicTracerProvider();

packages/opentelemetry-exporter-collector/README.md

@@ -36,7 +36,7 @@ provider.register();
 
 ```
 
-## Usage in Node
+## Usage in Node - GRPC
 
 The CollectorExporter in Node expects the URL to only be the hostname. It will not work with `/v1/trace`.
 
@@ -109,6 +109,29 @@ provider.register();
 
 Note, that this will only work if TLS is also configured on the server.
 
+## Usage in Node - JSON over http
+
+```js
+const { BasicTracerProvider, SimpleSpanProcessor } = require('@opentelemetry/tracing');
+const { CollectorExporter } =  require('@opentelemetry/exporter-collector');
+
+const collectorOptions = {
+  useJson: true, // this needs to be set to true
+  serviceName: 'basic-service',
+  url: '<opentelemetry-collector-url>', // url is optional and can be omitted - default is http://localhost:55678/v1/trace
+  headers: {
+    foo: 'bar'
+  }, //an optional object containing custom headers to be sent with each request will only work with json over http
+};
+
+const provider = new BasicTracerProvider();
+const exporter = new CollectorExporter(collectorOptions);
+provider.addSpanProcessor(new SimpleSpanProcessor(exporter));
+
+provider.register();
+
+```
+
 ## Running opentelemetry-collector locally to see the traces
 
 1. Go to examples/basic-tracer-node

packages/opentelemetry-exporter-collector/src/CollectorExporterBase.ts

@@ -50,7 +50,7 @@ export abstract class CollectorExporterBase<
    */
   constructor(config: T = {} as T) {
     this.serviceName = config.serviceName || DEFAULT_SERVICE_NAME;
-    this.url = this.getDefaultUrl(config.url);
+    this.url = this.getDefaultUrl(config);
     if (typeof config.hostName === 'string') {
       this.hostName = config.hostName;
     }
@@ -134,5 +134,5 @@ export abstract class CollectorExporterBase<
     onSuccess: () => void,
     onError: (error: CollectorExporterError) => void
   ): void;
-  abstract getDefaultUrl(url: string | undefined): string;
+  abstract getDefaultUrl(config: T): string;
 }

packages/opentelemetry-exporter-collector/src/platform/browser/CollectorExporter.ts

@@ -61,8 +61,8 @@ export class CollectorExporter extends CollectorExporterBase<
     window.removeEventListener('unload', this.shutdown);
   }
 
-  getDefaultUrl(url: string | undefined) {
-    return url || DEFAULT_COLLECTOR_URL;
+  getDefaultUrl(config: CollectorExporterConfig) {
+    return config.url || DEFAULT_COLLECTOR_URL;
   }
 
   sendSpans(

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

@@ -14,29 +14,34 @@
  * limitations under the License.
  */
 
-import * as protoLoader from '@grpc/proto-loader';
-import * as grpc from 'grpc';
-import * as path from 'path';
-import * as collectorTypes from '../../types';
-
 import { ReadableSpan } from '@opentelemetry/tracing';
+import * as grpc from 'grpc';
 import {
   CollectorExporterBase,
   CollectorExporterConfigBase,
 } from '../../CollectorExporterBase';
-import { CollectorExporterError } from '../../types';
-import { toCollectorExportTraceServiceRequest } from '../../transform';
+import * as collectorTypes from '../../types';
+import {
+  DEFAULT_COLLECTOR_URL_GRPC,
+  onInitWithGrpc,
+  sendSpansUsingGrpc,
+} from './utilWithGrpc';
+import {
+  DEFAULT_COLLECTOR_URL_JSON,
+  onInitWithJson,
+  sendSpansUsingJson,
+} from './utilWithJson';
 import { GRPCQueueItem, TraceServiceClient } from './types';
-import { removeProtocol } from './util';
-
-const DEFAULT_COLLECTOR_URL = 'localhost:55678';
 
 /**
  * Collector Exporter Config for Node
+ * headers will only work if useJson is set to true
  */
 export interface CollectorExporterConfig extends CollectorExporterConfigBase {
   credentials?: grpc.ChannelCredentials;
   metadata?: grpc.Metadata;
+  headers?: { [key: string]: string };
+  useJson?: boolean;
 }
 
 /**
@@ -45,17 +50,29 @@ export interface CollectorExporterConfig extends CollectorExporterConfigBase {
 export class CollectorExporter extends CollectorExporterBase<
   CollectorExporterConfig
 > {
+  DEFAULT_HEADERS: { [key: string]: string } = {
+    [collectorTypes.OT_REQUEST_HEADER]: '1',
+  };
   isShutDown: boolean = false;
   traceServiceClient?: TraceServiceClient = undefined;
   grpcSpansQueue: GRPCQueueItem[] = [];
   metadata?: grpc.Metadata;
+  headers: { [key: string]: string };
+  private readonly _useJson: boolean = false;
 
   /**
    * @param config
    */
   constructor(config: CollectorExporterConfig = {}) {
     super(config);
+    this._useJson = !!config.useJson;
+    if (this._useJson) {
+      this.logger.debug('CollectorExporter - using json over http');
+    } else {
+      this.logger.debug('CollectorExporter - using grpc');
+    }
     this.metadata = config.metadata;
+    this.headers = config.headers || this.DEFAULT_HEADERS;
   }
 
   onShutdown(): void {
@@ -67,82 +84,35 @@ export class CollectorExporter extends CollectorExporterBase<
 
   onInit(config: CollectorExporterConfig): void {
     this.isShutDown = false;
-    this.grpcSpansQueue = [];
-    const serverAddress = removeProtocol(this.url);
-    const credentials: grpc.ChannelCredentials =
-      config.credentials || grpc.credentials.createInsecure();
 
-    const traceServiceProtoPath =
-      'opentelemetry/proto/collector/trace/v1/trace_service.proto';
-    const includeDirs = [path.resolve(__dirname, 'protos')];
-
-    protoLoader
-      .load(traceServiceProtoPath, {
-        keepCase: false,
-        longs: String,
-        enums: String,
-        defaults: true,
-        oneofs: true,
-        includeDirs,
-      })
-      .then(packageDefinition => {
-        const packageObject: any = grpc.loadPackageDefinition(
-          packageDefinition
-        );
-        this.traceServiceClient = new packageObject.opentelemetry.proto.collector.trace.v1.TraceService(
-          serverAddress,
-          credentials
-        );
-        if (this.grpcSpansQueue.length > 0) {
-          const queue = this.grpcSpansQueue.splice(0);
-          queue.forEach((item: GRPCQueueItem) => {
-            this.sendSpans(item.spans, item.onSuccess, item.onError);
-          });
-        }
-      });
+    if (config.useJson) {
+      onInitWithJson(this, config);
+    } else {
+      onInitWithGrpc(this, config);
+    }
   }
 
   sendSpans(
     spans: ReadableSpan[],
     onSuccess: () => void,
-    onError: (error: CollectorExporterError) => void
+    onError: (error: collectorTypes.CollectorExporterError) => void
   ): void {
     if (this.isShutDown) {
       return;
     }
-    if (this.traceServiceClient) {
-      const exportTraceServiceRequest = toCollectorExportTraceServiceRequest(
-        spans,
-        this
-      );
-
-      this.traceServiceClient.export(
-        exportTraceServiceRequest,
-        this.metadata,
-        (
-          err: collectorTypes.opentelemetryProto.collector.trace.v1.ExportTraceServiceError
-        ) => {
-          if (err) {
-            this.logger.error(
-              'exportTraceServiceRequest',
-              exportTraceServiceRequest
-            );
-            onError(err);
-          } else {
-            onSuccess();
-          }
-        }
-      );
+    if (this._useJson) {
+      sendSpansUsingJson(this, spans, onSuccess, onError);
     } else {
-      this.grpcSpansQueue.push({
-        spans,
-        onSuccess,
-        onError,
-      });
+      sendSpansUsingGrpc(this, spans, onSuccess, onError);
     }
   }
 
-  getDefaultUrl(url: string | undefined): string {
-    return url || DEFAULT_COLLECTOR_URL;
+  getDefaultUrl(config: CollectorExporterConfig): string {
+    if (!config.url) {
+      return config.useJson
+        ? DEFAULT_COLLECTOR_URL_JSON
+        : DEFAULT_COLLECTOR_URL_GRPC;
+    }
+    return config.url;
   }
 }

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

@@ -0,0 +1,107 @@
+/*
+ * 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 * as protoLoader from '@grpc/proto-loader';
+import * as grpc from 'grpc';
+import * as path from 'path';
+import * as collectorTypes from '../../types';
+
+import { ReadableSpan } from '@opentelemetry/tracing';
+import { CollectorExporterError } from '../../types';
+import { toCollectorExportTraceServiceRequest } from '../../transform';
+import {
+  CollectorExporter,
+  CollectorExporterConfig,
+} from './CollectorExporter';
+import { GRPCQueueItem } from './types';
+import { removeProtocol } from './util';
+
+export const DEFAULT_COLLECTOR_URL_GRPC = 'localhost:55678';
+
+export function onInitWithGrpc(
+  collector: CollectorExporter,
+  config: CollectorExporterConfig
+): void {
+  collector.grpcSpansQueue = [];
+  const serverAddress = removeProtocol(collector.url);
+  const credentials: grpc.ChannelCredentials =
+    config.credentials || grpc.credentials.createInsecure();
+
+  const traceServiceProtoPath =
+    'opentelemetry/proto/collector/trace/v1/trace_service.proto';
+  const includeDirs = [path.resolve(__dirname, 'protos')];
+
+  protoLoader
+    .load(traceServiceProtoPath, {
+      keepCase: false,
+      longs: String,
+      enums: String,
+      defaults: true,
+      oneofs: true,
+      includeDirs,
+    })
+    .then(packageDefinition => {
+      const packageObject: any = grpc.loadPackageDefinition(packageDefinition);
+      collector.traceServiceClient = new packageObject.opentelemetry.proto.collector.trace.v1.TraceService(
+        serverAddress,
+        credentials
+      );
+      if (collector.grpcSpansQueue.length > 0) {
+        const queue = collector.grpcSpansQueue.splice(0);
+        queue.forEach((item: GRPCQueueItem) => {
+          collector.sendSpans(item.spans, item.onSuccess, item.onError);
+        });
+      }
+    });
+}
+
+export function sendSpansUsingGrpc(
+  collector: CollectorExporter,
+  spans: ReadableSpan[],
+  onSuccess: () => void,
+  onError: (error: CollectorExporterError) => void
+): void {
+  if (collector.traceServiceClient) {
+    const exportTraceServiceRequest = toCollectorExportTraceServiceRequest(
+      spans,
+      collector
+    );
+    collector.traceServiceClient.export(
+      exportTraceServiceRequest,
+      collector.metadata,
+      (
+        err: collectorTypes.opentelemetryProto.collector.trace.v1.ExportTraceServiceError
+      ) => {
+        if (err) {
+          collector.logger.error(
+            'exportTraceServiceRequest',
+            exportTraceServiceRequest
+          );
+          onError(err);
+        } else {
+          collector.logger.debug('spans sent');
+          onSuccess();
+        }
+      }
+    );
+  } else {
+    collector.grpcSpansQueue.push({
+      spans,
+      onSuccess,
+      onError,
+    });
+  }
+}