diff --git a/.env.example b/.env.example index 9916679a5..59e45945e 100644 --- a/.env.example +++ b/.env.example @@ -59,4 +59,8 @@ COORDINATOR_SECRET=coordinator-secret # generate the actual secret with `openssl # CONTAINER_REGISTRY_ORIGIN= # CONTAINER_REGISTRY_USERNAME= # CONTAINER_REGISTRY_PASSWORD= -# DEV_OTEL_EXPORTER_OTLP_ENDPOINT="http://0.0.0.0:4318" \ No newline at end of file +# DEV_OTEL_EXPORTER_OTLP_ENDPOINT="http://0.0.0.0:4318" +# These are needed for the object store (for handling large payloads/outputs) +# OBJECT_STORE_BASE_URL="https://{bucket}.{accountId}.r2.cloudflarestorage.com" +# OBJECT_STORE_ACCESS_KEY_ID= +# OBJECT_STORE_SECRET_ACCESS_KEY= \ No newline at end of file diff --git a/.nvmrc b/.nvmrc index b714151ef..2efc7e111 100644 --- a/.nvmrc +++ b/.nvmrc @@ -1 +1 @@ -v18.18.0 \ No newline at end of file +v20.11.1 \ No newline at end of file diff --git a/apps/webapp/app/components/code/CodeBlock.tsx b/apps/webapp/app/components/code/CodeBlock.tsx index 841ae80ea..a1a496ad5 100644 --- a/apps/webapp/app/components/code/CodeBlock.tsx +++ b/apps/webapp/app/components/code/CodeBlock.tsx @@ -367,7 +367,7 @@ function Chrome({ title }: { title?: string }) { ); } -function TitleRow({ title }: { title: string }) { +export function TitleRow({ title }: { title: string }) { return (
diff --git a/apps/webapp/app/components/primitives/Buttons.tsx b/apps/webapp/app/components/primitives/Buttons.tsx index 944144af2..2ec8bc541 100644 --- a/apps/webapp/app/components/primitives/Buttons.tsx +++ b/apps/webapp/app/components/primitives/Buttons.tsx @@ -285,7 +285,7 @@ export const Button = forwardRef( type LinkPropsType = Pick< LinkProps, - "to" | "target" | "onClick" | "onMouseDown" | "onMouseEnter" | "onMouseLeave" + "to" | "target" | "onClick" | "onMouseDown" | "onMouseEnter" | "onMouseLeave" | "download" > & React.ComponentProps; export const LinkButton = ({ @@ -294,6 +294,7 @@ export const LinkButton = ({ onMouseDown, onMouseEnter, onMouseLeave, + download, ...props }: LinkPropsType) => { const innerRef = useRef(null); @@ -308,7 +309,7 @@ export const LinkButton = ({ }); } - if (to.toString().startsWith("http")) { + if (to.toString().startsWith("http") || to.toString().startsWith("/resources")) { return ( @@ -332,6 +334,7 @@ export const LinkButton = ({ onMouseDown={onMouseDown} onMouseEnter={onMouseEnter} onMouseLeave={onMouseLeave} + download={download} > diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index b706da45b..e5da4a466 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -96,6 +96,9 @@ const EnvironmentSchema = z.object({ CONTAINER_REGISTRY_PASSWORD: z.string().optional(), DEPLOY_REGISTRY_HOST: z.string().optional(), DEV_OTEL_EXPORTER_OTLP_ENDPOINT: z.string().optional(), + OBJECT_STORE_BASE_URL: z.string().optional(), + OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(), + OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(), }); export type Environment = z.infer; diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts index 6dd1d3ff1..d58e604bc 100644 --- a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts @@ -1,13 +1,4 @@ -import { Attributes } from "@opentelemetry/api"; -import { - ExceptionEventProperties, - SemanticInternalAttributes, - SpanEvent, - SpanEvents, - correctErrorStackTrace, - isExceptionSpanEvent, -} from "@trigger.dev/core/v3"; -import { z } from "zod"; +import { prettyPrintPacket } from "@trigger.dev/core/v3"; import { PrismaClient, prisma } from "~/db.server"; import { eventRepository } from "~/v3/eventRepository.server"; @@ -48,12 +39,28 @@ export class SpanPresenter { throw new Error("Event not found"); } + const output = + span.outputType === "application/store" + ? `/resources/packets/${span.environmentId}/${span.output}` + : typeof span.output !== "undefined" && span.output !== null + ? prettyPrintPacket(span.output, span.outputType ?? undefined) + : undefined; + + const payload = + span.payloadType === "application/store" + ? `/resources/packets/${span.environmentId}/${span.payload}` + : typeof span.payload !== "undefined" && span.payload !== null + ? prettyPrintPacket(span.payload, span.payloadType ?? undefined) + : undefined; + return { event: { ...span, events: span.events, - output: span.output ? JSON.stringify(span.output, null, 2) : undefined, - payload: span.payload ? JSON.stringify(span.payload, null, 2) : undefined, + output, + outputType: span.outputType ?? "application/json", + payload, + payloadType: span.payloadType ?? "application/json", properties: span.properties ? JSON.stringify(span.properties, null, 2) : undefined, showActionBar: span.show?.actions === true, }, diff --git a/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts b/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts index 46affe010..9031540fa 100644 --- a/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts @@ -86,6 +86,8 @@ export class TestTaskPresenter { taskr."runtimeEnvironmentId" FROM taskruns AS taskr + WHERE + taskr."payloadType" = 'application/json' ORDER BY taskr."createdAt" DESC;`; diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx index edb752b34..c369697cc 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx @@ -1,4 +1,4 @@ -import { QueueListIcon, StopCircleIcon } from "@heroicons/react/20/solid"; +import { CloudArrowDownIcon, QueueListIcon, StopCircleIcon } from "@heroicons/react/20/solid"; import { useParams } from "@remix-run/react"; import { LoaderFunctionArgs } from "@remix-run/server-runtime"; import { formatDurationNanoseconds, nanosecondsToMilliseconds } from "@trigger.dev/core/v3"; @@ -12,7 +12,6 @@ import { Dialog, DialogTrigger } from "~/components/primitives/Dialog"; import { Header2 } from "~/components/primitives/Headers"; import { Paragraph } from "~/components/primitives/Paragraph"; import { Property, PropertyTable } from "~/components/primitives/PropertyTable"; -import { TextLink } from "~/components/primitives/TextLink"; import { CancelRunDialog } from "~/components/runs/v3/CancelRunDialog"; import { LiveTimer } from "~/components/runs/v3/LiveTimer"; import { RunIcon } from "~/components/runs/v3/RunIcon"; @@ -149,10 +148,10 @@ export default function Page() { {event.events !== undefined && } {event.payload !== undefined && ( - + )} {event.output !== undefined && ( - + )} {event.properties !== undefined && ( @@ -204,6 +203,31 @@ export default function Page() { ); } +function PacketDisplay({ + data, + dataType, + title, +}: { + data: string; + dataType: string; + title: string; +}) { + if (dataType === "application/store") { + return ( +
+ + {title} + + + Download + +
+ ); + } else { + return ; + } +} + type TimelineProps = { startTime: Date; duration: number; diff --git a/apps/webapp/app/routes/api.v1.packets.$.ts b/apps/webapp/app/routes/api.v1.packets.$.ts new file mode 100644 index 000000000..eee457be9 --- /dev/null +++ b/apps/webapp/app/routes/api.v1.packets.$.ts @@ -0,0 +1,101 @@ +import type { ActionFunctionArgs } from "@remix-run/server-runtime"; +import { json } from "@remix-run/server-runtime"; +import { z } from "zod"; +import { env } from "~/env.server"; +import { authenticateApiRequest } from "~/services/apiAuth.server"; +import { logger } from "~/services/logger.server"; +import { r2 } from "~/v3/r2.server"; + +const ParamsSchema = z.object({ + "*": z.string(), +}); + +export async function action({ request, params }: ActionFunctionArgs) { + // Ensure this is a POST request + if (request.method.toUpperCase() !== "PUT") { + return { status: 405, body: "Method Not Allowed" }; + } + + // Next authenticate the request + const authenticationResult = await authenticateApiRequest(request); + + if (!authenticationResult) { + return json({ error: "Invalid or Missing API key" }, { status: 401 }); + } + + const parsedParams = ParamsSchema.parse(params); + const filename = parsedParams["*"]; + + if (!env.OBJECT_STORE_BASE_URL) { + return json({ error: "Object store base URL is not set" }, { status: 500 }); + } + + if (!r2) { + return json({ error: "Object store credentials are not set" }, { status: 500 }); + } + + const url = new URL(env.OBJECT_STORE_BASE_URL); + url.pathname = `/packets/${authenticationResult.environment.project.externalRef}/${authenticationResult.environment.slug}/${filename}`; + url.searchParams.set("X-Amz-Expires", "300"); // 5 minutes + + const signed = await r2.sign( + new Request(url, { + method: "PUT", + }), + { + aws: { signQuery: true }, + } + ); + + logger.debug("Generated presigned URL", { + url: signed.url, + headers: Object.fromEntries(signed.headers), + }); + + // Caller can now use this URL to upload to that object. + return json({ presignedUrl: signed.url }); +} + +export async function loader({ request, params }: ActionFunctionArgs) { + // Next authenticate the request + const authenticationResult = await authenticateApiRequest(request); + + if (!authenticationResult) { + return json({ error: "Invalid or Missing API key" }, { status: 401 }); + } + + const parsedParams = ParamsSchema.parse(params); + const filename = parsedParams["*"]; + + if (!env.OBJECT_STORE_BASE_URL) { + return json({ error: "Object store base URL is not set" }, { status: 500 }); + } + + if (!r2) { + return json({ error: "Object store credentials are not set" }, { status: 500 }); + } + + const url = new URL(env.OBJECT_STORE_BASE_URL); + url.pathname = `/packets/${authenticationResult.environment.project.externalRef}/${authenticationResult.environment.slug}/${filename}`; + url.searchParams.set("X-Amz-Expires", "300"); // 5 minutes + + const signed = await r2.sign( + new Request(url, { + method: request.method, + }), + { + aws: { signQuery: true }, + } + ); + + logger.debug("Generated presigned URL", { + url: signed.url, + headers: Object.fromEntries(signed.headers), + }); + + const getUrl = new URL(url.href); + getUrl.searchParams.delete("X-Amz-Expires"); + + // Caller can now use this URL to upload to that object. + return json({ presignedUrl: signed.url }); +} diff --git a/apps/webapp/app/routes/otel.v1.logs.ts b/apps/webapp/app/routes/otel.v1.logs.ts index 8821e8440..c04be32b4 100644 --- a/apps/webapp/app/routes/otel.v1.logs.ts +++ b/apps/webapp/app/routes/otel.v1.logs.ts @@ -9,7 +9,7 @@ export async function action({ request }: ActionFunctionArgs) { if (contentType === "application/json") { const body = await request.json(); - const exportResponse = await otlpExporter.exportLogs(body as ExportLogsServiceRequest, true); + const exportResponse = await otlpExporter.exportLogs(body as ExportLogsServiceRequest, false); return json(exportResponse, { status: 200 }); } else if (contentType === "application/x-protobuf") { diff --git a/apps/webapp/app/routes/otel.v1.traces.ts b/apps/webapp/app/routes/otel.v1.traces.ts index 6d7667a9a..5d77314a1 100644 --- a/apps/webapp/app/routes/otel.v1.traces.ts +++ b/apps/webapp/app/routes/otel.v1.traces.ts @@ -11,7 +11,7 @@ export async function action({ request }: ActionFunctionArgs) { const exportResponse = await otlpExporter.exportTraces( body as ExportTraceServiceRequest, - true + false ); return json(exportResponse, { status: 200 }); diff --git a/apps/webapp/app/routes/resources.packets.$environmentId.$.ts b/apps/webapp/app/routes/resources.packets.$environmentId.$.ts new file mode 100644 index 000000000..101d03702 --- /dev/null +++ b/apps/webapp/app/routes/resources.packets.$environmentId.$.ts @@ -0,0 +1,70 @@ +import { LoaderFunctionArgs } from "@remix-run/node"; +import { basename } from "node:path"; +import { z } from "zod"; +import { prisma } from "~/db.server"; +import { env } from "~/env.server"; +import { requireUserId } from "~/services/session.server"; +import { r2 } from "~/v3/r2.server"; + +const ParamSchema = z.object({ + environmentId: z.string(), + "*": z.string(), +}); + +export async function loader({ request, params }: LoaderFunctionArgs) { + const userId = await requireUserId(request); + const { environmentId, "*": filename } = ParamSchema.parse(params); + + const environment = await prisma.runtimeEnvironment.findFirst({ + where: { + id: environmentId, + organization: { + members: { + some: { + userId, + }, + }, + }, + }, + include: { + project: true, + }, + }); + + if (!environment) { + return new Response("Not found", { status: 404 }); + } + + if (!env.OBJECT_STORE_BASE_URL) { + return new Response("Object store base URL is not set", { status: 500 }); + } + + if (!r2) { + return new Response("Object store credentials are not set", { status: 500 }); + } + + const url = new URL(env.OBJECT_STORE_BASE_URL); + url.pathname = `/packets/${environment.project.externalRef}/${environment.slug}/${filename}`; + url.searchParams.set("X-Amz-Expires", "30"); // 30 seconds + + const signed = await r2.sign( + new Request(url, { + method: "GET", + }), + { + aws: { signQuery: true }, + } + ); + + const response = await fetch(signed.url, { + headers: signed.headers, + }); + + return new Response(response.body, { + status: 200, + headers: { + "Content-Type": "application/octet-stream", + "Content-Disposition": `attachment; filename="${basename(url.pathname)}"`, + }, + }); +} diff --git a/apps/webapp/app/v3/eventRepository.server.ts b/apps/webapp/app/v3/eventRepository.server.ts index 6341c56d7..6c517ce53 100644 --- a/apps/webapp/app/v3/eventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository.server.ts @@ -10,30 +10,31 @@ import { SpanMessagingEvent, TaskEventStyle, correctErrorStackTrace, + createPackageAttributesAsJson, flattenAttributes, isExceptionSpanEvent, omit, - primitiveValueOrflattenedAttributes, unflattenAttributes, } from "@trigger.dev/core/v3"; import { Prisma, TaskEvent, TaskEventStatus, type TaskEventKind } from "@trigger.dev/database"; -import { createHash } from "node:crypto"; -import { PrismaClient, prisma } from "~/db.server"; -import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; -import { DynamicFlushScheduler } from "./dynamicFlushScheduler.server"; import Redis, { RedisOptions } from "ioredis"; -import { env } from "~/env.server"; +import { createHash } from "node:crypto"; import { EventEmitter } from "node:stream"; +import { PrismaClient, prisma } from "~/db.server"; +import { env } from "~/env.server"; +import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; +import { DynamicFlushScheduler } from "./dynamicFlushScheduler.server"; export type CreatableEvent = Omit< Prisma.TaskEventCreateInput, - "id" | "createdAt" | "properties" | "metadata" | "style" | "output" + "id" | "createdAt" | "properties" | "metadata" | "style" | "output" | "payload" > & { properties: Attributes; metadata: Attributes | undefined; style: Attributes | undefined; output: Attributes | string | boolean | number | undefined; + payload: Attributes | string | boolean | number | undefined; }; export type CreatableEventKind = TaskEventKind; @@ -49,12 +50,15 @@ export type TraceAttributes = Partial< | "runId" | "runIsTest" | "output" + | "outputType" | "metadata" | "properties" | "style" | "queueId" | "queueName" | "batchId" + | "payload" + | "payloadType" > >; @@ -182,7 +186,17 @@ export class EventRepository { const event = events[0]; - logger.debug("Completing event", { spanId, eventId: event.id }); + const output = options?.attributes.output + ? createPackageAttributesAsJson( + options?.attributes.output, + options?.attributes.outputType ?? "application/json" + ) + : undefined; + + logger.debug("Completing event", { + spanId, + eventId: event.id, + }); await this.insert({ ...omit(event, "id"), @@ -196,9 +210,13 @@ export class EventRepository { properties: event.properties as Attributes, metadata: event.metadata as Attributes, style: event.style as Attributes, - output: options?.attributes.output - ? primitiveValueOrflattenedAttributes(options.attributes.output, undefined) - : undefined, + output: output, + outputType: + options?.attributes.outputType === "application/store" + ? "application/store" + : "application/json", + payload: event.payload as Attributes, + payloadType: event.payloadType, }); } @@ -229,6 +247,9 @@ export class EventRepository { metadata: event.metadata as Attributes, style: event.style as Attributes, output: event.output as Attributes, + outputType: event.outputType, + payload: event.payload as Attributes, + payloadType: event.payloadType, }); } @@ -362,14 +383,14 @@ export class EventRepository { return; } - const payload = unflattenAttributes( - filteredAttributes(fullEvent.properties as Attributes, SemanticInternalAttributes.PAYLOAD) - )[SemanticInternalAttributes.PAYLOAD]; - const output = isEmptyJson(fullEvent.output) ? null : unflattenAttributes(fullEvent.output as Attributes); + const payload = isEmptyJson(fullEvent.payload) + ? null + : unflattenAttributes(fullEvent.payload as Attributes); + const show = unflattenAttributes( filteredAttributes(fullEvent.properties as Attributes, SemanticInternalAttributes.SHOW) )[SemanticInternalAttributes.SHOW] as @@ -497,6 +518,9 @@ export class EventRepository { metadata: metadata, style: stripAttributePrefix(style, SemanticInternalAttributes.STYLE), output: undefined, + outputType: undefined, + payload: undefined, + payloadType: undefined, }; if (options.immediate) { @@ -630,7 +654,10 @@ export class EventRepository { metadata: metadata, style: stripAttributePrefix(style, SemanticInternalAttributes.STYLE), output: undefined, + outputType: undefined, links: links as unknown as Prisma.InputJsonValue, + payload: options.attributes.payload, + payloadType: options.attributes.payloadType, }; if (options.immediate) { @@ -717,7 +744,7 @@ export class EventRepository { export const eventRepository = new EventRepository(prisma, { batchSize: 100, - batchInterval: 5000, + batchInterval: 1000, redis: { port: env.REDIS_PORT, host: env.REDIS_HOST, @@ -920,9 +947,6 @@ function isEmptyJson(json: Prisma.JsonValue) { if (json === null) { return true; } - if (Object.keys(json).length === 0) { - return true; - } return false; } diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts index a19d37ad5..e96f8f0c3 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/apps/webapp/app/v3/otlpExporter.server.ts @@ -189,6 +189,13 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array attribute.key === key); + + if (!attribute) return undefined; + + return isStringValue(attribute.value) ? attribute.value.stringValue : undefined; +} + function convertKeyValueItemsToMap( attributes: KeyValue[], filteredKeys: string[] = [], prefix?: string -): Record { - const result = attributes.reduce( - (map: Record, attribute) => { - if (filteredKeys.includes(attribute.key)) return map; +): Record | undefined { + if (!attributes) return; + if (!attributes.length) return; + const filteredAttributes = attributes.filter( + (attribute) => !filteredKeys.includes(attribute.key) + ); + + if (!filteredAttributes.length) return; + + const result = filteredAttributes.reduce( + (map: Record, attribute) => { map[`${prefix ? `${prefix}.` : ""}${attribute.key}`] = isStringValue(attribute.value) ? attribute.value.stringValue : isIntValue(attribute.value) @@ -373,9 +411,11 @@ function convertKeyValueItemsToMap( } function detectPrimitiveValue( - attributes: Record, + attributes: Record | undefined, sentinel: string ): Record | string | number | boolean | undefined { + if (!attributes) return undefined; + if (typeof attributes[sentinel] !== "undefined") { return attributes[sentinel]; } diff --git a/apps/webapp/app/v3/r2.server.ts b/apps/webapp/app/v3/r2.server.ts new file mode 100644 index 000000000..a38a77a1e --- /dev/null +++ b/apps/webapp/app/v3/r2.server.ts @@ -0,0 +1,52 @@ +import { AwsClient } from "aws4fetch"; +import { env } from "~/env.server"; +import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; +import { logger } from "~/services/logger.server"; +import { singleton } from "~/utils/singleton"; + +export const r2 = singleton("r2", initializeR2); + +function initializeR2() { + if (!env.OBJECT_STORE_ACCESS_KEY_ID || !env.OBJECT_STORE_SECRET_ACCESS_KEY) { + return; + } + + return new AwsClient({ + accessKeyId: env.OBJECT_STORE_ACCESS_KEY_ID, + secretAccessKey: env.OBJECT_STORE_SECRET_ACCESS_KEY, + }); +} + +export async function uploadToObjectStore( + filename: string, + data: string, + contentType: string, + environment: AuthenticatedEnvironment +): Promise { + if (!r2) { + throw new Error("Object store credentials are not set"); + } + + if (!env.OBJECT_STORE_BASE_URL) { + throw new Error("Object store base URL is not set"); + } + + const url = new URL(env.OBJECT_STORE_BASE_URL); + url.pathname = `/packets/${environment.project.externalRef}/${environment.slug}/${filename}`; + + logger.debug("Uploading to object store", { url: url.href }); + + const response = await r2.fetch(url.toString(), { + method: "PUT", + headers: { + "Content-Type": contentType, + }, + body: data, + }); + + if (!response.ok) { + throw new Error(`Failed to upload output to ${url}: ${response.statusText}`); + } + + return url.href; +} diff --git a/apps/webapp/app/v3/services/completeAttempt.server.ts b/apps/webapp/app/v3/services/completeAttempt.server.ts index 8a5e729d8..743ce1d2a 100644 --- a/apps/webapp/app/v3/services/completeAttempt.server.ts +++ b/apps/webapp/app/v3/services/completeAttempt.server.ts @@ -92,7 +92,10 @@ export class CompleteAttemptService extends BaseService { }, }); - logger.debug("Completed attempt successfully, ACKing message", taskRunAttempt); + logger.debug("Completed attempt successfully, ACKing message", { + serializedOutput: completion.output, + outputType: completion.outputType, + }); await marqs?.acknowledgeMessage(taskRunAttempt.taskRunId); @@ -101,7 +104,13 @@ export class CompleteAttemptService extends BaseService { endTime: new Date(), attributes: { isError: false, - output: completion.output ? (safeJsonParse(completion.output) as Attributes) : undefined, + output: + completion.outputType === "application/store" + ? completion.output + : completion.output + ? (safeJsonParse(completion.output) as Attributes) + : undefined, + outputType: completion.outputType, }, }); diff --git a/apps/webapp/app/v3/services/triggerTask.server.ts b/apps/webapp/app/v3/services/triggerTask.server.ts index 1e9a3c20d..aea699697 100644 --- a/apps/webapp/app/v3/services/triggerTask.server.ts +++ b/apps/webapp/app/v3/services/triggerTask.server.ts @@ -2,6 +2,7 @@ import { PRIMARY_VARIANT, SemanticInternalAttributes, TriggerTaskRequestBody, + packetRequiresOffloading, } from "@trigger.dev/core/v3"; import { nanoid } from "nanoid"; import { createHash } from "node:crypto"; @@ -11,6 +12,8 @@ import { eventRepository } from "../eventRepository.server"; import { generateFriendlyId } from "../friendlyIdentifiers"; import { marqs } from "../marqs.server"; import { BaseService } from "./baseService.server"; +import { uploadToObjectStore } from "../r2.server"; +import { logger } from "~/services/logger.server"; export type TriggerTaskServiceOptions = { idempotencyKey?: string; @@ -56,7 +59,6 @@ export class TriggerTaskService extends BaseService { taskSlug: taskId, attributes: { properties: { - [SemanticInternalAttributes.PAYLOAD]: body.payload, [SemanticInternalAttributes.SHOW_ACTIONS]: true, }, style: { @@ -98,17 +100,25 @@ export class TriggerTaskService extends BaseService { event.setAttribute("queueName", queueName); span.setAttribute("queueName", queueName); + const runFriendlyId = generateFriendlyId("run"); + + const payloadPacket = await this.#handlePayloadPacket( + body.payload, + runFriendlyId, + environment + ); + const taskRun = await tx.taskRun.create({ data: { status: "PENDING", number: counter.lastNumber, - friendlyId: generateFriendlyId("run"), + friendlyId: runFriendlyId, runtimeEnvironmentId: environment.id, projectId: environment.projectId, idempotencyKey, taskIdentifier: taskId, - payload: JSON.stringify(body.payload), - payloadType: "application/json", + payload: payloadPacket.data, + payloadType: payloadPacket.dataType, context: body.context, traceContext: traceContext, traceId: event.traceId, @@ -120,6 +130,16 @@ export class TriggerTaskService extends BaseService { }, }); + if (payloadPacket.data) { + if (payloadPacket.dataType === "application/json") { + event.setAttribute("payload", JSON.parse(payloadPacket.data) as any); + } else { + event.setAttribute("payload", payloadPacket.data); + } + + event.setAttribute("payloadType", payloadPacket.dataType); + } + event.setAttribute("runId", taskRun.friendlyId); span.setAttribute("runId", taskRun.friendlyId); @@ -166,6 +186,32 @@ export class TriggerTaskService extends BaseService { ); }); } + + async #handlePayloadPacket( + payload: any, + pathPrefix: string, + environment: AuthenticatedEnvironment + ) { + const packet = { + data: JSON.stringify(payload), + dataType: "application/json", + }; + + const { needsOffloading, size } = packetRequiresOffloading(packet); + + if (!needsOffloading) { + return packet; + } + + const filename = `${pathPrefix}/payload.json`; + + await uploadToObjectStore(filename, packet.data, packet.dataType, environment); + + return { + data: filename, + dataType: "application/store", + }; + } } function taskIdentifierToLockId(taskIdentifier: string): number { diff --git a/apps/webapp/package.json b/apps/webapp/package.json index 0afdd6f43..5a5424722 100644 --- a/apps/webapp/package.json +++ b/apps/webapp/package.json @@ -94,6 +94,7 @@ "@uiw/react-codemirror": "^4.19.5", "@upstash/ratelimit": "^1.0.1", "@whatwg-node/fetch": "^0.9.14", + "aws4fetch": "^1.0.18", "class-variance-authority": "^0.5.2", "clsx": "^1.2.1", "compression": "^1.7.4", @@ -145,6 +146,7 @@ "socket.io": "^4.7.4", "sonner": "^1.0.3", "sqs-consumer": "^7.4.0", + "superjson": "^2.2.1", "tailwind-merge": "^1.12.0", "tailwind-scrollbar-hide": "^1.1.7", "tailwindcss-animate": "^1.0.5", diff --git a/apps/webapp/remix.config.js b/apps/webapp/remix.config.js index d183b450b..9ca7e0830 100644 --- a/apps/webapp/remix.config.js +++ b/apps/webapp/remix.config.js @@ -19,6 +19,7 @@ module.exports = { "emails", "highlight.run", "random-words", + "superjson", ], watchPaths: async () => { return [ diff --git a/docker/Dockerfile b/docker/Dockerfile index aa1473360..abbbb34c0 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -1,4 +1,4 @@ -FROM node:18.18.2-bullseye-slim@sha256:21479df46c3173ee0cefc6b264928e10239152c4f74df872ca9369be01a245b7 AS pruner +FROM node:20.11.1-bullseye-slim@sha256:5a5a92b3a8d392691c983719dbdc65d9f30085d6dcd65376e7a32e6fe9bf4cbe AS pruner WORKDIR /triggerdotdev @@ -7,7 +7,7 @@ RUN npx -q turbo@1.10.9 prune --scope=webapp --docker RUN find . -name "node_modules" -type d -prune -exec rm -rf '{}' + # Base strategy to have layer caching -FROM node:18.18.2-bullseye-slim@sha256:21479df46c3173ee0cefc6b264928e10239152c4f74df872ca9369be01a245b7 AS base +FROM node:20.11.1-bullseye-slim@sha256:5a5a92b3a8d392691c983719dbdc65d9f30085d6dcd65376e7a32e6fe9bf4cbe AS base RUN apt-get update && apt-get install -y openssl dumb-init WORKDIR /triggerdotdev COPY --chown=node:node .gitignore .gitignore @@ -53,7 +53,7 @@ RUN pnpm run generate RUN pnpm run build --filter=webapp... # Runner -FROM node:18.18.2-bullseye-slim@sha256:21479df46c3173ee0cefc6b264928e10239152c4f74df872ca9369be01a245b7 AS runner +FROM node:20.11.1-bullseye-slim@sha256:5a5a92b3a8d392691c983719dbdc65d9f30085d6dcd65376e7a32e6fe9bf4cbe AS runner RUN apt-get update && apt-get install -y openssl WORKDIR /triggerdotdev RUN corepack enable diff --git a/packages/cli-v3/src/utilities/build.ts b/packages/cli-v3/src/utilities/build.ts index d1eea449f..9c21e2045 100644 --- a/packages/cli-v3/src/utilities/build.ts +++ b/packages/cli-v3/src/utilities/build.ts @@ -74,6 +74,14 @@ export function bundleDependenciesPlugin(config: ResolvedConfig): Plugin { return undefined; // let esbuild bundle it } + if (args.path === "superjson") { + logger.debug(`Bundling ${args.path} because its superjson`, { + ...args, + }); + + return undefined; // let esbuild bundle it + } + // Skip assets that are treated as files (.css, .svg, .png, etc.). // Otherwise, esbuild would emit code that would attempt to require() // or import these files --- which aren't JavaScript! diff --git a/packages/cli-v3/src/workers/dev/backgroundWorker.ts b/packages/cli-v3/src/workers/dev/backgroundWorker.ts index 0e2d250f4..3b700ddff 100644 --- a/packages/cli-v3/src/workers/dev/backgroundWorker.ts +++ b/packages/cli-v3/src/workers/dev/backgroundWorker.ts @@ -108,7 +108,7 @@ export class BackgroundWorkerCoordinator { } async handleMessage(id: string, message: BackgroundWorkerServerMessages) { - logger.debug(`Received message from worker ${id}`, { workerMessage: message }); + logger.debug(`Received message from worker ${id}`, JSON.stringify({ workerMessage: message })); switch (message.type) { case "EXECUTE_RUNS": { diff --git a/packages/core/package.json b/packages/core/package.json index f292d1af6..387a3383d 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -73,6 +73,7 @@ "humanize-duration": "^3.27.3", "socket.io": "^4.7.4", "socket.io-client": "^4.7.4", + "superjson": "^2.2.1", "ulidx": "^2.2.1", "zod": "3.22.3", "zod-error": "1.5.0" diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts index 6d10d9c98..bf2511eb2 100644 --- a/packages/core/src/v3/apiClient/index.ts +++ b/packages/core/src/v3/apiClient/index.ts @@ -8,6 +8,7 @@ import { TriggerTaskResponse, BatchTriggerTaskRequestBody, BatchTriggerTaskResponse, + CreateUploadPayloadUrlResponseBody, } from "../schemas"; export type TriggerOptions = { @@ -43,6 +44,28 @@ export class ApiClient { }); } + createUploadPayloadUrl(filename: string) { + return zodfetch( + CreateUploadPayloadUrlResponseBody, + `${this.baseUrl}/api/v1/packets/${filename}`, + { + method: "PUT", + headers: this.#getHeaders(false), + } + ); + } + + getPayloadUrl(filename: string) { + return zodfetch( + CreateUploadPayloadUrlResponseBody, + `${this.baseUrl}/api/v1/packets/${filename}`, + { + method: "GET", + headers: this.#getHeaders(false), + } + ); + } + #getHeaders(spanParentAsLink: boolean) { const headers: Record = { "Content-Type": "application/json", diff --git a/packages/core/src/v3/index.ts b/packages/core/src/v3/index.ts index d08a732e8..62ada3d7a 100644 --- a/packages/core/src/v3/index.ts +++ b/packages/core/src/v3/index.ts @@ -10,6 +10,7 @@ export * from "./errors"; export * from "./runtime-api"; export * from "./logger-api"; export * from "./types"; +export * from "./limits"; export { SemanticInternalAttributes } from "./semanticInternalAttributes"; export { iconStringForSeverity } from "./icons"; export { @@ -61,3 +62,14 @@ export { calculatePreciseDateHrTime, preciseDateOriginNow, } from "./utils/preciseDate"; +export { + parsePacket, + stringifyIO, + prettyPrintPacket, + createPacketAttributes, + createPackageAttributesAsJson, + conditionallyExportPacket, + conditionallyImportPacket, + packetRequiresOffloading, + type IOPacket, +} from "./utils/ioSerialization"; diff --git a/packages/core/src/v3/limits.ts b/packages/core/src/v3/limits.ts new file mode 100644 index 000000000..212446faa --- /dev/null +++ b/packages/core/src/v3/limits.ts @@ -0,0 +1,53 @@ +import { AttributeValue, Attributes } from "@opentelemetry/api"; + +export const OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT = 256; +export const OTEL_LOG_ATTRIBUTE_COUNT_LIMIT = 256; +export const OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT = 1028; +export const OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT = 1028; +export const OTEL_SPAN_EVENT_COUNT_LIMIT = 10; +export const OTEL_LINK_COUNT_LIMIT = 2; +export const OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT = 10; +export const OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT = 10; +export const OFFLOAD_IO_PACKET_LENGTH_LIMIT = 128 * 1024; + +export function imposeAttributeLimits(attributes: Attributes): Attributes { + const newAttributes: Attributes = {}; + + for (const [key, value] of Object.entries(attributes)) { + if (calculateAttributeValueLength(value) > OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT) { + continue; + } + + if (Object.keys(newAttributes).length >= OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT) { + break; + } + + newAttributes[key] = value; + } + + return newAttributes; +} + +function calculateAttributeValueLength(value: AttributeValue | undefined | null): number { + if (value === undefined || value === null) { + return 0; + } + + if (typeof value === "string") { + return value.length; + } + + if (typeof value === "number") { + return 8; + } + + if (typeof value === "boolean") { + return 4; + } + + if (Array.isArray(value)) { + return value.reduce((acc: number, v) => acc + calculateAttributeValueLength(v), 0); + } + + return 0; +} diff --git a/packages/core/src/v3/otel/tracingSDK.ts b/packages/core/src/v3/otel/tracingSDK.ts index 09ba3b8e0..611574ba7 100644 --- a/packages/core/src/v3/otel/tracingSDK.ts +++ b/packages/core/src/v3/otel/tracingSDK.ts @@ -25,6 +25,16 @@ import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions" import { SemanticInternalAttributes } from "../semanticInternalAttributes"; import { TaskContextLogProcessor, TaskContextSpanProcessor } from "../tasks/taskContextManager"; import { getEnvVar } from "../utils/getEnv"; +import { + OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT, + OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT, + OTEL_LINK_COUNT_LIMIT, + OTEL_LOG_ATTRIBUTE_COUNT_LIMIT, + OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT, + OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, + OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, + OTEL_SPAN_EVENT_COUNT_LIMIT, +} from "../limits"; class AsyncResourceDetector implements DetectorSync { private _promise: Promise; @@ -105,12 +115,12 @@ export class TracingSDK { forceFlushTimeoutMillis: config.forceFlushTimeoutMillis ?? 500, resource: commonResources, spanLimits: { - attributeCountLimit: 1000, - attributeValueLengthLimit: 1000, - eventCountLimit: 100, - attributePerEventCountLimit: 100, - linkCountLimit: 10, - attributePerLinkCountLimit: 100, + attributeCountLimit: OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT, + attributeValueLengthLimit: OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, + eventCountLimit: OTEL_SPAN_EVENT_COUNT_LIMIT, + attributePerEventCountLimit: OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT, + linkCountLimit: OTEL_LINK_COUNT_LIMIT, + attributePerLinkCountLimit: OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT, }, }); @@ -137,8 +147,8 @@ export class TracingSDK { const loggerProvider = new LoggerProvider({ resource: commonResources, logRecordLimits: { - attributeCountLimit: 1000, - attributeValueLengthLimit: 1000, + attributeCountLimit: OTEL_LOG_ATTRIBUTE_COUNT_LIMIT, + attributeValueLengthLimit: OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT, }, }); diff --git a/packages/core/src/v3/runtime/devRuntimeManager.ts b/packages/core/src/v3/runtime/devRuntimeManager.ts index 007749639..a4df98c89 100644 --- a/packages/core/src/v3/runtime/devRuntimeManager.ts +++ b/packages/core/src/v3/runtime/devRuntimeManager.ts @@ -5,6 +5,7 @@ import { TaskRunExecution, TaskRunExecutionResult, } from "../schemas"; +import { conditionallyImportPacket } from "../utils/ioSerialization"; import { RuntimeManager } from "./manager"; export class DevRuntimeManager implements RuntimeManager { diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts index 1f2f2d960..779de6309 100644 --- a/packages/core/src/v3/schemas/api.ts +++ b/packages/core/src/v3/schemas/api.ts @@ -155,7 +155,15 @@ export type InitializeDeploymentRequestBody = z.infer; + +export const CreateUploadPayloadUrlResponseBody = z.object({ + presignedUrl: z.string(), +}); + +export type CreateUploadPayloadUrlResponseBody = z.infer; diff --git a/packages/core/src/v3/semanticInternalAttributes.ts b/packages/core/src/v3/semanticInternalAttributes.ts index f2352de38..510581a94 100644 --- a/packages/core/src/v3/semanticInternalAttributes.ts +++ b/packages/core/src/v3/semanticInternalAttributes.ts @@ -21,6 +21,7 @@ export const SemanticInternalAttributes = { SPAN_PARTIAL: "$span.partial", SPAN_ID: "$span.span_id", OUTPUT: "$output", + OUTPUT_TYPE: "$mime_type_output", STYLE: "$style", STYLE_ICON: "$style.icon", STYLE_VARIANT: "$style.variant", @@ -28,6 +29,7 @@ export const SemanticInternalAttributes = { METADATA: "$metadata", TRIGGER: "$trigger", PAYLOAD: "$payload", + PAYLOAD_TYPE: "$mime_type_payload", SHOW: "$show", SHOW_ACTIONS: "$show.actions", WORKER_ID: "worker.id", diff --git a/packages/core/src/v3/tasks/taskContextManager.ts b/packages/core/src/v3/tasks/taskContextManager.ts index b65c5ce85..898203a11 100644 --- a/packages/core/src/v3/tasks/taskContextManager.ts +++ b/packages/core/src/v3/tasks/taskContextManager.ts @@ -5,7 +5,6 @@ import { SafeAsyncLocalStorage } from "../utils/safeAsyncLocalStorage"; type TaskContext = { ctx: TaskRunContext; - payload: any; worker: BackgroundWorkerProperties; }; @@ -21,11 +20,6 @@ export class TaskContextManager { return store?.ctx; } - get payload(): any | undefined { - const store = this.#getStore(); - return store?.payload; - } - get worker(): BackgroundWorkerProperties | undefined { const store = this.#getStore(); return store?.worker; @@ -36,22 +30,12 @@ export class TaskContextManager { return { ...this.contextAttributes, ...this.workerAttributes, - ...this.payloadAttributes, - [SemanticResourceAttributes.SERVICE_NAME]: this.ctx.task.id, }; } return {}; } - get payloadAttributes(): Attributes { - if (this.payload) { - return flattenAttributes(this.payload, "payload"); - } - - return {}; - } - get workerAttributes(): Attributes { if (this.worker) { return { @@ -104,10 +88,9 @@ export class TaskContextManager { export const taskContextManager = new TaskContextManager(); -import { Span, SpanProcessor } from "@opentelemetry/sdk-trace-base"; -import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions"; -import { SemanticInternalAttributes } from "../semanticInternalAttributes"; import { LogRecord, LogRecordProcessor } from "@opentelemetry/sdk-logs"; +import { Span, SpanProcessor } from "@opentelemetry/sdk-trace-base"; +import { SemanticInternalAttributes } from "../semanticInternalAttributes"; export class TaskContextSpanProcessor implements SpanProcessor { private _innerProcessor: SpanProcessor; diff --git a/packages/core/src/v3/utils/ioSerialization.ts b/packages/core/src/v3/utils/ioSerialization.ts new file mode 100644 index 000000000..3458b5468 --- /dev/null +++ b/packages/core/src/v3/utils/ioSerialization.ts @@ -0,0 +1,309 @@ +import { Attributes, Span } from "@opentelemetry/api"; +import { deserialize, parse, stringify } from "superjson"; +import { apiClientManager } from "../apiClient"; +import { OFFLOAD_IO_PACKET_LENGTH_LIMIT, imposeAttributeLimits } from "../limits"; +import { TaskRunExecutionResult } from "../schemas"; +import { SemanticInternalAttributes } from "../semanticInternalAttributes"; +import { TriggerTracer } from "../tracer"; +import { flattenAttributes } from "./flattenAttributes"; + +export type IOPacket = { + data?: string | undefined; + dataType: string; +}; + +export function parsePacket(value: IOPacket): any { + if (!value.data) { + return undefined; + } + + switch (value.dataType) { + case "application/json": + return JSON.parse(value.data); + case "application/super+json": + return parse(value.data); + case "text/plain": + return value.data; + default: + return value.data; + } +} + +export function stringifyIO(value: any): IOPacket { + if (value === undefined) { + return { dataType: "application/json" }; + } + + if (typeof value === "string") { + return { data: value, dataType: "text/plain" }; + } + + return { data: stringify(value), dataType: "application/super+json" }; +} + +export async function conditionallyExportPacket( + packet: IOPacket, + pathPrefix: string, + tracer?: TriggerTracer +): Promise { + if (apiClientManager.client) { + const { needsOffloading, size } = packetRequiresOffloading(packet); + + if (needsOffloading) { + if (!tracer) { + return await exportPacket(packet, pathPrefix); + } else { + const result = await tracer.startActiveSpan( + "store.uploadOutput", + async (span) => { + return await exportPacket(packet, pathPrefix); + }, + { + attributes: { + byteLength: size, + [SemanticInternalAttributes.STYLE_ICON]: "cloud-upload", + }, + } + ); + + return result ?? packet; + } + } + } + + return packet; +} + +export function packetRequiresOffloading(packet: IOPacket): { + needsOffloading: boolean; + size: number; +} { + if (!packet.data) { + return { + needsOffloading: false, + size: 0, + }; + } + + const byteSize = Buffer.byteLength(packet.data, "utf8"); + + return { + needsOffloading: byteSize >= OFFLOAD_IO_PACKET_LENGTH_LIMIT, + size: byteSize, + }; +} + +async function exportPacket(packet: IOPacket, pathPrefix: string): Promise { + // Offload the output + const filename = `${pathPrefix}.${getPacketExtension(packet.dataType)}`; + + const presignedResponse = await apiClientManager.client!.createUploadPayloadUrl(filename); + + if (presignedResponse.ok) { + const uploadResponse = await fetch(presignedResponse.data.presignedUrl, { + method: "PUT", + headers: { + "Content-Type": packet.dataType, + }, + body: packet.data, + }); + + if (!uploadResponse.ok) { + throw new Error( + `Failed to upload output to ${presignedResponse.data.presignedUrl}: ${uploadResponse.statusText}` + ); + } + + return { + data: filename, + dataType: "application/store", + }; + } + + return packet; +} + +export async function conditionallyImportPacket( + packet: IOPacket, + tracer?: TriggerTracer +): Promise { + if (packet.dataType !== "application/store") { + return packet; + } + + if (!tracer) { + return await importPacket(packet); + } else { + const result = await tracer.startActiveSpan( + "store.downloadPayload", + async (span) => { + return await importPacket(packet, span); + }, + { + attributes: { + [SemanticInternalAttributes.STYLE_ICON]: "cloud-download", + }, + } + ); + + return result ?? packet; + } +} + +async function importPacket(packet: IOPacket, span?: Span): Promise { + if (!packet.data) { + return packet; + } + + if (!apiClientManager.client) { + return packet; + } + + const presignedResponse = await apiClientManager.client.getPayloadUrl(packet.data); + + if (presignedResponse.ok) { + const response = await fetch(presignedResponse.data.presignedUrl); + + if (!response.ok) { + throw new Error( + `Failed to import packet ${presignedResponse.data.presignedUrl}: ${response.statusText}` + ); + } + + const data = await response.text(); + + span?.setAttribute("size", Buffer.byteLength(data, "utf8")); + + return { + data, + dataType: response.headers.get("content-type") ?? "application/json", + }; + } + + return packet; +} + +export function createPacketAttributes( + packet: IOPacket, + dataKey: string, + dataTypeKey: string +): Attributes { + if (!packet.data) { + return {}; + } + + switch (packet.dataType) { + case "application/json": + return { + ...flattenAttributes(packet, dataKey), + [dataTypeKey]: packet.dataType, + }; + case "application/super+json": + const parsed = parse(packet.data) as any; + const jsonified = JSON.parse(JSON.stringify(parsed, safeReplacer)); + + return { + ...flattenAttributes(jsonified, dataKey), + [dataTypeKey]: "application/json", + }; + case "application/store": + return { + [dataKey]: packet.data, + [dataTypeKey]: packet.dataType, + }; + case "text/plain": + return { + [SemanticInternalAttributes.OUTPUT]: packet.data, + [SemanticInternalAttributes.OUTPUT_TYPE]: packet.dataType, + }; + default: + return {}; + } +} + +export function createPackageAttributesAsJson(data: any, dataType: string): Attributes { + if ( + typeof data === "string" || + typeof data === "number" || + typeof data === "boolean" || + data === null || + data === undefined + ) { + return data; + } + + switch (dataType) { + case "application/json": + return imposeAttributeLimits(flattenAttributes(data, undefined)); + case "application/super+json": + const deserialized = deserialize(data) as any; + const jsonify = JSON.parse(JSON.stringify(deserialized, safeReplacer)); + + return imposeAttributeLimits(flattenAttributes(jsonify, undefined)); + case "application/store": + return data; + default: + return {}; + } +} + +export function prettyPrintPacket(rawData: any, dataType?: string): string { + if (rawData === undefined) { + return ""; + } + + if (dataType === "application/super+json") { + return prettyPrintPacket(deserialize(rawData), "application/json"); + } + + if (dataType === "application/json") { + return JSON.stringify(rawData, safeReplacer, 2); + } + + if (typeof rawData === "string") { + return rawData; + } + + return JSON.stringify(rawData, safeReplacer, 2); +} + +function safeReplacer(key: string, value: any) { + // If it is a BigInt + if (typeof value === "bigint") { + return value.toString(); // Convert to string + } + + // if it is a Regex + if (value instanceof RegExp) { + return value.toString(); // Convert to string + } + + // if it is a Set + if (value instanceof Set) { + return Array.from(value); // Convert to array + } + + // if it is a Map, convert it to an object + if (value instanceof Map) { + const obj: Record = {}; + value.forEach((v, k) => { + obj[k] = v; + }); + return obj; + } + + return value; // Otherwise return the value as is +} + +function getPacketExtension(outputType: string): string { + switch (outputType) { + case "application/json": + return "json"; + case "application/super+json": + return "json"; + case "text/plain": + return "txt"; + default: + return "txt"; + } +} diff --git a/packages/core/src/v3/workers/taskExecutor.ts b/packages/core/src/v3/workers/taskExecutor.ts index a93ada2a4..2b373d82d 100644 --- a/packages/core/src/v3/workers/taskExecutor.ts +++ b/packages/core/src/v3/workers/taskExecutor.ts @@ -1,4 +1,6 @@ import { SpanKind } from "@opentelemetry/api"; +import { ConsoleInterceptor } from "../consoleInterceptor"; +import { parseError } from "../errors"; import { TracingSDK, recordSpanException } from "../otel"; import { BackgroundWorkerProperties, @@ -10,14 +12,18 @@ import { TaskRunExecutionRetry, } from "../schemas"; import { SemanticInternalAttributes } from "../semanticInternalAttributes"; -import { HandleErrorFunction, ProjectConfig, TaskMetadataWithFunctions } from "../types"; -import { flattenAttributes } from "../utils/flattenAttributes"; -import { accessoryAttributes } from "../utils/styleAttributes"; -import { calculateNextRetryDelay } from "../utils/retries"; import { taskContextManager } from "../tasks/taskContextManager"; import { TriggerTracer } from "../tracer"; -import { ConsoleInterceptor } from "../consoleInterceptor"; -import { parseError } from "../errors"; +import { HandleErrorFunction, ProjectConfig, TaskMetadataWithFunctions } from "../types"; +import { + conditionallyExportPacket, + conditionallyImportPacket, + createPacketAttributes, + parsePacket, + stringifyIO, +} from "../utils/ioSerialization"; +import { calculateNextRetryDelay } from "../utils/retries"; +import { accessoryAttributes } from "../utils/styleAttributes"; export type TaskExecutorOptions = { tracingSDK: TracingSDK; @@ -53,14 +59,17 @@ export class TaskExecutor { worker: BackgroundWorkerProperties, traceContext: Record ): Promise { - const parsedPayload = JSON.parse(execution.run.payload); const ctx = TaskRunContext.parse(execution); const attemptMessage = `Attempt ${execution.attempt.number}`; + const originalPacket = { + data: execution.run.payload, + dataType: execution.run.payloadType, + }; + const result = await taskContextManager.runWith( { ctx, - payload: parsedPayload, worker, }, async () => { @@ -74,21 +83,40 @@ export class TaskExecutor { attemptMessage, async (span) => { return await this._consoleInterceptor.intercept(console, async () => { - const init = await this.#callTaskInit(parsedPayload, ctx); + let parsedPayload: any; + let initOutput: any; try { - const output = await this.#callRun(parsedPayload, ctx, init); + const payloadPacket = await conditionallyImportPacket(originalPacket, this._tracer); + + parsedPayload = parsePacket(payloadPacket); + + initOutput = await this.#callTaskInit(parsedPayload, ctx); + + const output = await this.#callRun(parsedPayload, ctx, initOutput); try { - span.setAttributes(flattenAttributes(output, SemanticInternalAttributes.OUTPUT)); + const stringifiedOutput = stringifyIO(output); - const serializedOutput = JSON.stringify(output); + const finalOutput = await conditionallyExportPacket( + stringifiedOutput, + `${execution.attempt.id}/output`, + this._tracer + ); + + span.setAttributes( + createPacketAttributes( + finalOutput, + SemanticInternalAttributes.OUTPUT, + SemanticInternalAttributes.OUTPUT_TYPE + ) + ); return { ok: true, id: execution.attempt.id, - output: serializedOutput, - outputType: "application/json", + output: finalOutput.data, + outputType: finalOutput.dataType, } satisfies TaskRunExecutionResult; } catch (stringifyError) { recordSpanException(span, stringifyError); @@ -148,7 +176,7 @@ export class TaskExecutor { } satisfies TaskRunExecutionResult; } } finally { - await this.#callTaskCleanup(parsedPayload, ctx, init); + await this.#callTaskCleanup(parsedPayload, ctx, initOutput); } }); }, @@ -156,7 +184,6 @@ export class TaskExecutor { kind: SpanKind.CONSUMER, attributes: { [SemanticInternalAttributes.STYLE_ICON]: "attempt", - ...flattenAttributes(parsedPayload, SemanticInternalAttributes.PAYLOAD), ...accessoryAttributes({ items: [ { diff --git a/packages/database/prisma/migrations/20240325224419_add_output_type_to_task_events/migration.sql b/packages/database/prisma/migrations/20240325224419_add_output_type_to_task_events/migration.sql new file mode 100644 index 000000000..3a17adb87 --- /dev/null +++ b/packages/database/prisma/migrations/20240325224419_add_output_type_to_task_events/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "TaskEvent" ADD COLUMN "outputType" TEXT; diff --git a/packages/database/prisma/migrations/20240326145956_add_payload_columns_to_task_event/migration.sql b/packages/database/prisma/migrations/20240326145956_add_payload_columns_to_task_event/migration.sql new file mode 100644 index 000000000..78e1893d3 --- /dev/null +++ b/packages/database/prisma/migrations/20240326145956_add_payload_columns_to_task_event/migration.sql @@ -0,0 +1,3 @@ +-- AlterTable +ALTER TABLE "TaskEvent" ADD COLUMN "payload" JSONB, +ADD COLUMN "payloadType" TEXT; diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index dd075354f..6b7483d46 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -1811,6 +1811,12 @@ model TaskEvent { /// This represents all span attributes in the $output namespace, like $output output Json? + /// This represents the mimetype of the output, such as application/json or application/super+json + outputType String? + + payload Json? + payloadType String? + createdAt DateTime @default(now()) } diff --git a/packages/trigger-sdk/src/v3/shared.ts b/packages/trigger-sdk/src/v3/shared.ts index 188317668..db65d84f7 100644 --- a/packages/trigger-sdk/src/v3/shared.ts +++ b/packages/trigger-sdk/src/v3/shared.ts @@ -1,5 +1,9 @@ import { SpanKind } from "@opentelemetry/api"; -import { SemanticAttributes } from "@opentelemetry/semantic-conventions"; +import { + SEMATTRS_MESSAGING_DESTINATION, + SEMATTRS_MESSAGING_OPERATION, + SEMATTRS_MESSAGING_SYSTEM, +} from "@opentelemetry/semantic-conventions"; import { HandleErrorFnParams, HandleErrorResult, @@ -12,11 +16,13 @@ import { SemanticInternalAttributes, SuccessFnParams, TaskRunContext, + TaskRunExecutionResult, accessoryAttributes, apiClientManager, + conditionallyImportPacket, createErrorTaskError, defaultRetryOptions, - flattenAttributes, + parsePacket, runtime, taskContextManager, } from "@trigger.dev/core/v3"; @@ -240,13 +246,12 @@ export function createTask( { kind: SpanKind.PRODUCER, attributes: { - [SemanticAttributes.MESSAGING_OPERATION]: "publish", + [SEMATTRS_MESSAGING_OPERATION]: "publish", [SemanticInternalAttributes.STYLE_ICON]: "trigger", ["messaging.client_id"]: taskContextManager.worker?.id, - [SemanticAttributes.MESSAGING_DESTINATION]: params.queue?.name ?? params.id, + [SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id, ["messaging.message.body.size"]: JSON.stringify(payload).length, - [SemanticAttributes.MESSAGING_SYSTEM]: "trigger.dev", - ...flattenAttributes(payload as any, SemanticInternalAttributes.PAYLOAD), + [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev", ...(taskMetadata ? accessoryAttributes({ items: [ @@ -302,14 +307,14 @@ export function createTask( { kind: SpanKind.PRODUCER, attributes: { - [SemanticAttributes.MESSAGING_OPERATION]: "publish", + [SEMATTRS_MESSAGING_OPERATION]: "publish", ["messaging.batch.message_count"]: items.length, ["messaging.client_id"]: taskContextManager.worker?.id, - [SemanticAttributes.MESSAGING_DESTINATION]: params.queue?.name ?? params.id, + [SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id, ["messaging.message.body.size"]: items .map((item) => JSON.stringify(item.payload)) .join("").length, - [SemanticAttributes.MESSAGING_SYSTEM]: "trigger.dev", + [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev", [SemanticInternalAttributes.STYLE_ICON]: "trigger", ...(taskMetadata ? accessoryAttributes({ @@ -347,7 +352,7 @@ export function createTask( taskMetadata ? "Trigger" : `${params.id} triggerAndWait()`, async (span) => { const response = await apiClient.triggerTask(params.id, { - payload: payload, + payload, options: { dependentAttempt: ctx.attempt.id, lockToVersion: taskContextManager.worker?.version, // Lock to current version because we're waiting for it to finish @@ -368,22 +373,22 @@ export function createTask( ctx, }); - if (!result.ok) { - throw createErrorTaskError(result.error); + const runResult = await handleTaskRunExecutionResult(result); + + if (!runResult.ok) { + throw runResult.error; } - return typeof result.output === "string" ? JSON.parse(result.output) : result.output; + return runResult.output; }, { kind: SpanKind.PRODUCER, attributes: { [SemanticInternalAttributes.STYLE_ICON]: "trigger", - [SemanticAttributes.MESSAGING_OPERATION]: "publish", + [SEMATTRS_MESSAGING_OPERATION]: "publish", ["messaging.client_id"]: taskContextManager.worker?.id, - [SemanticAttributes.MESSAGING_DESTINATION]: params.queue?.name ?? params.id, - ["messaging.message.body.size"]: JSON.stringify(payload).length, - [SemanticAttributes.MESSAGING_SYSTEM]: "trigger.dev", - ...flattenAttributes(payload as any, SemanticInternalAttributes.PAYLOAD), + [SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id, + [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev", ...(taskMetadata ? accessoryAttributes({ items: [ @@ -442,21 +447,7 @@ export function createTask( ctx, }); - const runs = result.items.map((item) => { - if (item.ok) { - return { - ok: true, - id: item.id, - output: typeof item.output === "string" ? JSON.parse(item.output) : item.output, - } satisfies TaskRunResult; - } else { - return { - ok: false, - id: item.id, - error: createErrorTaskError(item.error), - } satisfies TaskRunResult; - } - }); + const runs = await handleBatchTaskRunExecutionResult(result.items); return { id: result.id, @@ -466,14 +457,14 @@ export function createTask( { kind: SpanKind.PRODUCER, attributes: { - [SemanticAttributes.MESSAGING_OPERATION]: "publish", + [SEMATTRS_MESSAGING_OPERATION]: "publish", ["messaging.batch.message_count"]: items.length, ["messaging.client_id"]: taskContextManager.worker?.id, - [SemanticAttributes.MESSAGING_DESTINATION]: params.queue?.name ?? params.id, + [SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id, ["messaging.message.body.size"]: items .map((item) => JSON.stringify(item.payload)) .join("").length, - [SemanticAttributes.MESSAGING_SYSTEM]: "trigger.dev", + [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev", [SemanticInternalAttributes.STYLE_ICON]: "trigger", ...(taskMetadata ? accessoryAttributes({ @@ -511,3 +502,59 @@ export function createTask( return task; } + +async function handleBatchTaskRunExecutionResult( + items: Array +): Promise>> { + const someObjectStoreOutputs = items.some( + (item) => item.ok && item.outputType === "application/store" + ); + + if (!someObjectStoreOutputs) { + const results = await Promise.all( + items.map(async (item) => { + return await handleTaskRunExecutionResult(item); + }) + ); + + return results; + } + + return await tracer.startActiveSpan( + "store.downloadPayloads", + async (span) => { + const results = await Promise.all( + items.map(async (item) => { + return await handleTaskRunExecutionResult(item); + }) + ); + + return results; + }, + { + kind: SpanKind.INTERNAL, + [SemanticInternalAttributes.STYLE_ICON]: "cloud-download", + } + ); +} + +async function handleTaskRunExecutionResult( + execution: TaskRunExecutionResult +): Promise> { + if (execution.ok) { + const outputPacket = { data: execution.output, dataType: execution.outputType }; + const importedPacket = await conditionallyImportPacket(outputPacket, tracer); + + return { + ok: true, + id: execution.id, + output: parsePacket(importedPacket), + }; + } else { + return { + ok: false, + id: execution.id, + error: createErrorTaskError(execution.error), + }; + } +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 001e9f118..e98c52e47 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -242,6 +242,7 @@ importers: '@upstash/ratelimit': ^1.0.1 '@whatwg-node/fetch': ^0.9.14 autoprefixer: ^10.4.13 + aws4fetch: ^1.0.18 babel-loader: ^9.1.3 babel-preset-react-app: ^10.0.1 class-variance-authority: ^0.5.2 @@ -309,6 +310,7 @@ importers: sonner: ^1.0.3 sqs-consumer: ^7.4.0 style-loader: ^3.3.4 + superjson: ^2.2.1 tailwind-merge: ^1.12.0 tailwind-scrollbar: ^3.0.1 tailwind-scrollbar-hide: ^1.1.7 @@ -393,6 +395,7 @@ importers: '@uiw/react-codemirror': 4.19.5_th22fcplkuhrqjnlojwclcaim4 '@upstash/ratelimit': 1.0.1 '@whatwg-node/fetch': 0.9.14 + aws4fetch: 1.0.18 class-variance-authority: 0.5.2_typescript@5.2.2 clsx: 1.2.1 compression: 1.7.4 @@ -444,6 +447,7 @@ importers: socket.io: 4.7.4 sonner: 1.0.3_biqbaboplfbrettd7655fr4n2y sqs-consumer: 7.5.0_hzguu36iioy52ghs7zxwltx2ia + superjson: 2.2.1 tailwind-merge: 1.12.0 tailwind-scrollbar-hide: 1.1.7 tailwindcss-animate: 1.0.5_tailwindcss@3.4.1 @@ -1186,6 +1190,7 @@ importers: rimraf: ^3.0.2 socket.io: ^4.7.4 socket.io-client: ^4.7.4 + superjson: ^2.2.1 ts-jest: ^29.1.1 tsup: ^8.0.1 typescript: ^5.3.0 @@ -1208,6 +1213,7 @@ importers: humanize-duration: 3.27.3 socket.io: 4.7.4 socket.io-client: 4.7.4 + superjson: 2.2.1 ulidx: 2.2.1 zod: 3.22.3 zod-error: 1.5.0 @@ -14783,6 +14789,10 @@ packages: resolution: {integrity: sha512-NmWvPnx0F1SfrQbYwOi7OeaNGokp9XhzNioJ/CSBs8Qa4vxug81mhJEAVZwxXuBmYB5KDRfMq/F3RR0BIU7sWg==} dev: false + /aws4fetch/1.0.18: + resolution: {integrity: sha512-3Cf+YaUl07p24MoQ46rFwulAmiyCwH2+1zw1ZyPAX5OtJ34Hh185DwB8y/qRLb6cYYYtSFJ9pthyLc0MD4e8sQ==} + dev: false + /axe-core/4.6.2: resolution: {integrity: sha512-b1WlTV8+XKLj9gZy2DZXgQiyDp9xkkoe2a6U6UbYccScq2wgH/YwCeI2/Jq2mgo0HzQxqJOjWZBLeA/mqsk5Mg==} engines: {node: '>=4'} @@ -16156,6 +16166,13 @@ packages: resolution: {integrity: sha512-LDx6oHrK+PhzLKJU9j5S7/Y3jM/mUHvD/DeI1WQmJn652iPC5Y4TBzC9l+5OMOXlyTTA+SmVUPm0HQUwpD5Jqw==} dev: true + /copy-anything/3.0.5: + resolution: {integrity: sha512-yCEafptTtb4bk7GLEQoM8KVJpxAfdBJYaXyzQEgQQQgYrZiDp8SJmGKlYza6CYjEDNstAdNdKA3UuoULlEbS6w==} + engines: {node: '>=12.13'} + dependencies: + is-what: 4.1.16 + dev: false + /copy-descriptor/0.1.1: resolution: {integrity: sha512-XgZ0pFcakEUlbwQEVNg3+QAis1FyTL3Qel9FYy8pSkQqoG3PNoT0bOCQtOXcOkur21r2Eq2kI+IE+gsmAEVlYw==} engines: {node: '>=0.10.0'} @@ -21519,6 +21536,11 @@ packages: get-intrinsic: 1.1.3 dev: true + /is-what/4.1.16: + resolution: {integrity: sha512-ZhMwEosbFJkA0YhFnNDgTM4ZxDRsS6HqTo7qsZM08fehyRYIYa0yHu5R6mgo1n/8MgaPBXiPimPD77baVFYg+A==} + engines: {node: '>=12.13'} + dev: false + /is-whitespace/0.3.0: resolution: {integrity: sha512-RydPhl4S6JwAyj0JJjshWJEFG6hNye3pZFBRZaTUfZFwGHxzppNaNOVgQuS/E/SlhrApuMXrpnK1EEIXfdo3Dg==} engines: {node: '>=0.10.0'} @@ -29323,6 +29345,13 @@ packages: - supports-color dev: true + /superjson/2.2.1: + resolution: {integrity: sha512-8iGv75BYOa0xRJHK5vRLEjE2H/i4lulTjzpUXic3Eg8akftYjkmQDa8JARQ42rlczXyFR3IeRoeFCc7RxHsYZA==} + engines: {node: '>=16'} + dependencies: + copy-anything: 3.0.5 + dev: false + /supertest/6.3.3: resolution: {integrity: sha512-EMCG6G8gDu5qEqRQ3JjjPs6+FYT1a7Hv5ApHvtSghmOFJYtsU5S+pSb6Y2EUeCEY3CmEL3mmQ8YWlPOzQomabA==} engines: {node: '>=6.4.0'} diff --git a/references/v3-catalog/src/trigger/superjson.ts b/references/v3-catalog/src/trigger/superjson.ts new file mode 100644 index 000000000..a768fa5b5 --- /dev/null +++ b/references/v3-catalog/src/trigger/superjson.ts @@ -0,0 +1,99 @@ +import { logger, task } from "@trigger.dev/sdk/v3"; + +export const superParentTask = task({ + id: "super-parent-task", + run: async () => { + const result = await superChildTask.triggerAndWait({ + payload: { + foo: "bar", + }, + }); + + logger.log(`typeof result.date = ${typeof result.date}`); + logger.log(`typeof result.regex = ${typeof result.regex}`); + logger.log(`typeof result.bigint = ${typeof result.bigint}`); + logger.log(`typeof result.set = ${typeof result.set}`); + logger.log(`typeof result.map = ${typeof result.map}`); + logger.log(`typeof result.error = ${typeof result.error}`); + logger.log(`typeof result.url = ${typeof result.url}`); + + return { + result, + }; + }, +}); + +export const superChildTask = task({ + id: "super-child-task", + run: async () => { + return { + date: new Date(), + regex: /foo/, + bigint: BigInt(123), + set: new Set([1, 2, 3, 4, 5, 6, 7, 8, 9]), + map: new Map([ + ["foo", "bar"], + ["baz", "qux"], + ]), + error: new Error("foo"), + url: new URL("https://trigger.dev"), + }; + }, +}); + +export const superHugePayloadTask = task({ + id: "super-huge-payload-task", + run: async () => { + const largePayload = createLargeObject(1000, 100); + + const result = await superHugeOutputTask.triggerAndWait({ + payload: largePayload, + }); + + logger.log("Result from superHugeOutputTask: ", { result }); + + const batchResult = await superHugeOutputTask.batchTriggerAndWait({ + items: [ + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + { payload: largePayload }, + ], + }); + + logger.log("Result from superHugeOutputTask batchTriggerAndWait: ", { batchResult }); + + return { + result, + }; + }, +}); + +export const superHugeOutputTask = task({ + id: "super-huge-output-task", + run: async () => { + return createLargeObject(1000, 100); + }, +}); + +function createLargeObject(i: number, length: number) { + return Array.from({ length }, (_, i) => [i.toString(), i.toString().padStart(i, "0")]).reduce( + (acc, [key, value]) => { + acc[key] = value; + return acc; + }, + {} as Record + ); +}