1import * as emrc from "@distilled.cloud/aws/emr-containers";
2import * as Data from "effect/Data";
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 type { Providers } from "../Providers.ts";
14
15export interface JobTemplateSparkSubmitJobDriver {
16 /**
17 * The entry point of the Spark application — an S3 URI to a jar or Python
18 * script (e.g. `s3://my-bucket/scripts/etl.py`).
19 */
20 entryPoint: string;
21 /**
22 * Arguments passed to the entry point.
23 */
24 entryPointArguments?: string[];
25 /**
26 * `spark-submit` parameters (e.g. `--conf spark.executor.instances=2`).
27 */
28 sparkSubmitParameters?: string;
29}
30
31export interface JobTemplateSparkSqlJobDriver {
32 /**
33 * The S3 URI of the SQL file to execute.
34 */
35 entryPoint?: string;
36 /**
37 * Spark SQL parameters.
38 */
39 sparkSqlParameters?: string;
40}
41
42export interface JobTemplateJobDriver {
43 /**
44 * The Spark submit job driver.
45 */
46 sparkSubmitJobDriver?: JobTemplateSparkSubmitJobDriver;
47 /**
48 * The Spark SQL job driver.
49 */
50 sparkSqlJobDriver?: JobTemplateSparkSqlJobDriver;
51}
52
53export interface JobTemplateConfiguration {
54 /**
55 * The classification of the configuration (e.g. `spark-defaults`).
56 */
57 classification: string;
58 /**
59 * Configuration properties for the classification.
60 */
61 properties?: Record<string, string>;
62 /**
63 * Nested configurations.
64 */
65 configurations?: JobTemplateConfiguration[];
66}
67
68export interface JobTemplateMonitoringConfiguration {
69 /**
70 * Whether the persistent application UI (Spark history server) is enabled.
71 */
72 persistentAppUI?: "ENABLED" | "DISABLED" | (string & {});
73 /**
74 * CloudWatch log delivery for the job's containers.
75 */
76 cloudWatchMonitoringConfiguration?: {
77 /** The CloudWatch log group to deliver logs to. */
78 logGroupName?: string;
79 /** A prefix for the log stream names. */
80 logStreamNamePrefix?: string;
81 };
82 /**
83 * S3 log delivery for the job's containers.
84 */
85 s3MonitoringConfiguration?: {
86 /** The S3 URI to deliver logs to. */
87 logUri?: string;
88 };
89}
90
91export interface JobTemplateConfigurationOverrides {
92 /**
93 * Application configurations (Spark/Hadoop classification properties).
94 */
95 applicationConfiguration?: JobTemplateConfiguration[];
96 /**
97 * Monitoring (log delivery) configuration.
98 */
99 monitoringConfiguration?: JobTemplateMonitoringConfiguration;
100}
101
102export interface JobTemplateParameterConfiguration {
103 /**
104 * The data type of the template parameter.
105 * @default "STRING"
106 */
107 type?: "NUMBER" | "STRING" | (string & {});
108 /**
109 * The default value used when `StartJobRun` omits the parameter.
110 */
111 defaultValue?: string;
112}
113
114export interface JobTemplateDataProps {
115 /**
116 * The IAM execution role ARN the job runs as. May reference template
117 * parameters (e.g. `${ExecutionRoleArn}`).
118 */
119 executionRoleArn: string;
120 /**
121 * The EMR release label (e.g. `emr-7.5.0-latest`).
122 */
123 releaseLabel: string;
124 /**
125 * Configuration overrides applied to jobs started from the template.
126 */
127 configurationOverrides?: JobTemplateConfigurationOverrides;
128 /**
129 * The job driver (Spark submit or Spark SQL) for jobs started from the
130 * template.
131 */
132 jobDriver: JobTemplateJobDriver;
133 /**
134 * Declares the `${placeholders}` used in the template and their types /
135 * default values.
136 */
137 parameterConfiguration?: Record<string, JobTemplateParameterConfiguration>;
138 /**
139 * Tags applied to the *job runs* started from this template (distinct from
140 * the template's own `tags`).
141 */
142 jobTags?: Record<string, string>;
143}
144
145export interface JobTemplateProps {
146 /**
147 * Name of the job template (1-64 characters). Changing the name replaces
148 * the job template.
149 * @default a generated physical name
150 */
151 jobTemplateName?: string;
152 /**
153 * The StartJobRun values the template stores. Job templates are immutable —
154 * any change replaces the template.
155 */
156 jobTemplateData: JobTemplateDataProps;
157 /**
158 * The KMS key ARN used to encrypt the template. Changing it replaces the
159 * job template.
160 */
161 kmsKeyArn?: string;
162 /**
163 * Tags to apply to the job template. Merged with the internal Alchemy
164 * tags. Tags are set at creation only — the EMR containers `TagResource`
165 * API rejects job template ARNs (typed `InvalidResourceArn`), so changing
166 * tags replaces the template.
167 */
168 tags?: Record<string, string>;
169}
170
171export interface JobTemplate extends Resource<
172 "AWS.EMRContainers.JobTemplate",
173 JobTemplateProps,
174 {
175 /** The ID of the job template. */
176 jobTemplateId: string;
177 /** The name of the job template. */
178 jobTemplateName: string;
179 /** The ARN of the job template. */
180 jobTemplateArn: string;
181 },
182 {},
183 Providers
184> {}
185
186export class JobTemplateConsistencyError extends Data.TaggedError(
187 "JobTemplateConsistencyError",
188)<{
189 readonly jobTemplateId: string;
190 readonly jobTemplateName: string;
191 readonly operation: "create" | "delete";
192 readonly message: string;
193}> {}
194
195/**
196 * An Amazon EMR on EKS job template — a stored set of `StartJobRun` values
197 * (execution role, release label, job driver, configuration) that can be
198 * referenced by ID when starting job runs, optionally with `${placeholder}`
199 * parameters filled in per run.
200 *
201 * Job templates are account-level and fully immutable: any change (including
202 * tags, which the tagging API does not support post-create for templates)
203 * replaces the template. They pair with the
204 * {@link StartJobRun | AWS.EMRContainers.StartJobRun} binding — a Lambda can
205 * start a templated Spark job with just the template ID and parameter values.
206 *
207 * ### Creating Job Templates
208 * **Example:** A Spark Job Template
209 * ```typescript
210 * import * as AWS from "alchemy/AWS";
211 *
212 * const template = yield* AWS.EMRContainers.JobTemplate("EtlTemplate", {
213 * jobTemplateData: {
214 * executionRoleArn: jobRole.roleArn,
215 * releaseLabel: "emr-7.5.0-latest",
216 * jobDriver: {
217 * sparkSubmitJobDriver: {
218 * entryPoint: "s3://my-bucket/scripts/etl.py",
219 * sparkSubmitParameters: "--conf spark.executor.instances=2",
220 * },
221 * },
222 * },
223 * });
224 * ```
225 *
226 * **Example:** Parameterized Template
227 * ```typescript
228 * const template = yield* AWS.EMRContainers.JobTemplate("Parameterized", {
229 * jobTemplateData: {
230 * executionRoleArn: jobRole.roleArn,
231 * releaseLabel: "emr-7.5.0-latest",
232 * jobDriver: {
233 * sparkSubmitJobDriver: { entryPoint: "${EntryPoint}" },
234 * },
235 * parameterConfiguration: {
236 * EntryPoint: { type: "STRING" },
237 * },
238 * },
239 * });
240 * // StartJobRun with jobTemplateId + jobTemplateParameters: { EntryPoint: "s3://..." }
241 * ```
242 *
243 * @resource
244 */
245export const JobTemplate = Resource<JobTemplate>(
246 "AWS.EMRContainers.JobTemplate",
247);
248
249/** Convert declared props to the distilled wire shape. */
250const toJobTemplateData = (
251 data: JobTemplateDataProps,
252): emrc.JobTemplateData => ({
253 executionRoleArn: data.executionRoleArn,
254 releaseLabel: data.releaseLabel,
255 configurationOverrides: data.configurationOverrides,
256 jobDriver: data.jobDriver,
257 parameterConfiguration: data.parameterConfiguration,
258 jobTags: data.jobTags,
259});
260
261export const JobTemplateProvider = () =>
262 Provider.effect(
263 JobTemplate,
264 Effect.gen(function* () {
265 const createName = Effect.fn(function* (
266 id: string,
267 props: { jobTemplateName?: string | undefined },
268 ) {
269 return (
270 props.jobTemplateName ??
271 (yield* createPhysicalName({ id, maxLength: 64 }))
272 );
273 });
274
275 const toAttributes = (jt: emrc.JobTemplate) => ({
276 jobTemplateId: jt.id!,
277 jobTemplateName: jt.name!,
278 jobTemplateArn: jt.arn!,
279 });
280
281 const observeById = Effect.fn(function* (id: string) {
282 return yield* emrc.describeJobTemplate({ id }).pipe(
283 Effect.map((response) => response.jobTemplate),
284 Effect.catchTag(
285 ["ResourceNotFoundException", "ValidationException"],
286 // a malformed/unknown id is "not present", same as not-found
287 () => Effect.succeed(undefined),
288 ),
289 );
290 });
291
292 const observeByName = Effect.fn(function* (name: string) {
293 return yield* emrc.listJobTemplates.items({}).pipe(
294 Stream.filter((jt) => jt.name === name),
295 Stream.runHead,
296 Effect.map(Option.getOrUndefined),
297 );
298 });
299
300 const observeListedById = Effect.fn(function* (id: string) {
301 return yield* emrc.listJobTemplates.items({}).pipe(
302 Stream.filter((jt) => jt.id === id),
303 Stream.runHead,
304 Effect.map(Option.getOrUndefined),
305 );
306 });
307
308 const awaitListVisibility = (id: string) =>
309 Effect.repeat(observeListedById(id), {
310 schedule: Schedule.spaced("3 seconds"),
311 until: (jt) => jt !== undefined,
312 times: 10,
313 });
314
315 const awaitAbsent = (id: string) =>
316 Effect.repeat(
317 Effect.all([observeById(id), observeListedById(id)] as const),
318 {
319 schedule: Schedule.spaced("3 seconds"),
320 until: ([described, listed]) =>
321 described === undefined && listed === undefined,
322 times: 10,
323 },
324 );
325
326 const observe = Effect.fn(function* (
327 id: string | undefined,
328 name: string,
329 ) {
330 const byId = id !== undefined ? yield* observeById(id) : undefined;
331 return byId ?? (yield* observeByName(name));
332 });
333
334 return JobTemplate.Provider.of({
335 stables: ["jobTemplateId", "jobTemplateName", "jobTemplateArn"],
336
337 list: () =>
338 Effect.gen(function* () {
339 const pages = yield* emrc.listJobTemplates
340 .pages({})
341 .pipe(Stream.runCollect);
342 return Array.from(pages)
343 .flatMap((page) => page.templates ?? [])
344 .filter(
345 (jt) =>
346 jt.id !== undefined &&
347 jt.name !== undefined &&
348 jt.arn !== undefined,
349 )
350 .map(toAttributes);
351 }),
352
353 read: Effect.fn(function* ({ id, olds, output }) {
354 const name =
355 output?.jobTemplateName ?? (yield* createName(id, olds ?? {}));
356 const jt = yield* observe(output?.jobTemplateId, name);
357 if (jt?.id === undefined || jt.arn === undefined) {
358 return undefined;
359 }
360 const attrs = toAttributes(jt);
361 return (yield* hasAlchemyTags(id, jt.tags)) ? attrs : Unowned(attrs);
362 }),
363
364 diff: Effect.fn(function* ({ id, news, olds }) {
365 if (!isResolved(news)) return undefined;
366 const oldName = yield* createName(id, olds);
367 const newName = yield* createName(id, news);
368 const { upsert, removed } = diffTags(
369 olds.tags ?? {},
370 news.tags ?? {},
371 );
372 if (
373 oldName !== newName ||
374 JSON.stringify(toJobTemplateData(olds.jobTemplateData)) !==
375 JSON.stringify(toJobTemplateData(news.jobTemplateData)) ||
376 olds.kmsKeyArn !== news.kmsKeyArn ||
377 upsert.length > 0 ||
378 removed.length > 0
379 ) {
380 // job templates are fully immutable — even tags: the tagging API
381 // rejects job template ARNs (typed InvalidResourceArn), so tags
382 // are set at creation only and any change replaces
383 return { action: "replace" } as const;
384 }
385 }),
386
387 reconcile: Effect.fn(function* ({
388 id,
389 news,
390 output,
391 session,
392 instanceId,
393 }) {
394 const name = output?.jobTemplateName ?? (yield* createName(id, news));
395 const internalTags = yield* createInternalTags(id);
396 const desiredTags = { ...news.tags, ...internalTags };
397
398 // 1. OBSERVE — cloud state is authoritative; output is an id cache
399 let jt = yield* observe(output?.jobTemplateId, name);
400
401 // 2. ENSURE — create if missing (synchronous, available immediately)
402 if (jt === undefined) {
403 const created = yield* emrc.createJobTemplate({
404 // deterministic per instance: a retried create after a crashed
405 // reconcile never double-provisions
406 clientToken:
407 instanceId.replaceAll(/[^a-zA-Z0-9]/g, "").slice(0, 64) ||
408 "alchemy",
409 name,
410 jobTemplateData: toJobTemplateData(news.jobTemplateData),
411 kmsKeyArn: news.kmsKeyArn,
412 tags: desiredTags,
413 });
414 jt =
415 created.id !== undefined
416 ? yield* emrc
417 .describeJobTemplate({ id: created.id })
418 .pipe(Effect.map((r) => r.jobTemplate))
419 : yield* observeByName(name);
420 if (jt?.id === undefined) {
421 return yield* Effect.fail(
422 new emrc.ResourceNotFoundException({
423 message: `job template ${name} not visible after create`,
424 }),
425 );
426 }
427 const listed = yield* awaitListVisibility(jt.id);
428 if (listed?.id !== jt.id) {
429 return yield* Effect.fail(
430 new JobTemplateConsistencyError({
431 jobTemplateId: jt.id,
432 jobTemplateName: name,
433 operation: "create",
434 message: `job template ${name} (${jt.id}) was not visible in ListJobTemplates after 30 seconds`,
435 }),
436 );
437 }
438 }
439
440 // No tag sync: EMR containers' TagResource rejects job template
441 // ARNs (typed InvalidResourceArn) — tags exist at creation only,
442 // and diff() replaces the template on any tag change.
443
444 yield* session.note(jt.id!);
445 return toAttributes(jt);
446 }),
447
448 delete: Effect.fn(function* ({ output }) {
449 // deleteJobTemplate's typed union has no not-found tag — the API
450 // reports missing/already-deleted ids as validation errors, which
451 // makes the swallow below the idempotency guarantee.
452 yield* emrc
453 .deleteJobTemplate({ id: output.jobTemplateId })
454 .pipe(Effect.catchTag("ValidationException", () => Effect.void));
455 const [described, listed] = yield* awaitAbsent(output.jobTemplateId);
456 if (described !== undefined || listed !== undefined) {
457 return yield* Effect.fail(
458 new JobTemplateConsistencyError({
459 jobTemplateId: output.jobTemplateId,
460 jobTemplateName: output.jobTemplateName,
461 operation: "delete",
462 message: `job template ${output.jobTemplateName} (${output.jobTemplateId}) remained visible after delete for 30 seconds`,
463 }),
464 );
465 }
466 }),
467 });
468 }),
469 );