1import * as emrc from "@distilled.cloud/aws/emr-containers";
2import * as Effect from "effect/Effect";
3import * as Option from "effect/Option";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12
13export interface VirtualClusterContainerProvider {
14 /**
15 * The name of the underlying EKS cluster. Changing it replaces the virtual
16 * cluster.
17 */
18 id: string;
19 /**
20 * The container provider type.
21 * @default "EKS"
22 */
23 type?: "EKS";
24 /**
25 * The namespace registration for the underlying EKS cluster.
26 */
27 info?: {
28 eksInfo: {
29 /**
30 * The Kubernetes namespace the virtual cluster maps to. Changing it
31 * replaces the virtual cluster.
32 */
33 namespace?: string;
34 /**
35 * A node label selector restricting which nodes run the jobs.
36 */
37 nodeLabel?: string;
38 };
39 };
40}
41
42export interface VirtualClusterProps {
43 /**
44 * Name of the virtual cluster (1-64 characters: letters, digits, `.` `_`
45 * `/` `#` `-`). Changing the name replaces the virtual cluster.
46 * @default a generated physical name
47 */
48 virtualClusterName?: string;
49 /**
50 * The EKS cluster and namespace this virtual cluster maps to. Virtual
51 * clusters are immutable — any change replaces the virtual cluster.
52 */
53 containerProvider: VirtualClusterContainerProvider;
54 /**
55 * The ID of an EMR containers security configuration to attach. Changing
56 * it replaces the virtual cluster.
57 */
58 securityConfigurationId?: string;
59 /**
60 * Tags to apply to the virtual cluster. Merged with the internal Alchemy
61 * tags.
62 */
63 tags?: Record<string, string>;
64}
65
66export interface VirtualCluster extends Resource<
67 "AWS.EMRContainers.VirtualCluster",
68 VirtualClusterProps,
69 {
70 /** The ID of the virtual cluster. */
71 virtualClusterId: string;
72 /** The name of the virtual cluster. */
73 virtualClusterName: string;
74 /** The ARN of the virtual cluster. */
75 virtualClusterArn: string;
76 /** The name of the EKS cluster backing the virtual cluster. */
77 eksClusterName: string;
78 /** The virtual cluster state (e.g. `RUNNING`, `TERMINATING`). */
79 state?: string;
80 },
81 {},
82 Providers
83> {}
84
85/**
86 * An Amazon EMR on EKS virtual cluster — a registration of a Kubernetes
87 * namespace on an EKS cluster as an EMR job-submission target. Virtual
88 * clusters consume no resources themselves (no cost while idle); jobs
89 * submitted to the virtual cluster run as pods in the mapped namespace.
90 *
91 * The underlying EKS cluster must grant Amazon EMR on EKS access to the
92 * namespace (via EKS access entries when the cluster's authentication mode
93 * includes `API`, or the legacy `aws-auth` ConfigMap). Everything except tags
94 * is immutable — changes replace the virtual cluster.
95 *
96 * ### Creating Virtual Clusters
97 * **Example:** Register an EKS Namespace with EMR
98 * ```typescript
99 * import * as AWS from "alchemy/AWS";
100 *
101 * const cluster = yield* AWS.EKS.Cluster("Cluster", {
102 * roleArn: clusterRole.roleArn,
103 * resourcesVpcConfig: { subnetIds },
104 * accessConfig: { authenticationMode: "API_AND_CONFIG_MAP" },
105 * });
106 *
107 * const virtualCluster = yield* AWS.EMRContainers.VirtualCluster("Spark", {
108 * containerProvider: {
109 * id: cluster.clusterName,
110 * info: { eksInfo: { namespace: "emr" } },
111 * },
112 * });
113 * // virtualCluster.virtualClusterId is passed to StartJobRun
114 * ```
115 *
116 * @resource
117 */
118export const VirtualCluster = Resource<VirtualCluster>(
119 "AWS.EMRContainers.VirtualCluster",
120);
121
122export const VirtualClusterProvider = () =>
123 Provider.effect(
124 VirtualCluster,
125 Effect.gen(function* () {
126 const createName = Effect.fn(function* (
127 id: string,
128 props: { virtualClusterName?: string | undefined },
129 ) {
130 return (
131 props.virtualClusterName ??
132 (yield* createPhysicalName({ id, maxLength: 64 }))
133 );
134 });
135
136 const toContainerProvider = (
137 props: VirtualClusterContainerProvider,
138 ): emrc.ContainerProvider => ({
139 type: props.type ?? "EKS",
140 id: props.id,
141 info: props.info,
142 });
143
144 const toAttributes = (vc: emrc.VirtualCluster) => ({
145 virtualClusterId: vc.id!,
146 virtualClusterName: vc.name!,
147 virtualClusterArn: vc.arn!,
148 eksClusterName: vc.containerProvider?.id ?? "",
149 state: vc.state,
150 });
151
152 const observeById = Effect.fn(function* (id: string) {
153 return yield* emrc.describeVirtualCluster({ id }).pipe(
154 Effect.map((response) => response.virtualCluster),
155 Effect.catchTag(
156 ["ResourceNotFoundException", "ValidationException"],
157 // a malformed/unknown id is "not present", same as not-found
158 () => Effect.succeed(undefined),
159 ),
160 );
161 });
162
163 // Terminated virtual clusters linger in list results; only a RUNNING
164 // (or transiently ARRESTED) one with our name is "present".
165 const live = (state: string | undefined) =>
166 state === "RUNNING" || state === "ARRESTED";
167
168 const observeByName = Effect.fn(function* (name: string) {
169 return yield* emrc.listVirtualClusters
170 .items({ states: ["RUNNING", "ARRESTED"] })
171 .pipe(
172 Stream.filter((vc) => vc.name === name),
173 Stream.runHead,
174 Effect.map(Option.getOrUndefined),
175 );
176 });
177
178 const observe = Effect.fn(function* (
179 id: string | undefined,
180 name: string,
181 ) {
182 const byId = id !== undefined ? yield* observeById(id) : undefined;
183 if (byId !== undefined && live(byId.state)) {
184 return byId;
185 }
186 return yield* observeByName(name);
187 });
188
189 const syncTags = Effect.fn(function* (
190 vc: emrc.VirtualCluster,
191 desired: Record<string, string>,
192 ) {
193 // Diff against OBSERVED cloud tags (adoption may bring foreign tags).
194 const observed = Object.fromEntries(
195 Object.entries(vc.tags ?? {}).filter(
196 (entry): entry is [string, string] => entry[1] !== undefined,
197 ),
198 );
199 const { upsert, removed } = diffTags(observed, desired);
200 if (upsert.length > 0) {
201 yield* emrc.tagResource({
202 resourceArn: vc.arn!,
203 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
204 });
205 }
206 if (removed.length > 0) {
207 yield* emrc.untagResource({
208 resourceArn: vc.arn!,
209 tagKeys: removed,
210 });
211 }
212 });
213
214 return VirtualCluster.Provider.of({
215 stables: [
216 "virtualClusterId",
217 "virtualClusterName",
218 "virtualClusterArn",
219 "eksClusterName",
220 ],
221
222 list: () =>
223 Effect.gen(function* () {
224 const pages = yield* emrc.listVirtualClusters
225 .pages({ states: ["RUNNING", "ARRESTED"] })
226 .pipe(Stream.runCollect);
227 return Array.from(pages)
228 .flatMap((page) => page.virtualClusters ?? [])
229 .filter(
230 (vc) =>
231 vc.id !== undefined &&
232 vc.name !== undefined &&
233 vc.arn !== undefined,
234 )
235 .map(toAttributes);
236 }),
237
238 read: Effect.fn(function* ({ id, olds, output }) {
239 const name =
240 output?.virtualClusterName ?? (yield* createName(id, olds ?? {}));
241 const vc = yield* observe(output?.virtualClusterId, name);
242 if (vc?.id === undefined || vc.arn === undefined) {
243 return undefined;
244 }
245 const attrs = toAttributes(vc);
246 return (yield* hasAlchemyTags(id, vc.tags)) ? attrs : Unowned(attrs);
247 }),
248
249 diff: Effect.fn(function* ({ id, news, olds }) {
250 if (!isResolved(news)) return undefined;
251 const oldName = yield* createName(id, olds);
252 const newName = yield* createName(id, news);
253 if (
254 oldName !== newName ||
255 JSON.stringify(toContainerProvider(olds.containerProvider)) !==
256 JSON.stringify(toContainerProvider(news.containerProvider)) ||
257 olds.securityConfigurationId !== news.securityConfigurationId
258 ) {
259 // everything except tags is immutable
260 return { action: "replace" } as const;
261 }
262 }),
263
264 reconcile: Effect.fn(function* ({
265 id,
266 news,
267 output,
268 session,
269 instanceId,
270 }) {
271 const name =
272 output?.virtualClusterName ?? (yield* createName(id, news));
273 const internalTags = yield* createInternalTags(id);
274 const desiredTags = { ...news.tags, ...internalTags };
275
276 // 1. OBSERVE — cloud state is authoritative; output is an id cache
277 let vc = yield* observe(output?.virtualClusterId, name);
278
279 // 2. ENSURE — create if missing (synchronous; RUNNING immediately)
280 if (vc === undefined) {
281 const created = yield* emrc.createVirtualCluster({
282 // deterministic per instance: a retried create after a crashed
283 // reconcile never double-provisions
284 clientToken:
285 instanceId.replaceAll(/[^a-zA-Z0-9]/g, "").slice(0, 64) ||
286 "alchemy",
287 name,
288 containerProvider: toContainerProvider(news.containerProvider),
289 securityConfigurationId: news.securityConfigurationId,
290 tags: desiredTags,
291 });
292 vc =
293 created.id !== undefined
294 ? yield* emrc
295 .describeVirtualCluster({ id: created.id })
296 .pipe(Effect.map((r) => r.virtualCluster))
297 : yield* observeByName(name);
298 if (vc?.id === undefined) {
299 return yield* Effect.fail(
300 new emrc.ResourceNotFoundException({
301 message: `virtual cluster ${name} not visible after create`,
302 }),
303 );
304 }
305 }
306
307 // 3. SYNC TAGS — the only mutable aspect; diff observed vs desired
308 yield* syncTags(vc, desiredTags);
309
310 yield* session.note(vc.id!);
311 return toAttributes(vc);
312 }),
313
314 delete: Effect.fn(function* ({ output }) {
315 // deleteVirtualCluster is idempotent-by-observation: a terminated
316 // or missing virtual cluster reads as "not live" and is skipped
317 // (repeat deletes of a terminated id fail with ValidationException).
318 const vc = yield* emrc
319 .describeVirtualCluster({ id: output.virtualClusterId })
320 .pipe(
321 Effect.map((response) => response.virtualCluster),
322 Effect.catchTag(
323 ["ResourceNotFoundException", "ValidationException"],
324 () => Effect.succeed(undefined),
325 ),
326 );
327 if (vc === undefined || !live(vc.state)) {
328 return;
329 }
330 yield* emrc
331 .deleteVirtualCluster({ id: output.virtualClusterId })
332 .pipe(Effect.catchTag("ValidationException", () => Effect.void));
333 }),
334 });
335 }),
336 );