Merge 3f9dad2 into d5cf120

.gitmodules

@@ -4,3 +4,6 @@
 [submodule "experimental/packages/otlp-proto-exporter-base/protos"]
 	path = experimental/packages/otlp-proto-exporter-base/protos
 	url = https://github.com/open-telemetry/opentelemetry-proto.git
+[submodule "experimental/packages/exporter-trace-otlp-grpc/protos"]
+	path = experimental/packages/exporter-trace-otlp-grpc/protos
+	url = https://github.com/open-telemetry/opentelemetry-proto

experimental/CHANGELOG.md

@@ -19,6 +19,8 @@ All notable changes to experimental packages in this project will be documented
 ### :bug: (Bug Fix)
 
 * fix(histogram): fix maximum when only values < -1 are provided [#3086](https://github.com/open-telemetry/opentelemetry-js/pull/3086) @pichlermarc
+* fix(exporter-grpc): prevent trace and metrics grpc service clients from interfering with each other [#3100](https://github.com/open-telemetry/opentelemetry-js/pull/3100) @dyladan
+  * refactor trace grpc exporter to not use the grpc exporter base class
 * fix(sdk-metrics-base): fix PeriodicExportingMetricReader keeping Node.js process from exiting
   [#3106](https://github.com/open-telemetry/opentelemetry-js/pull/3106) @seemk
 

experimental/packages/exporter-trace-otlp-grpc/package.json

@@ -49,7 +49,7 @@
   "devDependencies": {
     "@babel/core": "7.16.0",
     "@opentelemetry/api": "^1.0.0",
-    "@opentelemetry/otlp-exporter-base": "0.30.0",
+    "@opentelemetry/resources": "1.4.0",
     "@types/mocha": "8.2.3",
     "@types/node": "14.17.33",
     "@types/sinon": "10.0.6",
@@ -67,12 +67,10 @@
     "@opentelemetry/api": "^1.0.0"
   },
   "dependencies": {
-    "@grpc/grpc-js": "^1.5.9",
-    "@grpc/proto-loader": "^0.6.9",
+    "@grpc/grpc-js": "1.5.9",
+    "@grpc/proto-loader": "0.6.9",
     "@opentelemetry/core": "1.4.0",
-    "@opentelemetry/otlp-grpc-exporter-base": "0.30.0",
     "@opentelemetry/otlp-transformer": "0.30.0",
-    "@opentelemetry/resources": "1.4.0",
     "@opentelemetry/sdk-trace-base": "1.4.0"
   }
 }

experimental/packages/exporter-trace-otlp-grpc/src/OTLPTraceExporter.ts

@@ -14,58 +14,100 @@
  * limitations under the License.
  */
 
-import { ReadableSpan, SpanExporter } from '@opentelemetry/sdk-trace-base';
-import { baggageUtils, getEnv } from '@opentelemetry/core';
-import { Metadata } from '@grpc/grpc-js';
-import {
-  OTLPGRPCExporterConfigNode,
-  OTLPGRPCExporterNodeBase,
-  ServiceClientType,
-  validateAndNormalizeUrl,
-  DEFAULT_COLLECTOR_URL
-} from '@opentelemetry/otlp-grpc-exporter-base';
-import { createExportTraceServiceRequest, IExportTraceServiceRequest } from '@opentelemetry/otlp-transformer';
+import * as grpc from '@grpc/grpc-js';
+import { loadSync } from '@grpc/proto-loader';
+import { ExportResult, ExportResultCode, getEnv } from '@opentelemetry/core';
+import { createExportTraceServiceRequest } from '@opentelemetry/otlp-transformer';
+import type { ReadableSpan, SpanExporter } from '@opentelemetry/sdk-trace-base';
+import * as path from 'path';
+import { RequestCounter } from './internal/request-counter';
+import { TraceServiceClient } from './internal/types';
+import type { OTLPGRPCTraceExporterConfig } from './types';
+import { getConnectionOptions } from './util';
 
 /**
  * OTLP Trace Exporter for Node
  */
-export class OTLPTraceExporter
-  extends OTLPGRPCExporterNodeBase<ReadableSpan,
-    IExportTraceServiceRequest>
-  implements SpanExporter {
+export class OTLPTraceExporter implements SpanExporter {
+  private _metadata: grpc.Metadata;
+  private _serviceClient: TraceServiceClient;
+  private _timeoutMillis: number;
+  private _requestCounter = new RequestCounter();
+  private _isShutdown = false;
 
-  constructor(config: OTLPGRPCExporterConfigNode = {}) {
-    super(config);
-    const headers = baggageUtils.parseKeyPairsIntoRecord(getEnv().OTEL_EXPORTER_OTLP_TRACES_HEADERS);
-    this.metadata ||= new Metadata();
-    for (const [k, v] of Object.entries(headers)) {
-      this.metadata.set(k, v);
-    }
-  }
+  constructor(config: OTLPGRPCTraceExporterConfig = {}) {
+    const { host, credentials, metadata, compression } = getConnectionOptions(config, getEnv());
+    this._metadata = metadata;
+    this._timeoutMillis = config.timeoutMillis ?? 10_000;
 
-  convert(spans: ReadableSpan[]): IExportTraceServiceRequest {
-    return createExportTraceServiceRequest(spans);
-  }
+    const packageDefinition = loadSync('opentelemetry/proto/collector/trace/v1/trace_service.proto', {
+      keepCase: false,
+      longs: String,
+      enums: String,
+      defaults: true,
+      oneofs: true,
+      includeDirs: [
+        // In webpack environments, the bundle may be at the same level as the protos/ directory
+        path.resolve(__dirname, 'protos'),
+        // When running typescript directly in tests or with ts-node, the protos/ directory is one level above the src/ directory
+        path.resolve(__dirname, '..', 'protos'),
+        // When running the compiled `js, the protos directory is two levels above the build/src/ directory
+        path.resolve(__dirname, '..', '..', 'protos'),
+      ],
+    });
 
-  getDefaultUrl(config: OTLPGRPCExporterConfigNode) {
-    return validateAndNormalizeUrl(this.getUrlFromConfig(config));
+    // any required here because the types returned by loadSync don't know our package structure
+    const packageObject: any = grpc.loadPackageDefinition(packageDefinition);
+    const channelOptions: grpc.ChannelOptions = {
+      'grpc.default_compression_algorithm': compression,
+    };
+    this._serviceClient = new packageObject.opentelemetry.proto.collector.trace.v1.TraceService(host, credentials, channelOptions);
   }
 
-  getServiceClientType() {
-    return ServiceClientType.SPANS;
-  }
+  export(spans: ReadableSpan[], resultCallback: (result: ExportResult) => void): void {
+    if (this._isShutdown) {
+      setImmediate(resultCallback, {
+        code: ExportResultCode.FAILED,
+        error: new Error('Cannot export after shutdown'),
+      });
+      return;
+    }
 
-  getServiceProtoPath(): string {
-    return 'opentelemetry/proto/collector/trace/v1/trace_service.proto';
+    const deadline = Date.now() + this._timeoutMillis;
+    this._requestCounter.startRequest();
+
+    const req = createExportTraceServiceRequest(spans);
+    this._serviceClient.export(
+      req,
+      this._metadata,
+      { deadline },
+      (error: Error) => {
+        this._requestCounter.endRequest();
+
+        if (error) {
+          resultCallback({
+            code: ExportResultCode.FAILED,
+            error,
+          });
+        } else {
+          resultCallback({
+            code: ExportResultCode.SUCCESS,
+          });
+        }
+      },
+    );
   }
 
-  getUrlFromConfig(config: OTLPGRPCExporterConfigNode): string {
-    if (typeof config.url === 'string') {
-      return config.url;
-    }
+  shutdown(): Promise<void> {
+    this._isShutdown = true;
+    return new Promise((resolve, _reject) => {
+      if (this._requestCounter.requests === 0) {
+        resolve();
+        return;
+      }
 
-    return getEnv().OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ||
-      getEnv().OTEL_EXPORTER_OTLP_ENDPOINT ||
-      DEFAULT_COLLECTOR_URL;
+      this._requestCounter.onEndLastRequest(resolve);
+    });
   }
 }
+

experimental/packages/exporter-trace-otlp-grpc/src/index.ts

@@ -14,4 +14,5 @@
  * limitations under the License.
  */
 
-export * from './OTLPTraceExporter';
+export { OTLPTraceExporter } from './OTLPTraceExporter';
+export { CompressionAlgorithm, OTLPGRPCTraceExporterConfig } from './types';

experimental/packages/exporter-trace-otlp-grpc/src/internal/request-counter.ts

@@ -0,0 +1,48 @@
+/*
+ * 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 EventEmitter = require('events');
+
+export class RequestCounter {
+  public requests = 0;
+  private _e = new EventEmitter();
+
+  startRequest() {
+    this.requests += 1;
+  }
+
+  endRequest() {
+    if (this.requests === 0) {
+      throw new Error('Tried to end more requests than were started');
+    }
+    this.requests -= 1;
+    this._e.emit('endRequest');
+  }
+
+  onEndLastRequest(cb: () => void) {
+    if (this.requests === 0) {
+      setImmediate(cb);
+    }
+
+    const self = this;
+    this._e.on('endRequest', function onEndRequest() {
+      if (self.requests === 0) {
+        self._e.removeListener('endRequest', onEndRequest);
+        cb();
+      }
+    });
+  }
+}

experimental/packages/exporter-trace-otlp-grpc/src/internal/types.ts

@@ -0,0 +1,34 @@
+/*
+ * 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 type * as grpc from '@grpc/grpc-js';
+
+export interface TraceServiceClient extends grpc.Client {
+  export: (
+    request: any,
+    metadata: grpc.Metadata,
+    options: grpc.CallOptions,
+    callback: Function
+  ) => {};
+}
+
+
+export type ConnectionOptions = {
+  host: string;
+  metadata: grpc.Metadata;
+  credentials: grpc.ChannelCredentials;
+  compression: grpc.compressionAlgorithms;
+};

experimental/packages/exporter-trace-otlp-grpc/src/types.ts

@@ -0,0 +1,33 @@
+/*
+ * 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 type * as grpc from '@grpc/grpc-js';
+
+export type OTLPGRPCTraceExporterConfig = {
+  credentials?: grpc.ChannelCredentials;
+  metadata?: grpc.Metadata;
+  compression?: CompressionAlgorithm;
+
+  url?: string;
+  /** Maximum time the OTLP exporter will wait for each batch export.
+   * The default value is 10000ms. */
+  timeoutMillis?: number;
+};
+
+export enum CompressionAlgorithm {
+  NONE = 'none',
+  GZIP = 'gzip'
+}