创建无服务器管道以提供可靠消息传递

作者：Chris McPeek 发表于2024年3月6日，内容包含在 、[Amazon Simple NotificationService (SNS)](https://aws.amazon.com/blogs/compute/category/messaging/amazon- simple-notification-service-sns/)、[Amazon Simple Queue Service (SQS)](https://aws.amazon.com/blogs/compute/category/messaging/amazon-simple- queue-service-sqs/)、 和 相关信息。

关键要点

• 审计系统需要跟踪全方位输入以防止欺诈和检测模型漂移。
• 使用无服务器架构可以创建可靠、可追踪和持久性的审计处理管道。
• 实现高性能和数据耐久性的四个基本需求。
• 检查和测试解决方案过程，确保消息的持久性和可靠性。

本文由 Jeff Harman（高级原型架构师）、Vaibhav Shah（高级解决方案架构师）和 Erik Olsen（高级技术客户经理）撰写。

很多行业需要为决策和交易系统提供审计跟踪。人工智能辅助的决策过程需要近实时监控决策系统的全部输入，以防止欺诈、检测模型漂移和歧视。现代系统通常使用更广泛的输入进行决策，包括图像、非结构化文本、历史数值和其他大型数据元素。这些大型数据元素对传统审计系统构成挑战，因为传统系统主要处理相对较小的结构化文本消息。本文展示了如何使用无服务器技术构建一个可靠、高性能、可追溯、耐用的审计处理管道。

概述

在开发审计记录摄取架构时，需要考虑以下四个需求：

下图展示了满足这些需求的架构，并模拟了审计记录在系统中的流动。

端点发起 API 调用。一个 函数接收该消息，并由 集群提供低延迟的初始存储机制。一旦数据存储在 ElastiCache 中， 工作流便负责协调通信和持久性功能。

订阅者会收到四个 Amazon Simple Notification Service (SNS) 通知，通知内容包括审计记录有效载荷的到达和存储、审计记录元数据的存储，以及审计记录归档的完成。用户可以将 Amazon Simple QueueService (SQS) 队列订阅到 SNS 主题，并利用 机制实现高可靠性。

1. 摄取消息的 Lambda 函数发送初始收据通知
2. 消息归档处理器 Lambda 函数通知有效载荷从 ElastiCache 储存到 Amazon Simple Storage Service (S3)
3. 消息元数据处理器 Lambda 函数通知将消息元数据存储到
4. 最终状态聚合 Lambda 函数通知审计记录已经归档。

三个基本处理步骤中的任何故障：摄取、数据归档和元数据归档，将触发一个 SQS死信队列（DLQ）中的消息，消息会包含原始请求和故障原因的说明。若摄取消息功能失败，则调用摄取消息失败处理函数，该函数会将原始参数存储到 S3失败消息存储桶中以供后续分析。

Step Functions工作流提供系统的编排和并行路径执行。下面是详细的工作流，展示了执行流程和通知操作。转换步骤将内部数据结构转换为消费者所需的格式。

。

摄取消息的 Lambda 函数生成一个消息 ID，用于跟踪消息有效载荷的生命周期。它将完整消息存储在 ElastiCache for Redis缓存中。摄取消息 Lambda 函数生成包含上述所有必要元素的内部消息。最后，Lambda 函数处理代码启动 Step Functions工作流，并传入内部消息有效载荷。

如果摄取消息的 Lambda 函数由于任何原因失败，该 Lambda 函数会调用摄取失败处理 Lambda 函数。此 Lambda 函数会将任何可恢复的 incoming 消息数据写入 S3 存储桶，并在摄取消息死信队列中发送通知。

然后 Step Functions 工作流会并行运行三个过程。

• 工作流触发消息归档数据处理 Lambda 函数，将消息数据从 ElastiCache 缓存持久化到 S3 存储桶。存储后，Lambda 函数返回 S3 存储桶引用和状态信息。可以选择两种方式从缓存中移除内部消息：在发送内部消息时立即从缓存中移除并更新 ElastiCache 缓存标志，或等待 ElastiCache 生命周期移除过期消息。本解决方案选择等待 ElastiCache 生命周期移除消息。
• 工作流触发消息元数据处理 Lambda 函数，将所有消息元数据和安全信息写入 DynamoDB。Lambda 函数回复 DynamoDB 引用信息。
• 最后，Step Functions 工作流向 SNS 主题发送消息，通知订阅者消息已到达，并且数据持久化过程已开始。

在每个 Lambda 函数处理完成后，Lambda 函数会向 SNS 通知主题发送通知，提醒订阅者每个操作已完成。当消息元数据和消息归档 Lambda函数完成后，最终聚合函数会对 DynamoDB 中的元数据进行最终更新，包括 S3 引用信息，移除 ElastiCache Redis 引用。

部署解决方案

前提条件：

1. 安装 AWS Serverless Application Model (AWS SAM)（请参见 ）
2. 具有适当权限以在目标 AWS 账户中运行 模板的 AWS 用户/凭证
3. Python 3.8 – 3.10
4. 安装 AWS SDK for Python (Boto3)
5. 安装 Python 库

此实现的源代码可以在 https://github.com/aws-samples/blog-serverless-reliable-messaging 找到。

安装解决方案：

1. 将 git 仓库克隆到本地目录
2. git clone https://github.com/aws-samples/blog-serverless-reliable-messaging.git
3. 进入由克隆操作创建的目录，通常为 blog_serverless_reliable_messaging
4. 执行命令：sam build
5. 执行命令：sam deploy --guided。系统会询问您提供以下参数：
6. 堆栈名称：此部署的名称（示例：serverless-messaging）
7. AWS 区域：要部署的区域（示例：us-east-1）
8. ElasticacheInstanceClass：要使用的 EC2 缓存实例类型（示例：cache.t3.small）
9. ElasticReplicaCount：ElastiCache 中应使用的副本数量（推荐最小值：2）
10. ProjectName：用于命名账户中的资源（示例：serverless-messaging）
11. MultiAZ：是否使用多个可用区的 True/False（建议：True）
12. 其余问题可以选择默认参数

测试：

部署堆栈后，您可以使用与部署输出中引用的 API 密钥相对应的 API Gateway 端点进行测试。可以通过 AWS 控制台（通过输出中的链接 - ApiKeyConsole）或通过 AWS CLI （通过输出中的 AWS CLI 引用 - APIKeyCLI）获取 API 密钥。

您还可以通过 Lambda 服务控制台直接测试，调用摄取消息功能。

项目根目录中有一个测试消息文件 test_message.json，用于直接测试摄取函数的 Lambda 功能。

1. 在控制台中导航到 Lambda 服务
2. 从可用函数列表中，选择 "<项目名称> -IngestMessageFunction-xxxxx" 函数
3. 在“函数概述”中选择“测试”选项卡
4. 输入您选择的事件名称
5. 将 test_message.json 的内容复制并粘贴到“事件 JSON”框中
6. 单击“保存”，然后在成功保存后单击“测试”
7. 如果成功，您应该在详细信息中看到以下类似内容：

json { "isBase64Encoded": false, "statusCode": 200, "headers": { "Access- Control-Allow-Headers": "Content-Type", "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "OPTIONS,POST" }, "body": "{\"messageID\": \"XXXXXXXXXXXXXX\"}" }

8. 在 S3 存储桶“<项目名称>-s3messagearchive-xxxxxx”中，找到原始 json 有效载荷，其键基于脚本执行的日期和时间，例如：YEAR/MONTH/DAY/HOUR/MINUTE，文件名称为 messageID
9. 在名为 metaDataTable 的 DynamoDB 表中，您应该找到一条记录，其 messageID 与上述 messageID 相等，并包含与有效载荷相关的所有元数据

测试文件夹中包含一个 Python 脚本：

1. 替换 test_client.py 文件中的 <Your API key key here> 和 <Your API Gateway URL here (IngestMessageApi)> 值，以正确对应您的环境
2. 使用安装了 requests 包的 Python 3.8 或更高版本执行测试脚本 示例执行（从 git 克隆的主目录）：
   python3 -m pip install -r ./test_client/requirements.txt python3 ./test_client/test_client.py
3. 成功的输出显示 messageID 和头的 JSON 有效载荷：

json { "messageID": " XXXXXXXXXXXXXX" }

4. 在 S3 存储桶“<项目名称>-s3messagearchive-xxxxxx”中，应该能够找到原始 json 有效载荷，其键基于脚本执行的日期和时间，例如：YEAR/MONTH/DAY/HOUR/MINUTE，文件名称为 messageID
5. 在名为 metaDataTable 的 DynamoDB 表中，您应该找到一条记录，其 messageID 与上述 messageID 相等，并包含与有效载荷相关的所有元数据

结论

本文描述了支持大型消息的高可靠性消息系统的架构模式、消息模式和数据结构。使用包括 Lambda 函数、StepFunctions、ElastiCache、DynamoDB 和 S3的无服务器服务，满足现代审计系统可扩展和可靠的要求。本文分享的架构适用于高度监管的环境，可以存储和跟踪比典型日志系统更大的消息，记录大小介于 256k 到 6MB 之间。该架构可作为蓝图，进一步扩展和适应其他无服务器用例。

为了获取无服务器学习资源，请访问 。

标签：，