1import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
2import * as ec2 from "@distilled.cloud/aws/ec2";
3import * as obs from "@distilled.cloud/aws/observabilityadmin";
4import * as Data from "effect/Data";
5import type * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Schedule from "effect/Schedule";
8import * as Stream from "effect/Stream";
9import { Unowned } from "../../AdoptPolicy.ts";
10import { isResolved } from "../../Diff.ts";
11import { createPhysicalName } from "../../PhysicalName.ts";
12import * as Provider from "../../Provider.ts";
13import { Resource } from "../../Resource.ts";
14import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
15import { toWireDays } from "../../Util/Duration.ts";
16import type { Providers } from "../Providers.ts";
17
18/**
19 * Where a telemetry rule delivers the telemetry it enables. Mirrors the wire
20 * `TelemetryDestinationConfiguration`, with the retention period expressed as
21 * a `Duration.Input` instead of the wire's `RetentionInDays`.
22 */
23export interface TelemetryRuleDestination extends Omit<
24 obs.TelemetryDestinationConfiguration,
25 "RetentionInDays"
26> {
27 /**
28 * How long CloudWatch Logs retains the telemetry delivered by this rule.
29 * Accepts any `Duration.Input` (e.g. `"30 days"`, `Duration.days(90)`);
30 * converted to whole days on the wire (`RetentionInDays`).
31 */
32 Retention?: Duration.Input;
33}
34
35export interface TelemetryRuleProps {
36 /**
37 * Name of the telemetry rule. If omitted, a deterministic physical name is
38 * generated. Changing the name replaces the rule.
39 */
40 ruleName?: string;
41 /**
42 * The type of telemetry the rule enables — `"Logs"`, `"Metrics"`, or
43 * `"Traces"`.
44 */
45 telemetryType: obs.TelemetryType;
46 /**
47 * The AWS resource type the rule applies to, e.g. `"AWS::EC2::VPC"`.
48 */
49 resourceType?: obs.ResourceType;
50 /**
51 * The telemetry sources the rule enables, e.g. `["VPC_FLOW_LOGS"]`.
52 */
53 telemetrySourceTypes?: obs.TelemetrySourceType[];
54 /**
55 * Where the enabled telemetry is delivered (destination type, pattern,
56 * retention, and per-source parameters such as VPC flow-log format).
57 */
58 destinationConfiguration?: TelemetryRuleDestination;
59 /**
60 * Organization-level scope selector (organization rules only).
61 */
62 scope?: string;
63 /**
64 * Criteria selecting which resources the rule configures (such as
65 * resource tags). The expression format is service-defined (the console's
66 * tag filter builder produces it); an invalid expression is rejected with
67 * `ValidationException: Invalid resource selection criteria`. Without
68 * selection criteria the rule applies to every resource of `resourceType`
69 * in the account.
70 */
71 selectionCriteria?: string;
72 /**
73 * Whether the rule's telemetry parameters may be updated after creation.
74 */
75 allowFieldUpdates?: boolean;
76 /**
77 * Regions the rule applies to. Mutually exclusive with `allRegions`.
78 */
79 regions?: string[];
80 /**
81 * Apply the rule in all regions. Mutually exclusive with `regions`.
82 */
83 allRegions?: boolean;
84 /**
85 * User-defined tags for the rule.
86 */
87 tags?: Record<string, string>;
88}
89
90export interface TelemetryRule extends Resource<
91 "AWS.ObservabilityAdmin.TelemetryRule",
92 TelemetryRuleProps,
93 {
94 /** The name of the telemetry rule. */
95 ruleName: string;
96 /** The ARN of the telemetry rule. */
97 ruleArn: string;
98 /** The telemetry type the rule enables. */
99 telemetryType: string | undefined;
100 /** The resource type the rule applies to. */
101 resourceType: string | undefined;
102 /** The tags applied to the rule. */
103 tags: Record<string, string>;
104 },
105 never,
106 Providers
107> {}
108
109/**
110 * A CloudWatch **telemetry rule** (Observability Admin) — automatically
111 * enables telemetry (such as VPC flow logs) for AWS resources in your
112 * account that match the rule's criteria.
113 *
114 * The account must be onboarded to CloudWatch telemetry config (see
115 * `ObservabilityAdmin.TelemetryConfig`) before rules can be created.
116 *
117 * ### Creating a Telemetry Rule
118 * **Example:** Enable VPC flow logs for the account's VPCs
119 * ```typescript
120 * import * as ObservabilityAdmin from "alchemy/AWS/ObservabilityAdmin";
121 *
122 * // Onboard the account to telemetry config first.
123 * const telemetry = yield* ObservabilityAdmin.TelemetryConfig("Telemetry");
124 *
125 * const rule = yield* ObservabilityAdmin.TelemetryRule("FlowLogs", {
126 * resourceType: "AWS::EC2::VPC",
127 * telemetryType: "Logs",
128 * telemetrySourceTypes: ["VPC_FLOW_LOGS"],
129 * destinationConfiguration: {
130 * DestinationType: "cloud-watch-logs",
131 * Retention: "30 days",
132 * },
133 * });
134 * ```
135 *
136 * **Example:** Custom flow-log parameters
137 * ```typescript
138 * const rule = yield* ObservabilityAdmin.TelemetryRule("FlowLogs", {
139 * resourceType: "AWS::EC2::VPC",
140 * telemetryType: "Logs",
141 * telemetrySourceTypes: ["VPC_FLOW_LOGS"],
142 * destinationConfiguration: {
143 * DestinationType: "cloud-watch-logs",
144 * Retention: "90 days",
145 * VPCFlowLogParameters: {
146 * TrafficType: "REJECT",
147 * MaxAggregationInterval: 600,
148 * },
149 * },
150 * });
151 * ```
152 *
153 * @resource
154 */
155export const TelemetryRule = Resource<TelemetryRule>(
156 "AWS.ObservabilityAdmin.TelemetryRule",
157);
158
159/** Normalize the wire tag map (values may be undefined) to a plain record. */
160const toTagRecord = (
161 tags: { [key: string]: string | undefined } | undefined,
162): Record<string, string> =>
163 Object.fromEntries(
164 Object.entries(tags ?? {}).filter(
165 (entry): entry is [string, string] => entry[1] !== undefined,
166 ),
167 );
168
169const sameJson = (a: unknown, b: unknown): boolean =>
170 JSON.stringify(a) === JSON.stringify(b);
171
172/**
173 * True when any field the desired rule declares differs from the observed
174 * rule. Fields the desired rule leaves unset are don't-cares (AWS echoes
175 * service defaults for them).
176 */
177const ruleNeedsUpdate = (
178 desired: obs.TelemetryRule,
179 observed: obs.TelemetryRule | undefined,
180): boolean => {
181 if (observed === undefined) return true;
182 const observedByKey = new Map<string, unknown>(Object.entries(observed));
183 return Object.entries(desired).some(
184 ([key, value]) => !sameJson(value, observedByKey.get(key)),
185 );
186};
187
188const vpcFlowLogGroupName = "/aws/vpc";
189
190const isVpcFlowLogRule = (rule: obs.TelemetryRule | undefined): boolean =>
191 rule?.ResourceType === "AWS::EC2::VPC" &&
192 rule.TelemetryType === "Logs" &&
193 rule.TelemetrySourceTypes?.includes("VPC_FLOW_LOGS") === true &&
194 rule.DestinationConfiguration?.DestinationType === "cloud-watch-logs";
195
196const isManagedVpcFlowLog = (
197 flowLog: ec2.FlowLog,
198): flowLog is ec2.FlowLog & { FlowLogId: string } =>
199 flowLog.FlowLogId !== undefined &&
200 flowLog.ResourceId?.startsWith("vpc-") === true &&
201 flowLog.LogGroupName === vpcFlowLogGroupName &&
202 flowLog.LogDestinationType === "cloud-watch-logs" &&
203 flowLog.Tags?.some(
204 (tag) =>
205 tag.Key === "CloudWatchTelemetryRuleManaged" && tag.Value === "true",
206 ) === true;
207
208class TelemetryRuleStillExists extends Data.TaggedError(
209 "TelemetryRuleStillExists",
210)<{ readonly ruleName: string }> {}
211
212class ManagedVpcFlowLogDeleteFailed extends Data.TaggedError(
213 "ManagedVpcFlowLogDeleteFailed",
214)<{ readonly failures: ec2.UnsuccessfulItem[] }> {}
215
216class ManagedVpcFlowLogsStillExist extends Data.TaggedError(
217 "ManagedVpcFlowLogsStillExist",
218)<{ readonly flowLogIds: string[] }> {}
219
220class ManagedVpcLogGroupStillExists extends Data.TaggedError(
221 "ManagedVpcLogGroupStillExists",
222)<{ readonly logGroupName: string }> {}
223
224const cleanupObservationSchedule = Schedule.max([
225 Schedule.fixed("1 second"),
226 Schedule.recurs(10),
227]);
228
229/** Project props to the wire `TelemetryRule` struct (dropping undefineds). */
230const toWireRule = (props: TelemetryRuleProps): obs.TelemetryRule =>
231 JSON.parse(
232 JSON.stringify({
233 ResourceType: props.resourceType,
234 TelemetryType: props.telemetryType,
235 TelemetrySourceTypes: props.telemetrySourceTypes,
236 DestinationConfiguration:
237 props.destinationConfiguration === undefined
238 ? undefined
239 : (() => {
240 const { Retention, ...rest } = props.destinationConfiguration;
241 return {
242 ...rest,
243 RetentionInDays: toWireDays(Retention),
244 };
245 })(),
246 Scope: props.scope,
247 SelectionCriteria: props.selectionCriteria,
248 AllowFieldUpdates: props.allowFieldUpdates,
249 Regions: props.regions,
250 AllRegions: props.allRegions,
251 }),
252 );
253
254export const TelemetryRuleProvider = () =>
255 Provider.effect(
256 TelemetryRule,
257 Effect.gen(function* () {
258 const createName = Effect.fn(function* (
259 id: string,
260 props: TelemetryRuleProps,
261 ) {
262 return props.ruleName ?? (yield* createPhysicalName({ id }));
263 });
264
265 const readRule = Effect.fn(function* (identifier: string) {
266 return yield* obs
267 .getTelemetryRule({ RuleIdentifier: identifier })
268 .pipe(
269 Effect.catchTag("ResourceNotFoundException", () =>
270 Effect.succeed(undefined),
271 ),
272 );
273 });
274
275 const waitUntilRuleAbsent = Effect.fn(function* (ruleName: string) {
276 yield* readRule(ruleName).pipe(
277 Effect.flatMap((rule) =>
278 rule === undefined
279 ? Effect.void
280 : Effect.fail(new TelemetryRuleStillExists({ ruleName })),
281 ),
282 Effect.retry({
283 while: (error) => error._tag === "TelemetryRuleStillExists",
284 schedule: cleanupObservationSchedule,
285 }),
286 );
287 });
288
289 const matchingRulesRemain = Effect.fn(function* () {
290 const summaries = yield* obs.listTelemetryRules.items({}).pipe(
291 Stream.runCollect,
292 Effect.map((items) => Array.from(items)),
293 );
294 const rules = yield* Effect.forEach(
295 summaries.flatMap((summary) =>
296 summary.RuleName === undefined ? [] : [summary.RuleName],
297 ),
298 readRule,
299 { concurrency: 4 },
300 );
301 return rules.some((rule) => isVpcFlowLogRule(rule?.TelemetryRule));
302 });
303
304 const listManagedVpcFlowLogs = () =>
305 ec2.describeFlowLogs
306 .items({
307 Filter: [
308 {
309 Name: "tag:CloudWatchTelemetryRuleManaged",
310 Values: ["true"],
311 },
312 { Name: "log-group-name", Values: [vpcFlowLogGroupName] },
313 ],
314 })
315 .pipe(
316 Stream.runCollect,
317 Effect.map((items) =>
318 Array.from(items).filter(isManagedVpcFlowLog),
319 ),
320 );
321
322 const waitUntilManagedFlowLogsAbsent = Effect.fn(function* (
323 deletedIds: string[],
324 ) {
325 const deleted = new Set(deletedIds);
326 yield* listManagedVpcFlowLogs().pipe(
327 Effect.flatMap((flowLogs) => {
328 const remaining = flowLogs
329 .map((flowLog) => flowLog.FlowLogId)
330 .filter((flowLogId) => deleted.has(flowLogId));
331 return remaining.length === 0
332 ? Effect.void
333 : Effect.fail(
334 new ManagedVpcFlowLogsStillExist({
335 flowLogIds: remaining,
336 }),
337 );
338 }),
339 Effect.retry({
340 while: (error) => error._tag === "ManagedVpcFlowLogsStillExist",
341 schedule: cleanupObservationSchedule,
342 }),
343 );
344 });
345
346 const waitUntilVpcLogGroupAbsent = logs
347 .describeLogGroups({
348 logGroupNamePrefix: vpcFlowLogGroupName,
349 limit: 1,
350 })
351 .pipe(
352 Effect.flatMap((response) =>
353 response.logGroups?.some(
354 (group) => group.logGroupName === vpcFlowLogGroupName,
355 ) === true
356 ? Effect.fail(
357 new ManagedVpcLogGroupStillExists({
358 logGroupName: vpcFlowLogGroupName,
359 }),
360 )
361 : Effect.void,
362 ),
363 Effect.retry({
364 while: (error) => error._tag === "ManagedVpcLogGroupStillExists",
365 schedule: cleanupObservationSchedule,
366 }),
367 );
368
369 const cleanupManagedVpcFlowLogs = Effect.fn(function* () {
370 const managed = yield* listManagedVpcFlowLogs();
371 const flowLogIds = managed.map((flowLog) => flowLog.FlowLogId);
372 if (flowLogIds.length > 0) {
373 yield* ec2.deleteFlowLogs({ FlowLogIds: flowLogIds }).pipe(
374 Effect.flatMap((response) => {
375 const failures = (response.Unsuccessful ?? []).filter(
376 (failure) =>
377 failure.Error?.Code !== "InvalidFlowLogId.NotFound",
378 );
379 return failures.length === 0
380 ? Effect.void
381 : Effect.fail(new ManagedVpcFlowLogDeleteFailed({ failures }));
382 }),
383 Effect.retry({
384 while: (error) => error._tag === "ManagedVpcFlowLogDeleteFailed",
385 schedule: cleanupObservationSchedule,
386 }),
387 );
388 yield* waitUntilManagedFlowLogsAbsent(flowLogIds);
389 }
390
391 // `/aws/vpc` is shared by all Observability Admin-managed VPC flow
392 // logs. Delete it only after no flow log of any ownership still uses
393 // the destination; otherwise deleting the group would destroy user
394 // logs or another rule's delivery target.
395 const groupConsumers = yield* ec2.describeFlowLogs
396 .items({
397 Filter: [{ Name: "log-group-name", Values: [vpcFlowLogGroupName] }],
398 })
399 .pipe(Stream.runCollect);
400 if (groupConsumers.length > 0) return;
401
402 yield* logs.deleteLogGroup({ logGroupName: vpcFlowLogGroupName }).pipe(
403 Effect.retry({
404 while: (error) =>
405 error._tag === "OperationAbortedException" ||
406 error._tag === "ServiceUnavailableException",
407 schedule: cleanupObservationSchedule,
408 }),
409 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
410 );
411 yield* waitUntilVpcLogGroupAbsent;
412 });
413
414 const readTags = Effect.fn(function* (ruleArn: string) {
415 return yield* obs.listTagsForResource({ ResourceARN: ruleArn }).pipe(
416 Effect.map((r) => toTagRecord(r.Tags)),
417 Effect.catchTag("ResourceNotFoundException", () =>
418 Effect.succeed({} as Record<string, string>),
419 ),
420 );
421 });
422
423 const toAttrs = (
424 rule: obs.GetTelemetryRuleOutput,
425 tags: Record<string, string>,
426 ) => ({
427 ruleName: rule.RuleName ?? "",
428 ruleArn: rule.RuleArn ?? "",
429 telemetryType: rule.TelemetryRule?.TelemetryType,
430 resourceType: rule.TelemetryRule?.ResourceType,
431 tags,
432 });
433
434 const readAttrs = Effect.fn(function* (identifier: string) {
435 const rule = yield* readRule(identifier);
436 if (rule === undefined || rule.RuleArn === undefined) return undefined;
437 const tags = yield* readTags(rule.RuleArn);
438 return toAttrs(rule, tags);
439 });
440
441 return TelemetryRule.Provider.of({
442 stables: ["ruleName", "ruleArn"],
443
444 list: () =>
445 obs.listTelemetryRules.items({}).pipe(
446 Stream.runCollect,
447 Effect.flatMap((summaries) =>
448 Effect.forEach(
449 Array.from(summaries).flatMap((s) =>
450 s.RuleName === undefined ? [] : [s.RuleName],
451 ),
452 (name) => readAttrs(name),
453 { concurrency: 4 },
454 ),
455 ),
456 Effect.map((attrs) =>
457 attrs.flatMap((a) => (a === undefined ? [] : [a])),
458 ),
459 ),
460
461 read: Effect.fn(function* ({ id, olds, output }) {
462 const name = output?.ruleName ?? (yield* createName(id, olds ?? {}));
463 const attrs = yield* readAttrs(name);
464 if (attrs === undefined) return undefined;
465 return (yield* hasAlchemyTags(id, attrs.tags))
466 ? attrs
467 : Unowned(attrs);
468 }),
469
470 diff: Effect.fn(function* ({ id, news, olds }) {
471 if (!isResolved(news)) return undefined;
472 // The name is the only immutable property; the rule configuration
473 // is updatable in place via UpdateTelemetryRule.
474 if ((yield* createName(id, olds)) !== (yield* createName(id, news))) {
475 return { action: "replace" } as const;
476 }
477 return undefined;
478 }),
479
480 reconcile: Effect.fn(function* ({ id, news, output, session }) {
481 const props = news!;
482 const name = output?.ruleName ?? (yield* createName(id, props));
483 const internalTags = yield* createInternalTags(id);
484 const desiredTags = { ...props.tags, ...internalTags };
485 const desiredRule = toWireRule(props);
486
487 // 1. Observe — cloud state is authoritative.
488 let observed = yield* readRule(name);
489
490 // 2. Ensure — create if missing; a ConflictException means a peer
491 // created the same-named rule concurrently, so re-observe.
492 if (observed === undefined) {
493 yield* obs
494 .createTelemetryRule({
495 RuleName: name,
496 Rule: desiredRule,
497 Tags: desiredTags,
498 })
499 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
500 // The read-after-create is eventually consistent; wait (bounded)
501 // for the rule to materialize.
502 observed = yield* readRule(name).pipe(
503 Effect.flatMap((rule) =>
504 rule === undefined
505 ? Effect.fail(
506 new Error(`telemetry rule '${name}' not yet visible`),
507 )
508 : Effect.succeed(rule),
509 ),
510 Effect.retry({
511 schedule: Schedule.max([
512 Schedule.fixed("2 seconds"),
513 Schedule.recurs(8),
514 ]),
515 }),
516 );
517 } else if (ruleNeedsUpdate(desiredRule, observed.TelemetryRule)) {
518 // 3. Sync — UpdateTelemetryRule replaces the whole rule config;
519 // apply only on an actual delta against OBSERVED state.
520 yield* obs.updateTelemetryRule({
521 RuleIdentifier: name,
522 Rule: desiredRule,
523 });
524 observed = (yield* readRule(name)) ?? observed;
525 }
526
527 // 3b. Sync tags — diff against OBSERVED cloud tags.
528 const ruleArn = observed.RuleArn!;
529 const observedTags = yield* readTags(ruleArn);
530 const { removed, upsert } = diffTags(observedTags, desiredTags);
531 if (upsert.length > 0) {
532 yield* obs.tagResource({
533 ResourceARN: ruleArn,
534 Tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
535 });
536 }
537 if (removed.length > 0) {
538 yield* obs.untagResource({
539 ResourceARN: ruleArn,
540 TagKeys: removed,
541 });
542 }
543
544 yield* session.note(name);
545 const tags = yield* readTags(ruleArn);
546 return toAttrs(observed, tags);
547 }),
548
549 delete: Effect.fn(function* ({ output, olds }) {
550 // Capture the live rule before deleting it so adoption and drift do
551 // not make cleanup decisions from stale desired state.
552 const observed = yield* readRule(output.ruleName);
553 const deletedRule =
554 observed?.TelemetryRule ??
555 (olds === undefined ? undefined : toWireRule(olds));
556 yield* obs
557 .deleteTelemetryRule({ RuleIdentifier: output.ruleName })
558 .pipe(
559 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
560 );
561 yield* waitUntilRuleAbsent(output.ruleName);
562
563 // Observability Admin asynchronously creates EC2 flow logs and the
564 // shared `/aws/vpc` destination for VPC_FLOW_LOGS rules, but deleting
565 // the rule does not remove them. Reap only when this was such a rule
566 // and no other live rule can own the shared managed resources.
567 if (
568 isVpcFlowLogRule(deletedRule) &&
569 !(yield* matchingRulesRemain())
570 ) {
571 yield* cleanupManagedVpcFlowLogs();
572 }
573 }),
574 });
575 }),
576 );