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,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 +192,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,61 @@ 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 { 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, `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 +374,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();
+ }
+};