feat: grpc export

README.md

@@ -9,7 +9,7 @@ It's intended to be included in Node.js core as a dependency, and wired up in th
 It seeks to enable:
 
 * Instrumentation of inbound and outbound HTTP transactions, following [OTel Semconv](https://opentelemetry.io/docs/specs/semconv/general/trace/).
-* Tracing data egress via OTLP/JSON. (JSON-only to avoid bringing in a gRPC dependency.)
+* Tracing data egress via OTLP/HTTP JSON or OTLP/gRPC, without runtime dependencies.
 * Distributed tracing via W3C Traceparent, including forwarding baggage (but not modifying it).
 
 Anything else, including the following, should be considered initially out of scope:
@@ -48,7 +48,8 @@ request it happened on carries on untraced or partly traced.
 
 | Option           | Type   | Default                      | Description |
 | ---------------- | ------ | ---------------------------- | ----------- |
-| `endpoint`       | string | `http://localhost:4318`      | Base URL of an OTLP/HTTP collector. Only `http:` and `https:` URLs are accepted. `/v1/traces` is appended to the endpoint's existing path. An endpoint already ending with `/v1/traces` is used as is. |
+| `protocol`       | string | `json`                 | Export protocol: `json` or `grpc`. |
+| `endpoint`       | string | protocol dependent          | Collector URL. Defaults to `http://localhost:4318` for `json`, or `http://localhost:4317` for `grpc`. Only `http:` and `https:` URLs are accepted. For HTTP/JSON, `/v1/traces` is appended to the existing path unless already present. For gRPC, the path is replaced with `/opentelemetry.proto.collector.trace.v1.TraceService/Export`. |
 | `serviceName`    | string | `unknown_service:node`      | Service name in exported resource attributes. The default follows the OpenTelemetry [semantic conventions](https://opentelemetry.io/docs/specs/semconv/general/trace/) for low-cardinality service names. |
 | `sdkName`        | string | `microtel`                   | `telemetry.sdk.name` resource attribute, and the name of the instrumentation scope on exported spans. |
 | `sdkVersion`    | string | this package's version      | `telemetry.sdk.version` resource attribute, and the scope version. |
@@ -68,6 +69,12 @@ per-process random `service.instance.id` UUID, `host.name` from
 can replace any of microtel's values except `service.name` and the
 `telemetry.sdk.*` keys.
 
+To enable gRPC export, set `protocol` to `grpc`:
+
+```js
+otel.init({ protocol: 'grpc' });
+```
+
 ### Exported functions
 
 * `init(options)` activates tracing. Returns nothing. Throws on invalid options.

benchmark/helpers/otlp.js

@@ -0,0 +1,56 @@
+'use strict';
+
+const { OtlpEncoder } = require('../../lib/otlp');
+const { Span, SPAN_KIND_CLIENT } = require('../../lib/span');
+const { version } = require('../../package.json');
+
+// Prebuild ended span models so benchmarks measure encoding/export rather
+// than instrumentation, ID generation, or span construction.
+function createInput(batchSize, exceptions) {
+  const spans = Array.from({ length: batchSize }, (_, i) => {
+    const span = new Span('GET', SPAN_KIND_CLIENT);
+    span.setAttribute('http.request.method', 'GET');
+    span.setAttribute('url.full', `https://example.com/items/${i}`);
+    span.setAttribute('server.address', 'example.com');
+    span.setAttribute('server.port', 443);
+    span.setAttribute('network.protocol.version', '1.1');
+    span.setAttribute('http.response.status_code', exceptions ? 503 : 200);
+    if (exceptions) {
+      span.setStatus(2, 'HTTP 503');
+      span.setAttribute('error.type', '503');
+      span.addEvent('exception', {
+        'exception.type': 'Error',
+        'exception.message': 'Unavailable 日本語 💥',
+        'exception.stacktrace':
+          'Error: unavailable\n' +
+          '    at handler (/app/server.js:42:10)\n'.repeat(20),
+      });
+    }
+    // Do not call end(): that would enqueue the span for automatic export.
+    span.endTime = span.startTime + 10;
+    return span;
+  });
+  const resource = {
+    attributes: Object.entries({
+      'service.name': 'benchmark',
+      'telemetry.sdk.name': 'microtel',
+      'telemetry.sdk.language': 'nodejs',
+      'telemetry.sdk.version': version,
+      'process.runtime.name': 'nodejs',
+      'process.runtime.version': process.version,
+      'process.pid': process.pid,
+    }).map(([key, value]) => ({
+      key,
+      value:
+        typeof value === 'number'
+          ? { intValue: String(value) }
+          : { stringValue: value },
+    })),
+  };
+  return {
+    spans,
+    encoder: new OtlpEncoder(resource, { name: 'microtel', version }),
+  };
+}
+
+module.exports = { createInput };

benchmark/otlp-encoding.bench.js

@@ -0,0 +1,32 @@
+'use strict';
+
+const assert = require('node:assert/strict');
+const { bench, suite } = require('node:bench');
+const { createInput } = require('./helpers/otlp');
+
+suite('OTLP encoding', () => {
+  for (const batchSize of [1, 100]) {
+    for (const exceptions of [false, true]) {
+      const { encoder, spans } = createInput(batchSize, exceptions);
+      const iterations = 100000 / batchSize;
+      const reference = encoder.encode(spans);
+      const expected = reference.length + reference[reference.length - 1];
+      bench(
+        `OTLP encode / batch=${batchSize} / ${exceptions ? 'exception' : 'HTTP'}`,
+        { samples: 30, warmup: 20, params: { batchSize, exceptions } },
+        (b) => {
+          let checksum = 0;
+          b.start();
+          for (let i = 0; i < iterations; i++) {
+            const buffer = encoder.encode(spans);
+            checksum += buffer.length + buffer[buffer.length - 1];
+          }
+          // One operation is a complete batch, including its sizing pass,
+          // allocation, protobuf encoding and gRPC message prefix.
+          b.end(iterations);
+          assert.equal(checksum, iterations * expected);
+        },
+      );
+    }
+  }
+});

benchmark/otlp-export.bench.js

@@ -0,0 +1,71 @@
+'use strict';
+
+const assert = require('node:assert/strict');
+const http2 = require('node:http2');
+const { before, after, bench, suite } = require('node:bench');
+const { GrpcExporter } = require('../lib/grpc');
+const { createInput } = require('./helpers/otlp');
+
+const kSuccessResponse = Buffer.from([0, 0, 0, 0, 0]);
+const kBatchesPerSample = 100;
+
+suite('OTLP export', () => {
+  let server;
+  let exporter;
+  let received = 0;
+
+  before(async () => {
+    // An in-process collector consumes the request and acknowledges it.
+    // It does not decode protobuf or depend on an external Collector.
+    server = http2.createServer();
+    server.on('stream', (stream) => {
+      stream.on('error', () => {});
+      stream.resume();
+      stream.on('end', () => {
+        received++;
+        stream.respond(
+          { ':status': 200, 'content-type': 'application/grpc+proto' },
+          { waitForTrailers: true },
+        );
+        stream.on('wantTrailers', () =>
+          stream.sendTrailers({ 'grpc-status': '0' }),
+        );
+        stream.end(kSuccessResponse);
+      });
+    });
+    await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
+    exporter = new GrpcExporter(
+      new URL(`http://127.0.0.1:${server.address().port}`),
+      10000,
+    );
+  });
+
+  after(async () => {
+    exporter?.close();
+    if (server) await new Promise((resolve) => server.close(resolve));
+  });
+
+  for (const batchSize of [1, 100]) {
+    for (const exceptions of [false, true]) {
+      const { encoder, spans } = createInput(batchSize, exceptions);
+      bench(
+        `OTLP export / batch=${batchSize} / ${exceptions ? 'exception' : 'HTTP'}`,
+        { samples: 30, warmup: 20, params: { batchSize, exceptions } },
+        async (b) => {
+          const before = received;
+          let result;
+          b.start();
+          for (let i = 0; i < kBatchesPerSample; i++) {
+            result = await exporter.send(encoder.encode(spans));
+          }
+          // One operation includes encoding and the complete local gRPC
+          // round trip, through response decoding and final status handling.
+          b.end(kBatchesPerSample);
+          assert.equal(received - before, kBatchesPerSample);
+          assert.equal(result.rejectedSpans, 0n);
+          assert.equal(result.errorMessage, '');
+        },
+      );
+    }
+  }
+});

lib/core.js

@@ -18,6 +18,7 @@ const kDefaultEndpoint = 'http://localhost:4318';
 const kDefaultServiceName = 'unknown_service:node';
 const kDefaultSdkName = 'microtel';
 
+let protocol = 'json';
 let endpoint = null;
 let active = false;
 let filter = null; // null = all modules enabled; Set = only listed modules.
@@ -173,7 +174,19 @@ function start(options = { __proto__: null }) {
 
   validateObject(options, 'options');
 
-  const endpointValue = optionalString(options, 'endpoint', kDefaultEndpoint);
+  const protocolValue = optionalString(options, 'protocol', 'json');
+  if (protocolValue !== 'json' && protocolValue !== 'grpc') {
+    throw invalidArgValueError(
+      'options.protocol',
+      protocolValue,
+      'must be json or grpc',
+    );
+  }
+  const endpointValue = optionalString(
+    options,
+    'endpoint',
+    protocolValue === 'grpc' ? 'http://localhost:4317' : kDefaultEndpoint,
+  );
 
   let parsed;
   try {
@@ -197,12 +210,15 @@ function start(options = { __proto__: null }) {
     );
   }
 
-  // /v1/traces is appended to the endpoint's path, unless the path already
-  // ends with it. The resolved URL is fixed at activation: it drives both
-  // the exports and the self-trace suppression.
-  let pathname = parsed.pathname.replace(/\/+$/, '');
-  if (!pathname.endsWith('/v1/traces')) pathname += '/v1/traces';
-  parsed.pathname = pathname;
+  // Resolve once for both the transport and self-trace suppression.
+  if (protocolValue === 'grpc') {
+    // gRPC method paths are absolute; endpoint paths are not HTTP prefixes.
+    parsed.pathname = require('./grpc').kTraceExportPath;
+  } else {
+    let pathname = parsed.pathname.replace(/\/+$/, '');
+    if (!pathname.endsWith('/v1/traces')) pathname += '/v1/traces';
+    parsed.pathname = pathname;
+  }
   parsed.search = '';
 
   if (options.maxBufferSize !== undefined)
@@ -234,6 +250,7 @@ function start(options = { __proto__: null }) {
     options.resourceAttributes,
   );
 
+  protocol = protocolValue;
   endpoint = endpointValue;
   filter = parsedFilter;
   exclude = parsedExclude;
@@ -259,6 +276,7 @@ module.exports = {
   start,
   isActive: () => active,
   getEndpoint: () => endpoint,
+  getProtocol: () => protocol,
   getSpanStorage,
   isModuleEnabled,
   getCollectorHost: () => collectorHost,

lib/flush.js

@@ -12,6 +12,7 @@ const { guard } = require('./guard');
 const {
   getEndpoint,
   getExportHeaders,
+  getProtocol,
   getExportUrl,
   getInstanceId,
   getResourceAttributes,
@@ -53,6 +54,8 @@ const inFlightExports = new Set();
 
 let cachedResource = null;
 let cachedScope = null;
+let otlpEncoder = null;
+let grpcExporter = null;
 
 // A user value replaces microtel's default for the same key, so each key
 // appears once. User keys microtel does not set are appended.
@@ -247,6 +250,43 @@ function exportBuffered() {
   const spans = spanBuffer;
   spanBuffer = [];
 
+  if (getProtocol() === 'grpc') {
+    try {
+      const { OtlpEncoder } = require('./otlp');
+      const { GrpcExporter } = require('./grpc');
+      otlpEncoder ??= new OtlpEncoder(getResource(), getScope());
+      grpcExporter ??= new GrpcExporter(
+        getExportUrl(),
+        exportTimeoutMs,
+        getExportHeaders(),
+      );
+      const body = otlpEncoder.encode(spans, (err, span) => {
+        warnExportFailure(
+          `Failed to serialize span "${span.name}": ${err.message}`,
+        );
+      });
+      if (!body) return settleAll(pending);
+      const exportDone = grpcExporter.send(body).then(
+        (result) => {
+          if (result.rejectedSpans !== 0n || result.errorMessage) {
+            warnExportFailure(
+              `OTLP collector rejected ${result.rejectedSpans} spans: ${result.errorMessage}`,
+            );
+          }
+        },
+        (err) => warnExportFailure(`Failed to export spans: ${err.message}`),
+      );
+      inFlightExports.add(exportDone);
+      exportDone.then(() => inFlightExports.delete(exportDone));
+      pending.push(exportDone);
+    } catch (err) {
+      warnExportFailure(
+        `Failed to serialize ${spans.length} spans: ${err.message}`,
+      );
+    }
+    return settleAll(pending);
+  }
+
   const otlpSpans = [];
   for (let i = 0; i < spans.length; i++) {
     try {
@@ -474,6 +514,9 @@ function resetCaches() {
   flushScheduled = false;
   flushIntervalMs = kDefaultFlushIntervalMs;
   exportTimeoutMs = kDefaultExportTimeoutMs;
+  grpcExporter?.close();
+  grpcExporter = null;
+  otlpEncoder = null;
   cachedResource = null;
   cachedScope = null;
   exportFailureCount = 0;