feat(kinesisfirehose-destinations): add support for prefixes in the S3 destination

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 ?? [])],
     };

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",

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();
+app.synth();

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