feat: go server backend (#42)

jest-opentelemetry.config.js

@@ -1,3 +1,4 @@
 module.exports = {
   timeout: 2000,
+  useRemoteOtelReceiver: false,
 };

packages/expect-opentelemetry/src/trace-loop/fetch-traces.ts

@@ -2,20 +2,23 @@ import { opentelemetry } from '@traceloop/otel-proto';
 import { setTimeout } from 'timers/promises';
 import { httpGetBinary } from '../utils';
 
-const TRACE_LOOP_ID_HEADER_OTEL_ATTRIBUTE = 'http.request.header.trace_loop_id';
+const TRACELOOP_ID_REQUEST_HEADER = 'http.request.header.traceloop_id';
+const TRACELOOP_ID_RESPONSE_HEADER = 'http.response.header.traceloop_id';
 
 export interface FetchTracesConfig {
   maxPollTime: number;
   pollInterval: number;
   awaitAllTracesTimeout: number;
   url: string;
+  customerId: string;
 }
 
 export const fetchTracesConfigBase: FetchTracesConfig = {
-  maxPollTime: 10000,
-  pollInterval: 500,
-  awaitAllTracesTimeout: 4000,
+  maxPollTime: 120000,
+  pollInterval: 1000,
+  awaitAllTracesTimeout: 1000,
   url: 'http://localhost:4123/v1/traces',
+  customerId: 'local',
 };
 
 /**
@@ -34,7 +37,25 @@ export const findTraceLoopIdMatch = (
       for (const span of scopeSpan.spans || []) {
         if (span.attributes) {
           for (const attribute of span.attributes) {
-            if (attribute.key === TRACE_LOOP_ID_HEADER_OTEL_ATTRIBUTE) {
+            // http: check in headers stringified json
+            if (attribute.key === 'http.request.headers') {
+              const matches = attribute.value?.stringValue?.match(
+                /"traceloop_id":"(.*)"/,
+              );
+              if (matches?.length > 1) {
+                if (matches[1] === traceLoopId) {
+                  return span.traceId
+                    ? Buffer.from(span.traceId).toString('hex')
+                    : undefined;
+                }
+              }
+            }
+
+            // check in specific header key
+            if (
+              attribute.key === TRACELOOP_ID_REQUEST_HEADER ||
+              attribute.key === TRACELOOP_ID_RESPONSE_HEADER
+            ) {
               if (
                 attribute.value?.arrayValue?.values?.[0]?.stringValue ===
                 traceLoopId
@@ -58,13 +79,20 @@ export const pollForTraceLoopIdMatch = async (
   let foundMatch = false;
   while (!foundMatch) {
     await setTimeout(config.pollInterval);
-    const response = await httpGetBinary(config.url);
-    const traces = opentelemetry.proto.trace.v1.TracesData.decode(response);
+    try {
+      const response = await httpGetBinary(config, traceLoopId);
+      const traces = opentelemetry.proto.trace.v1.TracesData.decode(response);
 
-    const traceId = findTraceLoopIdMatch(traces, traceLoopId);
-    if (traceId) {
-      foundMatch = true;
-      return traceId;
+      const traceId = findTraceLoopIdMatch(traces, traceLoopId);
+      if (traceId) {
+        foundMatch = true;
+        return traceId;
+      }
+    } catch (e) {
+      // retry on 400, else throw
+      if ((e as Error)?.message !== '400') {
+        throw e;
+      }
     }
   }
 };

packages/expect-opentelemetry/src/trace-loop/index.ts

@@ -17,7 +17,7 @@ import {
   byCustomAttribute,
 } from './filter-service-spans';
 
-const TRACE_LOOP_ID_HEADER = 'trace-loop-id';
+const TRACE_LOOP_ID_HEADER = 'traceloop_id';
 
 export class TraceLoop {
   private readonly _traceLoopId: string;
@@ -63,7 +63,7 @@ export class TraceLoop {
     // allow time for all spans for the current trace to be received
     await setTimeout(config.awaitAllTracesTimeout);
 
-    const response = await httpGetBinary(config.url);
+    const response = await httpGetBinary(config, this._traceLoopId);
     this._traceData = opentelemetry.proto.trace.v1.TracesData.decode(response);
     this._fetchedTrace = true;
   }

packages/expect-opentelemetry/src/utils.ts

@@ -1,5 +1,6 @@
 import http from 'http';
 import { opentelemetry } from '@traceloop/otel-proto';
+import { FetchTracesConfig } from './trace-loop/fetch-traces';
 
 export const getInstanceType = (instance: any) => {
   if (
@@ -81,13 +82,17 @@ export const generateStubData = () => {
  * @param url - url to make get request to (server that responds with Buffer)
  * @returns Buffer result
  */
-export function httpGetBinary(url: string): Promise<Buffer> {
+export function httpGetBinary(
+  config: FetchTracesConfig,
+  traceloopId: string,
+): Promise<Buffer> {
+  const url = `${config.url}/${traceloopId}`;
   return new Promise((resolve, reject) => {
-    http.get(url, (res) => {
+    http.get(url, { headers: { Authorization: config.customerId } }, (res) => {
       const { statusCode } = res;
 
       if (!statusCode || statusCode < 200 || statusCode >= 300) {
-        return reject(new Error('statusCode=' + res.statusCode));
+        return reject(new Error(`${res.statusCode}`));
       }
 
       const data: Uint8Array[] = [];

packages/instrument-opentelemetry/src/otel-custom/http.js

@@ -90,12 +90,12 @@ export const httpInstrumentationConfig = {
   responseHook: httpCustomAttributesOnResponse,
   headersToSpanAttributes: {
     client: {
-      requestHeaders: ['trace-loop-id'],
-      responseHeaders: ['trace-loop-id'],
+      requestHeaders: ['traceloop_id'],
+      responseHeaders: ['traceloop_id'],
     },
     server: {
-      requestHeaders: ['trace-loop-id'],
-      responseHeaders: ['trace-loop-id'],
+      requestHeaders: ['traceloop_id'],
+      responseHeaders: ['traceloop_id'],
     },
   },
 };

packages/jest-environment-otel/src/global.ts

@@ -18,6 +18,10 @@ export async function setup(jestConfig: JestConfig = {}) {
     didAlreadyRunInWatchMode = true;
   }
 
+  if (config.useRemoteOtelReceiver) {
+    return;
+  }
+
   try {
     await setupServer({
       command: 'node ./node_modules/@traceloop/otel-receiver/dist/index.js',

packages/otel-receiver/src/index.ts

@@ -21,7 +21,7 @@ const startServer = async () => {
     },
   );
 
-  app.get('/v1/traces', (_: Request, res: Response) => {
+  app.get('/v1/traces/:traceloopId', (_: Request, res: Response) => {
     res.send(getAll());
   });