Skip to content

feat(tools): consolidate OTLP routes to prioritize standard v1 endpoints #5225

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
MichaelDoyle merged 1 commit into from
May 6, 2026
+23 −67
Merged

feat(tools): consolidate OTLP routes to prioritize standard v1 endpoints #5225

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 23 additions & 67 deletions genkit-tools/telemetry-server/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,11 +241,14 @@ export async function startTelemetryServer(params: {
);

api.post(
'/api/otlp/:parentTraceId/:parentSpanId',
[
'/api/otlp',
'/api/otlp/v1/traces',
'/api/otlp/v1/logs',
'/api/otlp/v1/metrics',
],
async (request, response) => {
try {
const { parentTraceId, parentSpanId } = request.params;

if (
!request.body.resourceSpans?.length &&
!request.body.resourceLogs?.length
Expand All @@ -255,27 +258,29 @@ export async function startTelemetryServer(params: {
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;
}
for (const trace of traces) {
const traceData = TraceDataSchema.parse(trace);
await params.traceStore.save(traceData.traceId, traceData);

// Convert each span to an event and broadcast individually
for (const [_, span] of Object.entries(traceData.spans)) {
const event: {
type: 'span_start' | 'span_end';
traceId: string;
span: SpanData;
} = {
type: span.endTime > 0 ? 'span_end' : 'span_start',
traceId: traceData.traceId,
span,
};
broadcastManager.broadcast(traceData.traceId, event);
}
Comment thread
MichaelDoyle marked this conversation as resolved.
await params.traceStore.save(parentTraceId, traceData);
}

// TODO: Add real time support and broadcast log events
if (request.body.resourceLogs?.length) {
const logs = logDataFromOtlp(request.body);
if (logs.length > 0) {
for (const log of logs) {
log.traceId = parentTraceId;
if (!log.spanId) {
log.spanId = parentSpanId;
}
}
await params.logStore.save(logs);
}
}
Expand All @@ -292,55 +297,6 @@ export async function startTelemetryServer(params: {
}
);

api.post('/api/otlp', async (request, response) => {
try {
if (
!request.body.resourceSpans?.length &&
!request.body.resourceLogs?.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);

// Convert each span to an event and broadcast individually
for (const [_, span] of Object.entries(traceData.spans)) {
const event: {
type: 'span_start' | 'span_end';
traceId: string;
span: SpanData;
} = {
type: span.endTime > 0 ? 'span_end' : 'span_start',
traceId: traceData.traceId,
span,
};
broadcastManager.broadcast(traceData.traceId, event);
}
}

// TODO: Add real time support and broadcast log events
if (request.body.resourceLogs?.length) {
const logs = logDataFromOtlp(request.body);
if (logs.length > 0) {
await params.logStore.save(logs);
}
}

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;
Expand Down
Loading