feat(genkit-tools/telemetry-server): Add a new otlp endpoint to support nested traces (#3796)

Author: ssbushi

genkit-tools/common/src/eval/evaluate.ts

@@ -491,6 +491,8 @@ async function gatherEvalInput(params: {
 
   // Only the last collected trace to be used for evaluation.
   const traceId = traceIds.at(-1)!;
+  // Sleep to let traces persist.
+  await new Promise((resolve) => setTimeout(resolve, 2000));
   const trace = await manager.getTrace({
     traceId,
   });

genkit-tools/telemetry-server/src/index.ts

@@ -91,6 +91,41 @@ export async function startTelemetryServer(params: {
     }
   });
 
+  api.post(
+    '/api/otlp/:parentTraceId/:parentSpanId',
+    async (request, response) => {
+      try {
+        const { parentTraceId, parentSpanId } = request.params;
+
+        if (!request.body.resourceSpans?.length) {
+          // Acknowledge and ignore empty payloads.
+          response.status(200).json({});
+          return;
+        }
+        const traces = traceDataFromOtlp(request.body);
+        for (const traceData of traces) {
+          traceData.traceId = parentTraceId;
+          for (const span of Object.values(traceData.spans)) {
+            span.attributes['genkit:otlp-traceId'] = span.traceId;
+            span.traceId = parentTraceId;
+            if (!span.parentSpanId) {
+              span.parentSpanId = parentSpanId;
+            }
+          }
+          await params.traceStore.save(parentTraceId, traceData);
+        }
+        response.status(200).json({});
+      } catch (err) {
+        logger.error(`Error processing OTLP payload: ${err}`);
+        response.status(500).json({
+          code: 13, // INTERNAL
+          message:
+            'An internal error occurred while processing the OTLP payload.',
+        });
+      }
+    }
+  );
+
   api.post('/api/otlp', async (request, response) => {
     try {
       if (!request.body.resourceSpans?.length) {
@@ -99,8 +134,7 @@ export async function startTelemetryServer(params: {
         return;
       }
       const traces = traceDataFromOtlp(request.body);
-      for (const trace of traces) {
-        const traceData = TraceDataSchema.parse(trace);
+      for (const traceData of traces) {
         await params.traceStore.save(traceData.traceId, traceData);
       }
       response.status(200).json({});