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

v3: crash observability (#1002)

Browse Source 
* indexing resource limits

* refactor and storage limits

* handle and display worker crashes

* enable prod cancellation

* fix reconnect delay

* improve crash messages

* log and display crash events

* changeset
This commit is contained in:
nicktrn
2024-04-05 13:00:20 +01:00
committed by GitHub
parent 6642228f26
commit 7ea8532cce
20 changed files with 974 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/late-icons-lie.md
+7
View File
@@ -0,0 +1,7 @@
---
"@trigger.dev/core-apps": patch
"trigger.dev": patch
"@trigger.dev/core": patch
---
Display errors for runs and deployments
apps/kubernetes-provider/package.json
+1
View File
@@ -19,6 +19,7 @@
"@kubernetes/client-node": "^0.20.0",
"@trigger.dev/core": "workspace:*",
"@trigger.dev/core-apps": "workspace:*",
"p-queue": "^8.0.1",
"socket.io-client": "^4.7.4"
},
"devDependencies": {
apps/kubernetes-provider/src/index.ts
+150 -132
View File
@@ -9,6 +9,7 @@ import {
} from "@trigger.dev/core-apps";
import { Machine, PostStartCauses, PreStopCauses, EnvironmentType } from "@trigger.dev/core/v3";
import { randomUUID } from "crypto";
import { TaskMonitor } from "./taskMonitor";
const RUNTIME_ENV = process.env.KUBERNETES_PORT ? "kubernetes" : "local";
const NODE_NAME = process.env.NODE_NAME || "local";
@@ -24,6 +25,10 @@ type Namespace = {
};
};
type ComputeResources = {
[K in "cpu" | "memory" | "ephemeral-storage"]?: string;
};
class KubernetesTaskOperations implements TaskOperations {
#namespace: Namespace;
#k8sApi: {
@@ -55,25 +60,15 @@ class KubernetesTaskOperations implements TaskOperations {
template: {
metadata: {
labels: {
...this.#getSharedLabels(opts),
app: "task-index",
"app.kubernetes.io/part-of": "trigger-worker",
"app.kubernetes.io/component": "index",
env: opts.envId,
envtype: this.#envTypeToLabelValue(opts.envType),
org: opts.orgId,
project: opts.projectId,
deployment: opts.deploymentId,
},
},
spec: {
restartPolicy: "Never",
imagePullSecrets: [
{
name: "registry-trigger",
},
],
nodeSelector: {
nodetype: "worker",
},
...this.#defaultPodSpec,
containers: [
{
name: this.#getIndexContainerName(opts.shortCode),
@@ -83,12 +78,13 @@ class KubernetesTaskOperations implements TaskOperations {
containerPort: 8000,
},
],
// resources: {
// limits: {
// cpu: "100m",
// memory: "50Mi",
// },
// },
resources: {
limits: {
cpu: "250m",
memory: "0.5G",
"ephemeral-storage": "2Gi",
},
},
lifecycle: {
preStop: {
exec: {
@@ -97,10 +93,7 @@ class KubernetesTaskOperations implements TaskOperations {
},
},
env: [
{
name: "DEBUG",
value: "true",
},
...this.#getSharedEnv(opts.envId),
{
name: "INDEX_TASKS",
value: "true",
@@ -113,42 +106,6 @@ class KubernetesTaskOperations implements TaskOperations {
name: "TRIGGER_API_URL",
value: opts.apiUrl,
},
{
name: "TRIGGER_ENV_ID",
value: opts.envId,
},
{
name: "OTEL_EXPORTER_OTLP_ENDPOINT",
value: OTEL_EXPORTER_OTLP_ENDPOINT,
},
{
name: "HTTP_SERVER_PORT",
value: "8000",
},
{
name: "POD_NAME",
valueFrom: {
fieldRef: {
fieldPath: "metadata.name",
},
},
},
{
name: "COORDINATOR_HOST",
valueFrom: {
fieldRef: {
fieldPath: "status.hostIP",
},
},
},
{
name: "MACHINE_NAME",
valueFrom: {
fieldRef: {
fieldPath: "spec.nodeName",
},
},
},
],
},
],
@@ -167,26 +124,15 @@ class KubernetesTaskOperations implements TaskOperations {
name: this.#getRunContainerName(opts.runId),
namespace: this.#namespace.metadata.name,
labels: {
...this.#getSharedLabels(opts),
app: "task-run",
"app.kubernetes.io/part-of": "trigger-worker",
"app.kubernetes.io/component": "create",
env: opts.envId,
envtype: this.#envTypeToLabelValue(opts.envType),
org: opts.orgId,
project: opts.projectId,
run: opts.runId,
},
},
spec: {
restartPolicy: "Never",
imagePullSecrets: [
{
name: "registry-trigger",
},
],
nodeSelector: {
nodetype: "worker",
},
...this.#defaultPodSpec,
containers: [
{
name: this.#getRunContainerName(opts.runId),
@@ -197,7 +143,13 @@ class KubernetesTaskOperations implements TaskOperations {
},
],
resources: {
limits: this.#getResourcesFromMachineConfig(opts.machine),
requests: {
...this.#defaultResourceRequests,
},
limits: {
...this.#defaultResourceLimits,
...this.#getResourcesFromMachineConfig(opts.machine),
},
},
lifecycle: {
postStart: {
@@ -212,54 +164,11 @@ class KubernetesTaskOperations implements TaskOperations {
},
},
env: [
{
name: "DEBUG",
value: "true",
},
{
name: "HTTP_SERVER_PORT",
value: "8000",
},
{
name: "TRIGGER_ENV_ID",
value: opts.envId,
},
...this.#getSharedEnv(opts.envId),
{
name: "TRIGGER_RUN_ID",
value: opts.runId,
},
{
name: "TRIGGER_WORKER_VERSION",
value: opts.version,
},
{
name: "OTEL_EXPORTER_OTLP_ENDPOINT",
value: OTEL_EXPORTER_OTLP_ENDPOINT,
},
{
name: "POD_NAME",
valueFrom: {
fieldRef: {
fieldPath: "metadata.name",
},
},
},
{
name: "COORDINATOR_HOST",
valueFrom: {
fieldRef: {
fieldPath: "status.hostIP",
},
},
},
{
name: "NODE_NAME",
valueFrom: {
fieldRef: {
fieldPath: "spec.nodeName",
},
},
},
],
volumeMounts: [
{
@@ -288,27 +197,16 @@ class KubernetesTaskOperations implements TaskOperations {
name: `${this.#getRunContainerName(opts.runId)}-${randomUUID().slice(0, 8)}`,
namespace: this.#namespace.metadata.name,
labels: {
...this.#getSharedLabels(opts),
app: "task-run",
"app.kubernetes.io/part-of": "trigger-worker",
"app.kubernetes.io/component": "restore",
env: opts.envId,
envtype: this.#envTypeToLabelValue(opts.envType),
org: opts.orgId,
project: opts.projectId,
run: opts.runId,
checkpoint: opts.checkpointId,
},
},
spec: {
restartPolicy: "Never",
imagePullSecrets: [
{
name: "registry-trigger",
},
],
nodeSelector: {
nodetype: "worker",
},
...this.#defaultPodSpec,
initContainers: [
{
name: "pull-base-image",
@@ -348,7 +246,13 @@ class KubernetesTaskOperations implements TaskOperations {
},
],
resources: {
limits: this.#getResourcesFromMachineConfig(opts.machine),
requests: {
...this.#defaultResourceRequests,
},
limits: {
...this.#defaultResourceLimits,
...this.#getResourcesFromMachineConfig(opts.machine),
},
},
lifecycle: {
postStart: {
@@ -406,7 +310,90 @@ class KubernetesTaskOperations implements TaskOperations {
}
}
#getResourcesFromMachineConfig(config: Machine) {
get #defaultPodSpec(): Omit<k8s.V1PodSpec, "containers"> {
return {
restartPolicy: "Never",
automountServiceAccountToken: false,
imagePullSecrets: [
{
name: "registry-trigger",
},
],
nodeSelector: {
nodetype: "worker",
},
};
}
get #defaultResourceRequests(): ComputeResources {
return {
"ephemeral-storage": "2Gi",
};
}
get #defaultResourceLimits(): ComputeResources {
return {
"ephemeral-storage": "10Gi",
};
}
#getSharedEnv(envId: string): k8s.V1EnvVar[] {
return [
{
name: "TRIGGER_ENV_ID",
value: envId,
},
{
name: "DEBUG",
value: process.env.DEBUG ? "1" : "0",
},
{
name: "HTTP_SERVER_PORT",
value: "8000",
},
{
name: "OTEL_EXPORTER_OTLP_ENDPOINT",
value: OTEL_EXPORTER_OTLP_ENDPOINT,
},
{
name: "POD_NAME",
valueFrom: {
fieldRef: {
fieldPath: "metadata.name",
},
},
},
{
name: "COORDINATOR_HOST",
valueFrom: {
fieldRef: {
fieldPath: "status.hostIP",
},
},
},
{
name: "MACHINE_NAME",
valueFrom: {
fieldRef: {
fieldPath: "spec.nodeName",
},
},
},
];
}
#getSharedLabels(
opts: TaskOperationsIndexOptions | TaskOperationsCreateOptions | TaskOperationsRestoreOptions
): Record<string, string> {
return {
env: opts.envId,
envtype: this.#envTypeToLabelValue(opts.envType),
org: opts.orgId,
project: opts.projectId,
};
}
#getResourcesFromMachineConfig(config: Machine): ComputeResources {
return {
cpu: `${config.cpu}`,
memory: `${config.memory}G`,
@@ -516,3 +503,34 @@ const provider = new ProviderShell({
});
provider.listen();
const taskMonitor = new TaskMonitor({
runtimeEnv: RUNTIME_ENV,
onIndexFailure: async (deploymentId, failureInfo) => {
logger.log("Indexing failed", { deploymentId, failureInfo });
try {
provider.platformSocket.send("INDEXING_FAILED", {
deploymentId,
error: {
name: `Crashed with exit code ${failureInfo.exitCode}`,
message: failureInfo.reason,
stack: failureInfo.logs,
},
});
} catch (error) {
logger.error(error);
}
},
onRunFailure: async (runId, failureInfo) => {
logger.log("Run failed:", { runId, failureInfo });
try {
provider.platformSocket.send("WORKER_CRASHED", { runId, ...failureInfo });
} catch (error) {
logger.error(error);
}
},
});
taskMonitor.start();
apps/kubernetes-provider/src/taskMonitor.ts
+436
View File
@@ -0,0 +1,436 @@
import * as k8s from "@kubernetes/client-node";
import { SimpleLogger } from "@trigger.dev/core-apps";
import { setTimeout } from "timers/promises";
import PQueue from "p-queue";
type IndexFailureHandler = (
deploymentId: string,
failureInfo: {
exitCode: number;
reason: string;
logs: string;
}
) => Promise<any>;
type RunFailureHandler = (
runId: string,
failureInfo: {
exitCode: number;
reason: string;
logs: string;
}
) => Promise<any>;
type TaskMonitorOptions = {
runtimeEnv: "local" | "kubernetes";
onIndexFailure?: IndexFailureHandler;
onRunFailure?: RunFailureHandler;
namespace?: string;
};
export class TaskMonitor {
#enabled = false;
#logger = new SimpleLogger("[TaskMonitor]");
#taskInformer: ReturnType<typeof k8s.makeInformer<k8s.V1Pod>>;
#processedPods = new Map<string, number>();
#queue = new PQueue({ concurrency: 10 });
#k8sClient: {
core: k8s.CoreV1Api;
kubeConfig: k8s.KubeConfig;
};
private namespace = "default";
private fieldSelector = "status.phase=Failed";
private labelSelector = "app in (task-index, task-run)";
constructor(private opts: TaskMonitorOptions) {
this.#k8sClient = this.#createK8sClient();
this.#taskInformer = this.#createTaskInformer();
this.#taskInformer.on("connect", this.#onInformerConnected.bind(this));
this.#taskInformer.on("error", this.#onInformerError.bind(this));
this.#taskInformer.on("update", this.#enqueueOnPodUpdated.bind(this));
}
#createTaskInformer() {
const listTasks = () =>
this.#k8sClient.core.listNamespacedPod(
this.namespace,
undefined,
undefined,
undefined,
this.fieldSelector,
this.labelSelector
);
// Uses watch with local caching
// https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes
const informer = k8s.makeInformer(
this.#k8sClient.kubeConfig,
`/api/v1/namespaces/${this.namespace}/pods`,
listTasks,
this.labelSelector,
this.fieldSelector
);
return informer;
}
async #onInformerConnected() {
this.#logger.log("Connected");
}
async #onInformerError(error: any) {
this.#logger.error("Error:", error);
// Automatic reconnect
await setTimeout(2_000);
this.#taskInformer.start();
}
#enqueueOnPodUpdated(pod: k8s.V1Pod) {
this.#queue.add(async () => {
try {
// It would be better to only pass the cache key, but the pod may already be removed from the cache by the time we process it
await this.#onPodUpdated(pod);
} catch (error) {
this.#logger.error("Caught onPodUpdated() error:", error);
}
});
}
async #onPodUpdated(pod: k8s.V1Pod) {
this.#logger.debug(`Updated: ${pod.metadata?.name}`);
this.#logger.debug("Updated", JSON.stringify(pod, null, 2));
// We only care about failures
if (pod.status?.phase !== "Failed") {
return;
}
const podName = pod.metadata?.name;
if (!podName) {
this.#logger.error("Pod is nameless", { pod });
return;
}
const containerStatus = pod.status.containerStatuses?.[0];
if (!containerStatus?.state) {
this.#logger.error("Pod failed, but container status doesn't have state", {
status: pod.status,
});
return;
}
if (this.#processedPods.has(podName)) {
this.#logger.debug("Pod update already processed", {
podName,
timestamp: this.#processedPods.get(podName),
});
return;
}
this.#processedPods.set(podName, Date.now());
const podStatus = this.#getPodStatusSummary(pod.status);
const containerState = this.#getContainerStateSummary(containerStatus.state);
const rawLogs = await this.#getLogTail(podName);
this.#logger.log(`${podName} failed with:`, {
podStatus,
containerState,
rawLogs,
});
const exitCode = containerState.exitCode ?? -1;
const rawReason = podStatus.reason ?? containerState.reason ?? "";
const message = podStatus.message ?? containerState.message ?? "";
let reason = rawReason || "Unknown error";
let logs = rawLogs || "";
switch (rawReason) {
case "Error":
reason = "Unknown error.";
break;
case "Evicted":
if (message.startsWith("Pod ephemeral local storage usage")) {
reason = "Storage limit exceeded.";
} else if (message) {
reason = `Evicted: ${message}`;
} else {
reason = "Evicted for unknown reason.";
}
if (logs.startsWith("failed to try resolving symlinks")) {
logs = "";
}
break;
case "OOMKilled":
reason = "Out of memory! Try increasing the memory on this task.";
break;
default:
break;
}
const failureInfo = {
exitCode,
reason,
logs,
};
const app = pod.metadata?.labels?.app;
switch (app) {
case "task-index":
const deploymentId = pod.metadata?.labels?.deployment;
if (!deploymentId) {
this.#logger.error("Index is missing ID", { pod });
return;
}
if (this.opts.onIndexFailure) {
await this.opts.onIndexFailure(deploymentId, failureInfo);
}
break;
case "task-run":
const runId = pod.metadata?.labels?.run;
if (!runId) {
this.#logger.error("Run is missing ID", { pod });
return;
}
if (this.opts.onRunFailure) {
await this.opts.onRunFailure(runId, failureInfo);
}
break;
default:
this.#logger.error("Pod has invalid app label", { pod });
return;
}
await this.#deletePod(podName);
}
async #getLogTail(podName: string) {
try {
const logs = await this.#k8sClient.core.readNamespacedPodLog(
podName,
this.namespace,
undefined,
undefined,
undefined,
1024, // limitBytes
undefined,
undefined,
undefined,
20 // tailLines
);
const responseBody = logs.body ?? "";
if (responseBody.startsWith("unable to retrieve container logs")) {
return "";
}
// Type is wrong, body may be undefined
return responseBody;
} catch (error) {
this.#logger.error("Log tail error:", error instanceof Error ? error.message : "unknown");
return "";
}
}
#getPodStatusSummary(status: k8s.V1PodStatus) {
return {
reason: status.reason,
message: status.message,
};
}
#getContainerStateSummary(state: k8s.V1ContainerState) {
return {
reason: state.terminated?.reason,
exitCode: state.terminated?.exitCode,
message: state.terminated?.message,
};
}
#createK8sClient() {
const kubeConfig = new k8s.KubeConfig();
if (this.opts.runtimeEnv === "local") {
kubeConfig.loadFromDefault();
} else if (this.opts.runtimeEnv === "kubernetes") {
kubeConfig.loadFromCluster();
} else {
throw new Error(`Unsupported runtime environment: ${this.opts.runtimeEnv}`);
}
return {
core: kubeConfig.makeApiClient(k8s.CoreV1Api),
kubeConfig: kubeConfig,
};
}
#isRecord(candidate: unknown): candidate is Record<string, unknown> {
if (typeof candidate !== "object" || candidate === null) {
return false;
} else {
return true;
}
}
#logK8sError(err: unknown, debugOnly = false) {
if (debugOnly) {
this.#logger.debug("K8s API Error", err);
} else {
this.#logger.error("K8s API Error", err);
}
}
#handleK8sError(err: unknown) {
if (!this.#isRecord(err) || !this.#isRecord(err.body)) {
this.#logK8sError(err);
return;
}
this.#logK8sError(err, true);
if (typeof err.body.message === "string") {
this.#logK8sError({ message: err.body.message });
return;
}
this.#logK8sError({ body: err.body });
}
#printStats(includeMoreDetails = false) {
this.#logger.log("Stats:", {
cacheSize: this.#taskInformer.list().length,
totalProcessed: this.#processedPods.size,
...(includeMoreDetails && {
processedPods: this.#processedPods,
}),
});
}
async #deletePod(name: string) {
this.#logger.debug("Deleting pod:", name);
await this.#k8sClient.core
.deleteNamespacedPod(name, this.namespace)
.catch(this.#handleK8sError.bind(this));
}
async start() {
this.#enabled = true;
const interval = setInterval(() => {
if (!this.#enabled) {
clearInterval(interval);
return;
}
this.#printStats();
}, 300_000);
await this.#taskInformer.start();
// this.#launchTests();
}
async stop() {
if (!this.#enabled) {
return;
}
this.#enabled = false;
this.#logger.log("Shutting down..");
await this.#taskInformer.stop();
this.#printStats(true);
}
async #launchTests() {
const createPod = async (
container: k8s.V1Container,
name: string,
labels?: Record<string, string>
) => {
this.#logger.log("Creating pod:", name);
const pod = {
metadata: {
name,
labels,
},
spec: {
restartPolicy: "Never",
automountServiceAccountToken: false,
terminationGracePeriodSeconds: 1,
containers: [container],
},
} satisfies k8s.V1Pod;
await this.#k8sClient.core
.createNamespacedPod(this.namespace, pod)
.catch(this.#handleK8sError.bind(this));
};
const createOomPod = async (name: string, labels?: Record<string, string>) => {
const container = {
name,
image: "polinux/stress",
resources: {
limits: {
memory: "100Mi",
},
},
command: ["stress"],
args: ["--vm", "1", "--vm-bytes", "150M", "--vm-hang", "1"],
} satisfies k8s.V1Container;
await createPod(container, name, labels);
};
const createNonZeroExitPod = async (name: string, labels?: Record<string, string>) => {
const container = {
name,
image: "busybox",
command: ["sh"],
args: ["-c", "exit 1"],
} satisfies k8s.V1Container;
await createPod(container, name, labels);
};
const createOoDiskPod = async (name: string, labels?: Record<string, string>) => {
const container = {
name,
image: "busybox",
command: ["sh"],
args: [
"-c",
"echo creating huge-file..; head -c 1000m /dev/zero > huge-file; ls -lh huge-file; sleep infinity",
],
resources: {
limits: {
"ephemeral-storage": "500Mi",
},
},
} satisfies k8s.V1Container;
await createPod(container, name, labels);
};
await createNonZeroExitPod("non-zero-exit-task", { app: "task-run", run: "123" });
await createOomPod("oom-task", { app: "task-index", deployment: "456" });
await createOoDiskPod("ood-task", { app: "task-run", run: "abc" });
}
}
apps/webapp/app/components/runs/v3/RunFilters.tsx
+1
View File
@@ -35,6 +35,7 @@ export const allTaskRunStatuses = [
"COMPLETED_WITH_ERRORS",
"INTERRUPTED",
"SYSTEM_FAILURE",
"CRASHED",
] as TaskRunStatusType[];
export const TaskAttemptStatus = z.nativeEnum(TaskRunStatus);
apps/webapp/app/components/runs/v3/SpanEvents.tsx
+5 -1
View File
@@ -66,7 +66,11 @@ function SpanEventError({
}) {
return (
<div className="flex flex-col gap-2 rounded-sm border border-rose-500/50 p-3">
<SpanEventHeader title={"Error"} time={spanEvent.time} titleClassName="text-rose-500" />
<SpanEventHeader
title={exception.type ?? "Error"}
time={spanEvent.time}
titleClassName="text-rose-500"
/>
{exception.message && <Callout variant="error">{exception.message}</Callout>}
{exception.stacktrace && (
<CodeBlock
apps/webapp/app/components/runs/v3/TaskRunStatus.tsx
+8 -1
View File
@@ -3,7 +3,7 @@ import {
BoltSlashIcon,
BugAntIcon,
CheckCircleIcon,
ClockIcon,
FireIcon,
NoSymbolIcon,
PauseCircleIcon,
RectangleStackIcon,
@@ -25,6 +25,7 @@ const taskRunStatusDescriptions: Record<TaskRunStatus, string> = {
INTERRUPTED: "Task has failed because it was interrupted",
SYSTEM_FAILURE: "Task has failed due to a system failure",
PAUSED: "Task has been paused by the user",
CRASHED: "Task has crashed and won't be retried",
};
export function descriptionForTaskRunStatus(status: TaskRunStatus): string {
@@ -80,6 +81,8 @@ export function TaskRunStatusIcon({
return <XCircleIcon className={cn(runStatusClassNameColor(status), className)} />;
case "SYSTEM_FAILURE":
return <BugAntIcon className={cn(runStatusClassNameColor(status), className)} />;
case "CRASHED":
return <FireIcon className={cn(runStatusClassNameColor(status), className)} />;
default: {
const _exhaustiveCheck: never = status;
@@ -109,6 +112,8 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
return "text-error";
case "SYSTEM_FAILURE":
return "text-error";
case "CRASHED":
return "text-error";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
@@ -138,6 +143,8 @@ export function runStatusTitle(status: TaskRunStatus): string {
return "Failed";
case "SYSTEM_FAILURE":
return "System failure";
case "CRASHED":
return "Crashed";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
apps/webapp/app/v3/eventRepository.server.ts
+42
View File
@@ -3,6 +3,7 @@ import { RandomIdGenerator } from "@opentelemetry/sdk-trace-base";
import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions";
import {
ExceptionEventProperties,
ExceptionSpanEvent,
PRIMARY_VARIANT,
SemanticInternalAttributes,
SpanEvent,
@@ -255,6 +256,47 @@ export class EventRepository {
});
}
async crashEvent({
event,
crashedAt,
exception,
}: {
event: TaskEventRecord;
crashedAt: Date;
exception: ExceptionEventProperties;
}) {
if (!event.isPartial) {
return;
}
await this.insertImmediate({
...omit(event, "id"),
isPartial: false,
isError: true,
isCancelled: false,
status: "ERROR",
links: event.links ?? [],
events: [
{
name: "exception",
time: crashedAt,
properties: {
exception,
},
} satisfies ExceptionSpanEvent,
...((event.events as any[]) ?? []),
],
duration: calculateDurationFromStart(event.startTime, crashedAt),
properties: event.properties as Attributes,
metadata: event.metadata as Attributes,
style: event.style as Attributes,
output: event.output as Attributes,
outputType: event.outputType,
payload: event.payload as Attributes,
payloadType: event.payloadType,
});
}
async queryEvents(queryOptions: QueryOptions): Promise<TaskEventRecord[]> {
return await this.db.taskEvent.findMany({
where: queryOptions,
apps/webapp/app/v3/handleSocketIo.server.ts
+24 -1
View File
@@ -21,6 +21,7 @@ import { ResumeAttemptService } from "./services/resumeAttempt.server";
import { DeploymentIndexFailed } from "./services/deploymentIndexFailed.server";
import { Redis } from "ioredis";
import { createAdapter } from "@socket.io/redis-adapter";
import { CrashTaskRunService } from "./services/crashTaskRun.server";
export const socketIo = singleton("socketIo", initalizeIoServer);
@@ -135,7 +136,7 @@ function createCoordinatorNamespace(io: Server) {
await service.call(message.deploymentId, message.error);
} catch (e) {
logger.error("Error while indexing failed", { error: e });
logger.error("Error while indexing", { error: e });
}
},
},
@@ -151,6 +152,28 @@ function createProviderNamespace(io: Server) {
authToken: env.PROVIDER_SECRET,
clientMessages: ProviderToPlatformMessages,
serverMessages: PlatformToProviderMessages,
handlers: {
WORKER_CRASHED: async (message) => {
try {
const service = new CrashTaskRunService();
await service.call(message.runId, {
...message,
});
} catch (error) {
logger.error("Error while handling crashed worker", { error });
}
},
INDEXING_FAILED: async (message) => {
try {
const service = new DeploymentIndexFailed();
await service.call(message.deploymentId, message.error);
} catch (e) {
logger.error("Error while indexing", { error: e });
}
},
},
});
return provider.namespace;
apps/webapp/app/v3/services/cancelTaskRun.server.ts
+58 -47
View File
@@ -1,4 +1,4 @@
import { TaskRun, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
import { Prisma, TaskRun, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
import { eventRepository } from "../eventRepository.server";
import { marqs } from "~/v3/marqs/index.server";
import { devPubSub } from "../marqs/devPubSub.server";
@@ -23,6 +23,13 @@ const CANCELLABLE_ATTEMPT_STATUSES: Array<TaskRunAttemptStatus> = [
"PENDING",
];
type ExtendedTaskRunAttempt = Prisma.TaskRunAttemptGetPayload<{
include: {
runtimeEnvironment: true;
backgroundWorker: true;
};
}>;
export type CancelTaskRunServiceOptions = {
reason?: string;
cancelAttempts?: boolean;
@@ -87,56 +94,60 @@ export class CancelTaskRunService extends BaseService {
// Cancel any in progress attempts
if (opts.cancelAttempts) {
for (const attempt of cancelledTaskRun.attempts) {
if (attempt.runtimeEnvironment.type === "DEVELOPMENT") {
// Signal the task run attempt to stop
await devPubSub.publish(
`backgroundWorker:${attempt.backgroundWorkerId}:${attempt.id}`,
"CANCEL_ATTEMPT",
{
attemptId: attempt.friendlyId,
backgroundWorkerId: attempt.backgroundWorker.friendlyId,
taskRunId: cancelledTaskRun.friendlyId,
}
);
} else {
switch (attempt.status) {
case "EXECUTING": {
// We need to send a cancel message to the coordinator
socketIo.coordinatorNamespace.emit("REQUEST_ATTEMPT_CANCELLATION", {
version: "v1",
attemptId: attempt.id,
attemptFriendlyId: attempt.friendlyId,
});
await this.#cancelPotentiallyRunningAttempts(cancelledTaskRun, cancelledTaskRun.attempts);
}
}
break;
}
case "PENDING":
case "PAUSED": {
logger.debug("Cancelling pending or paused attempt", {
attempt,
});
async #cancelPotentiallyRunningAttempts(run: TaskRun, attempts: ExtendedTaskRunAttempt[]) {
for (const attempt of attempts) {
if (attempt.runtimeEnvironment.type === "DEVELOPMENT") {
// Signal the task run attempt to stop
await devPubSub.publish(
`backgroundWorker:${attempt.backgroundWorkerId}:${attempt.id}`,
"CANCEL_ATTEMPT",
{
attemptId: attempt.friendlyId,
backgroundWorkerId: attempt.backgroundWorker.friendlyId,
taskRunId: run.friendlyId,
}
);
} else {
switch (attempt.status) {
case "EXECUTING": {
// We need to send a cancel message to the coordinator
socketIo.coordinatorNamespace.emit("REQUEST_ATTEMPT_CANCELLATION", {
version: "v1",
attemptId: attempt.id,
attemptFriendlyId: attempt.friendlyId,
});
const service = new CancelAttemptService();
break;
}
case "PENDING":
case "PAUSED": {
logger.debug("Cancelling pending or paused attempt", {
attempt,
});
await service.call(
attempt.friendlyId,
taskRun.id,
new Date(),
"Task run was cancelled by user"
);
const service = new CancelAttemptService();
break;
}
case "CANCELED":
case "COMPLETED":
case "FAILED": {
// Do nothing
break;
}
default: {
assertUnreachable(attempt.status);
}
await service.call(
attempt.friendlyId,
run.id,
new Date(),
"Task run was cancelled by user"
);
break;
}
case "CANCELED":
case "COMPLETED":
case "FAILED": {
// Do nothing
break;
}
default: {
assertUnreachable(attempt.status);
}
}
}
apps/webapp/app/v3/services/crashTaskRun.server.ts
+157
View File
@@ -0,0 +1,157 @@
import {
TaskRun,
TaskRunAttempt,
TaskRunAttemptStatus,
TaskRunStatus,
} from "@trigger.dev/database";
import { eventRepository } from "../eventRepository.server";
import { marqs } from "~/v3/marqs/index.server";
import { BaseService } from "./baseService.server";
import { logger } from "~/services/logger.server";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server";
export const CRASHABLE_RUN_STATUSES: Array<TaskRunStatus> = [
"PENDING",
"EXECUTING",
"PAUSED",
"WAITING_TO_RESUME",
"PAUSED",
"RETRYING_AFTER_FAILURE",
];
const CRASHABLE_ATTEMPT_STATUSES: Array<TaskRunAttemptStatus> = ["EXECUTING", "PAUSED", "PENDING"];
export type CrashTaskRunServiceOptions = {
reason?: string;
exitCode?: number;
logs?: string;
crashAttempts?: boolean;
crashedAt?: Date;
};
export class CrashTaskRunService extends BaseService {
public async call(runId: string, options?: CrashTaskRunServiceOptions) {
const opts = {
reason: "Worker crashed",
crashAttempts: true,
crashedAt: new Date(),
...options,
};
const taskRun = await this._prisma.taskRun.findFirst({
where: {
id: runId,
},
});
if (!taskRun) {
logger.error("Task run not found", { runId });
return;
}
// Make sure the task run is in a crashable state
if (!CRASHABLE_RUN_STATUSES.includes(taskRun.status)) {
return;
}
// Remove the task run from the queue if it's there for some reason
await marqs?.acknowledgeMessage(taskRun.id);
// Set the task run status to crashed
const crashedTaskRun = await this._prisma.taskRun.update({
where: {
id: taskRun.id,
},
data: {
status: "CRASHED",
},
include: {
attempts: {
where: {
status: {
in: CRASHABLE_ATTEMPT_STATUSES,
},
},
include: {
backgroundWorker: true,
runtimeEnvironment: true,
},
},
dependency: true,
runtimeEnvironment: {
include: {
organization: true,
project: true,
},
},
},
});
const inProgressEvents = await eventRepository.queryIncompleteEvents({
runId: taskRun.friendlyId,
});
logger.debug("Crashing in-progress events", {
inProgressEvents: inProgressEvents.map((event) => event.id),
});
await Promise.all(
inProgressEvents.map((event) => {
return eventRepository.crashEvent({
event: event,
crashedAt: opts.crashedAt,
exception: {
type: "Worker crashed",
message: opts.reason,
stacktrace: opts.logs,
},
});
})
);
if (!opts.crashAttempts) {
return;
}
// Cancel any in progress attempts
for (const attempt of crashedTaskRun.attempts) {
await this.#failAttempt(
attempt,
crashedTaskRun,
new Date(),
crashedTaskRun.runtimeEnvironment
);
}
}
async #failAttempt(
attempt: TaskRunAttempt,
run: TaskRun,
failedAt: Date,
environment: AuthenticatedEnvironment
) {
return await this.traceWithEnv("failAttempt()", environment, async (span) => {
span.setAttribute("taskRunId", run.id);
span.setAttribute("attemptId", attempt.id);
await marqs?.acknowledgeMessage(run.id);
await this._prisma.taskRunAttempt.update({
where: {
id: attempt.id,
},
data: {
status: "FAILED",
completedAt: failedAt,
},
});
if (environment.type === "DEVELOPMENT") {
return;
}
await ResumeTaskRunDependenciesService.enqueue(attempt.id, this._prisma);
});
}
}
apps/webapp/app/v3/services/deploymentIndexFailed.server.ts
+10 -4
View File
@@ -2,13 +2,19 @@ import { BaseService } from "./baseService.server";
export class DeploymentIndexFailed extends BaseService {
public async call(
deploymentId: string,
maybeFriendlyId: string,
error: { name: string; message: string; stack?: string }
) {
const isFriendlyId = maybeFriendlyId.startsWith("deployment_");
const deployment = await this._prisma.workerDeployment.update({
where: {
friendlyId: deploymentId,
},
where: isFriendlyId
? {
friendlyId: maybeFriendlyId,
}
: {
id: maybeFriendlyId,
},
data: {
status: "FAILED",
failedAt: new Date(),
apps/webapp/app/v3/services/indexDeployment.server.ts
+1
View File
@@ -56,6 +56,7 @@ export class IndexDeploymentService extends BaseService {
envType: deployment.environment.type,
projectId: deployment.projectId,
orgId: deployment.environment.organizationId,
deploymentId: deployment.id,
});
logger.debug("Index ACK received", { responses });
packages/cli-v3/src/workers/prod/backgroundWorker.ts
+1 -1
View File
@@ -431,7 +431,7 @@ class TaskRunProcess {
resolver(result);
},
READY_TO_DISPOSE: async (message) => {
// noop
process.exit(0);
},
TASK_HEARTBEAT: async (message) => {
this.onTaskHeartbeat.post(message.id);
packages/cli-v3/src/workers/prod/entry-point.ts
+21 -19
View File
@@ -150,36 +150,38 @@ class ProdWorker {
this.#httpServer = this.#createHttpServer();
}
async #reconnect(isPostStart = false) {
async #reconnect(isPostStart = false, reconnectImmediately = false) {
if (isPostStart) {
this.waitForPostStart = false;
}
this.#coordinatorSocket.close();
if (!this.runningInKubernetes) {
this.#coordinatorSocket.connect();
return;
if (!reconnectImmediately) {
await setTimeout(1000);
}
let coordinatorHost = COORDINATOR_HOST;
try {
const coordinatorHost = (await readFile("/etc/taskinfo/coordinator-host", "utf-8")).replace(
"\n",
""
);
if (this.runningInKubernetes) {
coordinatorHost = (await readFile("/etc/taskinfo/coordinator-host", "utf-8")).replace(
"\n",
""
);
logger.log("reconnecting", {
coordinatorHost: {
fromEnv: COORDINATOR_HOST,
fromVolume: coordinatorHost,
current: this.#coordinatorSocket.socket.io.opts.hostname,
},
});
this.#coordinatorSocket = this.#createCoordinatorSocket(coordinatorHost);
logger.log("reconnecting", {
coordinatorHost: {
fromEnv: COORDINATOR_HOST,
fromVolume: coordinatorHost,
current: this.#coordinatorSocket.socket.io.opts.hostname,
},
});
}
} catch (error) {
logger.error("taskinfo read error during reconnect", { error });
this.#coordinatorSocket.connect();
} finally {
this.#coordinatorSocket = this.#createCoordinatorSocket(coordinatorHost);
}
}
@@ -618,7 +620,7 @@ class ProdWorker {
break;
}
case "restore": {
await this.#reconnect(true);
await this.#reconnect(true, true);
break;
}
default: {
packages/core-apps/src/provider.ts
+5 -3
View File
@@ -33,6 +33,7 @@ export interface TaskOperationsIndexOptions {
envType: EnvironmentType;
orgId: string;
projectId: string;
deploymentId: string;
}
export interface TaskOperationsCreateOptions {
@@ -86,7 +87,7 @@ export class ProviderShell implements Provider {
#httpPort: number;
#httpServer: ReturnType<typeof createServer>;
#platformSocket: ZodSocketConnection<
platformSocket: ZodSocketConnection<
typeof ProviderToPlatformMessages,
typeof PlatformToProviderMessages
>;
@@ -95,7 +96,7 @@ export class ProviderShell implements Provider {
this.tasks = options.tasks;
this.#httpPort = options.port ?? HTTP_SERVER_PORT;
this.#httpServer = this.#createHttpServer();
this.#platformSocket = this.#createPlatformSocket();
this.platformSocket = this.#createPlatformSocket();
this.#createSharedQueueSocket();
}
@@ -196,6 +197,7 @@ export class ProviderShell implements Provider {
envType: message.envType,
orgId: message.orgId,
projectId: message.projectId,
deploymentId: message.deploymentId,
});
} catch (error) {
logger.error("index failed", error);
@@ -272,7 +274,7 @@ export class ProviderShell implements Provider {
return reply.text(`${MACHINE_NAME}`);
}
case "/close": {
this.#platformSocket.close();
this.platformSocket.close();
return reply.text("platform socket closed");
}
case "/delete": {
packages/core/src/v3/schemas/schemas.ts
+22
View File
@@ -71,6 +71,27 @@ export const ProviderToPlatformMessages = {
status: z.literal("ok"),
}),
},
WORKER_CRASHED: {
message: z.object({
version: z.literal("v1").default("v1"),
runId: z.string(),
reason: z.string().optional(),
exitCode: z.number().optional(),
message: z.string().optional(),
logs: z.string().optional(),
}),
},
INDEXING_FAILED: {
message: z.object({
version: z.literal("v1").default("v1"),
deploymentId: z.string(),
error: z.object({
name: z.string(),
message: z.string(),
stack: z.string().optional(),
}),
}),
},
};
export const PlatformToProviderMessages = {
@@ -94,6 +115,7 @@ export const PlatformToProviderMessages = {
envType: EnvironmentType,
orgId: z.string(),
projectId: z.string(),
deploymentId: z.string(),
}),
callback: z.discriminatedUnion("success", [
z.object({
packages/database/prisma/migrations/20240404150051_add_crashed_task_run_status/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterEnum
ALTER TYPE "TaskRunStatus" ADD VALUE 'CRASHED';
packages/database/prisma/schema.prisma
+3
View File
@@ -1649,6 +1649,9 @@ enum TaskRunStatus {
/// Task has failed to complete, due to an error in the system
SYSTEM_FAILURE
/// Task has crashed and won't be retried, most likely the worker ran out of resources, e.g. memory or storage
CRASHED
}
model TaskRunDependency {
pnpm-lock.yaml
Generated
+20
View File
@@ -148,6 +148,9 @@ importers:
'@trigger.dev/core-apps':
specifier: workspace:*
version: link:../../packages/core-apps
p-queue:
specifier: ^8.0.1
version: 8.0.1
socket.io-client:
specifier: ^4.7.4
version: 4.7.4
@@ -20143,6 +20146,10 @@ packages:
resolution: {integrity: sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==}
dev: false
/eventemitter3@5.0.1:
resolution: {integrity: sha512-GWkBvjiSZK87ELrYOSESUYeVIc9mvLLf/nXalMOS5dYrgZq9o5OVkbZAVM06CVxYsCwH9BDZFPlQTlPA1j4ahA==}
dev: false
/events@3.3.0:
resolution: {integrity: sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==}
engines: {node: '>=0.8.x'}
@@ -26402,6 +26409,14 @@ packages:
p-timeout: 3.2.0
dev: false
/p-queue@8.0.1:
resolution: {integrity: sha512-NXzu9aQJTAzbBqOt2hwsR63ea7yvxJc0PwN/zobNAudYfb1B7R08SzB4TsLeSbUCuG467NhnoT0oO6w1qRO+BA==}
engines: {node: '>=18'}
dependencies:
eventemitter3: 5.0.1
p-timeout: 6.1.2
dev: false
/p-retry@4.6.2:
resolution: {integrity: sha512-312Id396EbJdvRONlngUx0NydfrIQ5lsYu0znKVUzVvArzEIt08V1qhtyESbGVd1FGX7UKtiFp5uwKZdM8wIuQ==}
engines: {node: '>=8'}
@@ -26435,6 +26450,11 @@ packages:
engines: {node: '>=12'}
dev: true
/p-timeout@6.1.2:
resolution: {integrity: sha512-UbD77BuZ9Bc9aABo74gfXhNvzC9Tx7SxtHSh1fxvx3jTLLYvmVhiQZZrJzqqU0jKbN32kb5VOKiLEQI/3bIjgQ==}
engines: {node: '>=14.16'}
dev: false
/p-try@2.2.0:
resolution: {integrity: sha512-R4nPAVTAU0B9D35/Gk3uJf/7XYbQcyohSKdvAxIRSNghFl4e71hVoGnBNQz9cWaXxO2I10KTC+3jMdvvoKw6dQ==}
engines: {node: '>=6'}