1import * as analytics from "@distilled.cloud/aws/kinesis-analytics-v2";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import type { Providers } from "../Providers.ts";
9
10export interface ApplicationCloudWatchLoggingOptionProps {
11 /**
12 * Name of the Managed Service for Apache Flink application to attach the
13 * logging option to. Changing the application replaces the option.
14 */
15 applicationName: string;
16 /**
17 * ARN of the CloudWatch Logs log stream that receives application
18 * messages. Changing the log stream replaces the option.
19 */
20 logStreamArn: string;
21}
22
23export interface ApplicationCloudWatchLoggingOption extends Resource<
24 "AWS.KinesisAnalyticsV2.ApplicationCloudWatchLoggingOption",
25 ApplicationCloudWatchLoggingOptionProps,
26 {
27 /**
28 * Name of the application the option is attached to.
29 */
30 applicationName: string;
31 /**
32 * ARN of the CloudWatch Logs log stream receiving application messages.
33 */
34 logStreamArn: string;
35 /**
36 * Service-assigned ID of the logging option within the application.
37 */
38 cloudWatchLoggingOptionId: string | undefined;
39 },
40 never,
41 Providers
42> {}
43
44/**
45 * Attaches an Amazon CloudWatch Logs log stream to a Managed Service for
46 * Apache Flink application so application messages (errors, job lifecycle
47 * events) are delivered to CloudWatch.
48 *
49 * The option is identified by the log stream it delivers to — changing
50 * either the application or the log stream replaces the option. The
51 * application's service execution role must be allowed to call
52 * `logs:PutLogEvents` / `logs:DescribeLogStreams` (the role auto-created by
53 * `Application` already is).
54 * ### Attaching Logging
55 * **Example:** Deliver application messages to a log stream
56 * ```typescript
57 * import * as AWS from "alchemy/AWS";
58 *
59 * const logGroup = yield* AWS.Logs.LogGroup("FlinkLogs");
60 * const logStream = yield* AWS.Logs.LogStream("FlinkLogStream", {
61 * logGroupName: logGroup.logGroupName,
62 * });
63 * const logging = yield* AWS.KinesisAnalyticsV2.ApplicationCloudWatchLoggingOption(
64 * "AppLogging",
65 * {
66 * applicationName: app.applicationName,
67 * logStreamArn: logStream.logStreamArn.as<string>(),
68 * },
69 * );
70 * ```
71 *
72 * @resource
73 */
74export const ApplicationCloudWatchLoggingOption =
75 Resource<ApplicationCloudWatchLoggingOption>(
76 "AWS.KinesisAnalyticsV2.ApplicationCloudWatchLoggingOption",
77 );
78
79/**
80 * The logging option could not be observed on the application after it was
81 * added — the add call succeeded but the option never appeared.
82 */
83export class LoggingOptionNotFound extends Data.TaggedError(
84 "LoggingOptionNotFound",
85)<{
86 readonly applicationName: string;
87 readonly logStreamArn: string;
88}> {}
89
90/**
91 * Retries `ResourceInUseException` / `ConcurrentModificationException`
92 * (application transitioning between statuses, or a concurrent operation in
93 * flight) on a bounded schedule. Explicit return annotation for the same
94 * declaration-emit reason as `retryThroughRolePropagation` in
95 * `Application.ts`.
96 */
97const retryWhileInUse = <A, E extends { _tag: string }, R>(
98 self: Effect.Effect<A, E, R>,
99): Effect.Effect<A, E, R> =>
100 Effect.retry(self, {
101 while: (e) =>
102 e._tag === "ResourceInUseException" ||
103 e._tag === "ConcurrentModificationException",
104 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(30)]),
105 });
106
107const describeApplicationDetail = Effect.fn(function* (
108 applicationName: string,
109) {
110 const response = yield* analytics
111 .describeApplication({ ApplicationName: applicationName })
112 .pipe(
113 Effect.catchTag("ResourceNotFoundException", () =>
114 Effect.succeed(undefined),
115 ),
116 );
117 return response?.ApplicationDetail;
118});
119
120const findOption = (
121 detail: analytics.ApplicationDetail | undefined,
122 logStreamArn: string,
123) =>
124 detail?.CloudWatchLoggingOptionDescriptions?.find(
125 (option) => option.LogStreamARN === logStreamArn,
126 );
127
128export const ApplicationCloudWatchLoggingOptionProvider = () =>
129 Provider.effect(
130 ApplicationCloudWatchLoggingOption,
131 Effect.gen(function* () {
132 return ApplicationCloudWatchLoggingOption.Provider.of({
133 stables: [
134 "applicationName",
135 "logStreamArn",
136 "cloudWatchLoggingOptionId",
137 ],
138
139 // Sub-resource keyed by its parent application — there is no
140 // account-level enumeration of logging options, so `list` is empty
141 // and refresh flows through `read` on known instances.
142 list: () => Effect.succeed([]),
143
144 read: Effect.fn(function* ({ olds, output }) {
145 const applicationName =
146 output?.applicationName ?? olds?.applicationName;
147 const logStreamArn = output?.logStreamArn ?? olds?.logStreamArn;
148 if (!applicationName || !logStreamArn) return undefined;
149 const detail = yield* describeApplicationDetail(applicationName);
150 const option = findOption(detail, logStreamArn);
151 if (!option) return undefined;
152 return {
153 applicationName,
154 logStreamArn,
155 cloudWatchLoggingOptionId: option.CloudWatchLoggingOptionId,
156 };
157 }),
158
159 diff: Effect.fn(function* ({ news, olds }) {
160 if (!isResolved(news)) return undefined;
161 // The option's identity is (application, log stream) — the only
162 // way to change either is to delete and re-add.
163 if (
164 olds?.applicationName !== news?.applicationName ||
165 olds?.logStreamArn !== news?.logStreamArn
166 ) {
167 return { action: "replace" } as const;
168 }
169 }),
170
171 // Existence-only sub-resource: observe → if missing, add. The add
172 // uses the freshly observed application version id as its
173 // compare-and-set token and is retried through concurrent
174 // modifications with a re-read.
175 reconcile: Effect.fn(function* ({ news, session }) {
176 const applicationName = news.applicationName;
177 const logStreamArn = news.logStreamArn;
178
179 const ensureOption = Effect.gen(function* () {
180 const detail = yield* describeApplicationDetail(applicationName);
181 const existing = findOption(detail, logStreamArn);
182 if (existing || !detail) return existing;
183 const response =
184 yield* analytics.addApplicationCloudWatchLoggingOption({
185 ApplicationName: applicationName,
186 CurrentApplicationVersionId: detail.ApplicationVersionId,
187 CloudWatchLoggingOption: { LogStreamARN: logStreamArn },
188 });
189 return response.CloudWatchLoggingOptionDescriptions?.find(
190 (option) => option.LogStreamARN === logStreamArn,
191 );
192 });
193 const option = yield* retryWhileInUse(ensureOption);
194 if (!option) {
195 return yield* Effect.fail(
196 new LoggingOptionNotFound({ applicationName, logStreamArn }),
197 );
198 }
199
200 yield* session.note(
201 `CloudWatch logging option ${option.CloudWatchLoggingOptionId ?? logStreamArn}`,
202 );
203 return {
204 applicationName,
205 logStreamArn,
206 cloudWatchLoggingOptionId: option.CloudWatchLoggingOptionId,
207 };
208 }),
209
210 delete: Effect.fn(function* ({ output }) {
211 // A missing application means the option is already gone.
212 const removeOption = Effect.gen(function* () {
213 const detail = yield* describeApplicationDetail(
214 output.applicationName,
215 );
216 const option = findOption(detail, output.logStreamArn);
217 const optionId =
218 option?.CloudWatchLoggingOptionId ??
219 output.cloudWatchLoggingOptionId;
220 if (!detail || !option || !optionId) return;
221 yield* analytics
222 .deleteApplicationCloudWatchLoggingOption({
223 ApplicationName: output.applicationName,
224 CurrentApplicationVersionId: detail.ApplicationVersionId,
225 CloudWatchLoggingOptionId: optionId,
226 })
227 .pipe(
228 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
229 );
230 });
231 yield* retryWhileInUse(removeOption);
232 }),
233 });
234 }),
235 );