EventBridge Pipes L2 Construct #473
Comments
|
I created a PoC implementation here: https://github.com/RaphaelManke/aws-cdk-pipes-rfc-473 |
PipeAWS EventBridge Pipe has itself is a fully managed service that does the heavy lifting of polling a source, then be able to filter out payloads based on filter criteria. This reduces the target invocations and can reduce costs. So a Pipe has the following components:
besides these (core) components that are used while processing data, there are additional attributes that describe a Pipe
graph LR
classDef required fill:#00941b
classDef optional fill:#5185f5
Source:::required
Filter:::optional
Enrichment_Input_Transformation[Input transformation]:::optional
Enrichment:::optional
Target_Input_Transformation[Input transformation]:::optional
Target:::required
Source --> Filter --> Enrichment_Input_Transformation --> Enrichment --> Target_Input_Transformation --> Target
Example implementationinterface PipeProps {
readonly source : PipeSource
readonly target : PipeTarget
readonly filter? : PipeFilter
readonly enrichment? : PipeEnrichment
readonly role? : IRole // role is optional, if not provided a new role is created
readonly description : string
readonly tags? : Tags
}
interface Pipe {
readonly role : IRole
readonly source : PipeSource
readonly target : PipeTarget
readonly filter? : PipeFilter
readonly enrichment? : PipeEnrichment
readonly description : string
readonly tags? : Tags
constructor(scope: Scope, id: string, props:PipeProps)
}Open questions
SourceA source is a AWS Service that needs to be polled.
The CfnPipe resource reference the source only by their ARN. Right now there is no validation in der CDK framework that checks if a ARN is valid or not. export abstract class PipeSource {
public readonly sourceArn: string;
public readonly sourceParameters?:
| CfnPipe.PipeSourceParametersProperty
| IResolvable;
constructor(
sourceArn: string,
props?: CfnPipe.PipeSourceParametersProperty,
) {
this.sourceArn = sourceArn;
this.sourceParameters = props;
}
public abstract grantRead(grantee: IRole): void;
}This PipeSource class has a sourceArn that is mapped to the CfnPipe {
sqsQueueParameters : {...}
}The specific Source class implementation hides this detail for the user and provide a interface with only the possible configuration options that are possible for the specific source. interface PipeSourceSqsQueueParametersProperty {
readonly batchSize?: number;
readonly maximumBatchingWindowInSeconds?: number;
}This interface for example is provided by the cloudformation specification and can be used as a base for possible configurations (additional validation can be added if useful). To be able to consume a source the EventBridge Pipe has a IAM-role. This role needs to have a policy to read from the source. Example implementationAn example api for a source that polls for a SQS-queue then can look like: export class SqsSource extends PipeSource {
private queue: IQueue;
constructor(queue: IQueue, props?:CfnPipe.PipeSourceSqsQueueParametersProperty) {
super(queue.queueArn, { sqsQueueParameters: props });
this.queue = queue;
}
public grantRead(grantee: IRole): void {
this.queue.grantConsumeMessages(grantee);
}
}It takes an existing SQS-queue and polling properties that are possible for that kind of source and does implement a grantRead method which creates the required IAM policy for the Pipe role. RoleA IAM role is required that can be assumed by the Open questions
FilterA filter does pattern matching based on the incoming payload and the specified filter criteria's. The matching is done in the same way the EventBridge pattern are matched. Example ImplementationThe implementation is split into two types.
interface IPipeFilterPattern {
pattern: string;
}
class PipeGenericFilterPattern {
static fromJson(patternObject: Record<string, any>) :IPipeFilterPattern {
return { pattern: JSON.stringify(patternObject) };
}
}
interface SqsMessageAttributes : {
messageId?: string;
receiptHandle?: string;
body?: any;
attributes?: {
ApproximateReceiveCount?: string;
SentTimestamp?: string;
SequenceNumber?: string;
MessageGroupId?: string;
SenderId?: string;
MessageDeduplicationId?: string;
ApproximateFirstReceiveTimestamp?: string;
};
messageAttributes?: any;
md5OfBody?: string;
}
class PipeSqsFilterPattern extends PipeGenericFilterPattern {
static fromSqsMessageAttributes(attributes: SqsMessageAttributes) :IPipeFilterPattern {
return {
pattern: JSON.stringify( attributes ),
};
}
}TargetA Target is the end of the Pipe. After the payload from the source is pulled, filtered and enriched it is forwarded to the target.
The CfnPipe resource reference the target only by their ARN. Right now there is no validation in der CDK framework that checks if a ARN is valid or not. The implementation is then similar to the Source implementation: Example implementationinterface IPipeTarget {
targetArn: string;
targetParameters: CfnPipe.PipeTargetParametersProperty;
grantPush(grantee: IRole): void;
};
export interface SqsTargetProps {
queue: IQueue;
sqsQueueParameters?: CfnPipe.PipeTargetSqsQueueParametersProperty;
}
export class SqsTarget implements IPipeTarget {
private queue: IQueue;
targetArn: string;
targetParameters: CfnPipe.PipeTargetParametersProperty;
constructor(props: SqsTargetProps) {
this.queue = props.queue;
this.targetArn = props.queue.queueArn;
this.targetParameters = { sqsQueueParameters: props.sqsQueueParameters };
}
public grantPush(grantee: IRole): void {
this.queue.grantSendMessages(grantee);
}
}EnrichmentIn the enrichment step the filtered payloads can be used to invoke one of the following services
The invocation is a synchron call to the service. The result of the enrichment step then can be used to combine it with the filtered payload to target.
The enrichment ARN is the AWS resource ARN that should be invoked. The Role must have access to invoke this ARN. Example implementationexport abstract class PipeEnrichment {
public readonly enrichmentArn: string;
public enrichmentParameters: CfnPipe.PipeEnrichmentParametersProperty;
constructor( enrichmentArn: string, props: CfnPipe.PipeEnrichmentParametersProperty) {
this.enrichmentParameters = props;
this.enrichmentArn = enrichmentArn;
}
abstract grantInvoke(grantee: IRole): void;
}
export class LambdaEnrichment extends PipeEnrichment {
private lambda : IFunction;
constructor(lambda: IFunction, props: {inputTransformation?: PipeInputTransformation}) {
super(lambda.functionArn, { inputTemplate: props.inputTransformation?.inputTemplate });
this.lambda = lambda;
}
grantInvoke(grantee: IRole): void {
this.lambda.grantInvoke(grantee);
}
}Input TransformationInput transformations are used to transform or extend payloads to a desired structure. This transformation mechanism can be used prior to the enrichment or target step. There are two types of mappings. Both types can be either static values or use values from the output of the previous step. Additionally there are a few values that come from the pipe itself (see
Example implementationenum reservedVariables {
PIPES_ARN = '<aws.pipes.pipe-arn>',
PIPES_NAME = '<aws.pipes.pipe-name>',
PIPES_TARGET_ARN = '<aws.pipes.target-arn>',
PIPE_EVENT_INGESTION_TIME = '<aws.pipes.event.ingestion-time>',
PIPE_EVENT = '<aws.pipes.event>',
PIPE_EVENT_JSON = '<aws.pipes.event.json>'
}
type StaticString = string;
type JsonPath = `<$.${string}>`;
type KeyValue = Record<string, string | reservedVariables>;
type StaticJsonFlat = Record<string, StaticString| JsonPath | KeyValue >;
type InputTransformJson = Record<string, StaticString| JsonPath | KeyValue | StaticJsonFlat>;
type PipeInputTransformationValue = StaticString | InputTransformJson
export interface IInputTransformationProps {
inputTemplate: PipeInputTransformationValue;
}
export class PipeInputTransformation {
static fromJson(inputTemplate: Record<string, any>): PipeInputTransformation {
return new PipeInputTransformation({ inputTemplate });
}
readonly inputTemplate: string;
constructor(props: IInputTransformationProps) {
this.inputTemplate = JSON.stringify(props);
}
}Open Question
|
|
I am an engineer on the Pipes service team. Thanks for contributing this! I just came back from vacation and will allocate some time in the next couple of weeks to review this. |
|
@nikp is there something I can do in the mean time? Should I start with a draft RFC Pull Request? Extend my PoC implementation? |
|
@RaphaelManke I'm truly sorry for the delay. I did not mean to become a bottleneck. I've been unable to allocate time to this due to some other emergent issues. I am not on the CDK team, and am a customer of them also. Please follow the process they recommend - in the checklist I think a bar raiser needs to be assigned. That said a draft RFC seems like the right move too. I don't want to overpromise and underdeliver again but I will get back to provide feedback as soon as I can. |
|
it's been a month , is there any news about this ? seems stuck in stage 2 for a few months now |
|
Yes there is 😃 @mrgrain got assigned as bar raiser. We will meet in the next weeks and discuss next steps. |
mrgrain
changed the title
|
@RaphaelManke thanks for this just implemented CfnPipe this weekend. The source (DynamoDB Stream) and target (EventBridge) were pretty easy to setup, I used the same setup with .grantRead and .grantPut on a custom role. The filter was also quite easy, the thing that tripped me up for a bit was how to transform the target input to EventBridge. I could not find a way to dynamically set the source and event-detail from the event itself. Would be helpful to understand what kind of default transformation Pipes is setting for each target. I can provide some code later to explain better. Looking forward to having this as a L2. |
|
Hi @RaphaelManke ! Thanks for your contribution - an L2 construct for Pipes would be a great asset. @nikp asked me to take a look to not make you wait longer. Would it be fair to separate the construct into several key concepts (which unsurprisingly match the Pipes stages)?
SourcesI'll have to double-check but I believe all sources have FiltersThe fields and expressions changing with the source is indeed a bit cumbersome (it also affects writing Enrichers) but I'd be worried that trying to wrap this into classes for each source could end up becoming a liability when Pipes supports more sources. EventBridge rules uses a common EventPatterns class for this reason. EnrichmentWould you plan on providing wrappers for each kind (I saw Lambda in your repo, so I assume yes)? TargetsI'd have to have a closer look but would favor reusing the Input Transformers from EB if possible. I'd be cautious again about tailoring them for each Target type fort the same reason as the sources. I'd be happy to have a live chat also. |
Sources
Nice 😃 didn't notice that yet. I updated the RFC PR to add this info.
My idea would be that the Source class constructor provide these source specific attributes as constructor params. Filters
Reusing would be a good idea, I am not sure if it matches 100 %. A source specific filter class would be an addition to make creating these pattern easier. TargetsReusing existing Input Transformers would be very time efficient because this parts is the trickiest of all due to the I am open to have a live chat 😃 you can reach me on the cdk slack or any other social media. |
|
Thanks for your great work and initiative @RaphaelMank Legitimately my favorite PR of all time |
|
Just wanted to give you an update: https://github.com/RaphaelManke/aws-cdk-pipes-rfc-473/tree/main/e2e I would be happy if you guys would play around with it and give feedback or report bugs. I am also happy to take PRs 😄 Note: this is a POC implementation and subject to change. I would not recommend it to use it in production. |
|
@RaphaelManke thanks for putting this up, will try it out! Is it possible to set the |
|
Thanks for this example. I checked and now think understand your question. You want to use an value from a source/enrichment step as an value for the target invocation. Source: SQS Source Event: {
"orderSystem" : "offline",
"orderId" : "Order-123"
}this will be a SQS message in the format: {
"messageId": "...",
"receiptHandle": "...",
"body": "{ \n \"orderSystem\" : \"offline\", \n \"orderId\" : \"Order-123\"\n }",
"attributes": {
"ApproximateReceiveCount": "...",
"SentTimestamp": "....",
"SenderId": "...",
"ApproximateFirstReceiveTimestamp": "..."
},
"messageAttributes": {},
"md5OfBody": "...",
"eventSource": "...",
"eventSourceARN": "...",
"awsRegion": "..."
}the target event should be in format: {
"version": "...",
"id": "...",
"detail-type": "newOrder", // <-- static string
"source": "offline", // <-- dynamic from the source event
"account": "...",
"time": "...",
"region": "...",
"resources": [],
"detail": {
"orderId" : "Order-123" // <-- dynamic from the source event
}
}AFAIK this is not possible with the tools provided in the AWS console because it lacks the capability to set the But it is possible via the AWS API or cloudformation. The solution requires two parts. {
"orderId" : <$.body.orderId>
}which needs to be stringyfied to: The second part is the the pipe target takes an object called For this example the parameters look like this: {
"EventBridgeEventBusParameters": {
"Source": "$.body.orderSystem",
"DetailType": "newOrder"
}
}This is described in the docs. Although this docs are not very clear and misses examples. In the CDK construct you can already use the const target = new EventBridgeEventBusTarget(targetEventBus, {
source: '$.body.orderSystem',
detailType: 'newOrder,
});The |
|
You got it right, interesting will test this out! |
|
Would be great to have this construct! |
|
@mrgrain sorry i am losing my Github notifications. Today, Pipe retry policy and dead letter destinations are only supported for Kinesis and DynamoDB Stream sources: |
|
The L2 construct is still in the making 😃 While using the the current api I noticed that it feels a little unintuitive to use the For example when using If I want to transform the body I need to know that this can be done by the How about making this explicit? See the following example: current implementation const enrichment = new ApiDestinationEnrichment(apiDestination, {
headerParameters: {
'Content-Type': 'application/json',
'Dynamic-Header': '$.messageId',
},
inputTransformation: PipeInputTransformation.fromJson({ // <-- generic transformation which will be the body
bodyFromSqs: '<$.body>',
rawSqsMessage: '<aws.pipes.event.json>',
}),
});alternative implementation const enrichment = new ApiDestinationEnrichment(apiDestination, {
headerParameters: {
'Content-Type': 'application/json',
'Dynamic-Header': '$.messageId',
},
body: PipeInputTransformation.fromJson({ // <-- the body is a attribute of the enrichment
bodyFromSqs: '<$.body>',
rawSqsMessage: '<aws.pipes.event.json>',
}),
});What do you think? |
Is it basically a rename from |
|
Hi folks, I've finally carved off some time and am doing a thorough review of this whole thread and will reply with detailed thoughts on all the outstanding work @RaphaelManke has done so far to push this forward But I'll go LIFO and reply to the last point first. You're very correct that the input transformation for an ApiDestination is the "body" of the request and is intrinsically tied to the rest of the target. Unfortunately things are a bit more fuzzy for the the other targets. The trouble would be with suggesting domain language that isn't valid for that target. The event goes into the SQS Message For some targets, the event is ignored entirely, or rather it's only used as a possible source for dynamic path parameters to the values of the Target - e.g. Redshift and ECS that let you execute data api queries and run tasks, respectively, but the event itself isn't passed through. That's just the nature of semantics of trying to be a universal resource for a variety of different AWS APIs. In the long term, Pipes will support "Universal targets" the way that EventBridge Scheduler does: https://docs.aws.amazon.com/scheduler/latest/UserGuide/managing-targets-universal.html and each invocation will become an "AWS SDK call" with the event as a source of parameter data for it. So I would suggest NOT renaming I will reply with more comments later tonight |
|
First of all, want to say kudos, @RaphaelManke, you've done really creative work on this RFC, and I think it's going to be a great user experience for CDK users everywhere. I'm learning a lot from your ideas, as this is not really my area of expertise building mostly web services. :) I welcome disagreement to any of my proposals or ideas below from you or anyone else in the community or anyone more experienced like @mrgrain Now onto some of the detailed comments
The object type i.e. IInputTransformation should certainly be the same, but the instances should be inside PipeEnrichment/PipeTarget separately, rather than at the root level - to make sure it’s really clear which one it’s transforming. As you noted in your latest comment, For ApiDestinations the InputTransformer is actually a key part of the target construction, and so we should strive for clarity by keeping them close. However, I'm noticing from looking at the code in your repo that you currently have it on individual Enrichment/Target objects - e.g. ApiDestinationEnrichment, LambdaEnrichment, etc. Is that intentional? There shouldn't be any reason why these can't be reused. Input transformer support is identical in syntax for both enrichment and Target. Speaking of reuse, just while we're here, the ApiDestination and ApiGateway parameters for both Enrichment and Target are identical, and could also be reused. In the API/CFN we call it
I don’t feel strongly about this one either way, there is good pros and cons to both. So I would default to maintaining consistency with CFN/API, and putting it inside PipeSource to minimize any user surprise
As a matter of guidance, I would suggest not pulling up common parameters for the individual sources to the top level, and keep them on the individual Source type objects. The reason is that there are subtle differences between even concepts that seem common. For example, Kinesis/DynamoDB stream batches are always going to be within a single shard. Whereas SQS batches are across the entire source. These are just the natural realities of a generic resource to integrate different types of services. The batch size limits are also quite different.
Good question, let’s look at some prior art for how SNS→SQS: https://github.com/aws/aws-cdk/blob/cea1039e3664fdfa89c6f00cdaeb1a0185a12678/packages/%40aws-cdk/aws-sns-subscriptions/lib/sqs.ts#L54 Would that work?
I don’t know enough about IRole vs Role in CDK-land to comment on this In a broad sense, the specific policies to configure permissions for different sources can actually get fairly complex, and I can provide some help with this. SQS and Kinesis polling is fairly straightforward but permissions to poll for instance a Self-Managed-Kafka source from a VPC can get fairly advanced, and diverse depending on the authorization types. I can share the policy templates the console uses for each scenario, and we can probably encode them into the logic of the L2 construct. Just to give you an example of that last one: A good starting point for the policies is the Would be happy to provide more help here
I can't quite tell if I'm agreeing or disagreeing with @spac3lord but I like the concept of strongly typed fields for each Filter - the Source documentation has payload examples for each source. For event buses pattern matching is a bit easier because each one has a standard envelope allowing us to use EventPattern. With Pipes each source is slightly different and we strived to be (mostly) backwards compatible with the payloads of Lambda EventSourceMapping, resulting in what you see here.
My suggestion would be to use the terminology
At first i wasn't following this point... Static and Dynamic mappings are the same, the only question is whether they contain inline variable usage. The string vs JSON transformer type may be considered somewhat distinct if only because a user may want to explicitly choose to choose JSON and want client-side validation. (This is extra desirable because a “JSON” input transformer template is itself not necessarily valid JSON to allow for the possibility of referencing JSON objects, such as: But then I saw your code, and I think I understand now - you're describing how you want to strongly type the validation for these right? Very cool!
This syntax is very compelling to me, can we use the same for the Enrichment/Target parameters that all support json value?
This one I will disagree with @spac3lord on, though I’m willing to hear other arguments. While the semantic purpose behind input transformation and the overall capabilities remain the same, the Pipe input transformation syntax was simplified by merging everything into one field (whereas EventBridge Rule transformers separate variable declaration from usage). Further, while Pipes are a part of the EventBridge “product”, and are in the same console, from an API and CFN perspective, they do not share common types, and are distinct. So maybe the CDK should stay distinct too?
That’s an interesting concept but I’m not sure how it would help. The best way would probably be to work around the fact that JSON-escaped |
I would like to see the existing Matcher being included in the design for filters. In reality this will have to be a subset because not all matchers are supported, but the difference is now well documented. For the purpose of the RFC you can call it |
To Do
|
@mrgrain FYI this 404s |
|
Fixed, thank you. |
|
I've updated to the corresponding PR and would like to move the discussion over to the PR so we can discuss around the RFC directly with the goal to get that one merged. So that we are enabled to start the work of an alpha module. |
mrgrain
added
status/final-comment-period
|
Thanks @RaphaelManke for the iterations! I'm very happy with this now and have progressed the RFC to final comments period. |
mrgrain
added
status/implementing
|
Approved and moving straight to implementing since you already have a prototype @RaphaelManke You are most welcome to start submitting PRs. The actual code review and merging will be handled by whoever is currently doing PR reviews, which is not me at the moment. However I will be hanging around to provide context if needed. Feel free to ping me if something is stuck. |
This PR is the starting point of the implementation for a L2 construct as defined in aws/aws-cdk-rfcs#473 In this PR the basic Pipe class is introduced including the api interfaces for how to define a pipe. Closes #23495 ---- *By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*
This PR is the starting point of the implementation for a L2 construct as defined in aws/aws-cdk-rfcs#473 In this PR the basic Pipe class is introduced including the api interfaces for how to define a pipe. Closes #23495 ---- *By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*
mrgrain
added
status/done
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
I'd suggest to build a L2 construct that gives the user guidance how to build pipes.
Possible class diagram
expand 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 --> StepFunctionEnrichmentExample usage of the L2 construct
PoC implementation
https://github.com/RaphaelManke/aws-cdk-pipes-rfc-473
Roles
Workflow
status/proposed)status/review)api-approvedapplied to pull request)status/final-comments-period)status/approved)status/planning)status/implementing)status/done)The text was updated successfully, but these errors were encountered: