Feature request: add support for large message handling (SQS, SNS, EventBridge, etc.)

未關閉
#1,615 12 則留言 7 個 reaction 已指派 0 人 在 GitHub 檢視

維護者通常 1 天內回覆

還沒有人認領這個 Issue。

評估

難度
5/5
預估耗時
一週以上
新手友好度
25/100
Issue 類型
功能
描述清晰度
基本清楚
活躍度
停滯
技術堆疊
aws, python

研究方向

Start by reading the existing aws_lambda_powertools.utilities.message_handler proposal in the issue and the current utilities.batch integration points; compare the example APIs with the repository’s utility patterns. Trace SQS record handling and persistence abstractions before deciding the supported service and metadata format. Done means an agreed, standalone SQS-first design exists, with producer, consumer, batch integration, configuration, and failure-handling behavior defined and covered by tests.

由索引模型根據 Issue 內容生成。

描述

help wanted need-customer-feedback need-rfc
Is this related to an existing feature request or issue?

No response

Which AWS Lambda Powertools utility does this relate to?

Batch processing

Summary

A common pain and friction point in developing Event-Driven Architectures using AWS Serverless managed services is the existing service limit on message size which generally equates to 256KB for messages passing through any of the messaging services - SQS, SNS, EventBridge, [insert others here].

In an enterprise integration landscape, 256KB is a challenging size. It requires the building of store-and-forward systems where the inbound message is stored in an S3 bucket (or another storage mechanism) and then pushing meta-data through the messaging service to enable processing by consumers, which all need to retrieve the actual message using the metadata before processing.

Use case

System Integration Use-case where data is submitted to an integration service in AWS which uses EventBridge to route event data to different destinations. API gateway to Lambda proxy integration allows up to 6MB of payload to be processed which is way larger then 256KB limit of event bridge.

In this scenario the Lambda behind the API is required to store the large payload first in S3 and then push meta-data for routing through EventBridge. Consumers then need to read the original large message and process it.

Proposal

Similar to the idempotency Utility, an abstract Persistence Class should be created to allow for the storing and retrieval of a message into an AWS storage Service (defaulting to S3 seems sensible).

Build out a message client handler like the sqs-extended-client-lib (which is based on the AWS Java implementation for large message sending to SQS and appears abandoned on GitHub) for storing, creating meta-data for forwarding through the messaging service.

Like idempotency would need to consider JMESPath for extracting meta-data or a mechanism for building the required message structure for submitting to AWS message service (SQS, SNS, Eventbridge, etc.).

Feels like a nice utility to cover more than just SQS but also others - ideally should start with the most common use case, and SQS seems logical in this regard since there are well-known implementations that already exist and match the existing user experience from the Java utility as a starting point seems reasonable.

Would also integrate the capability of detecting the large message events within the existing Powertools batch processing utilities so that the retrieval of the large message can be done as part of the batch processing utilities when they are used.

Need to provide a stand-alone mechanism for retrieving the large messages for customers who wish to partially adopt Lambda powertools so there are pathways for everyone to gain advantage from this feature.

Producer:


from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.utilities.message_handler import SQSMessageClient, S3MessagePersistenceLayer, MessageClientConfig
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = APIGatewayRestResolver()

config =  MessageClientConfig(
    message_key_jmespath="messageId", 
    use_local_cache=True,
)
persistence_layer = S3PersistenceLayer(bucket=os.environ.get("MessageBucket"))
sqs_client = SQSMessageClient(persistence_store=persistence_layer, config=config)

@app.post("/event")
@tracer.capture_method
def post_event():
    payload = app.current_event.json_body
    return sqs_client.send_message(queue=os.environ.get("MyMessageQueue"), message=payload)


# You can continue to use other utilities just as before
@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

Consumer:

import json

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType, batch_processor
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
from aws_lambda_powertools.utilities.typing import LambdaContext


processor = BatchProcessor(event_type=EventType.SQS)
tracer = Tracer()
logger = Logger()


@tracer.capture_method
def record_handler(record: SQSRecord):
    payload: str = record.body
    if payload:
        item: dict = json.loads(payload)
    ...

@logger.inject_lambda_context
@tracer.capture_lambda_handler
@batch_processor(record_handler=record_handler, processor=processor)
def lambda_handler(event, context: LambdaContext):
    return processor.response()

Out of scope

Initially a single messaging service should be targetted but consideration of other Messaging services (SNS, EventBridge, Kinesis, etc) should be considered through design to enable this feature on all batch processing utilities in the future.

Potential challenges

A customisable method of taking the large message and creating the meta-data for the messaging service with data detailing how to retrieve the large message is needed. JMESPath or JSON path can be considered for simple use cases but should also provide a custom function implementation since not every use case will be so straightforward.

Need to also consider how the consumer would know where to access the large message data given it will be in a different Lambda function and potentially without knowledge of the storage mechanism - if this could be embedded in meta-data for Powertools to determine from the consumer side it would lessen boilerplate considerations.

Dependencies and Integrations

Batch Utilities would need to understand how to retrieve large messages when this utility is in use. Having the retrieval as an automated mechanism removes a lot of boilerplate code from existing solutions.

Alternative solutions
[sqs-extended-client-lib](https://github.com/timothymugayi/boto3-sqs-extended-client-lib) is a consideration for SQS only and appears to be abandoned (no changes in over 14 months).
Acknowledgment
主要語言
Python
星號
3.3k
分支
511
平均合併
3 天 8 小時
30 天內合併 PR
92

環境準備

從這裡開始

  1. 先讀完整個 Issue,再讀專案的貢獻指南。
  2. 在 Issue 下留言說明你要接手 —— 這能避免兩個人做同樣的事。
  3. Fork 儲存庫,在一個分支上完成修改。
  4. 送出 Pull Request,並在描述裡引用這個 Issue 編號。

aws-powertools/powertools-lambda-python 的其他 Issue

查看 aws-powertools/powertools-lambda-python 的全部 Issue

相似的 Issue

更多 Python Issue

把新 issue 寄到你的電子郵件信箱

精選適合新手參與的 GitHub issue 摘要。