v3: superjson output support (triggerdotdev#971)

* superjson output support

* Better support for superjson in the task events, limiting output attributes

* Offload large outputs to object store (r2)

* Finishing up the offloading of large outputs/payloads to an object store

Author: ericallam

.env.example

@@ -59,4 +59,8 @@ COORDINATOR_SECRET=coordinator-secret # generate the actual secret with `openssl
 # CONTAINER_REGISTRY_ORIGIN=<Container registry origin e.g. https://registry.digitalocean.com>
 # CONTAINER_REGISTRY_USERNAME=<Container registry username e.g. Digital ocean email address>
 # CONTAINER_REGISTRY_PASSWORD=<Container registry password e.g. Digital ocean PAT>
-# DEV_OTEL_EXPORTER_OTLP_ENDPOINT="http://0.0.0.0:4318"
+# 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=

.nvmrc

@@ -1 +1 @@
-v18.18.0
+v20.11.1

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 (
     <div className="flex items-center justify-between px-4">
       <Paragraph variant="base/bright" className="w-full border-b border-grid-dimmed py-2.5">

apps/webapp/app/components/primitives/Buttons.tsx

@@ -285,7 +285,7 @@ export const Button = forwardRef<HTMLButtonElement, ButtonPropsType>(
 
 type LinkPropsType = Pick<
   LinkProps,
-  "to" | "target" | "onClick" | "onMouseDown" | "onMouseEnter" | "onMouseLeave"
+  "to" | "target" | "onClick" | "onMouseDown" | "onMouseEnter" | "onMouseLeave" | "download"
 > &
   React.ComponentProps<typeof ButtonContent>;
 export const LinkButton = ({
@@ -294,6 +294,7 @@ export const LinkButton = ({
   onMouseDown,
   onMouseEnter,
   onMouseLeave,
+  download,
   ...props
 }: LinkPropsType) => {
   const innerRef = useRef<HTMLAnchorElement>(null);
@@ -308,7 +309,7 @@ export const LinkButton = ({
     });
   }
 
-  if (to.toString().startsWith("http")) {
+  if (to.toString().startsWith("http") || to.toString().startsWith("/resources")) {
     return (
       <ExtLink
         href={to.toString()}
@@ -318,6 +319,7 @@ export const LinkButton = ({
         onMouseDown={onMouseDown}
         onMouseEnter={onMouseEnter}
         onMouseLeave={onMouseLeave}
+        download={download}
       >
         <ButtonContent {...props} />
       </ExtLink>
@@ -332,6 +334,7 @@ export const LinkButton = ({
         onMouseDown={onMouseDown}
         onMouseEnter={onMouseEnter}
         onMouseLeave={onMouseLeave}
+        download={download}
       >
         <ButtonContent {...props} />
       </Link>

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<typeof EnvironmentSchema>;

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,
       },

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;`;
 

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 && <SpanEvents spanEvents={event.events} />}
           {event.payload !== undefined && (
-            <CodeBlock rowTitle="Payload" code={event.payload} maxLines={20} />
+            <PacketDisplay data={event.payload} dataType={event.payloadType} title="Payload" />
           )}
           {event.output !== undefined && (
-            <CodeBlock rowTitle="Output" code={event.output} maxLines={20} />
+            <PacketDisplay data={event.output} dataType={event.outputType} title="Output" />
           )}
           {event.properties !== undefined && (
             <CodeBlock rowTitle="Properties" code={event.properties} maxLines={20} />
@@ -204,6 +203,31 @@ export default function Page() {
   );
 }
 
+function PacketDisplay({
+  data,
+  dataType,
+  title,
+}: {
+  data: string;
+  dataType: string;
+  title: string;
+}) {
+  if (dataType === "application/store") {
+    return (
+      <div className="flex flex-col">
+        <Paragraph variant="base/bright" className="w-full border-b border-grid-dimmed py-2.5">
+          {title}
+        </Paragraph>
+        <LinkButton LeadingIcon={CloudArrowDownIcon} to={data} variant="tertiary/medium" download>
+          Download
+        </LinkButton>
+      </div>
+    );
+  } else {
+    return <CodeBlock rowTitle={title} code={data} maxLines={20} />;
+  }
+}
+
 type TimelineProps = {
   startTime: Date;
   duration: number;

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 });
+}

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") {

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 });

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)}"`,
+    },
+  });
+}