CloudWatch Logs: new library

packages/@aws-cdk/kinesis/lib/stream.ts

@@ -1,6 +1,7 @@
-import { Construct, Output, PolicyStatement, Token } from '@aws-cdk/core';
-import { IIdentityResource } from '@aws-cdk/iam';
+import { Construct, FnConcat, FnSelect, FnSplit, FnSub, Output, PolicyStatement, ServicePrincipal, Stack, Token } from '@aws-cdk/core';
+import { IIdentityResource, Role } from '@aws-cdk/iam';
 import * as kms from '@aws-cdk/kms';
+import logs = require('@aws-cdk/logs');
 import { cloudformation, StreamArn } from './kinesis.generated';
 
 /**
@@ -37,7 +38,7 @@ export interface StreamRefProps {
  *     StreamRef.import(this, 'MyImportedStream', ref);
  *
  */
-export abstract class StreamRef extends Construct {
+export abstract class StreamRef extends Construct implements logs.ISubscriptionDestination {
     /**
      * Creates a Stream construct that represents an external stream.
      *
@@ -55,11 +56,21 @@ export abstract class StreamRef extends Construct {
      */
     public abstract readonly streamArn: StreamArn;
 
+    /**
+     * The name of the stream
+     */
+    public abstract readonly streamName: StreamName;
+
     /**
      * Optional KMS encryption key associated with this stream.
      */
     public abstract readonly encryptionKey?: kms.EncryptionKeyRef;
 
+    /**
+     * The role that can be used by CloudWatch logs to write to this stream
+     */
+    private cloudWatchLogsRole?: Role;
+
     /**
      * Exports this stream from the stack.
      */
@@ -159,6 +170,45 @@ export abstract class StreamRef extends Construct {
         );
     }
 
+    public subscriptionDestination(sourceLogGroup: logs.LogGroup): logs.SubscriptionDestination {
+        // Following example from https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html#DestinationKinesisExample
+        if (!this.cloudWatchLogsRole) {
+            // Create a role to be assumed by CWL that can write to this stream and pass itself.
+            this.cloudWatchLogsRole = new Role(this, 'CloudWatchLogsCanPutRecords', {
+                assumedBy: new ServicePrincipal(new FnSub('logs.${AWS::Region}.amazonaws.com')),
+            });
+            this.cloudWatchLogsRole.addToPolicy(new PolicyStatement().addAction('kinesis:PutRecord').addResource(this.streamArn));
+            this.cloudWatchLogsRole.addToPolicy(new PolicyStatement().addAction('iam:PassRole').addResource(this.cloudWatchLogsRole.roleArn));
+        }
+
+        // We've now made it possible for CloudWatch events to write to us. In case the LogGroup is in a
+        // different account, we must add a Destination in between as well.
+        const sourceStack = Stack.find(sourceLogGroup);
+        const thisStack = Stack.find(this);
+
+        // Case considered: if both accounts are undefined, we can't make any assumptions. Better
+        // to assume we don't need to do anything special.
+        const sameAccount = sourceStack.env.account === thisStack.env.account;
+
+        if (sameAccount) {
+            return { arn: this.streamArn, role: this.cloudWatchLogsRole };
+        }
+
+        // The destination lives in the target account
+        const dest = new logs.CrossAccountDestination(this, 'CloudWatchCrossAccountDestination', {
+            // Unfortunately destinationName is required so we have to invent one that won't conflict.
+            destinationName: new FnConcat(sourceLogGroup.logGroupName, 'To', this.streamName) as any,
+            targetArn: this.streamArn,
+            role: this.cloudWatchLogsRole
+        });
+        dest.addToPolicy(new PolicyStatement()
+            .addAction('logs:PutSubscriptionFilter')
+            .addAwsAccountPrincipal(sourceStack.env.account)
+            .addAllResources());
+
+        return dest.subscriptionDestination(sourceLogGroup);
+    }
+
     private grant(identity: IIdentityResource, actions: { streamActions: string[], keyActions: string[] }) {
         identity.addToPolicy(new PolicyStatement()
             .addResource(this.streamArn)
@@ -307,12 +357,17 @@ export class StreamName extends Token {}
 
 class ImportedStreamRef extends StreamRef {
     public readonly streamArn: StreamArn;
+    public readonly streamName: StreamName;
     public readonly encryptionKey?: kms.EncryptionKeyRef;
 
     constructor(parent: Construct, name: string, props: StreamRefProps) {
         super(parent, name);
 
         this.streamArn = props.streamArn;
+        // ARN always looks like: arn:aws:kinesis:us-east-2:123456789012:stream/mystream
+        // so we can get the name from the ARN.
+        this.streamName = new FnSelect(1, new FnSplit('/', this.streamArn));
+
         if (props.encryptionKey) {
             this.encryptionKey = kms.EncryptionKeyRef.import(parent, 'Key', props.encryptionKey);
         } else {

packages/@aws-cdk/kinesis/package.json

@@ -41,6 +41,7 @@
   "dependencies": {
     "@aws-cdk/core": "^0.7.3-beta",
     "@aws-cdk/iam": "^0.7.3-beta",
-    "@aws-cdk/kms": "^0.7.3-beta"
+    "@aws-cdk/kms": "^0.7.3-beta",
+    "@aws-cdk/logs": "^0.7.3-beta"
   }
 }

packages/@aws-cdk/kinesis/test/test.subscriptiondestination.ts

@@ -0,0 +1,80 @@
+import { expect, haveResource } from '@aws-cdk/assert';
+import { Stack } from '@aws-cdk/core';
+import { FilterPattern, LogGroup, SubscriptionFilter } from '@aws-cdk/logs';
+import { Test } from 'nodeunit';
+import { Stream } from '../lib';
+
+export = {
+    'stream can be subscription destination'(test: Test) {
+        // GIVEN
+        const stack = new Stack();
+        const stream = new Stream(stack, 'MyStream');
+        const logGroup = new LogGroup(stack, 'LogGroup');
+
+        // WHEN
+        new SubscriptionFilter(stack, 'Subscription', {
+            logGroup,
+            destination: stream,
+            filterPattern: FilterPattern.allEvents()
+        });
+
+        // THEN: subscription target is Stream
+        expect(stack).to(haveResource('AWS::Logs::SubscriptionFilter', {
+            DestinationArn: { "Fn::GetAtt": [ "MyStream5C050E93", "Arn" ] },
+            RoleArn: { "Fn::GetAtt": [ "MyStreamCloudWatchLogsCanPutRecords58498490", "Arn" ] },
+        }));
+
+        // THEN: we have a role to write to the Lambda
+        expect(stack).to(haveResource('AWS::IAM::Role', {
+            AssumeRolePolicyDocument: {
+                Statement: [{
+                    Action: "sts:AssumeRole",
+                    Principal: { Service: { "Fn::Sub": "logs.${AWS::Region}.amazonaws.com" }}
+                }],
+            }
+        }));
+
+        expect(stack).to(haveResource('AWS::IAM::Policy', {
+            PolicyDocument: {
+                Statement: [
+                    {
+                        Action: "kinesis:PutRecord",
+                        Effect: "Allow",
+                        Resource: { "Fn::GetAtt": [ "MyStream5C050E93", "Arn" ] }
+                    },
+                    {
+                        Action: "iam:PassRole",
+                        Effect: "Allow",
+                        Resource: { "Fn::GetAtt": [ "MyStreamCloudWatchLogsCanPutRecords58498490", "Arn" ] }
+                    }
+                ],
+            }
+        }));
+
+        test.done();
+    },
+
+    'cross-account stream can be subscription destination with Destination'(test: Test) {
+        // GIVEN
+        const sourceStack = new Stack(undefined, undefined, { env: { account: '12345' }});
+        const logGroup = new LogGroup(sourceStack, 'LogGroup');
+
+        const destStack = new Stack(undefined, undefined, { env: { account: '67890' }});
+        const stream = new Stream(destStack, 'MyStream');
+
+        // WHEN
+        new SubscriptionFilter(sourceStack, 'Subscription', {
+            logGroup,
+            destination: stream,
+            filterPattern: FilterPattern.allEvents()
+        });
+
+        // THEN: the source stack has a Destination object that the subscription points to
+        expect(destStack).to(haveResource('AWS::Logs::Destination', {
+            TargetArn: { "Fn::GetAtt": [ "MyStream5C050E93", "Arn" ] },
+            RoleArn: { "Fn::GetAtt": [ "MyStreamCloudWatchLogsCanPutRecords58498490", "Arn" ] },
+        }));
+
+        test.done();
+    }
+};

packages/@aws-cdk/lambda/lib/lambda-ref.ts

@@ -1,8 +1,9 @@
 import { Metric, MetricCustomization } from '@aws-cdk/cloudwatch';
-import { AccountPrincipal, Arn, Construct, FnSelect, FnSplit, PolicyPrincipal,
-         PolicyStatement, resolve, ServicePrincipal, Token } from '@aws-cdk/core';
+import { AccountPrincipal, Arn, Construct, FnSelect, FnSplit, FnSub,
+         PolicyPrincipal, PolicyStatement, resolve, ServicePrincipal, Token } from '@aws-cdk/core';
 import { EventRuleTarget, IEventRuleTarget } from '@aws-cdk/events';
 import { Role } from '@aws-cdk/iam';
+import logs = require('@aws-cdk/logs');
 import { cloudformation, FunctionArn } from './lambda.generated';
 import { LambdaPermission } from './permission';
 
@@ -23,7 +24,7 @@ export interface LambdaRefProps {
     role?: Role;
 }
 
-export abstract class LambdaRef extends Construct implements IEventRuleTarget {
+export abstract class LambdaRef extends Construct implements IEventRuleTarget, logs.ISubscriptionDestination {
     /**
      * Creates a Lambda function object which represents a function not defined
      * within this stack.
@@ -113,6 +114,11 @@ export abstract class LambdaRef extends Construct implements IEventRuleTarget {
      */
     private eventRuleTargetPolicyAdded = false;
 
+    /**
+     * Indicates if the policy that allows CloudWatch logs to publish to this topic has been added.
+     */
+    private logSubscriptionDestinationPolicyAddedFor: logs.LogGroupArn[] = [];
+
     /**
      * Adds a permission to the Lambda resource policy.
      * @param name A name for the permission construct
@@ -212,6 +218,23 @@ export abstract class LambdaRef extends Construct implements IEventRuleTarget {
         return this.metric('Throttles', { statistic: 'sum', ...props });
     }
 
+    public subscriptionDestination(sourceLogGroup: logs.LogGroup): logs.SubscriptionDestination {
+        const arn = sourceLogGroup.logGroupArn;
+
+        if (this.logSubscriptionDestinationPolicyAddedFor.indexOf(arn) === -1) {
+            // NOTE: the use of {AWS::Region} limits this to the same region, which shouldn't really be an issue,
+            // since the Lambda must be in the same region as the SubscriptionFilter anyway.
+            //
+            // (Wildcards in principals are unfortunately not supported.
+            this.addPermission('InvokedByCloudWatchLogs', {
+                principal: new ServicePrincipal(new FnSub('logs.${AWS::Region}.amazonaws.com')),
+                sourceArn: arn
+            });
+            this.logSubscriptionDestinationPolicyAddedFor.push(arn);
+        }
+        return { arn: this.functionArn };
+    }
+
     private parsePermissionPrincipal(principal?: PolicyPrincipal) {
         if (!principal) {
             return undefined;

packages/@aws-cdk/lambda/package.json

@@ -44,6 +44,7 @@
     "@aws-cdk/core": "^0.7.3-beta",
     "@aws-cdk/events": "^0.7.3-beta",
     "@aws-cdk/iam": "^0.7.3-beta",
-    "@aws-cdk/s3": "^0.7.3-beta"
+    "@aws-cdk/s3": "^0.7.3-beta",
+    "@aws-cdk/logs": "^0.7.3-beta"
   }
 }

packages/@aws-cdk/lambda/test/test.subscriptiondestination.ts

@@ -0,0 +1,39 @@
+import { expect, haveResource } from '@aws-cdk/assert';
+import { Stack } from '@aws-cdk/core';
+import { FilterPattern, LogGroup, SubscriptionFilter } from '@aws-cdk/logs';
+import { Test } from 'nodeunit';
+import { Lambda, LambdaInlineCode, LambdaRuntime } from '../lib';
+
+export = {
+    'lambda can be used as metric subscription destination'(test: Test) {
+        // GIVEN
+        const stack = new Stack();
+        const lambda = new Lambda(stack, 'MyLambda', {
+            code: new LambdaInlineCode('foo'),
+            handler: 'index.handler',
+            runtime: LambdaRuntime.NodeJS610,
+        });
+        const logGroup = new LogGroup(stack, 'LogGroup');
+
+        // WHEN
+        new SubscriptionFilter(stack, 'Subscription', {
+            logGroup,
+            destination: lambda,
+            filterPattern: FilterPattern.allEvents()
+        });
+
+        // THEN: subscription target is Lambda
+        expect(stack).to(haveResource('AWS::Logs::SubscriptionFilter', {
+            DestinationArn: { "Fn::GetAtt": [ "MyLambdaCCE802FB", "Arn" ] },
+        }));
+
+        // THEN: Lambda has permissions to be invoked by CWL
+        expect(stack).to(haveResource('AWS::Lambda::Permission', {
+            Action: "lambda:InvokeFunction",
+            FunctionName: { Ref: "MyLambdaCCE802FB" },
+            Principal: { "Fn::Sub": "logs.${AWS::Region}.amazonaws.com" }
+        }));
+
+        test.done();
+    }
+};

packages/@aws-cdk/logs/.gitignore

@@ -1,9 +1,10 @@
+*.d.ts
+*.generated.ts
 *.js
 *.js.map
-*.d.ts
+.jsii
+dist
+lib/generated/resources.ts
+node_modules
 tsconfig.json
 tslint.json
-node_modules
-*.generated.ts
-dist
-.jsii