1import * as dataproc from "@distilled.cloud/gcp/dataproc_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { tagRecord } from "../../Tags.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import {
11 createInternalLabels,
12 diffLabels,
13 hasAlchemyLabels,
14 toLabels,
15} from "../Labels.ts";
16import type { Providers } from "../Providers.ts";
17import {
18 collectPages,
19 LIST_LOCATIONS,
20 MAX_MANAGED_CLUSTER_PREFIX_LENGTH,
21 MAX_POLICY_ID_LENGTH,
22 defaultWorkflowJobs,
23 defaultWorkflowPlacement,
24 emptyOnMissing,
25 fingerprint,
26 hasAlchemyLabelMap,
27 locationParent,
28 normalizeLocation,
29 parseResourceName,
30 rfc1035,
31 toPhysicalId,
32 userLabels,
33 waitUntilGone,
34} from "./internal.ts";
35
36export type WorkflowTemplateProps = {
37 /**
38 * Template id (the `{template}` segment of
39 * `projects/{project}/locations/{location}/workflowTemplates/{template}`).
40 * If omitted, a unique RFC1035 name is generated. Immutable — changing
41 * it replaces the template.
42 */
43 templateId?: string;
44 /**
45 * Location (`us-central1`, …). Immutable — changing it replaces the
46 * template. `US-CENTRAL1` is accepted and normalized to `us-central1`.
47 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
48 */
49 location?: string;
50 /**
51 * User labels. Alchemy ownership labels are merged in automatically.
52 */
53 labels?: Record<string, string>;
54 /**
55 * Directed acyclic graph of jobs. Defaults to a SparkPi example step.
56 */
57 jobs?: dataproc.OrderedJobList;
58 /**
59 * Target cluster (managed cluster or selector). Defaults to a 1-master
60 * 2-worker managed cluster template; creating the workflow does not
61 * provision the cluster until the template is instantiated.
62 */
63 placement?: dataproc.WorkflowTemplatePlacement;
64 /**
65 * DAG timeout (JSON Duration, 10m-24h).
66 */
67 dagTimeout?: string;
68 /**
69 * Cloud KMS key for encrypting job arguments.
70 */
71 kmsKey?: string;
72 /**
73 * Template parameters substituted at instantiate time.
74 */
75 parameters?: dataproc.TemplateParameterList;
76};
77
78export type WorkflowTemplate = Resource<
79 "GCP.Dataproc.WorkflowTemplate",
80 WorkflowTemplateProps,
81 {
82 /** Full resource name `projects/{project}/locations/{location}/workflowTemplates/{template}`. */
83 name: string;
84 /** Template id (last path segment). */
85 templateId: string;
86 /** Project id. */
87 project: string;
88 /** Location id. */
89 location: string;
90 /** User labels (Alchemy ownership labels stripped). */
91 labels: Record<string, string>;
92 /** Server version used for optimistic updates. */
93 version: number | undefined;
94 /** DAG timeout. */
95 dagTimeout: string | undefined;
96 /** RFC3339 creation timestamp. */
97 createTime: string | undefined;
98 /** RFC3339 last-update timestamp. */
99 updateTime: string | undefined;
100 },
101 never,
102 Providers
103>;
104
105/**
106 * A Dataproc workflow template (locations API).
107 *
108 * Stores a DAG of jobs and a cluster placement. Instantiating the template
109 * (out of band) is what actually creates a cluster and runs jobs.
110 *
111 * Changing `templateId` or `location` replaces the template. Jobs,
112 * placement, labels, and timeout update in place (optimistic version).
113 *
114 * ### Creating a Template
115 * **Example:** Generated name
116 * ```typescript
117 * const template = yield* GCP.Dataproc.WorkflowTemplate("Nightly", {});
118 * ```
119 *
120 * **Example:** Explicit jobs
121 * ```typescript
122 * const template = yield* GCP.Dataproc.WorkflowTemplate("Nightly", {
123 * templateId: "nightly-etl",
124 * location: "us-central1",
125 * labels: { env: "prod" },
126 * jobs: [{
127 * stepId: "spark-pi",
128 * sparkJob: {
129 * mainClass: "org.apache.spark.examples.SparkPi",
130 * jarFileUris: ["file:///usr/lib/spark/examples/jars/spark-examples.jar"],
131 * },
132 * }],
133 * });
134 * ```
135 *
136 * @resource
137 * @category Dataproc
138 */
139export const WorkflowTemplate = Resource<WorkflowTemplate>(
140 "GCP.Dataproc.WorkflowTemplate",
141);
142
143export class WorkflowTemplateNotResolved extends Data.TaggedError(
144 "GCP.Dataproc.WorkflowTemplateNotResolved",
145)<{
146 name: string;
147}> {}
148
149const resourceName = (project: string, location: string, templateId: string) =>
150 `${locationParent(project, location)}/workflowTemplates/${templateId}`;
151
152const toAttrs = (
153 template: dataproc.WorkflowTemplate,
154 project: string,
155 location: string,
156) => {
157 const name = template.name ?? "";
158 const parsed = parseResourceName(name, "workflowTemplates", location);
159 return {
160 name,
161 templateId: template.id ?? parsed.id,
162 project: parsed.project || project,
163 location: parsed.location || location,
164 labels: userLabels(template.labels),
165 version: template.version,
166 dagTimeout: template.dagTimeout,
167 createTime: template.createTime,
168 updateTime: template.updateTime,
169 };
170};
171
172const desiredBody = (
173 news: WorkflowTemplateProps,
174 templateId: string,
175 name: string,
176 desiredLabels: Record<string, string>,
177 version?: number,
178): dataproc.WorkflowTemplate => ({
179 id: templateId,
180 name,
181 labels: desiredLabels,
182 version,
183 jobs: news.jobs ?? defaultWorkflowJobs(),
184 placement:
185 news.placement ??
186 defaultWorkflowPlacement(
187 rfc1035(`${templateId}-c`, MAX_MANAGED_CLUSTER_PREFIX_LENGTH, "cluster"),
188 ),
189 dagTimeout: news.dagTimeout,
190 parameters: news.parameters,
191 encryptionConfig: news.kmsKey ? { kmsKey: news.kmsKey } : undefined,
192});
193
194const getByName = (name: string) =>
195 dataproc
196 .getProjectsLocationsWorkflowTemplates({ name })
197 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
198
199const listLocation = (project: string, location: string) =>
200 emptyOnMissing(
201 collectPages(
202 dataproc.listProjectsLocationsWorkflowTemplates.pages({
203 parent: locationParent(project, location),
204 pageSize: 1000,
205 }),
206 (page) => page.templates,
207 ).pipe(
208 Effect.map((items) =>
209 items
210 .filter((template) => hasAlchemyLabelMap(template.labels))
211 .map((template) => toAttrs(template, project, location)),
212 ),
213 ),
214 );
215
216const templateChanged = (
217 current: dataproc.WorkflowTemplate,
218 desired: dataproc.WorkflowTemplate,
219) =>
220 fingerprint(current.jobs) !== fingerprint(desired.jobs) ||
221 fingerprint(current.placement) !== fingerprint(desired.placement) ||
222 (current.dagTimeout ?? "") !== (desired.dagTimeout ?? "") ||
223 fingerprint(current.parameters) !== fingerprint(desired.parameters) ||
224 (current.encryptionConfig?.kmsKey ?? "") !==
225 (desired.encryptionConfig?.kmsKey ?? "");
226
227export const WorkflowTemplateProvider = () =>
228 Provider.succeed(WorkflowTemplate, {
229 stables: ["name", "templateId", "project", "location", "createTime"],
230
231 diff: Effect.fn(function* ({ news, olds, output }) {
232 if (!isResolved(news)) return undefined;
233 const env = yield* GcpEnvironment.current;
234 const previousId = olds?.templateId ?? output?.templateId;
235 const nextId = news.templateId ?? previousId;
236 const previousLocation = normalizeLocation(
237 olds?.location ?? output?.location,
238 env.region,
239 );
240 const nextLocation = normalizeLocation(
241 news.location ?? olds?.location ?? output?.location,
242 env.region,
243 );
244 if (
245 (previousId !== undefined &&
246 nextId !== undefined &&
247 previousId !== nextId) ||
248 (output !== undefined && previousLocation !== nextLocation)
249 ) {
250 return {
251 action: "replace" as const,
252 deleteFirst:
253 previousLocation === nextLocation &&
254 previousId !== undefined &&
255 nextId === previousId,
256 };
257 }
258 return undefined;
259 }),
260
261 read: Effect.fn(function* ({ id, olds, output }) {
262 const env = yield* GcpEnvironment.current;
263 const location = normalizeLocation(
264 olds?.location ?? output?.location,
265 env.region,
266 );
267 const templateId = yield* toPhysicalId(
268 id,
269 olds?.templateId,
270 output?.templateId,
271 MAX_POLICY_ID_LENGTH,
272 "template",
273 );
274 const name =
275 output?.name ?? resourceName(env.project, location, templateId);
276 const existing = yield* getByName(name);
277 if (existing === undefined) return undefined;
278 const attrs = toAttrs(existing, env.project, location);
279 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
280 ? attrs
281 : Unowned(attrs);
282 }),
283
284 list: () =>
285 Effect.gen(function* () {
286 const env = yield* GcpEnvironment.current;
287 const pages = yield* Effect.forEach(
288 [...new Set([env.region, ...LIST_LOCATIONS])],
289 (location) => listLocation(env.project, location),
290 { concurrency: 4 },
291 );
292 return pages.flat();
293 }),
294
295 reconcile: Effect.fn(function* ({ id, news, output }) {
296 const env = yield* GcpEnvironment.current;
297 const location = normalizeLocation(
298 news.location ?? output?.location,
299 env.region,
300 );
301 const templateId = yield* toPhysicalId(
302 id,
303 news.templateId,
304 output?.templateId,
305 MAX_POLICY_ID_LENGTH,
306 "template",
307 );
308 const name = resourceName(env.project, location, templateId);
309 const desiredLabels = {
310 ...toLabels(news.labels),
311 ...(yield* createInternalLabels(id)),
312 };
313
314 let current = yield* getByName(name);
315
316 if (current === undefined) {
317 const created = yield* dataproc
318 .createProjectsLocationsWorkflowTemplates({
319 parent: locationParent(env.project, location),
320 body: desiredBody(news, templateId, name, desiredLabels),
321 })
322 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
323 current = created ?? undefined;
324 }
325
326 if (current === undefined) {
327 return yield* new WorkflowTemplateNotResolved({ name });
328 }
329
330 const desired = desiredBody(
331 news,
332 templateId,
333 current.name ?? name,
334 desiredLabels,
335 current.version,
336 );
337 const observedLabels = tagRecord(current.labels);
338 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
339 const labelsChanged = upsert.length > 0 || removed.length > 0;
340
341 if (labelsChanged || templateChanged(current, desired)) {
342 current = yield* dataproc.updateProjectsLocationsWorkflowTemplates({
343 name: current.name ?? name,
344 body: desired,
345 });
346 }
347
348 return toAttrs(current, env.project, location);
349 }),
350
351 delete: Effect.fn(function* ({ output }) {
352 if (!output.name) return;
353 yield* dataproc
354 .deleteProjectsLocationsWorkflowTemplates({ name: output.name })
355 .pipe(Effect.catchTag("NotFound", () => Effect.void));
356 yield* waitUntilGone(getByName(output.name), output.name);
357 }),
358 });