1import * as AWS from "@/AWS";
2import {
3 Application,
4 ApplicationCloudWatchLoggingOption,
5} from "@/AWS/KinesisAnalyticsV2";
6import * as Test from "@/Test/Alchemy";
7import * as analytics from "@distilled.cloud/aws/kinesis-analytics-v2";
8import { describe, expect } from "alchemy-test";
9import * as Data from "effect/Data";
10import * as Effect from "effect/Effect";
11import * as Schedule from "effect/Schedule";
12import {
13 codeKey,
14 deleteCodeBucketIdempotent,
15 provisionCodeBucket,
16} from "./code-bucket.ts";
17
18const { test } = Test.make({ providers: AWS.providers() });
19
20// Deterministic out-of-band code bucket (see code-bucket.ts) — an in-stack
21// bucket orphans on a crashed run because its generated name is lost with
22// the state.
23const loggingCodeBucket = "alchemy-test-kav2-logging-code";
24
25class ApplicationStillExists extends Data.TaggedError(
26 "ApplicationStillExists",
27) {}
28
29const assertApplicationDeleted = Effect.fn(function* (applicationName: string) {
30 yield* analytics
31 .describeApplication({ ApplicationName: applicationName })
32 .pipe(
33 Effect.flatMap(() => Effect.fail(new ApplicationStillExists())),
34 Effect.retry({
35 while: (e: { _tag: string }) => e._tag === "ApplicationStillExists",
36 schedule: Schedule.max([
37 Schedule.fixed("3 seconds"),
38 Schedule.recurs(40),
39 ]),
40 }),
41 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
42 );
43});
44
45describe.skipIf(!!process.env.FAST)(
46 "AWS.KinesisAnalyticsV2.ApplicationCloudWatchLoggingOption",
47 {
48 tags: [
49 "provider:aws",
50 "provider:aws:kinesisanalyticsv2",
51 "provider:aws:logs",
52 "provider:aws:s3",
53 "live",
54 ],
55 },
56 () => {
57 test.provider(
58 "attach a CloudWatch logging option, detach it, destroy",
59 (stack) =>
60 Effect.gen(function* () {
61 yield* stack.destroy();
62
63 // Stage the code object BEFORE the application exists — the service
64 // reads (and hashes) the object at create/update time.
65 const codeBucketArn = yield* provisionCodeBucket(loggingCodeBucket);
66
67 const makeBase = Effect.gen(function* () {
68 const logGroup = yield* AWS.Logs.LogGroup("FlinkLogGroup", {});
69 const logStream = yield* AWS.Logs.LogStream("FlinkLogStream", {
70 logGroupName: logGroup.logGroupName,
71 });
72 const app = yield* Application("LoggedFlinkApp", {
73 runtimeEnvironment: "FLINK-1_20",
74 code: { bucketArn: codeBucketArn, fileKey: codeKey },
75 });
76 return { logGroup, logStream, app };
77 });
78
79 const deployed = yield* stack.deploy(
80 Effect.gen(function* () {
81 const base = yield* makeBase;
82 const logging = yield* ApplicationCloudWatchLoggingOption(
83 "FlinkLogging",
84 {
85 applicationName: base.app.applicationName,
86 logStreamArn: base.logStream.logStreamArn.as<string>(),
87 },
88 );
89 return { ...base, logging };
90 }),
91 );
92
93 expect(deployed.logging.applicationName).toEqual(
94 deployed.app.applicationName,
95 );
96 expect(deployed.logging.cloudWatchLoggingOptionId).toBeDefined();
97 expect(deployed.logging.logStreamArn).toContain(":log-stream:");
98
99 // Out-of-band: the option is visible on the application.
100 const described = yield* analytics.describeApplication({
101 ApplicationName: deployed.app.applicationName,
102 });
103 const options =
104 described.ApplicationDetail.CloudWatchLoggingOptionDescriptions ??
105 [];
106 expect(options).toHaveLength(1);
107 expect(options[0]?.LogStreamARN).toEqual(
108 deployed.logging.logStreamArn,
109 );
110
111 // Remove the option while the application stays deployed — the
112 // engine deletes just the sub-resource.
113 const detached = yield* stack.deploy(makeBase);
114 expect(detached.app.applicationName).toEqual(
115 deployed.app.applicationName,
116 );
117
118 const afterDetach = yield* analytics.describeApplication({
119 ApplicationName: deployed.app.applicationName,
120 });
121 expect(
122 afterDetach.ApplicationDetail.CloudWatchLoggingOptionDescriptions ??
123 [],
124 ).toHaveLength(0);
125
126 yield* stack.destroy();
127
128 yield* assertApplicationDeleted(deployed.app.applicationName);
129 }).pipe(Effect.ensuring(deleteCodeBucketIdempotent(loggingCodeBucket))),
130 { timeout: 300_000 },
131 );
132 },
133);