feat(kinesisfirehose): add DeliveryStream with Redshift destination

packages/@aws-cdk/aws-ec2/lib/connections.ts

@@ -20,6 +20,9 @@ import { ISecurityGroup } from './security-group';
  * An object that has a Connections object
  */
 export interface IConnectable {
+  /**
+   * The connections associated with this resource.
+   */
   readonly connections: Connections;
 }
 

packages/@aws-cdk/aws-events-targets/lib/kinesis-firehose-stream.ts

@@ -43,4 +43,4 @@ export class KinesisFirehoseStream implements events.IRuleTarget {
       targetResource: this.stream,
     };
   }
-}
+}

packages/@aws-cdk/aws-kinesisfirehose/README.md

@@ -9,8 +9,44 @@
 >
 > [CFN Resources]: https://docs.aws.amazon.com/cdk/latest/guide/constructs.html#constructs_lib
 
+![cdk-constructs: Experimental](https://img.shields.io/badge/cdk--constructs-experimental-important.svg?style=for-the-badge)
+
+> The APIs of higher level constructs in this module are experimental and under active development.
+> They are subject to non-backward compatible changes or removal in any future version. These are
+> not subject to the [Semantic Versioning](https://semver.org/) model and breaking changes will be
+> announced in the release notes. This means that while you may use them, you may need to update
+> your source code when upgrading to a newer version of this package.
+
 ---
 
 <!--END STABILITY BANNER-->
 
-This module is part of the [AWS Cloud Development Kit](https://github.com/aws/aws-cdk) project.
+This module is part of the [AWS Cloud Development Kit](https://github.com/aws/aws-cdk) project. It allows you to create Kinesis Data Firehose delivery streams.
+
+```ts
+import * as ec2 from '@aws-cdk/aws-ec2';
+import * as redshift from '@aws-cdk/aws-redshift';
+
+// Given a Redshift cluster
+const vpc = new ec2.Vpc(this, 'Vpc');
+const database = 'my_db';
+const cluster = new redshift.Cluster(this, 'Cluster', {
+  vpc: vpc,
+  masterUser: {
+    masterUsername: 'master',
+  },
+  defaultDatabaseName: database,
+});
+
+// Create a delivery stream that publishes data to the cluster
+new DeliveryStream(this, 'Delivery Stream', {
+  destination: new RedshiftDestination({
+    cluster: cluster,
+    user: {
+      username: 'firehose',
+    },
+    database: database,
+    tableName: 'table',
+  }),
+});
+```

packages/@aws-cdk/aws-kinesisfirehose/lib/delivery-stream.ts

@@ -0,0 +1,246 @@
+import * as cloudwatch from '@aws-cdk/aws-cloudwatch';
+import * as ec2 from '@aws-cdk/aws-ec2';
+import * as iam from '@aws-cdk/aws-iam';
+import * as kinesis from '@aws-cdk/aws-kinesis';
+import * as kms from '@aws-cdk/aws-kms';
+import { CfnMapping, Fn, IResource, Resource, Stack } from '@aws-cdk/core';
+import { RegionInfo } from '@aws-cdk/region-info';
+import { Construct } from 'constructs';
+import { IDestination } from './destination';
+import { CfnDeliveryStream } from './kinesisfirehose.generated';
+
+/**
+ * Represents a Kinesis Data Firehose delivery stream.
+ */
+export interface IDeliveryStream extends IResource, iam.IGrantable, ec2.IConnectable {
+  /**
+   * Name of the delivery stream.
+   *
+   * @attribute
+   */
+  readonly deliveryStreamName: string;
+
+  /**
+   * ARN of the delivery stream.
+   *
+   * @attribute
+   */
+  readonly deliveryStreamArn: string;
+
+  /**
+   * Grant the given identity permissions to perform the given actions.
+   */
+  grant(grantee: iam.IGrantable, ...actions: string[]): iam.Grant;
+
+  /**
+   * Grant the given identity permissions to write data to this stream.
+   */
+  grantWrite(grantee: iam.IGrantable): iam.Grant;
+
+  /**
+   * Return the given named metric for this delivery stream
+   */
+  metric(metricName: string, props?: cloudwatch.MetricOptions): cloudwatch.Metric;
+}
+
+/**
+ * Options for server-side encryption of a delivery stream
+ */
+export enum StreamEncryption {
+  /**
+   * Data in the stream is stored unencrypted.
+   */
+  UNENCRYPTED,
+
+  /**
+   * Data in the stream is stored encrypted by a KMS key managed by the customer.
+   */
+  CUSTOMER_MANAGED,
+
+  /**
+   * Data in the stream is stored encrypted by a KMS key owned by AWS and managed for use in multiple AWS accounts.
+   */
+  AWS_OWNED
+}
+
+/**
+ * Properties for a new delivery stream
+ */
+export interface DeliveryStreamProps {
+  /**
+   * The destination that this delivery stream will deliver data to.
+   *
+   * TODO: figure out if multiple destinations are supported (API return value seems to indicate so) and convert this to a list
+   */
+  readonly destination: IDestination;
+
+  /**
+   * A name for the delivery stream.
+   *
+   * @default - a name is generated by CloudFormation.
+   */
+  readonly deliveryStreamName?: string;
+
+  /**
+   * The Kinesis data stream to use as a source for this delivery stream.
+   *
+   * @default - data is written to the delivery stream via a direct put.
+   */
+  readonly sourceStream?: kinesis.IStream;
+
+  /**
+   * The IAM role assumed by Kinesis Firehose to read from sources, invoke processors, and write to destinations
+   *
+   * @default - a role will be created with default permissions.
+   */
+  readonly role?: iam.IRole;
+
+  /**
+   * Indicates the type of customer master key (CMK) to use for server-side encryption, if any.
+   *
+   * If `encryptionKey` is provided, this will be implicitly set to `CUSTOMER_MANAGED`.
+   *
+   * @default StreamEncryption.UNENCRYPTED.
+   */
+  readonly encryption?: StreamEncryption;
+
+  /**
+   * Customer managed key to server-side encrypt data in the stream.
+   *
+   * @default - if `encryption` is set to `CUSTOMER_MANAGED`, a KMS key will be created for you.
+   */
+  readonly encryptionKey?: kms.IKey;
+
+  // TODO: tags?
+}
+
+/**
+ * Base class for new and imported Kinesis Data Firehose delivery streams
+ */
+export abstract class DeliveryStreamBase extends Resource implements IDeliveryStream {
+
+  abstract readonly deliveryStreamName: string;
+
+  abstract readonly deliveryStreamArn: string;
+
+  abstract readonly grantPrincipal: iam.IPrincipal;
+
+  abstract readonly connections: ec2.Connections;
+
+  public grant(grantee: iam.IGrantable, ...actions: string[]): iam.Grant {
+    return iam.Grant.addToPrincipal({
+      resourceArns: [this.deliveryStreamArn],
+      grantee: grantee,
+      actions: actions,
+    });
+  }
+
+  public grantWrite(grantee: iam.IGrantable): iam.Grant {
+    return this.grant(grantee, 'firehose:PutRecord', 'firehose:PutRecordBatch');
+  }
+
+  // TODO: use canned metrics
+  public metric(metricName: string, props?: cloudwatch.MetricOptions): cloudwatch.Metric {
+    return new cloudwatch.Metric({
+      namespace: 'AWS/Firehose',
+      metricName: metricName,
+      dimensions: {
+        DeliveryStreamName: this.deliveryStreamName,
+      },
+      ...props,
+    });
+  }
+}
+
+/**
+ * Create a Kinesis Data Firehose delivery stream
+ *
+ * @resource AWS::KinesisFirehose::DeliveryStream
+ */
+export class DeliveryStream extends DeliveryStreamBase {
+  /**
+   * Import an existing delivery stream from its name.
+   */
+  static fromDeliveryStreamName(scope: Construct, id: string, deliveryStreamName: string): IDeliveryStream {
+    class Import extends DeliveryStreamBase {
+      public readonly deliveryStreamName = deliveryStreamName;
+      public readonly deliveryStreamArn = Stack.of(scope).formatArn({
+        service: 'firehose',
+        resource: 'deliverystream',
+        resourceName: deliveryStreamName,
+      })
+      public readonly grantPrincipal = new iam.UnknownPrincipal({ resource: this });
+      public readonly connections = setConnections(this);
+    }
+    return new Import(scope, id);
+  }
+
+  readonly deliveryStreamName: string;
+
+  readonly deliveryStreamArn: string;
+
+  readonly grantPrincipal: iam.IPrincipal;
+
+  readonly connections: ec2.Connections;
+
+  constructor(scope: Construct, id: string, props: DeliveryStreamProps) {
+    super(scope, id);
+
+    const role = props.role ?? new iam.Role(this, 'Service Role', {
+      assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
+    });
+    this.grantPrincipal = role;
+
+    this.connections = setConnections(this);
+
+    const encryptionKey = props.encryptionKey ?? (props.encryption === StreamEncryption.CUSTOMER_MANAGED ? new kms.Key(this, 'Key') : undefined);
+    const encryptionConfig = (encryptionKey || (props.encryption === StreamEncryption.AWS_OWNED)) ? {
+      keyArn: encryptionKey?.keyArn,
+      keyType: encryptionKey ? 'CUSTOMER_MANAGED_CMK' : 'AWS_OWNED_CMK',
+    } : undefined;
+    // TODO: we definitely need to grant access to the role
+
+    props.sourceStream?.grantRead(role); // TODO: may need to be DescribeStream instead of DescribeStreamSummary
+    const streamSourceConfig = props.sourceStream ? {
+      kinesisStreamArn: props.sourceStream?.streamArn,
+      roleArn: role.roleArn,
+    } : undefined;
+
+    const destination = props.destination.bind(this, this);
+
+    const resource = new CfnDeliveryStream(this, 'Resource', {
+      deliveryStreamEncryptionConfigurationInput: encryptionConfig,
+      deliveryStreamName: props.deliveryStreamName,
+      deliveryStreamType: props.sourceStream ? 'KinesisStreamAsSource' : 'DirectPut',
+      kinesisStreamSourceConfiguration: streamSourceConfig,
+      ...destination.properties,
+    });
+
+    this.deliveryStreamName = resource.ref;
+    this.deliveryStreamArn = resource.attrArn;
+  }
+}
+
+function setConnections(scope: Construct) {
+  const region = Stack.of(scope).region;
+  let cidrBlock = RegionInfo.get(region).firehoseCidrBlock;
+  if (!cidrBlock) {
+    const mapping: {[region: string]: { FirehoseCidrBlock: string }} = {};
+    RegionInfo.regions.forEach((regionInfo) => {
+      if (regionInfo.firehoseCidrBlock) {
+        mapping[regionInfo.name] = {
+          FirehoseCidrBlock: regionInfo.firehoseCidrBlock,
+        };
+      }
+    });
+    const cfnMapping = new CfnMapping(scope, 'Firehose CIDR Mapping', {
+      mapping,
+    });
+    cidrBlock = Fn.findInMap(cfnMapping.logicalId, region, 'FirehoseCidrBlock');
+    // TODO: this fails deployment if the region isn't configured, is that acceptable?
+  }
+
+  return new ec2.Connections({
+    peer: ec2.Peer.ipv4(cidrBlock),
+  });
+}