From 50b6bc9f636659e615aa824e6dd839143ae3973c Mon Sep 17 00:00:00 2001
From: Madeline Kusters <80541297+madeline-k@users.noreply.github.com>
Date: Wed, 28 Jul 2021 15:45:50 -0700
Subject: [PATCH] feat(kinesisfirehose-destinations): add support for prefixes
 in the S3 destination (#15552)

Closes #15551

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*
---
 .../lib/s3-bucket.ts                          | 22 +++++++++++++++++++
 .../test/integ.s3-bucket.expected.json        |  2 ++
 .../test/integ.s3-bucket.ts                   |  4 +++-
 .../@aws-cdk/aws-kinesisfirehose/README.md    | 13 +++++++++++
 4 files changed, 40 insertions(+), 1 deletion(-)

diff --git a/packages/@aws-cdk/aws-kinesisfirehose-destinations/lib/s3-bucket.ts b/packages/@aws-cdk/aws-kinesisfirehose-destinations/lib/s3-bucket.ts
index 809eed9f293dd..2dec865f55728 100644
--- a/packages/@aws-cdk/aws-kinesisfirehose-destinations/lib/s3-bucket.ts
+++ b/packages/@aws-cdk/aws-kinesisfirehose-destinations/lib/s3-bucket.ts
@@ -20,6 +20,26 @@ export interface S3BucketProps extends CommonDestinationProps {
    * @default - no compression is applied
    */
   readonly compression?: Compression;
+
+  /**
+   * A prefix that Kinesis Data Firehose evaluates and adds to failed records before writing them to S3.
+   *
+   * This prefix appears immediately following the bucket name.
+   * @see https://docs.aws.amazon.com/firehose/latest/dev/s3-prefixes.html
+   *
+   * @default "YYYY/MM/DD/HH"
+   */
+  readonly errorOutputPrefix?: string;
+
+  /**
+   * A prefix that Kinesis Data Firehose evaluates and adds to records before writing them to S3.
+   *
+   * This prefix appears immediately following the bucket name.
+   * @see https://docs.aws.amazon.com/firehose/latest/dev/s3-prefixes.html
+   *
+   * @default "YYYY/MM/DD/HH"
+   */
+  readonly dataOutputPrefix?: string;
 }
 
 /**
@@ -48,6 +68,8 @@ export class S3Bucket implements firehose.IDestination {
         roleArn: role.roleArn,
         bucketArn: this.bucket.bucketArn,
         compressionFormat: this.props.compression?.value,
+        errorOutputPrefix: this.props.errorOutputPrefix,
+        prefix: this.props.dataOutputPrefix,
       },
       dependables: [bucketGrant, ...(loggingDependables ?? [])],
     };
diff --git a/packages/@aws-cdk/aws-kinesisfirehose-destinations/test/integ.s3-bucket.expected.json b/packages/@aws-cdk/aws-kinesisfirehose-destinations/test/integ.s3-bucket.expected.json
index e5aaa3da4e40e..c058076f7760f 100644
--- a/packages/@aws-cdk/aws-kinesisfirehose-destinations/test/integ.s3-bucket.expected.json
+++ b/packages/@aws-cdk/aws-kinesisfirehose-destinations/test/integ.s3-bucket.expected.json
@@ -300,6 +300,8 @@
             }
           },
           "CompressionFormat": "GZIP",
+          "ErrorOutputPrefix": "errorPrefix",
+          "Prefix": "regularPrefix",
           "RoleARN": {
             "Fn::GetAtt": [
               "DeliveryStreamS3DestinationRole500FC089",
diff --git a/packages/@aws-cdk/aws-kinesisfirehose-destinations/test/integ.s3-bucket.ts b/packages/@aws-cdk/aws-kinesisfirehose-destinations/test/integ.s3-bucket.ts
index 545d293fdb264..1aa90595853f0 100644
--- a/packages/@aws-cdk/aws-kinesisfirehose-destinations/test/integ.s3-bucket.ts
+++ b/packages/@aws-cdk/aws-kinesisfirehose-destinations/test/integ.s3-bucket.ts
@@ -23,7 +23,9 @@ new firehose.DeliveryStream(stack, 'Delivery Stream', {
     logging: true,
     logGroup: logGroup,
     compression: destinations.Compression.GZIP,
+    dataOutputPrefix: 'regularPrefix',
+    errorOutputPrefix: 'errorPrefix',
   })],
 });
 
-app.synth();
\ No newline at end of file
+app.synth();
diff --git a/packages/@aws-cdk/aws-kinesisfirehose/README.md b/packages/@aws-cdk/aws-kinesisfirehose/README.md
index 130008ac331fa..e4e2f2ad31d3a 100644
--- a/packages/@aws-cdk/aws-kinesisfirehose/README.md
+++ b/packages/@aws-cdk/aws-kinesisfirehose/README.md
@@ -109,6 +109,19 @@ new DeliveryStream(this, 'Delivery Stream', {
 });
 ```
 
+The S3 destination also supports custom dynamic prefixes. `prefix` will be used for files
+successfully delivered to S3. `errorOutputPrefix` will be added to failed records before
+writing them to S3.
+
+```ts fixture=with-bucket
+const s3Destination = new destinations.S3Bucket(bucket, {
+  dataOutputPrefix: 'myFirehose/DeliveredYear=!{timestamp:yyyy}/anyMonth/rand=!{firehose:random-string}',
+  errorOutputPrefix: 'myFirehoseFailures/!{firehose:error-output-type}/!{timestamp:yyyy}/anyMonth/!{timestamp:dd}',
+});
+```
+
+See: [Custom S3 Prefixes](https://docs.aws.amazon.com/firehose/latest/dev/s3-prefixes.html) in the *Kinesis Data Firehose Developer Guide*.
+
 ## Server-side Encryption
 
 Enabling server-side encryption (SSE) requires Kinesis Data Firehose to encrypt all data