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

Author: pichlermarc

experimental/CHANGELOG.md

@@ -83,6 +83,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/src/common/protobuf/common-serializer.ts

@@ -0,0 +1,157 @@
+/*
+ * 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 { 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 split into 32-bit parts
+  // We use the fact that 1e9 < 2^30, so multiplication is safe for reasonable timestamps
+
+  // For the low 32 bits: (seconds * 1e9 + nanos) & 0xFFFFFFFF
+  // For the high 32 bits: floor((seconds * 1e9 + nanos) / 2^32)
+
+  // Calculate seconds * 1e9 split into parts
+  const secNanos = seconds * 1e9;
+  const secNanosLow = secNanos >>> 0; // Low 32 bits of seconds * 1e9
+  const secNanosHigh = (secNanos / 0x100000000) >>> 0; // High bits from seconds * 1e9
+
+  // Add nanoseconds to the low part
+  const totalLow = (secNanosLow + nanos) >>> 0;
+
+  // Check for overflow from low to high (carry bit)
+  const carry = secNanosLow + nanos >= 0x100000000 ? 1 : 0;
+  const totalHigh = (secNanosHigh + 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);
+    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);
+  writer.writeString(key);
+  writer.writeTag(2, 2);
+  const valueStart = writer.startLengthDelimited();
+  const startPos = writer.pos;
+  writeAnyValue(writer, value);
+  writer.finishLengthDelimited(valueStart, writer.pos - startPos);
+}
+
+/**
+ * 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);
+    writer.writeString(value as string);
+  } else if (t === 'boolean') {
+    writer.writeTag(2, 0);
+    writer.writeVarint((value as boolean) ? 1 : 0);
+  } else if (t === 'number') {
+    // Use isSafeInteger to avoid precision loss with large integers
+    // Numbers outside the safe integer range should be serialized as doubles
+    if (!Number.isSafeInteger(value as number)) {
+      writer.writeTag(4, 1);
+      writer.writeDouble(value as number);
+    } else {
+      writer.writeTag(3, 0);
+      writer.writeVarint(value as number);
+    }
+  } else if (value instanceof Uint8Array) {
+    writer.writeTag(7, 2);
+    writer.writeBytes(value);
+  } else if (Array.isArray(value)) {
+    writer.writeTag(5, 2);
+    const arrayStart = writer.startLengthDelimited();
+    const arrayStartPos = writer.pos;
+    for (const item of value) {
+      writer.writeTag(1, 2);
+      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);
+    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);
+      const kvStart = writer.startLengthDelimited();
+      const kvStartPos = writer.pos;
+      writer.writeTag(1, 2);
+      writer.writeString(k);
+      writer.writeTag(2, 2);
+      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,28 @@
+/*
+ * 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.
+ */
+
+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,91 @@
+/*
+ * 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 { IProtobufWriter } from './i-protobuf-writer';
+import { sizeAsVarint } 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 += sizeAsVarint(length);
+  }
+
+  writeVarint(value: number): void {
+    this.pos += sizeAsVarint(value);
+  }
+
+  writeFixed32(_value: number): void {
+    this.pos += 4;
+  }
+
+  writeFixed64(_low: number, _high: number): void {
+    this.pos += 8;
+  }
+
+  writeBytes(bytes: Uint8Array): void {
+    this.pos += sizeAsVarint(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 += sizeAsVarint(byteLen);
+    this.pos += byteLen;
+  }
+}