1import * as cloudwatch from "@distilled.cloud/aws/cloudwatch";
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 { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import type { Providers } from "../Providers.ts";
10import {
11 detectorIdentity,
12 matchesDetectorIdentity,
13 retryConcurrent,
14} from "./common.ts";
15
16class AnomalyDetectorNotVisible extends Data.TaggedError(
17 "AnomalyDetectorNotVisible",
18)<{
19 message: string;
20}> {}
21
22export interface AnomalyDetectorProps
23 extends cloudwatch.PutAnomalyDetectorInput {}
24
25export interface AnomalyDetector extends Resource<
26 "AWS.CloudWatch.AnomalyDetector",
27 AnomalyDetectorProps,
28 {
29 /** Synthetic identifier derived from the detector's metric identity. */
30 detectorId: string;
31 /** The full AnomalyDetector description as last read from CloudWatch. */
32 anomalyDetector: cloudwatch.AnomalyDetector;
33 },
34 never,
35 Providers
36> {}
37
38/**
39 * A CloudWatch anomaly detector — trains a model on a metric's historical
40 * data and computes an expected-value band, which alarms can use via the
41 * `ANOMALY_DETECTION_BAND` metric-math function.
42 * ### Creating Detectors
43 * **Example:** Single Metric Detector
44 * ```typescript
45 * const detector = yield* AnomalyDetector("ErrorsDetector", {
46 * Namespace: "AWS/Lambda",
47 * MetricName: "Errors",
48 * Stat: "Sum",
49 * });
50 * ```
51 *
52 * **Example:** Detector on a Custom Metric
53 * ```typescript
54 * // pair with PutMetricData publishing to the same namespace/metric
55 * const detector = yield* AnomalyDetector("PaymentsDetector", {
56 * Namespace: "MyApp/Payments",
57 * MetricName: "PaymentProcessed",
58 * Stat: "Sum",
59 * });
60 * ```
61 *
62 * ### Reading Detectors at Runtime
63 * **Example:** List Detectors from a Function
64 * ```typescript
65 * // init — see DescribeAnomalyDetectors
66 * const describeAnomalyDetectors = yield* AWS.CloudWatch.DescribeAnomalyDetectors();
67 *
68 * // runtime
69 * const result = yield* describeAnomalyDetectors({ Namespace: "MyApp/Payments" });
70 * ```
71 *
72 * @resource
73 */
74export const AnomalyDetector = Resource<AnomalyDetector>(
75 "AWS.CloudWatch.AnomalyDetector",
76);
77
78const toDescribeRequest = (
79 input: cloudwatch.PutAnomalyDetectorInput,
80): cloudwatch.DescribeAnomalyDetectorsInput => {
81 if (input.MetricMathAnomalyDetector) {
82 return {
83 AnomalyDetectorTypes: ["METRIC_MATH"],
84 };
85 }
86
87 return {
88 Namespace: input.Namespace,
89 MetricName: input.MetricName,
90 Dimensions: input.Dimensions,
91 AnomalyDetectorTypes: ["SINGLE_METRIC"],
92 };
93};
94
95const toDeleteRequest = (
96 input: Pick<
97 cloudwatch.AnomalyDetector,
98 | "Namespace"
99 | "MetricName"
100 | "Dimensions"
101 | "Stat"
102 | "SingleMetricAnomalyDetector"
103 | "MetricMathAnomalyDetector"
104 >,
105): cloudwatch.DeleteAnomalyDetectorInput => {
106 if (input.MetricMathAnomalyDetector) {
107 return {
108 MetricMathAnomalyDetector: input.MetricMathAnomalyDetector,
109 };
110 }
111
112 if (input.SingleMetricAnomalyDetector) {
113 return {
114 SingleMetricAnomalyDetector: input.SingleMetricAnomalyDetector,
115 };
116 }
117
118 return {
119 Namespace: input.Namespace,
120 MetricName: input.MetricName,
121 Dimensions: input.Dimensions,
122 Stat: input.Stat,
123 };
124};
125
126const detectorReadinessSchedule = Schedule.max([
127 Schedule.exponential(200),
128 Schedule.recurs(8),
129]);
130
131const describeDetector = Effect.fn(function* (
132 props: cloudwatch.PutAnomalyDetectorInput,
133 options?: {
134 resourceId?: string;
135 attempt?: number;
136 logMisses?: boolean;
137 },
138) {
139 const request = toDescribeRequest(props);
140 // Stop paginating at the first identity match; a miss drains every page
141 // anyway, which is exactly what the miss diagnostics below need.
142 const detectors = yield* cloudwatch.describeAnomalyDetectors
143 .items(request)
144 .pipe(
145 Stream.takeUntil((candidate) =>
146 matchesDetectorIdentity(candidate, props),
147 ),
148 Stream.runCollect,
149 Effect.map((chunk) => Array.from(chunk)),
150 );
151 const detector = detectors.find((candidate) =>
152 matchesDetectorIdentity(candidate, props),
153 );
154
155 if (!detector && options?.logMisses) {
156 const prefix = options.resourceId
157 ? `${options.resourceId}: anomaly detector not yet visible`
158 : "anomaly detector not yet visible";
159 const attempt =
160 options.attempt === undefined ? "" : ` (attempt ${options.attempt})`;
161
162 yield* Effect.logInfo(
163 `${prefix}${attempt}; request=${JSON.stringify(request)} candidates=${JSON.stringify(
164 detectors.map((candidate) => ({
165 identity: detectorIdentity(candidate),
166 Namespace: candidate.Namespace,
167 MetricName: candidate.MetricName,
168 Stat: candidate.Stat,
169 SingleMetricAnomalyDetector: candidate.SingleMetricAnomalyDetector,
170 MetricMathAnomalyDetector: candidate.MetricMathAnomalyDetector,
171 })),
172 )}`,
173 );
174 }
175
176 return detector;
177});
178
179export const AnomalyDetectorProvider = () =>
180 Provider.succeed(AnomalyDetector, {
181 stables: ["detectorId"],
182 list: () =>
183 // `describeAnomalyDetectors` is paginated and account/region-scoped;
184 // collect every page and flatten the `AnomalyDetectors` array into the
185 // full `Attributes` shape `read` produces (keyed by detector identity).
186 cloudwatch.describeAnomalyDetectors.pages({}).pipe(
187 Stream.runCollect,
188 Effect.map((chunk) =>
189 Array.from(chunk).flatMap((page) =>
190 (page.AnomalyDetectors ?? []).map((detector) => ({
191 detectorId: detectorIdentity(detector),
192 anomalyDetector: detector,
193 })),
194 ),
195 ),
196 ),
197 diff: Effect.fn(function* ({ olds = {}, news = {} }) {
198 if (!isResolved(news)) return undefined;
199 if (detectorIdentity(olds) !== detectorIdentity(news)) {
200 return { action: "replace" } as const;
201 }
202 }),
203 read: Effect.fn(function* ({ olds, output }) {
204 const props = output?.anomalyDetector ?? olds;
205 if (!props) {
206 return undefined;
207 }
208
209 const detector = yield* describeDetector(props);
210
211 if (!detector) {
212 return undefined;
213 }
214
215 return {
216 detectorId: detectorIdentity(props),
217 anomalyDetector: detector,
218 };
219 }),
220 reconcile: Effect.fn(function* ({ news, session }) {
221 // Ensure — `putAnomalyDetector` is an upsert keyed by the detector's
222 // identity (Namespace+MetricName+Dimensions+Stat or MetricMath
223 // expression). Sending desired props every reconcile converges the
224 // cloud regardless of whether this is first-create or an update.
225 yield* retryConcurrent(cloudwatch.putAnomalyDetector(news));
226 const detectorId = detectorIdentity(news);
227 yield* session.note(detectorId);
228
229 // Sync — describe the detector after the put. CloudWatch is
230 // eventually consistent for `DescribeAnomalyDetectors`, so we retry
231 // until the detector matching our identity is visible.
232 let attempt = 0;
233 const state = yield* Effect.suspend(() => {
234 attempt += 1;
235 return describeDetector(news, {
236 resourceId: "AnomalyDetector",
237 attempt,
238 logMisses: true,
239 }).pipe(
240 Effect.flatMap((state) =>
241 state
242 ? Effect.succeed(state)
243 : Effect.fail(
244 new AnomalyDetectorNotVisible({
245 message: "Anomaly detector not yet visible",
246 }),
247 ),
248 ),
249 );
250 }).pipe(
251 Effect.retry({
252 while: (error) => error._tag === "AnomalyDetectorNotVisible",
253 schedule: detectorReadinessSchedule,
254 }),
255 );
256
257 return {
258 detectorId,
259 anomalyDetector: state,
260 };
261 }),
262 delete: Effect.fn(function* ({ output }) {
263 yield* retryConcurrent(
264 cloudwatch.deleteAnomalyDetector(
265 toDeleteRequest(output.anomalyDetector),
266 ),
267 ).pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
268 }),
269 });