feat(kinesis): support stream consumers

[humanzz]

Issue # (if applicable)

Closes #32050

Reason for this change

Support Enhanced fan-out consumers via AWS::Kinesis::StreamConsumer and facilitate cross-account stream consumption via Lambda

Description of changes

• introduce StreamConsumer construct to model AWS::Kinesis::StreamConsumer
  • introduce addToResourcePolicy to enable creating/configuring a resource policy for the consumer
  • introduce grant and grantRead for granting permissions
  • leverage iam.Grant.addToPrincipalOrResource in grant to be able to use grant methods cross environments to update the grantee's iam policy and the consumer's resource policy as needed
• update ResourcePolicy to support both Stream and StreamConsumer
• update Stream's grant to leverage iam.Grant.addToPrincipalOrResource for cross-environment support
• introduce KinesisConsumerEventSource to lambda-event-sources for use with the newly introduced StreamConsumer

Useful links

• https://docs.aws.amazon.com/streams/latest/dev/enhanced-consumers.html
• https://aws.amazon.com/blogs/big-data/invoke-aws-lambda-functions-from-cross-account-amazon-kinesis-data-streams/

Description of how you validated changes

unit and integration tests

Checklist

• My code adheres to the CONTRIBUTING GUIDE and DESIGN GUIDELINES

---

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

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 78.46%. Comparing base (24adca3) to head (f1f7ef6).
Report is 1 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #32087   +/-   ##
=======================================
  Coverage   78.46%   78.46%           
=======================================
  Files         106      106           
  Lines        7208     7208           
  Branches     1323     1323           
=======================================
  Hits         5656     5656           
  Misses       1364     1364           
  Partials      188      188           

Flag	Coverage Δ
suite.unit	78.46% <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

Components	Coverage Δ
packages/aws-cdk	78.46% <ø> (ø)

[humanzz]

One other thing I noticed, is that in lambda event sources, the KinesisEventSource assumes an IStream, and I think that will need to be changed to allow for StreamConsumer

aws-cdk/packages/aws-cdk-lib/aws-lambda-event-sources/lib/kinesis.ts

Line 23 in 6878126

constructor(readonly stream: kinesis.IStream, props: KinesisEventSourceProps) {

A potential implementation

• Extract the logic into a private base class e.g. KinesisEventSourceBase which takes as input a private KinesisSource
• Update existing KinesisEventSource to extend KinesisEventSourceBase
• Introduce a new KinesisConsumerEventSource also extending KinesisEventSourceBase

import * as constructs from 'constructs';
import { StreamEventSource, StreamEventSourceProps } from './stream';
import * as iam from '../../aws-iam';
import * as kinesis from '../../aws-kinesis';
import * as lambda from '../../aws-lambda';
import * as cdk from '../../core';

export interface KinesisEventSourceProps extends StreamEventSourceProps {
  /**
   * The time from which to start reading, in Unix time seconds.
   *
   * @default - no timestamp
   */
  readonly startingPositionTimestamp?: number;
}

interface KinesisSource {
  readonly node: constructs.Node;
  readonly sourceArn: string;
  readonly eventSourceName: string;
  grantRead(grantee: iam.IGrantable): iam.Grant;
}

abstract class KinesisEventSourceBase extends StreamEventSource {
  private _eventSourceMappingId?: string = undefined;
  private _eventSourceMappingArn?: string = undefined;
  private startingPositionTimestamp?: number;

  constructor(readonly source: KinesisSource, props: KinesisEventSourceProps) {
    super(props);
    this.startingPositionTimestamp = props.startingPositionTimestamp;

    this.props.batchSize !== undefined && cdk.withResolved(this.props.batchSize, batchSize => {
      if (batchSize < 1 || batchSize > 10000) {
        throw new Error(`Maximum batch size must be between 1 and 10000 inclusive (given ${this.props.batchSize})`);
      }
    });
  }

  public bind(target: lambda.IFunction) {
    const eventSourceMapping = target.addEventSourceMapping(`${this.source.eventSourceName}:${cdk.Names.nodeUniqueId(this.source.node)}`,
      this.enrichMappingOptions({
        eventSourceArn: this.source.sourceArn,
        startingPositionTimestamp: this.startingPositionTimestamp,
        metricsConfig: this.props.metricsConfig,
      }),
    );
    this._eventSourceMappingId = eventSourceMapping.eventSourceMappingId;
    this._eventSourceMappingArn = eventSourceMapping.eventSourceMappingArn;

    this.source.grantRead(target);
  }

  /**
   * The identifier for this EventSourceMapping
   */
  public get eventSourceMappingId(): string {
    if (!this._eventSourceMappingId) {
      throw new Error(`${this.source.eventSourceName} is not yet bound to an event source mapping`);
    }
    return this._eventSourceMappingId;
  }

  /**
   * The ARN for this EventSourceMapping
   */
  public get eventSourceMappingArn(): string {
    if (!this._eventSourceMappingArn) {
      throw new Error(`${this.source.eventSourceName} is not yet bound to an event source mapping`);
    }
    return this._eventSourceMappingArn;
  }
}

/**
 * Use an Amazon Kinesis stream as an event source for AWS Lambda.
 */
export class KinesisEventSource extends KinesisEventSourceBase {
  constructor(stream: kinesis.IStream, props: KinesisEventSourceProps) {
    super({ ...stream, eventSourceName: 'KinesisEventSource', sourceArn: stream.streamArn }, props);
  }
}

/**
 * Use an Amazon Kinesis stream consumer as an event source for AWS Lambda.
 */
export class KinesisConsumerEventSource extends KinesisEventSourceBase {
  constructor(streamConsumer: kinesis.IStreamConsumer, props: KinesisEventSourceProps) {
    super({ ...streamConsumer, eventSourceName: 'KinesisConsumerEventSource', sourceArn: streamConsumer.streamConsumerArn }, props);
  }
}

[humanzz]

This permission is now already part of grantRead

[aws-cdk-automation]

AWS CodeBuild CI Report

• CodeBuild project: AutoBuildv2Project1C6BFA3F-wQm2hXv2jqQv
• Commit ID: f1f7ef6
• Result: SUCCEEDED
• Build Logs (available for 30 days)

Powered by github-codebuild-logs, available on the AWS Serverless Application Repository