1import * as dataplex from "@distilled.cloud/gcp/dataplex_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { tagRecord } from "../../Tags.ts";
10import { GcpEnvironment } from "../Environment.ts";
11import {
12 createInternalLabels,
13 diffLabels,
14 hasAlchemyLabels,
15 toLabels,
16} from "../Labels.ts";
17import type { Providers } from "../Providers.ts";
18import { waitForOperation } from "./operations.ts";
19import {
20 DataplexNotResolved,
21 expandParent,
22 fingerprint,
23 hasAlchemyLabelMap,
24 isPendingState,
25 listChildResources,
26 listLakes,
27 listTasks,
28 normalizeLocation,
29 parseResourceName,
30 replaceIfChanged,
31 toPhysicalRfc1035,
32 userLabels,
33} from "./shared.ts";
34
35const DEFAULT_TRIGGER = "ON_DEMAND";
36
37export type TaskTriggerSpec = {
38 /**
39 * Trigger type. Immutable — changing it replaces the task.
40 * @default "ON_DEMAND"
41 */
42 type?: dataplex.GoogleCloudDataplexV1TaskTriggerSpecTypeEnum | (string & {});
43 /** First run after this RFC3339 timestamp. */
44 startTime?: string;
45 /** Temporarily disable RECURRING execution. */
46 disabled?: boolean;
47 /** Retry attempts before aborting. */
48 maxRetries?: number;
49 /** Cron schedule. Required for `RECURRING`. */
50 schedule?: string;
51};
52
53export type TaskExecutionSpec = {
54 /** Arguments interpolated before passing to the driver. */
55 args?: Record<string, string>;
56 /**
57 * Service account used to execute the task. If omitted, the project
58 * Compute Engine default service account is used.
59 */
60 serviceAccount?: string;
61 /** Project in which jobs run. Defaults to the lake project. */
62 project?: string;
63 /** Maximum job duration (e.g. `"3600s"`). */
64 maxJobExecutionLifetime?: string;
65 /** Cloud KMS key used to encrypt job data. */
66 kmsKey?: string;
67};
68
69export type TaskInfrastructureSpec = {
70 /** Dataproc Serverless executor counts. */
71 batch?: {
72 executorsCount?: number;
73 maxExecutorsCount?: number;
74 };
75 /** Container image runtime. */
76 containerImage?: {
77 image?: string;
78 javaJars?: string[];
79 pythonPackages?: string[];
80 properties?: Record<string, string>;
81 };
82 /** VPC network used to run the job. */
83 vpcNetwork?: {
84 network?: string;
85 subNetwork?: string;
86 networkTags?: string[];
87 };
88};
89
90export type TaskSparkConfig = {
91 /** Cloud Storage URI of the jar containing the main class. */
92 mainJarFileUri?: string;
93 /** Driver main class. */
94 mainClass?: string;
95 /** Cloud Storage URI of the main Python file. */
96 pythonScriptFile?: string;
97 /** Cloud Storage URI of a SQL query file. */
98 sqlScriptFile?: string;
99 /** Inline SQL query text. */
100 sqlScript?: string;
101 /** Working-directory files. */
102 fileUris?: string[];
103 /** Archives extracted into the working directory. */
104 archiveUris?: string[];
105 /** Infrastructure used to run the Spark job. */
106 infrastructureSpec?: TaskInfrastructureSpec;
107};
108
109export type TaskNotebookConfig = {
110 /** Path to the input notebook. */
111 notebook: string;
112 /** Infrastructure used to run the notebook. */
113 infrastructureSpec?: TaskInfrastructureSpec;
114 /** Working-directory files. */
115 fileUris?: string[];
116 /** Archives extracted into the working directory. */
117 archiveUris?: string[];
118};
119
120export type LakesTaskProps = {
121 /**
122 * Parent lake. Full name
123 * `projects/{project}/locations/{location}/lakes/{lake}` or the lake id
124 * (combined with `location`). Immutable — changing it replaces the task.
125 */
126 lake: string;
127 /**
128 * Region used when `lake` is a bare id.
129 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
130 */
131 location?: string;
132 /**
133 * Task id. Immutable — changing it replaces the task.
134 */
135 taskId?: string;
136 /**
137 * Human-readable description.
138 */
139 description?: string;
140 /**
141 * User-friendly display name.
142 */
143 displayName?: string;
144 /**
145 * User labels. Alchemy ownership labels are merged in automatically.
146 */
147 labels?: Record<string, string>;
148 /**
149 * How often and when the task runs.
150 * @default { type: "ON_DEMAND" }
151 */
152 triggerSpec?: TaskTriggerSpec;
153 /**
154 * How the task is executed (service account, args, encryption).
155 */
156 executionSpec?: TaskExecutionSpec;
157 /**
158 * Spark workload. Mutually exclusive with `notebook`.
159 */
160 spark?: TaskSparkConfig;
161 /**
162 * Notebook workload. Mutually exclusive with `spark`.
163 */
164 notebook?: TaskNotebookConfig;
165};
166
167export type LakesTask = Resource<
168 "GCP.Dataplex.LakesTask",
169 LakesTaskProps,
170 {
171 /** Full resource name `.../lakes/{lake}/tasks/{task}`. */
172 name: string;
173 /** Task id. */
174 taskId: string;
175 /** Parent lake resource name. */
176 lake: string;
177 /** Project id. */
178 project: string;
179 /** Location id. */
180 location: string;
181 /** Description. */
182 description: string | undefined;
183 /** User-friendly display name. */
184 displayName: string | undefined;
185 /** User labels (Alchemy ownership labels stripped). */
186 labels: Record<string, string>;
187 /** Trigger type. */
188 triggerType: string | undefined;
189 /** Execution service account. */
190 serviceAccount: string | undefined;
191 /** Lifecycle state. */
192 state: string | undefined;
193 /** Server-assigned uid. */
194 uid: string | undefined;
195 /** RFC3339 creation timestamp. */
196 createTime: string | undefined;
197 /** RFC3339 last-update timestamp. */
198 updateTime: string | undefined;
199 },
200 never,
201 Providers
202>;
203
204/**
205 * A Dataplex lake task — a Spark or notebook job scheduled against a lake.
206 *
207 * Changing `lake`, `taskId`, `location`, or trigger type replaces the
208 * task. Description, labels, execution spec, and Spark/notebook config
209 * update in place.
210 *
211 * ### Creating a Task
212 * **Example:** On-demand Spark SQL
213 * ```typescript
214 * const task = yield* GCP.Dataplex.LakesTask("SelectOne", {
215 * lake: lake.name,
216 * executionSpec: { serviceAccount },
217 * spark: { sqlScript: "SELECT 1" },
218 * });
219 * ```
220 *
221 * **Example:** Recurring notebook
222 * ```typescript
223 * const task = yield* GCP.Dataplex.LakesTask("Nightly", {
224 * lake: lake.name,
225 * triggerSpec: { type: "RECURRING", schedule: "0 2 * * *" },
226 * executionSpec: { serviceAccount },
227 * notebook: { notebook: "gs://bucket/job.ipynb" },
228 * });
229 * ```
230 *
231 * @resource
232 * @category Dataplex
233 */
234export const LakesTask = Resource<LakesTask>("GCP.Dataplex.LakesTask");
235
236export class LakesTaskNotResolved extends Data.TaggedError(
237 "GCP.Dataplex.LakesTaskNotResolved",
238)<{
239 name: string;
240}> {}
241
242export class LakesTaskStillExists extends Data.TaggedError(
243 "GCP.Dataplex.LakesTaskStillExists",
244)<{
245 name: string;
246}> {}
247
248const lakeOf = (lake: string, project: string, location: string) =>
249 expandParent(lake, project, location, "lakes");
250
251const resourceName = (lake: string, taskId: string) =>
252 `${lake}/tasks/${taskId}`;
253
254const triggerBody = (
255 spec: TaskTriggerSpec | undefined,
256): dataplex.GoogleCloudDataplexV1TaskTriggerSpec => ({
257 type: (spec?.type ?? DEFAULT_TRIGGER).toUpperCase(),
258 startTime: spec?.startTime,
259 disabled: spec?.disabled,
260 maxRetries: spec?.maxRetries,
261 schedule: spec?.schedule,
262});
263
264const executionBody = (
265 spec: TaskExecutionSpec | undefined,
266): dataplex.GoogleCloudDataplexV1TaskExecutionSpec => ({
267 args: spec?.args,
268 serviceAccount: spec?.serviceAccount,
269 project: spec?.project,
270 maxJobExecutionLifetime: spec?.maxJobExecutionLifetime,
271 kmsKey: spec?.kmsKey,
272});
273
274const toAttrs = (task: dataplex.GoogleCloudDataplexV1Task, project: string) => {
275 const name = task.name ?? "";
276 const parsed = parseResourceName(name, "tasks");
277 return {
278 name,
279 taskId: parsed.id,
280 lake: parsed.parent,
281 project: parsed.project || project,
282 location: parsed.location,
283 description: task.description,
284 displayName: task.displayName,
285 labels: userLabels(task.labels),
286 triggerType: task.triggerSpec?.type,
287 serviceAccount: task.executionSpec?.serviceAccount,
288 state: task.state,
289 uid: task.uid,
290 createTime: task.createTime,
291 updateTime: task.updateTime,
292 };
293};
294
295const getByName = (name: string) =>
296 name.length === 0 || name.includes("//")
297 ? Effect.succeed(undefined)
298 : dataplex
299 .getProjectsLocationsLakesTasks({ name })
300 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
301
302const waitUntilReady = (name: string) =>
303 getByName(name).pipe(
304 Effect.filterOrFail(
305 (task): task is dataplex.GoogleCloudDataplexV1Task => task !== undefined,
306 () => new LakesTaskNotResolved({ name }),
307 ),
308 Effect.filterOrFail(
309 (task) => !isPendingState(task.state),
310 () => new DataplexNotResolved({ name }),
311 ),
312 Effect.retry({
313 while: (error) =>
314 error._tag === "GCP.Dataplex.LakesTaskNotResolved" ||
315 error._tag === "GCP.Dataplex.NotResolved" ||
316 error._tag === "TooManyRequests",
317 times: 10,
318 schedule: Schedule.spaced("10 seconds"),
319 }),
320 );
321
322const waitUntilGone = (name: string) =>
323 getByName(name).pipe(
324 Effect.flatMap((task) =>
325 task === undefined
326 ? Effect.void
327 : Effect.fail(new LakesTaskStillExists({ name })),
328 ),
329 Effect.retry({
330 while: (error) => error._tag === "GCP.Dataplex.LakesTaskStillExists",
331 times: 10,
332 schedule: Schedule.spaced("3 seconds"),
333 }),
334 );
335
336export const LakesTaskProvider = () =>
337 Provider.succeed(LakesTask, {
338 stables: [
339 "name",
340 "taskId",
341 "lake",
342 "project",
343 "location",
344 "uid",
345 "createTime",
346 ],
347
348 diff: Effect.fn(function* ({ news, olds, output }) {
349 if (!isResolved(news)) return undefined;
350 const env = yield* GcpEnvironment.current;
351 const previousId = olds?.taskId ?? output?.taskId;
352 const nextId = news.taskId ?? previousId;
353 const previousLake = olds?.lake ?? output?.lake;
354 const nextLake = news.lake ?? previousLake;
355 const previousLocation = normalizeLocation(
356 olds?.location ?? output?.location,
357 env.region,
358 );
359 const nextLocation = normalizeLocation(
360 news.location ?? olds?.location ?? output?.location,
361 env.region,
362 );
363 const previousTrigger = (
364 olds?.triggerSpec?.type ??
365 output?.triggerType ??
366 DEFAULT_TRIGGER
367 ).toUpperCase();
368 const nextTrigger = (
369 news.triggerSpec?.type ?? previousTrigger
370 ).toUpperCase();
371 if (
372 replaceIfChanged(previousId, nextId) ||
373 replaceIfChanged(previousLake, nextLake) ||
374 (output !== undefined && previousLocation !== nextLocation) ||
375 previousTrigger !== nextTrigger
376 ) {
377 return {
378 action: "replace" as const,
379 deleteFirst:
380 previousLake === nextLake &&
381 previousLocation === nextLocation &&
382 previousId !== undefined &&
383 nextId === previousId,
384 };
385 }
386 return undefined;
387 }),
388
389 read: Effect.fn(function* ({ id, olds, output }) {
390 const env = yield* GcpEnvironment.current;
391 const location = normalizeLocation(
392 olds?.location ?? output?.location,
393 env.region,
394 );
395 const lake = lakeOf(
396 olds?.lake ?? output?.lake ?? "",
397 env.project,
398 location,
399 );
400 const taskId = yield* toPhysicalRfc1035(id, olds?.taskId, output?.taskId);
401 const name = output?.name ?? resourceName(lake, taskId);
402 const existing = yield* getByName(name);
403 if (existing === undefined) return undefined;
404 const attrs = toAttrs(existing, env.project);
405 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
406 ? attrs
407 : Unowned(attrs);
408 }),
409
410 list: () =>
411 Effect.gen(function* () {
412 const env = yield* GcpEnvironment.current;
413 const lakes = yield* listLakes(env.project, env.region);
414 const tasks = yield* listChildResources(lakes, listTasks);
415 return tasks
416 .filter((task) => hasAlchemyLabelMap(task.labels))
417 .map((task) => toAttrs(task, env.project));
418 }),
419
420 reconcile: Effect.fn(function* ({ id, news, output }) {
421 const env = yield* GcpEnvironment.current;
422 const location = normalizeLocation(
423 news.location ?? output?.location,
424 env.region,
425 );
426 const lake = lakeOf(news.lake, env.project, location);
427 const taskId = yield* toPhysicalRfc1035(id, news.taskId, output?.taskId);
428 const name = output?.name ?? resourceName(lake, taskId);
429 const desiredLabels = {
430 ...toLabels(news.labels),
431 ...(yield* createInternalLabels(id)),
432 };
433 const triggerSpec = triggerBody(news.triggerSpec);
434 const executionSpec = executionBody(news.executionSpec);
435
436 let current = yield* getByName(name);
437
438 if (current === undefined) {
439 const created = yield* dataplex
440 .createProjectsLocationsLakesTasks({
441 parent: lake,
442 taskId,
443 body: {
444 description: news.description,
445 displayName: news.displayName,
446 labels: desiredLabels,
447 triggerSpec,
448 executionSpec,
449 spark: news.spark,
450 notebook: news.notebook,
451 },
452 })
453 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
454 if (created !== undefined) {
455 yield* waitForOperation(created);
456 }
457 current = yield* waitUntilReady(name);
458 }
459
460 if (current === undefined) {
461 return yield* new LakesTaskNotResolved({ name });
462 }
463
464 const observedLabels = tagRecord(current.labels);
465 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
466 const labelsChanged = upsert.length > 0 || removed.length > 0;
467 const descriptionChanged =
468 (current.description ?? "") !== (news.description ?? "");
469 const displayNameChanged =
470 (current.displayName ?? "") !== (news.displayName ?? "");
471 const triggerChanged =
472 fingerprint(current.triggerSpec) !== fingerprint(triggerSpec);
473 const executionChanged =
474 fingerprint(current.executionSpec) !== fingerprint(executionSpec);
475 const sparkChanged =
476 fingerprint(current.spark) !== fingerprint(news.spark);
477 const notebookChanged =
478 fingerprint(current.notebook) !== fingerprint(news.notebook);
479
480 if (
481 labelsChanged ||
482 descriptionChanged ||
483 displayNameChanged ||
484 triggerChanged ||
485 executionChanged ||
486 sparkChanged ||
487 notebookChanged
488 ) {
489 const updateMask = [
490 labelsChanged ? "labels" : undefined,
491 descriptionChanged ? "description" : undefined,
492 displayNameChanged ? "display_name" : undefined,
493 triggerChanged ? "trigger_spec" : undefined,
494 executionChanged ? "execution_spec" : undefined,
495 sparkChanged ? "spark" : undefined,
496 notebookChanged ? "notebook" : undefined,
497 ].filter((field): field is string => field !== undefined);
498 const operation = yield* dataplex.patchProjectsLocationsLakesTasks({
499 name: current.name ?? name,
500 updateMask: updateMask.join(","),
501 body: {
502 description: news.description,
503 displayName: news.displayName,
504 labels: desiredLabels,
505 triggerSpec,
506 executionSpec,
507 spark: news.spark,
508 notebook: news.notebook,
509 },
510 });
511 yield* waitForOperation(operation);
512 current = yield* waitUntilReady(current.name ?? name);
513 }
514
515 return toAttrs(current, env.project);
516 }),
517
518 delete: Effect.fn(function* ({ output }) {
519 if (!output.name || output.name.includes("//")) return;
520 const operation = yield* dataplex
521 .deleteProjectsLocationsLakesTasks({ name: output.name })
522 .pipe(
523 Effect.retry({
524 while: (error) => error._tag === "Conflict",
525 times: 8,
526 schedule: Schedule.spaced("2 seconds"),
527 }),
528 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
529 Effect.catchTag("BadRequest", () => Effect.succeed(undefined)),
530 );
531 if (operation !== undefined) {
532 yield* waitForOperation(operation, { notFoundOk: true });
533 }
534 yield* waitUntilGone(output.name);
535 }),
536 });