Addressing more review comments

- Rename SubscriptionDestionation->LogSubscriptionDestination.
- Introducing Arn.parseToken()
- Use FnConcat instead of FnSub to build a region-aware service principal
- Add an additional safeguard in the cross-account subscription generation
  for if one account is unset.

Don't forget to mention this fixes #174.

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

@@ -1,4 +1,6 @@
 import { AwsAccountId, AwsPartition, AwsRegion, FnConcat, Token } from '..';
+import { FnSplit } from '../../../iam/node_modules/@aws-cdk/core/lib/cloudformation/fn';
+import { FnSelect } from '../../../iam/node_modules/@aws-cdk/core/lib/cloudformation/fn';
 
 /**
  * An Amazon Resource Name (ARN).
@@ -122,6 +124,64 @@ 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.
+     * @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 +193,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,4 +1,5 @@
-import { Construct, FnConcat, FnSelect, FnSplit, FnSub, Output, PolicyStatement, ServicePrincipal, Stack, Token } from '@aws-cdk/core';
+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');
@@ -38,7 +39,7 @@ export interface StreamRefProps {
  *     StreamRef.import(this, 'MyImportedStream', ref);
  *
  */
-export abstract class StreamRef extends Construct implements logs.ISubscriptionDestination {
+export abstract class StreamRef extends Construct implements logs.ILogSubscriptionDestination {
     /**
      * Creates a Stream construct that represents an external stream.
      *
@@ -170,12 +171,12 @@ export abstract class StreamRef extends Construct implements logs.ISubscriptionD
         );
     }
 
-    public subscriptionDestination(sourceLogGroup: logs.LogGroup): logs.SubscriptionDestination {
+    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 FnSub('logs.${AWS::Region}.amazonaws.com')),
+                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));
@@ -194,19 +195,35 @@ export abstract class StreamRef extends Construct implements logs.ISubscriptionD
             return { arn: this.streamArn, role: this.cloudWatchLogsRole };
         }
 
+        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');
+        }
+
+        return this.crossAccountLogSubscriptionDestination(sourceLogGroup);
+    }
+
+    /**
+     * 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);
+
+        // 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, '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,
+        const dest = new logs.CrossAccountDestination(this, `CWLDestination${uniqueId}`, {
             targetArn: this.streamArn,
-            role: this.cloudWatchLogsRole
+            role: this.cloudWatchLogsRole!
         });
+
         dest.addToPolicy(new PolicyStatement()
             .addAction('logs:PutSubscriptionFilter')
             .addAwsAccountPrincipal(sourceStack.env.account)
             .addAllResources());
 
-        return dest.subscriptionDestination(sourceLogGroup);
+        return dest.logSubscriptionDestination(sourceLogGroup);
     }
 
     private grant(identity: IIdentityResource, actions: { streamActions: string[], keyActions: string[] }) {
@@ -364,9 +381,8 @@ class ImportedStreamRef extends StreamRef {
         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));
+        // 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);

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

@@ -29,7 +29,7 @@ export = {
             AssumeRolePolicyDocument: {
                 Statement: [{
                     Action: "sts:AssumeRole",
-                    Principal: { Service: { "Fn::Sub": "logs.${AWS::Region}.amazonaws.com" }}
+                    Principal: { Service: { "Fn::Join": ["", ["logs.", {Ref: "AWS::Region"}, ".amazonaws.com"]] }}
                 }],
             }
         }));

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

@@ -1,5 +1,5 @@
 import { Metric, MetricCustomization } from '@aws-cdk/cloudwatch';
-import { AccountPrincipal, Arn, Construct, FnSelect, FnSplit, FnSub,
+import { AccountPrincipal, Arn, AwsRegion, Construct, FnConcat, FnSelect, FnSplit,
          PolicyPrincipal, PolicyStatement, resolve, ServicePrincipal, Token } from '@aws-cdk/core';
 import { EventRuleTarget, IEventRuleTarget } from '@aws-cdk/events';
 import { Role } from '@aws-cdk/iam';
@@ -24,7 +24,7 @@ export interface LambdaRefProps {
     role?: Role;
 }
 
-export abstract class LambdaRef extends Construct implements IEventRuleTarget, logs.ISubscriptionDestination {
+export abstract class LambdaRef extends Construct implements IEventRuleTarget, logs.ILogSubscriptionDestination {
     /**
      * Creates a Lambda function object which represents a function not defined
      * within this stack.
@@ -110,12 +110,12 @@ export abstract class LambdaRef extends Construct implements IEventRuleTarget, l
 
     /**
      * Indicates if the resource policy that allows CloudWatch events to publish
-     * notifications to this topic have been added.
+     * notifications to this lambda have been added.
      */
     private eventRuleTargetPolicyAdded = false;
 
     /**
-     * Indicates if the policy that allows CloudWatch logs to publish to this topic has been added.
+     * Indicates if the policy that allows CloudWatch logs to publish to this lambda has been added.
      */
     private logSubscriptionDestinationPolicyAddedFor: logs.LogGroupArn[] = [];
 
@@ -218,7 +218,7 @@ export abstract class LambdaRef extends Construct implements IEventRuleTarget, l
         return this.metric('Throttles', { statistic: 'sum', ...props });
     }
 
-    public subscriptionDestination(sourceLogGroup: logs.LogGroup): logs.SubscriptionDestination {
+    public logSubscriptionDestination(sourceLogGroup: logs.LogGroup): logs.LogSubscriptionDestination {
         const arn = sourceLogGroup.logGroupArn;
 
         if (this.logSubscriptionDestinationPolicyAddedFor.indexOf(arn) === -1) {
@@ -227,7 +227,7 @@ export abstract class LambdaRef extends Construct implements IEventRuleTarget, l
             //
             // (Wildcards in principals are unfortunately not supported.
             this.addPermission('InvokedByCloudWatchLogs', {
-                principal: new ServicePrincipal(new FnSub('logs.${AWS::Region}.amazonaws.com')),
+                principal: new ServicePrincipal(new FnConcat('logs.', new AwsRegion(), '.amazonaws.com')),
                 sourceArn: arn
             });
             this.logSubscriptionDestinationPolicyAddedFor.push(arn);

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

@@ -31,7 +31,7 @@ export = {
         expect(stack).to(haveResource('AWS::Lambda::Permission', {
             Action: "lambda:InvokeFunction",
             FunctionName: { Ref: "MyLambdaCCE802FB" },
-            Principal: { "Fn::Sub": "logs.${AWS::Region}.amazonaws.com" }
+            Principal: { "Fn::Join": ["", ["logs.", {Ref: "AWS::Region"}, ".amazonaws.com"]] }
         }));
 
         test.done();

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

@@ -27,11 +27,10 @@ any amount of days, or `Infinity` to keep the data in the log group forever.
 Log events matching a particular filter can be sent to either a Lambda function
 or a Kinesis stream.
 
-* If the Kinesis stream lives in a different account, you have to also create a
-  `Destination` object in the current account which will act as a proxy for the
-  remote Kinesis stream.
-* If the filter destination is either a Lambda or a Kinesis stream in the
-  current account, they can be subscribed directly.
+If the Kinesis stream lives in a different account, a `CrossAccountDestination`
+object needs to be added in the destination account which will act as a proxy
+for the remote Kinesis stream. This object is automatically created for you
+if you use the CDK Kinesis library.
 
 Create a `SubscriptionFilter`, initialize it with an appropriate `Pattern` (see
 below) and supply the intended destination: