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