Refactor api-proxy OpenTelemetry into serialization/exporter modules

containers/api-proxy/otel-exporters.js

@@ -0,0 +1,178 @@
+'use strict';
+
+const fs = require('fs');
+const http = require('http');
+const https = require('https');
+
+const { HttpsProxyAgent } = require('https-proxy-agent');
+
+const { buildResourceSpans } = require('./otel-serialization');
+
+const EXPORT_TIMEOUT_MS = 10_000;
+const AWF_VERSION = process.env.AWF_VERSION || '0.0.0-dev';
+const OTEL_SPAN_SCHEMA = `otel-span/v${AWF_VERSION}`;
+
+/**
+ * A minimal SpanExporter that POSTs OTLP/JSON to an HTTP(S) collector,
+ * routing traffic through the Squid proxy when HTTPS_PROXY is set.
+ *
+ * This replaces the standard `@opentelemetry/exporter-trace-otlp-http` solely
+ * to gain `HttpsProxyAgent` support — the api-proxy container's iptables rules
+ * require all external traffic to exit via Squid.
+ */
+class ProxyAwareOtlpExporter {
+  /**
+   * @param {object} opts
+   * @param {string} opts.url        - OTLP base URL (with or without /v1/traces suffix)
+   * @param {Record<string,string>} opts.headers - Extra request headers (auth etc.)
+   * @param {string|null} opts.httpsProxy - Squid proxy URL, or falsy to connect directly
+   * @param {import('@opentelemetry/resources').Resource} opts.resource
+   */
+  constructor({ url, headers, httpsProxy, resource }) {
+    const parsed = new URL(url);
+    const trimmedPath = parsed.pathname.replace(/\/+$/, '');
+    if (trimmedPath === '' || trimmedPath === '/') {
+      parsed.pathname = '/v1/traces';
+    } else if (trimmedPath.endsWith('/v1/traces')) {
+      parsed.pathname = trimmedPath;
+    } else {
+      parsed.pathname = `${trimmedPath}/v1/traces`;
+    }
+    this._parsedUrl = parsed;
+    this._headers = headers || {};
+    this._agent = httpsProxy ? new HttpsProxyAgent(httpsProxy) : undefined;
+    this._resource = resource;
+  }
+
+  /**
+   * @param {import('@opentelemetry/sdk-trace-base').ReadableSpan[]} spans
+   * @param {(result: import('@opentelemetry/core').ExportResult) => void} resultCallback
+   */
+  export(spans, resultCallback) {
+    if (!spans || spans.length === 0) { resultCallback({ code: 0 }); return; }
+    let settled = false;
+    const settle = (result) => {
+      if (settled) return;
+      settled = true;
+      resultCallback(result);
+    };
+
+    let bodyBuf;
+    try {
+      bodyBuf = Buffer.from(JSON.stringify({
+        resourceSpans: buildResourceSpans(spans, this._resource),
+      }), 'utf8');
+    } catch (err) {
+      settle({ code: 1, error: err });
+      return;
+    }
+
+    const isHttps = this._parsedUrl.protocol === 'https:';
+    const Transport = isHttps ? https : http;
+    const port = this._parsedUrl.port
+      ? parseInt(this._parsedUrl.port, 10)
+      : (isHttps ? 443 : 80);
+
+    const reqOptions = {
+      hostname: this._parsedUrl.hostname,
+      port,
+      path: `${this._parsedUrl.pathname}${this._parsedUrl.search || ''}`,
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        'Content-Length': bodyBuf.length,
+        ...this._headers,
+      },
+    };
+    if (this._agent) reqOptions.agent = this._agent;
+
+    let req;
+    try {
+      req = Transport.request(reqOptions, (res) => {
+        res.on('data', () => {});
+        res.on('error', (err) => { settle({ code: 1, error: err }); });
+        res.on('end', () => {
+          const ok = res.statusCode >= 200 && res.statusCode < 300;
+          settle({ code: ok ? 0 : 1 });
+        });
+      });
+    } catch (err) {
+      settle({ code: 1, error: err });
+      return;
+    }
+    req.setTimeout(EXPORT_TIMEOUT_MS, () => {
+      req.destroy(new Error(`OTLP export timeout after ${EXPORT_TIMEOUT_MS}ms`));
+    });
+    req.on('error', (err) => { settle({ code: 1, error: err }); });
+    req.write(bodyBuf);
+    req.end();
+  }
+
+  shutdown() { return Promise.resolve(); }
+}
+
+/**
+ * Writes span data as NDJSON to /var/log/api-proxy/otel.jsonl.
+ * Used when no OTLP endpoint is configured.  Mirrors the MCPG pattern at
+ * /tmp/gh-aw/otel.jsonl.  All writes are best-effort — errors are silently
+ * swallowed so a missing or unwritable log directory never breaks the proxy.
+ */
+class FileSpanExporter {
+  constructor(filePath) {
+    this._filePath = filePath;
+    this._stream = null;
+  }
+
+  _getStream() {
+    if (this._stream) return this._stream;
+    try {
+      this._stream = fs.createWriteStream(this._filePath, { flags: 'a' });
+      this._stream.on('error', () => { this._stream = null; });
+    } catch { return null; }
+    return this._stream;
+  }
+
+  export(spans, resultCallback) {
+    const stream = this._getStream();
+    if (stream) {
+      for (const span of spans || []) {
+        try {
+          const ctx = span.spanContext();
+          const record = {
+            _schema: OTEL_SPAN_SCHEMA,
+            timestamp: new Date().toISOString(),
+            event: 'otel_span',
+            traceId: ctx.traceId,
+            spanId: ctx.spanId,
+            parentSpanId: span.parentSpanId || null,
+            name: span.name,
+            kind: span.kind,
+            startTimeMs: span.startTime[0] * 1000 + Math.round(span.startTime[1] / 1e6),
+            endTimeMs: span.endTime[0] * 1000 + Math.round(span.endTime[1] / 1e6),
+            attributes: span.attributes || {},
+            events: (span.events || []).map(e => ({ name: e.name, attributes: e.attributes || {} })),
+            status: span.status,
+          };
+          stream.write(`${JSON.stringify(record)}\n`);
+        } catch { /* best-effort */ }
+      }
+    }
+    resultCallback({ code: 0 });
+  }
+
+  shutdown() {
+    return new Promise((resolve) => {
+      if (this._stream) {
+        this._stream.end(resolve);
+        this._stream = null;
+      } else {
+        resolve();
+      }
+    });
+  }
+}
+
+module.exports = {
+  ProxyAwareOtlpExporter,
+  FileSpanExporter,
+};

containers/api-proxy/otel-serialization.js

@@ -0,0 +1,116 @@
+'use strict';
+
+const SCOPE_NAME = 'awf-api-proxy';
+
+/**
+ * Parse the OTEL_EXPORTER_OTLP_HEADERS format: comma-separated "key=value" pairs.
+ * @param {string} raw
+ * @returns {Record<string, string>}
+ */
+function parseOtlpHeaders(raw) {
+  const headers = {};
+  if (!raw) return headers;
+  for (const part of raw.split(',')) {
+    const idx = part.indexOf('=');
+    if (idx < 0) continue;
+    const key = part.slice(0, idx).trim();
+    const value = part.slice(idx + 1).trim();
+    if (key) headers[key] = value;
+  }
+  return headers;
+}
+
+/** @param {[number, number]} hrTime  HrTime as [seconds, nanoseconds] */
+function hrTimeToNanoString(hrTime) {
+  return String(BigInt(hrTime[0]) * 1_000_000_000n + BigInt(hrTime[1]));
+}
+
+function serializeAttrValue(val) {
+  if (typeof val === 'string') return { stringValue: val };
+  if (typeof val === 'boolean') return { boolValue: val };
+  if (typeof val === 'number') {
+    return Number.isInteger(val) ? { intValue: String(val) } : { doubleValue: val };
+  }
+  if (Array.isArray(val)) {
+    return { arrayValue: { values: val.map(serializeAttrValue) } };
+  }
+  return { stringValue: String(val) };
+}
+
+function serializeAttributes(attrs) {
+  return Object.entries(attrs || {}).map(([key, val]) => ({
+    key,
+    value: serializeAttrValue(val),
+  }));
+}
+
+function serializeEvent(event) {
+  return {
+    name: event.name,
+    timeUnixNano: hrTimeToNanoString(event.time),
+    attributes: serializeAttributes(event.attributes || {}),
+    droppedAttributesCount: event.droppedAttributesCount || 0,
+  };
+}
+
+// OTLP proto SpanKind is 1-indexed relative to the OTEL SDK's 0-indexed enum.
+function toOtlpKind(kind) { return kind + 1; }
+
+// OTLP StatusCode matches SpanStatusCode: 0=Unset 1=Ok 2=Error.
+function serializeStatus(status) {
+  const obj = { code: status.code || 0 };
+  if (status.message) obj.message = status.message;
+  return obj;
+}
+
+function serializeSpan(span) {
+  const ctx = span.spanContext();
+  const out = {
+    traceId: ctx.traceId,
+    spanId: ctx.spanId,
+    name: span.name,
+    kind: toOtlpKind(span.kind),
+    startTimeUnixNano: hrTimeToNanoString(span.startTime),
+    endTimeUnixNano: hrTimeToNanoString(span.endTime),
+    attributes: serializeAttributes(span.attributes),
+    events: (span.events || []).map(serializeEvent),
+    links: [],
+    status: serializeStatus(span.status),
+    droppedAttributesCount: span._droppedAttributesCount || 0,
+    droppedEventsCount: span._droppedEventsCount || 0,
+    droppedLinksCount: span._droppedLinksCount || 0,
+  };
+  if (span.parentSpanId) out.parentSpanId = span.parentSpanId;
+  return out;
+}
+
+/**
+ * Build the OTLP/JSON `resourceSpans` envelope.
+ * @param {import('@opentelemetry/sdk-trace-base').ReadableSpan[]} spans
+ * @param {import('@opentelemetry/resources').Resource} resource
+ * @returns {object[]}
+ */
+function buildResourceSpans(spans, resource) {
+  return [{
+    resource: {
+      attributes: serializeAttributes(resource.attributes),
+      droppedAttributesCount: 0,
+    },
+    scopeSpans: [{
+      scope: { name: SCOPE_NAME, version: '' },
+      spans: spans.map(serializeSpan),
+    }],
+  }];
+}
+
+module.exports = {
+  parseOtlpHeaders,
+  hrTimeToNanoString,
+  serializeAttrValue,
+  serializeAttributes,
+  serializeEvent,
+  toOtlpKind,
+  serializeStatus,
+  serializeSpan,
+  buildResourceSpans,
+};