feat(opentelemetry): add protobuf protocol support for OTLP exporters

Add support for Protocol Buffers binary encoding as an alternative to JSON
for OTLP trace, metrics, and log exports. This enables more efficient
wire format when communicating with OpenTelemetry collectors.

- Add `protocol` option ("json" | "protobuf") to Otlp.layer and individual
  OtlpTracer, OtlpMetrics, OtlpLogger layers
- Implement protobuf wire format encoding following opentelemetry-proto specs
- Set appropriate Content-Type header (application/x-protobuf vs application/json)
- No new dependencies - protobuf encoding implemented from scratch

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: davidgoli

.changeset/add-otlp-protobuf-support.md

@@ -0,0 +1,22 @@
+---
+"@effect/opentelemetry": minor
+---
+
+Add protobuf protocol support for OTLP exporters.
+
+This adds a `protocol` option to `Otlp.layer`, `OtlpTracer.layer`, `OtlpMetrics.layer`, and `OtlpLogger.layer` that allows choosing between JSON (default) and Protocol Buffers binary encoding when exporting telemetry data to OpenTelemetry collectors.
+
+```typescript
+import { Otlp } from "@effect/opentelemetry"
+
+// Use protobuf encoding for more efficient wire format
+Otlp.layer({
+  baseUrl: "http://localhost:4318",
+  protocol: "protobuf",
+  resource: { serviceName: "my-service" }
+})
+```
+
+- No new dependencies - protobuf encoding implemented from scratch
+- Sets appropriate Content-Type header (`application/x-protobuf` vs `application/json`)
+- Follows opentelemetry-proto specifications

packages/opentelemetry/src/Otlp.ts

@@ -8,10 +8,22 @@ import type * as Duration from "effect/Duration"
 import * as Layer from "effect/Layer"
 import type * as Logger from "effect/Logger"
 import type * as Tracer from "effect/Tracer"
+import type { OtlpProtocol as _OtlpProtocol } from "./internal/otlpExporter.js"
 import * as OtlpLogger from "./OtlpLogger.js"
 import * as OtlpMetrics from "./OtlpMetrics.js"
 import * as OtlpTracer from "./OtlpTracer.js"
 
+/**
+ * OTLP protocol type for encoding telemetry data.
+ *
+ * - `"json"`: JSON encoding (default) - uses `application/json` content type
+ * - `"protobuf"`: Protocol Buffers binary encoding - uses `application/x-protobuf` content type
+ *
+ * @since 1.0.0
+ * @category Models
+ */
+export type OtlpProtocol = _OtlpProtocol
+
 /**
  * @since 1.0.0
  * @category Layers
@@ -32,6 +44,7 @@ export const layer = (options: {
   readonly metricsExportInterval?: Duration.DurationInput | undefined
   readonly tracerExportInterval?: Duration.DurationInput | undefined
   readonly shutdownTimeout?: Duration.DurationInput | undefined
+  readonly protocol?: OtlpProtocol | undefined
 }): Layer.Layer<never, never, HttpClient.HttpClient> => {
   const baseReq = HttpClientRequest.get(options.baseUrl)
   const url = (path: string) => HttpClientRequest.appendUrl(baseReq, path).url
@@ -44,14 +57,16 @@ export const layer = (options: {
       exportInterval: options.loggerExportInterval,
       maxBatchSize: options.maxBatchSize,
       shutdownTimeout: options.shutdownTimeout,
-      excludeLogSpans: options.loggerExcludeLogSpans
+      excludeLogSpans: options.loggerExcludeLogSpans,
+      protocol: options.protocol
     }),
     OtlpMetrics.layer({
       url: url("/v1/metrics"),
       resource: options.resource,
       headers: options.headers,
       exportInterval: options.metricsExportInterval,
-      shutdownTimeout: options.shutdownTimeout
+      shutdownTimeout: options.shutdownTimeout,
+      protocol: options.protocol
     }),
     OtlpTracer.layer({
       url: url("/v1/traces"),
@@ -60,7 +75,8 @@ export const layer = (options: {
       exportInterval: options.tracerExportInterval,
       maxBatchSize: options.maxBatchSize,
       context: options.tracerContext,
-      shutdownTimeout: options.shutdownTimeout
+      shutdownTimeout: options.shutdownTimeout,
+      protocol: options.protocol
     })
   )
 }

packages/opentelemetry/src/OtlpLogger.ts

@@ -18,6 +18,8 @@ import * as Option from "effect/Option"
 import type * as Scope from "effect/Scope"
 import * as Tracer from "effect/Tracer"
 import * as Exporter from "./internal/otlpExporter.js"
+import type { OtlpProtocol } from "./internal/otlpExporter.js"
+import * as OtlpProtobuf from "./internal/otlpProtobuf.js"
 import type { AnyValue, Fixed64, KeyValue, Resource } from "./OtlpResource.js"
 import * as OtlpResource from "./OtlpResource.js"
 
@@ -38,6 +40,7 @@ export const make: (
     readonly maxBatchSize?: number | undefined
     readonly shutdownTimeout?: Duration.DurationInput | undefined
     readonly excludeLogSpans?: boolean | undefined
+    readonly protocol?: OtlpProtocol | undefined
   }
 ) => Effect.Effect<
   Logger.Logger<unknown, void>,
@@ -55,6 +58,7 @@ export const make: (
     headers: options.headers,
     maxBatchSize: options.maxBatchSize ?? 1000,
     exportInterval: options.exportInterval ?? Duration.seconds(1),
+    protocol: options.protocol,
     body: (data): IExportLogsServiceRequest => ({
       resourceLogs: [{
         resource: otelResource,
@@ -64,6 +68,29 @@ export const make: (
         }]
       }]
     }),
+    bodyProtobuf: (data: Array<ILogRecord>): Uint8Array =>
+      OtlpProtobuf.encodeLogsData({
+        resourceLogs: [{
+          resource: otelResource,
+          scopeLogs: [{
+            scope,
+            logRecords: data.map((record) => ({
+              timeUnixNano: String(record.timeUnixNano),
+              observedTimeUnixNano: record.observedTimeUnixNano !== undefined
+                ? String(record.observedTimeUnixNano)
+                : undefined,
+              severityNumber: record.severityNumber,
+              severityText: record.severityText,
+              body: record.body,
+              attributes: record.attributes,
+              droppedAttributesCount: record.droppedAttributesCount,
+              flags: record.flags,
+              traceId: typeof record.traceId === "string" ? record.traceId : undefined,
+              spanId: typeof record.spanId === "string" ? record.spanId : undefined
+            }))
+          }]
+        }]
+      }),
     shutdownTimeout: options.shutdownTimeout ?? Duration.seconds(3)
   })
 
@@ -92,6 +119,7 @@ export const layer = (options: {
   readonly maxBatchSize?: number | undefined
   readonly shutdownTimeout?: Duration.DurationInput | undefined
   readonly excludeLogSpans?: boolean | undefined
+  readonly protocol?: OtlpProtocol | undefined
 }): Layer.Layer<never, never, HttpClient.HttpClient> =>
   options.replaceLogger ? Logger.replaceScoped(options.replaceLogger, make(options)) : Logger.addScoped(make(options))
 

packages/opentelemetry/src/OtlpMetrics.ts

@@ -13,6 +13,8 @@ import * as MetricState from "effect/MetricState"
 import * as Option from "effect/Option"
 import type * as Scope from "effect/Scope"
 import * as Exporter from "./internal/otlpExporter.js"
+import type { OtlpProtocol } from "./internal/otlpExporter.js"
+import * as OtlpProtobuf from "./internal/otlpProtobuf.js"
 import type { Fixed64, KeyValue } from "./OtlpResource.js"
 import * as OtlpResource from "./OtlpResource.js"
 
@@ -30,6 +32,7 @@ export const make: (options: {
   readonly headers?: Headers.Input | undefined
   readonly exportInterval?: Duration.DurationInput | undefined
   readonly shutdownTimeout?: Duration.DurationInput | undefined
+  readonly protocol?: OtlpProtocol | undefined
 }) => Effect.Effect<
   void,
   never,
@@ -265,13 +268,72 @@ export const make: (options: {
     }
   }
 
+  const snapshotProtobuf = (): Uint8Array => {
+    const data = snapshot()
+    return OtlpProtobuf.encodeMetricsData({
+      resourceMetrics: data.resourceMetrics.map((rm) => ({
+        resource: rm.resource!,
+        scopeMetrics: rm.scopeMetrics.map((sm) => ({
+          scope: sm.scope!,
+          metrics: sm.metrics.map((m) => ({
+            name: m.name,
+            description: m.description,
+            unit: m.unit,
+            gauge: m.gauge
+              ? {
+                dataPoints: m.gauge.dataPoints.map((dp) => ({
+                  attributes: dp.attributes,
+                  startTimeUnixNano: String(dp.startTimeUnixNano ?? "0"),
+                  timeUnixNano: String(dp.timeUnixNano ?? "0"),
+                  asDouble: dp.asDouble !== null ? dp.asDouble : undefined,
+                  asInt: dp.asInt !== undefined ? String(dp.asInt) : undefined
+                }))
+              }
+              : undefined,
+            sum: m.sum
+              ? {
+                dataPoints: m.sum.dataPoints.map((dp) => ({
+                  attributes: dp.attributes,
+                  startTimeUnixNano: String(dp.startTimeUnixNano ?? "0"),
+                  timeUnixNano: String(dp.timeUnixNano ?? "0"),
+                  asDouble: dp.asDouble !== null ? dp.asDouble : undefined,
+                  asInt: dp.asInt !== undefined ? String(dp.asInt) : undefined
+                })),
+                aggregationTemporality: m.sum.aggregationTemporality,
+                isMonotonic: m.sum.isMonotonic
+              }
+              : undefined,
+            histogram: m.histogram
+              ? {
+                dataPoints: m.histogram.dataPoints.map((dp) => ({
+                  attributes: dp.attributes ?? [],
+                  startTimeUnixNano: String(dp.startTimeUnixNano ?? "0"),
+                  timeUnixNano: String(dp.timeUnixNano ?? "0"),
+                  count: dp.count ?? 0,
+                  sum: dp.sum,
+                  bucketCounts: dp.bucketCounts ?? [],
+                  explicitBounds: dp.explicitBounds ?? [],
+                  min: dp.min,
+                  max: dp.max
+                })),
+                aggregationTemporality: m.histogram.aggregationTemporality ?? OtlpProtobuf.AggregationTemporality.Cumulative
+              }
+              : undefined
+          }))
+        }))
+      }))
+    })
+  }
+
   yield* Exporter.make({
     label: "OtlpMetrics",
     url: options.url,
     headers: options.headers,
     maxBatchSize: "disabled",
     exportInterval: options.exportInterval ?? Duration.seconds(10),
+    protocol: options.protocol,
     body: snapshot,
+    bodyProtobuf: snapshotProtobuf,
     shutdownTimeout: options.shutdownTimeout ?? Duration.seconds(3)
   })
 })
@@ -290,6 +352,7 @@ export const layer = (options: {
   readonly headers?: Headers.Input | undefined
   readonly exportInterval?: Duration.DurationInput | undefined
   readonly shutdownTimeout?: Duration.DurationInput | undefined
+  readonly protocol?: OtlpProtocol | undefined
 }): Layer.Layer<never, never, HttpClient.HttpClient> => Layer.scopedDiscard(make(options))
 
 // internal

packages/opentelemetry/src/OtlpTracer.ts

@@ -14,6 +14,8 @@ import type * as Scope from "effect/Scope"
 import * as Tracer from "effect/Tracer"
 import type { ExtractTag } from "effect/Types"
 import * as Exporter from "./internal/otlpExporter.js"
+import type { OtlpProtocol } from "./internal/otlpExporter.js"
+import * as OtlpProtobuf from "./internal/otlpProtobuf.js"
 import type { KeyValue, Resource } from "./OtlpResource.js"
 import { entriesToAttributes } from "./OtlpResource.js"
 import * as OtlpResource from "./OtlpResource.js"
@@ -39,6 +41,7 @@ export const make: (
     readonly maxBatchSize?: number | undefined
     readonly context?: (<X>(f: () => X, span: Tracer.AnySpan) => X) | undefined
     readonly shutdownTimeout?: Duration.DurationInput | undefined
+    readonly protocol?: OtlpProtocol | undefined
   }
 ) => Effect.Effect<
   Tracer.Tracer,
@@ -56,6 +59,7 @@ export const make: (
     headers: options.headers,
     exportInterval: options.exportInterval ?? Duration.seconds(5),
     maxBatchSize: options.maxBatchSize ?? 1000,
+    protocol: options.protocol,
     body(spans) {
       const data: TraceData = {
         resourceSpans: [{
@@ -68,6 +72,17 @@ export const make: (
       }
       return data
     },
+    bodyProtobuf(spans) {
+      return OtlpProtobuf.encodeTracesData({
+        resourceSpans: [{
+          resource: otelResource,
+          scopeSpans: [{
+            scope,
+            spans
+          }]
+        }]
+      })
+    },
     shutdownTimeout: options.shutdownTimeout ?? Duration.seconds(3)
   })
 
@@ -119,6 +134,7 @@ export const layer = (options: {
   readonly maxBatchSize?: number | undefined
   readonly context?: (<X>(f: () => X, span: Tracer.AnySpan) => X) | undefined
   readonly shutdownTimeout?: Duration.DurationInput | undefined
+  readonly protocol?: OtlpProtocol | undefined
 }): Layer.Layer<never, never, HttpClient.HttpClient> => Layer.unwrapScoped(Effect.map(make(options), Layer.setTracer))
 
 // internal

packages/opentelemetry/src/internal/otlpExporter.ts

@@ -1,4 +1,5 @@
 import * as Headers from "@effect/platform/Headers"
+import * as HttpBody from "@effect/platform/HttpBody"
 import * as HttpClient from "@effect/platform/HttpClient"
 import * as HttpClientError from "@effect/platform/HttpClientError"
 import * as HttpClientRequest from "@effect/platform/HttpClientRequest"
@@ -10,6 +11,12 @@ import * as Option from "effect/Option"
 import * as Schedule from "effect/Schedule"
 import * as Scope from "effect/Scope"
 
+/**
+ * OTLP protocol type for encoding
+ * @internal
+ */
+export type OtlpProtocol = "json" | "protobuf"
+
 const policy = Schedule.forever.pipe(
   Schedule.passthrough,
   Schedule.addDelay((error) => {
@@ -37,6 +44,8 @@ export const make: (
     readonly exportInterval: Duration.DurationInput
     readonly maxBatchSize: number | "disabled"
     readonly body: (data: Array<any>) => unknown
+    readonly bodyProtobuf?: ((data: Array<any>) => Uint8Array) | undefined
+    readonly protocol?: OtlpProtocol | undefined
     readonly shutdownTimeout: Duration.DurationInput
   }
 ) => Effect.Effect<
@@ -53,8 +62,14 @@ export const make: (
     HttpClient.retryTransient({ schedule: policy, times: 3 })
   )
 
+  const protocol = options.protocol ?? "json"
+  const contentType = protocol === "protobuf"
+    ? "application/x-protobuf"
+    : "application/json"
+
   let headers = Headers.unsafeFromRecord({
-    "user-agent": `effect-opentelemetry-${options.label}/0.0.0`
+    "user-agent": `effect-opentelemetry-${options.label}/0.0.0`,
+    "content-type": contentType
   })
   if (options.headers) {
     headers = Headers.merge(Headers.fromInput(options.headers), headers)
@@ -75,9 +90,14 @@ export const make: (
       }
       buffer = []
     }
-    return client.execute(
-      HttpClientRequest.bodyUnsafeJson(request, options.body(items))
-    ).pipe(
+    const requestWithBody = protocol === "protobuf" && options.bodyProtobuf
+      ? HttpClientRequest.setBody(
+        request,
+        HttpBody.uint8Array(options.bodyProtobuf(items), contentType)
+      )
+      : HttpClientRequest.bodyUnsafeJson(request, options.body(items))
+
+    return client.execute(requestWithBody).pipe(
       Effect.asVoid,
       Effect.withTracerEnabled(false)
     )