Description

Amazon EventBridge Pipes (User Guide, CloudFormation resource) enable connections between several aws services and has filtering, transforming and enrichment capabilities that makes connecting aws services much easier.

Although there is no L2 support for this new aws feature yet.
This results in an user experience that can be improved to have a similar experience than the aws console has.

The current L1 cfn constructs give user no hint which services can be connected and what needs to be done for example in regards to iam permissions.
The AWS console provides a nice ui that is split into four phases:

This source, enrichment and target have a dropdown list of possible options.
On top of that the console creates a iam policy that is needed to access all the configured sources.

The current L1 construct api has no type safety and gives the user no clue which sources, enrichments and targets can be used. On top of that the user has to create iam roles and permissions by itself.

Example of L1 construct connecting two sqs queues

const pipeRole = new Role(this, "PipeRole", {
  assumedBy: new ServicePrincipal("pipes.amazonaws.com", {}),
});

const sourceQueue = new Queue(this, "SourceQueue");
const targetQueue = new Queue(this, "TargetQueue");

const Pipe = new CfnPipe(this, "MyPipe", {
  roleArn: pipeRole.roleArn,
  source: sourceQueue.queueArn,
  target: targetQueue.queueArn,
});
sourceQueue.grantConsumeMessages(pipeRole);
targetQueue.grantSendMessages(pipeRole);

I'd suggest to build a L2 construct that gives the user guidance how to build pipes.

Possible class diagram

classDiagram 
  direction LR
  
  class Pipe {
    source PipeSource
    target PipeTarget
    filter? PipeFilter
    enrichment? PipeEnrichment
  }
  
  Pipe --> PipeSource
  Pipe --> PipeTarget
  Pipe --> PipeFilter
  Pipe --> PipeEnrichment
  
  PipeSource --> DynamoDBSource
  PipeSource --> KinesisSource
  PipeSource --> AmazonMqSource
  PipeSource --> AmazonMSKSource
  PipeSource --> SelfManagedKafkaSource
  PipeSource --> SelfManagedKafkaSource
  PipeSource --> SqsSource
  
  PipeTarget --> ApiDestinationTarget
  PipeTarget --> ApiGatewayTarget
  PipeTarget --> BatchJobQueueTarget
  PipeTarget --> CloudwatchLoggroupTarget
  PipeTarget --> EcsTaskTarget
  PipeTarget --> AllOtherTarget
  
  
  class PipeFilter {
    fromObject()
    fromString()
  }
  
  class PipeEnrichment 
  
  PipeEnrichment --> ApiDestinationEnrichment
  PipeEnrichment --> ApiGatewayEnrichment
  PipeEnrichment --> LambdaEnrichment
  PipeEnrichment --> StepFunctionEnrichment

Example usage of the L2 construct

const sourceQueue = new Queue(this, "SourceQueue");
const targetQueue = new Queue(this, "TargetQueue");
const lambdaFunction = new NodejsFunction(this, "LambdaFunction")

const Pipe = new Pipe(this, "MyPipe", {
  source: new SqsSource(sourceQueue),
  target: new SqsTarget(targetQueue, {
    inputTemplate: JSON.stringify({
      body: "<$.body>",
      messageId: "<$.messageId>",
      messageAttributes: "<$.messageAttributes>",
      nestedBody: {
        body: "<$.body>",
      },
    }),
  }),
  filter: PipeFilter.fromObject({ body: [{ prefix: "Test" }] }),
  enrichment: new LambdaEnrichment(lambdaFunction)
});

PoC implementation

https://github.com/RaphaelManke/aws-cdk-pipes-rfc-473

Roles

See RFC Process for details

Workflow

• Tracking issue created (label: status/proposed)
• API bar raiser assigned (ping us at #aws-cdk-rfcs if needed)
• Kick off meeting
• RFC pull request submitted (label: status/review)
• Community reach out (via Slack and/or Twitter)
• API signed-off (label api-approved applied to pull request)
• Final comments period (label: status/final-comments-period)
• Approved and merged (label: status/approved)
• Execution plan submitted (label: status/planning)
• Plan approved and merged (label: status/implementing)
• Implementation complete (label: status/done)

Author is responsible to progress the RFC according to this checklist, and
apply the relevant labels to this issue so that the RFC table in README gets
updated.