Run One-Off ECS Tasks With AWS SDK v3 (RunTask)

// run-task.ts: run a one-off ECS task on Fargate, wait for it to stop, return its exit code and logs.
import { randomUUID } from "node:crypto";
import {
  ECSClient,
  RunTaskCommand,
  DescribeTasksCommand,
  waitUntilTasksStopped,
  type Failure,
  type RunTaskCommandInput,
} from "@aws-sdk/client-ecs";
import { CloudWatchLogsClient, paginateFilterLogEvents } from "@aws-sdk/client-cloudwatch-logs";

export interface OneOffTask {
  cluster: string;
  taskDefinition: string; // "family:revision", or "family" for the latest ACTIVE revision
  container: string; // the container whose command and exit code you care about
  command?: string[];
  environment?: Record<string, string>;
  subnets: string[];
  securityGroups: string[];
  assignPublicIp?: boolean; // only for public subnets without a NAT gateway
  spot?: boolean; // FARGATE_SPOT through a capacity provider strategy
  startedBy?: string;
  maxWaitSeconds?: number;
  logGroup?: string; // awslogs-group of the container
  logStreamPrefix?: string; // awslogs-stream-prefix of the container
}

export interface TaskResult {
  taskArn: string;
  exitCode?: number;
  stopCode?: string;
  stoppedReason?: string;
  containerReason?: string;
  logs: string[];
}

export class RunTaskFailed extends Error {
  readonly failures: Failure[];
  constructor(failures: Failure[]) {
    super(`RunTask placed no task: ${failures.map((f) => `${f.reason ?? "unknown"} ${f.detail ?? ""} ${f.arn ?? ""}`.trim()).join("; ") || "no tasks returned"}`);
    this.name = "RunTaskFailed";
    this.failures = failures;
  }
}

const ecs = new ECSClient({});
const logs = new CloudWatchLogsClient({});

export async function runOneOffTask(t: OneOffTask, waiterMinDelay = 6): Promise<TaskResult> {
  const input: RunTaskCommandInput = {
    cluster: t.cluster,
    taskDefinition: t.taskDefinition,
    count: 1,
    clientToken: randomUUID(), // makes a retried request return the same task instead of starting a second one
    startedBy: t.startedBy ?? "one-off",
    networkConfiguration: {
      awsvpcConfiguration: { subnets: t.subnets, securityGroups: t.securityGroups, assignPublicIp: t.assignPublicIp ? "ENABLED" : "DISABLED" },
    },
    overrides: {
      containerOverrides: [
        {
          name: t.container,
          command: t.command,
          environment: Object.entries(t.environment ?? {}).map(([name, value]) => ({ name, value })),
        },
      ],
    },
  };
  // A task uses either a launch type or a capacity provider strategy, never both.
  if (t.spot) input.capacityProviderStrategy = [{ capacityProvider: "FARGATE_SPOT", weight: 1 }];
  else input.launchType = "FARGATE";

  const run = await ecs.send(new RunTaskCommand(input));
  const taskArn = run.tasks?.[0]?.taskArn;
  if (!taskArn || run.failures?.length) throw new RunTaskFailed(run.failures ?? []);

  await waitUntilTasksStopped(
    { client: ecs, maxWaitTime: t.maxWaitSeconds ?? 3600, minDelay: waiterMinDelay },
    { cluster: t.cluster, tasks: [taskArn] },
  );

  const described = await ecs.send(new DescribeTasksCommand({ cluster: t.cluster, tasks: [taskArn] }));
  const task = described.tasks?.[0];
  const container = task?.containers?.find((c) => c.name === t.container);
  const result: TaskResult = {
    taskArn,
    exitCode: container?.exitCode,
    stopCode: task?.stopCode,
    stoppedReason: task?.stoppedReason,
    containerReason: container?.reason,
    logs: [],
  };

  if (t.logGroup && t.logStreamPrefix) {
    // awslogs names the stream prefix/container-name/task-id
    const taskId = taskArn.split("/").pop() ?? "";
    const stream = `${t.logStreamPrefix}/${t.container}/${taskId}`;
    for await (const page of paginateFilterLogEvents({ client: logs }, { logGroupName: t.logGroup, logStreamNames: [stream] })) {
      for (const e of page.events ?? []) result.logs.push(e.message ?? "");
    }
  }
  return result;
}

// run-migration.ts: run database migrations as a one-off ECS task from CI and fail the job if they fail.
// Usage: CLUSTER=prod SUBNETS=subnet-0a,subnet-0b SECURITY_GROUP=sg-0abc npx tsx run-migration.ts
import { runOneOffTask, RunTaskFailed } from "./run-task.ts";

const env = (name: string): string => {
  const v = process.env[name];
  if (!v) throw new Error(`${name} is not set`);
  return v;
};

try {
  const result = await runOneOffTask({
    cluster: env("CLUSTER"),
    taskDefinition: process.env.TASK_DEFINITION ?? "api",
    container: "app",
    command: ["npm", "run", "migrate"],
    environment: { LOG_LEVEL: "info" },
    subnets: env("SUBNETS").split(","),
    securityGroups: [env("SECURITY_GROUP")],
    startedBy: `ci-${process.env.GITHUB_RUN_ID ?? "local"}`,
    maxWaitSeconds: 1800,
    logGroup: "/ecs/api",
    logStreamPrefix: "api",
  });
  for (const line of result.logs) console.log(line);
  console.log(`task ${result.taskArn} stopped (${result.stopCode}), exit code ${result.exitCode}`);
  if (result.exitCode !== 0) console.error("stopped reason:", result.stoppedReason, "| container reason:", result.containerReason);
  process.exit(result.exitCode === 0 ? 0 : 1);
} catch (err) {
  if (err instanceof RunTaskFailed) console.error(err.message, err.failures);
  else console.error(err instanceof Error ? `${err.name}: ${err.message}` : err);
  process.exit(1);
}

> [email protected] migrate
Applied 20260928_add_invoice_index
Applied 20260929_backfill_currency
2 migrations applied
task arn:aws:ecs:us-east-1:123456789012:task/prod/9f3c1e7a52b84d0f stopped (EssentialContainerExited), exit code 0

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "RunApiTaskInProdCluster",
      "Effect": "Allow",
      "Action": "ecs:RunTask",
      "Resource": "arn:aws:ecs:us-east-1:123456789012:task-definition/api:*",
      "Condition": { "ArnEquals": { "ecs:cluster": "arn:aws:ecs:us-east-1:123456789012:cluster/prod" } }
    },
    {
      "Sid": "WatchTasks",
      "Effect": "Allow",
      "Action": "ecs:DescribeTasks",
      "Resource": "arn:aws:ecs:us-east-1:123456789012:task/prod/*"
    },
    {
      "Sid": "PassTaskRoles",
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": [
        "arn:aws:iam::123456789012:role/api-task-execution",
        "arn:aws:iam::123456789012:role/api-task"
      ],
      "Condition": { "StringEquals": { "iam:PassedToService": "ecs-tasks.amazonaws.com" } }
    },
    {
      "Sid": "ReadTaskLogs",
      "Effect": "Allow",
      "Action": "logs:FilterLogEvents",
      "Resource": "arn:aws:logs:us-east-1:123456789012:log-group:/ecs/api"
    }
  ]
}

// run-task.test.ts: exercises run-task.ts with aws-sdk-client-mock (run with: npx tsx test/run-task.test.ts)
import assert from "node:assert/strict";
import { mockClient } from "aws-sdk-client-mock";
import { ECSClient, RunTaskCommand, DescribeTasksCommand } from "@aws-sdk/client-ecs";
import { CloudWatchLogsClient, FilterLogEventsCommand } from "@aws-sdk/client-cloudwatch-logs";
import { runOneOffTask, RunTaskFailed, type OneOffTask } from "../src/run-task.ts";

const ecs = mockClient(ECSClient);
const logs = mockClient(CloudWatchLogsClient);
const arn = "arn:aws:ecs:us-east-1:123456789012:task/prod/0a1b2c3d4e5f";
const job: OneOffTask = {
  cluster: "prod", taskDefinition: "api:42", container: "app", command: ["npm", "run", "migrate"],
  subnets: ["subnet-0a"], securityGroups: ["sg-0abc"], logGroup: "/ecs/api", logStreamPrefix: "api",
};

ecs.on(RunTaskCommand).resolves({ tasks: [{ taskArn: arn, lastStatus: "PROVISIONING" }], failures: [] });
ecs.on(DescribeTasksCommand)
  .resolvesOnce({ tasks: [{ taskArn: arn, lastStatus: "RUNNING" }] })
  .resolves({
    tasks: [{ taskArn: arn, lastStatus: "STOPPED", stopCode: "EssentialContainerExited",
      containers: [{ name: "app", exitCode: 0 }] }],
  });
logs.on(FilterLogEventsCommand)
  .resolvesOnce({ events: [{ message: "Running 3 migrations" }], nextToken: "t1" })
  .resolves({ events: [{ message: "Done" }] });

const ok = await runOneOffTask(job, 1);
assert.equal(ok.exitCode, 0);
assert.deepEqual(ok.logs, ["Running 3 migrations", "Done"]);
const sent = ecs.commandCalls(RunTaskCommand)[0].args[0].input;
assert.equal(sent.launchType, "FARGATE");
assert.equal(sent.capacityProviderStrategy, undefined);
assert.equal(sent.networkConfiguration?.awsvpcConfiguration?.assignPublicIp, "DISABLED");
assert.deepEqual(sent.overrides?.containerOverrides?.[0].command, ["npm", "run", "migrate"]);
assert.equal(typeof sent.clientToken, "string");
assert.deepEqual(logs.commandCalls(FilterLogEventsCommand)[0].args[0].input.logStreamNames, ["api/app/0a1b2c3d4e5f"]);

// Spot uses a capacity provider strategy instead of a launch type.
ecs.on(DescribeTasksCommand).resolves({ tasks: [{ taskArn: arn, lastStatus: "STOPPED", containers: [{ name: "app", exitCode: 3 }] }] });
const failed = await runOneOffTask({ ...job, spot: true, logGroup: undefined }, 1);
assert.equal(failed.exitCode, 3);
const spot = ecs.commandCalls(RunTaskCommand)[1].args[0].input;
assert.equal(spot.launchType, undefined);
assert.deepEqual(spot.capacityProviderStrategy, [{ capacityProvider: "FARGATE_SPOT", weight: 1 }]);

// Placement failures come back in the response, not as an exception.
ecs.on(RunTaskCommand).resolves({ tasks: [], failures: [{ reason: "ATTRIBUTE", arn: "arn:aws:ecs:us-east-1:123456789012:cluster/prod" }] });
await assert.rejects(runOneOffTask(job, 1), (err: unknown) => err instanceof RunTaskFailed && err.failures[0].reason === "ATTRIBUTE");
console.log("run-task tests passed");