Merge pull request #39 from DEVtheOPS/feat/dynamic-otlp-headers

Author: dialupdisaster

README.md

@@ -15,6 +15,7 @@ An [opencode](https://opencode.ai) plugin that exports telemetry via OpenTelemet
 - [Configuration](#configuration)
   - [Quick start](#quick-start)
   - [Headers and resource attributes](#headers-and-resource-attributes)
+  - [Dynamic headers](#dynamic-headers)
   - [Disabling specific metrics](#disabling-specific-metrics)
   - [Datadog example](#datadog-example)
   - [Honeycomb example](#honeycomb-example)
@@ -89,6 +90,7 @@ All configuration is via environment variables. Set them in your shell profile (
 | `OPENCODE_METRIC_PREFIX` | `opencode.` | Prefix for all metric names (e.g. set to `claude_code.` for Claude Code dashboard compatibility) |
 | `OPENCODE_DISABLE_METRICS` | _(unset)_ | Comma-separated list of metric name suffixes to disable (e.g. `cache.count,session.duration`) |
 | `OPENCODE_OTLP_HEADERS` | _(unset)_ | Comma-separated `key=value` headers added to all OTLP exports. **Keep out of version control — may contain sensitive auth tokens.** |
+| `OPENCODE_OTLP_HEADERS_HELPER` | _(unset)_ | Executable script/binary that returns dynamic OTLP headers as JSON after an auth failure. Helper headers override `OPENCODE_OTLP_HEADERS`. |
 | `OPENCODE_RESOURCE_ATTRIBUTES` | _(unset)_ | Comma-separated `key=value` pairs merged into the OTel resource. Example: `service.version=1.2.3,deployment.environment=production` |
 | `OPENCODE_OTLP_METRICS_TEMPORALITY` | _(unset)_ | Metrics aggregation temporality: `delta`, `cumulative`, or `lowmemory`. Required for Datadog (`delta`). Copied to `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE`. |
 
@@ -115,6 +117,25 @@ export OPENCODE_RESOURCE_ATTRIBUTES="service.version=1.2.3,deployment.environmen
 
 > **Security note:** `OPENCODE_OTLP_HEADERS` typically contains auth tokens. Set it in your shell profile (`~/.zshrc`, `~/.bashrc`) or a secrets manager — never commit it to version control or print it in CI logs.
 
+### Dynamic headers
+
+Use `OPENCODE_OTLP_HEADERS_HELPER` when your collector requires short-lived authentication tokens. The helper is run only after an OTLP export fails with an authentication error (`401`/`403` for HTTP or `UNAUTHENTICATED`/`PERMISSION_DENIED` for gRPC). The plugin refreshes headers, rebuilds the exporter, and retries the failed export once.
+
+```bash
+export OPENCODE_OTLP_HEADERS_HELPER=/path/to/opencode-otel-headers.sh
+```
+
+The helper must be executable and print a JSON object to stdout:
+
+```bash
+#!/bin/sh
+printf '{"Authorization":"Bearer %s"}' "$(get-token.sh)"
+```
+
+For a Cloud Run collector using IAM authentication, `get-token.sh` might be `gcloud auth print-identity-token`.
+
+If `OPENCODE_OTLP_HEADERS` is also set, helper-provided headers override static headers with the same name. Header values are never logged.
+
 ### Disabling specific metrics
 
 Use `OPENCODE_DISABLE_METRICS` to suppress individual metrics. The value is a comma-separated list of metric name suffixes (without the prefix).

src/config.ts

@@ -14,6 +14,7 @@ export type PluginConfig = {
   logsInterval: number
   metricPrefix: string
   otlpHeaders: string | undefined
+  otlpHeadersHelper: string | undefined
   resourceAttributes: string | undefined
   metricsTemporality: MetricsTemporality | undefined
   disabledMetrics: Set<string>
@@ -38,6 +39,7 @@ export function parseEnvInt(key: string, fallback: number): number {
  */
 export function loadConfig(): PluginConfig {
   const otlpHeaders = process.env["OPENCODE_OTLP_HEADERS"]
+  const otlpHeadersHelper = process.env["OPENCODE_OTLP_HEADERS_HELPER"]
   const resourceAttributes = process.env["OPENCODE_RESOURCE_ATTRIBUTES"]
   const rawTemporality = process.env["OPENCODE_OTLP_METRICS_TEMPORALITY"]
   const protocol = process.env["OPENCODE_OTLP_PROTOCOL"]
@@ -81,6 +83,7 @@ export function loadConfig(): PluginConfig {
     logsInterval: parseEnvInt("OPENCODE_OTLP_LOGS_INTERVAL", 5000),
     metricPrefix: process.env["OPENCODE_METRIC_PREFIX"] ?? "opencode.",
     otlpHeaders,
+    otlpHeadersHelper,
     resourceAttributes,
     metricsTemporality,
     disabledMetrics,

src/headers.ts

@@ -0,0 +1,289 @@
+import { createRequire } from "module"
+import { ExportResultCode, type ExportResult } from "@opentelemetry/core"
+import type { PushMetricExporter, ResourceMetrics } from "@opentelemetry/sdk-metrics"
+import type { SpanExporter, ReadableSpan } from "@opentelemetry/sdk-trace-base"
+import type { LogRecordExporter, ReadableLogRecord } from "@opentelemetry/sdk-logs"
+import type { Metadata } from "@grpc/grpc-js"
+
+const require = createRequire(import.meta.url)
+const DEFAULT_HELPER_TIMEOUT_MS = 5000
+
+type Exporter<T> = {
+  export(items: T, resultCallback: (result: ExportResult) => void): void
+  shutdown(): Promise<void>
+  forceFlush?(): Promise<void>
+}
+
+type SelectAggregation = NonNullable<PushMetricExporter["selectAggregation"]>
+type SelectAggregationTemporality = NonNullable<PushMetricExporter["selectAggregationTemporality"]>
+
+export type HeadersMap = Record<string, string>
+
+export function parseOtlpHeaders(raw: string | undefined): HeadersMap {
+  if (!raw) return {}
+  const headers: HeadersMap = {}
+  for (const pair of raw.split(",")) {
+    const idx = pair.indexOf("=")
+    if (idx > 0) {
+      const key = pair.slice(0, idx).trim()
+      const val = pair.slice(idx + 1).trim()
+      if (key) headers[key] = val
+    }
+  }
+  return headers
+}
+
+export function createGrpcMetadata(headers: HeadersMap): Metadata {
+  const { Metadata } = require("@grpc/grpc-js") as typeof import("@grpc/grpc-js")
+  const metadata = new Metadata()
+  for (const [key, value] of Object.entries(headers)) metadata.set(key, value)
+  return metadata
+}
+
+export function isAuthFailure(error: unknown): boolean {
+  if (!error) return false
+  const err = error as Record<string, unknown>
+  const numericStatus = Number(err["status"] ?? err["statusCode"] ?? err["code"])
+  if ([401, 403, 7, 16].includes(numericStatus)) return true
+  const message = error instanceof Error ? error.message : String(error)
+  return /\b(401|403)\b|unauthenticated|permission denied|unauthorized|forbidden/i.test(message)
+}
+
+export class DynamicHeaders {
+  private headers: HeadersMap
+  private version = 0
+  private refreshPromise: Promise<number> | undefined
+
+  constructor(
+    private readonly staticHeaders: HeadersMap,
+    private readonly helper: string | undefined,
+    private readonly helperTimeoutMs = DEFAULT_HELPER_TIMEOUT_MS,
+  ) {
+    this.headers = { ...staticHeaders }
+  }
+
+  current(): HeadersMap {
+    return { ...this.headers }
+  }
+
+  currentVersion(): number {
+    return this.version
+  }
+
+  refresh(): Promise<number> {
+    if (!this.helper) return Promise.resolve(this.version)
+    if (!this.refreshPromise) {
+      this.refreshPromise = this.runHelper()
+        .then((helperHeaders) => {
+          const next = { ...this.staticHeaders, ...helperHeaders }
+          const changed = headerSignature(next) !== headerSignature(this.headers)
+          this.headers = next
+          if (changed) this.version += 1
+          return this.version
+        })
+        .finally(() => {
+          this.refreshPromise = undefined
+        })
+    }
+    return this.refreshPromise
+  }
+
+  private async runHelper(): Promise<HeadersMap> {
+    const proc = Bun.spawn([this.helper!], {
+      stdout: "pipe",
+      stderr: "pipe",
+      timeout: this.helperTimeoutMs,
+      killSignal: "SIGTERM",
+    })
+    const [stdout, stderr, exitCode] = await Promise.all([
+      new Response(proc.stdout).text(),
+      new Response(proc.stderr).text(),
+      proc.exited,
+    ])
+    if (proc.signalCode) {
+      throw new Error(`OTLP headers helper was terminated by ${proc.signalCode}`)
+    }
+    if (exitCode !== 0) {
+      const detail = stderr.trim() || `exit code ${exitCode}`
+      throw new Error(`OTLP headers helper failed: ${detail}`)
+    }
+    const parsed = JSON.parse(stdout) as unknown
+    if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
+      throw new Error("OTLP headers helper must return a JSON object")
+    }
+    const headers: HeadersMap = {}
+    for (const [key, value] of Object.entries(parsed)) {
+      if (typeof value !== "string") throw new Error(`OTLP headers helper returned non-string value for ${key}`)
+      headers[key] = value
+    }
+    return headers
+  }
+}
+
+export class RefreshingMetricExporter implements PushMetricExporter {
+  private exporter: PushMetricExporter
+  private headersVersion: number
+
+  constructor(
+    private readonly createExporter: (headers: HeadersMap) => PushMetricExporter,
+    private readonly dynamicHeaders: DynamicHeaders,
+  ) {
+    this.exporter = createExporter(dynamicHeaders.current())
+    this.headersVersion = dynamicHeaders.currentVersion()
+  }
+
+  export(metrics: ResourceMetrics, resultCallback: (result: ExportResult) => void): void {
+    exportWithAuthRetry(this, metrics, resultCallback)
+  }
+
+  forceFlush(): Promise<void> {
+    return this.exporter.forceFlush()
+  }
+
+  shutdown(): Promise<void> {
+    return this.exporter.shutdown()
+  }
+
+  selectAggregationTemporality(
+    instrumentType: Parameters<SelectAggregationTemporality>[0],
+  ): ReturnType<SelectAggregationTemporality> {
+    return this.exporter.selectAggregationTemporality!(instrumentType)
+  }
+
+  selectAggregation(instrumentType: Parameters<SelectAggregation>[0]): ReturnType<SelectAggregation> {
+    return this.exporter.selectAggregation!(instrumentType)
+  }
+
+  _exporter(): PushMetricExporter {
+    return this.exporter
+  }
+
+  _replaceExporter(version: number): void {
+    const old = this.exporter
+    this.exporter = this.createExporter(this.dynamicHeaders.current())
+    this.headersVersion = version
+    old.shutdown().catch(() => {})
+  }
+
+  _refreshHeaders(): Promise<number> {
+    return this.dynamicHeaders.refresh()
+  }
+
+  _headersVersion(): number {
+    return this.headersVersion
+  }
+}
+
+export class RefreshingSpanExporter implements SpanExporter {
+  private exporter: SpanExporter
+  private headersVersion: number
+
+  constructor(
+    private readonly createExporter: (headers: HeadersMap) => SpanExporter,
+    private readonly dynamicHeaders: DynamicHeaders,
+  ) {
+    this.exporter = createExporter(dynamicHeaders.current())
+    this.headersVersion = dynamicHeaders.currentVersion()
+  }
+
+  export(spans: ReadableSpan[], resultCallback: (result: ExportResult) => void): void {
+    exportWithAuthRetry(this, spans, resultCallback)
+  }
+
+  forceFlush(): Promise<void> {
+    return this.exporter.forceFlush?.() ?? Promise.resolve()
+  }
+
+  shutdown(): Promise<void> {
+    return this.exporter.shutdown()
+  }
+
+  _exporter(): SpanExporter {
+    return this.exporter
+  }
+
+  _replaceExporter(version: number): void {
+    const old = this.exporter
+    this.exporter = this.createExporter(this.dynamicHeaders.current())
+    this.headersVersion = version
+    old.shutdown().catch(() => {})
+  }
+
+  _refreshHeaders(): Promise<number> {
+    return this.dynamicHeaders.refresh()
+  }
+
+  _headersVersion(): number {
+    return this.headersVersion
+  }
+}
+
+export class RefreshingLogExporter implements LogRecordExporter {
+  private exporter: LogRecordExporter
+  private headersVersion: number
+
+  constructor(
+    private readonly createExporter: (headers: HeadersMap) => LogRecordExporter,
+    private readonly dynamicHeaders: DynamicHeaders,
+  ) {
+    this.exporter = createExporter(dynamicHeaders.current())
+    this.headersVersion = dynamicHeaders.currentVersion()
+  }
+
+  export(logs: ReadableLogRecord[], resultCallback: (result: ExportResult) => void): void {
+    exportWithAuthRetry(this, logs, resultCallback)
+  }
+
+  shutdown(): Promise<void> {
+    return this.exporter.shutdown()
+  }
+
+  _exporter(): LogRecordExporter {
+    return this.exporter
+  }
+
+  _replaceExporter(version: number): void {
+    const old = this.exporter
+    this.exporter = this.createExporter(this.dynamicHeaders.current())
+    this.headersVersion = version
+    old.shutdown().catch(() => {})
+  }
+
+  _refreshHeaders(): Promise<number> {
+    return this.dynamicHeaders.refresh()
+  }
+
+  _headersVersion(): number {
+    return this.headersVersion
+  }
+}
+
+function exportWithAuthRetry<T>(
+  wrapper: {
+    _exporter(): Exporter<T>
+    _replaceExporter(version: number): void
+    _refreshHeaders(): Promise<number>
+    _headersVersion(): number
+  },
+  items: T,
+  resultCallback: (result: ExportResult) => void,
+): void {
+  wrapper._exporter().export(items, (result) => {
+    if (result.code !== ExportResultCode.FAILED || !isAuthFailure(result.error)) {
+      resultCallback(result)
+      return
+    }
+    wrapper._refreshHeaders()
+      .then((version) => {
+        if (version !== wrapper._headersVersion()) wrapper._replaceExporter(version)
+        wrapper._exporter().export(items, resultCallback)
+      })
+      .catch((error: unknown) => {
+        resultCallback({ code: ExportResultCode.FAILED, error: error instanceof Error ? error : new Error(String(error)) })
+      })
+  })
+}
+
+function headerSignature(headers: HeadersMap): string {
+  return JSON.stringify(Object.entries(headers).sort(([a], [b]) => a.localeCompare(b)))
+}

src/index.ts

@@ -53,10 +53,12 @@ export const OtelPlugin: Plugin = async ({ project, client }) => {
     metricsInterval: config.metricsInterval,
     logsInterval: config.logsInterval,
     metricPrefix: config.metricPrefix,
+    headersHelperSet: !!config.otlpHeadersHelper,
   })
 
   await log("debug", "config loaded", {
     headersSet: !!config.otlpHeaders,
+    headersHelperSet: !!config.otlpHeadersHelper,
     resourceAttributesSet: !!config.resourceAttributes,
   })
 
@@ -76,6 +78,8 @@ export const OtelPlugin: Plugin = async ({ project, client }) => {
     config.metricsInterval,
     config.logsInterval,
     PLUGIN_VERSION,
+    config.otlpHeaders,
+    config.otlpHeadersHelper,
   )
   await log("info", "OTel SDK initialized")