1import * as emr from "@distilled.cloud/aws/emr";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import { createPhysicalName } from "../../PhysicalName.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
13import { toWireSeconds } from "../../Util/Duration.ts";
14import type { Providers } from "../Providers.ts";
15
16/**
17 * EC2 instance topology for an EMR {@link Cluster} — a master (primary)
18 * instance group plus an optional core instance group, using the uniform
19 * instance-groups configuration.
20 */
21export interface ClusterInstancesConfig {
22 /**
23 * EC2 instance type of the single master (primary) node. Changing this
24 * replaces the cluster.
25 * @default "m5.xlarge"
26 */
27 masterInstanceType?: string;
28 /**
29 * EC2 instance type of the core nodes. Changing this replaces the cluster.
30 * @default "m5.xlarge"
31 */
32 coreInstanceType?: string;
33 /**
34 * Number of core nodes. `0` launches a master-only cluster. Resizing
35 * between non-zero counts is applied in place; crossing the `0` boundary
36 * (adding or removing the core group entirely) replaces the cluster.
37 * @default 1
38 */
39 coreInstanceCount?: number;
40 /**
41 * VPC subnet the cluster instances launch in. Changing the subnet replaces
42 * the cluster.
43 * @default an EMR-chosen subnet of the account's default VPC
44 */
45 ec2SubnetId?: string;
46 /**
47 * EC2 key pair for SSH access to the master node. Changing this replaces
48 * the cluster.
49 */
50 ec2KeyName?: string;
51 /**
52 * Whether the cluster stays alive (transitions to `WAITING`) after all
53 * steps complete rather than auto-terminating. Updateable in place.
54 * @default true
55 */
56 keepJobFlowAliveWhenNoSteps?: boolean;
57 /**
58 * Whether the cluster's EC2 instances are locked against termination by
59 * API calls or user intervention. Updateable in place (and always lifted
60 * before the provider terminates the cluster on destroy).
61 * @default false
62 */
63 terminationProtected?: boolean;
64 /**
65 * Whether EMR automatically replaces unhealthy core/task nodes.
66 * Updateable in place.
67 */
68 unhealthyNodeReplacement?: boolean;
69 /**
70 * EMR-managed security group for the master node. Changing this replaces
71 * the cluster.
72 * @default a security group created by EMR
73 */
74 emrManagedMasterSecurityGroup?: string;
75 /**
76 * EMR-managed security group for core/task nodes. Changing this replaces
77 * the cluster.
78 * @default a security group created by EMR
79 */
80 emrManagedSlaveSecurityGroup?: string;
81 /**
82 * Additional security groups applied to the master node. Changing these
83 * replaces the cluster.
84 */
85 additionalMasterSecurityGroups?: string[];
86 /**
87 * Additional security groups applied to core/task nodes. Changing these
88 * replaces the cluster.
89 */
90 additionalSlaveSecurityGroups?: string[];
91}
92
93export interface ClusterProps {
94 /**
95 * Display name of the cluster. If omitted, a deterministic physical name is
96 * generated. Changing the name replaces the cluster.
97 */
98 clusterName?: string;
99 /**
100 * Amazon EMR release, e.g. `"emr-7.5.0"`. Changing the release replaces the
101 * cluster.
102 */
103 releaseLabel: string;
104 /**
105 * Applications installed on the cluster, e.g. `["Spark", "Hadoop"]`.
106 * Changing the set replaces the cluster.
107 */
108 applications?: string[];
109 /**
110 * EC2 instance topology (master/core instance groups, subnet, key pair,
111 * security groups).
112 */
113 instances?: ClusterInstancesConfig;
114 /**
115 * IAM service role EMR assumes to manage cluster resources, e.g. a role
116 * trusting `elasticmapreduce.amazonaws.com`. Changing the role replaces
117 * the cluster.
118 */
119 serviceRole: string;
120 /**
121 * IAM instance profile (also called the job-flow role or EC2 instance
122 * profile) assumed by the cluster's EC2 instances. Changing it replaces
123 * the cluster.
124 */
125 jobFlowRole: string;
126 /**
127 * S3 URI where cluster logs are written, e.g. `"s3://my-bucket/logs/"`.
128 * Changing it replaces the cluster.
129 */
130 logUri?: string;
131 /**
132 * Name of an EMR {@link SecurityConfiguration} applied at launch. Changing
133 * it replaces the cluster.
134 */
135 securityConfiguration?: string;
136 /**
137 * Application configuration overrides (raw EMR `Configuration` objects,
138 * e.g. `[{ Classification: "spark-defaults", Properties: {...} }]`).
139 * Changing them replaces the cluster.
140 */
141 configurations?: emr.Configuration[];
142 /**
143 * Whether all IAM principals in the account can see the cluster.
144 * Updateable in place.
145 * @default true
146 */
147 visibleToAllUsers?: boolean;
148 /**
149 * Size of the EBS root volume (GiB) of each instance. Changing it replaces
150 * the cluster.
151 */
152 ebsRootVolumeSize?: number;
153 /**
154 * Custom AMI ID for the cluster instances. Changing it replaces the
155 * cluster.
156 */
157 customAmiId?: string;
158 /**
159 * Number of steps that may execute concurrently (1-256). Updateable in
160 * place.
161 * @default 1
162 */
163 stepConcurrencyLevel?: number;
164 /**
165 * Auto-termination policy — the cluster terminates itself after being idle
166 * for the configured duration (60 seconds to 7 days). Updateable in place;
167 * removing the prop detaches the policy.
168 */
169 autoTerminationPolicy?: {
170 /**
171 * Idle time after which the cluster auto-terminates — e.g. `"1 hour"` or
172 * `Duration.hours(1)`. Sent to AWS as whole seconds.
173 */
174 idleTimeout: Duration.Input;
175 };
176 /**
177 * User-defined tags for the cluster.
178 */
179 tags?: Record<string, string>;
180}
181
182export interface Cluster extends Resource<
183 "AWS.EMR.Cluster",
184 ClusterProps,
185 {
186 /** The ID of the cluster (e.g. `j-2AXXXXXXGAPLF`). */
187 clusterId: string;
188 /** The ARN of the cluster. */
189 clusterArn: string;
190 /** The name of the cluster. */
191 clusterName: string;
192 /** The cluster state (e.g. `STARTING`, `RUNNING`, `WAITING`). */
193 state: string;
194 /** The public DNS name of the primary node, when reachable. */
195 masterPublicDnsName: string | undefined;
196 /** The tags applied to the cluster. */
197 tags: Record<string, string>;
198 },
199 never,
200 Providers
201> {}
202
203/**
204 * A provisioned Amazon EMR cluster (job flow) running open-source big-data
205 * frameworks such as Apache Spark and Hadoop on EC2 instances.
206 *
207 * Clusters take roughly 10-15 minutes to reach `WAITING` and bill per
208 * instance-hour while they exist. Each cluster needs an EMR service role and
209 * an EC2 instance profile (job-flow role); destroy clusters you are not
210 * using, or set `autoTerminationPolicy` as a safety net.
211 * ### Creating a Cluster
212 * **Example:** Spark Cluster in a Default-VPC Subnet
213 * ```typescript
214 * const cluster = yield* Cluster("Analytics", {
215 * releaseLabel: "emr-7.5.0",
216 * applications: ["Spark", "Hadoop"],
217 * serviceRole: serviceRole.roleName,
218 * jobFlowRole: instanceProfile.instanceProfileName,
219 * logUri: Output.interpolate`s3://${logsBucket.bucketName}/logs/`,
220 * instances: {
221 * masterInstanceType: "m5.xlarge",
222 * coreInstanceType: "m5.xlarge",
223 * coreInstanceCount: 1,
224 * ec2SubnetId: subnetId,
225 * },
226 * });
227 * ```
228 *
229 * **Example:** Cluster with an Auto-Termination Safety Net
230 * ```typescript
231 * const cluster = yield* Cluster("Batch", {
232 * releaseLabel: "emr-7.5.0",
233 * applications: ["Spark"],
234 * serviceRole: serviceRole.roleName,
235 * jobFlowRole: instanceProfile.instanceProfileName,
236 * autoTerminationPolicy: { idleTimeout: "1 hour" },
237 * stepConcurrencyLevel: 4,
238 * });
239 * ```
240 *
241 * ### Applying a Security Configuration
242 * **Example:** Cluster with Encryption Settings
243 * ```typescript
244 * const config = yield* SecurityConfiguration("Encryption", {
245 * securityConfiguration: {
246 * EncryptionConfiguration: {
247 * EnableInTransitEncryption: false,
248 * EnableAtRestEncryption: false,
249 * },
250 * },
251 * });
252 * const cluster = yield* Cluster("Secure", {
253 * releaseLabel: "emr-7.5.0",
254 * serviceRole: serviceRole.roleName,
255 * jobFlowRole: instanceProfile.instanceProfileName,
256 * securityConfiguration: config.securityConfigurationName,
257 * });
258 * ```
259 *
260 * @resource
261 */
262export const Cluster = Resource<Cluster>("AWS.EMR.Cluster");
263
264// States in which a cluster is (still) a live, converging resource.
265const ACTIVE_STATES = [
266 "STARTING",
267 "BOOTSTRAPPING",
268 "RUNNING",
269 "WAITING",
270] as const;
271
272// Terminal states: the cluster is gone or irreversibly going away.
273const TERMINAL_STATES = new Set([
274 "TERMINATING",
275 "TERMINATED",
276 "TERMINATED_WITH_ERRORS",
277]);
278
279const toTagRecord = (
280 tags: ReadonlyArray<{ Key?: string; Value?: string }> | undefined,
281): Record<string, string> =>
282 Object.fromEntries(
283 (tags ?? [])
284 .filter(
285 (tag): tag is { Key: string; Value: string } =>
286 typeof tag.Key === "string" && typeof tag.Value === "string",
287 )
288 .map((tag) => [tag.Key, tag.Value]),
289 );
290
291const sameStringSet = (
292 a: readonly string[] | undefined,
293 b: readonly string[] | undefined,
294): boolean => {
295 const left = [...(a ?? [])].sort();
296 const right = [...(b ?? [])].sort();
297 return left.length === right.length && left.every((v, i) => v === right[i]);
298};
299
300/** Build the master/core instance-group configuration for RunJobFlow. */
301const buildInstanceGroups = (
302 instances: ClusterInstancesConfig | undefined,
303): emr.InstanceGroupConfig[] => {
304 const coreCount = instances?.coreInstanceCount ?? 1;
305 return [
306 {
307 Name: "Master",
308 InstanceRole: "MASTER",
309 InstanceType: instances?.masterInstanceType ?? "m5.xlarge",
310 InstanceCount: 1,
311 },
312 ...(coreCount > 0
313 ? [
314 {
315 Name: "Core",
316 InstanceRole: "CORE" as const,
317 InstanceType: instances?.coreInstanceType ?? "m5.xlarge",
318 InstanceCount: coreCount,
319 },
320 ]
321 : []),
322 ];
323};
324
325export const ClusterProvider = () =>
326 Provider.effect(
327 Cluster,
328 Effect.gen(function* () {
329 const toName = (id: string, props: Partial<ClusterProps>) =>
330 props.clusterName
331 ? Effect.succeed(props.clusterName)
332 : createPhysicalName({ id, maxLength: 80 });
333
334 // Describe a cluster by id; a terminal or missing cluster reads as
335 // undefined — a terminated EMR cluster cannot be revived or updated.
336 const readCluster = Effect.fn(function* (clusterId: string) {
337 const response = yield* emr
338 .describeCluster({ ClusterId: clusterId })
339 .pipe(
340 Effect.catchTag("ClusterNotFound", () => Effect.succeed(undefined)),
341 );
342 const cluster = response?.Cluster;
343 if (!cluster?.Id) return undefined;
344 if (TERMINAL_STATES.has(cluster.Status?.State ?? "")) return undefined;
345 return cluster;
346 });
347
348 // Cluster names are not unique in EMR; the deterministic physical name
349 // is. Find the live (non-terminal) cluster carrying our name — used
350 // when state was lost and we only know the derived name.
351 const findClusterByName = Effect.fn(function* (name: string) {
352 const summary = yield* emr.listClusters
353 .items({ ClusterStates: [...ACTIVE_STATES] })
354 .pipe(
355 Stream.filter((summary) => summary.Name === name),
356 Stream.runHead,
357 Effect.map(Option.getOrUndefined),
358 );
359 return summary?.Id ? yield* readCluster(summary.Id) : undefined;
360 });
361
362 // Bounded readiness wait. A Spark/Hadoop cluster typically reaches
363 // WAITING in 10-15 minutes; budget ~30 min (90 * 20s). A terminal
364 // state stops the wait immediately (non-retryable).
365 const waitForReady = Effect.fn(function* (clusterId: string) {
366 const policy = Schedule.max([
367 Schedule.fixed("20 seconds"),
368 Schedule.recurs(90),
369 ]);
370 return yield* emr.describeCluster({ ClusterId: clusterId }).pipe(
371 Effect.flatMap((response) => {
372 const cluster = response.Cluster;
373 const state = cluster?.Status?.State ?? "STARTING";
374 if (TERMINAL_STATES.has(state)) {
375 const reason = cluster?.Status?.StateChangeReason;
376 return Effect.fail(
377 new Error(
378 `EMR cluster '${clusterId}' reached terminal state '${state}'` +
379 ` (${reason?.Code ?? "unknown"}: ${reason?.Message ?? "no message"})`,
380 ),
381 );
382 }
383 if (state !== "WAITING" && state !== "RUNNING") {
384 return Effect.fail(
385 new Error(
386 `EMR cluster '${clusterId}' not ready (state: ${state})`,
387 ),
388 );
389 }
390 return Effect.succeed(cluster!);
391 }),
392 Effect.retry({
393 while: (e) =>
394 e instanceof Error && !e.message.includes("terminal state"),
395 schedule: policy,
396 }),
397 );
398 });
399
400 const toAttrs = Effect.fn(function* (cluster: emr.Cluster) {
401 if (!cluster.Id || !cluster.ClusterArn || !cluster.Name) {
402 return yield* Effect.fail(
403 new Error(`EMR cluster '${cluster.Id}' is missing its ARN or name`),
404 );
405 }
406 return {
407 clusterId: cluster.Id,
408 clusterArn: cluster.ClusterArn,
409 clusterName: cluster.Name,
410 state: cluster.Status?.State ?? "STARTING",
411 masterPublicDnsName: cluster.MasterPublicDnsName,
412 tags: toTagRecord(cluster.Tags),
413 };
414 });
415
416 return {
417 stables: ["clusterId", "clusterArn", "clusterName"],
418
419 diff: Effect.fn(function* ({ id, olds, news }) {
420 if (!isResolved(news)) return undefined;
421 const n = news;
422 const o = olds;
423 if (n === undefined || o === undefined) return undefined;
424 if ((yield* toName(id, o)) !== (yield* toName(id, n))) {
425 return { action: "replace" } as const;
426 }
427 // Create-only cluster-level properties force a replacement.
428 if (n.releaseLabel !== o.releaseLabel) {
429 return { action: "replace" } as const;
430 }
431 if (!sameStringSet(n.applications, o.applications)) {
432 return { action: "replace" } as const;
433 }
434 if (n.serviceRole !== o.serviceRole) {
435 return { action: "replace" } as const;
436 }
437 if (n.jobFlowRole !== o.jobFlowRole) {
438 return { action: "replace" } as const;
439 }
440 if ((n.logUri ?? undefined) !== (o.logUri ?? undefined)) {
441 return { action: "replace" } as const;
442 }
443 if (
444 (n.securityConfiguration ?? undefined) !==
445 (o.securityConfiguration ?? undefined)
446 ) {
447 return { action: "replace" } as const;
448 }
449 if (
450 JSON.stringify(n.configurations ?? []) !==
451 JSON.stringify(o.configurations ?? [])
452 ) {
453 return { action: "replace" } as const;
454 }
455 if (
456 (n.ebsRootVolumeSize ?? undefined) !==
457 (o.ebsRootVolumeSize ?? undefined)
458 ) {
459 return { action: "replace" } as const;
460 }
461 if ((n.customAmiId ?? undefined) !== (o.customAmiId ?? undefined)) {
462 return { action: "replace" } as const;
463 }
464 // Create-only instance topology.
465 const ni = n.instances ?? {};
466 const oi = o.instances ?? {};
467 if (
468 (ni.masterInstanceType ?? "m5.xlarge") !==
469 (oi.masterInstanceType ?? "m5.xlarge") ||
470 (ni.coreInstanceType ?? "m5.xlarge") !==
471 (oi.coreInstanceType ?? "m5.xlarge") ||
472 (ni.ec2SubnetId ?? undefined) !== (oi.ec2SubnetId ?? undefined) ||
473 (ni.ec2KeyName ?? undefined) !== (oi.ec2KeyName ?? undefined) ||
474 (ni.emrManagedMasterSecurityGroup ?? undefined) !==
475 (oi.emrManagedMasterSecurityGroup ?? undefined) ||
476 (ni.emrManagedSlaveSecurityGroup ?? undefined) !==
477 (oi.emrManagedSlaveSecurityGroup ?? undefined) ||
478 !sameStringSet(
479 ni.additionalMasterSecurityGroups,
480 oi.additionalMasterSecurityGroups,
481 ) ||
482 !sameStringSet(
483 ni.additionalSlaveSecurityGroups,
484 oi.additionalSlaveSecurityGroups,
485 )
486 ) {
487 return { action: "replace" } as const;
488 }
489 // The core group can resize in place, but crossing the 0 boundary
490 // adds/removes the whole group — replacement.
491 const newCore = ni.coreInstanceCount ?? 1;
492 const oldCore = oi.coreInstanceCount ?? 1;
493 if ((newCore === 0) !== (oldCore === 0)) {
494 return { action: "replace" } as const;
495 }
496 }),
497
498 read: Effect.fn(function* ({ id, olds, output }) {
499 const cluster = output?.clusterId
500 ? yield* readCluster(output.clusterId)
501 : yield* findClusterByName(yield* toName(id, olds ?? {}));
502 if (!cluster) return undefined;
503 const attrs = yield* toAttrs(cluster);
504 return (yield* hasAlchemyTags(id, attrs.tags))
505 ? attrs
506 : Unowned(attrs);
507 }),
508
509 reconcile: Effect.fn(function* ({ id, news, output, session }) {
510 const props = news!;
511 const name = output?.clusterName ?? (yield* toName(id, props));
512 const internalTags = yield* createInternalTags(id);
513 const desiredTags = { ...internalTags, ...props.tags };
514
515 // 1. Observe — cloud state is authoritative; output is only an id
516 // cache. A terminated cluster reads as missing.
517 let observed = output?.clusterId
518 ? yield* readCluster(output.clusterId)
519 : yield* findClusterByName(name);
520
521 // 2. Ensure — launch the job flow if missing. Create errors
522 // (invalid roles, subnet, release) propagate directly.
523 let clusterId = observed?.Id;
524 if (clusterId === undefined) {
525 const instances = props.instances ?? {};
526 const created = yield* emr.runJobFlow({
527 Name: name,
528 ReleaseLabel: props.releaseLabel,
529 Applications: (props.applications ?? []).map((Name) => ({
530 Name,
531 })),
532 Instances: {
533 InstanceGroups: buildInstanceGroups(props.instances),
534 Ec2SubnetId: instances.ec2SubnetId,
535 Ec2KeyName: instances.ec2KeyName,
536 KeepJobFlowAliveWhenNoSteps:
537 instances.keepJobFlowAliveWhenNoSteps ?? true,
538 TerminationProtected: instances.terminationProtected ?? false,
539 UnhealthyNodeReplacement: instances.unhealthyNodeReplacement,
540 EmrManagedMasterSecurityGroup:
541 instances.emrManagedMasterSecurityGroup,
542 EmrManagedSlaveSecurityGroup:
543 instances.emrManagedSlaveSecurityGroup,
544 AdditionalMasterSecurityGroups:
545 instances.additionalMasterSecurityGroups,
546 AdditionalSlaveSecurityGroups:
547 instances.additionalSlaveSecurityGroups,
548 },
549 ServiceRole: props.serviceRole,
550 JobFlowRole: props.jobFlowRole,
551 LogUri: props.logUri,
552 SecurityConfiguration: props.securityConfiguration,
553 Configurations: props.configurations,
554 VisibleToAllUsers: props.visibleToAllUsers ?? true,
555 EbsRootVolumeSize: props.ebsRootVolumeSize,
556 CustomAmiId: props.customAmiId,
557 StepConcurrencyLevel: props.stepConcurrencyLevel,
558 AutoTerminationPolicy: props.autoTerminationPolicy
559 ? {
560 IdleTimeout: toWireSeconds(
561 props.autoTerminationPolicy.idleTimeout,
562 ),
563 }
564 : undefined,
565 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
566 Key,
567 Value,
568 })),
569 });
570 if (!created.JobFlowId) {
571 return yield* Effect.fail(
572 new Error(`RunJobFlow for '${name}' returned no JobFlowId`),
573 );
574 }
575 clusterId = created.JobFlowId;
576 }
577
578 // Provisioning surfaces as STARTING/BOOTSTRAPPING; wait (bounded)
579 // for WAITING/RUNNING so modify calls don't race bootstrap.
580 observed = yield* waitForReady(clusterId);
581
582 // 3. Sync — per mutable aspect, diff OBSERVED state against the
583 // desired props and apply only the delta.
584 const desiredVisible = props.visibleToAllUsers ?? true;
585 if ((observed.VisibleToAllUsers ?? true) !== desiredVisible) {
586 yield* emr.setVisibleToAllUsers({
587 JobFlowIds: [clusterId],
588 VisibleToAllUsers: desiredVisible,
589 });
590 }
591
592 const desiredKeepAlive =
593 props.instances?.keepJobFlowAliveWhenNoSteps ?? true;
594 const observedKeepAlive = !(observed.AutoTerminate ?? false);
595 if (observedKeepAlive !== desiredKeepAlive) {
596 yield* emr.setKeepJobFlowAliveWhenNoSteps({
597 JobFlowIds: [clusterId],
598 KeepJobFlowAliveWhenNoSteps: desiredKeepAlive,
599 });
600 }
601
602 const desiredProtected = props.instances?.terminationProtected;
603 if (
604 desiredProtected !== undefined &&
605 (observed.TerminationProtected ?? false) !== desiredProtected
606 ) {
607 yield* emr.setTerminationProtection({
608 JobFlowIds: [clusterId],
609 TerminationProtected: desiredProtected,
610 });
611 }
612
613 const desiredUnhealthy = props.instances?.unhealthyNodeReplacement;
614 if (
615 desiredUnhealthy !== undefined &&
616 (observed.UnhealthyNodeReplacement ?? false) !== desiredUnhealthy
617 ) {
618 yield* emr.setUnhealthyNodeReplacement({
619 JobFlowIds: [clusterId],
620 UnhealthyNodeReplacement: desiredUnhealthy,
621 });
622 }
623
624 if (
625 props.stepConcurrencyLevel !== undefined &&
626 (observed.StepConcurrencyLevel ?? 1) !== props.stepConcurrencyLevel
627 ) {
628 yield* emr.modifyCluster({
629 ClusterId: clusterId,
630 StepConcurrencyLevel: props.stepConcurrencyLevel,
631 });
632 }
633
634 // Auto-termination policy: observe, then put/remove the delta.
635 const observedPolicy = yield* emr
636 .getAutoTerminationPolicy({ ClusterId: clusterId })
637 .pipe(
638 Effect.map((r) => r.AutoTerminationPolicy),
639 Effect.catch(() => Effect.succeed(undefined)),
640 );
641 if (props.autoTerminationPolicy !== undefined) {
642 const desiredIdleTimeout = toWireSeconds(
643 props.autoTerminationPolicy.idleTimeout,
644 );
645 if (observedPolicy?.IdleTimeout !== desiredIdleTimeout) {
646 yield* emr.putAutoTerminationPolicy({
647 ClusterId: clusterId,
648 AutoTerminationPolicy: {
649 IdleTimeout: desiredIdleTimeout,
650 },
651 });
652 }
653 } else if (observedPolicy !== undefined) {
654 yield* emr.removeAutoTerminationPolicy({ ClusterId: clusterId });
655 }
656
657 // Core instance-group resize (in place between non-zero counts).
658 const desiredCore = props.instances?.coreInstanceCount ?? 1;
659 if (desiredCore > 0) {
660 const core = yield* emr.listInstanceGroups
661 .items({ ClusterId: clusterId })
662 .pipe(
663 Stream.filter((g) => g.InstanceGroupType === "CORE"),
664 Stream.runHead,
665 Effect.map(Option.getOrUndefined),
666 );
667 if (
668 core?.Id !== undefined &&
669 (core.RequestedInstanceCount ?? 0) !== desiredCore
670 ) {
671 yield* emr.modifyInstanceGroups({
672 ClusterId: clusterId,
673 InstanceGroups: [
674 { InstanceGroupId: core.Id, InstanceCount: desiredCore },
675 ],
676 });
677 }
678 }
679
680 // 3b. Sync tags — diff against OBSERVED cloud tags.
681 const observedTags = toTagRecord(observed.Tags);
682 const { removed, upsert } = diffTags(observedTags, desiredTags);
683 if (upsert.length > 0) {
684 yield* emr.addTags({ ResourceId: clusterId, Tags: upsert });
685 }
686 if (removed.length > 0) {
687 yield* emr.removeTags({ ResourceId: clusterId, TagKeys: removed });
688 }
689
690 // 4. Return fresh attributes.
691 const final = yield* readCluster(clusterId);
692 yield* session.note(clusterId);
693 return yield* toAttrs(final ?? observed);
694 }),
695
696 delete: Effect.fn(function* ({ output }) {
697 const clusterId = output.clusterId;
698 // A cluster that is already terminal (or gone) is success.
699 const observed = yield* readCluster(clusterId);
700 if (observed === undefined) return;
701 // Lift termination protection so TerminateJobFlows succeeds.
702 if (observed.TerminationProtected === true) {
703 yield* emr
704 .setTerminationProtection({
705 JobFlowIds: [clusterId],
706 TerminationProtected: false,
707 })
708 .pipe(Effect.catchTag("JobFlowNotFound", () => Effect.void));
709 }
710 yield* emr
711 .terminateJobFlows({ JobFlowIds: [clusterId] })
712 .pipe(Effect.catchTag("JobFlowNotFound", () => Effect.void));
713 // Termination is irreversible once initiated; wait (bounded) for
714 // the state to leave the active set. Full instance teardown takes
715 // 5-10 more minutes server-side and needs no babysitting.
716 yield* emr.describeCluster({ ClusterId: clusterId }).pipe(
717 Effect.flatMap((response) => {
718 const state = response.Cluster?.Status?.State ?? "TERMINATED";
719 return TERMINAL_STATES.has(state)
720 ? Effect.void
721 : Effect.fail(
722 new Error(
723 `EMR cluster '${clusterId}' still active (state: ${state})`,
724 ),
725 );
726 }),
727 Effect.catchTag("ClusterNotFound", () => Effect.void),
728 Effect.retry({
729 schedule: Schedule.max([
730 Schedule.fixed("10 seconds"),
731 Schedule.recurs(30),
732 ]),
733 }),
734 );
735 }),
736
737 list: () =>
738 emr.listClusters.items({ ClusterStates: [...ACTIVE_STATES] }).pipe(
739 Stream.runCollect,
740 Effect.map((chunk) =>
741 Array.from(chunk).flatMap((summary) =>
742 summary.Id ? [summary.Id] : [],
743 ),
744 ),
745 Effect.flatMap(
746 Effect.forEach((clusterId) => readCluster(clusterId), {
747 concurrency: 4,
748 }),
749 ),
750 Effect.map((clusters) =>
751 clusters.flatMap((cluster) =>
752 cluster === undefined ? [] : [cluster],
753 ),
754 ),
755 Effect.flatMap(
756 Effect.forEach((cluster) => toAttrs(cluster), {
757 concurrency: 4,
758 }),
759 ),
760 ),
761 };
762 }),
763 );