Skip to content

Create new partitioned TaskEvent table, and switch to it gradually as new runs are created #1696

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Feb 12, 2025
Merged

Create new partitioned TaskEvent table, and switch to it gradually as new runs are created #1696

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
50444de
Create new partitioned TaskEvent table, and switch to it gradually as…
ericallam Feb 12, 2025
b734ee2
Add env var for partition window in seconds
ericallam Feb 12, 2025
e353738
Make startCreatedAt required in task event store
ericallam Feb 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 0 additions & 39 deletions apps/webapp/app/components/runs/v3/RunInspector.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,14 +32,11 @@ import { cn } from "~/utils/cn";
import { formatCurrencyAccurate } from "~/utils/numberFormatter";
import {
v3RunDownloadLogsPath,
v3RunPath,
v3RunSpanPath,
v3RunsPath,
v3SchedulePath,
v3TraceSpanPath,
} from "~/utils/pathBuilder";
import { TraceSpan } from "~/utils/taskEvent";
import { SpanLink } from "~/v3/eventRepository.server";
import { isFailedRunStatus, isFinalRunStatus } from "~/v3/taskStatus";
import { RunTimelineEvent, RunTimelineLine } from "./InspectorTimeline";
import { RunTag } from "./RunTag";
Expand Down Expand Up @@ -317,18 +314,6 @@ export function RunInspector({
)}
</Property.Value>
</Property.Item>
{span?.links && span.links.length > 0 && (
<Property.Item>
<Property.Label>Links</Property.Label>
<Property.Value>
<div className="space-y-1">
{span.links.map((link, index) => (
<SpanLinkElement key={index} link={link} />
))}
</div>
</Property.Value>
</Property.Item>
)}
<Property.Item>
<Property.Label>Max duration</Property.Label>
<Property.Value>
Expand Down Expand Up @@ -647,27 +632,3 @@ function PacketDisplay({
}
}
}

function SpanLinkElement({ link }: { link: SpanLink }) {
const organization = useOrganization();
const project = useProject();

switch (link.type) {
case "run": {
return (
<TextLink to={v3RunPath(organization, project, { friendlyId: link.runId })}>
{link.title}
</TextLink>
);
}
case "span": {
return (
<TextLink to={v3TraceSpanPath(organization, project, link.traceId, link.spanId)}>
{link.title}
</TextLink>
);
}
}

return null;
}
55 changes: 3 additions & 52 deletions apps/webapp/app/components/runs/v3/SpanInspector.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
import { formatDuration, nanosecondsToMilliseconds } from "@trigger.dev/core/v3";
import { ExitIcon } from "~/assets/icons/ExitIcon";
import { CodeBlock } from "~/components/code/CodeBlock";
import { Button } from "~/components/primitives/Buttons";
import { DateTimeAccurate } from "~/components/primitives/DateTime";
import { Header2 } from "~/components/primitives/Headers";
import * as Property from "~/components/primitives/PropertyTable";
import { Spinner } from "~/components/primitives/Spinner";
import { TabButton, TabContainer } from "~/components/primitives/Tabs";
import { TextLink } from "~/components/primitives/TextLink";
import { InfoIconTooltip, SimpleTooltip } from "~/components/primitives/Tooltip";
Expand All @@ -15,13 +17,10 @@ import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { useSearchParams } from "~/hooks/useSearchParam";
import { cn } from "~/utils/cn";
import { v3RunPath, v3RunsPath, v3TraceSpanPath } from "~/utils/pathBuilder";
import { v3RunsPath } from "~/utils/pathBuilder";
import { TraceSpan } from "~/utils/taskEvent";
import { SpanLink } from "~/v3/eventRepository.server";
import { RunTimelineEvent, RunTimelineLine } from "./InspectorTimeline";
import { Spinner } from "~/components/primitives/Spinner";
import { LiveTimer } from "./LiveTimer";
import { formatDuration, nanosecondsToMilliseconds } from "@trigger.dev/core/v3";

export function SpanInspector({
span,
Expand Down Expand Up @@ -150,18 +149,6 @@ export function SpanInspector({
)}
</Property.Value>
</Property.Item>
{span.links && span.links.length > 0 && (
<Property.Item>
<Property.Label>Links</Property.Label>
<Property.Value>
<div className="space-y-1">
{span.links.map((link, index) => (
<SpanLinkElement key={index} link={link} />
))}
</div>
</Property.Value>
</Property.Item>
)}
</Property.Table>
</div>
) : (
Expand Down Expand Up @@ -203,18 +190,6 @@ export function SpanInspector({
<Property.Label>Message</Property.Label>
<Property.Value>{span.message}</Property.Value>
</Property.Item>
{span.links && span.links.length > 0 && (
<Property.Item>
<Property.Label>Links</Property.Label>
<Property.Value>
<div className="space-y-1">
{span.links.map((link, index) => (
<SpanLinkElement key={index} link={link} />
))}
</div>
</Property.Value>
</Property.Item>
)}
</Property.Table>

{span.events !== undefined && <SpanEvents spanEvents={span.events} />}
Expand Down Expand Up @@ -287,27 +262,3 @@ export function SpanTimeline({ startTime, duration, inProgress, isError }: Timel
</>
);
}

function SpanLinkElement({ link }: { link: SpanLink }) {
const organization = useOrganization();
const project = useProject();

switch (link.type) {
case "run": {
return (
<TextLink to={v3RunPath(organization, project, { friendlyId: link.runId })}>
{link.title}
</TextLink>
);
}
case "span": {
return (
<TextLink to={v3TraceSpanPath(organization, project, link.traceId, link.spanId)}>
{link.title}
</TextLink>
);
}
}

return null;
}
3 changes: 3 additions & 0 deletions apps/webapp/app/env.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -481,6 +481,9 @@ const EnvironmentSchema = z.object({
.transform((v) => v ?? process.env.REDIS_PASSWORD),
COMMON_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
COMMON_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

TASK_EVENT_PARTITIONING_ENABLED: z.string().default("0"),
TASK_EVENT_PARTITIONED_WINDOW_IN_SECONDS: z.coerce.number().int().default(60), // 1 minute
});

export type Environment = z.infer<typeof EnvironmentSchema>;
Expand Down
10 changes: 9 additions & 1 deletion apps/webapp/app/presenters/v3/RunPresenter.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import { createTreeFromFlatItems, flattenTree } from "~/components/primitives/Tr
import { PrismaClient, prisma } from "~/db.server";
import { getUsername } from "~/utils/username";
import { eventRepository } from "~/v3/eventRepository.server";
import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
import { isFinalRunStatus } from "~/v3/taskStatus";

type Result = Awaited<ReturnType<RunPresenter["call"]>>;
Expand Down Expand Up @@ -32,6 +33,8 @@ export class RunPresenter {
const run = await this.#prismaClient.taskRun.findFirstOrThrow({
select: {
id: true,
createdAt: true,
taskEventStore: true,
number: true,
traceId: true,
spanId: true,
Expand Down Expand Up @@ -105,7 +108,12 @@ export class RunPresenter {
}

// get the events
const traceSummary = await eventRepository.getTraceSummary(run.traceId);
const traceSummary = await eventRepository.getTraceSummary(
getTaskEventStoreTableForRun(run),
run.traceId,
run.createdAt,
run.completedAt ?? undefined
);
if (!traceSummary) {
return {
run: runData,
Expand Down
21 changes: 19 additions & 2 deletions apps/webapp/app/presenters/v3/SpanPresenter.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import { machinePresetFromName } from "~/v3/machinePresets.server";
import { FINAL_ATTEMPT_STATUSES, isFailedRunStatus, isFinalRunStatus } from "~/v3/taskStatus";
import { BasePresenter } from "./basePresenter.server";
import { getMaxDuration } from "~/v3/utils/maxDuration";
import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";

type Result = Awaited<ReturnType<SpanPresenter["call"]>>;
export type Span = NonNullable<NonNullable<Result>["span"]>;
Expand Down Expand Up @@ -71,6 +72,7 @@ export class SpanPresenter extends BasePresenter {
friendlyId: true,
isTest: true,
maxDurationInSeconds: true,
taskEventStore: true,
tags: {
select: {
name: true,
Expand Down Expand Up @@ -205,7 +207,13 @@ export class SpanPresenter extends BasePresenter {
}
}

const span = await eventRepository.getSpan(spanId, run.traceId);
const span = await eventRepository.getSpan(
getTaskEventStoreTableForRun(run),
spanId,
run.traceId,
run.createdAt,
run.completedAt ?? undefined
);

const metadata = run.metadata
? await prettyPrintPacket(run.metadata, run.metadataType, {
Expand Down Expand Up @@ -342,6 +350,9 @@ export class SpanPresenter extends BasePresenter {
const run = await this._prisma.taskRun.findFirst({
select: {
traceId: true,
createdAt: true,
completedAt: true,
taskEventStore: true,
},
where: {
friendlyId: runFriendlyId,
Expand All @@ -352,7 +363,13 @@ export class SpanPresenter extends BasePresenter {
return;
}

const span = await eventRepository.getSpan(spanId, run.traceId);
const span = await eventRepository.getSpan(
getTaskEventStoreTableForRun(run),
spanId,
run.traceId,
run.createdAt,
run.completedAt ?? undefined
);

if (!span) {
return;
Expand Down
35 changes: 0 additions & 35 deletions ....orgs.$organizationSlug.projects.v3.$projectParam.traces.$traceId.spans.$spanId/route.tsx
View file
Open in desktop

This file was deleted.

28 changes: 0 additions & 28 deletions ...rgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,6 @@ import { RunTag } from "~/components/runs/v3/RunTag";
import { SpanEvents } from "~/components/runs/v3/SpanEvents";
import { SpanTitle } from "~/components/runs/v3/SpanTitle";
import { TaskRunAttemptStatusCombo } from "~/components/runs/v3/TaskRunAttemptStatus";
import { TaskRunsTable } from "~/components/runs/v3/TaskRunsTable";
import { TaskRunStatusCombo } from "~/components/runs/v3/TaskRunStatus";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
Expand All @@ -57,16 +56,13 @@ import { cn } from "~/utils/cn";
import { formatCurrencyAccurate } from "~/utils/numberFormatter";
import {
v3BatchPath,
v3BatchRunsPath,
v3RunDownloadLogsPath,
v3RunPath,
v3RunSpanPath,
v3RunsPath,
v3SchedulePath,
v3SpanParamsSchema,
v3TraceSpanPath,
} from "~/utils/pathBuilder";
import { SpanLink } from "~/v3/eventRepository.server";

export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
Expand Down Expand Up @@ -1152,27 +1148,3 @@ function classNameForState(state: TimelineState) {
}
}
}

function SpanLinkElement({ link }: { link: SpanLink }) {
const organization = useOrganization();
const project = useProject();

switch (link.type) {
case "run": {
return (
<TextLink to={v3RunPath(organization, project, { friendlyId: link.runId })}>
{link.title}
</TextLink>
);
}
case "span": {
return (
<TextLink to={v3TraceSpanPath(organization, project, link.traceId, link.spanId)}>
{link.title}
</TextLink>
);
}
}

return null;
}
8 changes: 7 additions & 1 deletion apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import { createGzip } from "zlib";
import { Readable } from "stream";
import { formatDurationMilliseconds } from "@trigger.dev/core/v3/utils/durations";
import { getDateFromNanoseconds } from "~/utils/taskEvent";
import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";

export async function loader({ params, request }: LoaderFunctionArgs) {
const userId = await requireUserId(request);
Expand All @@ -31,7 +32,12 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
return new Response("Not found", { status: 404 });
}

const runEvents = await eventRepository.getRunEvents(run.friendlyId);
const runEvents = await eventRepository.getRunEvents(
getTaskEventStoreTableForRun(run),
run.friendlyId,
run.createdAt,
run.completedAt ?? undefined
);

// Create a Readable stream from the runEvents array
const readable = new Readable({
Expand Down
9 changes: 0 additions & 9 deletions apps/webapp/app/utils/pathBuilder.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -396,15 +396,6 @@ export function v3RunSpanPath(
return `${v3RunPath(organization, project, run)}?span=${span.spanId}`;
}

export function v3TraceSpanPath(
organization: OrgForPath,
project: ProjectForPath,
traceId: string,
spanId: string
) {
return `${v3ProjectPath(organization, project)}/traces/${traceId}/spans/${spanId}`;
}

export function v3RunStreamingPath(
organization: OrgForPath,
project: ProjectForPath,
Expand Down
2 changes: 0 additions & 2 deletions apps/webapp/app/utils/taskEvent.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,6 @@ export function prepareTrace(events: TaskEvent[]): TraceSummary | undefined {
);

const span = {
recordId: event.id,
id: event.spanId,
parentId: event.parentId ?? undefined,
runId: event.runId,
Expand Down Expand Up @@ -247,7 +246,6 @@ export function createSpanFromEvent(events: TaskEvent[], event: PreparedEvent) {
}

const span = {
recordId: event.id,
id: event.spanId,
parentId: event.parentId ?? undefined,
runId: event.runId,
Expand Down
Loading