Implement simple 7 day log retention and blank states (#976)

Author: ericallam

apps/webapp/app/env.server.ts

@@ -101,6 +101,9 @@ const EnvironmentSchema = z.object({
   OBJECT_STORE_BASE_URL: z.string().optional(),
   OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),
   OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
+  EVENTS_BATCH_SIZE: z.coerce.number().int().default(100),
+  EVENTS_BATCH_INTERVAL: z.coerce.number().int().default(1000),
+  EVENTS_DEFAULT_LOG_RETENTION: z.coerce.number().int().default(7),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/presenters/v3/RunPresenter.server.ts

@@ -6,7 +6,7 @@ import { eventRepository } from "~/v3/eventRepository.server";
 
 type Result = Awaited<ReturnType<RunPresenter["call"]>>;
 export type Run = Result["run"];
-export type RunEvent = Result["events"][0];
+export type RunEvent = NonNullable<Result["trace"]>["events"][0];
 
 export class RunPresenter {
   #prismaClient: PrismaClient;
@@ -64,7 +64,19 @@ export class RunPresenter {
     const traceSummary = await eventRepository.getTraceSummary(run.traceId);
 
     if (!traceSummary) {
-      throw new Error("Trace not found");
+      return {
+        run: {
+          number: run.number,
+          friendlyId: run.friendlyId,
+          environment: {
+            type: run.runtimeEnvironment.type,
+            slug: run.runtimeEnvironment.slug,
+            userId: run.runtimeEnvironment.orgMember?.user.id,
+            userName: getUsername(run.runtimeEnvironment.orgMember?.user),
+          },
+        },
+        trace: undefined,
+      };
     }
 
     //this tree starts at the passed in span (hides parent elements if there are any)
@@ -115,12 +127,14 @@ export class RunPresenter {
           userName: getUsername(run.runtimeEnvironment.orgMember?.user),
         },
       },
-      rootSpanStatus,
-      events: events,
-      parentRunFriendlyId:
-        tree?.id === traceSummary.rootSpan.id ? undefined : traceSummary.rootSpan.runId,
-      duration: totalDuration,
-      rootStartedAt: tree?.data.startTime,
+      trace: {
+        rootSpanStatus,
+        events: events,
+        parentRunFriendlyId:
+          tree?.id === traceSummary.rootSpan.id ? undefined : traceSummary.rootSpan.runId,
+        duration: totalDuration,
+        rootStartedAt: tree?.data.startTime,
+      },
     };
   }
 }

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -3,7 +3,7 @@ import { PrismaClient, prisma } from "~/db.server";
 import { eventRepository } from "~/v3/eventRepository.server";
 
 type Result = Awaited<ReturnType<SpanPresenter["call"]>>;
-export type Span = Result["event"];
+export type Span = NonNullable<Result>["event"];
 
 export class SpanPresenter {
   #prismaClient: PrismaClient;
@@ -36,7 +36,7 @@ export class SpanPresenter {
     const span = await eventRepository.getSpan(spanId);
 
     if (!span) {
-      throw new Error("Event not found");
+      return;
     }
 
     const output =

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -21,6 +21,7 @@ import { TaskPath } from "~/components/runs/v3/TaskPath";
 import { TaskRunAttemptStatusCombo } from "~/components/runs/v3/TaskRunAttemptStatus";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
+import { redirectWithErrorMessage } from "~/models/message.server";
 import { SpanPresenter } from "~/presenters/v3/SpanPresenter.server";
 import { requireUserId } from "~/services/session.server";
 import { cn } from "~/utils/cn";
@@ -29,7 +30,7 @@ import { SpanLink } from "~/v3/eventRepository.server";
 
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
-  const { projectParam, organizationSlug, spanParam } = v3SpanParamsSchema.parse(params);
+  const { projectParam, organizationSlug, runParam, spanParam } = v3SpanParamsSchema.parse(params);
 
   const presenter = new SpanPresenter();
   const span = await presenter.call({
@@ -39,6 +40,15 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     spanId: spanParam,
   });
 
+  if (!span) {
+    // We're going to redirect to the run page
+    return redirectWithErrorMessage(
+      v3RunPath({ slug: organizationSlug }, { slug: projectParam }, { friendlyId: runParam }),
+      request,
+      `Event not found.`
+    );
+  }
+
   return typedjson({ span });
 };
 

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam/route.tsx

@@ -17,8 +17,9 @@ import { useEffect, useRef, useState } from "react";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
 import { ShowParentIcon, ShowParentIconSelected } from "~/assets/icons/ShowParentIcon";
 import tileBgPath from "~/assets/images/[email protected]";
+import { BlankstateInstructions } from "~/components/BlankstateInstructions";
 import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
-import { PageBody } from "~/components/layout/AppLayout";
+import { MainCenteredContainer, PageBody } from "~/components/layout/AppLayout";
 import { Badge } from "~/components/primitives/Badge";
 import { LinkButton } from "~/components/primitives/Buttons";
 import { Input } from "~/components/primitives/Input";
@@ -85,29 +86,56 @@ function getSpanId(path: string): string | undefined {
 }
 
 export default function Page() {
-  const {
-    run,
-    events,
-    parentRunFriendlyId,
-    resizeSettings,
-    duration,
-    rootSpanStatus,
-    rootStartedAt,
-  } = useTypedLoaderData<typeof loader>();
+  const { run, trace, resizeSettings } = useTypedLoaderData<typeof loader>();
   const navigate = useNavigate();
   const organization = useOrganization();
   const pathName = usePathName();
   const project = useProject();
   const user = useUser();
 
+  const usernameForEnv = user.id !== run.environment.userId ? run.environment.userName : undefined;
+
+  if (!trace) {
+    return (
+      <>
+        <NavBar>
+          <PageTitle
+            backButton={{
+              to: v3RunsPath(organization, project),
+              text: "Runs",
+            }}
+            title={`Run #${run.number}`}
+          />
+          <PageAccessories>
+            <EnvironmentLabel
+              size="large"
+              environment={run.environment}
+              userName={usernameForEnv}
+            />
+          </PageAccessories>
+        </NavBar>
+        <PageBody>
+          <MainCenteredContainer className="max-w-prose">
+            <BlankstateInstructions title="These logs have taken a walk">
+              <Paragraph spacing>
+                Looks like the logs from this run have wandered off after their 7-day stay. We tidy
+                up older logs to keep things running smoothly.
+              </Paragraph>
+            </BlankstateInstructions>
+          </MainCenteredContainer>
+        </PageBody>
+      </>
+    );
+  }
+
+  const { events, parentRunFriendlyId, duration, rootSpanStatus, rootStartedAt } = trace;
+
   const selectedSpanId = getSpanId(pathName);
 
   const changeToSpan = useDebounce((selectedSpan: string) => {
     navigate(v3RunSpanPath(organization, project, run, { spanId: selectedSpan }));
   }, 250);
 
-  const usernameForEnv = user.id !== run.environment.userId ? run.environment.userName : undefined;
-
   const revalidator = useRevalidator();
   const streamedEvents = useEventSource(v3RunStreamingPath(organization, project, run), {
     event: "message",

apps/webapp/app/services/worker.server.ts

@@ -34,6 +34,7 @@ import { ResumeTaskRunDependenciesService } from "~/v3/services/resumeTaskRunDep
 import { ResumeBatchRunService } from "~/v3/services/resumeBatchRun.server";
 import { ResumeTaskDependencyService } from "~/v3/services/resumeTaskDependency.server";
 import { TimeoutDeploymentService } from "~/v3/services/timeoutDeployment.server";
+import { eventRepository } from "~/v3/eventRepository.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -250,6 +251,13 @@ function getWorkerQueue() {
           });
         },
       },
+      // Run this every hour at the 13 minute mark
+      purgeOldTaskEvents: {
+        pattern: "47 * * * *",
+        handler: async (payload, job) => {
+          await eventRepository.truncateEvents();
+        },
+      },
     },
     tasks: {
       "events.invokeDispatcher": {
@@ -301,8 +309,8 @@ function getWorkerQueue() {
                 graphileJob.id,
                 payload.orphanedEvents
                   ? {
-                    event: payload.orphanedEvents,
-                  }
+                      event: payload.orphanedEvents,
+                    }
                   : undefined
               );
               break;

apps/webapp/app/utils/pathBuilder.ts

@@ -82,7 +82,7 @@ export const v3RunParamsSchema = ProjectParamSchema.extend({
   runParam: z.string(),
 });
 
-export const v3SpanParamsSchema = ProjectParamSchema.extend({
+export const v3SpanParamsSchema = v3RunParamsSchema.extend({
   spanParam: z.string(),
 });
 

apps/webapp/app/v3/eventRepository.server.ts

@@ -87,6 +87,7 @@ export type EventRepoConfig = {
   batchSize: number;
   batchInterval: number;
   redis: RedisOptions;
+  retentionInDays: number;
 };
 
 export type QueryOptions = Prisma.TaskEventWhereInput;
@@ -723,6 +724,16 @@ export class EventRepository {
     return this._randomIdGenerator.generateSpanId();
   }
 
+  public async truncateEvents() {
+    await this.db.taskEvent.deleteMany({
+      where: {
+        createdAt: {
+          lt: new Date(Date.now() - this._config.retentionInDays * 24 * 60 * 60 * 1000),
+        },
+      },
+    });
+  }
+
   /**
    * Returns a deterministically random 8-byte span ID formatted/encoded as a 16 lowercase hex
    * characters corresponding to 64 bits, based on the trace ID and seed.
@@ -743,8 +754,9 @@ export class EventRepository {
 }
 
 export const eventRepository = new EventRepository(prisma, {
-  batchSize: 100,
-  batchInterval: 1000,
+  batchSize: env.EVENTS_BATCH_SIZE,
+  batchInterval: env.EVENTS_BATCH_INTERVAL,
+  retentionInDays: env.EVENTS_DEFAULT_LOG_RETENTION,
   redis: {
     port: env.REDIS_PORT,
     host: env.REDIS_HOST,