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 { 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 { createInternalLabels, hasAlchemyLabels, toLabels } from "../Labels.ts";
12import type { Providers } from "../Providers.ts";
13import {
14 collectPages,
15 LIST_LOCATIONS,
16 MAX_POLLS,
17 MAX_WORKLOAD_ID_LENGTH,
18 emptyOnMissing,
19 hasAlchemyLabelMap,
20 locationParent,
21 normalizeLocation,
22 parseResourceName,
23 toPhysicalId,
24 userLabels,
25 waitForOperation,
26 waitUntilExists,
27 waitUntilGone,
28} from "./internal.ts";
29
30const READY = new Set(["ACTIVE", "TERMINATING", "TERMINATED", "FAILED"]);
31
32export type SessionProps = {
33 /**
34 * Session id (the `{session}` segment of
35 * `projects/{project}/locations/{location}/sessions/{session}`). If
36 * omitted, a unique RFC1035 name is generated. Immutable — changing it
37 * replaces the session.
38 */
39 sessionId?: string;
40 /**
41 * Location (`us-central1`, …). Immutable — changing it replaces the
42 * session. `US-CENTRAL1` is accepted and normalized to `us-central1`.
43 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
44 */
45 location?: string;
46 /**
47 * User labels. Alchemy ownership labels are merged in automatically.
48 */
49 labels?: Record<string, string>;
50 /**
51 * Existing session template resource name. If omitted, `jupyterSession`
52 * (default PYTHON) is used.
53 */
54 sessionTemplate?: string;
55 /**
56 * Jupyter kernel config.
57 */
58 jupyterSession?: dataproc.JupyterConfig;
59 /**
60 * Spark Connect session config.
61 */
62 sparkConnectSession?: dataproc.SparkConnectConfig;
63 /**
64 * Runtime configuration (version, properties, container image).
65 */
66 runtimeConfig?: dataproc.RuntimeConfig;
67 /**
68 * Environment configuration (execution and peripherals).
69 */
70 environmentConfig?: dataproc.EnvironmentConfig;
71 /**
72 * Email of the user who owns the session.
73 */
74 user?: string;
75};
76
77export type Session = Resource<
78 "GCP.Dataproc.Session",
79 SessionProps,
80 {
81 /** Full resource name `projects/{project}/locations/{location}/sessions/{session}`. */
82 name: string;
83 /** Session id (last path segment). */
84 sessionId: string;
85 /** Project id. */
86 project: string;
87 /** Location id. */
88 location: string;
89 /** User labels (Alchemy ownership labels stripped). */
90 labels: Record<string, string>;
91 /** Server-reported state (`ACTIVE`, `CREATING`, …). */
92 state: string | undefined;
93 /** Extra status text, if any. */
94 stateMessage: string | undefined;
95 /** Session template used, if any. */
96 sessionTemplate: string | undefined;
97 /** Server-assigned uuid. */
98 uuid: string | undefined;
99 /** Creator email. */
100 creator: string | undefined;
101 /** RFC3339 creation timestamp. */
102 createTime: string | undefined;
103 },
104 never,
105 Providers
106>;
107
108/**
109 * A Dataproc serverless interactive session.
110 *
111 * Sessions are immutable after create. Changing identity or config
112 * replaces the session. Provisioning typically takes several minutes.
113 *
114 * ### Creating a Session
115 * **Example:** Jupyter session
116 * ```typescript
117 * const session = yield* GCP.Dataproc.Session("Notebook", {
118 * jupyterSession: { kernel: "PYTHON" },
119 * environmentConfig: { executionConfig: { idleTtl: "600s", ttl: "3600s" } },
120 * });
121 * ```
122 *
123 * **Example:** From a session template
124 * ```typescript
125 * const session = yield* GCP.Dataproc.Session("Notebook", {
126 * sessionTemplate: template.name,
127 * });
128 * ```
129 *
130 * @resource
131 * @category Dataproc
132 */
133export const Session = Resource<Session>("GCP.Dataproc.Session");
134
135export class SessionNotResolved extends Data.TaggedError(
136 "GCP.Dataproc.SessionNotResolved",
137)<{
138 name: string;
139}> {}
140
141export class SessionFailed extends Data.TaggedError(
142 "GCP.Dataproc.SessionFailed",
143)<{
144 name: string;
145 state: string | undefined;
146 detail: string | undefined;
147}> {}
148
149export class SessionNotReady extends Data.TaggedError(
150 "GCP.Dataproc.SessionNotReady",
151)<{
152 name: string;
153 state: string | undefined;
154}> {}
155
156const resourceName = (project: string, location: string, sessionId: string) =>
157 `${locationParent(project, location)}/sessions/${sessionId}`;
158
159const defaultJupyter = (
160 news: SessionProps,
161): dataproc.JupyterConfig | undefined => {
162 if (news.sessionTemplate !== undefined) return news.jupyterSession;
163 if (news.sparkConnectSession !== undefined) return undefined;
164 return news.jupyterSession ?? { kernel: "PYTHON" };
165};
166
167const toAttrs = (
168 session: dataproc.Session,
169 project: string,
170 location: string,
171) => {
172 const name = session.name ?? "";
173 const parsed = parseResourceName(name, "sessions", location);
174 return {
175 name,
176 sessionId: parsed.id,
177 project: parsed.project || project,
178 location: parsed.location || location,
179 labels: userLabels(session.labels),
180 state: session.state,
181 stateMessage: session.stateMessage,
182 sessionTemplate: session.sessionTemplate,
183 uuid: session.uuid,
184 creator: session.creator,
185 createTime: session.createTime,
186 };
187};
188
189// Dataproc rejects a session create whose body omits `name` with a bare
190// `INVALID_ARGUMENT`.
191const desiredBody = (
192 news: SessionProps,
193 name: string,
194 desiredLabels: Record<string, string>,
195): dataproc.Session => ({
196 name,
197 labels: desiredLabels,
198 sessionTemplate: news.sessionTemplate,
199 jupyterSession: defaultJupyter(news),
200 sparkConnectSession: news.sparkConnectSession,
201 runtimeConfig: news.runtimeConfig,
202 environmentConfig: news.environmentConfig ?? {
203 executionConfig: { idleTtl: "600s", ttl: "3600s" },
204 },
205 user: news.user,
206});
207
208const getByName = (name: string) =>
209 dataproc
210 .getProjectsLocationsSessions({ name })
211 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
212
213const waitUntilReady = (name: string) =>
214 getByName(name).pipe(
215 Effect.flatMap((session) =>
216 session
217 ? Effect.succeed(session)
218 : Effect.fail(new SessionNotResolved({ name })),
219 ),
220 Effect.filterOrFail(
221 (session) => session.state !== "FAILED",
222 (session) =>
223 new SessionFailed({
224 name,
225 state: session.state,
226 detail: session.stateMessage,
227 }),
228 ),
229 Effect.filterOrFail(
230 (session) => READY.has(session.state ?? ""),
231 (session) =>
232 new SessionNotReady({
233 name,
234 state: session.state,
235 }),
236 ),
237 Effect.retry({
238 while: (error) =>
239 error._tag === "GCP.Dataproc.SessionNotReady" ||
240 error._tag === "GCP.Dataproc.SessionNotResolved",
241 times: MAX_POLLS,
242 schedule: Schedule.spaced("5 seconds"),
243 }),
244 );
245
246const listLocation = (project: string, location: string) =>
247 emptyOnMissing(
248 collectPages(
249 dataproc.listProjectsLocationsSessions.pages({
250 parent: locationParent(project, location),
251 pageSize: 1000,
252 }),
253 (page) => page.sessions,
254 ).pipe(
255 Effect.map((items) =>
256 items
257 .filter((session) => hasAlchemyLabelMap(session.labels))
258 .map((session) => toAttrs(session, project, location)),
259 ),
260 ),
261 );
262
263export const SessionProvider = () =>
264 Provider.succeed(Session, {
265 stables: ["name", "sessionId", "project", "location", "uuid", "createTime"],
266
267 diff: Effect.fn(function* ({ news, olds, output }) {
268 if (!isResolved(news)) return undefined;
269 const env = yield* GcpEnvironment.current;
270 const previousId = olds?.sessionId ?? output?.sessionId;
271 const nextId = news.sessionId ?? previousId;
272 const previousLocation = normalizeLocation(
273 olds?.location ?? output?.location,
274 env.region,
275 );
276 const nextLocation = normalizeLocation(
277 news.location ?? olds?.location ?? output?.location,
278 env.region,
279 );
280 if (
281 (previousId !== undefined &&
282 nextId !== undefined &&
283 previousId !== nextId) ||
284 (output !== undefined && previousLocation !== nextLocation)
285 ) {
286 return {
287 action: "replace" as const,
288 deleteFirst:
289 previousLocation === nextLocation &&
290 previousId !== undefined &&
291 nextId === previousId,
292 };
293 }
294 return undefined;
295 }),
296
297 read: Effect.fn(function* ({ id, olds, output }) {
298 const env = yield* GcpEnvironment.current;
299 const location = normalizeLocation(
300 olds?.location ?? output?.location,
301 env.region,
302 );
303 const sessionId = yield* toPhysicalId(
304 id,
305 olds?.sessionId,
306 output?.sessionId,
307 MAX_WORKLOAD_ID_LENGTH,
308 "session",
309 );
310 const name =
311 output?.name ?? resourceName(env.project, location, sessionId);
312 const existing = yield* getByName(name);
313 if (existing === undefined) return undefined;
314 const attrs = toAttrs(existing, env.project, location);
315 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
316 ? attrs
317 : Unowned(attrs);
318 }),
319
320 list: () =>
321 Effect.gen(function* () {
322 const env = yield* GcpEnvironment.current;
323 const pages = yield* Effect.forEach(
324 [...new Set([env.region, ...LIST_LOCATIONS])],
325 (location) => listLocation(env.project, location),
326 { concurrency: 4 },
327 );
328 return pages.flat();
329 }),
330
331 reconcile: Effect.fn(function* ({ id, news, output }) {
332 const env = yield* GcpEnvironment.current;
333 const location = normalizeLocation(
334 news.location ?? output?.location,
335 env.region,
336 );
337 const sessionId = yield* toPhysicalId(
338 id,
339 news.sessionId,
340 output?.sessionId,
341 MAX_WORKLOAD_ID_LENGTH,
342 "session",
343 );
344 const name = resourceName(env.project, location, sessionId);
345 const desiredLabels = {
346 ...toLabels(news.labels),
347 ...(yield* createInternalLabels(id)),
348 };
349 const desired = desiredBody(news, name, desiredLabels);
350
351 let current = yield* getByName(name);
352
353 if (current === undefined) {
354 const created = yield* dataproc
355 .createProjectsLocationsSessions({
356 parent: locationParent(env.project, location),
357 sessionId,
358 body: desired,
359 })
360 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
361 if (created) {
362 yield* waitForOperation(created);
363 }
364 current = yield* waitUntilExists(getByName(name), name);
365 current = yield* waitUntilReady(name);
366 }
367
368 if (current === undefined) {
369 return yield* new SessionNotResolved({ name });
370 }
371
372 return toAttrs(current, env.project, location);
373 }),
374
375 delete: Effect.fn(function* ({ output }) {
376 if (!output.name) return;
377 const operation = yield* dataproc
378 .deleteProjectsLocationsSessions({ name: output.name })
379 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
380 if (operation !== undefined) {
381 yield* waitForOperation(operation, { notFoundOk: true });
382 }
383 yield* waitUntilGone(getByName(output.name), output.name);
384 }),
385 });