CloudWatch Logs: new library

packages/@aws-cdk/core/lib/cloudformation/arn.ts

@@ -1,4 +1,5 @@
 import { AwsAccountId, AwsPartition, AwsRegion, FnConcat, Token } from '..';
+import { FnSelect, FnSplit } from '../cloudformation/fn';
 
 /**
  * An Amazon Resource Name (ARN).
@@ -122,6 +123,66 @@ export class Arn extends Token {
 
         return result;
     }
+
+    /**
+     * Given a Token evaluating to ARN, parses it and returns components.
+     *
+     * The ARN cannot be validated, since we don't have the actual value yet
+     * at the time of this function call. You will have to know the separator
+     * and the type of ARN.
+     *
+     * The resulting `ArnComponents` object will contain tokens for the
+     * subexpressions of the ARN, not string literals.
+     *
+     * WARNING: this function cannot properly parse the complete final
+     * resourceName (path) out of ARNs that use '/' to both separate the
+     * 'resource' from the 'resourceName' AND to subdivide the resourceName
+     * further. For example, in S3 ARNs:
+     *
+     *      arn:aws:s3:::my_corporate_bucket/path/to/exampleobject.png
+     *
+     * After parsing the resourceName will not contain 'path/to/exampleobject.png'
+     * but simply 'path'. This is a limitation because there is no slicing
+     * functionality in CloudFormation templates.
+     *
+     * @param arn The input token that contains an ARN
+     * @param sep The separator used to separate resource from resourceName
+     * @param hasName Whether there is a name component in the ARN at all.
+     * For example, SNS Topics ARNs have the 'resource' component contain the
+     * topic name, and no 'resourceName' component.
+     * @returns an ArnComponents object which allows access to the various
+     * components of the ARN.
+     */
+    public static parseToken(arn: Token, sep: string = '/', hasName: boolean = true): ArnComponents {
+        // Arn ARN looks like:
+        // arn:partition:service:region:account-id:resource
+        // arn:partition:service:region:account-id:resourcetype/resource
+        // arn:partition:service:region:account-id:resourcetype:resource
+
+        // We need the 'hasName' argument because {Fn::Select}ing a nonexistent field
+        // throws an error.
+
+        const components = new FnSplit(':', arn);
+
+        const partition = new FnSelect(1, components);
+        const service = new FnSelect(2, components);
+        const region = new FnSelect(3, components);
+        const account = new FnSelect(4, components);
+
+        if (sep === ':') {
+            const resource = new FnSelect(5, components);
+            const resourceName = hasName ? new FnSelect(6, components) : undefined;
+
+            return { partition, service, region, account, resource, resourceName, sep };
+        } else {
+            const lastComponents = new FnSplit(sep, new FnSelect(5, components));
+
+            const resource = new FnSelect(0, lastComponents);
+            const resourceName = hasName ? new FnSelect(1, lastComponents) : undefined;
+
+            return { partition, service, region, account, resource, resourceName, sep };
+        }
+    }
 }
 
 export interface ArnComponents {
@@ -133,21 +194,21 @@ export interface ArnComponents {
      *
      * @default The AWS partition the stack is deployed to.
      */
-    partition?: string;
+    partition?: any;
 
     /**
      * The service namespace that identifies the AWS product (for example,
      * 's3', 'iam', 'codepipline').
      */
-    service: string;
+    service: any;
 
     /**
      * The region the resource resides in. Note that the ARNs for some resources
      * do not require a region, so this component might be omitted.
      *
      * @default The region the stack is deployed to.
      */
-    region?: string;
+    region?: any;
 
     /**
      * The ID of the AWS account that owns the resource, without the hyphens.

packages/@aws-cdk/core/test/cloudformation/test.arn.ts

@@ -1,5 +1,5 @@
 import { Test } from 'nodeunit';
-import { Arn, ArnComponents, resolve } from '../../lib';
+import { Arn, ArnComponents, resolve, Token } from '../../lib';
 
 export = {
     'create from components with defaults'(test: Test) {
@@ -187,7 +187,36 @@ export = {
             });
 
             test.done();
-        }
+        },
+
+        'a Token with : separator'(test: Test) {
+            const theToken = { Ref: 'SomeParameter' };
+            const parsed = Arn.parseToken(new Token(() => theToken), ':');
+
+            test.deepEqual(resolve(parsed.partition), { 'Fn::Select': [ 1, { 'Fn::Split': [ ':', theToken ]} ]});
+            test.deepEqual(resolve(parsed.service), { 'Fn::Select': [ 2, { 'Fn::Split': [ ':', theToken ]} ]});
+            test.deepEqual(resolve(parsed.region), { 'Fn::Select': [ 3, { 'Fn::Split': [ ':', theToken ]} ]});
+            test.deepEqual(resolve(parsed.account), { 'Fn::Select': [ 4, { 'Fn::Split': [ ':', theToken ]} ]});
+            test.deepEqual(resolve(parsed.resource), { 'Fn::Select': [ 5, { 'Fn::Split': [ ':', theToken ]} ]});
+            test.deepEqual(resolve(parsed.resourceName), { 'Fn::Select': [ 6, { 'Fn::Split': [ ':', theToken ]} ]});
+            test.equal(parsed.sep, ':');
+
+            test.done();
+        },
 
+        'a Token with / separator'(test: Test) {
+            const theToken = { Ref: 'SomeParameter' };
+            const parsed = Arn.parseToken(new Token(() => theToken));
+
+            test.equal(parsed.sep, '/');
+
+            // tslint:disable-next-line:max-line-length
+            test.deepEqual(resolve(parsed.resource), { 'Fn::Select': [ 0, { 'Fn::Split': [ '/', { 'Fn::Select': [ 5, { 'Fn::Split': [ ':', theToken ]} ]} ]} ]});
+            // tslint:disable-next-line:max-line-length
+            test.deepEqual(resolve(parsed.resourceName), { 'Fn::Select': [ 1, { 'Fn::Split': [ '/', { 'Fn::Select': [ 5, { 'Fn::Split': [ ':', theToken ]} ]} ]} ]});
+
+            test.done();
+        }
     },
-};
+
+};

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

@@ -1,6 +1,8 @@
-import { Construct, Output, PolicyStatement, Token } from '@aws-cdk/core';
-import { IIdentityResource } from '@aws-cdk/iam';
+import { Arn, AwsRegion, Construct, FnConcat, HashedAddressingScheme, 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 +39,7 @@ export interface StreamRefProps {
  *     StreamRef.import(this, 'MyImportedStream', ref);
  *
  */
-export abstract class StreamRef extends Construct {
+export abstract class StreamRef extends Construct implements logs.ILogSubscriptionDestination {
     /**
      * Creates a Stream construct that represents an external stream.
      *
@@ -55,11 +57,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 +171,62 @@ export abstract class StreamRef extends Construct {
         );
     }
 
+    public logSubscriptionDestination(sourceLogGroup: logs.LogGroup): logs.LogSubscriptionDestination {
+        // 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 FnConcat('logs.', new AwsRegion(), '.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 this.crossAccountLogSubscriptionDestination(sourceLogGroup);
+        }
+
+        return { arn: this.streamArn, role: this.cloudWatchLogsRole };
+    }
+
+    /**
+     * Generate a CloudWatch Logs Destination and return the properties in the form o a subscription destination
+     */
+    private crossAccountLogSubscriptionDestination(sourceLogGroup: logs.LogGroup): logs.LogSubscriptionDestination {
+        const sourceStack = Stack.find(sourceLogGroup);
+        const thisStack = Stack.find(this);
+
+        if (!sourceStack.env.account || !thisStack.env.account) {
+            throw new Error('SubscriptionFilter stack and Destination stack must either both have accounts defined, or both not have accounts');
+        }
+
+        // Take some effort to construct a unique ID for the destination that is unique to the
+        // combination of (stream, loggroup).
+        const uniqueId =  new HashedAddressingScheme().allocateAddress([sourceLogGroup.path.replace('/', ''), sourceStack.env.account!]);
+
+        // The destination lives in the target account
+        const dest = new logs.CrossAccountDestination(this, `CWLDestination${uniqueId}`, {
+            targetArn: this.streamArn,
+            role: this.cloudWatchLogsRole!
+        });
+
+        dest.addToPolicy(new PolicyStatement()
+            .addAction('logs:PutSubscriptionFilter')
+            .addAwsAccountPrincipal(sourceStack.env.account)
+            .addAllResources());
+
+        return dest.logSubscriptionDestination(sourceLogGroup);
+    }
+
     private grant(identity: IIdentityResource, actions: { streamActions: string[], keyActions: string[] }) {
         identity.addToPolicy(new PolicyStatement()
             .addResource(this.streamArn)
@@ -307,12 +375,16 @@ 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;
+        // Get the name from the ARN
+        this.streamName = Arn.parseToken(props.streamArn).resourceName;
+
         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::Join": ["", ["logs.", {Ref: "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();
+    }
+};