feat(telemetry-server): implemented OTLP receiver (experimental)

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

@@ -22,6 +22,7 @@ import { logger } from '@genkit-ai/tools-common/utils';
 import express from 'express';
 import type * as http from 'http';
 import type { TraceStore } from './types';
+import { traceDataFromOtlp } from './utils/otlp';
 
 export { LocalFileTraceStore } from './file-trace-store.js';
 export { TraceQuerySchema, type TraceQuery, type TraceStore } from './types';
@@ -90,6 +91,29 @@ export async function startTelemetryServer(params: {
     }
   });
 
+  api.post('/api/otlp', async (request, response) => {
+    try {
+      if (!request.body.resourceSpans?.length) {
+        // Acknowledge and ignore empty payloads.
+        response.status(200).json({});
+        return;
+      }
+      const traces = traceDataFromOtlp(request.body);
+      for (const trace of traces) {
+        const traceData = TraceDataSchema.parse(trace);
+        await params.traceStore.save(traceData.traceId, 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.use((err: any, req: any, res: any, next: any) => {
     logger.error(err.stack);
     const error = err as Error;

genkit-tools/telemetry-server/src/utils/otlp.ts

@@ -0,0 +1,162 @@
+/**
+ * Copyright 2025 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { SpanData, TraceData } from '@genkit-ai/tools-common';
+
+// These interfaces are based on the OTLP JSON format.
+// A full definition can be found at:
+// https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto
+
+interface OtlpValue {
+  stringValue?: string;
+  intValue?: number;
+  boolValue?: boolean;
+  arrayValue?: {
+    values: OtlpValue[];
+  };
+}
+
+interface OtlpAttribute {
+  key: string;
+  value: OtlpValue;
+}
+
+interface OtlpSpan {
+  traceId: string;
+  spanId: string;
+  parentSpanId?: string;
+  name: string;
+  kind: number;
+  startTimeUnixNano: string;
+  endTimeUnixNano: string;
+  attributes: OtlpAttribute[];
+  droppedAttributesCount: number;
+  events: any[];
+  droppedEventsCount: number;
+  status?: {
+    code: number;
+    message?: string;
+  };
+  links: any[];
+  droppedLinksCount: number;
+}
+
+interface OtlpScopeSpan {
+  scope: {
+    name: string;
+    version: string;
+  };
+  spans: OtlpSpan[];
+}
+
+interface OtlpResourceSpan {
+  resource: {
+    attributes: OtlpAttribute[];
+    droppedAttributesCount: number;
+  };
+  scopeSpans: OtlpScopeSpan[];
+}
+
+interface OtlpPayload {
+  resourceSpans: OtlpResourceSpan[];
+}
+
+function toMillis(nano: string): number {
+  return Math.round(parseInt(nano) / 1_000_000);
+}
+
+function toSpanData(span: OtlpSpan, scope: OtlpScopeSpan['scope']): SpanData {
+  const attributes: Record<string, any> = {};
+  span.attributes.forEach((attr) => {
+    if (attr.value.stringValue) {
+      attributes[attr.key] = attr.value.stringValue;
+    } else if (attr.value.intValue) {
+      attributes[attr.key] = attr.value.intValue;
+    } else if (attr.value.boolValue) {
+      attributes[attr.key] = attr.value.boolValue;
+    }
+  });
+
+  let spanKind: string;
+  switch (span.kind) {
+    case 1:
+      spanKind = 'INTERNAL';
+      break;
+    case 2:
+      spanKind = 'SERVER';
+      break;
+    case 3:
+      spanKind = 'CLIENT';
+      break;
+    case 4:
+      spanKind = 'PRODUCER';
+      break;
+    case 5:
+      spanKind = 'CONSUMER';
+      break;
+    default:
+      spanKind = 'UNSPECIFIED';
+      break;
+  }
+
+  const spanData: SpanData = {
+    traceId: span.traceId,
+    spanId: span.spanId,
+    parentSpanId: span.parentSpanId,
+    startTime: toMillis(span.startTimeUnixNano),
+    endTime: toMillis(span.endTimeUnixNano),
+    displayName: span.name,
+    attributes,
+    instrumentationLibrary: {
+      name: scope.name,
+      version: scope.version,
+    },
+    spanKind,
+  };
+  if (span.status && span.status.code !== 0) {
+    const status: { code: number; message?: string } = {
+      code: span.status.code,
+    };
+    if (span.status.message) {
+      status.message = span.status.message;
+    }
+    spanData.status = status;
+  }
+  return spanData;
+}
+
+export function traceDataFromOtlp(otlpData: OtlpPayload): TraceData[] {
+  const traces: Record<string, TraceData> = {};
+
+  otlpData.resourceSpans.forEach((resourceSpan) => {
+    resourceSpan.scopeSpans.forEach((scopeSpan) => {
+      scopeSpan.spans.forEach((span) => {
+        if (!traces[span.traceId]) {
+          traces[span.traceId] = {
+            traceId: span.traceId,
+            spans: {},
+          };
+        }
+        traces[span.traceId].spans[span.spanId] = toSpanData(
+          span,
+          scopeSpan.scope
+        );
+      });
+    });
+  });
+
+  return Object.values(traces);
+}

genkit-tools/telemetry-server/tests/file_store_test.ts

@@ -16,16 +16,17 @@
 
 import type { TraceData, TraceQueryFilter } from '@genkit-ai/tools-common';
 import * as assert from 'assert';
+import fs from 'fs';
 import getPort from 'get-port';
 import { afterEach, beforeEach, describe, it } from 'node:test';
 import os from 'os';
 import path from 'path';
+import { Index } from '../src/file-trace-store';
 import {
   LocalFileTraceStore,
   startTelemetryServer,
   stopTelemetryApi,
 } from '../src/index';
-import { Index } from '../src/localFileTraceStore';
 import { sleep, span } from './utils';
 
 const TRACE_ID = '1234';
@@ -37,10 +38,10 @@ const SPAN_B = 'bcd';
 const SPAN_C = 'cde';
 
 describe('local-file-store', () => {
-  let port;
-  let storeRoot;
-  let indexRoot;
-  let url;
+  let port: number;
+  let storeRoot: string;
+  let indexRoot: string;
+  let url: string;
 
   beforeEach(async () => {
     port = await getPort();
@@ -276,18 +277,22 @@ describe('local-file-store', () => {
 });
 
 describe('index', () => {
-  let indexRoot;
+  let indexRoot: string;
   let index: Index;
 
   beforeEach(async () => {
     indexRoot = path.resolve(
       os.tmpdir(),
-      `./telemetry-server-api-test-${Date.now()}/traces_idx`
+      `./telemetry-server-api-test-${Date.now()}-${Math.floor(Math.random() * 1000)}/traces_idx`
     );
 
     index = new Index(indexRoot);
   });
 
+  afterEach(() => {
+    fs.rmSync(indexRoot, { recursive: true, force: true });
+  });
+
   it('should index and search spans', () => {
     const spanA = span(TRACE_ID_1, SPAN_A, 100, 100);
     spanA.displayName = 'spanA';