1import * as amp from "@distilled.cloud/aws/amp";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
12import { toWireSeconds } from "../../Util/Duration.ts";
13import type { Providers } from "../Providers.ts";
14import { syncAmpTags, toTagRecord } from "./internal.ts";
15
16/**
17 * Tolerance band around the expected value inside which deviations are not
18 * flagged — either an absolute `amount` or a `ratio` of the expected value.
19 */
20export type AnomalyDetectorIgnoreNearExpected =
21 | { amount: number }
22 | { ratio: number };
23
24/** What the detector does with evaluation windows that have no data. */
25export type AnomalyDetectorMissingDataAction =
26 | { markAsAnomaly: boolean }
27 | { skip: boolean };
28
29export interface AnomalyDetectorProps {
30 /**
31 * Id of the AMP workspace the detector runs in. Changing the workspace
32 * replaces the detector.
33 */
34 workspaceId: string;
35 /**
36 * Human-readable alias for the detector, unique within the workspace.
37 * Changing the alias replaces the detector.
38 */
39 alias: string;
40 /**
41 * The PromQL query producing the time series the Random Cut Forest
42 * algorithm scores for anomalies. Updating it retrains the detector.
43 */
44 query: string;
45 /**
46 * Number of consecutive data points combined into one sample (shingle)
47 * for scoring. If omitted, the service default is used.
48 */
49 shingleSize?: number;
50 /**
51 * Number of points sampled per Random Cut Forest tree. If omitted, the
52 * service default is used.
53 */
54 sampleSize?: number;
55 /**
56 * Suppress anomalies above the expected value that fall within this
57 * tolerance band.
58 */
59 ignoreNearExpectedFromAbove?: AnomalyDetectorIgnoreNearExpected;
60 /**
61 * Suppress anomalies below the expected value that fall within this
62 * tolerance band.
63 */
64 ignoreNearExpectedFromBelow?: AnomalyDetectorIgnoreNearExpected;
65 /**
66 * How often the detector evaluates the query. Accepts any
67 * `Duration.Input` (e.g. `"1 minute"`, `Duration.minutes(1)`; a bare
68 * number is milliseconds); the wire unit is whole seconds
69 * (`evaluationIntervalInSeconds`). If omitted, the service default is
70 * used.
71 */
72 evaluationInterval?: Duration.Input;
73 /**
74 * What to do when an evaluation window has no data: mark it as an
75 * anomaly (`{ markAsAnomaly: true }`) or skip it (`{ skip: true }`).
76 */
77 missingDataAction?: AnomalyDetectorMissingDataAction;
78 /**
79 * Extra labels attached to the anomaly metrics the detector emits.
80 */
81 labels?: Record<string, string>;
82 /**
83 * User-defined tags for the detector.
84 */
85 tags?: Record<string, string>;
86}
87
88export interface AnomalyDetector extends Resource<
89 "AWS.AMP.AnomalyDetector",
90 AnomalyDetectorProps,
91 {
92 workspaceId: string;
93 anomalyDetectorId: string;
94 anomalyDetectorArn: string;
95 alias: string;
96 status: string;
97 },
98 never,
99 Providers
100> {}
101
102/**
103 * A Random Cut Forest anomaly detector inside an Amazon Managed Service for
104 * Prometheus workspace — continuously evaluates a PromQL query and emits
105 * anomaly scores as new metrics in the same workspace.
106 *
107 * ### Creating an Anomaly Detector
108 * **Example:** Detect Anomalies on a Request-Rate Query
109 * ```typescript
110 * const workspace = yield* AMP.Workspace("Metrics", {});
111 * const detector = yield* AMP.AnomalyDetector("RequestSpikes", {
112 * workspaceId: workspace.workspaceId,
113 * alias: "request-spikes",
114 * query: 'rate(http_requests_total{job="api"}[5m])',
115 * evaluationInterval: "1 minute",
116 * missingDataAction: { skip: true },
117 * });
118 * ```
119 *
120 * @resource
121 */
122export const AnomalyDetector = Resource<AnomalyDetector>(
123 "AWS.AMP.AnomalyDetector",
124);
125
126export const AnomalyDetectorProvider = () =>
127 Provider.effect(
128 AnomalyDetector,
129 Effect.gen(function* () {
130 const toAttrs = (
131 workspaceId: string,
132 detector: amp.AnomalyDetectorDescription,
133 ) => ({
134 workspaceId,
135 anomalyDetectorId: detector.anomalyDetectorId,
136 anomalyDetectorArn: detector.arn,
137 alias: detector.alias,
138 status: detector.status.statusCode,
139 });
140
141 /** Describe a detector by id; typed not-found → undefined. */
142 const describe = Effect.fn(function* (
143 workspaceId: string,
144 anomalyDetectorId: string,
145 ) {
146 const response = yield* amp
147 .describeAnomalyDetector({ workspaceId, anomalyDetectorId })
148 .pipe(
149 Effect.catchTag("ResourceNotFoundException", () =>
150 Effect.succeed(undefined),
151 ),
152 );
153 return response?.anomalyDetector;
154 });
155
156 /**
157 * Find a detector by its alias (the deterministic identity within a
158 * workspace) — used when no id is cached in `output`.
159 */
160 const findByAlias = Effect.fn(function* (
161 workspaceId: string,
162 alias: string,
163 ) {
164 const summary = yield* amp.listAnomalyDetectors
165 .items({ workspaceId, alias })
166 .pipe(
167 Stream.filter((s) => s.alias === alias),
168 Stream.runHead,
169 Effect.map(Option.getOrUndefined),
170 Effect.catchTag("ResourceNotFoundException", () =>
171 Effect.succeed(undefined),
172 ),
173 );
174 return summary === undefined
175 ? undefined
176 : yield* describe(workspaceId, summary.anomalyDetectorId);
177 });
178
179 /**
180 * Bounded best-effort wait toward ACTIVE. Training/backfill can
181 * outlast the window; the last observed description is returned and
182 * a still-training detector converges on a later reconcile.
183 */
184 const waitSettled = Effect.fn(function* (
185 workspaceId: string,
186 anomalyDetectorId: string,
187 ) {
188 return yield* amp
189 .describeAnomalyDetector({ workspaceId, anomalyDetectorId })
190 .pipe(
191 Effect.map((r) => r.anomalyDetector),
192 Effect.repeat({
193 schedule: Schedule.max([
194 Schedule.fixed("3 seconds"),
195 Schedule.recurs(20),
196 ]),
197 until: (d): boolean =>
198 d.status.statusCode !== "CREATING" &&
199 d.status.statusCode !== "UPDATING",
200 }),
201 );
202 });
203
204 const canonical = (value: unknown) =>
205 JSON.stringify(value, (_, v: unknown) =>
206 v !== null && typeof v === "object" && !Array.isArray(v)
207 ? Object.fromEntries(
208 Object.entries(v as Record<string, unknown>).sort(([a], [b]) =>
209 a.localeCompare(b),
210 ),
211 )
212 : v,
213 );
214
215 /**
216 * Detect drift between desired props and the observed detector.
217 * Only user-specified fields participate — omitted optional fields
218 * defer to service defaults and never force a (full-retrain) put.
219 */
220 const configDrifts = (
221 news: AnomalyDetectorProps,
222 observed: amp.AnomalyDetectorDescription,
223 ): boolean => {
224 const rcf = observed.configuration?.randomCutForest;
225 const drift =
226 news.query !== rcf?.query ||
227 (news.shingleSize !== undefined &&
228 news.shingleSize !== rcf?.shingleSize) ||
229 (news.sampleSize !== undefined &&
230 news.sampleSize !== rcf?.sampleSize) ||
231 (news.ignoreNearExpectedFromAbove !== undefined &&
232 canonical(news.ignoreNearExpectedFromAbove) !==
233 canonical(rcf?.ignoreNearExpectedFromAbove)) ||
234 (news.ignoreNearExpectedFromBelow !== undefined &&
235 canonical(news.ignoreNearExpectedFromBelow) !==
236 canonical(rcf?.ignoreNearExpectedFromBelow)) ||
237 (news.evaluationInterval !== undefined &&
238 toWireSeconds(news.evaluationInterval) !==
239 observed.evaluationIntervalInSeconds) ||
240 (news.missingDataAction !== undefined &&
241 canonical(news.missingDataAction) !==
242 canonical(observed.missingDataAction)) ||
243 (news.labels !== undefined &&
244 canonical(news.labels) !== canonical(toTagRecord(observed.labels)));
245 return drift;
246 };
247
248 const toConfiguration = (news: AnomalyDetectorProps) => ({
249 randomCutForest: {
250 query: news.query,
251 shingleSize: news.shingleSize,
252 sampleSize: news.sampleSize,
253 ignoreNearExpectedFromAbove: news.ignoreNearExpectedFromAbove,
254 ignoreNearExpectedFromBelow: news.ignoreNearExpectedFromBelow,
255 },
256 });
257
258 return {
259 stables: [
260 "workspaceId",
261 "anomalyDetectorId",
262 "anomalyDetectorArn",
263 "alias",
264 ],
265
266 diff: Effect.fn(function* ({ olds, news }) {
267 if (!isResolved(news)) return undefined;
268 if (
269 olds?.workspaceId !== news.workspaceId ||
270 olds?.alias !== news.alias
271 ) {
272 return { action: "replace" } as const;
273 }
274 }),
275
276 read: Effect.fn(function* ({ id, olds, output }) {
277 const workspaceId = output?.workspaceId ?? olds?.workspaceId;
278 if (!workspaceId) return undefined;
279 const detector =
280 output?.anomalyDetectorId !== undefined
281 ? yield* describe(workspaceId, output.anomalyDetectorId)
282 : olds?.alias !== undefined
283 ? yield* findByAlias(workspaceId, olds.alias)
284 : undefined;
285 if (detector === undefined) return undefined;
286 const attrs = toAttrs(workspaceId, detector);
287 const tags = toTagRecord(detector.tags);
288 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
289 }),
290
291 reconcile: Effect.fn(function* ({ id, news, output, session }) {
292 const workspaceId = news!.workspaceId;
293 const internalTags = yield* createInternalTags(id);
294 const desiredTags = { ...internalTags, ...news!.tags };
295
296 // 1. Observe — by cached id, falling back to the alias.
297 let detector =
298 output?.anomalyDetectorId !== undefined
299 ? yield* describe(workspaceId, output.anomalyDetectorId)
300 : yield* findByAlias(workspaceId, news!.alias);
301
302 // 2. Ensure — create if missing.
303 if (detector === undefined) {
304 const created = yield* amp.createAnomalyDetector({
305 workspaceId,
306 alias: news!.alias,
307 configuration: toConfiguration(news!),
308 evaluationIntervalInSeconds: toWireSeconds(
309 news!.evaluationInterval,
310 ),
311 missingDataAction: news!.missingDataAction,
312 labels: news!.labels,
313 tags: desiredTags,
314 });
315 detector = yield* waitSettled(
316 workspaceId,
317 created.anomalyDetectorId,
318 );
319 } else if (configDrifts(news!, detector)) {
320 // 3. Sync — `putAnomalyDetector` triggers a full retrain, so it
321 // is only called when an observed aspect actually drifts.
322 yield* amp.putAnomalyDetector({
323 workspaceId,
324 anomalyDetectorId: detector.anomalyDetectorId,
325 configuration: toConfiguration(news!),
326 evaluationIntervalInSeconds: toWireSeconds(
327 news!.evaluationInterval,
328 ),
329 missingDataAction: news!.missingDataAction,
330 labels: news!.labels,
331 });
332 detector = yield* waitSettled(
333 workspaceId,
334 detector.anomalyDetectorId,
335 );
336 }
337
338 // 3b. Sync tags — diff against OBSERVED cloud tags.
339 yield* syncAmpTags(detector.arn, desiredTags);
340
341 yield* session.note(`${workspaceId}/${detector.anomalyDetectorId}`);
342 return toAttrs(workspaceId, detector);
343 }),
344
345 delete: Effect.fn(function* ({ output }) {
346 yield* amp
347 .deleteAnomalyDetector({
348 workspaceId: output.workspaceId,
349 anomalyDetectorId: output.anomalyDetectorId,
350 })
351 .pipe(
352 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
353 // A detector mid-training rejects deletion; retry briefly.
354 Effect.retry({
355 while: (e) => e._tag === "ConflictException",
356 schedule: Schedule.max([
357 Schedule.fixed("4 seconds"),
358 Schedule.recurs(25),
359 ]),
360 }),
361 );
362 }),
363
364 // Sub-resource keyed by its parent workspace — not independently
365 // enumerable across the account.
366 list: () => Effect.succeed([]),
367 };
368 }),
369 );