Skip to content

Commit d69eac2

Browse files
0skiTrigger.dev RepoOps
authored andcommitted
feat(webapp,run-engine): mint metering dimensions into the usage JWT
The signed usage token now includes the environment type and region, and invocation usage reports send a signed token too. Mono-RevId: 54895cd41d9ee94b993fa4ac26d4e2f9aab51f78
1 parent 2b1dd89 commit d69eac2

12 files changed

Lines changed: 247 additions & 45 deletions

‎apps/webapp/app/routes/engine.v1.dev.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import { defaultMachine } from "~/services/platform.v3.server";
1414
import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
1515
import { resolveVariablesForEnvironment } from "~/v3/environmentVariables/environmentVariablesRepository.server";
1616
import { machinePresetFromName } from "~/v3/machinePresets.server";
17+
import { meteringClaims } from "~/v3/utils/meteringClaims.server";
1718
import { engine } from "~/v3/runEngine.server";
1819
import { runStore } from "~/v3/runStore.server";
1920

@@ -59,6 +60,11 @@ const { action } = createActionApiRoute(
5960
authentication.environment,
6061
engineResult.run.id,
6162
engineResult.execution.machine ?? defaultMachinePreset,
63+
meteringClaims({
64+
environmentType: authentication.environment.type,
65+
region: run.region,
66+
workerQueue: run.workerQueue,
67+
}),
6268
engineResult.run.taskEventStore
6369
);
6470

@@ -80,11 +86,13 @@ async function getEnvVars(
8086
environment: AuthenticatedEnvironment,
8187
runId: string,
8288
machinePreset: MachinePreset,
89+
metering: Record<string, string>,
8390
taskEventStore?: string
8491
): Promise<Record<string, string>> {
8592
const variables = await resolveVariablesForEnvironment(environment);
8693

8794
const jwt = await generateJWTTokenForEnvironment(environment, {
95+
...metering,
8896
run_id: runId,
8997
machine_preset: machinePreset.name,
9098
});

‎apps/webapp/app/services/apiAuth.server.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -859,7 +859,7 @@ const JWT_ALGORITHM = "HS256";
859859
const DEFAULT_JWT_EXPIRATION_IN_MS = 1000 * 60 * 60; // 1 hour
860860

861861
export async function generateJWTTokenForEnvironment(
862-
environment: RuntimeEnvironmentForEnvRepo,
862+
environment: Pick<RuntimeEnvironmentForEnvRepo, "id" | "organizationId" | "projectId">,
863863
payload: Record<string, string>
864864
) {
865865
const jwt = await new SignJWT({

‎apps/webapp/app/services/platform.v3.server.ts‎

Lines changed: 21 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import {
1616
type UsageResult,
1717
type UsageSeriesParams,
1818
type CurrentPlan,
19+
ReportInvocationUsageResult,
1920
} from "@trigger.dev/platform";
2021
import {
2122
BillingLimitResultSchema,
@@ -771,25 +772,33 @@ export async function getUsageSeries(organizationId: string, params: UsageSeries
771772
export async function reportInvocationUsage(
772773
organizationId: string,
773774
costInCents: number,
775+
jwt: string,
774776
additionalData?: Record<string, any>
775777
) {
776778
if (!client) return undefined;
777779

778-
try {
779-
const result = await client.reportInvocationUsage({
780-
organizationId,
781-
costInCents,
782-
additionalData,
783-
});
784-
if (!result.success) {
785-
recordPlatformFailure("reportInvocationUsage", "no_success");
786-
return undefined;
787-
}
788-
return result;
789-
} catch (_e) {
780+
const [error, result] = await tryCatch(
781+
client.fetch("/api/v1/usage/ingest/invocation", ReportInvocationUsageResult, {
782+
method: "POST",
783+
headers: {
784+
"Content-Type": "application/json",
785+
"x-trigger-jwt": jwt,
786+
},
787+
body: JSON.stringify({ organizationId, costInCents, additionalData }),
788+
})
789+
);
790+
791+
if (error) {
790792
recordPlatformFailure("reportInvocationUsage", "caught");
791793
return undefined;
792794
}
795+
796+
if (!result.success) {
797+
recordPlatformFailure("reportInvocationUsage", "no_success");
798+
return undefined;
799+
}
800+
801+
return result;
793802
}
794803

795804
export async function reportComputeUsage(request: Request) {

‎apps/webapp/app/v3/runEngineHandlers.server.ts‎

Lines changed: 35 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import type { CompleteBatchResult } from "@internal/run-engine";
1+
import type { CompleteBatchResult, EventBusEventArgs } from "@internal/run-engine";
22
import { SpanKind } from "@internal/tracing";
33
import { tryCatch } from "@trigger.dev/core/utils";
44
import { createJsonErrorObject, sanitizeError, TaskRunErrorCodes } from "@trigger.dev/core/v3";
@@ -16,13 +16,15 @@ import { env } from "~/env.server";
1616
import { findEnvironmentById, findEnvironmentFromRun } from "~/models/runtimeEnvironment.server";
1717
import { TriggerFailedTaskService } from "~/runEngine/services/triggerFailedTask.server";
1818
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
19+
import { generateJWTTokenForEnvironment } from "~/services/apiAuth.server";
1920
import { logger } from "~/services/logger.server";
2021
import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
2122
import { reportInvocationUsage } from "~/services/platform.v3.server";
2223
import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server";
2324
import { MetadataTooLargeError } from "~/utils/packets";
2425
import { QueueSizeLimitExceededError } from "~/v3/services/common.server";
2526
import { TriggerTaskService } from "~/v3/services/triggerTask.server";
27+
import { meteringClaims } from "~/v3/utils/meteringClaims.server";
2628
import { tracer } from "~/v3/tracer.server";
2729
import { createExceptionPropertiesFromError } from "./eventRepository/common.server";
2830
import { getEventRepositoryForStore, recordRunDebugLog } from "./eventRepository/index.server";
@@ -581,12 +583,16 @@ export function registerRunEngineEventBusHandlers() {
581583
}
582584
);
583585

584-
engine.eventBus.on("runAttemptStarted", async ({ time, run, organization }) => {
585-
try {
586-
if (run.attemptNumber === 1 && run.baseCostInCents > 0) {
587-
await reportInvocationUsage(organization.id, run.baseCostInCents, { runId: run.id });
588-
}
589-
} catch (error) {
586+
engine.eventBus.on("runAttemptStarted", async (event) => {
587+
const { run, organization } = event;
588+
589+
if (run.attemptNumber !== 1 || run.baseCostInCents <= 0) {
590+
return;
591+
}
592+
593+
const [error] = await tryCatch(reportAttemptInvocationUsage(event));
594+
595+
if (error) {
590596
logger.error("[runAttemptStarted] Failed to report invocation usage", {
591597
error: error instanceof Error ? error.message : error,
592598
runId: run.id,
@@ -1100,3 +1106,25 @@ function normalizePayload(payload: unknown, payloadType?: string): unknown {
11001106

11011107
return payload;
11021108
}
1109+
1110+
async function reportAttemptInvocationUsage({
1111+
run,
1112+
organization,
1113+
project,
1114+
environment,
1115+
}: EventBusEventArgs<"runAttemptStarted">[0]) {
1116+
const jwt = await generateJWTTokenForEnvironment(
1117+
{ id: environment.id, organizationId: organization.id, projectId: project.id },
1118+
{
1119+
...meteringClaims({
1120+
environmentType: environment.type,
1121+
region: run.region,
1122+
workerQueue: run.workerQueue,
1123+
}),
1124+
run_id: run.id,
1125+
machine_preset: run.machinePreset,
1126+
}
1127+
);
1128+
1129+
return reportInvocationUsage(organization.id, run.baseCostInCents, jwt, { runId: run.id });
1130+
}

‎apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ import { defaultMachine } from "~/services/platform.v3.server";
3737
import { singleton } from "~/utils/singleton";
3838
import { resolveVariablesForEnvironment } from "~/v3/environmentVariables/environmentVariablesRepository.server";
3939
import { machinePresetFromName } from "~/v3/machinePresets.server";
40+
import { meteringClaims } from "~/v3/utils/meteringClaims.server";
4041
import type { WithRunEngineOptions } from "../baseService.server";
4142
import { WithRunEngine } from "../baseService.server";
4243

@@ -563,6 +564,11 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
563564
environment,
564565
engineResult.run.id,
565566
engineResult.execution.machine ?? defaultMachinePreset,
567+
meteringClaims({
568+
environmentType: environment.type,
569+
region: engineResult.run.region,
570+
workerQueue: engineResult.run.workerQueue,
571+
}),
566572
environment.parentEnvironment ?? undefined,
567573
engineResult.run.taskEventStore ?? undefined
568574
)
@@ -712,12 +718,14 @@ export class AuthenticatedWorkerInstance extends WithRunEngine {
712718
environment: RuntimeEnvironment,
713719
runId: string,
714720
machinePreset: MachinePreset,
721+
metering: Record<string, string>,
715722
parentEnvironment?: RuntimeEnvironment,
716723
taskEventStore?: string
717724
): Promise<Record<string, string>> {
718725
const variables = await resolveVariablesForEnvironment(environment, parentEnvironment);
719726

720727
const jwt = await generateJWTTokenForEnvironment(environment, {
728+
...metering,
721729
run_id: runId,
722730
machine_preset: machinePreset.name,
723731
});
Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
import { describe, expect, it } from "vitest";
2+
import { meteringClaims, regionForMetering } from "./meteringClaims.server";
3+
4+
describe("regionForMetering", () => {
5+
it("returns the explicit region when set", () => {
6+
expect(regionForMetering("eu-central-1", "eu-central-1-microvm")).toBe("eu-central-1");
7+
});
8+
9+
it("returns the worker queue when it has no colon", () => {
10+
expect(regionForMetering(null, "us-nyc-3")).toBe("us-nyc-3");
11+
});
12+
13+
it("returns the worker queue prefix before the first colon", () => {
14+
expect(regionForMetering(undefined, "us-nyc-3:scheduled")).toBe("us-nyc-3");
15+
expect(regionForMetering(undefined, "us-nyc-3:a:b")).toBe("us-nyc-3");
16+
});
17+
18+
it("falls back to the worker queue when the region is empty", () => {
19+
expect(regionForMetering("", "us-nyc-3")).toBe("us-nyc-3");
20+
});
21+
22+
it("returns undefined when neither is known", () => {
23+
expect(regionForMetering(null, null)).toBeUndefined();
24+
expect(regionForMetering(undefined, "")).toBeUndefined();
25+
});
26+
});
27+
28+
describe("meteringClaims", () => {
29+
it("lowercases the environment type", () => {
30+
expect(meteringClaims({ environmentType: "PRODUCTION", region: "us-east-1" })).toEqual({
31+
environment_type: "production",
32+
region: "us-east-1",
33+
});
34+
expect(meteringClaims({ environmentType: "STAGING" }).environment_type).toBe("staging");
35+
expect(meteringClaims({ environmentType: "PREVIEW" }).environment_type).toBe("preview");
36+
expect(meteringClaims({ environmentType: "DEVELOPMENT" }).environment_type).toBe("development");
37+
});
38+
39+
it("prefers the run's region over its worker queue", () => {
40+
expect(
41+
meteringClaims({
42+
environmentType: "PRODUCTION",
43+
region: "eu-central-1",
44+
workerQueue: "eu-central-1-microvm",
45+
})
46+
).toEqual({ environment_type: "production", region: "eu-central-1" });
47+
});
48+
49+
it("falls back to the worker queue prefix before the first colon", () => {
50+
expect(
51+
meteringClaims({
52+
environmentType: "PRODUCTION",
53+
region: null,
54+
workerQueue: "us-nyc-3:scheduled",
55+
})
56+
).toEqual({ environment_type: "production", region: "us-nyc-3" });
57+
});
58+
59+
it("treats an empty region as unknown", () => {
60+
expect(
61+
meteringClaims({ environmentType: "STAGING", region: "", workerQueue: "us-nyc-3" })
62+
).toEqual({ environment_type: "staging", region: "us-nyc-3" });
63+
});
64+
65+
it("omits region when neither region nor worker queue is known", () => {
66+
const claims = meteringClaims({ environmentType: "PRODUCTION", region: null, workerQueue: "" });
67+
68+
expect(claims).toEqual({ environment_type: "production" });
69+
expect("region" in claims).toBe(false);
70+
});
71+
72+
it("always reports development runs as local", () => {
73+
expect(
74+
meteringClaims({
75+
environmentType: "DEVELOPMENT",
76+
region: "cm_dev_environment_id",
77+
workerQueue: "cm_dev_environment_id",
78+
})
79+
).toEqual({ environment_type: "development", region: "local" });
80+
expect(meteringClaims({ environmentType: "DEVELOPMENT" })).toEqual({
81+
environment_type: "development",
82+
region: "local",
83+
});
84+
});
85+
});
Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
import type { RuntimeEnvironmentType } from "@trigger.dev/database";
2+
3+
const DEVELOPMENT_METERING_REGION = "local";
4+
5+
type MeteringClaimsInput = {
6+
environmentType: RuntimeEnvironmentType;
7+
region?: string | null;
8+
workerQueue?: string | null;
9+
};
10+
11+
export function regionForMetering(
12+
region: string | null | undefined,
13+
workerQueue: string | null | undefined
14+
): string | undefined {
15+
if (region) {
16+
return region;
17+
}
18+
19+
if (!workerQueue) {
20+
return undefined;
21+
}
22+
23+
const colon = workerQueue.indexOf(":");
24+
return colon === -1 ? workerQueue : workerQueue.slice(0, colon);
25+
}
26+
27+
export function meteringClaims({
28+
environmentType,
29+
region,
30+
workerQueue,
31+
}: MeteringClaimsInput): Record<string, string> {
32+
const claims = { environment_type: environmentType.toLowerCase() };
33+
34+
if (environmentType === "DEVELOPMENT") {
35+
return { ...claims, region: DEVELOPMENT_METERING_REGION };
36+
}
37+
38+
const resolvedRegion = regionForMetering(region, workerQueue);
39+
40+
if (!resolvedRegion) {
41+
return claims;
42+
}
43+
44+
return { ...claims, region: resolvedRegion };
45+
}

‎internal-packages/run-engine/src/engine/eventBus.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -136,6 +136,9 @@ export type EventBusEvents = {
136136
executedAt: Date | undefined;
137137
runTags: string[];
138138
batchId: string | null;
139+
region: string | null;
140+
workerQueue: string;
141+
machinePreset: string;
139142
};
140143
organization: {
141144
id: string;
@@ -145,6 +148,7 @@ export type EventBusEvents = {
145148
};
146149
environment: {
147150
id: string;
151+
type: RuntimeEnvironmentType;
148152
};
149153
},
150154
];

‎internal-packages/run-engine/src/engine/index.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@ import {
1515
type DequeuedMessage,
1616
type ExecutionResult,
1717
type RunExecutionData,
18-
type StartRunAttemptResult,
1918
type TaskRunContext,
2019
type TaskRunExecutionResult,
2120
type TaskRunInternalError,
@@ -100,6 +99,7 @@ import {
10099
import { TtlSystem } from "./systems/ttlSystem.js";
101100
import { WaitpointSystem } from "./systems/waitpointSystem.js";
102101
import type {
102+
EngineStartRunAttemptResult,
103103
EngineWorker,
104104
HeartbeatTimeouts,
105105
ReportableQueue,
@@ -1660,7 +1660,7 @@ export class RunEngine {
16601660
// Carried from the DequeuedMessage via the worker's start-attempt request; see RunAttemptSystem.
16611661
snapshotRoute?: SnapshotRouteWire;
16621662
tx?: PrismaClientOrTransaction;
1663-
}): Promise<StartRunAttemptResult> {
1663+
}): Promise<EngineStartRunAttemptResult> {
16641664
return this.runAttemptSystem.startRunAttempt({
16651665
runId,
16661666
snapshotId,

0 commit comments

Comments
 (0)