feat(aws-logs-destinations): add Firehose Log Subscription Destination

packages/@aws-cdk/aws-logs-destinations/README.md

@@ -1,4 +1,4 @@
-# CDK Construct Libray for AWS XXX
+# CDK Construct Library for AWS CloudWatch Logs Subscription Destinations
 <!--BEGIN STABILITY BANNER-->
 
 ---
@@ -9,4 +9,51 @@
 
 <!--END STABILITY BANNER-->
 
-A short description here.
+This library supplies constructs for working with CloudWatch Logs Subscription Destinations.
+
+## Logs Destinations
+
+Log events matching a particular filter can be sent to Lambda, Kinesis or Kinesis Data Firehose.
+
+Logs that are sent to a receiving service through a subscription filter are Base64 encoded and compressed with the gzip format. For further information, see the [Using CloudWatch Logs Subscription Filters](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html).
+
+Create a `SubscriptionFilter`, initialize it with an appropriate `Pattern` and supply the intended destination.
+
+### Lambda Destination
+
+```ts
+const fn = new lambda.Function(this, 'Lambda', { ... });
+const logGroup = new LogGroup(this, 'LogGroup', { ... });
+
+new SubscriptionFilter(this, 'Subscription', {
+  logGroup,
+  destination: new LogsDestinations.LambdaDestination(fn),
+  filterPattern: FilterPattern.allTerms("ERROR", "MainThread")
+});
+```
+
+### Kinesis Destination
+
+```ts
+const stream = new kinesis.Stream(stack, 'KinesisStream', { ... });
+const logGroup = new LogGroup(this, 'LogGroup', { ... });
+
+new SubscriptionFilter(this, 'Subscription', {
+  logGroup,
+  destination: new LogsDestinations.KinesisDestination(stream),
+  filterPattern: FilterPattern.allTerms("ERROR", "MainThread")
+});
+```
+
+### Kinesis Firehose Destination
+
+```ts
+const deliveryStream = new firehose.CfnDeliveryStream(stack, 'Firehose', { ... });
+const logGroup = new LogGroup(this, 'LogGroup', { ... });
+
+new SubscriptionFilter(this, 'Subscription', {
+  logGroup,
+  destination: new LogsDestinations.KinesisFirehoseDestination(deliveryStream),
+  filterPattern: FilterPattern.allTerms("ERROR", "MainThread")
+});
+```

packages/@aws-cdk/aws-logs-destinations/lib/index.ts

@@ -1,2 +1,3 @@
 export * from './lambda';
 export * from './kinesis';
+export * from './kinesisfirehose';

packages/@aws-cdk/aws-logs-destinations/lib/kinesisfirehose.ts

@@ -0,0 +1,44 @@
+import * as iam from '@aws-cdk/aws-iam';
+import * as firehose from '@aws-cdk/aws-kinesisfirehose';
+import * as logs from '@aws-cdk/aws-logs';
+import { Stack } from '@aws-cdk/core';
+
+// keep this import separate from other imports to reduce chance for merge conflicts with v2-main
+// eslint-disable-next-line no-duplicate-imports, import/order
+import { Construct } from '@aws-cdk/core';
+
+/**
+ * Use a Kinesis Firehose as the destination for a log subscription
+ */
+export class KinesisFirehoseDestination implements logs.ILogSubscriptionDestination {
+  constructor(private readonly deliveryStream: firehose.CfnDeliveryStream) {
+  }
+
+  public bind(scope: Construct, _sourceLogGroup: logs.ILogGroup): logs.LogSubscriptionDestinationConfig {
+    const { region } = Stack.of(scope);
+
+    // Following example from https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html#FirehoseExample
+    // Create a role to be assumed by CWL that can write to this Firehose.
+    const roleId = 'CloudWatchLogsCanPutRecordsIntoKinesisFirehose';
+    const role = scope.node.tryFindChild(roleId) as iam.IRole ?? new iam.Role(scope, roleId, {
+      assumedBy: new iam.ServicePrincipal(`logs.${region}.amazonaws.com`),
+    });
+
+    iam.Grant.addToPrincipal({
+      scope,
+      grantee: role,
+      actions: [
+        'firehose:PutRecord',
+        'firehose:PutRecordBatch',
+      ],
+      resourceArns: [
+        this.deliveryStream.attrArn,
+      ],
+    });
+
+    return {
+      arn: this.deliveryStream.attrArn,
+      role,
+    };
+  }
+}

packages/@aws-cdk/aws-logs-destinations/lib/lambda.ts

@@ -4,7 +4,7 @@ import * as logs from '@aws-cdk/aws-logs';
 import { Construct } from '@aws-cdk/core';
 
 /**
- * Use a Lamda Function as the destination for a log subscription
+ * Use a Lambda Function as the destination for a log subscription
  */
 export class LambdaDestination implements logs.ILogSubscriptionDestination {
   constructor(private readonly fn: lambda.IFunction) {

packages/@aws-cdk/aws-logs-destinations/package.json

@@ -68,11 +68,13 @@
     "cfn2ts": "0.0.0",
     "jest": "^26.6.3",
     "pkglint": "0.0.0",
-    "@aws-cdk/assert-internal": "0.0.0"
+    "@aws-cdk/assert-internal": "0.0.0",
+    "@aws-cdk/aws-s3": "0.0.0"
   },
   "dependencies": {
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-kinesis": "0.0.0",
+    "@aws-cdk/aws-kinesisfirehose": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",
     "@aws-cdk/aws-logs": "0.0.0",
     "@aws-cdk/core": "0.0.0",
@@ -82,6 +84,7 @@
   "peerDependencies": {
     "@aws-cdk/aws-iam": "0.0.0",
     "@aws-cdk/aws-kinesis": "0.0.0",
+    "@aws-cdk/aws-kinesisfirehose": "0.0.0",
     "@aws-cdk/aws-lambda": "0.0.0",
     "@aws-cdk/aws-logs": "0.0.0",
     "@aws-cdk/core": "0.0.0",

packages/@aws-cdk/aws-logs-destinations/test/integ.kinesisfirehose-to-s3.expected.json

@@ -0,0 +1,174 @@
+{
+  "Resources": {
+    "Bucket83908E77": {
+      "Type": "AWS::S3::Bucket",
+      "UpdateReplacePolicy": "Delete",
+      "DeletionPolicy": "Delete"
+    },
+    "FirehoseRoleAA67C190": {
+      "Type": "AWS::IAM::Role",
+      "Properties": {
+        "AssumeRolePolicyDocument": {
+          "Statement": [
+            {
+              "Action": "sts:AssumeRole",
+              "Effect": "Allow",
+              "Principal": {
+                "Service": "firehose.amazonaws.com"
+              }
+            }
+          ],
+          "Version": "2012-10-17"
+        },
+        "Policies": [
+          {
+            "PolicyDocument": {
+              "Statement": [
+                {
+                  "Action": [
+                    "s3:AbortMultipartUpload",
+                    "s3:GetBucketLocation",
+                    "s3:GetObject",
+                    "s3:ListBucket",
+                    "s3:ListBucketMultipartUploads",
+                    "s3:PutObject"
+                  ],
+                  "Effect": "Allow",
+                  "Resource": [
+                    {
+                      "Fn::GetAtt": [
+                        "Bucket83908E77",
+                        "Arn"
+                      ]
+                    },
+                    {
+                      "Fn::Join": [
+                        "",
+                        [
+                          {
+                            "Fn::GetAtt": [
+                              "Bucket83908E77",
+                              "Arn"
+                            ]
+                          },
+                          "/*"
+                        ]
+                      ]
+                    }
+                  ]
+                }
+              ],
+              "Version": "2012-10-17"
+            },
+            "PolicyName": "allowToDeliverToS3Bucket"
+          }
+        ]
+      }
+    },
+    "MyFirehose": {
+      "Type": "AWS::KinesisFirehose::DeliveryStream",
+      "Properties": {
+        "ExtendedS3DestinationConfiguration": {
+          "BucketARN": {
+            "Fn::GetAtt": [
+              "Bucket83908E77",
+              "Arn"
+            ]
+          },
+          "RoleARN": {
+            "Fn::GetAtt": [
+              "FirehoseRoleAA67C190",
+              "Arn"
+            ]
+          }
+        }
+      }
+    },
+    "LogGroupF5B46931": {
+      "Type": "AWS::Logs::LogGroup",
+      "Properties": {
+        "RetentionInDays": 731
+      },
+      "UpdateReplacePolicy": "Delete",
+      "DeletionPolicy": "Delete"
+    },
+    "SubscriptionCloudWatchLogsCanPutRecordsIntoKinesisFirehoseC45D3D89": {
+      "Type": "AWS::IAM::Role",
+      "Properties": {
+        "AssumeRolePolicyDocument": {
+          "Statement": [
+            {
+              "Action": "sts:AssumeRole",
+              "Effect": "Allow",
+              "Principal": {
+                "Service": {
+                  "Fn::Join": [
+                    "",
+                    [
+                      "logs.",
+                      {
+                        "Ref": "AWS::Region"
+                      },
+                      ".amazonaws.com"
+                    ]
+                  ]
+                }
+              }
+            }
+          ],
+          "Version": "2012-10-17"
+        }
+      }
+    },
+    "SubscriptionCloudWatchLogsCanPutRecordsIntoKinesisFirehoseDefaultPolicy8571150D": {
+      "Type": "AWS::IAM::Policy",
+      "Properties": {
+        "PolicyDocument": {
+          "Statement": [
+            {
+              "Action": [
+                "firehose:PutRecord",
+                "firehose:PutRecordBatch"
+              ],
+              "Effect": "Allow",
+              "Resource": {
+                "Fn::GetAtt": [
+                  "MyFirehose",
+                  "Arn"
+                ]
+              }
+            }
+          ],
+          "Version": "2012-10-17"
+        },
+        "PolicyName": "SubscriptionCloudWatchLogsCanPutRecordsIntoKinesisFirehoseDefaultPolicy8571150D",
+        "Roles": [
+          {
+            "Ref": "SubscriptionCloudWatchLogsCanPutRecordsIntoKinesisFirehoseC45D3D89"
+          }
+        ]
+      }
+    },
+    "Subscription391C9821": {
+      "Type": "AWS::Logs::SubscriptionFilter",
+      "Properties": {
+        "DestinationArn": {
+          "Fn::GetAtt": [
+            "MyFirehose",
+            "Arn"
+          ]
+        },
+        "FilterPattern": "",
+        "LogGroupName": {
+          "Ref": "LogGroupF5B46931"
+        },
+        "RoleArn": {
+          "Fn::GetAtt": [
+            "SubscriptionCloudWatchLogsCanPutRecordsIntoKinesisFirehoseC45D3D89",
+            "Arn"
+          ]
+        }
+      }
+    }
+  }
+}

packages/@aws-cdk/aws-logs-destinations/test/integ.kinesisfirehose-to-s3.ts

@@ -0,0 +1,70 @@
+#!/usr/bin/env node
+import * as iam from '@aws-cdk/aws-iam';
+import * as firehose from '@aws-cdk/aws-kinesisfirehose';
+import * as logs from '@aws-cdk/aws-logs';
+import * as s3 from '@aws-cdk/aws-s3';
+import * as cdk from '@aws-cdk/core';
+import { Construct } from 'constructs';
+import * as dests from '../lib';
+
+/*
+ * Stack verification steps:
+ * 1. `aws logs create-log-stream --log-group-name <log-group-name> --log-stream-name MyIntegTestLogStream`
+ * 2. `aws logs put-log-events --log-group-name <log-group-name> --log-stream-name MyIntegTestLogStream --log-events timestamp=$(date +%s000),message=hello`
+ * 3. Wait a few minutes for Kinesis to deliver log to S3
+ * 4. `aws s3 ls s3://<bucket-name> --recursive` should list one object
+ */
+
+class TestStack extends cdk.Stack {
+  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
+    super(scope, id, props);
+
+    // Create Firehose Stream that will deliver to S3 bucket
+    const bucket = new s3.Bucket(this, 'Bucket', {
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+    });
+
+    // Following example from https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html#FirehoseExample
+    const firehoseRole = new iam.Role(this, 'FirehoseRole', {
+      assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
+      inlinePolicies: {
+        allowToDeliverToS3Bucket: new iam.PolicyDocument({
+          statements: [new iam.PolicyStatement({
+            actions: [
+              's3:AbortMultipartUpload',
+              's3:GetBucketLocation',
+              's3:GetObject',
+              's3:ListBucket',
+              's3:ListBucketMultipartUploads',
+              's3:PutObject',
+            ],
+            resources: [
+              bucket.bucketArn,
+              `${bucket.bucketArn}/*`,
+            ],
+          })],
+        }),
+      },
+    });
+
+    const deliveryStream = new firehose.CfnDeliveryStream(this, 'MyFirehose', {
+      extendedS3DestinationConfiguration: {
+        bucketArn: bucket.bucketArn,
+        roleArn: firehoseRole.roleArn,
+      },
+    });
+    const logGroup = new logs.LogGroup(this, 'LogGroup', {
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+    });
+
+    new logs.SubscriptionFilter(this, 'Subscription', {
+      logGroup,
+      destination: new dests.KinesisFirehoseDestination(deliveryStream),
+      filterPattern: logs.FilterPattern.allEvents(),
+    });
+  }
+}
+
+const app = new cdk.App();
+new TestStack(app, 'cdk-logs-destinations-firehose-to-s3');
+app.synth();