feat(otlp-transformer): add custom logs protobuf serializer

experimental/CHANGELOG.md

@@ -135,6 +135,7 @@ For notes on migrating to 2.x / 0.200.x see [the upgrade guide](doc/upgrade-to-2
 * feat(opentelemetry-sdk-node): set instrumentation and propagators for experimental start [#6148](https://github.com/open-telemetry/opentelemetry-js/pull/6148) @maryliag
 * refactor(configuration): set console exporter as empty object [#6164](https://github.com/open-telemetry/opentelemetry-js/pull/6164) @maryliag
 * feat(instrumentation-http, instrumentation-fetch, instrumentation-xml-http-request): support "QUERY" as a known HTTP method
+* feat(otlp-transformer): add custom protobuf logs serializer [#6228](https://github.com/open-telemetry/opentelemetry-js/pull/6228) @pichlermarc
 
 ### :bug: Bug Fixes
 

experimental/packages/otlp-transformer/.eslintignore

@@ -1,2 +1,3 @@
 build
 src/generated
+test/generated

experimental/packages/otlp-transformer/.gitignore

@@ -1,3 +1,5 @@
 src/generated/*
+test/generated/*
 !src/generated/.gitkeep
+!test/generated/.gitkeep
 !src/logs

experimental/packages/otlp-transformer/package.json

@@ -17,8 +17,8 @@
     "compile": "tsc --build tsconfig.json tsconfig.esm.json tsconfig.esnext.json",
     "clean": "tsc --build --clean tsconfig.json tsconfig.esm.json tsconfig.esnext.json",
     "protos": "npm run submodule && npm run protos:generate",
-    "protos:generate:js": "pbjs -t static-module -p ./protos -w commonjs --null-defaults -o ./src/generated/root.js ./protos/opentelemetry/proto/common/v1/common.proto ./protos/opentelemetry/proto/resource/v1/resource.proto ./protos/opentelemetry/proto/trace/v1/trace.proto ./protos/opentelemetry/proto/collector/trace/v1/trace_service.proto ./protos/opentelemetry/proto/metrics/v1/metrics.proto ./protos/opentelemetry/proto/collector/metrics/v1/metrics_service.proto ./protos/opentelemetry/proto/logs/v1/logs.proto ./protos/opentelemetry/proto/collector/logs/v1/logs_service.proto",
-    "protos:generate:ts": "pbts -o ./src/generated/root.d.ts ./src/generated/root.js",
+    "protos:generate:js": "pbjs -t static-module -p ./protos -w commonjs --null-defaults -o ./src/generated/root.js ./protos/opentelemetry/proto/common/v1/common.proto ./protos/opentelemetry/proto/resource/v1/resource.proto ./protos/opentelemetry/proto/trace/v1/trace.proto ./protos/opentelemetry/proto/collector/trace/v1/trace_service.proto ./protos/opentelemetry/proto/metrics/v1/metrics.proto ./protos/opentelemetry/proto/collector/metrics/v1/metrics_service.proto ./protos/opentelemetry/proto/logs/v1/logs.proto ./protos/opentelemetry/proto/collector/logs/v1/logs_service.proto && pbjs -t static-module -p ./test/fixtures -w commonjs --null-defaults -o ./test/generated/testbed.js ./test/fixtures/testbed.proto",
+    "protos:generate:ts": "pbts -o ./src/generated/root.d.ts ./src/generated/root.js && pbts -o ./test/generated/testbed.d.ts ./test/generated/testbed.js",
     "protos:generate": "npm run protos:generate:js && npm run protos:generate:ts",
     "lint": "eslint . --ext .ts",
     "lint:fix": "eslint . --ext .ts --fix",
@@ -65,6 +65,7 @@
   "devDependencies": {
     "@opentelemetry/api": "1.9.0",
     "@types/mocha": "10.0.10",
+    "@types/sinon": "^17.0.4",
     "@types/webpack-env": "1.16.3",
     "babel-plugin-istanbul": "7.0.1",
     "karma": "6.4.4",

experimental/packages/otlp-transformer/src/common/protobuf/common-serializer.ts

@@ -0,0 +1,189 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+import type { Attributes, HrTime } from '@opentelemetry/api';
+import type { AnyValue, LogAttributes } from '@opentelemetry/api-logs';
+import type { IProtobufWriter } from './i-protobuf-writer';
+
+/**
+ * Write HrTime [seconds, nanoseconds] directly as fixed64 to the serializer.
+ * Converts to nanoseconds and writes as 64-bit little-endian integer without allocations.
+ *
+ * HrTime represents: total_nanos = seconds * 1_000_000_000 + nanoseconds
+ * We need to split this into low (bits 0-31) and high (bits 32-63).
+ *
+ * @param serializer - The protobuf writer
+ * @param hrTime - HrTime tuple [seconds, nanoseconds]
+ */
+export function writeHrTimeAsFixed64(
+  serializer: IProtobufWriter,
+  hrTime: HrTime
+): void {
+  const seconds = hrTime[0];
+  const nanos = hrTime[1];
+
+  // Calculate total nanoseconds (seconds * 1_000_000_000 + nanos) split into [low32, high32].
+  //
+  // We cannot use `seconds * 1e9` directly because for Unix timestamps
+  // (seconds > ~9_007_199) the product exceeds Number.MAX_SAFE_INTEGER and loses
+  // integer precision in IEEE 754 double arithmetic.
+  //
+  // So we split `seconds` into its lower 16 bits and remaining upper bits so
+  // every multiplication stays
+  // within the safe-integer range (< 2^53):
+  //   secondsLower16Bits * 1e9 <= 65535 * 1e9 ≈ 6.55e13 < 2^53
+  //   secondsUpperBits   * 1e9 <= 65535 * 1e9 ≈ 6.55e13 < 2^53
+  //
+  // Then recombine:
+  //   seconds * 1e9 =
+  //     secondsUpperBits * 1e9 * 2^16 +
+  //     secondsLower16Bits * 1e9
+
+  const nanosPerSecond = 1_000_000_000;
+  // Split `seconds` into low 16 bits and the remaining upper bits.
+  // This avoids the signed 32-bit seconds limit behind the Year 2038 problem:
+  // the practical limit here is the encoded fixed64 range. Callers must keep
+  // `seconds * 1e9 + nanos` within uint64, i.e. up to
+  // [18_446_744_073, 709_551_615]. Beyond that, the serialized value wraps.
+  const secondsLower16Bits = seconds & 0xffff; // bits 0-15 of seconds
+  const secondsUpperBits = (seconds / 0x10000) >>> 0; // bits 16+ of seconds
+
+  const nanosFromLower16Bits = secondsLower16Bits * nanosPerSecond; // exact integer, < 2^53
+  const nanosFromUpperBits = secondsUpperBits * nanosPerSecond; // exact integer, < 2^53
+
+  // Split the lower-16-bit contribution into [low32, high32].
+  const lower16ContributionLow32 = nanosFromLower16Bits >>> 0;
+  const lower16ContributionHigh32 = Math.floor(
+    nanosFromLower16Bits / 0x100000000
+  );
+
+  // The upper-bits contribution is shifted left by 16 bits when recombined.
+  const upperBitsContributionLow32 =
+    ((nanosFromUpperBits & 0xffff) * 0x10000) >>> 0;
+  const upperBitsContributionHigh32 = (nanosFromUpperBits / 0x10000) >>> 0;
+
+  // Add the two contributions plus the sub-second nanoseconds with carry propagation.
+  const low32WithCarry =
+    lower16ContributionLow32 + upperBitsContributionLow32 + nanos;
+  const totalLow = low32WithCarry >>> 0;
+  const carry = Math.floor(low32WithCarry / 0x100000000); // 0, 1, or 2
+  const totalHigh =
+    (lower16ContributionHigh32 + upperBitsContributionHigh32 + carry) >>> 0;
+
+  serializer.writeFixed64(totalLow, totalHigh);
+}
+
+/**
+ * Write Attributes directly to protobuf as repeated KeyValue
+ */
+export function writeAttributes(
+  writer: IProtobufWriter,
+  attributes: Attributes | LogAttributes,
+  fieldNumber: number
+): void {
+  for (const key in attributes) {
+    if (!Object.prototype.hasOwnProperty.call(attributes, key)) {
+      continue;
+    }
+    const value = attributes[key];
+    writer.writeTag(fieldNumber, 2); // repeated KeyValue attributes (field varies, length-delimited)
+    const kvStart = writer.startLengthDelimited();
+    const startPos = writer.pos;
+    writeKeyValue(writer, key, value);
+    writer.finishLengthDelimited(kvStart, writer.pos - startPos);
+  }
+}
+
+/**
+ * Write a KeyValue pair directly to protobuf
+ */
+export function writeKeyValue(
+  writer: IProtobufWriter,
+  key: string,
+  value: AnyValue
+): void {
+  writer.writeTag(1, 2); // KeyValue.key (field 1, string, length-delimited)
+  writer.writeString(key);
+  writer.writeTag(2, 2); // KeyValue.value (field 2, AnyValue, length-delimited)
+  const valueStart = writer.startLengthDelimited();
+  const startPos = writer.pos;
+  writeAnyValue(writer, value);
+  writer.finishLengthDelimited(valueStart, writer.pos - startPos);
+}
+
+// int64 range bounds for int_value vs double_value encoding.
+// -(2^63) is the signed int64 minimum; 2^63 is one past the maximum.
+// Both are exact IEEE 754 doubles (powers of two), so comparisons are precise.
+const MIN_64_BIT_INT = -(2 ** 63);
+const MAX_64_BIT_INT = 2 ** 63;
+
+/**
+ * Write an AnyValue directly from raw attribute value to protobuf
+ */
+export function writeAnyValue(writer: IProtobufWriter, value: AnyValue): void {
+  const t = typeof value;
+  if (t === 'string') {
+    writer.writeTag(1, 2); // AnyValue.string_value (field 1, length-delimited)
+    writer.writeString(value as string);
+  } else if (t === 'boolean') {
+    writer.writeTag(2, 0); // AnyValue.bool_value (field 2, varint)
+    writer.writeVarint((value as boolean) ? 1 : 0);
+  } else if (t === 'number') {
+    const numValue = value as number;
+    // Encode as int_value (int64) when the number is an integer within the int64
+    // range. Number.isSafeInteger() would be too conservative — integers beyond
+    // 2^53-1 that are exactly representable in IEEE 754 (e.g. powers of two)
+    // still fit in int64 and must not be downgraded to double_value.
+    if (
+      Number.isInteger(numValue) &&
+      numValue >= MIN_64_BIT_INT &&
+      numValue < MAX_64_BIT_INT
+    ) {
+      writer.writeTag(3, 0); // AnyValue.int_value (field 3, varint)
+      writer.writeVarint(numValue);
+    } else {
+      writer.writeTag(4, 1); // AnyValue.double_value (field 4, fixed64)
+      writer.writeDouble(numValue);
+    }
+  } else if (value instanceof Uint8Array) {
+    writer.writeTag(7, 2); // AnyValue.bytes_value (field 7, length-delimited)
+    writer.writeBytes(value);
+  } else if (Array.isArray(value)) {
+    writer.writeTag(5, 2); // AnyValue.array_value (field 5, ArrayValue, length-delimited)
+    const arrayStart = writer.startLengthDelimited();
+    const arrayStartPos = writer.pos;
+    for (const item of value) {
+      writer.writeTag(1, 2); // ArrayValue.values (field 1, AnyValue, length-delimited)
+      const itemStart = writer.startLengthDelimited();
+      const itemStartPos = writer.pos;
+      writeAnyValue(writer, item);
+      writer.finishLengthDelimited(itemStart, writer.pos - itemStartPos);
+    }
+    writer.finishLengthDelimited(arrayStart, writer.pos - arrayStartPos);
+  } else if (t === 'object' && value != null) {
+    writer.writeTag(6, 2); // AnyValue.kvlist_value (field 6, KeyValueList, length-delimited)
+    const kvlistStart = writer.startLengthDelimited();
+    const kvlistStartPos = writer.pos;
+    const obj = value as Record<string, AnyValue>;
+    for (const k in obj) {
+      if (!Object.prototype.hasOwnProperty.call(obj, k)) {
+        continue;
+      }
+      const v = obj[k];
+      writer.writeTag(1, 2); // KeyValueList.values (field 1, KeyValue, length-delimited)
+      const kvStart = writer.startLengthDelimited();
+      const kvStartPos = writer.pos;
+      writer.writeTag(1, 2); // KeyValue.key (field 1, string, length-delimited)
+      writer.writeString(k);
+      writer.writeTag(2, 2); // KeyValue.value (field 2, AnyValue, length-delimited)
+      const valueStart = writer.startLengthDelimited();
+      const valueStartPos = writer.pos;
+      writeAnyValue(writer, v);
+      writer.finishLengthDelimited(valueStart, writer.pos - valueStartPos);
+      writer.finishLengthDelimited(kvStart, writer.pos - kvStartPos);
+    }
+    writer.finishLengthDelimited(kvlistStart, writer.pos - kvlistStartPos);
+  }
+  // Else: unsupported type, write nothing
+}

experimental/packages/otlp-transformer/src/common/protobuf/i-protobuf-writer.ts

@@ -0,0 +1,17 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+export interface IProtobufWriter {
+  pos: number;
+  writeTag(fieldNumber: number, wireType: number): void;
+  writeVarint(value: number): void;
+  writeFixed32(value: number): void;
+  writeFixed64(low: number, high: number): void;
+  writeBytes(bytes: Uint8Array): void;
+  writeString(str: string): void;
+  writeDouble(value: number): void;
+  startLengthDelimited(): number;
+  finishLengthDelimited(pos: number, length: number): void;
+}

experimental/packages/otlp-transformer/src/common/protobuf/protobuf-size-estimator.ts

@@ -0,0 +1,80 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import type { IProtobufWriter } from './i-protobuf-writer';
+import { estimateVarintSize } from './utils';
+
+/**
+ * Calculate UTF-8 byte length without encoding
+ * @param str valid UTF-16 string
+ */
+function utf8ByteLength(str: string): number {
+  // No quick path for ASCII, since we need to loop though all chars anyway.
+  const len = str.length;
+  let byteLen = 0;
+  for (let i = 0; i < len; i++) {
+    const code = str.charCodeAt(i);
+    if (code < 0x80) {
+      byteLen += 1;
+    } else if (code < 0x800) {
+      byteLen += 2;
+    } else if (code < 0xd800 || code >= 0xe000) {
+      byteLen += 3;
+    } else {
+      // Surrogate pair
+      i++; // Skip the next character
+      byteLen += 4;
+    }
+  }
+  return byteLen;
+}
+
+/**
+ * Size estimator for protobuf messages.
+ * Implements the same interface as ProtobufWriter but only counts bytes without allocating a buffer.
+ * @internal
+ */
+export class ProtobufSizeEstimator implements IProtobufWriter {
+  public pos: number = 0;
+
+  startLengthDelimited(): number {
+    return this.pos;
+  }
+
+  finishLengthDelimited(_: number, length: number): void {
+    this.pos += estimateVarintSize(length);
+  }
+
+  writeVarint(value: number): void {
+    this.pos += estimateVarintSize(value);
+  }
+
+  writeFixed32(_value: number): void {
+    this.pos += 4;
+  }
+
+  writeFixed64(_low: number, _high: number): void {
+    this.pos += 8;
+  }
+
+  writeBytes(bytes: Uint8Array): void {
+    this.pos += estimateVarintSize(bytes.length);
+    this.pos += bytes.length;
+  }
+
+  writeTag(fieldNumber: number, wireType: number): void {
+    this.writeVarint((fieldNumber << 3) | wireType);
+  }
+
+  writeDouble(_value: number): void {
+    this.pos += 8;
+  }
+
+  writeString(str: string): void {
+    const byteLen = utf8ByteLength(str);
+    this.pos += estimateVarintSize(byteLen);
+    this.pos += byteLen;
+  }
+}