1import * as AWS from "@/AWS";
2import { Cluster } from "@/AWS/EMR/Cluster.ts";
3import * as Output from "@/Output";
4import * as Test from "@/Test/Alchemy";
5import * as EC2 from "@distilled.cloud/aws/ec2";
6import * as emr from "@distilled.cloud/aws/emr";
7import { expect } from "alchemy-test";
8import * as Duration from "effect/Duration";
9import * as Effect from "effect/Effect";
10import * as Schedule from "effect/Schedule";
11import { getDefaultVpc } from "../DefaultVpc.ts";
12
13const { test } = Test.make({ providers: AWS.providers() });
14
15// Ungated typed-error probes: prove the distilled emr error union carries the
16// not-found tags this provider's read/reconcile/delete paths depend on.
17// DescribeCluster surfaces a missing cluster as InvalidRequestException
18// ("Cluster id 'j-…' is not valid.") → synthetic ClusterNotFound, and
19// TerminateJobFlows as ValidationException ("Specified job flow ID not
20// valid.") → synthetic JobFlowNotFound.
21test.provider(
22 "describeCluster on a nonexistent id fails with ClusterNotFound",
23 () =>
24 Effect.gen(function* () {
25 const error = yield* Effect.flip(
26 emr.describeCluster({ ClusterId: "j-1K48XAOQ4XHCB" }),
27 );
28 expect(error._tag).toBe("ClusterNotFound");
29 }),
30 { tags: ["provider:aws", "provider:aws:emr", "live"] },
31);
32
33test.provider(
34 "terminateJobFlows on a nonexistent id fails with JobFlowNotFound",
35 () =>
36 Effect.gen(function* () {
37 const error = yield* Effect.flip(
38 emr.terminateJobFlows({ JobFlowIds: ["j-1K48XAOQ4XHCB"] }),
39 );
40 expect(error._tag).toBe("JobFlowNotFound");
41 }),
42 { tags: ["provider:aws", "provider:aws:emr", "live"] },
43);
44
45// Resolve a subnet of the account's default VPC (public subnet — EMR
46// clusters launch fine there and the factory never creates NAT gateways).
47const resolveSubnet = Effect.gen(function* () {
48 const vpc = yield* getDefaultVpc;
49 const subnets = yield* EC2.describeSubnets({
50 Filters: [
51 { Name: "vpc-id", Values: [vpc.vpcId] },
52 { Name: "default-for-az", Values: ["true"] },
53 ],
54 });
55 const subnetId = (subnets.Subnets ?? []).flatMap((s) =>
56 s.SubnetId ? [s.SubnetId] : [],
57 )[0];
58 return { subnetId };
59});
60
61// A provisioned EMR cluster takes ~10-15 minutes to reach WAITING, bills per
62// instance-hour for 3 m5.xlarge-class instances while it exists, and takes
63// ~5-10 more minutes to terminate. The full lifecycle is gated behind
64// AWS_TEST_SLOW=1 and always destroys what it created.
65test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
66 "create Spark cluster, verify WAITING, update in place, destroy, verify terminating",
67 (stack) =>
68 Effect.gen(function* () {
69 yield* stack.destroy();
70
71 const { subnetId } = yield* resolveSubnet;
72 expect(subnetId).toBeDefined();
73
74 const deploy = (props: {
75 stepConcurrencyLevel?: number;
76 idleTimeoutSeconds?: number;
77 }) =>
78 stack.deploy(
79 Effect.gen(function* () {
80 const logs = yield* AWS.S3.Bucket("EmrLogs", {
81 forceDestroy: true,
82 });
83 // EMR service role (legacy managed policy — the v2 policy
84 // requires tag-scoped resources).
85 const serviceRole = yield* AWS.IAM.Role("EmrServiceRole", {
86 assumeRolePolicyDocument: {
87 Version: "2012-10-17",
88 Statement: [
89 {
90 Effect: "Allow",
91 Principal: { Service: ["elasticmapreduce.amazonaws.com"] },
92 Action: ["sts:AssumeRole"],
93 },
94 ],
95 },
96 managedPolicyArns: [
97 "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole",
98 ],
99 });
100 // EC2 instance role + profile (the job-flow role).
101 const ec2Role = yield* AWS.IAM.Role("EmrEc2Role", {
102 assumeRolePolicyDocument: {
103 Version: "2012-10-17",
104 Statement: [
105 {
106 Effect: "Allow",
107 Principal: { Service: ["ec2.amazonaws.com"] },
108 Action: ["sts:AssumeRole"],
109 },
110 ],
111 },
112 managedPolicyArns: [
113 "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role",
114 ],
115 });
116 const instanceProfile = yield* AWS.IAM.InstanceProfile(
117 "EmrEc2Profile",
118 { roleName: ec2Role.roleName },
119 );
120 const cluster = yield* Cluster("Spark", {
121 releaseLabel: "emr-7.5.0",
122 applications: ["Spark", "Hadoop"],
123 serviceRole: serviceRole.roleName,
124 jobFlowRole: instanceProfile.instanceProfileName,
125 logUri: Output.interpolate`s3://${logs.bucketName}/logs/`,
126 instances: {
127 masterInstanceType: "m5.xlarge",
128 coreInstanceType: "m5.xlarge",
129 coreInstanceCount: 1,
130 ec2SubnetId: subnetId,
131 keepJobFlowAliveWhenNoSteps: true,
132 },
133 stepConcurrencyLevel: props.stepConcurrencyLevel,
134 autoTerminationPolicy: props.idleTimeoutSeconds
135 ? {
136 idleTimeout: Duration.seconds(props.idleTimeoutSeconds),
137 }
138 : undefined,
139 tags: { fixture: "emr-cluster" },
140 });
141 return { cluster };
142 }),
143 );
144
145 // Create — reconcile waits (bounded) for WAITING/RUNNING.
146 const { cluster } = yield* deploy({ idleTimeoutSeconds: 3600 });
147 expect(cluster.clusterId).toMatch(/^j-/);
148 expect(cluster.clusterArn).toContain(":cluster/");
149 expect(["WAITING", "RUNNING"]).toContain(cluster.state);
150
151 // Out-of-band verification via distilled.
152 const created = yield* emr.describeCluster({
153 ClusterId: cluster.clusterId,
154 });
155 expect(created.Cluster?.ReleaseLabel).toBe("emr-7.5.0");
156 expect(
157 (created.Cluster?.Applications ?? []).map((a) => a.Name).sort(),
158 ).toEqual(["Hadoop", "Spark"]);
159 const tags = Object.fromEntries(
160 (created.Cluster?.Tags ?? []).map((t) => [t.Key, t.Value]),
161 );
162 expect(tags.fixture).toBe("emr-cluster");
163 const policy = yield* emr.getAutoTerminationPolicy({
164 ClusterId: cluster.clusterId,
165 });
166 expect(policy.AutoTerminationPolicy?.IdleTimeout).toBe(3600);
167
168 // Update in place — step concurrency and auto-termination policy sync
169 // without replacement.
170 const { cluster: updated } = yield* deploy({
171 stepConcurrencyLevel: 4,
172 idleTimeoutSeconds: 7200,
173 });
174 expect(updated.clusterId).toBe(cluster.clusterId);
175 const afterUpdate = yield* emr.describeCluster({
176 ClusterId: cluster.clusterId,
177 });
178 expect(afterUpdate.Cluster?.StepConcurrencyLevel).toBe(4);
179 const updatedPolicy = yield* emr.getAutoTerminationPolicy({
180 ClusterId: cluster.clusterId,
181 });
182 expect(updatedPolicy.AutoTerminationPolicy?.IdleTimeout).toBe(7200);
183
184 // Destroy — the provider initiates termination (irreversible) and
185 // waits for the cluster to leave the active states.
186 yield* stack.destroy();
187 yield* assertClusterTerminating(cluster.clusterId);
188 }),
189 // create (~10-15 min) + update + termination initiation, one test.
190 {
191 tags: [
192 "provider:aws",
193 "provider:aws:ec2",
194 "provider:aws:emr",
195 "provider:aws:iam",
196 "provider:aws:s3",
197 "live",
198 ],
199 timeout: 2_700_000,
200 },
201);
202
203// Termination is verified as INITIATED (TERMINATING, irreversible) or fully
204// TERMINATED; full instance teardown takes ~5-10 more minutes server-side.
205const assertClusterTerminating = (clusterId: string) =>
206 Effect.gen(function* () {
207 const state = yield* emr.describeCluster({ ClusterId: clusterId }).pipe(
208 Effect.map((r) => r.Cluster?.Status?.State ?? "gone"),
209 Effect.catchTag("ClusterNotFound", () => Effect.succeed("gone" as const)),
210 );
211 if (
212 state !== "gone" &&
213 state !== "TERMINATING" &&
214 state !== "TERMINATED" &&
215 state !== "TERMINATED_WITH_ERRORS"
216 ) {
217 return yield* Effect.fail(
218 new Error(`EMR cluster '${clusterId}' still active (state: ${state})`),
219 );
220 }
221 }).pipe(
222 Effect.retry({
223 schedule: Schedule.max([
224 Schedule.fixed("10 seconds"),
225 Schedule.recurs(20),
226 ]),
227 }),
228 );