Use custom trace extractor for synapse

src/tracer.ts

@@ -3,8 +3,9 @@ import { NodeSDK } from "@opentelemetry/sdk-node";
 import { AlwaysOnSampler, Sampler, SamplingDecision } from '@opentelemetry/sdk-trace-base';
 import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
 import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
-import { DiagConsoleLogger, DiagLogLevel, Attributes, SpanKind, diag } from '@opentelemetry/api';
+import { DiagConsoleLogger, DiagLogLevel, Attributes, SpanKind, diag, TextMapPropagator, Context, TextMapGetter, TextMapSetter } from '@opentelemetry/api';
 import { SemanticAttributes } from '@opentelemetry/semantic-conventions';
+import * as api from '@opentelemetry/api';
 
 export enum DRAUPNIR_SYSTEM_TYPES {
     APPSERVICE = "appservice",
@@ -21,6 +22,100 @@ export enum DRAUPNIR_TRACING_ATTRIBUTES {
     PROVISION_OUTCOME = "draupnir.provision.outcome"
 }
 
+// Value is expected to be of form: `{trace_id}:{span_id}:{parent_id}:{flags}`
+const SYNAPSE_TRACE_HEADER = "uber-trace-id";
+
+const SYNAPSE_BAGGAGE_HEADER_PREFIX = "uberctx-";
+
+const FIELDS = [SYNAPSE_TRACE_HEADER];
+
+function readHeader(
+    carrier: unknown,
+    getter: TextMapGetter,
+    key: string
+): string {
+    let header = getter.get(carrier, key);
+    if (Array.isArray(header)) [header] = header;
+    return header || '';
+}
+
+const VALID_HEADER_NAME_CHARS = /^[\^_`a-zA-Z\-0-9!#$%&'*+.|~]+$/;
+
+function isValidHeaderName(name: string): boolean {
+    return VALID_HEADER_NAME_CHARS.test(name);
+}
+
+const INVALID_HEADER_VALUE_CHARS = /[^\t\x20-\x7e\x80-\xff]/;
+
+function isValidHeaderValue(value: string): boolean {
+    return !INVALID_HEADER_VALUE_CHARS.test(value);
+}
+
+class SynapseTracePropargator implements TextMapPropagator {
+    inject(context: Context, carrier: any, setter: TextMapSetter<any>): void {
+        const spanContext = api.trace.getSpan(context)?.spanContext();
+        if (!spanContext || !api.isSpanContextValid(spanContext)) return;
+
+        setter.set(carrier, SYNAPSE_TRACE_HEADER, `${spanContext.traceId}:${spanContext.spanId}:0:${spanContext.traceFlags}`);
+        const baggage = api.propagation.getBaggage(context);
+        if (!baggage) return;
+        baggage.getAllEntries().forEach(([k, v]) => {
+            if (!isValidHeaderName(k) || !isValidHeaderValue(v.value)) return;
+            setter.set(carrier, `${SYNAPSE_BAGGAGE_HEADER_PREFIX}${k}`, v.value);
+        });
+    }
+    extract(context: Context, carrier: any, getter: TextMapGetter<any>): Context {
+        const header = readHeader(carrier, getter, SYNAPSE_TRACE_HEADER);
+        if (header.split(':').length - 1 !== 4) {
+            return context;
+        }
+        const trace_data = readHeader(carrier, getter, SYNAPSE_TRACE_HEADER).split(':');
+        const traceId = trace_data[0];
+        const spanId = trace_data[1];
+        let parentId: string | null = trace_data[1];
+        if (parentId === "0") {
+            parentId = null;
+        }
+        const traceFlags = Number(trace_data[2]);
+
+        context = api.trace.setSpan(
+            context,
+            api.trace.wrapSpanContext({
+                traceId,
+                spanId,
+                isRemote: true,
+                traceFlags,
+            })
+        );
+
+        let baggage: api.Baggage =
+            api.propagation.getBaggage(context) || api.propagation.createBaggage();
+
+        getter.keys(carrier).forEach(k => {
+            if (!k.startsWith(SYNAPSE_BAGGAGE_HEADER_PREFIX)) return;
+            const value = readHeader(carrier, getter, k);
+            baggage = baggage.setEntry(k.substr(SYNAPSE_BAGGAGE_HEADER_PREFIX.length), {
+                value,
+            });
+        });
+
+        if (baggage.getAllEntries().length > 0) {
+            context = api.propagation.setBaggage(context, baggage);
+        }
+
+
+        return context;
+    }
+    /**
+     * Note: fields does not include baggage headers as they are dependent on
+     * carrier instance. Attempting to reuse a carrier by clearing fields could
+     * result in a memory leak.
+     */
+    fields(): string[] {
+        return FIELDS.slice();
+    }
+}
+
 export default function initTracer(serviceName: string) {
     /**
      * This starts instrumentation for the app
@@ -93,8 +188,19 @@ export default function initTracer(serviceName: string) {
                 '@opentelemetry/instrumentation-fs': {
                     enabled: false,
                 },
+                // Ignore health and metrics endpoints
+                '@opentelemetry/instrumentation-http': {
+                    ignoreIncomingRequestHook(req) {
+                        // Ignore spans from healthz.
+                        const isHealthz = !!req.url?.match(/^\/healthz$/);
+                        // Ignore spans from metrics
+                        const isMetrics = !!req.url?.match(/^\/metrics$/);
+                        return isHealthz || isMetrics;
+                    }
+                }
             })]
         });
+        api.propagation.setGlobalPropagator(new SynapseTracePropargator());
 
         sdk.start();
 

yarn.lock

@@ -1104,6 +1104,16 @@
   resolved "https://registry.yarnpkg.com/@types/crypto-js/-/crypto-js-4.1.2.tgz#fb56b34f397d9ae2335611e416f15e7d65e276e6"
   integrity sha512-t33RNmTu5ufG/sorROIafiCVJMx3jz95bXUMoPAZcUD14fxMXnuTzqzXZoxpR0tNx2xpw11Dlmem9vGCsrSOfA==
 
+"@types/express-serve-static-core@^4.17.18":
+  version "4.17.39"
+  resolved "https://registry.yarnpkg.com/@types/express-serve-static-core/-/express-serve-static-core-4.17.39.tgz#2107afc0a4b035e6cb00accac3bdf2d76ae408c8"
+  integrity sha512-BiEUfAiGCOllomsRAZOiMFP7LAnrifHpt56pc4Z7l9K6ACyN06Ns1JLMBxwkfLOjJRlSf06NwWsT7yzfpaVpyQ==
+  dependencies:
+    "@types/node" "*"
+    "@types/qs" "*"
+    "@types/range-parser" "*"
+    "@types/send" "*"
+
 "@types/express-serve-static-core@^4.17.33":
   version "4.17.36"
   resolved "https://registry.npmjs.org/@types/express-serve-static-core/-/express-serve-static-core-4.17.36.tgz"