Open History to see the last commit
1import * as ec2 from "@distilled.cloud/aws/ec2";
2import { isResolved } from "alchemy/Diff";
3import * as Provider from "alchemy/Provider";
4import { Resource } from "alchemy/Resource";
5import { createInternalTags, createTagsList, diffTags } from "alchemy/Tags";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as Stream from "effect/Stream";
10import type { AccountID } from "../Environment.ts";
11import { AWSEnvironment } from "../Environment.ts";
12import type { Providers } from "../Providers.ts";
13import type { RegionID } from "../Region.ts";
14
15export type FlowLogId<ID extends string = string> = `fl-${ID}`;
16
17export type FlowLogArn<ID extends FlowLogId = FlowLogId> =
18 `arn:aws:ec2:${RegionID}:${AccountID}:vpc-flow-log/${ID}`;
19
20/**
21 * Raised when `createFlowLogs`/`deleteFlowLogs` report an item in the
22 * `Unsuccessful` array (these APIs never throw for per-resource failures).
23 */
24export class FlowLogOperationFailed extends Data.TaggedError("FlowLogOperationFailed")<{
25 readonly code: string;
26 readonly message: string;
27}> {}
28
29export interface FlowLogProps {
30 /**
31 * The type of resource to monitor.
32 */
33 resourceType: "VPC" | "Subnet" | "NetworkInterface";
34
35 /**
36 * The ID of the VPC, subnet, or network interface to capture flow logs for.
37 * Immutable — changing it replaces the flow log.
38 */
39 resourceId: string;
40
41 /**
42 * The type of traffic to capture.
43 * @default "ALL"
44 */
45 trafficType?: "ACCEPT" | "REJECT" | "ALL";
46
47 /**
48 * Where to deliver the flow logs.
49 * @default "cloud-watch-logs"
50 */
51 logDestinationType?: "cloud-watch-logs" | "s3" | "kinesis-data-firehose";
52
53 /**
54 * The name of the CloudWatch Logs log group. Required (and only valid) when
55 * `logDestinationType` is `cloud-watch-logs`.
56 */
57 logGroupName?: string;
58
59 /**
60 * The ARN of the IAM role that permits EC2 to publish flow logs to the
61 * CloudWatch Logs group. Required when `logDestinationType` is
62 * `cloud-watch-logs`.
63 */
64 deliverLogsPermissionArn?: string;
65
66 /**
67 * The ARN of the destination for the flow logs when `logDestinationType` is
68 * `s3` (a bucket ARN, optionally with a key prefix) or
69 * `kinesis-data-firehose` (a delivery stream ARN).
70 */
71 logDestination?: string;
72
73 /**
74 * The maximum interval, in seconds, during which a flow of packets is
75 * captured and aggregated into a flow log record. Either `60` or `600`.
76 * @default 600
77 */
78 maxAggregationInterval?: 60 | 600;
79
80 /**
81 * A custom format string for the flow log records. If omitted, the default
82 * format is used.
83 */
84 logFormat?: string;
85
86 /**
87 * Tags to assign to the flow log.
88 */
89 tags?: Record<string, string>;
90}
91
92export interface FlowLog extends Resource<
93 "AWS.EC2.FlowLog",
94 FlowLogProps,
95 {
96 /**
97 * The ID of the flow log (prefixed `fl-`).
98 */
99 flowLogId: FlowLogId;
100
101 /**
102 * The Amazon Resource Name (ARN) of the flow log.
103 */
104 flowLogArn: FlowLogArn;
105
106 /**
107 * The ID of the monitored resource.
108 */
109 resourceId: string;
110
111 /**
112 * The type of traffic captured.
113 */
114 trafficType: "ACCEPT" | "REJECT" | "ALL";
115
116 /**
117 * Where the flow logs are delivered.
118 */
119 logDestinationType: "cloud-watch-logs" | "s3" | "kinesis-data-firehose";
120 },
121 never,
122 Providers
123> {}
124
125/**
126 * A flow log captures information about the IP traffic going to and from a VPC,
127 * subnet, or network interface, and publishes it to CloudWatch Logs, S3, or a
128 * Kinesis Data Firehose delivery stream. Use it for network monitoring, traffic
129 * analysis, and troubleshooting security-group / NACL rules.
130 *
131 * A flow log is immutable: every property except `tags` is fixed at creation,
132 * so changing the monitored resource, destination, or traffic type replaces the
133 * flow log. For CloudWatch Logs delivery you must supply a `logGroupName` and a
134 * `deliverLogsPermissionArn` — an IAM role that EC2's `vpc-flow-logs` service
135 * principal can assume to write to the group.
136 *
137 * ### Creating a Flow Log
138 * **Example:** VPC Flow Log to CloudWatch Logs
139 * ```typescript
140 * const logGroup = yield* AWS.Logs.LogGroup("FlowLogs", {});
141 *
142 * const role = yield* AWS.IAM.Role("FlowLogRole", {
143 * assumeRolePolicyDocument: {
144 * Version: "2012-10-17",
145 * Statement: [{
146 * Effect: "Allow",
147 * Principal: { Service: "vpc-flow-logs.amazonaws.com" },
148 * Action: "sts:AssumeRole",
149 * }],
150 * },
151 * inlinePolicies: {
152 * deliver: {
153 * Version: "2012-10-17",
154 * Statement: [{
155 * Effect: "Allow",
156 * Action: [
157 * "logs:CreateLogStream",
158 * "logs:PutLogEvents",
159 * "logs:DescribeLogStreams",
160 * ],
161 * Resource: "*",
162 * }],
163 * },
164 * },
165 * });
166 *
167 * const flowLog = yield* AWS.EC2.FlowLog("VpcFlowLog", {
168 * resourceType: "VPC",
169 * resourceId: myVpc.vpcId,
170 * logGroupName: logGroup.logGroupName,
171 * deliverLogsPermissionArn: role.roleArn,
172 * });
173 * ```
174 * Captures all traffic for the VPC and delivers it to the CloudWatch Logs
175 * group via the delivery role.
176 *
177 * **Example:** S3 Flow Log for Rejected Traffic
178 * ```typescript
179 * const flowLog = yield* AWS.EC2.FlowLog("RejectedTraffic", {
180 * resourceType: "Subnet",
181 * resourceId: mySubnet.subnetId,
182 * trafficType: "REJECT",
183 * logDestinationType: "s3",
184 * logDestination: bucket.bucketArn,
185 * });
186 * ```
187 * Delivers only rejected-traffic records for a subnet directly to an S3 bucket
188 * (no IAM role required for S3 delivery).
189 *
190 * @resource
191 */
192export const FlowLog = Resource<FlowLog>("AWS.EC2.FlowLog");
193
194export const FlowLogProvider = () =>
195 Provider.effect(
196 FlowLog,
197 Effect.gen(function* () {
198 const createTags = Effect.fn(function* (id: string, tags?: Record<string, string>) {
199 return {
200 Name: id,
201 ...(yield* createInternalTags(id)),
202 ...tags,
203 };
204 });
205
206 const describeFlowLog = (flowLogId: string) =>
207 ec2.describeFlowLogs({ FlowLogIds: [flowLogId] }).pipe(
208 Effect.map((r) => r.FlowLogs?.[0]),
209 Effect.catchTag("InvalidFlowLogId.NotFound", () => Effect.undefined),
210 );
211
212 const toAttrs = (fl: ec2.FlowLog) =>
213 AWSEnvironment.current.pipe(
214 Effect.map((env) => ({
215 flowLogId: fl.FlowLogId as FlowLogId,
216 flowLogArn:
217 `arn:aws:ec2:${env.region}:${env.accountId}:vpc-flow-log/${fl.FlowLogId}` as FlowLogArn,
218 resourceId: fl.ResourceId!,
219 trafficType: (fl.TrafficType ?? "ALL") as "ACCEPT" | "REJECT" | "ALL",
220 logDestinationType: (fl.LogDestinationType ?? "cloud-watch-logs") as
221 | "cloud-watch-logs"
222 | "s3"
223 | "kinesis-data-firehose",
224 })),
225 );
226
227 return {
228 stables: ["flowLogId", "flowLogArn"],
229
230 list: () =>
231 Effect.gen(function* () {
232 const env = yield* AWSEnvironment.current;
233 const items = yield* ec2.describeFlowLogs.pages({}).pipe(
234 Stream.runCollect,
235 Effect.map((chunk) =>
236 Array.from(chunk).flatMap((page) =>
237 (page.FlowLogs ?? [])
238 .filter((fl): fl is ec2.FlowLog & { FlowLogId: string } => fl.FlowLogId != null)
239 .map((fl) => ({
240 flowLogId: fl.FlowLogId as FlowLogId,
241 flowLogArn:
242 `arn:aws:ec2:${env.region}:${env.accountId}:vpc-flow-log/${fl.FlowLogId}` as FlowLogArn,
243 resourceId: fl.ResourceId!,
244 trafficType: (fl.TrafficType ?? "ALL") as "ACCEPT" | "REJECT" | "ALL",
245 logDestinationType: (fl.LogDestinationType ?? "cloud-watch-logs") as
246 | "cloud-watch-logs"
247 | "s3"
248 | "kinesis-data-firehose",
249 })),
250 ),
251 ),
252 );
253 return items satisfies FlowLog["Attributes"][];
254 }),
255
256 read: Effect.fn(function* ({ output }) {
257 if (!output) return undefined;
258 const fl = yield* describeFlowLog(output.flowLogId);
259 if (!fl) return undefined;
260 return yield* toAttrs(fl);
261 }),
262
263 diff: Effect.fn(function* ({ news, olds }) {
264 if (!isResolved(news)) return;
265 // Flow logs are immutable except for tags — any structural change
266 // replaces the resource.
267 if (
268 news.resourceType !== olds.resourceType ||
269 news.resourceId !== olds.resourceId ||
270 (news.trafficType ?? "ALL") !== (olds.trafficType ?? "ALL") ||
271 (news.logDestinationType ?? "cloud-watch-logs") !==
272 (olds.logDestinationType ?? "cloud-watch-logs") ||
273 news.logGroupName !== olds.logGroupName ||
274 news.logDestination !== olds.logDestination ||
275 news.deliverLogsPermissionArn !== olds.deliverLogsPermissionArn ||
276 (news.maxAggregationInterval ?? 600) !== (olds.maxAggregationInterval ?? 600) ||
277 news.logFormat !== olds.logFormat
278 ) {
279 return { action: "replace" };
280 }
281 // Only tags are mutable in place.
282 }),
283
284 reconcile: Effect.fn(function* ({ id, news, output, session }) {
285 const desiredTags = yield* createTags(id, news.tags);
286
287 // Observe — find the flow log via the cached id, else create.
288 let fl: ec2.FlowLog | undefined;
289 if (output?.flowLogId) {
290 fl = yield* describeFlowLog(output.flowLogId);
291 }
292
293 // Ensure — create if missing. createFlowLogs reports per-resource
294 // failures in the Unsuccessful array rather than throwing.
295 if (fl === undefined) {
296 yield* session.note("Creating flow log...");
297 // A freshly-created delivery role may not yet be assumable by the
298 // flow-logs service (IAM eventual consistency) — createFlowLogs
299 // reports that in Unsuccessful. Retry that (but not a genuine
300 // "already exists") on a bounded schedule.
301 const result = yield* ec2
302 .createFlowLogs({
303 ResourceType: news.resourceType,
304 ResourceIds: [news.resourceId],
305 TrafficType: news.trafficType ?? "ALL",
306 LogDestinationType: news.logDestinationType ?? "cloud-watch-logs",
307 LogGroupName: news.logGroupName,
308 DeliverLogsPermissionArn: news.deliverLogsPermissionArn,
309 LogDestination: news.logDestination,
310 MaxAggregationInterval: news.maxAggregationInterval,
311 LogFormat: news.logFormat,
312 TagSpecifications: [
313 {
314 ResourceType: "vpc-flow-log",
315 Tags: createTagsList(desiredTags),
316 },
317 ],
318 })
319 .pipe(
320 Effect.flatMap((r) => {
321 const failure = r.Unsuccessful?.[0];
322 return failure
323 ? Effect.fail(
324 new FlowLogOperationFailed({
325 code: failure.Error?.Code ?? "Unknown",
326 message: failure.Error?.Message ?? "createFlowLogs failed",
327 }),
328 )
329 : Effect.succeed(r);
330 }),
331 Effect.retry({
332 while: (e) =>
333 e._tag === "FlowLogOperationFailed" && e.code !== "FlowLogAlreadyExists",
334 schedule: Schedule.max([Schedule.fixed(3000), Schedule.recurs(15)]),
335 }),
336 );
337 const flowLogId = result.FlowLogIds![0]!;
338 yield* session.note(`Flow log created: ${flowLogId}`);
339 fl = yield* describeFlowLog(flowLogId);
340 if (!fl) {
341 // Freshly-created flow log not yet visible — synthesize attrs.
342 return yield* toAttrs({
343 FlowLogId: flowLogId,
344 ResourceId: news.resourceId,
345 TrafficType: news.trafficType ?? "ALL",
346 LogDestinationType: news.logDestinationType ?? "cloud-watch-logs",
347 } as ec2.FlowLog);
348 }
349 }
350
351 const flowLogId = fl.FlowLogId!;
352
353 // Sync tags — observed cloud tags vs desired.
354 const currentTags =
355 (yield* ec2
356 .describeTags({
357 Filters: [
358 { Name: "resource-id", Values: [flowLogId] },
359 { Name: "resource-type", Values: ["vpc-flow-log"] },
360 ],
361 })
362 .pipe(
363 Effect.map(
364 (r) =>
365 Object.fromEntries(r.Tags?.map((t) => [t.Key!, t.Value!]) ?? []) as Record<
366 string,
367 string
368 >,
369 ),
370 )) ?? {};
371 const { removed, upsert } = diffTags(currentTags, desiredTags);
372 if (removed.length > 0) {
373 yield* ec2.deleteTags({
374 Resources: [flowLogId],
375 Tags: removed.map((key) => ({ Key: key })),
376 });
377 }
378 if (upsert.length > 0) {
379 yield* ec2.createTags({ Resources: [flowLogId], Tags: upsert });
380 }
381
382 const final = yield* describeFlowLog(flowLogId);
383 return yield* toAttrs(final ?? fl);
384 }),
385
386 delete: Effect.fn(function* ({ output, session }) {
387 const flowLogId = output.flowLogId;
388 yield* session.note(`Deleting flow log: ${flowLogId}`);
389 // deleteFlowLogs is idempotent — a missing flow log is reported in
390 // Unsuccessful, which we ignore.
391 yield* ec2.deleteFlowLogs({ FlowLogIds: [flowLogId] });
392 }),
393 };
394 }),
395 );