1import * as dataproc from "@distilled.cloud/gcp/dataproc_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { tagRecord } from "../../Tags.ts";
12import { GcpEnvironment } from "../Environment.ts";
13import { OperationFailed } from "../Operation.ts";
14import {
15 createInternalLabels,
16 diffLabels,
17 hasAlchemyLabels,
18 stripInternalLabels,
19 toLabels,
20} from "../Labels.ts";
21import type { Providers } from "../Providers.ts";
22import { MAX_POLLS, waitForOperation } from "./internal.ts";
23
24const DEFAULT_CLUSTER_TYPE = "SINGLE_NODE";
25const DEFAULT_MACHINE_TYPE = "e2-standard-2";
26const DEFAULT_BOOT_DISK_GB = 30;
27const DEFAULT_BOOT_DISK_TYPE = "pd-standard";
28const DEFAULT_MASTER_INSTANCES = 1;
29const DEFAULT_STANDARD_WORKERS = 2;
30const MAX_NAME_LENGTH = 51;
31
32const LIST_REGIONS = [
33 "us-central1",
34 "us-east1",
35 "us-east4",
36 "us-west1",
37 "europe-west1",
38 "asia-east1",
39] as const;
40
41export type InitializationAction = {
42 /** Cloud Storage URI of the executable (`gs://...`). */
43 executableFile: string;
44 /**
45 * How long the executable may run (JSON Duration, e.g. `"10m"`).
46 * @default "10m"
47 */
48 executionTimeout?: string;
49};
50
51export type ClusterProps = {
52 /**
53 * Cluster id unique within the project. If omitted, a unique RFC1035
54 * name is generated from the stack, stage, and logical id. Must start
55 * with a lowercase letter, contain only lowercase letters, numbers, and
56 * hyphens, and not end with a hyphen (up to 51 characters). Immutable —
57 * changing it replaces the cluster.
58 */
59 clusterName?: string;
60 /**
61 * Dataproc region (`us-central1`, `us-east1`, …). Immutable — changing
62 * it replaces the cluster. `US-CENTRAL1` is accepted and normalized to
63 * `us-central1`.
64 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
65 */
66 region?: string;
67 /**
68 * User labels. Alchemy ownership labels are merged in automatically.
69 */
70 labels?: Record<string, string>;
71 /**
72 * Cluster topology. `SINGLE_NODE` is a 1-VM cluster (no workers).
73 * Immutable — changing it replaces the cluster.
74 * @default "SINGLE_NODE"
75 */
76 clusterType?: dataproc.ClusterConfigClusterTypeEnum | (string & {});
77 /**
78 * Dataproc image version (`2.2`, `2.2-debian12`, …). Prefix matches
79 * the server-resolved version. Immutable — changing it replaces the
80 * cluster.
81 */
82 imageVersion?: string;
83 /**
84 * Optional components to activate (`JUPYTER`, `ZEPPELIN`, `TRINO`, …).
85 * Immutable — changing them replaces the cluster.
86 */
87 optionalComponents?: dataproc.SoftwareConfigOptionalComponentsItemEnumList;
88 /**
89 * Daemon properties (`core:hadoop.tmp.dir`, `spark:spark.executor.memory`,
90 * …). Immutable — changing them replaces the cluster.
91 */
92 properties?: Record<string, string>;
93 /**
94 * Compute Engine zone (`us-central1-a`). Immutable. If omitted,
95 * Dataproc picks a zone in `region`.
96 */
97 zone?: string;
98 /**
99 * VPC network name or URI. Cannot be set with `subnetwork`. Immutable.
100 */
101 network?: string;
102 /**
103 * Subnetwork name or URI. Cannot be set with `network`. Immutable.
104 */
105 subnetwork?: string;
106 /**
107 * Restrict VMs to internal IPs. Immutable. Defaults to `false` so the
108 * cluster can reach Google APIs without Private Google Access.
109 * @default false
110 */
111 internalIpOnly?: boolean;
112 /**
113 * VM service account email. Immutable.
114 */
115 serviceAccount?: string;
116 /**
117 * Extra OAuth scopes for cluster VMs. Immutable.
118 */
119 serviceAccountScopes?: string[];
120 /**
121 * Network tags applied to every VM. Immutable.
122 */
123 tags?: string[];
124 /**
125 * Compute Engine metadata entries on every VM. Immutable.
126 */
127 metadata?: Record<string, string>;
128 /**
129 * Master instances. `1` for standard or single-node, `3` for HA.
130 * Immutable — changing it replaces the cluster.
131 * @default 1
132 */
133 masterNumInstances?: number;
134 /**
135 * Master machine type short name or URI.
136 * @default "e2-standard-2"
137 */
138 masterMachineType?: string;
139 /**
140 * Master boot disk size in GB.
141 * @default 30
142 */
143 masterBootDiskSizeGb?: number;
144 /**
145 * Master boot disk type (`pd-standard`, `pd-balanced`, `pd-ssd`).
146 * @default "pd-standard"
147 */
148 masterBootDiskType?: string;
149 /**
150 * Primary worker count. `0` for single-node. Resizes in place.
151 * @default 0
152 */
153 workerNumInstances?: number;
154 /**
155 * Primary worker machine type short name or URI. Immutable.
156 * @default "e2-standard-2"
157 */
158 workerMachineType?: string;
159 /**
160 * Primary worker boot disk size in GB. Immutable.
161 * @default 30
162 */
163 workerBootDiskSizeGb?: number;
164 /**
165 * Primary worker boot disk type. Immutable.
166 * @default "pd-standard"
167 */
168 workerBootDiskType?: string;
169 /**
170 * Secondary (preemptible or non-preemptible) worker count. Resizes in
171 * place.
172 */
173 secondaryWorkerNumInstances?: number;
174 /**
175 * Secondary worker machine type. Immutable.
176 */
177 secondaryWorkerMachineType?: string;
178 /**
179 * Secondary worker boot disk size in GB. Immutable.
180 */
181 secondaryWorkerBootDiskSizeGb?: number;
182 /**
183 * Secondary worker preemptibility (`PREEMPTIBLE`, `SPOT`,
184 * `NON_PREEMPTIBLE`). Immutable.
185 */
186 secondaryWorkerPreemptibility?:
187 | dataproc.InstanceGroupConfigPreemptibilityEnum
188 | (string & {});
189 /**
190 * Autoscaling policy resource name. Updated in place.
191 */
192 autoscalingPolicyUri?: string;
193 /**
194 * Idle TTL before the cluster is deleted (JSON Duration). Immutable.
195 */
196 idleDeleteTtl?: string;
197 /**
198 * Lifetime TTL before the cluster is deleted (JSON Duration). Immutable.
199 */
200 autoDeleteTtl?: string;
201 /**
202 * Staging bucket name (not a `gs://` URI). Immutable.
203 */
204 configBucket?: string;
205 /**
206 * Temp bucket name (not a `gs://` URI). Immutable.
207 */
208 tempBucket?: string;
209 /**
210 * Enable Component Gateway HTTP port access.
211 * @default false
212 */
213 enableHttpPortAccess?: boolean;
214 /**
215 * Customer-managed KMS key for PD and job-argument encryption.
216 * Immutable.
217 */
218 kmsKey?: string;
219 /**
220 * Scripts to run on each node after config completes. Immutable.
221 */
222 initializationActions?: InitializationAction[];
223};
224
225export type Cluster = Resource<
226 "GCP.Dataproc.Cluster",
227 ClusterProps,
228 {
229 /** Synthetic name `projects/{project}/regions/{region}/clusters/{cluster}`. */
230 name: string;
231 /** Cluster id. */
232 clusterName: string;
233 /** Project id. */
234 project: string;
235 /** Region id (`us-central1`, …). */
236 region: string;
237 /** Server-assigned cluster UUID. */
238 clusterUuid: string | undefined;
239 /** Server-reported state (`RUNNING`, `CREATING`, …). */
240 state: string | undefined;
241 /** Extra status text, if any. */
242 statusDetail: string | undefined;
243 /** User labels (Alchemy ownership labels stripped). */
244 labels: Record<string, string>;
245 /** Cluster topology (`SINGLE_NODE`, `STANDARD`, …). */
246 clusterType: string;
247 /** Resolved image version. */
248 imageVersion: string | undefined;
249 /** Provisioned zone. */
250 zone: string | undefined;
251 /** VPC network. */
252 network: string | undefined;
253 /** Subnetwork. */
254 subnetwork: string | undefined;
255 /** Master instance count. */
256 masterNumInstances: number | undefined;
257 /** Master machine type short name. */
258 masterMachineType: string | undefined;
259 /** Master boot disk size in GB. */
260 masterBootDiskSizeGb: number | undefined;
261 /** Master boot disk type. */
262 masterBootDiskType: string | undefined;
263 /** Primary worker count. */
264 workerNumInstances: number | undefined;
265 /** Primary worker machine type short name. */
266 workerMachineType: string | undefined;
267 /** Primary worker boot disk size in GB. */
268 workerBootDiskSizeGb: number | undefined;
269 /** Primary worker boot disk type. */
270 workerBootDiskType: string | undefined;
271 /** Whether VMs use internal IPs only. */
272 internalIpOnly: boolean | undefined;
273 /** VM service account email. */
274 serviceAccount: string | undefined;
275 /** Secondary worker count. */
276 secondaryWorkerNumInstances: number | undefined;
277 /** Autoscaling policy resource name. */
278 autoscalingPolicyUri: string | undefined;
279 /** Staging bucket name. */
280 configBucket: string | undefined;
281 /** Temp bucket name. */
282 tempBucket: string | undefined;
283 /** Component Gateway HTTP ports, when enabled. */
284 httpPorts: Record<string, string>;
285 },
286 never,
287 Providers
288>;
289
290/**
291 * A Dataproc cluster of Compute Engine VMs.
292 *
293 * Defaults to a single-node cluster (`e2-standard-2`, 30 GB boot disk) so
294 * a bare `Cluster("Spark", {})` stays cheap. Set `clusterType: "STANDARD"`
295 * and `workerNumInstances` for a multi-VM cluster.
296 *
297 * Changing `clusterName`, `region`, topology, image, machine types, disks,
298 * network, zone, or software config replaces the cluster. Labels, primary
299 * worker count, secondary worker count, and the autoscaling policy URI
300 * update in place.
301 *
302 * Provisioning typically takes several minutes.
303 *
304 * ### Creating a Cluster
305 * **Example:** Generated name, single-node
306 * ```typescript
307 * const cluster = yield* GCP.Dataproc.Cluster("Spark", {});
308 * ```
309 *
310 * **Example:** Explicit id, labels, and machine type
311 * ```typescript
312 * const cluster = yield* GCP.Dataproc.Cluster("Spark", {
313 * clusterName: "app-spark",
314 * region: "us-central1",
315 * clusterType: "SINGLE_NODE",
316 * masterMachineType: "e2-standard-2",
317 * masterBootDiskSizeGb: 30,
318 * labels: { env: "prod" },
319 * });
320 * ```
321 *
322 * ### Standard topology
323 * **Example:** One master and two workers
324 * ```typescript
325 * const cluster = yield* GCP.Dataproc.Cluster("Spark", {
326 * clusterType: "STANDARD",
327 * workerNumInstances: 2,
328 * workerMachineType: "e2-standard-2",
329 * workerBootDiskSizeGb: 30,
330 * });
331 * ```
332 *
333 * @resource
334 * @category Dataproc
335 */
336export const Cluster = Resource<Cluster>("GCP.Dataproc.Cluster");
337
338export class ClusterNotResolved extends Data.TaggedError(
339 "GCP.Dataproc.ClusterNotResolved",
340)<{
341 clusterName: string;
342 region: string;
343}> {}
344
345export class ClusterFailed extends Data.TaggedError(
346 "GCP.Dataproc.ClusterFailed",
347)<{
348 clusterName: string;
349 state: string | undefined;
350 detail: string | undefined;
351}> {}
352
353export class ClusterNotReady extends Data.TaggedError(
354 "GCP.Dataproc.ClusterNotReady",
355)<{
356 clusterName: string;
357 state: string | undefined;
358}> {}
359
360export class ClusterStillExists extends Data.TaggedError(
361 "GCP.Dataproc.ClusterStillExists",
362)<{
363 clusterName: string;
364}> {}
365
366const lastSegment = (value: string) => {
367 const trimmed = value.replace(/\/+$/, "");
368 const parts = trimmed.split("/");
369 return parts[parts.length - 1] || trimmed;
370};
371
372const normalizeRegion = (region: string | undefined, fallback: string) =>
373 lastSegment(region ?? fallback).toLowerCase();
374
375const linkKey = (value: string | undefined) =>
376 value === undefined || value === "" ? "" : lastSegment(value).toLowerCase();
377
378const normalizeClusterType = (
379 type: string | undefined,
380 workerNumInstances: number | undefined,
381) => {
382 const value = (type ?? "").toUpperCase();
383 if (
384 value !== "" &&
385 value !== "CLUSTER_TYPE_UNSPECIFIED" &&
386 value !== "UNSPECIFIED"
387 ) {
388 return value;
389 }
390 if (workerNumInstances === 0) return "SINGLE_NODE";
391 return DEFAULT_CLUSTER_TYPE;
392};
393
394const desiredClusterType = (news: ClusterProps) =>
395 normalizeClusterType(news.clusterType, news.workerNumInstances);
396
397const rfc1035 = (name: string): string => {
398 let next = name
399 .toLowerCase()
400 .replace(/[^a-z0-9-]/g, "-")
401 .replace(/-+/g, "-")
402 .replace(/^-+|-+$/g, "");
403 if (!/^[a-z]/.test(next)) next = `c${next}`;
404 next = next.slice(0, MAX_NAME_LENGTH).replace(/-+$/g, "");
405 if (next.length === 0) return "cluster";
406 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, MAX_NAME_LENGTH - 1)}0`;
407 return next.slice(0, MAX_NAME_LENGTH);
408};
409
410const resourceName = (project: string, region: string, clusterName: string) =>
411 `projects/${project}/regions/${region}/clusters/${clusterName}`;
412
413const parseName = (name: string, fallbackRegion: string) => {
414 const parts = name.split("/").filter((part) => part.length > 0);
415 const clustersAt = parts.lastIndexOf("clusters");
416 const regionsAt = parts.lastIndexOf("regions");
417 const projectsAt = parts.lastIndexOf("projects");
418 return {
419 project:
420 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
421 region:
422 regionsAt >= 0 && parts[regionsAt + 1]
423 ? parts[regionsAt + 1]!
424 : fallbackRegion,
425 clusterName:
426 clustersAt >= 0 && parts[clustersAt + 1]
427 ? parts[clustersAt + 1]!
428 : lastSegment(name),
429 };
430};
431
432const userLabels = (
433 labels: Record<string, string | undefined> | null | undefined,
434): Record<string, string> => stripInternalLabels(tagRecord(labels));
435
436const toId = (id: string, clusterName: string | undefined, existing?: string) =>
437 Effect.gen(function* () {
438 return (
439 clusterName ??
440 existing ??
441 rfc1035(
442 yield* createPhysicalName({
443 id,
444 maxLength: MAX_NAME_LENGTH,
445 lowercase: true,
446 }),
447 )
448 );
449 });
450
451const configsOf = (
452 configs: Record<string, string | undefined> | null | undefined,
453): Record<string, string> =>
454 Object.fromEntries(
455 Object.entries(configs ?? {}).filter(
456 (entry): entry is [string, string] => entry[1] !== undefined,
457 ),
458 );
459
460const sortedKey = (values: readonly string[] | undefined) =>
461 JSON.stringify(
462 [...(values ?? [])].map((value) => value.toUpperCase()).sort(),
463 );
464
465const propertiesKey = (
466 properties: Record<string, string | undefined> | null | undefined,
467) =>
468 JSON.stringify(
469 Object.fromEntries(
470 Object.entries(configsOf(properties)).sort(([a], [b]) =>
471 a.localeCompare(b),
472 ),
473 ),
474 );
475
476const metadataKey = (
477 metadata: Record<string, string | undefined> | null | undefined,
478) => propertiesKey(metadata);
479
480const initKey = (actions: InitializationAction[] | undefined) =>
481 JSON.stringify(
482 (actions ?? []).map((action) => ({
483 executableFile: action.executableFile,
484 executionTimeout: action.executionTimeout ?? "",
485 })),
486 );
487
488const imageMatches = (
489 desired: string | undefined,
490 observed: string | undefined,
491) => {
492 if (desired === undefined || desired === "") return true;
493 if (observed === undefined || observed === "") return false;
494 const next = desired.toLowerCase();
495 const current = observed.toLowerCase();
496 return (
497 current === next ||
498 current.startsWith(`${next}-`) ||
499 current.startsWith(`${next}.`)
500 );
501};
502
503const groupNum = (group: dataproc.InstanceGroupConfig | undefined) =>
504 group?.numInstances;
505
506const toAttrs = (
507 cluster: dataproc.Cluster,
508 project: string,
509 region: string,
510) => {
511 const clusterName = cluster.clusterName ?? "";
512 const config = cluster.config;
513 const master = config?.masterConfig;
514 const worker = config?.workerConfig;
515 const secondary = config?.secondaryWorkerConfig;
516 return {
517 name: resourceName(project, region, clusterName),
518 clusterName,
519 project: cluster.projectId || project,
520 region,
521 clusterUuid: cluster.clusterUuid,
522 state: cluster.status?.state,
523 statusDetail: cluster.status?.detail,
524 labels: userLabels(cluster.labels),
525 clusterType: normalizeClusterType(
526 config?.clusterType,
527 worker?.numInstances,
528 ),
529 imageVersion: config?.softwareConfig?.imageVersion,
530 zone: config?.gceClusterConfig?.zoneUri
531 ? lastSegment(config.gceClusterConfig.zoneUri)
532 : undefined,
533 network: config?.gceClusterConfig?.networkUri
534 ? lastSegment(config.gceClusterConfig.networkUri)
535 : undefined,
536 subnetwork: config?.gceClusterConfig?.subnetworkUri
537 ? lastSegment(config.gceClusterConfig.subnetworkUri)
538 : undefined,
539 masterNumInstances: master?.numInstances,
540 masterMachineType: master?.machineTypeUri
541 ? lastSegment(master.machineTypeUri)
542 : undefined,
543 masterBootDiskSizeGb: master?.diskConfig?.bootDiskSizeGb,
544 masterBootDiskType: master?.diskConfig?.bootDiskType,
545 workerNumInstances: worker?.numInstances,
546 workerMachineType: worker?.machineTypeUri
547 ? lastSegment(worker.machineTypeUri)
548 : undefined,
549 workerBootDiskSizeGb: worker?.diskConfig?.bootDiskSizeGb,
550 workerBootDiskType: worker?.diskConfig?.bootDiskType,
551 internalIpOnly: config?.gceClusterConfig?.internalIpOnly,
552 serviceAccount: config?.gceClusterConfig?.serviceAccount,
553 secondaryWorkerNumInstances: secondary?.numInstances,
554 autoscalingPolicyUri: config?.autoscalingConfig?.policyUri,
555 configBucket: config?.configBucket,
556 tempBucket: config?.tempBucket,
557 httpPorts: configsOf(config?.endpointConfig?.httpPorts),
558 };
559};
560
561const getById = (projectId: string, region: string, clusterName: string) =>
562 dataproc
563 .getProjectsRegionsClusters({ projectId, region, clusterName })
564 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
565
566const waitUntilExists = (
567 projectId: string,
568 region: string,
569 clusterName: string,
570) =>
571 getById(projectId, region, clusterName).pipe(
572 Effect.flatMap((cluster) =>
573 cluster
574 ? Effect.succeed(cluster)
575 : Effect.fail(new ClusterNotResolved({ clusterName, region })),
576 ),
577 Effect.retry({
578 while: (error) => error._tag === "GCP.Dataproc.ClusterNotResolved",
579 times: 8,
580 schedule: Schedule.spaced("1 second"),
581 }),
582 );
583
584const terminalError = (state: string | undefined) =>
585 state === "ERROR" || state === "ERROR_DUE_TO_UPDATE";
586
587const waitUntilRunning = (
588 projectId: string,
589 region: string,
590 clusterName: string,
591) =>
592 getById(projectId, region, clusterName).pipe(
593 Effect.filterOrFail(
594 (cluster): cluster is dataproc.Cluster => cluster !== undefined,
595 () => new ClusterNotResolved({ clusterName, region }),
596 ),
597 Effect.filterOrFail(
598 (cluster) => !terminalError(cluster.status?.state),
599 (cluster) =>
600 new ClusterFailed({
601 clusterName,
602 state: cluster.status?.state,
603 detail: cluster.status?.detail,
604 }),
605 ),
606 Effect.filterOrFail(
607 (cluster) => (cluster.status?.state ?? "") === "RUNNING",
608 (cluster) =>
609 new ClusterNotReady({
610 clusterName,
611 state: cluster.status?.state,
612 }),
613 ),
614 Effect.retry({
615 while: (error) =>
616 error._tag === "GCP.Dataproc.ClusterNotReady" ||
617 error._tag === "GCP.Dataproc.ClusterNotResolved",
618 times: MAX_POLLS,
619 schedule: Schedule.spaced("5 seconds"),
620 }),
621 );
622
623const waitUntilGone = (
624 projectId: string,
625 region: string,
626 clusterName: string,
627) =>
628 getById(projectId, region, clusterName).pipe(
629 Effect.flatMap((cluster) =>
630 cluster === undefined
631 ? Effect.void
632 : Effect.fail(new ClusterStillExists({ clusterName })),
633 ),
634 Effect.retry({
635 while: (error) => error._tag === "GCP.Dataproc.ClusterStillExists",
636 times: MAX_POLLS,
637 schedule: Schedule.spaced("5 seconds"),
638 }),
639 );
640
641const deleteCluster = (
642 projectId: string,
643 region: string,
644 clusterName: string,
645) =>
646 Effect.gen(function* () {
647 const operation = yield* dataproc
648 .deleteProjectsRegionsClusters({ projectId, region, clusterName })
649 .pipe(
650 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
651 Effect.retry({
652 while: (error) => error._tag === "Conflict",
653 times: 8,
654 schedule: Schedule.spaced("5 seconds"),
655 }),
656 );
657 if (operation !== undefined) {
658 yield* waitForOperation(operation, { notFoundOk: true });
659 }
660 yield* waitUntilGone(projectId, region, clusterName);
661 });
662
663/**
664 * A create operation that fails because the VPC subnet is transiently busy
665 * (`The resource '.../subnetworks/default' is not ready`, raised while other
666 * operations mutate the shared subnet). LRO errors carry only a code and a
667 * message, so the message is the only discriminator.
668 */
669const isSubnetNotReady = (error: { readonly _tag: string }) =>
670 error instanceof OperationFailed &&
671 error.message.includes("/subnetworks/") &&
672 error.message.includes("is not ready");
673
674/**
675 * Create the cluster and wait for the create operation. A create that fails
676 * on a busy subnet leaves the cluster in `ERROR`; it is deleted and the
677 * create retried.
678 */
679const createCluster = (
680 projectId: string,
681 region: string,
682 clusterName: string,
683 body: dataproc.Cluster,
684) =>
685 Effect.gen(function* () {
686 const existing = yield* getById(projectId, region, clusterName);
687 if (existing !== undefined && terminalError(existing.status?.state)) {
688 yield* deleteCluster(projectId, region, clusterName);
689 }
690 const created = yield* dataproc
691 .createProjectsRegionsClusters({ projectId, region, body })
692 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
693 if (created !== undefined) {
694 yield* waitForOperation(created);
695 }
696 }).pipe(
697 Effect.retry({
698 while: isSubnetNotReady,
699 times: 6,
700 schedule: Schedule.spaced("30 seconds"),
701 }),
702 );
703
704const instanceGroup = (options: {
705 numInstances?: number;
706 machineType?: string;
707 bootDiskSizeGb?: number;
708 bootDiskType?: string;
709 preemptibility?:
710 | dataproc.InstanceGroupConfigPreemptibilityEnum
711 | (string & {});
712}): dataproc.InstanceGroupConfig | undefined => {
713 const group: dataproc.InstanceGroupConfig = {};
714 if (options.numInstances !== undefined) {
715 group.numInstances = options.numInstances;
716 }
717 if (options.machineType !== undefined) {
718 group.machineTypeUri = options.machineType;
719 }
720 const disk: dataproc.DiskConfig = {};
721 if (options.bootDiskSizeGb !== undefined) {
722 disk.bootDiskSizeGb = options.bootDiskSizeGb;
723 }
724 if (options.bootDiskType !== undefined) {
725 disk.bootDiskType = options.bootDiskType;
726 }
727 if (Object.keys(disk).length > 0) {
728 group.diskConfig = disk;
729 }
730 if (options.preemptibility !== undefined) {
731 group.preemptibility = options.preemptibility;
732 }
733 return Object.keys(group).length > 0 ? group : undefined;
734};
735
736const toCreateBody = (
737 news: ClusterProps,
738 projectId: string,
739 clusterName: string,
740 desiredLabels: Record<string, string>,
741): dataproc.Cluster => {
742 const clusterType = desiredClusterType(news);
743 const singleNode = clusterType === "SINGLE_NODE";
744 const workerNum =
745 news.workerNumInstances ?? (singleNode ? 0 : DEFAULT_STANDARD_WORKERS);
746 const masterNum = news.masterNumInstances ?? DEFAULT_MASTER_INSTANCES;
747 const masterMachine = news.masterMachineType ?? DEFAULT_MACHINE_TYPE;
748 const workerMachine =
749 news.workerMachineType ?? news.masterMachineType ?? DEFAULT_MACHINE_TYPE;
750 const masterDiskGb = news.masterBootDiskSizeGb ?? DEFAULT_BOOT_DISK_GB;
751 const workerDiskGb =
752 news.workerBootDiskSizeGb ??
753 news.masterBootDiskSizeGb ??
754 DEFAULT_BOOT_DISK_GB;
755 const masterDiskType = news.masterBootDiskType ?? DEFAULT_BOOT_DISK_TYPE;
756 const workerDiskType =
757 news.workerBootDiskType ??
758 news.masterBootDiskType ??
759 DEFAULT_BOOT_DISK_TYPE;
760
761 const gce: dataproc.GceClusterConfig = {};
762 if (news.zone !== undefined) gce.zoneUri = news.zone;
763 if (news.network !== undefined) gce.networkUri = news.network;
764 if (news.subnetwork !== undefined) gce.subnetworkUri = news.subnetwork;
765 gce.internalIpOnly = news.internalIpOnly === true;
766 if (news.serviceAccount !== undefined)
767 gce.serviceAccount = news.serviceAccount;
768 if (news.serviceAccountScopes !== undefined) {
769 gce.serviceAccountScopes = news.serviceAccountScopes;
770 }
771 if (news.tags !== undefined) gce.tags = news.tags;
772 if (news.metadata !== undefined) gce.metadata = news.metadata;
773
774 const software: dataproc.SoftwareConfig = {};
775 if (news.imageVersion !== undefined)
776 software.imageVersion = news.imageVersion;
777 if (news.optionalComponents !== undefined) {
778 software.optionalComponents = news.optionalComponents;
779 }
780 if (news.properties !== undefined) software.properties = news.properties;
781
782 const lifecycle: dataproc.LifecycleConfig = {};
783 if (news.idleDeleteTtl !== undefined)
784 lifecycle.idleDeleteTtl = news.idleDeleteTtl;
785 if (news.autoDeleteTtl !== undefined)
786 lifecycle.autoDeleteTtl = news.autoDeleteTtl;
787
788 const config: dataproc.ClusterConfig = {
789 clusterType,
790 masterConfig: instanceGroup({
791 numInstances: masterNum,
792 machineType: masterMachine,
793 bootDiskSizeGb: masterDiskGb,
794 bootDiskType: masterDiskType,
795 }),
796 workerConfig: instanceGroup({
797 numInstances: workerNum,
798 machineType: singleNode ? undefined : workerMachine,
799 bootDiskSizeGb: singleNode ? undefined : workerDiskGb,
800 bootDiskType: singleNode ? undefined : workerDiskType,
801 }),
802 };
803
804 if (Object.keys(gce).length > 0) config.gceClusterConfig = gce;
805 if (Object.keys(software).length > 0) config.softwareConfig = software;
806 if (Object.keys(lifecycle).length > 0) config.lifecycleConfig = lifecycle;
807 if (news.autoscalingPolicyUri !== undefined) {
808 config.autoscalingConfig = { policyUri: news.autoscalingPolicyUri };
809 }
810 if (news.configBucket !== undefined) config.configBucket = news.configBucket;
811 if (news.tempBucket !== undefined) config.tempBucket = news.tempBucket;
812 if (news.enableHttpPortAccess === true) {
813 config.endpointConfig = { enableHttpPortAccess: true };
814 }
815 if (news.kmsKey !== undefined) {
816 config.encryptionConfig = { kmsKey: news.kmsKey };
817 }
818 if (news.initializationActions !== undefined) {
819 config.initializationActions = news.initializationActions;
820 }
821 if (
822 news.secondaryWorkerNumInstances !== undefined ||
823 news.secondaryWorkerMachineType !== undefined ||
824 news.secondaryWorkerBootDiskSizeGb !== undefined ||
825 news.secondaryWorkerPreemptibility !== undefined
826 ) {
827 config.secondaryWorkerConfig = instanceGroup({
828 numInstances: news.secondaryWorkerNumInstances,
829 machineType: news.secondaryWorkerMachineType,
830 bootDiskSizeGb: news.secondaryWorkerBootDiskSizeGb,
831 preemptibility: news.secondaryWorkerPreemptibility,
832 });
833 }
834
835 return {
836 projectId,
837 clusterName,
838 labels: desiredLabels,
839 config,
840 };
841};
842
843const listRegion = (projectId: string, region: string) =>
844 dataproc.listProjectsRegionsClusters
845 .pages({
846 projectId,
847 region,
848 pageSize: 1000,
849 })
850 .pipe(
851 Stream.flatMap((page) => Stream.fromIterable(page.clusters ?? [])),
852 Stream.filter((cluster) =>
853 Object.keys(cluster.labels ?? {}).some((key) =>
854 key.startsWith("alchemy-"),
855 ),
856 ),
857 Stream.map((cluster) => toAttrs(cluster, projectId, region)),
858 Stream.runCollect,
859 Effect.map((chunk) => Array.from(chunk)),
860 Effect.catchTag("NotFound", () => Effect.succeed([])),
861 );
862
863export const ClusterProvider = () =>
864 Provider.succeed(Cluster, {
865 stables: ["name", "clusterName", "project", "region", "clusterUuid"],
866
867 diff: Effect.fn(function* ({ news, olds, output }) {
868 if (!isResolved(news)) return undefined;
869 const env = yield* GcpEnvironment.current;
870
871 const previousName = olds?.clusterName ?? output?.clusterName;
872 const nextName = news.clusterName ?? previousName;
873 const previousRegion = normalizeRegion(
874 olds?.region ?? output?.region,
875 env.region,
876 );
877 const nextRegion = normalizeRegion(
878 news.region ?? output?.region,
879 env.region,
880 );
881 const previousType = normalizeClusterType(
882 olds?.clusterType ?? output?.clusterType,
883 olds?.workerNumInstances ?? output?.workerNumInstances,
884 );
885 const nextType = desiredClusterType({
886 clusterType: news.clusterType ?? previousType,
887 workerNumInstances:
888 news.workerNumInstances ?? output?.workerNumInstances,
889 });
890
891 const replace =
892 (previousName !== undefined &&
893 nextName !== undefined &&
894 nextName !== previousName) ||
895 previousRegion !== nextRegion ||
896 previousType !== nextType ||
897 (news.imageVersion !== undefined &&
898 !imageMatches(
899 news.imageVersion,
900 olds?.imageVersion ?? output?.imageVersion,
901 )) ||
902 (news.masterNumInstances !== undefined &&
903 news.masterNumInstances !==
904 (olds?.masterNumInstances ?? output?.masterNumInstances)) ||
905 (news.masterMachineType !== undefined &&
906 linkKey(news.masterMachineType) !==
907 linkKey(olds?.masterMachineType ?? output?.masterMachineType)) ||
908 (news.workerMachineType !== undefined &&
909 linkKey(news.workerMachineType) !==
910 linkKey(olds?.workerMachineType ?? output?.workerMachineType)) ||
911 (news.masterBootDiskSizeGb !== undefined &&
912 news.masterBootDiskSizeGb !==
913 (olds?.masterBootDiskSizeGb ?? output?.masterBootDiskSizeGb)) ||
914 (news.workerBootDiskSizeGb !== undefined &&
915 news.workerBootDiskSizeGb !==
916 (olds?.workerBootDiskSizeGb ?? output?.workerBootDiskSizeGb)) ||
917 (news.masterBootDiskType !== undefined &&
918 news.masterBootDiskType.toLowerCase() !==
919 (
920 olds?.masterBootDiskType ??
921 output?.masterBootDiskType ??
922 ""
923 ).toLowerCase()) ||
924 (news.workerBootDiskType !== undefined &&
925 news.workerBootDiskType.toLowerCase() !==
926 (
927 olds?.workerBootDiskType ??
928 output?.workerBootDiskType ??
929 ""
930 ).toLowerCase()) ||
931 (news.zone !== undefined &&
932 linkKey(news.zone) !== linkKey(olds?.zone ?? output?.zone)) ||
933 (news.network !== undefined &&
934 linkKey(news.network) !==
935 linkKey(olds?.network ?? output?.network)) ||
936 (news.subnetwork !== undefined &&
937 linkKey(news.subnetwork) !==
938 linkKey(olds?.subnetwork ?? output?.subnetwork)) ||
939 (news.internalIpOnly !== undefined &&
940 news.internalIpOnly !==
941 (olds?.internalIpOnly ?? output?.internalIpOnly)) ||
942 (news.serviceAccount !== undefined &&
943 news.serviceAccount !==
944 (olds?.serviceAccount ?? output?.serviceAccount)) ||
945 (news.serviceAccountScopes !== undefined &&
946 sortedKey(news.serviceAccountScopes) !==
947 sortedKey(olds?.serviceAccountScopes)) ||
948 (news.tags !== undefined &&
949 sortedKey(news.tags) !== sortedKey(olds?.tags)) ||
950 (news.metadata !== undefined &&
951 metadataKey(news.metadata) !== metadataKey(olds?.metadata)) ||
952 (news.optionalComponents !== undefined &&
953 sortedKey(news.optionalComponents) !==
954 sortedKey(olds?.optionalComponents)) ||
955 (news.properties !== undefined &&
956 propertiesKey(news.properties) !== propertiesKey(olds?.properties)) ||
957 (news.secondaryWorkerMachineType !== undefined &&
958 linkKey(news.secondaryWorkerMachineType) !==
959 linkKey(olds?.secondaryWorkerMachineType)) ||
960 (news.secondaryWorkerBootDiskSizeGb !== undefined &&
961 news.secondaryWorkerBootDiskSizeGb !==
962 (olds?.secondaryWorkerBootDiskSizeGb ?? undefined)) ||
963 (news.secondaryWorkerPreemptibility !== undefined &&
964 (news.secondaryWorkerPreemptibility ?? "").toUpperCase() !==
965 (olds?.secondaryWorkerPreemptibility ?? "").toUpperCase()) ||
966 (news.idleDeleteTtl !== undefined &&
967 news.idleDeleteTtl !== (olds?.idleDeleteTtl ?? undefined)) ||
968 (news.autoDeleteTtl !== undefined &&
969 news.autoDeleteTtl !== (olds?.autoDeleteTtl ?? undefined)) ||
970 (news.configBucket !== undefined &&
971 news.configBucket !== (olds?.configBucket ?? output?.configBucket)) ||
972 (news.tempBucket !== undefined &&
973 news.tempBucket !== (olds?.tempBucket ?? output?.tempBucket)) ||
974 (news.enableHttpPortAccess !== undefined &&
975 news.enableHttpPortAccess !==
976 (olds?.enableHttpPortAccess ?? undefined)) ||
977 (news.kmsKey !== undefined && news.kmsKey !== (olds?.kmsKey ?? "")) ||
978 (news.initializationActions !== undefined &&
979 initKey(news.initializationActions) !==
980 initKey(olds?.initializationActions));
981
982 if (!replace) return undefined;
983 return {
984 action: "replace" as const,
985 deleteFirst:
986 previousRegion === nextRegion &&
987 previousName !== undefined &&
988 nextName === previousName,
989 };
990 }),
991
992 read: Effect.fn(function* ({ id, olds, output }) {
993 const env = yield* GcpEnvironment.current;
994 const clusterName = yield* toId(
995 id,
996 olds?.clusterName,
997 output?.clusterName,
998 );
999 const region = normalizeRegion(
1000 olds?.region ?? output?.region,
1001 env.region,
1002 );
1003 const parsed = output?.name ? parseName(output.name, region) : undefined;
1004 const project = parsed?.project || env.project;
1005 const existing = yield* getById(project, region, clusterName);
1006 if (existing === undefined) return undefined;
1007 const attrs = toAttrs(existing, project, region);
1008 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
1009 ? attrs
1010 : Unowned(attrs);
1011 }),
1012
1013 list: () =>
1014 Effect.gen(function* () {
1015 const env = yield* GcpEnvironment.current;
1016 const pages = yield* Effect.forEach(
1017 [...new Set([env.region, ...LIST_REGIONS])],
1018 (region) => listRegion(env.project, region),
1019 { concurrency: 4 },
1020 );
1021 return pages.flat();
1022 }),
1023
1024 reconcile: Effect.fn(function* ({ id, news, output }) {
1025 const env = yield* GcpEnvironment.current;
1026 const clusterName = yield* toId(
1027 id,
1028 news.clusterName,
1029 output?.clusterName,
1030 );
1031 const region = normalizeRegion(news.region ?? output?.region, env.region);
1032 const parsed = output?.name ? parseName(output.name, region) : undefined;
1033 const projectId = parsed?.project || env.project;
1034 const desiredLabels = {
1035 ...toLabels(news.labels),
1036 ...(yield* createInternalLabels(id)),
1037 };
1038
1039 let current = yield* getById(
1040 projectId,
1041 region,
1042 output?.clusterName ?? clusterName,
1043 );
1044
1045 if (current?.status?.state === "DELETING") {
1046 yield* waitUntilGone(projectId, region, clusterName);
1047 current = undefined;
1048 }
1049
1050 if (current === undefined) {
1051 yield* createCluster(
1052 projectId,
1053 region,
1054 clusterName,
1055 toCreateBody(news, projectId, clusterName, desiredLabels),
1056 );
1057 current = yield* waitUntilExists(projectId, region, clusterName);
1058 }
1059
1060 if (current === undefined) {
1061 return yield* new ClusterNotResolved({ clusterName, region });
1062 }
1063
1064 const state = current.status?.state ?? "";
1065 if (terminalError(state)) {
1066 return yield* new ClusterFailed({
1067 clusterName,
1068 state,
1069 detail: current.status?.detail,
1070 });
1071 }
1072
1073 if (state === "STOPPED" || state === "STOPPING") {
1074 const started = yield* dataproc.startProjectsRegionsClusters({
1075 projectId,
1076 region,
1077 clusterName,
1078 });
1079 yield* waitForOperation(started);
1080 }
1081
1082 if ((current.status?.state ?? "") !== "RUNNING") {
1083 current = yield* waitUntilRunning(projectId, region, clusterName);
1084 }
1085
1086 const observedLabels = tagRecord(current.labels);
1087 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
1088 const labelsChanged = upsert.length > 0 || removed.length > 0;
1089
1090 const observedWorkers = groupNum(current.config?.workerConfig);
1091 const desiredWorkers = news.workerNumInstances;
1092 const workersChanged =
1093 desiredWorkers !== undefined && observedWorkers !== desiredWorkers;
1094
1095 const observedSecondary = groupNum(current.config?.secondaryWorkerConfig);
1096 const desiredSecondary = news.secondaryWorkerNumInstances;
1097 const secondaryChanged =
1098 desiredSecondary !== undefined &&
1099 observedSecondary !== desiredSecondary;
1100
1101 const observedPolicy = current.config?.autoscalingConfig?.policyUri ?? "";
1102 const desiredPolicy = news.autoscalingPolicyUri;
1103 const policyChanged =
1104 desiredPolicy !== undefined && observedPolicy !== desiredPolicy;
1105
1106 if (
1107 labelsChanged ||
1108 workersChanged ||
1109 secondaryChanged ||
1110 policyChanged
1111 ) {
1112 const updateMask = [
1113 labelsChanged ? "labels" : undefined,
1114 workersChanged ? "config.worker_config.num_instances" : undefined,
1115 secondaryChanged
1116 ? "config.secondary_worker_config.num_instances"
1117 : undefined,
1118 policyChanged ? "config.autoscaling_config.policy_uri" : undefined,
1119 ].filter((field): field is string => field !== undefined);
1120
1121 const patched = yield* dataproc
1122 .patchProjectsRegionsClusters({
1123 projectId,
1124 region,
1125 clusterName,
1126 updateMask: updateMask.join(","),
1127 body: {
1128 projectId,
1129 clusterName,
1130 labels: desiredLabels,
1131 config: {
1132 workerConfig:
1133 desiredWorkers !== undefined
1134 ? { numInstances: desiredWorkers }
1135 : undefined,
1136 secondaryWorkerConfig:
1137 desiredSecondary !== undefined
1138 ? { numInstances: desiredSecondary }
1139 : undefined,
1140 autoscalingConfig:
1141 desiredPolicy !== undefined
1142 ? { policyUri: desiredPolicy }
1143 : undefined,
1144 },
1145 },
1146 })
1147 .pipe(
1148 Effect.retry({
1149 while: (error) => error._tag === "Conflict",
1150 times: 8,
1151 schedule: Schedule.spaced("5 seconds"),
1152 }),
1153 );
1154 yield* waitForOperation(patched);
1155 current = yield* waitUntilRunning(projectId, region, clusterName);
1156 }
1157
1158 return toAttrs(current, projectId, region);
1159 }),
1160
1161 delete: Effect.fn(function* ({ output }) {
1162 yield* deleteCluster(output.project, output.region, output.clusterName);
1163 }),
1164 });