github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Support json traces and logs in the otel endpoints (#954)

Browse Source
This commit is contained in:
Eric Allam
2024-03-19 10:59:50 +00:00
committed by GitHub
parent 6afda40c35
commit 86c24afb44
5 changed files with 85 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/routes/otel.v1.logs.ts
+17 -5
View File
@@ -1,13 +1,25 @@
import { ActionFunctionArgs } from "@remix-run/server-runtime";
import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
import { ExportLogsServiceRequest, ExportLogsServiceResponse } from "@trigger.dev/otlp-importer";
import { otlpExporter } from "~/v3/otlpExporter.server";
export async function action({ request }: ActionFunctionArgs) {
const buffer = await request.arrayBuffer();
const contentType = request.headers.get("content-type");
const exportRequest = ExportLogsServiceRequest.decode(new Uint8Array(buffer));
if (contentType === "application/json") {
const body = await request.json();
const exportResponse = await otlpExporter.exportLogs(exportRequest);
const exportResponse = await otlpExporter.exportLogs(body as ExportLogsServiceRequest);
return new Response(ExportLogsServiceResponse.encode(exportResponse).finish(), { status: 200 });
return json(exportResponse, { status: 200 })
} else if (contentType === "application/x-protobuf") {
const buffer = await request.arrayBuffer();
const exportRequest = ExportLogsServiceRequest.decode(new Uint8Array(buffer));
const exportResponse = await otlpExporter.exportLogs(exportRequest);
return new Response(ExportLogsServiceResponse.encode(exportResponse).finish(), { status: 200 });
} else {
return new Response("Unsupported content type. Must be either application/x-protobuf or application/json", { status: 400 });
}
}
apps/webapp/app/routes/otel.v1.traces.ts
+17 -5
View File
@@ -1,13 +1,25 @@
import { ActionFunctionArgs } from "@remix-run/server-runtime";
import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
import { ExportTraceServiceRequest, ExportTraceServiceResponse } from "@trigger.dev/otlp-importer";
import { otlpExporter } from "~/v3/otlpExporter.server";
export async function action({ request }: ActionFunctionArgs) {
const buffer = await request.arrayBuffer();
const contentType = request.headers.get("content-type");
const exportRequest = ExportTraceServiceRequest.decode(new Uint8Array(buffer));
if (contentType === "application/json") {
const body = await request.json();
const exportResponse = await otlpExporter.exportTraces(exportRequest);
const exportResponse = await otlpExporter.exportTraces(body as ExportTraceServiceRequest);
return new Response(ExportTraceServiceResponse.encode(exportResponse).finish(), { status: 200 });
return json(exportResponse, { status: 200 })
} else if (contentType === "application/x-protobuf") {
const buffer = await request.arrayBuffer();
const exportRequest = ExportTraceServiceRequest.decode(new Uint8Array(buffer));
const exportResponse = await otlpExporter.exportTraces(exportRequest);
return new Response(ExportTraceServiceResponse.encode(exportResponse).finish(), { status: 200 });
} else {
return new Response("Unsupported content type. Must be either application/x-protobuf or application/json", { status: 400 });
}
}
apps/webapp/app/v3/otlpExporter.server.ts
+38 -37
View File
@@ -35,7 +35,7 @@ class OTLPExporter {
constructor(
private readonly _eventRepository: EventRepository,
private readonly _verbose: boolean
) {}
) { }
async exportTraces(request: ExportTraceServiceRequest): Promise<ExportTraceServiceResponse> {
this.#logExportTracesVerbose(request);
@@ -109,7 +109,7 @@ class OTLPExporter {
if (!triggerAttribute) return false;
return isBoolValue(triggerAttribute.value) ? triggerAttribute.value.value.boolValue : false;
return isBoolValue(triggerAttribute.value) ? triggerAttribute.value.boolValue : false;
});
}
@@ -123,7 +123,7 @@ class OTLPExporter {
if (!attribute) return false;
return isBoolValue(attribute.value) ? attribute.value.value.boolValue : false;
return isBoolValue(attribute.value) ? attribute.value.boolValue : false;
});
}
}
@@ -141,7 +141,7 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array<Creatab
traceId: binaryToHex(log.traceId),
spanId: eventRepository.generateSpanId(),
parentId: binaryToHex(log.spanId),
message: isStringValue(log.body) ? log.body.value.stringValue : `${log.severityText} log`,
message: isStringValue(log.body) ? log.body.stringValue : `${log.severityText} log`,
isPartial: false,
kind: "INTERNAL",
level: logLevelToEventLevel(log.severityNumber),
@@ -201,10 +201,10 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
traceId: binaryToHex(span.traceId),
spanId: isPartial
? extractStringAttribute(
span?.attributes ?? [],
SemanticInternalAttributes.SPAN_ID,
binaryToHex(span.spanId)
)
span?.attributes ?? [],
SemanticInternalAttributes.SPAN_ID,
binaryToHex(span.spanId)
)
: binaryToHex(span.spanId),
parentId: binaryToHex(span.parentSpanId),
message: span.name,
@@ -327,16 +327,16 @@ function convertKeyValueItemsToMap(
if (filteredKeys.includes(attribute.key)) return map;
map[`${prefix ? `${prefix}.` : ""}${attribute.key}`] = isStringValue(attribute.value)
? attribute.value.value.stringValue
? attribute.value.stringValue
: isIntValue(attribute.value)
? Number(attribute.value.value.intValue)
: isDoubleValue(attribute.value)
? attribute.value.value.doubleValue
: isBoolValue(attribute.value)
? attribute.value.value.boolValue
: isBytesValue(attribute.value)
? binaryToHex(attribute.value.value.bytesValue)
: undefined;
? Number(attribute.value.intValue)
: isDoubleValue(attribute.value)
? attribute.value.doubleValue
: isBoolValue(attribute.value)
? attribute.value.boolValue
: isBytesValue(attribute.value)
? binaryToHex(attribute.value.bytesValue)
: undefined;
return map;
},
@@ -505,8 +505,8 @@ function logLevelToEventStatus(level: SeverityNumber): CreatableEventStatus {
}
}
function convertUnixNanoToDate(unixNano: bigint): Date {
return new Date(Number(unixNano / BigInt(1_000_000)));
function convertUnixNanoToDate(unixNano: bigint | number): Date {
return new Date(Number(BigInt(unixNano) / BigInt(1_000_000)));
}
function extractStringAttribute(attributes: KeyValue[], name: string): string | undefined;
@@ -520,7 +520,7 @@ function extractStringAttribute(
if (!attribute) return fallback;
return isStringValue(attribute?.value) ? attribute.value.value.stringValue : fallback;
return isStringValue(attribute?.value) ? attribute.value.stringValue : fallback;
}
function extractNumberAttribute(attributes: KeyValue[], name: string): number | undefined;
@@ -534,7 +534,7 @@ function extractNumberAttribute(
if (!attribute) return fallback;
return isIntValue(attribute?.value) ? Number(attribute.value.value.intValue) : fallback;
return isIntValue(attribute?.value) ? Number(attribute.value.intValue) : fallback;
}
function extractBooleanAttribute(attributes: KeyValue[], name: string): boolean | undefined;
@@ -548,7 +548,7 @@ function extractBooleanAttribute(
if (!attribute) return fallback;
return isBoolValue(attribute?.value) ? attribute.value.value.boolValue : fallback;
return isBoolValue(attribute?.value) ? attribute.value.boolValue : fallback;
}
function isPartialSpan(span: Span): boolean {
@@ -560,58 +560,59 @@ function isPartialSpan(span: Span): boolean {
if (!attribute) return false;
return isBoolValue(attribute.value) ? attribute.value.value.boolValue : false;
return isBoolValue(attribute.value) ? attribute.value.boolValue : false;
}
function isBoolValue(
value: AnyValue | undefined
): value is { value: { $case: "boolValue"; boolValue: boolean } } {
): value is { boolValue: boolean } {
if (!value) return false;
return (value.value && value.value.$case === "boolValue")!!;
return typeof value.boolValue === "boolean";
}
function isStringValue(
value: AnyValue | undefined
): value is { value: { $case: "stringValue"; stringValue: string } } {
): value is { stringValue: string } {
if (!value) return false;
return (value.value && value.value.$case === "stringValue")!!;
return typeof value.stringValue === "string";
}
function isIntValue(
value: AnyValue | undefined
): value is { value: { $case: "intValue"; intValue: bigint } } {
): value is { intValue: bigint } {
if (!value) return false;
return (value.value && value.value.$case === "intValue")!!;
return typeof value.intValue === "number";
}
function isDoubleValue(
value: AnyValue | undefined
): value is { value: { $case: "doubleValue"; doubleValue: number } } {
): value is { doubleValue: number } {
if (!value) return false;
return (value.value && value.value.$case === "doubleValue")!!;
return typeof value.doubleValue === "number";
}
function isBytesValue(
value: AnyValue | undefined
): value is { value: { $case: "bytesValue"; bytesValue: Buffer } } {
): value is { bytesValue: Buffer } {
if (!value) return false;
return (value.value && value.value.$case === "bytesValue")!!;
return Buffer.isBuffer(value.bytesValue);
}
function binaryToHex(buffer: Buffer): string;
function binaryToHex(buffer: Buffer | undefined): string | undefined;
function binaryToHex(buffer: Buffer | undefined): string | undefined {
function binaryToHex(buffer: Buffer | string): string;
function binaryToHex(buffer: Buffer | string | undefined): string | undefined;
function binaryToHex(buffer: Buffer | string | undefined): string | undefined {
if (!buffer) return undefined;
if (typeof buffer === "string") return buffer;
return Buffer.from(Array.from(buffer)).toString("hex");
}
export const otlpExporter = new OTLPExporter(
eventRepository,
process.env.OTL_EXPORTER_VERBOSE === "1"
process.env.OTLP_EXPORTER_VERBOSE === "1"
);
packages/cli-v3/src/workers/dev/backgroundWorker.ts
+13 -9
View File
@@ -167,8 +167,8 @@ export class BackgroundWorkerCoordinator {
!completion.ok && completion.skippedRetrying
? " (retrying skipped)"
: !completion.ok && completion.retry !== undefined
? ` (retrying in ${completion.retry.delay}ms)`
: "";
? ` (retrying in ${completion.retry.delay}ms)`
: "";
const resultText = !completion.ok
? completion.error.type === "INTERNAL_ERROR" &&
@@ -181,8 +181,8 @@ export class BackgroundWorkerCoordinator {
const errorText = !completion.ok
? this.#formatErrorLog(completion.error)
: "retry" in completion
? `retry in ${completion.retry}ms`
: "";
? `retry in ${completion.retry}ms`
: "";
const elapsedText = chalk.dim(`(${elapsed.toFixed(2)}ms)`);
@@ -263,7 +263,7 @@ export class BackgroundWorker {
constructor(
public path: string,
private params: BackgroundWorkerParams
) {}
) { }
close() {
if (this._closed) {
@@ -536,6 +536,11 @@ class TaskRunProcess {
}
async initialize() {
logger.debug("initializing task run process", {
env: this.env,
path: this.path,
})
this._child = fork(this.path, {
stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"],
cwd: dirname(this.path),
@@ -544,6 +549,7 @@ class TaskRunProcess {
OTEL_RESOURCE_ATTRIBUTES: JSON.stringify({
[SemanticInternalAttributes.PROJECT_DIR]: this.worker.projectConfig.projectDir,
}),
OTEL_EXPORTER_OTLP_COMPRESSION: "none",
...(this.worker.debugOtel ? { OTEL_LOG_LEVEL: "debug" } : {}),
},
execArgv: this.worker.debuggerOn
@@ -688,8 +694,7 @@ class TaskRunProcess {
}
logger.log(
`[${this.metadata.version}][${this._currentExecution.run.id}.${
this._currentExecution.attempt.number
`[${this.metadata.version}][${this._currentExecution.run.id}.${this._currentExecution.attempt.number
}] ${data.toString()}`
);
}
@@ -706,8 +711,7 @@ class TaskRunProcess {
}
logger.error(
`[${this.metadata.version}][${this._currentExecution.run.id}.${
this._currentExecution.attempt.number
`[${this.metadata.version}][${this._currentExecution.run.id}.${this._currentExecution.attempt.number
}] ${data.toString()}`
);
}
packages/otlp-importer/scripts/generate-protos.mjs
-1
View File
@@ -46,7 +46,6 @@ for (const proto of protos) {
`--ts_proto_opt=env=node ` +
`--ts_proto_opt=removeEnumPrefix=true ` +
`--ts_proto_opt=lowerCaseServiceMethods=true ` +
`--ts_proto_opt=oneof=unions ` +
`--experimental_allow_proto3_optional ` +
`"${path.join(protosPath, proto)}"`;
try {