1import * as osis from "@distilled.cloud/aws/osis";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 jsonEquals,
13 readPipelineTags,
14 retryWhilePipelineConflict,
15 waitForPipelineSettled,
16} from "./internal.ts";
17
18export interface PipelineLogPublishingOptions {
19 /**
20 * Whether pipeline logs are published to CloudWatch Logs.
21 */
22 isLoggingEnabled?: boolean;
23 /**
24 * CloudWatch Logs destination. The log group name must start with
25 * `/aws/vendedlogs/`.
26 */
27 cloudWatchLogDestination?: {
28 /** Name of the CloudWatch Logs log group, e.g. `/aws/vendedlogs/OpenSearchIngestion/my-pipeline`. */
29 logGroup: string;
30 };
31}
32
33export interface PipelineBufferOptions {
34 /**
35 * Whether persistent buffering is enabled for the pipeline.
36 */
37 persistentBufferEnabled: boolean;
38}
39
40export interface PipelineEncryptionAtRestOptions {
41 /**
42 * ARN of the customer-managed KMS key used to encrypt buffer data.
43 */
44 kmsKeyArn: string;
45}
46
47export interface PipelineVpcOptions {
48 /**
49 * Subnet IDs the pipeline's VPC endpoint is placed into. Changing VPC
50 * options replaces the pipeline.
51 */
52 subnetIds: string[];
53 /**
54 * Security group IDs applied to the pipeline's VPC endpoint.
55 */
56 securityGroupIds?: string[];
57}
58
59export interface PipelineProps {
60 /**
61 * Name of the pipeline. 3-28 characters; lowercase letters, numbers, and
62 * hyphens. If omitted, a deterministic physical name is generated.
63 * Changing the name replaces the pipeline.
64 */
65 pipelineName?: string;
66 /**
67 * Minimum number of Ingestion OpenSearch Compute Units (OCUs) the pipeline
68 * scales down to. Minimum 1.
69 */
70 minUnits: number;
71 /**
72 * Maximum number of Ingestion OpenSearch Compute Units (OCUs) the pipeline
73 * scales up to.
74 */
75 maxUnits: number;
76 /**
77 * Data Prepper pipeline configuration in YAML (must start with
78 * `version: "2"`). Blueprints for common source/sink topologies are
79 * available via the `listPipelineBlueprints`/`getPipelineBlueprint` APIs.
80 */
81 pipelineConfigurationBody: string;
82 /**
83 * CloudWatch Logs publishing configuration for pipeline logs.
84 */
85 logPublishingOptions?: PipelineLogPublishingOptions;
86 /**
87 * Persistent buffering for the pipeline's ingest data.
88 */
89 bufferOptions?: PipelineBufferOptions;
90 /**
91 * Customer-managed KMS encryption for buffer data.
92 * @default an AWS-owned key
93 */
94 encryptionAtRestOptions?: PipelineEncryptionAtRestOptions;
95 /**
96 * VPC placement for the pipeline's ingest endpoint. Omit for a public
97 * endpoint. Changing VPC options replaces the pipeline.
98 */
99 vpcOptions?: PipelineVpcOptions;
100 /**
101 * IAM role the pipeline assumes to write to its sinks (overrides the
102 * `sts_role_arn` inside the configuration body where supported).
103 */
104 pipelineRoleArn?: string;
105 /**
106 * User-defined tags for the pipeline.
107 */
108 tags?: Record<string, string>;
109}
110
111export interface Pipeline extends Resource<
112 "AWS.OSIS.Pipeline",
113 PipelineProps,
114 {
115 /**
116 * Name of the pipeline.
117 */
118 pipelineName: string;
119 /**
120 * ARN of the pipeline.
121 */
122 pipelineArn: string;
123 /**
124 * Pipeline status (e.g. `ACTIVE`, `CREATING`, `UPDATING`).
125 */
126 status: string;
127 /**
128 * Minimum Ingestion OCUs the pipeline scales down to.
129 */
130 minUnits: number | undefined;
131 /**
132 * Maximum Ingestion OCUs the pipeline scales up to.
133 */
134 maxUnits: number | undefined;
135 /**
136 * URLs to ingest data into the pipeline.
137 */
138 ingestEndpointUrls: string[] | undefined;
139 /**
140 * Tags on the pipeline.
141 */
142 tags: Record<string, string>;
143 },
144 never,
145 Providers
146> {}
147
148/**
149 * An Amazon OpenSearch Ingestion (OSIS) pipeline — a managed Data Prepper
150 * pipeline that ingests, transforms, and delivers data to OpenSearch domains,
151 * serverless collections, or S3.
152 *
153 * Pipelines take roughly 5-10 minutes to provision and are billed per
154 * Ingestion-OCU-hour while they exist (minimum 1 OCU). Destroy pipelines you
155 * are not using.
156 * ### Creating a Pipeline
157 * **Example:** HTTP Source to S3 Sink
158 * ```typescript
159 * const pipeline = yield* Pipeline("Logs", {
160 * minUnits: 1,
161 * maxUnits: 1,
162 * pipelineConfigurationBody: Output.interpolate`version: "2"
163 * log-pipeline:
164 * source:
165 * http:
166 * path: "/logs"
167 * sink:
168 * - s3:
169 * aws:
170 * sts_role_arn: "${role.roleArn}"
171 * region: "us-west-2"
172 * bucket: "${bucket.bucketName}"
173 * threshold:
174 * event_collect_timeout: "60s"
175 * codec:
176 * ndjson:
177 * `,
178 * });
179 * ```
180 *
181 * **Example:** Pipeline with CloudWatch Logging
182 * ```typescript
183 * const pipeline = yield* Pipeline("Logs", {
184 * minUnits: 1,
185 * maxUnits: 2,
186 * pipelineConfigurationBody: configYaml,
187 * logPublishingOptions: {
188 * isLoggingEnabled: true,
189 * cloudWatchLogDestination: {
190 * logGroup: "/aws/vendedlogs/OpenSearchIngestion/logs",
191 * },
192 * },
193 * });
194 * ```
195 *
196 * @resource
197 */
198export const Pipeline = Resource<Pipeline>("AWS.OSIS.Pipeline");
199
200const toLogPublishingOptions = (
201 options: PipelineLogPublishingOptions | undefined,
202): osis.LogPublishingOptions | undefined =>
203 options === undefined
204 ? undefined
205 : {
206 IsLoggingEnabled: options.isLoggingEnabled,
207 CloudWatchLogDestination:
208 options.cloudWatchLogDestination !== undefined
209 ? { LogGroup: options.cloudWatchLogDestination.logGroup }
210 : undefined,
211 };
212
213export const PipelineProvider = () =>
214 Provider.effect(
215 Pipeline,
216 Effect.gen(function* () {
217 const toName = (id: string, props: Partial<PipelineProps>) =>
218 props.pipelineName
219 ? Effect.succeed(props.pipelineName)
220 : createPhysicalName({ id, maxLength: 28, lowercase: true });
221
222 const readPipeline = Effect.fn(function* (name: string) {
223 return yield* osis.getPipeline({ PipelineName: name }).pipe(
224 Effect.map((response) => response.Pipeline),
225 Effect.catchTag("ResourceNotFoundException", () =>
226 Effect.succeed(undefined),
227 ),
228 );
229 });
230
231 const waitForActive = Effect.fn(function* (name: string) {
232 const pipeline = yield* waitForPipelineSettled(
233 name,
234 readPipeline(name),
235 );
236 if (pipeline === undefined) {
237 return yield* Effect.fail(
238 new Error(`OSIS pipeline '${name}' not found while waiting`),
239 );
240 }
241 return pipeline;
242 });
243
244 const toAttrs = Effect.fn(function* (pipeline: osis.Pipeline) {
245 if (!pipeline.PipelineName || !pipeline.PipelineArn) {
246 return yield* Effect.fail(
247 new Error(
248 `OSIS pipeline '${pipeline.PipelineName}' is missing its ARN`,
249 ),
250 );
251 }
252 return {
253 pipelineName: pipeline.PipelineName,
254 pipelineArn: pipeline.PipelineArn,
255 status: pipeline.Status ?? "ACTIVE",
256 minUnits: pipeline.MinUnits,
257 maxUnits: pipeline.MaxUnits,
258 ingestEndpointUrls: pipeline.IngestEndpointUrls,
259 tags: yield* readPipelineTags(pipeline.PipelineArn),
260 };
261 });
262
263 return {
264 stables: ["pipelineName", "pipelineArn"],
265
266 diff: Effect.fn(function* ({ id, olds, news }) {
267 if (!isResolved(news)) return undefined;
268 const o: Partial<PipelineProps> = olds ?? {};
269 const n: Partial<PipelineProps> = news ?? {};
270 if ((yield* toName(id, o)) !== (yield* toName(id, n))) {
271 return { action: "replace" } as const;
272 }
273 // VPC options are create-only.
274 if (!jsonEquals(o.vpcOptions, n.vpcOptions)) {
275 return { action: "replace" } as const;
276 }
277 }),
278
279 read: Effect.fn(function* ({ id, olds, output }) {
280 const name = output?.pipelineName ?? (yield* toName(id, olds ?? {}));
281 const pipeline = yield* readPipeline(name);
282 if (pipeline === undefined) return undefined;
283 const attrs = yield* toAttrs(pipeline);
284 return (yield* hasAlchemyTags(id, attrs.tags))
285 ? attrs
286 : Unowned(attrs);
287 }),
288
289 reconcile: Effect.fn(function* ({ id, news, output, session }) {
290 const props = news!;
291 const name = output?.pipelineName ?? (yield* toName(id, props));
292 const internalTags = yield* createInternalTags(id);
293 const desiredTags = { ...internalTags, ...props.tags };
294
295 // 1. Observe — cloud state is authoritative.
296 let observed = yield* readPipeline(name);
297
298 // 2. Ensure — create if missing; tolerate AlreadyExists as a race.
299 if (observed === undefined) {
300 yield* osis
301 .createPipeline({
302 PipelineName: name,
303 MinUnits: props.minUnits,
304 MaxUnits: props.maxUnits,
305 PipelineConfigurationBody: props.pipelineConfigurationBody,
306 LogPublishingOptions: toLogPublishingOptions(
307 props.logPublishingOptions,
308 ),
309 BufferOptions:
310 props.bufferOptions !== undefined
311 ? {
312 PersistentBufferEnabled:
313 props.bufferOptions.persistentBufferEnabled,
314 }
315 : undefined,
316 EncryptionAtRestOptions:
317 props.encryptionAtRestOptions !== undefined
318 ? { KmsKeyArn: props.encryptionAtRestOptions.kmsKeyArn }
319 : undefined,
320 VpcOptions:
321 props.vpcOptions !== undefined
322 ? {
323 SubnetIds: props.vpcOptions.subnetIds,
324 SecurityGroupIds: props.vpcOptions.securityGroupIds,
325 }
326 : undefined,
327 PipelineRoleArn: props.pipelineRoleArn,
328 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
329 Key,
330 Value,
331 })),
332 })
333 .pipe(
334 Effect.catchTag(
335 "ResourceAlreadyExistsException",
336 () => Effect.void,
337 ),
338 );
339 }
340
341 // Creation and in-flight updates both surface as transitional
342 // statuses; wait (bounded) so update calls do not conflict.
343 observed = yield* waitForActive(name);
344
345 // 3. Sync — compute the update delta from OBSERVED state.
346 const update: osis.UpdatePipelineRequest = { PipelineName: name };
347 let mutated = false;
348 if (props.minUnits !== observed.MinUnits) {
349 update.MinUnits = props.minUnits;
350 mutated = true;
351 }
352 if (props.maxUnits !== observed.MaxUnits) {
353 update.MaxUnits = props.maxUnits;
354 mutated = true;
355 }
356 if (
357 props.pipelineConfigurationBody.trim() !==
358 (observed.PipelineConfigurationBody ?? "").trim()
359 ) {
360 update.PipelineConfigurationBody = props.pipelineConfigurationBody;
361 mutated = true;
362 }
363 const desiredLogging = toLogPublishingOptions(
364 props.logPublishingOptions,
365 );
366 if (
367 desiredLogging !== undefined &&
368 !jsonEquals(
369 {
370 IsLoggingEnabled:
371 observed.LogPublishingOptions?.IsLoggingEnabled,
372 CloudWatchLogDestination:
373 observed.LogPublishingOptions?.CloudWatchLogDestination,
374 },
375 desiredLogging,
376 )
377 ) {
378 update.LogPublishingOptions = desiredLogging;
379 mutated = true;
380 }
381 if (
382 props.bufferOptions !== undefined &&
383 props.bufferOptions.persistentBufferEnabled !==
384 observed.BufferOptions?.PersistentBufferEnabled
385 ) {
386 update.BufferOptions = {
387 PersistentBufferEnabled:
388 props.bufferOptions.persistentBufferEnabled,
389 };
390 mutated = true;
391 }
392 if (
393 props.encryptionAtRestOptions !== undefined &&
394 props.encryptionAtRestOptions.kmsKeyArn !==
395 observed.EncryptionAtRestOptions?.KmsKeyArn
396 ) {
397 update.EncryptionAtRestOptions = {
398 KmsKeyArn: props.encryptionAtRestOptions.kmsKeyArn,
399 };
400 mutated = true;
401 }
402 if (
403 props.pipelineRoleArn !== undefined &&
404 props.pipelineRoleArn !== observed.PipelineRoleArn
405 ) {
406 update.PipelineRoleArn = props.pipelineRoleArn;
407 mutated = true;
408 }
409
410 if (mutated) {
411 yield* osis.updatePipeline(update);
412 observed = yield* waitForActive(name);
413 }
414
415 // 3b. Sync tags — diff against OBSERVED cloud tags.
416 const arn = observed.PipelineArn;
417 if (arn) {
418 const observedTags = yield* readPipelineTags(arn);
419 const { removed, upsert } = diffTags(observedTags, desiredTags);
420 if (upsert.length > 0) {
421 yield* osis.tagResource({ Arn: arn, Tags: upsert });
422 }
423 if (removed.length > 0) {
424 yield* osis.untagResource({ Arn: arn, TagKeys: removed });
425 }
426 }
427
428 yield* session.note(name);
429 return yield* toAttrs(observed);
430 }),
431
432 delete: Effect.fn(function* ({ output }) {
433 const name = output.pipelineName;
434 // A pipeline mid-create/update rejects deletion with
435 // ConflictException — wait (bounded, tolerant) for it to settle,
436 // then retry conflicts through any residual transition.
437 yield* waitForPipelineSettled(name, readPipeline(name)).pipe(
438 Effect.catch(() => Effect.succeed(undefined)),
439 );
440 yield* retryWhilePipelineConflict(
441 osis
442 .deletePipeline({ PipelineName: name })
443 .pipe(
444 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
445 ),
446 );
447 }),
448
449 list: () =>
450 osis.listPipelines.pages({}).pipe(
451 Stream.runCollect,
452 Effect.map((chunk) =>
453 Array.from(chunk).flatMap((page) =>
454 (page.Pipelines ?? [])
455 .map((summary) => summary.PipelineName)
456 .filter((name): name is string => name !== undefined),
457 ),
458 ),
459 Effect.flatMap(
460 Effect.forEach(
461 (name) =>
462 osis.getPipeline({ PipelineName: name }).pipe(
463 Effect.map((response) => response.Pipeline),
464 Effect.catchTag("ResourceNotFoundException", () =>
465 Effect.succeed(undefined),
466 ),
467 ),
468 { concurrency: 4 },
469 ),
470 ),
471 Effect.map((pipelines) =>
472 pipelines.filter(
473 (pipeline): pipeline is osis.Pipeline => pipeline !== undefined,
474 ),
475 ),
476 Effect.flatMap(
477 Effect.forEach((pipeline) => toAttrs(pipeline), {
478 concurrency: 4,
479 }),
480 ),
481 ),
482 };
483 }),
484 );