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

Merge branch 'main' into v3/infra-updates

Browse Source
This commit is contained in:
nicktrn
2024-03-19 19:22:27 +00:00
parent 5ff738a18c 0e347b001b
commit a320f17bc4
44 changed files with 791 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/e2e.yml
+1 -1
View File
@@ -4,7 +4,7 @@ on:
jobs:
e2e:
name: "🧪 E2E Tests"
runs-on: buildjet-4vcpu-ubuntu-2204
runs-on: buildjet-16vcpu-ubuntu-2204
steps:
- name: 🐳 Login to Docker Hub
uses: docker/login-action@v2
.github/workflows/publish-docker.yml
+3
View File
@@ -8,6 +8,9 @@ jobs:
version: ${{ steps.get_version.outputs.version }}
short_sha: ${{ steps.get_commit.outputs.sha_short }}
steps:
- name: Setup Depot CLI
uses: depot/setup-action@v1
- name: ⬇️ Checkout repo
uses: actions/checkout@v3
with:
.github/workflows/publish-infra.yml
+1 -1
View File
@@ -40,7 +40,7 @@ jobs:
strategy:
matrix:
package: [coordinator, kubernetes-provider]
runs-on: buildjet-4vcpu-ubuntu-2204
runs-on: buildjet-16vcpu-ubuntu-2204
env:
DOCKER_BUILDKIT: "1"
steps:
.github/workflows/release.yml
+1 -1
View File
@@ -12,7 +12,7 @@ on:
jobs:
release:
name: 🦋 Changesets Release
runs-on: buildjet-4vcpu-ubuntu-2204
runs-on: buildjet-16vcpu-ubuntu-2204
if: |
github.repository == 'triggerdotdev/trigger.dev'
outputs:
.github/workflows/typecheck.yml
+1 -1
View File
@@ -3,7 +3,7 @@ on:
workflow_call:
jobs:
typecheck:
runs-on: buildjet-4vcpu-ubuntu-2204
runs-on: buildjet-16vcpu-ubuntu-2204
steps:
- name: ⬇️ Checkout repo
.github/workflows/unit-tests.yml
+1 -1
View File
@@ -4,7 +4,7 @@ on:
jobs:
unitTests:
name: "🧪 Unit Tests"
runs-on: buildjet-4vcpu-ubuntu-2204
runs-on: buildjet-16vcpu-ubuntu-2204
steps:
- name: ⬇️ Checkout repo
uses: actions/checkout@v3
apps/coordinator/tsconfig.json
+1
View File
@@ -5,6 +5,7 @@
"target": "es2016",
"module": "commonjs",
"esModuleInterop": true,
"resolveJsonModule": true,
"forceConsistentCasingInFileNames": true,
"strict": true,
"skipLibCheck": true,
apps/docker-provider/tsconfig.json
+1
View File
@@ -4,6 +4,7 @@
"module": "commonjs",
"esModuleInterop": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true,
"strict": true,
"skipLibCheck": true,
"paths": {
apps/webapp/app/components/SetupCommands.tsx
+7 -6
View File
@@ -131,6 +131,7 @@ export function TriggerDevStep({ extra }: { extra?: string }) {
}
// Trigger.dev version 3 setup commands
const v3PackageTag = "v3";
export function InitCommandV3() {
const project = useProject();
@@ -147,7 +148,7 @@ export function InitCommandV3() {
variant="primary/medium"
iconButton
className="mb-4"
value={`npx trigger.dev@latest init -p ${projectRef}`}
value={`npx trigger.dev@${v3PackageTag} init -p ${projectRef}`}
/>
</ClientTabsContent>
<ClientTabsContent value={"pnpm"}>
@@ -155,7 +156,7 @@ export function InitCommandV3() {
variant="primary/medium"
iconButton
className="mb-4"
value={`pnpm dlx trigger.dev@latest init -p ${projectRef}`}
value={`pnpm dlx trigger.dev@${v3PackageTag} init -p ${projectRef}`}
/>
</ClientTabsContent>
<ClientTabsContent value={"yarn"}>
@@ -163,7 +164,7 @@ export function InitCommandV3() {
variant="primary/medium"
iconButton
className="mb-4"
value={`yarn dlx trigger.dev@latest init -p ${projectRef}`}
value={`yarn dlx trigger.dev@${v3PackageTag} init -p ${projectRef}`}
/>
</ClientTabsContent>
</ClientTabs>
@@ -183,7 +184,7 @@ export function TriggerDevStepV3() {
variant="primary/medium"
iconButton
className="mb-4"
value={`npx trigger.dev@latest dev`}
value={`npx trigger.dev@${v3PackageTag} dev`}
/>
</ClientTabsContent>
<ClientTabsContent value={"pnpm"}>
@@ -191,7 +192,7 @@ export function TriggerDevStepV3() {
variant="primary/medium"
iconButton
className="mb-4"
value={`pnpm dlx trigger.dev@latest dev`}
value={`pnpm dlx trigger.dev@${v3PackageTag} dev`}
/>
</ClientTabsContent>
<ClientTabsContent value={"yarn"}>
@@ -199,7 +200,7 @@ export function TriggerDevStepV3() {
variant="primary/medium"
iconButton
className="mb-4"
value={`yarn dlx trigger.dev@latest dev`}
value={`yarn dlx trigger.dev@${v3PackageTag} dev`}
/>
</ClientTabsContent>
</ClientTabs>
apps/webapp/app/components/primitives/TreeView/TreeView.tsx
+4 -1
View File
@@ -23,6 +23,9 @@ export type TreeViewProps<TData> = {
onScroll?: (scrollTop: number) => void;
} & Pick<UseTreeStateOutput, "getTreeProps" | "getNodeProps">;
export type GetTreePropsFn = UseTreeStateOutput["getTreeProps"];
export type GetNodePropsFn = UseTreeStateOutput["getNodeProps"];
export function TreeView<TData>({
tree,
renderNode,
@@ -144,7 +147,7 @@ type HTMLAttributes = Omit<
"onAnimationStart" | "onDragStart" | "onDragEnd" | "onDrag"
>;
type UseTreeStateOutput = {
export type UseTreeStateOutput = {
selected: string | undefined;
nodes: NodesState;
virtualizer: Virtualizer<HTMLElement, Element>;
apps/webapp/app/presenters/v3/RunPresenter.server.ts
+11 -13
View File
@@ -79,25 +79,22 @@ export class RunPresenter {
n.data.startTime.getTime() - treeRootStartTimeMs
);
totalDuration = Math.max(totalDuration, offset + n.data.duration);
return { ...n, data: { ...n.data, offset, isRoot: n.id === traceSummary.rootSpan.id } };
return {
...n,
data: {
...n.data,
//set partial nodes to null duration
duration: n.data.isPartial ? null : n.data.duration,
offset,
isRoot: n.id === traceSummary.rootSpan.id,
},
};
})
: [];
//if any elements are partial we want the total duration to represent all of the time until now
totalDuration = events.some((e) => e.data.isPartial)
? millisecondsToNanoseconds(Date.now() - treeRootStartTimeMs)
: totalDuration;
//total duration should be a minimum of 1ms
totalDuration = Math.max(totalDuration, millisecondsToNanoseconds(1));
//we need to adjust any partial nodes so they run the full duration
for (const event of events) {
if (event.data.isPartial) {
event.data.duration = totalDuration - event.data.offset;
}
}
let rootSpanStatus: "executing" | "completed" | "failed" = "executing";
if (events[0]) {
if (events[0].data.isError) {
@@ -123,6 +120,7 @@ export class RunPresenter {
parentRunFriendlyId:
tree?.id === traceSummary.rootSpan.id ? undefined : traceSummary.rootSpan.runId,
duration: totalDuration,
rootStartedAt: tree?.data.startTime,
};
}
}
apps/webapp/app/presenters/v3/TasksStreamPresenter.server.ts
+115
View File
@@ -0,0 +1,115 @@
import { TaskRun, TaskRunAttempt } from "@trigger.dev/database";
import { eventStream } from "remix-utils/sse/server";
import { PrismaClient, prisma } from "~/db.server";
import { logger } from "~/services/logger.server";
import { eventRepository } from "~/v3/eventRepository.server";
import { projectPubSub } from "~/v3/services/projectPubSub.server";
type RunWithAttempts = {
updatedAt: Date;
attempts: {
status: TaskRunAttempt["status"];
updatedAt: Date;
}[];
};
const pingInterval = 1000;
export class TasksStreamPresenter {
#prismaClient: PrismaClient;
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
public async call({
request,
organizationSlug,
projectSlug,
userId,
}: {
request: Request;
organizationSlug: string;
projectSlug: string;
userId: string;
}) {
const project = await this.#prismaClient.project.findUnique({
where: {
slug: projectSlug,
organization: {
slug: organizationSlug,
members: {
some: {
userId,
},
},
},
},
select: {
id: true,
},
});
if (!project) {
return new Response("Not found", { status: 404 });
}
logger.info("TasksStreamPresenter.call", {
projectSlug,
});
let pinger: NodeJS.Timer | undefined = undefined;
const subscriber = await projectPubSub.subscribe(`project:${project.id}:*`);
return eventStream(request.signal, (send, close) => {
const safeSend = (args: { event?: string; data: string }) => {
try {
send(args);
} catch (error) {
if (error instanceof Error) {
if (error.name !== "TypeError") {
logger.debug("Error sending SSE, aborting", {
error: {
name: error.name,
message: error.message,
stack: error.stack,
},
args,
});
}
} else {
logger.debug("Unknown error sending SSE, aborting", {
error,
args,
});
}
close();
}
};
subscriber.on("WORKER_CREATED", async (message) => {
safeSend({ data: message.createdAt.toISOString() });
});
pinger = setInterval(() => {
if (request.signal.aborted) {
return close();
}
safeSend({ event: "ping", data: new Date().toISOString() });
}, pingInterval);
return async function clear() {
logger.info("TasksStreamPresenter.abort", {
projectSlug,
});
clearInterval(pinger);
await subscriber.stopListening();
};
});
}
}
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam._index/route.tsx
+17 -3
View File
@@ -1,6 +1,8 @@
import { ChatBubbleLeftRightIcon } from "@heroicons/react/20/solid";
import { useRevalidator } from "@remix-run/react";
import { LoaderFunctionArgs } from "@remix-run/server-runtime";
import { TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
import { useEffect } from "react";
import { typedjson, useTypedLoaderData } from "remix-typedjson";
import invariant from "tiny-invariant";
import { Feedback } from "~/components/Feedback";
@@ -28,13 +30,14 @@ import {
import { TaskFunctionName, TaskPath } from "~/components/runs/v3/TaskPath";
import { TaskRunStatusCombo } from "~/components/runs/v3/TaskRunStatus";
import { useDevEnvironment } from "~/hooks/useEnvironments";
import { useEventSource } from "~/hooks/useEventSource";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { useUser } from "~/hooks/useUser";
import { TaskListPresenter } from "~/presenters/v3/TaskListPresenter.server";
import { requireUserId } from "~/services/session.server";
import { cn } from "~/utils/cn";
import { ProjectParamSchema, v3RunsPath } from "~/utils/pathBuilder";
import { ProjectParamSchema, v3RunsPath, v3TasksStreamingPath } from "~/utils/pathBuilder";
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
@@ -67,6 +70,19 @@ export default function Page() {
const { tasks } = useTypedLoaderData<typeof loader>();
const hasTasks = tasks.length > 0;
//live reload the page when the tasks change
const revalidator = useRevalidator();
const streamedEvents = useEventSource(v3TasksStreamingPath(organization, project), {
event: "message",
});
useEffect(() => {
if (streamedEvents !== null) {
revalidator.revalidate();
}
// WARNING Don't put the revalidator in the useEffect deps array or bad things will happen
}, [streamedEvents]); // eslint-disable-line react-hooks/exhaustive-deps
return (
<PageContainer>
<NavBar>
@@ -179,8 +195,6 @@ function classForTaskRunStatus(status: TaskRunStatus) {
}
function CreateTaskInstructions() {
const devEnvironment = useDevEnvironment();
invariant(devEnvironment, "Dev environment must be defined");
return (
<MainCenteredContainer className="max-w-prose">
<div className="mb-6 flex items-center justify-between border-b">
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam/route.tsx
+272 -185
View File
@@ -4,9 +4,15 @@ import {
MagnifyingGlassMinusIcon,
MagnifyingGlassPlusIcon,
} from "@heroicons/react/20/solid";
import { Link, Outlet, useNavigate, useParams, useRevalidator } from "@remix-run/react";
import { Outlet, useNavigate, useParams, useRevalidator } from "@remix-run/react";
import { LoaderFunctionArgs } from "@remix-run/server-runtime";
import { formatDurationMilliseconds, nanosecondsToMilliseconds } from "@trigger.dev/core/v3";
import { Virtualizer } from "@tanstack/react-virtual";
import {
formatDurationMilliseconds,
millisecondsToNanoseconds,
nanosecondsToMilliseconds,
} from "@trigger.dev/core/v3";
import { motion } from "framer-motion";
import { useEffect, useRef, useState } from "react";
import { typedjson, useTypedLoaderData } from "remix-typedjson";
import { ShowParentIcon, ShowParentIconSelected } from "~/assets/icons/ShowParentIcon";
@@ -26,7 +32,8 @@ import {
import { Slider } from "~/components/primitives/Slider";
import { Switch } from "~/components/primitives/Switch";
import * as Timeline from "~/components/primitives/Timeline";
import { TreeView, useTree } from "~/components/primitives/TreeView/TreeView";
import { TreeView, UseTreeStateOutput, useTree } from "~/components/primitives/TreeView/TreeView";
import { NodesState } from "~/components/primitives/TreeView/reducer";
import { RunIcon } from "~/components/runs/v3/RunIcon";
import { SpanTitle, eventBackgroundClassName } from "~/components/runs/v3/SpanTitle";
import { TaskRunStatusIcon, runStatusClassNameColor } from "~/components/runs/v3/TaskRunStatus";
@@ -55,7 +62,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const { projectParam, organizationSlug, runParam } = v3RunParamsSchema.parse(params);
const presenter = new RunPresenter();
const { run, events, parentRunFriendlyId, duration, rootSpanStatus } = await presenter.call({
const result = await presenter.call({
userId,
organizationSlug,
projectSlug: projectParam,
@@ -66,12 +73,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const resizeSettings = await getResizableRunSettings(request);
return typedjson({
run,
events,
parentRunFriendlyId,
...result,
resizeSettings,
duration,
rootSpanStatus,
});
};
@@ -82,8 +85,15 @@ function getSpanId(path: string): string | undefined {
}
export default function Page() {
const { run, events, parentRunFriendlyId, resizeSettings, duration, rootSpanStatus } =
useTypedLoaderData<typeof loader>();
const {
run,
events,
parentRunFriendlyId,
resizeSettings,
duration,
rootSpanStatus,
rootStartedAt,
} = useTypedLoaderData<typeof loader>();
const navigate = useNavigate();
const organization = useOrganization();
const pathName = usePathName();
@@ -142,6 +152,7 @@ export default function Page() {
}}
totalDuration={duration}
rootSpanStatus={rootSpanStatus}
rootStartedAt={rootStartedAt}
/>
) : (
<ResizablePanelGroup
@@ -169,6 +180,7 @@ export default function Page() {
}}
totalDuration={duration}
rootSpanStatus={rootSpanStatus}
rootStartedAt={rootStartedAt}
/>
</ResizablePanel>
<ResizableHandle withHandle />
@@ -183,7 +195,15 @@ export default function Page() {
);
}
const tickCount = 5;
type TasksTreeViewProps = {
events: RunEvent[];
selectedId?: string;
parentRunFriendlyId?: string;
onSelectedIdChanged: (selectedId: string | undefined) => void;
totalDuration: number;
rootSpanStatus: "executing" | "completed" | "failed";
rootStartedAt: Date | undefined;
};
function TasksTreeView({
events,
@@ -192,14 +212,8 @@ function TasksTreeView({
onSelectedIdChanged,
totalDuration,
rootSpanStatus,
}: {
events: RunEvent[];
selectedId?: string;
parentRunFriendlyId?: string;
onSelectedIdChanged: (selectedId: string | undefined) => void;
totalDuration: number;
rootSpanStatus: "executing" | "completed" | "failed";
}) {
rootStartedAt,
}: TasksTreeViewProps) {
const [filterText, setFilterText] = useState("");
const [errorsOnly, setErrorsOnly] = useState(false);
const [showDurations, setShowDurations] = useState(false);
@@ -207,8 +221,6 @@ function TasksTreeView({
const parentRef = useRef<HTMLDivElement>(null);
const treeScrollRef = useRef<HTMLDivElement>(null);
const timelineScrollRef = useRef<HTMLDivElement>(null);
const timelineContainerRef = useRef<HTMLDivElement>(null);
const initialTimelineDimensions = useInitialDimensions(timelineContainerRef);
const {
nodes,
@@ -238,9 +250,6 @@ function TasksTreeView({
},
});
const minTimelineWidth = initialTimelineDimensions?.width ?? 300;
const maxTimelineWidth = minTimelineWidth * 10;
return (
<div className="grid h-full grid-rows-[2.5rem_1fr] overflow-hidden">
<div className="mx-3 flex items-center justify-between gap-2 border-b border-grid-dimmed">
@@ -378,170 +387,247 @@ function TasksTreeView({
<ResizableHandle withHandle />
{/* Timeline */}
<ResizablePanel order={2} minSize={20} defaultSize={50}>
<div
className="h-full overflow-x-auto overflow-y-hidden scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600"
ref={timelineContainerRef}
>
<Timeline.Root
durationMs={nanosecondsToMilliseconds(totalDuration * 1.05)}
scale={scale}
className="h-full overflow-hidden"
minWidth={minTimelineWidth}
maxWidth={maxTimelineWidth}
>
{/* Follows the cursor */}
<CurrentTimeIndicator totalDuration={totalDuration} />
<TimelineView
totalDuration={totalDuration}
scale={scale}
events={events}
rootSpanStatus={rootSpanStatus}
rootStartedAt={rootStartedAt}
parentRef={parentRef}
timelineScrollRef={timelineScrollRef}
nodes={nodes}
getNodeProps={getNodeProps}
getTreeProps={getTreeProps}
showDurations={showDurations}
treeScrollRef={treeScrollRef}
virtualizer={virtualizer}
toggleNodeSelection={toggleNodeSelection}
/>
</ResizablePanel>
</ResizablePanelGroup>
</div>
);
}
<Timeline.Row className="grid h-full grid-rows-[2rem_1fr]">
{/* The duration labels */}
<Timeline.Row>
<Timeline.Row className="h-6">
<Timeline.EquallyDistribute count={tickCount}>
{(ms: number, index: number) => {
if (index === tickCount - 1) return null;
return (
<Timeline.Point
ms={ms}
className={"relative bottom-[2px] text-xxs text-text-dimmed"}
>
{(ms) => (
<div
className={cn(
"whitespace-nowrap",
index === 0
? "ml-1"
: index === tickCount - 1
? "-ml-1 -translate-x-full"
: "-translate-x-1/2"
)}
>
{formatDurationMilliseconds(ms, {
style: "short",
maxDecimalPoints: ms < 1000 ? 0 : 1,
})}
</div>
)}
</Timeline.Point>
);
}}
</Timeline.EquallyDistribute>
{rootSpanStatus !== "executing" && (
<Timeline.Point
ms={nanosecondsToMilliseconds(totalDuration)}
className={cn(
"relative bottom-[2px] text-xxs",
rootSpanStatus === "completed" ? "text-success" : "text-error"
)}
>
type TimelineViewProps = Pick<
TasksTreeViewProps,
"totalDuration" | "rootSpanStatus" | "events" | "rootStartedAt"
> & {
scale: number;
parentRef: React.RefObject<HTMLDivElement>;
timelineScrollRef: React.RefObject<HTMLDivElement>;
virtualizer: Virtualizer<HTMLElement, Element>;
nodes: NodesState;
getNodeProps: UseTreeStateOutput["getNodeProps"];
getTreeProps: UseTreeStateOutput["getTreeProps"];
toggleNodeSelection: UseTreeStateOutput["toggleNodeSelection"];
showDurations: boolean;
treeScrollRef: React.RefObject<HTMLDivElement>;
};
const tickCount = 5;
function TimelineView({
totalDuration,
scale,
rootSpanStatus,
rootStartedAt,
parentRef,
timelineScrollRef,
virtualizer,
events,
nodes,
getNodeProps,
getTreeProps,
toggleNodeSelection,
showDurations,
treeScrollRef,
}: TimelineViewProps) {
const timelineContainerRef = useRef<HTMLDivElement>(null);
const initialTimelineDimensions = useInitialDimensions(timelineContainerRef);
const minTimelineWidth = initialTimelineDimensions?.width ?? 300;
const maxTimelineWidth = minTimelineWidth * 10;
//we want to live-update the duration if the root span is still executing
const [duration, setDuration] = useState(totalDuration);
useEffect(() => {
if (rootSpanStatus !== "executing" || !rootStartedAt) {
setDuration(totalDuration);
return;
}
const interval = setInterval(() => {
setDuration(millisecondsToNanoseconds(Date.now() - rootStartedAt.getTime()));
}, 500);
return () => clearInterval(interval);
}, [totalDuration, rootSpanStatus]);
return (
<div
className="h-full overflow-x-auto overflow-y-hidden scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600"
ref={timelineContainerRef}
>
<Timeline.Root
durationMs={nanosecondsToMilliseconds(duration * 1.05)}
scale={scale}
className="h-full overflow-hidden"
minWidth={minTimelineWidth}
maxWidth={maxTimelineWidth}
>
{/* Follows the cursor */}
<CurrentTimeIndicator totalDuration={duration} />
<Timeline.Row className="grid h-full grid-rows-[2rem_1fr]">
{/* The duration labels */}
<Timeline.Row>
<Timeline.Row className="h-6">
<Timeline.EquallyDistribute count={tickCount}>
{(ms: number, index: number) => {
if (index === tickCount - 1) return null;
return (
<Timeline.Point
ms={ms}
className={"relative bottom-[2px] text-xxs text-text-dimmed"}
>
{(ms) => (
<div
className={cn(
"whitespace-nowrap",
index === 0
? "ml-1"
: index === tickCount - 1
? "-ml-1 -translate-x-full"
: "-translate-x-1/2"
)}
>
{formatDurationMilliseconds(ms, {
style: "short",
maxDecimalPoints: ms < 1000 ? 0 : 1,
})}
</div>
)}
</Timeline.Point>
);
}}
</Timeline.EquallyDistribute>
{rootSpanStatus !== "executing" && (
<Timeline.Point
ms={nanosecondsToMilliseconds(duration)}
className={cn(
"relative bottom-[2px] text-xxs",
rootSpanStatus === "completed" ? "text-success" : "text-error"
)}
>
{(ms) => (
<div className={cn("-translate-x-1/2 whitespace-nowrap")}>
{formatDurationMilliseconds(ms, {
style: "short",
maxDecimalPoints: ms < 1000 ? 0 : 1,
})}
</div>
)}
</Timeline.Point>
)}
</Timeline.Row>
<Timeline.Row className="h-2">
<Timeline.EquallyDistribute count={tickCount}>
{(ms: number, index: number) => {
if (index === 0 || index === tickCount - 1) return null;
return (
<Timeline.Point ms={ms} className={"h-full border-r border-grid-dimmed"} />
);
}}
</Timeline.EquallyDistribute>
<Timeline.Point
ms={nanosecondsToMilliseconds(duration)}
className={cn(
"h-full border-r",
rootSpanStatus === "completed" ? "border-success/30" : "border-error/30"
)}
/>
</Timeline.Row>
</Timeline.Row>
{/* Main timeline body */}
<Timeline.Row className="overflow-hidden">
{/* The vertical tick lines */}
<Timeline.EquallyDistribute count={tickCount}>
{(ms: number, index: number) => {
if (index === 0) return null;
return <Timeline.Point ms={ms} className={"h-full border-r border-grid-dimmed"} />;
}}
</Timeline.EquallyDistribute>
{/* The completed line */}
{rootSpanStatus !== "executing" && (
<Timeline.Point
ms={nanosecondsToMilliseconds(duration)}
className={cn(
"h-full border-r",
rootSpanStatus === "completed" ? "border-success/30" : "border-error/30"
)}
/>
)}
<TreeView
parentRef={parentRef}
scrollRef={timelineScrollRef}
virtualizer={virtualizer}
tree={events}
nodes={nodes}
getNodeProps={getNodeProps}
getTreeProps={getTreeProps}
parentClassName="h-full scrollbar-hide"
renderNode={({ node, state, index, virtualizer, virtualItem }) => {
return (
<Timeline.Row
key={index}
className={cn(
"group flex h-8 items-center",
state.selected
? "bg-grid-dimmed hover:bg-grid-bright"
: "bg-transparent hover:bg-grid-dimmed"
)}
// onMouseOver={() => console.log(`hover ${index}`)}
onClick={(e) => {
toggleNodeSelection(node.id);
}}
>
{node.data.level === "TRACE" ? (
<SpanWithDuration
showDuration={state.selected ? true : showDurations}
startMs={nanosecondsToMilliseconds(node.data.offset)}
durationMs={
node.data.duration
? nanosecondsToMilliseconds(node.data.duration)
: nanosecondsToMilliseconds(duration - node.data.offset)
}
node={node}
/>
) : (
<Timeline.Point ms={nanosecondsToMilliseconds(node.data.offset)}>
{(ms) => (
<div className={cn("-translate-x-1/2 whitespace-nowrap")}>
{formatDurationMilliseconds(ms, {
style: "short",
maxDecimalPoints: ms < 1000 ? 0 : 1,
})}
</div>
<motion.div
className={cn(
"-ml-1 h-3 w-3 rounded-full border-2 border-background-bright",
eventBackgroundClassName(node.data)
)}
layoutId={node.id}
/>
)}
</Timeline.Point>
)}
</Timeline.Row>
<Timeline.Row className="h-2">
<Timeline.EquallyDistribute count={tickCount}>
{(ms: number, index: number) => {
if (index === 0 || index === tickCount - 1) return null;
return (
<Timeline.Point
ms={ms}
className={"h-full border-r border-grid-dimmed"}
/>
);
}}
</Timeline.EquallyDistribute>
<Timeline.Point
ms={nanosecondsToMilliseconds(totalDuration)}
className={cn(
"h-full border-r",
rootSpanStatus === "completed" ? "border-success/30" : "border-error/30"
)}
/>
</Timeline.Row>
</Timeline.Row>
{/* Main timeline body */}
<Timeline.Row className="overflow-hidden">
{/* The vertical tick lines */}
<Timeline.EquallyDistribute count={tickCount}>
{(ms: number, index: number) => {
if (index === 0) return null;
return (
<Timeline.Point ms={ms} className={"h-full border-r border-grid-dimmed"} />
);
}}
</Timeline.EquallyDistribute>
{/* The completed line */}
{rootSpanStatus !== "executing" && (
<Timeline.Point
ms={nanosecondsToMilliseconds(totalDuration)}
className={cn(
"h-full border-r",
rootSpanStatus === "completed" ? "border-success/30" : "border-error/30"
)}
/>
)}
<TreeView
parentRef={parentRef}
scrollRef={timelineScrollRef}
virtualizer={virtualizer}
tree={events}
nodes={nodes}
getNodeProps={getNodeProps}
getTreeProps={getTreeProps}
parentClassName="h-full scrollbar-hide"
renderNode={({ node, state, index, virtualizer, virtualItem }) => {
return (
<Timeline.Row
key={index}
className={cn(
"group flex h-8 items-center",
state.selected
? "bg-grid-dimmed hover:bg-grid-bright"
: "bg-transparent hover:bg-grid-dimmed"
)}
// onMouseOver={() => console.log(`hover ${index}`)}
onClick={(e) => {
toggleNodeSelection(node.id);
}}
>
{node.data.level === "TRACE" ? (
<SpanWithDuration
showDuration={state.selected ? true : showDurations}
startMs={nanosecondsToMilliseconds(node.data.offset)}
durationMs={nanosecondsToMilliseconds(node.data.duration)}
node={node}
/>
) : (
<Timeline.Point
ms={nanosecondsToMilliseconds(node.data.offset)}
className={cn(
"-ml-1 h-3 w-3 rounded-full border-2 border-background-bright",
eventBackgroundClassName(node.data)
)}
/>
)}
</Timeline.Row>
);
}}
onScroll={(scrollTop) => {
//sync the scroll to the tree
if (treeScrollRef.current && treeScrollRef.current.scrollTop !== scrollTop) {
treeScrollRef.current.scrollTop = scrollTop;
}
}}
/>
</Timeline.Row>
</Timeline.Row>
</Timeline.Root>
</div>
</ResizablePanel>
</ResizablePanelGroup>
);
}}
onScroll={(scrollTop) => {
//sync the scroll to the tree
if (treeScrollRef.current && treeScrollRef.current.scrollTop !== scrollTop) {
treeScrollRef.current.scrollTop = scrollTop;
}
}}
/>
</Timeline.Row>
</Timeline.Row>
</Timeline.Root>
</div>
);
}
@@ -659,11 +745,12 @@ function SpanWithDuration({
}: Timeline.SpanProps & { node: RunEvent; showDuration: boolean }) {
return (
<Timeline.Span {...props}>
<div
<motion.div
className={cn(
"relative flex h-4 w-full min-w-[2px] items-center rounded-sm",
eventBackgroundClassName(node.data)
)}
layoutId={node.id}
>
{node.data.isPartial && (
<div
@@ -684,7 +771,7 @@ function SpanWithDuration({
})}
</div>
</div>
</div>
</motion.div>
</Timeline.Span>
);
}
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.tasks.stream/route.tsx
+13
View File
@@ -0,0 +1,13 @@
import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
import { TasksStreamPresenter } from "~/presenters/v3/TasksStreamPresenter.server";
import { requireUserId } from "~/services/session.server";
import { ProjectParamSchema } from "~/utils/pathBuilder";
export async function loader({ request, params }: LoaderFunctionArgs) {
const userId = await requireUserId(request);
const { organizationSlug, projectParam } = ProjectParamSchema.parse(params);
const presenter = new TasksStreamPresenter();
return presenter.call({ request, projectSlug: projectParam, organizationSlug, userId });
}
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, true);
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, true);
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/utils/pathBuilder.ts
+4
View File
@@ -301,6 +301,10 @@ export function v3ProjectPath(organization: OrgForPath, project: ProjectForPath)
return `/orgs/${organizationParam(organization)}/projects/v3/${projectParam(project)}`;
}
export function v3TasksStreamingPath(organization: OrgForPath, project: ProjectForPath) {
return `${v3ProjectPath(organization, project)}/tasks/stream`;
}
export function v3ApiKeysPath(organization: OrgForPath, project: ProjectForPath) {
return `${v3ProjectPath(organization, project)}/apikeys`;
}
apps/webapp/app/v3/eventRepository.server.ts
+38 -23
View File
@@ -67,7 +67,7 @@ export type TraceEventOptions = {
attributes: TraceAttributes;
environment: AuthenticatedEnvironment;
taskSlug: string;
startTime?: Date;
startTime?: bigint;
endTime?: Date;
immediate?: boolean;
};
@@ -153,6 +153,10 @@ export class EventRepository {
this._flushScheduler.addToBatch(events);
}
async insertManyImmediate(events: CreatableEvent[]) {
return await this.#flushBatch(events);
}
async completeEvent(spanId: string, options?: UpdateEventOptions) {
const events = await this.queryIncompleteEvents({ spanId });
@@ -172,8 +176,7 @@ export class EventRepository {
status: options?.attributes.isError ? "ERROR" : "OK",
links: event.links ?? [],
events: event.events ?? [],
duration:
((options?.endTime ?? new Date()).getTime() - event.startTime.getTime()) * 1_000_000, // convert to nanoseconds
duration: calculateDurationFromStart(event.startTime, options?.endTime),
properties: event.properties as Attributes,
metadata: event.metadata as Attributes,
style: event.style as Attributes,
@@ -205,7 +208,7 @@ export class EventRepository {
},
...((event.events as any[]) ?? []),
],
duration: (cancelledAt.getTime() - event.startTime.getTime()) * 1_000_000, // convert to nanoseconds
duration: calculateDurationFromStart(event.startTime, cancelledAt),
properties: event.properties as Attributes,
metadata: event.metadata as Attributes,
style: event.style as Attributes,
@@ -285,7 +288,7 @@ export class EventRepository {
isError: event.isError,
isPartial: ancestorCancelled ? false : event.isPartial,
isCancelled: event.isCancelled === true ? true : event.isPartial && ancestorCancelled,
startTime: event.startTime,
startTime: getDateFromNanoseconds(event.startTime),
level: event.level,
events: event.events,
},
@@ -368,8 +371,8 @@ export class EventRepository {
public async recordEvent(message: string, options: TraceEventOptions) {
const propagatedContext = extractContextFromCarrier(options.context ?? {});
const startTime = options.startTime ?? new Date();
const durationInMs = options.endTime ? options.endTime.getTime() - startTime.getTime() : 100;
const startTime = options.startTime ?? getNowInNanoseconds();
const duration = options.endTime ? calculateDurationFromStart(startTime, options.endTime) : 100;
const traceId = propagatedContext?.traceparent?.traceId ?? this.generateTraceId();
const parentId = propagatedContext?.traceparent?.spanId;
@@ -414,7 +417,7 @@ export class EventRepository {
status: "OK",
startTime,
isPartial: false,
duration: durationInMs * 1_000_000, // convert to nanoseconds
duration, // convert to nanoseconds
environmentId: options.environment.id,
environmentType: options.environment.type,
organizationId: options.environment.organizationId,
@@ -459,7 +462,7 @@ export class EventRepository {
const propagatedContext = extractContextFromCarrier(options.context ?? {});
const start = process.hrtime.bigint();
const startTime = new Date();
const startTime = getNowInNanoseconds();
const traceId = options.spanParentAsLink
? this.generateTraceId()
@@ -477,14 +480,14 @@ export class EventRepository {
const links: Link[] =
options.spanParentAsLink && propagatedContext?.traceparent
? [
{
context: {
traceId: propagatedContext.traceparent.traceId,
spanId: propagatedContext.traceparent.spanId,
traceFlags: TraceFlags.SAMPLED,
},
{
context: {
traceId: propagatedContext.traceparent.traceId,
spanId: propagatedContext.traceparent.spanId,
traceFlags: TraceFlags.SAMPLED,
},
]
},
]
: [];
const eventBuilder = {
@@ -548,7 +551,7 @@ export class EventRepository {
level: "TRACE",
kind: options.kind,
status: "OK",
startTime: startTime,
startTime,
environmentId: options.environment.id,
environmentType: options.environment.type,
organizationId: options.environment.organizationId,
@@ -739,9 +742,9 @@ function prepareEvent(event: QueriedEvent): PreparedEvent {
function parseEventsField(events: Prisma.JsonValue): SpanEvents {
const eventsUnflattened = events
? (events as any[]).map((e) => ({
...e,
properties: unflattenAttributes(e.properties as Attributes),
}))
...e,
properties: unflattenAttributes(e.properties as Attributes),
}))
: undefined;
const spanEvents = SpanEvents.safeParse(eventsUnflattened);
@@ -810,7 +813,7 @@ function calculateDurationIfAncestorIsCancelled(
);
if (cancellationEvent) {
return (cancellationEvent.time.getTime() - event.startTime.getTime()) * 1_000_000;
return calculateDurationFromStart(event.startTime, cancellationEvent.time);
}
}
}
@@ -935,8 +938,8 @@ function transformException(
...exception,
stacktrace: exception.stacktrace
? correctErrorStackTrace(exception.stacktrace, projectDirAttributeValue, {
removeFirstLine: true,
})
removeFirstLine: true,
})
: undefined,
};
}
@@ -952,3 +955,15 @@ function filteredAttributes(attributes: Attributes, prefix: string): Attributes
return result;
}
function calculateDurationFromStart(startTime: bigint, endTime: Date = new Date()) {
return Number(BigInt(endTime.getTime() * 1_000_000) - startTime);
}
function getNowInNanoseconds(): bigint {
return BigInt(new Date().getTime() * 1_000_000);
}
function getDateFromNanoseconds(nanoseconds: bigint) {
return new Date(Number(nanoseconds) / 1_000_000);
}
apps/webapp/app/v3/marqs/devPubSub.server.ts
+1 -8
View File
@@ -26,13 +26,6 @@ function initializeDevPubSub() {
enableAutoPipelining: true,
...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
},
schema: {
CANCEL_ATTEMPT: z.object({
version: z.literal("v1").default("v1"),
backgroundWorkerId: z.string(),
attemptId: z.string(),
taskRunId: z.string(),
}),
},
schema: messageCatalog,
});
}
apps/webapp/app/v3/otlpExporter.server.ts
+52 -43
View File
@@ -35,9 +35,9 @@ class OTLPExporter {
constructor(
private readonly _eventRepository: EventRepository,
private readonly _verbose: boolean
) {}
) { }
async exportTraces(request: ExportTraceServiceRequest): Promise<ExportTraceServiceResponse> {
async exportTraces(request: ExportTraceServiceRequest, immediate: boolean = false): Promise<ExportTraceServiceResponse> {
this.#logExportTracesVerbose(request);
const events = this.#filterResourceSpans(request.resourceSpans).flatMap((resourceSpan) => {
@@ -46,12 +46,16 @@ class OTLPExporter {
this.#logEventsVerbose(events);
this._eventRepository.insertMany(events);
if (immediate) {
await this._eventRepository.insertManyImmediate(events);
} else {
await this._eventRepository.insertMany(events);
}
return ExportTraceServiceResponse.create();
}
async exportLogs(request: ExportLogsServiceRequest): Promise<ExportLogsServiceResponse> {
async exportLogs(request: ExportLogsServiceRequest, immediate: boolean = false): Promise<ExportLogsServiceResponse> {
this.#logExportLogsVerbose(request);
const events = this.#filterResourceLogs(request.resourceLogs).flatMap((resourceLog) => {
@@ -60,7 +64,11 @@ class OTLPExporter {
this.#logEventsVerbose(events);
this._eventRepository.insertMany(events);
if (immediate) {
await this._eventRepository.insertManyImmediate(events);
} else {
await this._eventRepository.insertMany(events);
}
return ExportLogsServiceResponse.create();
}
@@ -109,7 +117,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 +131,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,13 +149,13 @@ 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),
isError: logLevel === "ERROR",
status: logLevelToEventStatus(log.severityNumber),
startTime: convertUnixNanoToDate(log.timeUnixNano),
startTime: log.timeUnixNano,
properties: {
...convertKeyValueItemsToMap(log.attributes ?? [], [
SemanticInternalAttributes.SPAN_ID,
@@ -201,10 +209,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,
@@ -213,7 +221,7 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): Array<Crea
kind: spanKindToEventKind(span.kind),
level: "TRACE",
status: spanStatusToEventStatus(span.status),
startTime: convertUnixNanoToDate(span.startTimeUnixNano),
startTime: span.startTimeUnixNano,
links: spanLinksToEventLinks(span.links ?? []),
events: spanEventsToEventEvents(span.events ?? []),
duration: span.endTimeUnixNano - span.startTimeUnixNano,
@@ -327,16 +335,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 +513,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 +528,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 +542,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 +556,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 +568,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"
);
apps/webapp/app/v3/services/createBackgroundWorker.server.ts
+10
View File
@@ -7,6 +7,7 @@ import { generateFriendlyId } from "../friendlyIdentifiers";
import { marqs } from "../marqs.server";
import { calculateNextBuildVersion } from "../utils/calculateNextBuildVersion";
import { BaseService } from "./baseService.server";
import { projectPubSub } from "./projectPubSub.server";
export class CreateBackgroundWorkerService extends BaseService {
public async call(
@@ -67,6 +68,15 @@ export class CreateBackgroundWorkerService extends BaseService {
await createBackgroundTasks(body.metadata.tasks, backgroundWorker, environment, this._prisma);
//send a notification that a new worker has been created
await projectPubSub.publish(`project:${project.id}:env:${environment.id}`, "WORKER_CREATED", {
environmentId: environment.id,
environmentType: environment.type,
createdAt: backgroundWorker.createdAt,
taskCount: body.metadata.tasks.length,
type: "local",
});
return backgroundWorker;
});
}
apps/webapp/app/v3/services/createDeployedBackgroundWorker.server.ts
+14
View File
@@ -5,6 +5,7 @@ import { generateFriendlyId } from "../friendlyIdentifiers";
import { BaseService } from "./baseService.server";
import { createBackgroundTasks } from "./createBackgroundWorker.server";
import { CURRENT_DEPLOYMENT_LABEL } from "~/consts";
import { projectPubSub } from "./projectPubSub.server";
export class CreateDeployedBackgroundWorkerService extends BaseService {
public async call(
@@ -71,6 +72,19 @@ export class CreateDeployedBackgroundWorkerService extends BaseService {
},
});
//send a notification that a new worker has been created
await projectPubSub.publish(
`project:${environment.projectId}:env:${environment.id}`,
"WORKER_CREATED",
{
environmentId: environment.id,
environmentType: environment.type,
createdAt: backgroundWorker.createdAt,
taskCount: body.metadata.tasks.length,
type: "deployed",
}
);
return backgroundWorker;
});
}
apps/webapp/app/v3/services/projectPubSub.server.ts
+32
View File
@@ -0,0 +1,32 @@
import { z } from "zod";
import { singleton } from "~/utils/singleton";
import { ZodPubSub, ZodSubscriber } from "../utils/zodPubSub.server";
import { env } from "~/env.server";
const messageCatalog = {
WORKER_CREATED: z.object({
environmentId: z.string(),
environmentType: z.string(),
createdAt: z.coerce.date(),
taskCount: z.number(),
type: z.union([z.literal("local"), z.literal("deployed")]),
}),
};
export type ProjectSubscriber = ZodSubscriber<typeof messageCatalog>;
export const projectPubSub = singleton("projectPubSub", initializeProjectPubSub);
function initializeProjectPubSub() {
return new ZodPubSub({
redis: {
port: env.REDIS_PORT,
host: env.REDIS_HOST,
username: env.REDIS_USERNAME,
password: env.REDIS_PASSWORD,
enableAutoPipelining: true,
...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
},
schema: messageCatalog,
});
}
apps/webapp/server.ts
+1
View File
@@ -71,6 +71,7 @@ if (process.env.HTTP_SERVER_DISABLED !== "true") {
if (process.env.DASHBOARD_AND_API_DISABLED !== "true") {
app.all(
"*",
// @ts-ignore
createRequestHandler({
build,
mode: MODE,
packages/cli-v3/src/commands/deploy.ts
+2 -2
View File
@@ -1,7 +1,7 @@
import { intro, log, outro, spinner } from "@clack/prompts";
import { depot } from "@depot/cli";
import { context, trace } from "@opentelemetry/api";
import { ResolvedConfig, flattenAttributes, recordSpanException } from "@trigger.dev/core/v3";
import { ResolvedConfig, detectDependencyVersion, flattenAttributes, recordSpanException } from "@trigger.dev/core/v3";
import chalk from "chalk";
import { Command, Option as CommandOption } from "commander";
import { Metafile, build } from "esbuild";
@@ -1225,7 +1225,7 @@ function gatherRequiredDependencies(
const internalDependencyVersion = (packageJson.dependencies as Record<string, string>)[
packageName
];
] ?? detectDependencyVersion(packageName);
if (internalDependencyVersion) {
dependencies[packageName] = internalDependencyVersion;
packages/cli-v3/src/commands/dev.tsx
+4 -4
View File
@@ -5,6 +5,7 @@ import {
ZodMessageHandler,
ZodMessageSender,
clientWebsocketMessages,
detectDependencyVersion,
serverWebsocketMessages,
} from "@trigger.dev/core/v3";
import chalk from "chalk";
@@ -33,7 +34,6 @@ import { isLoggedIn } from "../utilities/session.js";
import { createTaskFileImports, gatherTaskFiles } from "../utilities/taskFiles";
import { UncaughtExceptionError } from "../workers/common/errors";
import { BackgroundWorker, BackgroundWorkerCoordinator } from "../workers/dev/backgroundWorker.js";
import { fromZodError } from "zod-validation-error";
let apiClient: CliApiClient | undefined;
@@ -692,9 +692,9 @@ function gatherRequiredDependencies(outputMeta: Metafile["outputs"][string]) {
continue;
}
const internalDependencyVersion = (packageJson.dependencies as Record<string, string>)[
packageName
];
const internalDependencyVersion =
(packageJson.dependencies as Record<string, string>)[packageName] ??
detectDependencyVersion(packageName);
if (internalDependencyVersion) {
dependencies[packageName] = internalDependencyVersion;
packages/cli-v3/src/utilities/installPackages.ts
+3 -1
View File
@@ -12,6 +12,8 @@ export async function installPackages(
) {
const cwd = options?.cwd ?? process.cwd();
logger.debug(`Installing packages at ${cwd}:`, { packages });
// Make sure the cwd has a package.json file (if not create a barebones one)
try {
await readJSONFile(join(cwd, "package.json"));
@@ -49,7 +51,7 @@ export async function installPackages(
return;
}
logger.debug(`Installing packages at ${cwd}:`);
logger.debug(`Found installable packages`);
logger.table(
Object.entries(installablePackages).map(([name, version]) => ({ name, version })),
"debug"
packages/cli-v3/src/workers/dev/backgroundWorker.ts
+21 -11
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) {
@@ -302,8 +302,8 @@ export class BackgroundWorker {
const child = fork(this.path, {
stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"],
env: {
...this.#readEnvVars(),
...this.params.env,
...this.#readEnvVars(),
},
});
@@ -475,13 +475,19 @@ export class BackgroundWorker {
}
#readEnvVars() {
const result = {};
const result: { [key: string]: string } = {};
dotenv.config({
processEnv: result,
path: [".env", ".env.local", ".env.development.local"].map((p) => resolve(process.cwd(), p)),
});
process.env.TRIGGER_API_URL && (result.TRIGGER_API_URL = process.env.TRIGGER_API_URL);
// remove TRIGGER_API_URL and TRIGGER_SECRET_KEY, since those should be coming from the worker
delete result.TRIGGER_API_URL;
delete result.TRIGGER_SECRET_KEY;
return result;
}
@@ -536,6 +542,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 +555,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 +700,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 +717,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/cli-v3/src/workers/dev/worker-facade.ts
+5 -2
View File
@@ -1,4 +1,4 @@
import { Config, ProjectConfig, TaskExecutor, type TracingSDK } from "@trigger.dev/core/v3";
import { Config, ProjectConfig, TaskExecutor, preciseDateOriginNow, type TracingSDK } from "@trigger.dev/core/v3";
import "source-map-support/register.js";
__WORKER_SETUP__;
@@ -35,8 +35,10 @@ import { TaskMetadataWithFunctions } from "../../types.js";
declare const sender: ZodMessageSender<typeof childToWorkerMessages>;
const preciseDateOrigin = preciseDateOriginNow();
const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
const consoleInterceptor = new ConsoleInterceptor(otelLogger);
const consoleInterceptor = new ConsoleInterceptor(otelLogger, preciseDateOrigin);
const devRuntimeManager = new DevRuntimeManager();
@@ -46,6 +48,7 @@ const otelTaskLogger = new OtelTaskLogger({
logger: otelLogger,
tracer: tracer,
level: "info",
preciseDateOrigin
});
logger.setGlobalTaskLogger(otelTaskLogger);
packages/cli-v3/src/workers/prod/worker-facade.ts
+5 -1
View File
@@ -6,6 +6,7 @@ import {
TaskExecutor,
ZodIpcConnection,
type TracingSDK,
preciseDateOriginNow,
} from "@trigger.dev/core/v3";
import "source-map-support/register.js";
@@ -37,13 +38,16 @@ import * as packageJson from "../../../package.json";
import { TaskMetadataWithFunctions } from "../../types";
const preciseDateOrigin = preciseDateOriginNow();
const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
const consoleInterceptor = new ConsoleInterceptor(otelLogger);
const consoleInterceptor = new ConsoleInterceptor(otelLogger, preciseDateOrigin);
const otelTaskLogger = new OtelTaskLogger({
logger: otelLogger,
tracer: tracer,
level: "info",
preciseDateOrigin
});
logger.setGlobalTaskLogger(otelTaskLogger);
packages/core/package.json
+1
View File
@@ -58,6 +58,7 @@
"test": "jest"
},
"dependencies": {
"@google-cloud/precise-date": "^4.0.0",
"@opentelemetry/api": "^1.7.0",
"@opentelemetry/api-logs": "^0.48.0",
"@opentelemetry/auto-instrumentations-node": "^0.40.3",
packages/core/src/v3/consoleInterceptor.ts
+12 -3
View File
@@ -1,12 +1,14 @@
import type * as logsAPI from "@opentelemetry/api-logs";
import { SeverityNumber } from "@opentelemetry/api-logs";
import util from "node:util";
import { flattenAttributes } from "./utils/flattenAttributes";
import { SemanticInternalAttributes } from "./semanticInternalAttributes";
import { iconStringForSeverity } from "./icons";
import { SemanticInternalAttributes } from "./semanticInternalAttributes";
import { flattenAttributes } from "./utils/flattenAttributes";
import { type PreciseDateOrigin, calculatePreciseDateHrTime } from "./utils/preciseDate";
export class ConsoleInterceptor {
constructor(private readonly logger: logsAPI.Logger) {}
constructor(private readonly logger: logsAPI.Logger, private readonly preciseDateOrigin: PreciseDateOrigin) { }
// Intercept the console and send logs to the OpenTelemetry logger
// during the execution of the callback
@@ -54,6 +56,7 @@ export class ConsoleInterceptor {
#handleLog(severityNumber: SeverityNumber, severityText: string, ...args: unknown[]): void {
const body = util.format(...args);
const timestamp = this.#getTimestampInHrTime();
const parsed = tryParseJSON(body);
@@ -63,6 +66,7 @@ export class ConsoleInterceptor {
severityText,
body: getLogMessage(parsed.value, severityText),
attributes: { ...this.#getAttributes(severityNumber), ...flattenAttributes(parsed.value) },
timestamp,
});
return;
@@ -73,9 +77,14 @@ export class ConsoleInterceptor {
severityText,
body,
attributes: this.#getAttributes(severityNumber),
timestamp,
});
}
#getTimestampInHrTime(): [number, number] {
return calculatePreciseDateHrTime(this.preciseDateOrigin);
}
#getAttributes(severityNumber: SeverityNumber): logsAPI.LogAttributes {
const icon = iconStringForSeverity(severityNumber);
let result: logsAPI.LogAttributes = {};
packages/core/src/v3/index.ts
+2
View File
@@ -50,3 +50,5 @@ export { eventFilterMatches } from "../eventFilterMatches";
export { omit } from "./utils/omit";
export { TracingSDK, type TracingDiagnosticLogLevel, recordSpanException } from "./otel";
export { TaskExecutor, type TaskExecutorOptions } from "./workers/taskExecutor";
export { detectDependencyVersion } from "./utils/detectDependencyVersion";
export { type PreciseDateOrigin, calculatePreciseDateHrTime, preciseDateOriginNow } from "./utils/preciseDate";
packages/core/src/v3/logger/taskLogger.ts
+16 -7
View File
@@ -1,9 +1,10 @@
import { Logger, SeverityNumber } from "@opentelemetry/api-logs";
import { flattenAttributes } from "../utils/flattenAttributes";
import { Attributes, Span, SpanOptions } from "@opentelemetry/api";
import { Logger, SeverityNumber } from "@opentelemetry/api-logs";
import { iconStringForSeverity } from "../icons";
import { SemanticInternalAttributes } from "../semanticInternalAttributes";
import { TriggerTracer } from "../tracer";
import { flattenAttributes } from "../utils/flattenAttributes";
import { PreciseDateOrigin, calculatePreciseDateHrTime } from "../utils/preciseDate";
export type LogLevel = "log" | "error" | "warn" | "info" | "debug";
@@ -13,6 +14,7 @@ export type TaskLoggerConfig = {
logger: Logger;
tracer: TriggerTracer;
level: LogLevel;
preciseDateOrigin: PreciseDateOrigin;
};
export interface TaskLogger {
@@ -67,6 +69,8 @@ export class OtelTaskLogger implements TaskLogger {
severityNumber: SeverityNumber,
properties?: Record<string, unknown>
) {
const timestamp = this.#getTimestampInHrTime();
let attributes: Attributes = { ...flattenAttributes(properties) };
const icon = iconStringForSeverity(severityNumber);
@@ -79,20 +83,25 @@ export class OtelTaskLogger implements TaskLogger {
severityText,
body: message,
attributes,
timestamp
});
}
trace<T>(name: string, fn: (span: Span) => Promise<T>, options?: SpanOptions): Promise<T> {
return this._config.tracer.startActiveSpan(name, fn, options);
}
#getTimestampInHrTime(): [number, number] {
return calculatePreciseDateHrTime(this._config.preciseDateOrigin);
}
}
export class NoopTaskLogger implements TaskLogger {
debug() {}
log() {}
info() {}
warn() {}
error() {}
debug() { }
log() { }
info() { }
warn() { }
error() { }
trace<T>(name: string, fn: (span: Span) => Promise<T>): Promise<T> {
return fn({} as Span);
}
packages/core/src/v3/utils/detectDependencyVersion.ts
+5
View File
@@ -0,0 +1,5 @@
import { dependencies } from "../../../package.json"
export function detectDependencyVersion(dependency: string): string | undefined {
return (dependencies as Record<string, string>)[dependency]
}
packages/core/src/v3/utils/preciseDate.ts
+23
View File
@@ -0,0 +1,23 @@
import { PreciseDate } from "@google-cloud/precise-date";
export type PreciseDateOrigin = {
hrtime: [number, number];
timestamp: PreciseDate
}
export function preciseDateOriginNow(): PreciseDateOrigin {
return {
hrtime: process.hrtime(),
timestamp: new PreciseDate()
}
}
export function calculatePreciseDateHrTime(origin: PreciseDateOrigin): [number, number] {
const elapsedHrTime = process.hrtime(origin.hrtime);
const elapsedNanoseconds = BigInt(elapsedHrTime[0]) * BigInt(1e9) + BigInt(elapsedHrTime[1]);
const preciseDate = new PreciseDate(origin.timestamp.getFullTime() + elapsedNanoseconds)
const dateStruct = preciseDate.toStruct();
return [dateStruct.seconds, dateStruct.nanos];
}
packages/core/tsconfig.json
+1
View File
@@ -6,6 +6,7 @@
"emitDecoratorMetadata": true,
"declaration": false,
"declarationMap": false,
"resolveJsonModule": true,
"types": ["jest"],
"lib": ["DOM", "DOM.Iterable"],
"paths": {
packages/database/prisma/migrations/20240319120645_convert_start_time_to_nanoseconds_since_epoch/migration.sql
+34
View File
@@ -0,0 +1,34 @@
/*
Warnings:
- Changed the type of `startTime` on the `TaskEvent` table. No cast exists, the column would be dropped and recreated, which cannot be done if there is data, since the column is required.
*/
-- AlterTable
BEGIN;
-- Step 1: Add a new column
ALTER TABLE
"TaskEvent"
ADD
COLUMN "temporary_startTime" BIGINT;
-- Step 2: Convert the data in the current "startTime" column to nanoseconds
UPDATE
"TaskEvent"
SET
"temporary_startTime" = EXTRACT(
EPOCH
FROM
"startTime"
) * 1000000000;
-- Step 3: Drop the original column
ALTER TABLE
"TaskEvent" DROP COLUMN "startTime";
-- Step 4: Rename the new column
ALTER TABLE
"TaskEvent" RENAME COLUMN "temporary_startTime" TO "startTime";
COMMIT;
packages/database/prisma/migrations/20240319121124_what_migration_is_this/migration.sql
+8
View File
@@ -0,0 +1,8 @@
/*
Warnings:
- Made the column `startTime` on table `TaskEvent` required. This step will fail if there are existing NULL values in that column.
*/
-- AlterTable
ALTER TABLE "TaskEvent" ALTER COLUMN "startTime" SET NOT NULL;
packages/database/prisma/schema.prisma
+2 -1
View File
@@ -1762,7 +1762,8 @@ model TaskEvent {
links Json?
events Json?
startTime DateTime
/// This is the time the event started in nanoseconds since the epoch
startTime BigInt
/// This is the duration of the event in nanoseconds
duration BigInt @default(0)
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 {
pnpm-lock.yaml
Generated
+8 -1
View File
@@ -1198,6 +1198,7 @@ importers:
packages/core:
specifiers:
'@google-cloud/precise-date': ^4.0.0
'@opentelemetry/api': ^1.7.0
'@opentelemetry/api-logs': ^0.48.0
'@opentelemetry/auto-instrumentations-node': ^0.40.3
@@ -1230,6 +1231,7 @@ importers:
zod: 3.22.3
zod-error: 1.5.0
dependencies:
'@google-cloud/precise-date': 4.0.0
'@opentelemetry/api': 1.7.0
'@opentelemetry/api-logs': 0.48.0
'@opentelemetry/auto-instrumentations-node': 0.40.3_@opentelemetry+api@1.7.0
@@ -7599,6 +7601,11 @@ packages:
resolution: {integrity: sha512-k2Ty1JcVojjJFwrg/ThKi2ujJ7XNLYaFGNB/bWT9wGR+oSMJHMa5w+CUq6p/pVrKeNNgA7pCqEcjSnHVoqJQFw==}
dev: true
/@google-cloud/precise-date/4.0.0:
resolution: {integrity: sha512-1TUx3KdaU3cN7nfCdNf+UVqA/PSX29Cjcox3fZZBtINlRrXVTmUkQnCKv2MbBUbCopbK4olAT1IHl76uZyCiVA==}
engines: {node: '>=14.0.0'}
dev: false
/@graphile/logger/0.2.0:
resolution: {integrity: sha512-jjcWBokl9eb1gVJ85QmoaQ73CQ52xAaOCF29ukRbYNl6lY+ts0ErTaDYOBlejcbUs2OpaiqYLO5uDhyLFzWw4w==}
dev: false
@@ -37081,7 +37088,7 @@ packages:
dependencies:
bs-logger: 0.2.6
fast-json-stable-stringify: 2.1.0
jest: 29.6.2_@types+node@18.15.13
jest: 29.6.2_@types+node@18.17.1
jest-util: 29.6.2
json5: 2.2.3
lodash.memoize: 4.1.2
references/v3-catalog/src/trigger/simple.ts
+4
View File
@@ -72,9 +72,13 @@ export const parentTask = task({
logger.info("Parent task payload", { payload });
console.info("This is an info message");
logger.info("This is an info message from logger.info");
console.log(JSON.stringify({ ctx, message: "This is the parent task context" }));
logger.log(JSON.stringify({ ctx, message: "This is the parent task context from logger.log" }));
console.warn("You've been warned buddy");
logger.warn("You've been warned buddy from logger.warn");
console.error("This is an error message");
logger.error("This is an error message from logger.error");
await wait.for({ seconds: 5 });