1import * as aiplatform from "@distilled.cloud/gcp/aiplatform_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 * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { tagRecord } from "../../Tags.ts";
11import { GcpEnvironment } from "../Environment.ts";
12import { hasAlchemyLabels } from "../Labels.ts";
13import type { Providers } from "../Providers.ts";
14import {
15 alchemyIdFilter,
16 createInternalLabels,
17 hasAlchemyPrefix,
18 labelsDiffer,
19 listLocations,
20 locationOf,
21 lastSegment,
22 normalizeLocation,
23 projectOf,
24 stableJson,
25 toDisplayName,
26 toLabels,
27 userLabels,
28} from "./names.ts";
29import { waitForOperation } from "./operations.ts";
30
31export type ModelDeploymentMonitoringJobProps = {
32 /**
33 * Vertex AI location. Immutable — changing it replaces the job.
34 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
35 */
36 location?: string;
37 /**
38 * Display name (max 128 Unicode characters). Generated from the stack,
39 * stage, and logical id when omitted.
40 */
41 displayName?: string;
42 /**
43 * User labels. Alchemy ownership labels are merged in automatically.
44 */
45 labels?: Record<string, string>;
46 /**
47 * Endpoint resource name
48 * `projects/{project}/locations/{location}/endpoints/{endpoint}`.
49 * Immutable.
50 */
51 endpoint: string;
52 /**
53 * Sample strategy for logging prediction requests.
54 */
55 loggingSamplingStrategy: aiplatform.GoogleCloudAiplatformV1SamplingStrategy;
56 /**
57 * Schedule config for running the monitoring job.
58 */
59 modelDeploymentMonitoringScheduleConfig: aiplatform.GoogleCloudAiplatformV1ModelDeploymentMonitoringScheduleConfig;
60 /**
61 * Per-DeployedModel monitoring objectives.
62 */
63 modelDeploymentMonitoringObjectiveConfigs: aiplatform.GoogleCloudAiplatformV1ModelDeploymentMonitoringObjectiveConfigList;
64 /**
65 * Alert config for model monitoring.
66 */
67 modelMonitoringAlertConfig?: aiplatform.GoogleCloudAiplatformV1ModelMonitoringAlertConfig;
68 /**
69 * YAML schema URI describing a single predict instance.
70 */
71 predictInstanceSchemaUri?: string;
72 /**
73 * YAML schema URI describing a TFDV analysis instance.
74 */
75 analysisInstanceSchemaUri?: string;
76 /**
77 * Sample predict instance used instead of `predictInstanceSchemaUri`.
78 */
79 samplePredictInstance?: unknown;
80 /**
81 * GCS folder for stats anomalies.
82 */
83 statsAnomaliesBaseDirectory?: aiplatform.GoogleCloudAiplatformV1GcsDestination;
84 /**
85 * TTL of BigQuery log tables (duration string).
86 */
87 logTtl?: string;
88 /**
89 * Send monitoring pipeline logs to Cloud Logging.
90 * @default false
91 */
92 enableMonitoringPipelineLogs?: boolean;
93 /**
94 * Customer-managed encryption key. Immutable.
95 */
96 encryptionSpec?: aiplatform.GoogleCloudAiplatformV1EncryptionSpec;
97};
98
99export type ModelDeploymentMonitoringJob = Resource<
100 "GCP.AIPlatform.ModelDeploymentMonitoringJob",
101 ModelDeploymentMonitoringJobProps,
102 {
103 /** Full resource name. */
104 name: string;
105 /** Job id (last path segment). */
106 modelDeploymentMonitoringJobId: string;
107 /** Project id. */
108 project: string;
109 /** Location id. */
110 location: string;
111 /** Display name. */
112 displayName: string | undefined;
113 /** User labels (Alchemy ownership labels stripped). */
114 labels: Record<string, string>;
115 /** Monitored endpoint resource name. */
116 endpoint: string | undefined;
117 /** Job state. */
118 state: string | undefined;
119 /** Schedule state while running. */
120 scheduleState: string | undefined;
121 /** Whether pipeline logs are enabled. */
122 enableMonitoringPipelineLogs: boolean;
123 /** RFC3339 creation timestamp. */
124 createTime: string | undefined;
125 /** RFC3339 last-update timestamp. */
126 updateTime: string | undefined;
127 /** RFC3339 next schedule time. */
128 nextScheduleTime: string | undefined;
129 },
130 never,
131 Providers
132>;
133
134/**
135 * A Vertex AI ModelDeploymentMonitoringJob that periodically scores a
136 * deployed model for skew and drift.
137 *
138 * Changing `location` or `endpoint` replaces the job. Display name,
139 * labels, schedule, sampling, alerts, and objective configs update in
140 * place (long-running). Delete pauses the job first.
141 *
142 * ### Creating a ModelDeploymentMonitoringJob
143 * **Example:** Hourly sampling
144 * ```typescript
145 * const job = yield* GCP.AIPlatform.ModelDeploymentMonitoringJob("Watch", {
146 * endpoint: endpoint.name,
147 * loggingSamplingStrategy: { randomSampleConfig: { sampleRate: 0.1 } },
148 * modelDeploymentMonitoringScheduleConfig: { monitorInterval: "3600s" },
149 * modelDeploymentMonitoringObjectiveConfigs: [{
150 * deployedModelId: deployed.id,
151 * objectiveConfig: {
152 * predictionDriftDetectionConfig: {},
153 * },
154 * }],
155 * });
156 * ```
157 *
158 * @resource
159 * @category AIPlatform
160 */
161export const ModelDeploymentMonitoringJob =
162 Resource<ModelDeploymentMonitoringJob>(
163 "GCP.AIPlatform.ModelDeploymentMonitoringJob",
164 );
165
166export class ModelDeploymentMonitoringJobNotResolved extends Data.TaggedError(
167 "GCP.AIPlatform.ModelDeploymentMonitoringJobNotResolved",
168)<{
169 name: string;
170}> {}
171
172export class ModelDeploymentMonitoringJobStillExists extends Data.TaggedError(
173 "GCP.AIPlatform.ModelDeploymentMonitoringJobStillExists",
174)<{
175 name: string;
176}> {}
177
178const toAttrs = (
179 job: aiplatform.GoogleCloudAiplatformV1ModelDeploymentMonitoringJob,
180 project: string,
181) => {
182 const name = job.name ?? "";
183 return {
184 name,
185 modelDeploymentMonitoringJobId: lastSegment(name),
186 project: projectOf(name, project),
187 location: locationOf(name),
188 displayName: job.displayName,
189 labels: userLabels(job.labels),
190 endpoint: job.endpoint,
191 state: job.state,
192 scheduleState: job.scheduleState,
193 enableMonitoringPipelineLogs: job.enableMonitoringPipelineLogs === true,
194 createTime: job.createTime,
195 updateTime: job.updateTime,
196 nextScheduleTime: job.nextScheduleTime,
197 };
198};
199
200const getByName = (name: string) =>
201 aiplatform
202 .getProjectsLocationsModelDeploymentMonitoringJobs({ name })
203 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
204
205const listPage = (parent: string, filter?: string) =>
206 aiplatform.listProjectsLocationsModelDeploymentMonitoringJobs
207 .pages({ parent, pageSize: 100, filter })
208 .pipe(
209 Stream.runCollect,
210 Effect.map((pages) =>
211 Array.from(pages).flatMap(
212 (page) => page.modelDeploymentMonitoringJobs ?? [],
213 ),
214 ),
215 Effect.catchTag("NotFound", () =>
216 Effect.succeed(
217 [] as aiplatform.GoogleCloudAiplatformV1ModelDeploymentMonitoringJob[],
218 ),
219 ),
220 );
221
222const findOwned = (
223 project: string,
224 location: string,
225 labels: Record<string, string>,
226) =>
227 listPage(
228 `projects/${project}/locations/${location}`,
229 alchemyIdFilter(labels),
230 ).pipe(
231 Effect.map(
232 (items) =>
233 items.find((item) => hasAlchemyPrefix(item.labels)) ?? undefined,
234 ),
235 );
236
237const waitUntilExists = (name: string) =>
238 getByName(name).pipe(
239 Effect.filterOrFail(
240 (
241 job,
242 ): job is aiplatform.GoogleCloudAiplatformV1ModelDeploymentMonitoringJob =>
243 job !== undefined,
244 () => new ModelDeploymentMonitoringJobNotResolved({ name }),
245 ),
246 Effect.retry({
247 while: (error) =>
248 error._tag === "GCP.AIPlatform.ModelDeploymentMonitoringJobNotResolved",
249 times: 8,
250 schedule: Schedule.spaced("2 seconds"),
251 }),
252 );
253
254const waitUntilGone = (name: string) =>
255 getByName(name).pipe(
256 Effect.filterOrFail(
257 (job) => job === undefined,
258 () => new ModelDeploymentMonitoringJobStillExists({ name }),
259 ),
260 Effect.asVoid,
261 Effect.retry({
262 while: (error) =>
263 error._tag === "GCP.AIPlatform.ModelDeploymentMonitoringJobStillExists",
264 times: 10,
265 schedule: Schedule.spaced("2 seconds"),
266 }),
267 );
268
269export const ModelDeploymentMonitoringJobProvider = () =>
270 Provider.succeed(ModelDeploymentMonitoringJob, {
271 stables: [
272 "name",
273 "modelDeploymentMonitoringJobId",
274 "project",
275 "location",
276 "endpoint",
277 "createTime",
278 ],
279
280 diff: Effect.fn(function* ({ news, olds, output }) {
281 const env = yield* GcpEnvironment.current;
282 if (!isResolved(news)) return undefined;
283 const previousLocation = normalizeLocation(
284 olds?.location ?? output?.location,
285 env.region,
286 );
287 const nextLocation = normalizeLocation(
288 news.location ?? olds?.location ?? output?.location,
289 env.region,
290 );
291 const previousEndpoint = olds?.endpoint ?? output?.endpoint;
292 const nextEndpoint = news.endpoint ?? previousEndpoint;
293 if (
294 previousLocation !== nextLocation ||
295 (previousEndpoint !== undefined &&
296 nextEndpoint !== undefined &&
297 nextEndpoint !== previousEndpoint)
298 ) {
299 return { action: "replace" as const, deleteFirst: false };
300 }
301 return undefined;
302 }),
303
304 read: Effect.fn(function* ({ id, olds, output }) {
305 const env = yield* GcpEnvironment.current;
306 const location = normalizeLocation(
307 olds?.location ?? output?.location,
308 env.region,
309 );
310 const ownership = yield* createInternalLabels(id);
311 const existing =
312 (output?.name !== undefined
313 ? yield* getByName(output.name)
314 : undefined) ?? (yield* findOwned(env.project, location, ownership));
315 if (existing === undefined) return undefined;
316 const attrs = toAttrs(existing, env.project);
317 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
318 ? attrs
319 : Unowned(attrs);
320 }),
321
322 list: () =>
323 Effect.gen(function* () {
324 const env = yield* GcpEnvironment.current;
325 const pages = yield* Effect.forEach(
326 listLocations(env.region),
327 (location) =>
328 listPage(`projects/${env.project}/locations/${location}`),
329 { concurrency: 4 },
330 );
331 return pages
332 .flat()
333 .filter((job) => hasAlchemyPrefix(job.labels))
334 .map((job) => toAttrs(job, env.project));
335 }),
336
337 reconcile: Effect.fn(function* ({ id, news, output }) {
338 const env = yield* GcpEnvironment.current;
339 const location = normalizeLocation(
340 news.location ?? output?.location,
341 env.region,
342 );
343 const displayName = yield* toDisplayName(
344 id,
345 news.displayName,
346 output?.displayName,
347 );
348 const desiredLabels = {
349 ...toLabels(news.labels),
350 ...(yield* createInternalLabels(id)),
351 };
352 const enableLogs = news.enableMonitoringPipelineLogs === true;
353
354 let current =
355 (output?.name !== undefined
356 ? yield* getByName(output.name)
357 : undefined) ??
358 (yield* findOwned(env.project, location, desiredLabels));
359
360 if (current === undefined) {
361 const created = yield* aiplatform
362 .createProjectsLocationsModelDeploymentMonitoringJobs({
363 parent: `projects/${env.project}/locations/${location}`,
364 body: {
365 displayName,
366 labels: desiredLabels,
367 endpoint: news.endpoint,
368 loggingSamplingStrategy: news.loggingSamplingStrategy,
369 modelDeploymentMonitoringScheduleConfig:
370 news.modelDeploymentMonitoringScheduleConfig,
371 modelDeploymentMonitoringObjectiveConfigs:
372 news.modelDeploymentMonitoringObjectiveConfigs,
373 modelMonitoringAlertConfig: news.modelMonitoringAlertConfig,
374 predictInstanceSchemaUri: news.predictInstanceSchemaUri,
375 analysisInstanceSchemaUri: news.analysisInstanceSchemaUri,
376 samplePredictInstance: news.samplePredictInstance,
377 statsAnomaliesBaseDirectory: news.statsAnomaliesBaseDirectory,
378 logTtl: news.logTtl,
379 enableMonitoringPipelineLogs: enableLogs,
380 encryptionSpec: news.encryptionSpec,
381 },
382 })
383 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
384 current =
385 created ?? (yield* findOwned(env.project, location, desiredLabels));
386 }
387
388 if (current === undefined || current.name === undefined) {
389 return yield* new ModelDeploymentMonitoringJobNotResolved({
390 name: output?.name ?? `${location}/modelDeploymentMonitoringJobs`,
391 });
392 }
393
394 const name = current.name;
395 const displayChanged = (current.displayName ?? "") !== displayName;
396 const labelsChanged = labelsDiffer(current.labels, desiredLabels);
397 const logsChanged =
398 (current.enableMonitoringPipelineLogs === true) !== enableLogs;
399 const samplingChanged =
400 stableJson(current.loggingSamplingStrategy) !==
401 stableJson(news.loggingSamplingStrategy);
402 const scheduleChanged =
403 stableJson(current.modelDeploymentMonitoringScheduleConfig) !==
404 stableJson(news.modelDeploymentMonitoringScheduleConfig);
405 const objectivesChanged =
406 stableJson(current.modelDeploymentMonitoringObjectiveConfigs) !==
407 stableJson(news.modelDeploymentMonitoringObjectiveConfigs);
408 const alertChanged =
409 news.modelMonitoringAlertConfig !== undefined &&
410 stableJson(current.modelMonitoringAlertConfig) !==
411 stableJson(news.modelMonitoringAlertConfig);
412 const logTtlChanged = (current.logTtl ?? "") !== (news.logTtl ?? "");
413
414 if (
415 displayChanged ||
416 labelsChanged ||
417 logsChanged ||
418 samplingChanged ||
419 scheduleChanged ||
420 objectivesChanged ||
421 alertChanged ||
422 logTtlChanged
423 ) {
424 const operation =
425 yield* aiplatform.patchProjectsLocationsModelDeploymentMonitoringJobs(
426 {
427 name,
428 updateMask: [
429 displayChanged ? "display_name" : undefined,
430 labelsChanged ? "labels" : undefined,
431 logsChanged ? "enable_monitoring_pipeline_logs" : undefined,
432 samplingChanged ? "logging_sampling_strategy" : undefined,
433 scheduleChanged
434 ? "model_deployment_monitoring_schedule_config"
435 : undefined,
436 objectivesChanged
437 ? "model_deployment_monitoring_objective_configs"
438 : undefined,
439 alertChanged ? "model_monitoring_alert_config" : undefined,
440 logTtlChanged ? "log_ttl" : undefined,
441 ]
442 .filter((field): field is string => field !== undefined)
443 .join(","),
444 body: {
445 name,
446 displayName,
447 labels: desiredLabels,
448 enableMonitoringPipelineLogs: enableLogs,
449 loggingSamplingStrategy: news.loggingSamplingStrategy,
450 modelDeploymentMonitoringScheduleConfig:
451 news.modelDeploymentMonitoringScheduleConfig,
452 modelDeploymentMonitoringObjectiveConfigs:
453 news.modelDeploymentMonitoringObjectiveConfigs,
454 modelMonitoringAlertConfig: news.modelMonitoringAlertConfig,
455 logTtl: news.logTtl,
456 },
457 },
458 );
459 yield* waitForOperation(operation);
460 current = yield* waitUntilExists(name);
461 }
462
463 return toAttrs(current, env.project);
464 }),
465
466 delete: Effect.fn(function* ({ output }) {
467 yield* aiplatform
468 .pauseProjectsLocationsModelDeploymentMonitoringJobs({
469 name: output.name,
470 body: {},
471 })
472 .pipe(
473 Effect.catchTag(
474 ["NotFound", "BadRequest", "Conflict"],
475 () => Effect.void,
476 ),
477 );
478 const operation = yield* aiplatform
479 .deleteProjectsLocationsModelDeploymentMonitoringJobs({
480 name: output.name,
481 })
482 .pipe(
483 Effect.retry({
484 while: (error) => error._tag === "Conflict",
485 times: 8,
486 schedule: Schedule.spaced("2 seconds"),
487 }),
488 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
489 );
490 if (operation !== undefined) {
491 yield* waitForOperation(operation, { notFoundOk: true });
492 }
493 yield* waitUntilGone(output.name);
494 }),
495 });