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_POLICY_ID_LENGTH,
21 defaultWorkerConfig,
22 defaultYarnConfig,
23 emptyOnMissing,
24 fingerprint,
25 hasAlchemyLabelMap,
26 locationParent,
27 normalizeLocation,
28 parseResourceName,
29 sameJson,
30 toPhysicalId,
31 userLabels,
32 waitUntilGone,
33} from "./internal.ts";
34
35export type AutoscalingPolicyProps = {
36 /**
37 * Policy id (the `{policy}` segment of
38 * `projects/{project}/locations/{location}/autoscalingPolicies/{policy}`).
39 * If omitted, a unique RFC1035 name is generated. 3-50 characters;
40 * letters, numbers, hyphens, underscores. Immutable — changing it
41 * replaces the policy.
42 */
43 policyId?: string;
44 /**
45 * Location (`us-central1`, …). Immutable — changing it replaces the
46 * policy. `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 * Cluster topology this policy applies to (`STANDARD`, `ZERO_SCALE`).
56 */
57 clusterType?: dataproc.AutoscalingPolicyClusterTypeEnum | (string & {});
58 /**
59 * Primary worker bounds. Defaults to min 2, max 3, weight 1.
60 */
61 workerConfig?: dataproc.InstanceGroupAutoscalingPolicyConfig;
62 /**
63 * Secondary worker bounds. Optional; omitted means no secondary workers.
64 */
65 secondaryWorkerConfig?: dataproc.InstanceGroupAutoscalingPolicyConfig;
66 /**
67 * Duration between scaling events (JSON Duration). Bounds: 2m-1d.
68 * @default "2m"
69 */
70 cooldownPeriod?: string;
71 /**
72 * YARN autoscaling configuration. Defaults to scale-up 0.5, no
73 * scale-down, 3600s graceful decommission.
74 */
75 yarnConfig?: dataproc.BasicYarnAutoscalingConfig;
76 /**
77 * Spark Standalone autoscaling configuration (Spark Standalone clusters).
78 */
79 sparkStandaloneConfig?: dataproc.SparkStandaloneAutoscalingConfig;
80};
81
82export type AutoscalingPolicy = Resource<
83 "GCP.Dataproc.AutoscalingPolicy",
84 AutoscalingPolicyProps,
85 {
86 /** Full resource name `projects/{project}/locations/{location}/autoscalingPolicies/{policy}`. */
87 name: string;
88 /** Policy id (last path segment). */
89 policyId: string;
90 /** Project id. */
91 project: string;
92 /** Location id. */
93 location: string;
94 /** User labels (Alchemy ownership labels stripped). */
95 labels: Record<string, string>;
96 /** Cluster topology this policy applies to. */
97 clusterType: string | undefined;
98 /** Primary worker min instances. */
99 workerMinInstances: number | undefined;
100 /** Primary worker max instances. */
101 workerMaxInstances: number | undefined;
102 /** Cooldown period. */
103 cooldownPeriod: string | undefined;
104 },
105 never,
106 Providers
107>;
108
109/**
110 * A Dataproc autoscaling policy (locations API).
111 *
112 * Changing `policyId` or `location` replaces the policy. Worker bounds,
113 * algorithm, and labels update in place via a full-resource replace.
114 *
115 * ### Creating a Policy
116 * **Example:** Generated name
117 * ```typescript
118 * const policy = yield* GCP.Dataproc.AutoscalingPolicy("SparkScale", {});
119 * ```
120 *
121 * **Example:** Explicit bounds
122 * ```typescript
123 * const policy = yield* GCP.Dataproc.AutoscalingPolicy("SparkScale", {
124 * policyId: "spark-scale",
125 * location: "us-central1",
126 * workerConfig: { minInstances: 2, maxInstances: 6 },
127 * labels: { env: "prod" },
128 * });
129 * ```
130 *
131 * @resource
132 * @category Dataproc
133 */
134export const AutoscalingPolicy = Resource<AutoscalingPolicy>(
135 "GCP.Dataproc.AutoscalingPolicy",
136);
137
138export class AutoscalingPolicyNotResolved extends Data.TaggedError(
139 "GCP.Dataproc.AutoscalingPolicyNotResolved",
140)<{
141 name: string;
142}> {}
143
144const resourceName = (project: string, location: string, policyId: string) =>
145 `${locationParent(project, location)}/autoscalingPolicies/${policyId}`;
146
147const toAttrs = (
148 policy: dataproc.AutoscalingPolicy,
149 project: string,
150 location: string,
151) => {
152 const name = policy.name ?? "";
153 const parsed = parseResourceName(name, "autoscalingPolicies", location);
154 return {
155 name,
156 policyId: policy.id ?? parsed.id,
157 project: parsed.project || project,
158 location: parsed.location || location,
159 labels: userLabels(policy.labels),
160 clusterType: policy.clusterType,
161 workerMinInstances: policy.workerConfig?.minInstances,
162 workerMaxInstances: policy.workerConfig?.maxInstances,
163 cooldownPeriod: policy.basicAlgorithm?.cooldownPeriod,
164 };
165};
166
167const desiredBody = (
168 news: AutoscalingPolicyProps,
169 policyId: string,
170 name: string,
171 desiredLabels: Record<string, string>,
172): dataproc.AutoscalingPolicy => ({
173 id: policyId,
174 name,
175 labels: desiredLabels,
176 clusterType: news.clusterType,
177 workerConfig: news.workerConfig ?? defaultWorkerConfig(),
178 secondaryWorkerConfig: news.secondaryWorkerConfig,
179 basicAlgorithm: {
180 cooldownPeriod: news.cooldownPeriod,
181 yarnConfig: news.yarnConfig ?? defaultYarnConfig(),
182 sparkStandaloneConfig: news.sparkStandaloneConfig,
183 },
184});
185
186const getByName = (name: string) =>
187 dataproc
188 .getProjectsLocationsAutoscalingPolicies({ name })
189 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
190
191const listLocation = (project: string, location: string) =>
192 emptyOnMissing(
193 collectPages(
194 dataproc.listProjectsLocationsAutoscalingPolicies.pages({
195 parent: locationParent(project, location),
196 pageSize: 1000,
197 }),
198 (page) => page.policies,
199 ).pipe(
200 Effect.map((items) =>
201 items
202 .filter((policy) => hasAlchemyLabelMap(policy.labels))
203 .map((policy) => toAttrs(policy, project, location)),
204 ),
205 ),
206 );
207
208const policyChanged = (
209 current: dataproc.AutoscalingPolicy,
210 desired: dataproc.AutoscalingPolicy,
211) =>
212 !sameJson(current.workerConfig, desired.workerConfig) ||
213 !sameJson(current.secondaryWorkerConfig, desired.secondaryWorkerConfig) ||
214 (current.clusterType ?? "") !== (desired.clusterType ?? "") ||
215 fingerprint(current.basicAlgorithm) !== fingerprint(desired.basicAlgorithm);
216
217export const AutoscalingPolicyProvider = () =>
218 Provider.succeed(AutoscalingPolicy, {
219 stables: ["name", "policyId", "project", "location"],
220
221 diff: Effect.fn(function* ({ news, olds, output }) {
222 if (!isResolved(news)) return undefined;
223 const env = yield* GcpEnvironment.current;
224 const previousId = olds?.policyId ?? output?.policyId;
225 const nextId = news.policyId ?? previousId;
226 const previousLocation = normalizeLocation(
227 olds?.location ?? output?.location,
228 env.region,
229 );
230 const nextLocation = normalizeLocation(
231 news.location ?? olds?.location ?? output?.location,
232 env.region,
233 );
234 if (
235 (previousId !== undefined &&
236 nextId !== undefined &&
237 previousId !== nextId) ||
238 (output !== undefined && previousLocation !== nextLocation)
239 ) {
240 return {
241 action: "replace" as const,
242 deleteFirst:
243 previousLocation === nextLocation &&
244 previousId !== undefined &&
245 nextId === previousId,
246 };
247 }
248 return undefined;
249 }),
250
251 read: Effect.fn(function* ({ id, olds, output }) {
252 const env = yield* GcpEnvironment.current;
253 const location = normalizeLocation(
254 olds?.location ?? output?.location,
255 env.region,
256 );
257 const policyId = yield* toPhysicalId(
258 id,
259 olds?.policyId,
260 output?.policyId,
261 MAX_POLICY_ID_LENGTH,
262 "policy",
263 );
264 const name =
265 output?.name ?? resourceName(env.project, location, policyId);
266 const existing = yield* getByName(name);
267 if (existing === undefined) return undefined;
268 const attrs = toAttrs(existing, env.project, location);
269 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
270 ? attrs
271 : Unowned(attrs);
272 }),
273
274 list: () =>
275 Effect.gen(function* () {
276 const env = yield* GcpEnvironment.current;
277 const pages = yield* Effect.forEach(
278 [...new Set([env.region, ...LIST_LOCATIONS])],
279 (location) => listLocation(env.project, location),
280 { concurrency: 4 },
281 );
282 return pages.flat();
283 }),
284
285 reconcile: Effect.fn(function* ({ id, news, output }) {
286 const env = yield* GcpEnvironment.current;
287 const location = normalizeLocation(
288 news.location ?? output?.location,
289 env.region,
290 );
291 const policyId = yield* toPhysicalId(
292 id,
293 news.policyId,
294 output?.policyId,
295 MAX_POLICY_ID_LENGTH,
296 "policy",
297 );
298 const name = resourceName(env.project, location, policyId);
299 const desiredLabels = {
300 ...toLabels(news.labels),
301 ...(yield* createInternalLabels(id)),
302 };
303 const desired = desiredBody(news, policyId, name, desiredLabels);
304
305 let current = yield* getByName(name);
306
307 if (current === undefined) {
308 const created = yield* dataproc
309 .createProjectsLocationsAutoscalingPolicies({
310 parent: locationParent(env.project, location),
311 body: desired,
312 })
313 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
314 current = created ?? undefined;
315 }
316
317 if (current === undefined) {
318 return yield* new AutoscalingPolicyNotResolved({ name });
319 }
320
321 const observedLabels = tagRecord(current.labels);
322 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
323 const labelsChanged = upsert.length > 0 || removed.length > 0;
324
325 if (labelsChanged || policyChanged(current, desired)) {
326 current = yield* dataproc.updateProjectsLocationsAutoscalingPolicies({
327 name: current.name ?? name,
328 body: {
329 ...desired,
330 name: current.name ?? name,
331 },
332 });
333 }
334
335 return toAttrs(current, env.project, location);
336 }),
337
338 delete: Effect.fn(function* ({ output }) {
339 if (!output.name) return;
340 yield* dataproc
341 .deleteProjectsLocationsAutoscalingPolicies({ name: output.name })
342 .pipe(Effect.catchTag("NotFound", () => Effect.void));
343 yield* waitUntilGone(getByName(output.name), output.name);
344 }),
345 });