feat(logs): custom Role for Kinesis destination (#13553)

* [x] Testing
  - Unit test added.
* [x] Docs
  - __jsdocs__: All public APIs documented
  - __README__: README and/or documentation topic updated

When creating a logs destination, an optional `role` can be passed to be assumed when writing logs in the destination.
 
Closes #7661.

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

Author: gmeligio

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

@@ -9,4 +9,7 @@
 
 <!--END STABILITY BANNER-->
 
-A short description here.
+This library contains destinations for AWS CloudWatch Logs SubscriptionFilters. You
+can send log data to Kinesis Streams or Lambda Functions.
+
+See the documentation of the `logs` module for more information.

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

@@ -3,18 +3,35 @@ import * as kinesis from '@aws-cdk/aws-kinesis';
 import * as logs from '@aws-cdk/aws-logs';
 import { Construct } from '@aws-cdk/core';
 
+/**
+ * Customize the Kinesis Logs Destination
+ */
+export interface KinesisDestinationProps {
+  /**
+   * The role to assume to write log events to the destination
+   *
+   * @default - A new Role is created
+   */
+  readonly role?: iam.IRole;
+}
+
 /**
  * Use a Kinesis stream as the destination for a log subscription
  */
 export class KinesisDestination implements logs.ILogSubscriptionDestination {
-  constructor(private readonly stream: kinesis.IStream) {
+  /**
+   * @param stream The Kinesis stream to use as destination
+   * @param props The Kinesis Destination properties
+   *
+   */
+  constructor(private readonly stream: kinesis.IStream, private readonly props: KinesisDestinationProps = {}) {
   }
 
   public bind(scope: Construct, _sourceLogGroup: logs.ILogGroup): logs.LogSubscriptionDestinationConfig {
     // Following example from https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html#DestinationKinesisExample
     // Create a role to be assumed by CWL that can write to this stream and pass itself.
     const id = 'CloudWatchLogsCanPutRecords';
-    const role = scope.node.tryFindChild(id) as iam.IRole || new iam.Role(scope, id, {
+    const role = this.props.role ?? scope.node.tryFindChild(id) as iam.IRole ?? new iam.Role(scope, id, {
       assumedBy: new iam.ServicePrincipal('logs.amazonaws.com'),
     });
     this.stream.grantWrite(role);

packages/@aws-cdk/aws-logs-destinations/test/kinesis.test.ts

@@ -1,4 +1,5 @@
 import { Template } from '@aws-cdk/assertions';
+import * as iam from '@aws-cdk/aws-iam';
 import * as kinesis from '@aws-cdk/aws-kinesis';
 import * as logs from '@aws-cdk/aws-logs';
 import * as cdk from '@aws-cdk/core';
@@ -136,3 +137,54 @@ test('stream can be subscription destination twice, without duplicating permissi
     },
   });
 });
+
+test('an existing IAM role can be passed to new destination instance instead of auto-created ', ()=> {
+  // GIVEN
+  const stack = new cdk.Stack();
+  const stream = new kinesis.Stream(stack, 'MyStream');
+  const logGroup = new logs.LogGroup(stack, 'LogGroup');
+
+  const importedRole = iam.Role.fromRoleArn(stack, 'ImportedRole', 'arn:aws:iam::123456789012:role/ImportedRoleKinesisDestinationTest');
+
+  const kinesisDestination = new dests.KinesisDestination(stream, { role: importedRole });
+
+  new logs.SubscriptionFilter(logGroup, 'MySubscriptionFilter', {
+    logGroup: logGroup,
+    destination: kinesisDestination,
+    filterPattern: logs.FilterPattern.allEvents(),
+  });
+
+  // THEN
+  const template = Template.fromStack(stack);
+  template.resourceCountIs('AWS::IAM::Role', 0);
+  template.hasResourceProperties('AWS::Logs::SubscriptionFilter', {
+    RoleArn: importedRole.roleArn,
+  });
+});
+
+test('creates a new IAM Role if not passed on new destination instance', ()=> {
+  // GIVEN
+  const stack = new cdk.Stack();
+  const stream = new kinesis.Stream(stack, 'MyStream');
+  const logGroup = new logs.LogGroup(stack, 'LogGroup');
+
+  const kinesisDestination = new dests.KinesisDestination(stream);
+
+  new logs.SubscriptionFilter(logGroup, 'MySubscriptionFilter', {
+    logGroup: logGroup,
+    destination: kinesisDestination,
+    filterPattern: logs.FilterPattern.allEvents(),
+  });
+
+  // THEN
+  const template = Template.fromStack(stack);
+  template.resourceCountIs('AWS::IAM::Role', 1);
+  template.hasResourceProperties('AWS::Logs::SubscriptionFilter', {
+    RoleArn: {
+      'Fn::GetAtt': [
+        'LogGroupMySubscriptionFilterCloudWatchLogsCanPutRecords9112BD02',
+        'Arn',
+      ],
+    },
+  });
+});