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 type { AutoscalingPolicyProps } from "./AutoscalingPolicy.ts";
18import {
19 collectPages,
20 LIST_LOCATIONS,
21 MAX_POLICY_ID_LENGTH,
22 defaultWorkerConfig,
23 defaultYarnConfig,
24 emptyOnMissing,
25 fingerprint,
26 hasAlchemyLabelMap,
27 normalizeLocation,
28 parseResourceName,
29 regionParent,
30 sameJson,
31 toPhysicalId,
32 userLabels,
33 waitUntilGone,
34} from "./internal.ts";
35
36export type RegionsAutoscalingPolicyProps = Omit<
37 AutoscalingPolicyProps,
38 "location"
39> & {
40 /**
41 * Region (`us-central1`, …). Immutable — changing it replaces the
42 * policy. `US-CENTRAL1` is accepted and normalized to `us-central1`.
43 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
44 */
45 region?: string;
46};
47
48export type RegionsAutoscalingPolicy = Resource<
49 "GCP.Dataproc.RegionsAutoscalingPolicy",
50 RegionsAutoscalingPolicyProps,
51 {
52 /** Full resource name `projects/{project}/regions/{region}/autoscalingPolicies/{policy}`. */
53 name: string;
54 /** Policy id (last path segment). */
55 policyId: string;
56 /** Project id. */
57 project: string;
58 /** Region id. */
59 region: string;
60 /** User labels (Alchemy ownership labels stripped). */
61 labels: Record<string, string>;
62 /** Cluster topology this policy applies to. */
63 clusterType: string | undefined;
64 /** Primary worker min instances. */
65 workerMinInstances: number | undefined;
66 /** Primary worker max instances. */
67 workerMaxInstances: number | undefined;
68 /** Cooldown period. */
69 cooldownPeriod: string | undefined;
70 },
71 never,
72 Providers
73>;
74
75/**
76 * A Dataproc autoscaling policy (regions API).
77 *
78 * Same resource as {@link AutoscalingPolicy} addressed via
79 * `projects/{project}/regions/{region}/autoscalingPolicies/{policy}`.
80 *
81 * ### Creating a Policy
82 * **Example:** Generated name
83 * ```typescript
84 * const policy = yield* GCP.Dataproc.RegionsAutoscalingPolicy("SparkScale", {});
85 * ```
86 *
87 * **Example:** Explicit bounds
88 * ```typescript
89 * const policy = yield* GCP.Dataproc.RegionsAutoscalingPolicy("SparkScale", {
90 * policyId: "spark-scale-reg",
91 * region: "us-central1",
92 * workerConfig: { minInstances: 2, maxInstances: 6 },
93 * labels: { env: "prod" },
94 * });
95 * ```
96 *
97 * @resource
98 * @category Dataproc
99 */
100export const RegionsAutoscalingPolicy = Resource<RegionsAutoscalingPolicy>(
101 "GCP.Dataproc.RegionsAutoscalingPolicy",
102);
103
104export class RegionsAutoscalingPolicyNotResolved extends Data.TaggedError(
105 "GCP.Dataproc.RegionsAutoscalingPolicyNotResolved",
106)<{
107 name: string;
108}> {}
109
110const resourceName = (project: string, region: string, policyId: string) =>
111 `${regionParent(project, region)}/autoscalingPolicies/${policyId}`;
112
113const toAttrs = (
114 policy: dataproc.AutoscalingPolicy,
115 project: string,
116 region: string,
117) => {
118 const name = policy.name ?? "";
119 const parsed = parseResourceName(name, "autoscalingPolicies", region);
120 return {
121 name,
122 policyId: policy.id ?? parsed.id,
123 project: parsed.project || project,
124 region: parsed.location || region,
125 labels: userLabels(policy.labels),
126 clusterType: policy.clusterType,
127 workerMinInstances: policy.workerConfig?.minInstances,
128 workerMaxInstances: policy.workerConfig?.maxInstances,
129 cooldownPeriod: policy.basicAlgorithm?.cooldownPeriod,
130 };
131};
132
133const desiredBody = (
134 news: RegionsAutoscalingPolicyProps,
135 policyId: string,
136 name: string,
137 desiredLabels: Record<string, string>,
138): dataproc.AutoscalingPolicy => ({
139 id: policyId,
140 name,
141 labels: desiredLabels,
142 clusterType: news.clusterType,
143 workerConfig: news.workerConfig ?? defaultWorkerConfig(),
144 secondaryWorkerConfig: news.secondaryWorkerConfig,
145 basicAlgorithm: {
146 cooldownPeriod: news.cooldownPeriod,
147 yarnConfig: news.yarnConfig ?? defaultYarnConfig(),
148 sparkStandaloneConfig: news.sparkStandaloneConfig,
149 },
150});
151
152const getByName = (name: string) =>
153 dataproc
154 .getProjectsRegionsAutoscalingPolicies({ name })
155 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
156
157const listRegion = (project: string, region: string) =>
158 emptyOnMissing(
159 collectPages(
160 dataproc.listProjectsRegionsAutoscalingPolicies.pages({
161 parent: regionParent(project, region),
162 pageSize: 1000,
163 }),
164 (page) => page.policies,
165 ).pipe(
166 Effect.map((items) =>
167 items
168 .filter((policy) => hasAlchemyLabelMap(policy.labels))
169 .map((policy) => toAttrs(policy, project, region)),
170 ),
171 ),
172 );
173
174const policyChanged = (
175 current: dataproc.AutoscalingPolicy,
176 desired: dataproc.AutoscalingPolicy,
177) =>
178 !sameJson(current.workerConfig, desired.workerConfig) ||
179 !sameJson(current.secondaryWorkerConfig, desired.secondaryWorkerConfig) ||
180 (current.clusterType ?? "") !== (desired.clusterType ?? "") ||
181 fingerprint(current.basicAlgorithm) !== fingerprint(desired.basicAlgorithm);
182
183export const RegionsAutoscalingPolicyProvider = () =>
184 Provider.succeed(RegionsAutoscalingPolicy, {
185 stables: ["name", "policyId", "project", "region"],
186
187 diff: Effect.fn(function* ({ news, olds, output }) {
188 if (!isResolved(news)) return undefined;
189 const env = yield* GcpEnvironment.current;
190 const previousId = olds?.policyId ?? output?.policyId;
191 const nextId = news.policyId ?? previousId;
192 const previousRegion = normalizeLocation(
193 olds?.region ?? output?.region,
194 env.region,
195 );
196 const nextRegion = normalizeLocation(
197 news.region ?? olds?.region ?? output?.region,
198 env.region,
199 );
200 if (
201 (previousId !== undefined &&
202 nextId !== undefined &&
203 previousId !== nextId) ||
204 (output !== undefined && previousRegion !== nextRegion)
205 ) {
206 return {
207 action: "replace" as const,
208 deleteFirst:
209 previousRegion === nextRegion &&
210 previousId !== undefined &&
211 nextId === previousId,
212 };
213 }
214 return undefined;
215 }),
216
217 read: Effect.fn(function* ({ id, olds, output }) {
218 const env = yield* GcpEnvironment.current;
219 const region = normalizeLocation(
220 olds?.region ?? output?.region,
221 env.region,
222 );
223 const policyId = yield* toPhysicalId(
224 id,
225 olds?.policyId,
226 output?.policyId,
227 MAX_POLICY_ID_LENGTH,
228 "policy",
229 );
230 const name = output?.name ?? resourceName(env.project, region, policyId);
231 const existing = yield* getByName(name);
232 if (existing === undefined) return undefined;
233 const attrs = toAttrs(existing, env.project, region);
234 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
235 ? attrs
236 : Unowned(attrs);
237 }),
238
239 list: () =>
240 Effect.gen(function* () {
241 const env = yield* GcpEnvironment.current;
242 const pages = yield* Effect.forEach(
243 [...new Set([env.region, ...LIST_LOCATIONS])],
244 (region) => listRegion(env.project, region),
245 { concurrency: 4 },
246 );
247 return pages.flat();
248 }),
249
250 reconcile: Effect.fn(function* ({ id, news, output }) {
251 const env = yield* GcpEnvironment.current;
252 const region = normalizeLocation(
253 news.region ?? output?.region,
254 env.region,
255 );
256 const policyId = yield* toPhysicalId(
257 id,
258 news.policyId,
259 output?.policyId,
260 MAX_POLICY_ID_LENGTH,
261 "policy",
262 );
263 const name = resourceName(env.project, region, policyId);
264 const desiredLabels = {
265 ...toLabels(news.labels),
266 ...(yield* createInternalLabels(id)),
267 };
268 const desired = desiredBody(news, policyId, name, desiredLabels);
269
270 let current = yield* getByName(name);
271
272 if (current === undefined) {
273 const created = yield* dataproc
274 .createProjectsRegionsAutoscalingPolicies({
275 parent: regionParent(env.project, region),
276 body: desired,
277 })
278 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
279 current = created ?? undefined;
280 }
281
282 if (current === undefined) {
283 return yield* new RegionsAutoscalingPolicyNotResolved({ name });
284 }
285
286 const observedLabels = tagRecord(current.labels);
287 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
288 const labelsChanged = upsert.length > 0 || removed.length > 0;
289
290 if (labelsChanged || policyChanged(current, desired)) {
291 current = yield* dataproc.updateProjectsRegionsAutoscalingPolicies({
292 name: current.name ?? name,
293 body: {
294 ...desired,
295 name: current.name ?? name,
296 },
297 });
298 }
299
300 return toAttrs(current, env.project, region);
301 }),
302
303 delete: Effect.fn(function* ({ output }) {
304 if (!output.name) return;
305 yield* dataproc
306 .deleteProjectsRegionsAutoscalingPolicies({ name: output.name })
307 .pipe(Effect.catchTag("NotFound", () => Effect.void));
308 yield* waitUntilGone(getByName(output.name), output.name);
309 }),
310 });