)
EventBridge Pipes 完全仿真指南Floci 中的管道配置、Kafka 橋接與 Enrichment 實戰(zhàn)【免費下載鏈接】flociLight, fluffy, and always free - The AWS Local Emulator alternative項目地址: https://gitcode.com/gh_mirrors/fl/flociFlociLight, fluffy, and always free — AWS 本地模擬器替代品提供了一套與 AWS EventBridge Pipes 線協(xié)議兼容的 REST-JSON API讓你在本地的 SQS、Kinesis、DynamoDB Streams、Kafka 等源與 Lambda、SQS、SNS、Kinesis、Step Functions 等目標之間搭建可過濾、可富化Enrichment的事件管道。本指南以 docs/services/pipes.md 為主線結(jié)合 PipesController.java、PipesService.java、PipesPoller.java 等源碼幫助你掌握創(chuàng)建、描述、更新、啟停與刪除 Pipe 的全部操作理解 Kafka 源如何借助 Karapace 邊車工作以及 Enrichment、Filter、ParallelizationFactor 等參數(shù)的精確語義與當前限制。概覽協(xié)議、端點與支持的動作EventBridge Pipes 在 Floci 中通過REST-JSON協(xié)議暴露與需要X-Amz-Target的 JSON 1.1 協(xié)議不同固定端點為POST http://localhost:4566/從 PipesController.java 可以看到控制器使用標準 HTTP 動詞 JSON 請求體并統(tǒng)一以application/json生產(chǎn)/消費。端點全部掛載在/v1/pipes下與 AWS CLI 的pipes子命令一一對應(yīng)。支持的動作如下表ActionHTTP 路由說明CreatePipePOST /v1/pipes/{name}以 source、target 與可選 enrichment 創(chuàng)建新的管道DescribePipeGET /v1/pipes/{name}獲取管道詳情含狀態(tài)與配置UpdatePipePUT /v1/pipes/{name}更新管道配置source、target、role、enrichment、desired state 等DeletePipeDELETE /v1/pipes/{name}刪除管道ListPipesGET /v1/pipes列出管道支持按狀態(tài)與前綴過濾StartPipePOST /v1/pipes/{name}/start啟動已停止的管道StopPipePOST /v1/pipes/{name}/stop停止運行中的管道請求體字段采用 AWS 線格式的大駝峰命名Source、Target、RoleArn、Description、Enrichment、DesiredState、SourceParameters、TargetParameters、EnrichmentParameters、Tags?？刂破鲗@些字段逐個解析并透傳給 PipesService。快速上手SQS 到 Lambda 的管道準備環(huán)境先設(shè)置 AWS 端點環(huán)境變量指向本地運行的 Flociexport AWS_ENDPOINT_URLhttp://localhost:4566創(chuàng)建管道SQS → Lambda# Create a pipe (SQS to Lambda) aws pipes create-pipe \ --name my-pipe \ --source arn:aws:sqs:us-east-1:000000000000:source-queue \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --endpoint-url $AWS_ENDPOINT_URLCreatePipe有四個必填字段PipesService.createPipe 中會嚴格校驗Name管道名稱不能為空Source源 ARN不能為空Target目標 ARN不能為空RoleArnIAM 角色 ARN不能為空。任一缺失都會返回ValidationExceptionHTTP 400。若同名管道已存在按region::name作為存儲鍵則返回ConflictExceptionHTTP 409。創(chuàng)建時若未指定DesiredState默認值為RUNNING管道創(chuàng)建后會立即被 PipesPoller 調(diào)度輪詢。描述 / 列表 / 啟停 / 更新 / 刪除# Describe a pipe aws pipes describe-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # List all pipes aws pipes list-pipes \ --endpoint-url $AWS_ENDPOINT_URL # Start a pipe aws pipes start-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # Stop a pipe aws pipes stop-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # Update a pipe aws pipes update-pipe \ --name my-pipe \ --target arn:aws:lambda:us-east-1:000000000000:function:new-function \ --endpoint-url $AWS_ENDPOINT_URL # Delete a pipe aws pipes delete-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL幾個值得注意的細節(jié)describe-pipe返回完整的管道配置含SourceParameters、TargetParameters、EnrichmentParameters、Tags、CreationTime、LastModifiedTime等見 PipesController.buildPipeResponselist-pipes只返回摘要ARN、名稱、源、目標、期望狀態(tài)、當前狀態(tài)、時間戳不含SourceParameters與 AWS 行為一致見 buildPipeListEntrylist-pipes支持--name-prefix、--source-prefix、--target-prefix、--desired-state、--current-state過濾對應(yīng) PipesService.listPipesupdate-pipe采用“未提供的屬性保持原值”的語義只有顯式傳入的字段才會被覆蓋null 屬性不會清空既有值見 writePipeConfigurationdelete-pipe會先停止輪詢再刪除存儲記錄stop-pipe會取消輪詢定時器、清空 Kinesis/DynamoDB 迭代器并關(guān)閉 Kafka 消費者見 PipesPoller.stopPolling。Floci 的pipes服務(wù)還實現(xiàn)了資源標簽?zāi)芰agResource/untagResource/listTags并作為pipes:pipe資源類型接入 Resource Explorer見 PipesService.getSupportedResourceTypes。管道狀態(tài)機DesiredState與CurrentState兩個概念需要區(qū)分DesiredStateRUNNING或STOPPED表示期望狀態(tài)CurrentState實際狀態(tài)。PipeState 枚舉完整定義了以下狀態(tài)STARTING— 管道正在啟動RUNNING— 管道正在活躍處理事件STOPPING— 管道正在停止STOPPED— 管道已停止不再處理事件DELETED— 管道已被刪除源碼中枚舉還包含CREATING、UPDATING、DELETING、CREATE_FAILED、UPDATE_FAILED、START_FAILED、STOP_FAILED、DELETE_FAILED等完整狀態(tài)集與 AWS 線格式保持一致。狀態(tài)流轉(zhuǎn)邏輯在 PipesService 中創(chuàng)建時未指定DesiredState則默認為RUNNING此時CurrentState直接取RUNNING并啟動輪詢createPipe創(chuàng)建時指定--desired-state STOPPEDCurrentState即為STOPPED不會啟動輪詢start-pipe將 DesiredState 與 CurrentState 都置為RUNNING并恢復(fù)輪詢startPipestop-pipe先停止輪詢再將兩個狀態(tài)置為STOPPEDstopPipe。兼容性測試 pipes.bats 中驗證了--desired-state STOPPED創(chuàng)建后CurrentState STOPPED的行為。另外Floci 重啟后會自動恢復(fù)CurrentState RUNNING的管道輪詢startPersistedPollers保證模擬環(huán)境中斷后管道能繼續(xù)消費。支持的源與目標Floci 模擬的 EventBridge Pipes 支持以下源與目標類型由 PipesPoller.pollAndInvoke 依據(jù)源 ARN 中的服務(wù)段自動分派Sources源Amazon SQS 隊列arn:aws:sqs:...Amazon Kinesis 流arn:aws:kinesis:...Amazon DynamoDB 流arn:aws:dynamodb:...Kafka topics —— MSKarn:aws:kafka:...與自管理smk://協(xié)議Targets目標Lambda 函數(shù):lambda:或:function:SQS 隊列SNS 主題Kinesis 流Step Functions 狀態(tài)機:states:目標分派邏輯在 PipesTargetInvoker.invoke按 ARN 特征路由到對應(yīng)服務(wù)Lambda以RequestResponse同步調(diào)用若返回FunctionError視為投遞失敗并轉(zhuǎn)入 DLQinvokeLambdaSQS將 payload 作為消息體發(fā)送到隊列SNS以SubjectPipes發(fā)布到主題invokeSnsEventBridge向事件總線PutEvents默認Sourceaws.pipes、DetailTypePipeForwarded可通過EventBridgeEventBusParameters覆蓋invokeEventBridgeStep Functions以隨機執(zhí)行名pipes-uuid啟動執(zhí)行invokeStepFunctions。目標側(cè)還支持TargetParameters.InputTemplate模板中用$.path形式的 JSONPath 占位符投遞前替換為從事件中提取的值實現(xiàn)在 applyInputTemplate / extractJsonPath。Kafka 源與 Karapace REST 橋接Kafka 源MSK 或自管理smk://是 Floci Pipes 中最特殊的一條路徑它按需啟動一個 Karapace REST Proxy 邊車容器通過 HTTP 輪詢 Kafka而不是在 Floci 進程內(nèi)嵌入kafka-clients參見 KarapaceManager 的注釋關(guān)聯(lián) issue #2916。工作原理每個不同的bootstrap.servers目標對應(yīng)一個 Karapace 容器多個讀取同一集群的 Pipe共享這一個實例首次使用時ensureStarted按需啟動通過 SHA-256 摘要的前 8 個十六進制字符生成目標鍵并用該鍵構(gòu)造容器名pipes-kafka-rest-hash容器以KARAPACE_*環(huán)境變量配置KARAPACE_HOST0.0.0.0、KARAPACE_PORT8082、KARAPACE_BOOTSTRAP_URIbootstrap、KARAPACE_ADVERTISED_PROTOCOLhttp并顯式指定入口/venv/bin/karapace_rest_proxy鏡像默認入口是無操作的python3見 KarapaceManager.startContainer啟動后通過GET /_health探測就緒30 秒超時隨后 Floci 的 PipesKafkaRestClient 使用 Kafka REST Proxy v2 API 子集消費并提交偏移量地址改寫若bootstrap.servers是localhost/127.0.0.1會改寫為 Docker 主機可達地址host.docker.internal別名并在 Linux 原生 Docker 上顯式注入 host-gateway 映射見 resolveForContainer進程退出時PreDestroy統(tǒng)一停止并刪除全部 Karapace 容器。相關(guān)配置環(huán)境變量默認值說明FLOCI_SERVICES_PIPES_ENABLEDtrue啟用或禁用該服務(wù)FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_DEFAULT_IMAGEghcr.io/aiven-open/karapace:latestKafka 源管道按需啟動的 Karapace REST Proxy 邊車鏡像FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_HOST_PORT_BASE9500分配給 Karapace 邊車的主機端口段起點FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_HOST_PORT_MAX9599分配給 Karapace 邊車的主機端口段終點配置實現(xiàn)在 EmulatorConfig.PipesServiceConfig / KafkaRestBridgeConfigKafkaRestBridgeConfig定義了默認鏡像與端口段KarapaceManager 通過PortAllocator在[9500, 9599]區(qū)間分配主機端口容器內(nèi)運行時不綁定主機端口僅暴露 8082。重要前提帶 Kafka 源的管道依賴 Docker 環(huán)境因為需要啟動 Karapace 邊車其余源SQS/Kinesis/DynamoDB Streams不需要 Docker。Kafka 源的參數(shù)要求創(chuàng)建 Kafka 源管道時SourceParameters必須攜帶對應(yīng)的參數(shù)塊否則返回ValidationException見 validateSourceConfigurationsmk://前綴 → 必須提供SelfManagedKafkaParameters:kafka:ARN → 必須提供ManagedStreamingKafkaParameters且參數(shù)塊內(nèi)TopicName為必填項。示例aws pipes create-pipe \ --name my-kafka-pipe \ --source arn:aws:kafka:us-east-1:000000000000:cluster/my-msk/b4f3c2a1-0000-4000-8000-000000000000-9 \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters { ManagedStreamingKafkaParameters: { TopicName: orders, StartingPosition: TRIM_HORIZON } } \ --endpoint-url $AWS_ENDPOINT_URL自管理 Kafka 的smk://源同理但使用SelfManagedKafkaParameters并在其中給出BootstrapServers。ParallelizationFactor參數(shù)語義與當前限制CreatePipe與UpdatePipe在KinesisStreamParameters和DynamoDBStreamParameters源參數(shù)塊中接受1 到 10 之間的整數(shù)ParallelizationFactor與 AWS 線格式一致。DescribePipe會在SourceParameters中原樣回顯ListPipes只返回摘要、不含SourceParameters與 AWS 一致。aws pipes create-pipe \ --name kinesis-pipe \ --source arn:aws:kinesis:us-east-1:000000000000:stream/events \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters {KinesisStreamParameters:{StartingPosition:TRIM_HORIZON,ParallelizationFactor:4}} \ --endpoint-url $AWS_ENDPOINT_URL校驗邏輯嚴格對齊 AWSvalidateParallelizationFactor值超出 1~10 →ValidationException字段出現(xiàn)在與源不匹配的參數(shù)塊上例如 SQS 源攜帶KinesisStreamParameters.ParallelizationFactor→ValidationException非數(shù)字、非整數(shù)、超出 int 范圍的寬值如2^64 5也會被拒絕——源碼通過canConvertToInt()防止 BigInteger 溢出繞過邊界檢查。Enforcement status重要限制ParallelizationFactor目前僅持久化并在線格式上回顯輪詢器尚未按分片并發(fā)處理批次。Floci 對每個 Kinesis / DynamoDB Streams 管道只在單個分片shardId-000000000000上打開迭代器、每次只投遞一個批次無論配置值是多少見 initKinesisIterator 與 initDynamoDbIterator。多分片輪詢與真正的分片級并發(fā)被列為后續(xù)跟進項。Filter事件過濾與匹配語義管道支持 AWS 風(fēng)格的FilterCriteria過濾位于SourceParameters.FilterCriteria.Filters每個過濾條件包含PatternJSON 匹配模式。實現(xiàn)位于 PipesFilterMatcher多條 filter 之間是OR關(guān)系任一匹配即通過單條 filter 內(nèi)的多個字段是AND關(guān)系模式值必須是數(shù)組或?qū)ο髽肆磕Ｊ綍慌卸榉欠ú⒕芙^該記錄支持字符串精確匹配、數(shù)字精確匹配、null字段不存在、對象嵌套支持 AWS 特殊模式{prefix: ...}、{suffix: ...}、{equals-ignore-case: ...}、{anything-but: ...}含anything-butprefix、{exists: true|false}、{numeric: [, 5, , 1]}等。SQS 路徑下未匹配過濾條件的消息會被立即從源隊列刪除AWS 行為被過濾掉的消息視為已消費見 PipesPoller.pollSqs。Kafka 路徑下未匹配的記錄會推進提交偏移量相當于消費掉見 deliverKafkaRecords。EnrichmentLambda 富化與 DLQ 兜底管道的可選富化步驟source → filter → enrichment → target目前僅為 Lambda enrichment 提供模擬過濾后的批次被同步調(diào)用RequestResponse富化函數(shù)的響應(yīng)成為目標的輸入。實現(xiàn)在 PipesTargetInvoker.applyEnrichment。行為細則與 AWS 對齊空響應(yīng)跳過目標空 body、null、{}、[]都會消費源記錄而不調(diào)用目標非空數(shù)組如[{}]仍會調(diào)用目標攜帶一個空 payload 元素。注意源碼會先解析 JSON 以區(qū)分{ }、[ ]等空白變體Lambda 富化返回FunctionError會判定批次失敗源記錄被路由到管道的死信隊列DLQ而非靜默消費非 Lambda 富化類型API destinations、API Gateway、Step Functions Express在 AWS 上合法但 Floci不模擬配置了此類富化的管道會把批次判失敗并轉(zhuǎn)入 DLQ而不是投遞未富化的 payload富化目前只在 SQS 源路徑生效Kinesis、DynamoDB Streams 與 Kafka 源會把過濾后的記錄直接投遞給目標配置在這些源上的 enrichment 暫不生效見 PipesPoller.deliverRecords 的注釋。富化響應(yīng)到目標的形狀轉(zhuǎn)換富化返回的是單個 JSON 值非數(shù)組時代表一條事件會被包裝成單元素數(shù)組返回數(shù)組則視為已是批次、原樣轉(zhuǎn)發(fā)。該邏輯在 asEventArray。同時Lambda 目標接收的輸入保持{Records: [...]}批次形狀而 Step Functions、SQS、SNS、EventBridge 目標接收富化響應(yīng)的原始內(nèi)容數(shù)組包裝會破壞它們的輸入見 deliverEnrichedBatch。DLQ死信隊列配置管道的 DLQ 通過源參數(shù)塊中的DeadLetterConfig.Arn配置支持SqsQueueParameters、KinesisStreamParameters、DynamoDBStreamParameters三個塊getDlqArnaws pipes create-pipe \ --name my-pipe \ --source arn:aws:sqs:us-east-1:000000000000:source-queue \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters { SqsQueueParameters: { BatchSize: 10, DeadLetterConfig: {Arn: arn:aws:sqs:us-east-1:000000000000:dlq-queue} } } \ --endpoint-url $AWS_ENDPOINT_URL投遞失敗時sendToDeadLetterQueue 會把失敗 payload 發(fā)送到 DLQ若未配置 DLQ則返回投遞失敗計數(shù)并記錄告警日志。批次大小與輪詢行為默認BatchSize為 10可通過各源參數(shù)塊的BatchSize字段調(diào)整getBatchSize輪詢間隔固定為 1 秒POLL_INTERVAL_MS 1000通過 Vert.x 周期定時器驅(qū)動并使用獨立線程池執(zhí)行保證同一管道不會并發(fā)輪詢activePolls防重入見 PipesPollerKinesis 迭代器過期ExpiredIteratorException或 DynamoDB 流數(shù)據(jù)被裁剪TrimmedDataAccessException時會自動重置迭代器并重新拉取pollKinesis、pollDynamoDbStreamsKafka 路徑下REST Proxy 一次輪詢可能返回超過一個BatchSize的記錄因此一個輪詢周期會順序投遞多個批次一旦某分區(qū)投遞失敗該分區(qū)在后續(xù)批次中也不會提交偏移量保證“先失敗記錄重投”的順序語義deliverKafkaBatch。與 CloudFormation 的集成Pipes 服務(wù)實現(xiàn)了ResourceProvider接口支持 CloudFormation 對AWS::Pipes::Pipe資源的創(chuàng)建、更新與刪除提供器位于 cloudformation/provisioners 下的PipesCfnProvisioner并配套了回滾恢復(fù)測試 PipesCfnRollbackRestoresExactConfigurationTest.java 與 PipesCfnProvisionerTest.java。restorePipe路徑采用“null 即清空”的全量恢復(fù)語義確保 CloudFormation 回滾時能把管道精確還原到快照狀態(tài)PipesService.restorePipe。測試與驗證倉庫提供了多層測試覆蓋 pipes 服務(wù)單元測試PipesServiceTest.java、PipesPollerTest.java、PipesTargetInvokerTest.java、PipesFilterMatcherTest.java、KarapaceManagerTest.java、PipesKafkaRestClientTest.java集成測試PipesIntegrationTest.java、PipesPollerIntegrationTest.javaAWS CLI 兼容性測試pipes.bats 覆蓋 create/describe/list/update/start/stop 全流程SDK 各語言套件sdk-test-go、sdk-test-node、sdk-test-java、sdk-test-python中的pipes用例驗證多語言 SDK 兼容性。結(jié)語Floci 的 EventBridge Pipes 實現(xiàn)覆蓋了從創(chuàng)建、啟停、更新、刪除到列表過濾的完整生命周期并對 SQS/Kinesis/DynamoDB Streams/Kafka 四種源與 Lambda/SQS/SNS/Kinesis/Step Functions/EventBridge 多種目標提供事件搬運能力Kafka 源通過按需啟動的 Karapace REST 邊車實現(xiàn)零內(nèi)嵌依賴的輪詢。在使用時請?zhí)貏e注意當前仿真邊界ParallelizationFactor僅為線格式兼容單分片順序輪詢、enrichment 僅支持 Lambda 且僅作用于 SQS 源路徑、Kafka 源依賴 Docker。這些限制都以代碼注釋和文檔!!! note塊的形式明確標注便于你在搭建本地事件管道時做出準確預(yù)期?！久赓M下載鏈接】flociLight, fluffy, and always free - The AWS Local Emulator alternative項目地址: https://gitcode.com/gh_mirrors/fl/floci創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 EventBridge Pipes 完全仿真指南：Floci 中的管道配置、Kafka 橋接與 Enrichment 實戰(zhàn)
EventBridge Pipes 完全仿真指南Floci 中的管道配置、Kafka 橋接與 Enrichment 實戰(zhàn)【免費下載鏈接】flociLight, fluffy, and always free - The AWS Local Emulator alternative項目地址: https://gitcode.com/gh_mirrors/fl/flociFlociLight, fluffy, and always free — AWS 本地模擬器替代品提供了一套與 AWS EventBridge Pipes 線協(xié)議兼容的 REST-JSON API讓你在本地的 SQS、Kinesis、DynamoDB Streams、Kafka 等源與 Lambda、SQS、SNS、Kinesis、Step Functions 等目標之間搭建可過濾、可富化Enrichment的事件管道。本指南以 docs/services/pipes.md 為主線結(jié)合 PipesController.java、PipesService.java、PipesPoller.java 等源碼幫助你掌握創(chuàng)建、描述、更新、啟停與刪除 Pipe 的全部操作理解 Kafka 源如何借助 Karapace 邊車工作以及 Enrichment、Filter、ParallelizationFactor 等參數(shù)的精確語義與當前限制。概覽協(xié)議、端點與支持的動作EventBridge Pipes 在 Floci 中通過REST-JSON協(xié)議暴露與需要X-Amz-Target的 JSON 1.1 協(xié)議不同固定端點為POST http://localhost:4566/從 PipesController.java 可以看到控制器使用標準 HTTP 動詞 JSON 請求體并統(tǒng)一以application/json生產(chǎn)/消費。端點全部掛載在/v1/pipes下與 AWS CLI 的pipes子命令一一對應(yīng)。支持的動作如下表ActionHTTP 路由說明CreatePipePOST /v1/pipes/{name}以 source、target 與可選 enrichment 創(chuàng)建新的管道DescribePipeGET /v1/pipes/{name}獲取管道詳情含狀態(tài)與配置UpdatePipePUT /v1/pipes/{name}更新管道配置source、target、role、enrichment、desired state 等DeletePipeDELETE /v1/pipes/{name}刪除管道ListPipesGET /v1/pipes列出管道支持按狀態(tài)與前綴過濾StartPipePOST /v1/pipes/{name}/start啟動已停止的管道StopPipePOST /v1/pipes/{name}/stop停止運行中的管道請求體字段采用 AWS 線格式的大駝峰命名Source、Target、RoleArn、Description、Enrichment、DesiredState、SourceParameters、TargetParameters、EnrichmentParameters、Tags?？刂破鲗@些字段逐個解析并透傳給 PipesService。快速上手SQS 到 Lambda 的管道準備環(huán)境先設(shè)置 AWS 端點環(huán)境變量指向本地運行的 Flociexport AWS_ENDPOINT_URLhttp://localhost:4566創(chuàng)建管道SQS → Lambda# Create a pipe (SQS to Lambda) aws pipes create-pipe \ --name my-pipe \ --source arn:aws:sqs:us-east-1:000000000000:source-queue \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --endpoint-url $AWS_ENDPOINT_URLCreatePipe有四個必填字段PipesService.createPipe 中會嚴格校驗Name管道名稱不能為空Source源 ARN不能為空Target目標 ARN不能為空RoleArnIAM 角色 ARN不能為空。任一缺失都會返回ValidationExceptionHTTP 400。若同名管道已存在按region::name作為存儲鍵則返回ConflictExceptionHTTP 409。創(chuàng)建時若未指定DesiredState默認值為RUNNING管道創(chuàng)建后會立即被 PipesPoller 調(diào)度輪詢。描述 / 列表 / 啟停 / 更新 / 刪除# Describe a pipe aws pipes describe-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # List all pipes aws pipes list-pipes \ --endpoint-url $AWS_ENDPOINT_URL # Start a pipe aws pipes start-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # Stop a pipe aws pipes stop-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # Update a pipe aws pipes update-pipe \ --name my-pipe \ --target arn:aws:lambda:us-east-1:000000000000:function:new-function \ --endpoint-url $AWS_ENDPOINT_URL # Delete a pipe aws pipes delete-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL幾個值得注意的細節(jié)describe-pipe返回完整的管道配置含SourceParameters、TargetParameters、EnrichmentParameters、Tags、CreationTime、LastModifiedTime等見 PipesController.buildPipeResponselist-pipes只返回摘要ARN、名稱、源、目標、期望狀態(tài)、當前狀態(tài)、時間戳不含SourceParameters與 AWS 行為一致見 buildPipeListEntrylist-pipes支持--name-prefix、--source-prefix、--target-prefix、--desired-state、--current-state過濾對應(yīng) PipesService.listPipesupdate-pipe采用“未提供的屬性保持原值”的語義只有顯式傳入的字段才會被覆蓋null 屬性不會清空既有值見 writePipeConfigurationdelete-pipe會先停止輪詢再刪除存儲記錄stop-pipe會取消輪詢定時器、清空 Kinesis/DynamoDB 迭代器并關(guān)閉 Kafka 消費者見 PipesPoller.stopPolling。Floci 的pipes服務(wù)還實現(xiàn)了資源標簽?zāi)芰agResource/untagResource/listTags并作為pipes:pipe資源類型接入 Resource Explorer見 PipesService.getSupportedResourceTypes。管道狀態(tài)機DesiredState與CurrentState兩個概念需要區(qū)分DesiredStateRUNNING或STOPPED表示期望狀態(tài)CurrentState實際狀態(tài)。PipeState 枚舉完整定義了以下狀態(tài)STARTING— 管道正在啟動RUNNING— 管道正在活躍處理事件STOPPING— 管道正在停止STOPPED— 管道已停止不再處理事件DELETED— 管道已被刪除源碼中枚舉還包含CREATING、UPDATING、DELETING、CREATE_FAILED、UPDATE_FAILED、START_FAILED、STOP_FAILED、DELETE_FAILED等完整狀態(tài)集與 AWS 線格式保持一致。狀態(tài)流轉(zhuǎn)邏輯在 PipesService 中創(chuàng)建時未指定DesiredState則默認為RUNNING此時CurrentState直接取RUNNING并啟動輪詢createPipe創(chuàng)建時指定--desired-state STOPPEDCurrentState即為STOPPED不會啟動輪詢start-pipe將 DesiredState 與 CurrentState 都置為RUNNING并恢復(fù)輪詢startPipestop-pipe先停止輪詢再將兩個狀態(tài)置為STOPPEDstopPipe。兼容性測試 pipes.bats 中驗證了--desired-state STOPPED創(chuàng)建后CurrentState STOPPED的行為。另外Floci 重啟后會自動恢復(fù)CurrentState RUNNING的管道輪詢startPersistedPollers保證模擬環(huán)境中斷后管道能繼續(xù)消費。支持的源與目標Floci 模擬的 EventBridge Pipes 支持以下源與目標類型由 PipesPoller.pollAndInvoke 依據(jù)源 ARN 中的服務(wù)段自動分派Sources源Amazon SQS 隊列arn:aws:sqs:...Amazon Kinesis 流arn:aws:kinesis:...Amazon DynamoDB 流arn:aws:dynamodb:...Kafka topics —— MSKarn:aws:kafka:...與自管理smk://協(xié)議Targets目標Lambda 函數(shù):lambda:或:function:SQS 隊列SNS 主題Kinesis 流Step Functions 狀態(tài)機:states:目標分派邏輯在 PipesTargetInvoker.invoke按 ARN 特征路由到對應(yīng)服務(wù)Lambda以RequestResponse同步調(diào)用若返回FunctionError視為投遞失敗并轉(zhuǎn)入 DLQinvokeLambdaSQS將 payload 作為消息體發(fā)送到隊列SNS以SubjectPipes發(fā)布到主題invokeSnsEventBridge向事件總線PutEvents默認Sourceaws.pipes、DetailTypePipeForwarded可通過EventBridgeEventBusParameters覆蓋invokeEventBridgeStep Functions以隨機執(zhí)行名pipes-uuid啟動執(zhí)行invokeStepFunctions。目標側(cè)還支持TargetParameters.InputTemplate模板中用$.path形式的 JSONPath 占位符投遞前替換為從事件中提取的值實現(xiàn)在 applyInputTemplate / extractJsonPath。Kafka 源與 Karapace REST 橋接Kafka 源MSK 或自管理smk://是 Floci Pipes 中最特殊的一條路徑它按需啟動一個 Karapace REST Proxy 邊車容器通過 HTTP 輪詢 Kafka而不是在 Floci 進程內(nèi)嵌入kafka-clients參見 KarapaceManager 的注釋關(guān)聯(lián) issue #2916。工作原理每個不同的bootstrap.servers目標對應(yīng)一個 Karapace 容器多個讀取同一集群的 Pipe共享這一個實例首次使用時ensureStarted按需啟動通過 SHA-256 摘要的前 8 個十六進制字符生成目標鍵并用該鍵構(gòu)造容器名pipes-kafka-rest-hash容器以KARAPACE_*環(huán)境變量配置KARAPACE_HOST0.0.0.0、KARAPACE_PORT8082、KARAPACE_BOOTSTRAP_URIbootstrap、KARAPACE_ADVERTISED_PROTOCOLhttp并顯式指定入口/venv/bin/karapace_rest_proxy鏡像默認入口是無操作的python3見 KarapaceManager.startContainer啟動后通過GET /_health探測就緒30 秒超時隨后 Floci 的 PipesKafkaRestClient 使用 Kafka REST Proxy v2 API 子集消費并提交偏移量地址改寫若bootstrap.servers是localhost/127.0.0.1會改寫為 Docker 主機可達地址host.docker.internal別名并在 Linux 原生 Docker 上顯式注入 host-gateway 映射見 resolveForContainer進程退出時PreDestroy統(tǒng)一停止并刪除全部 Karapace 容器。相關(guān)配置環(huán)境變量默認值說明FLOCI_SERVICES_PIPES_ENABLEDtrue啟用或禁用該服務(wù)FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_DEFAULT_IMAGEghcr.io/aiven-open/karapace:latestKafka 源管道按需啟動的 Karapace REST Proxy 邊車鏡像FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_HOST_PORT_BASE9500分配給 Karapace 邊車的主機端口段起點FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_HOST_PORT_MAX9599分配給 Karapace 邊車的主機端口段終點配置實現(xiàn)在 EmulatorConfig.PipesServiceConfig / KafkaRestBridgeConfigKafkaRestBridgeConfig定義了默認鏡像與端口段KarapaceManager 通過PortAllocator在[9500, 9599]區(qū)間分配主機端口容器內(nèi)運行時不綁定主機端口僅暴露 8082。重要前提帶 Kafka 源的管道依賴 Docker 環(huán)境因為需要啟動 Karapace 邊車其余源SQS/Kinesis/DynamoDB Streams不需要 Docker。Kafka 源的參數(shù)要求創(chuàng)建 Kafka 源管道時SourceParameters必須攜帶對應(yīng)的參數(shù)塊否則返回ValidationException見 validateSourceConfigurationsmk://前綴 → 必須提供SelfManagedKafkaParameters:kafka:ARN → 必須提供ManagedStreamingKafkaParameters且參數(shù)塊內(nèi)TopicName為必填項。示例aws pipes create-pipe \ --name my-kafka-pipe \ --source arn:aws:kafka:us-east-1:000000000000:cluster/my-msk/b4f3c2a1-0000-4000-8000-000000000000-9 \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters { ManagedStreamingKafkaParameters: { TopicName: orders, StartingPosition: TRIM_HORIZON } } \ --endpoint-url $AWS_ENDPOINT_URL自管理 Kafka 的smk://源同理但使用SelfManagedKafkaParameters并在其中給出BootstrapServers。ParallelizationFactor參數(shù)語義與當前限制CreatePipe與UpdatePipe在KinesisStreamParameters和DynamoDBStreamParameters源參數(shù)塊中接受1 到 10 之間的整數(shù)ParallelizationFactor與 AWS 線格式一致。DescribePipe會在SourceParameters中原樣回顯ListPipes只返回摘要、不含SourceParameters與 AWS 一致。aws pipes create-pipe \ --name kinesis-pipe \ --source arn:aws:kinesis:us-east-1:000000000000:stream/events \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters {KinesisStreamParameters:{StartingPosition:TRIM_HORIZON,ParallelizationFactor:4}} \ --endpoint-url $AWS_ENDPOINT_URL校驗邏輯嚴格對齊 AWSvalidateParallelizationFactor值超出 1~10 →ValidationException字段出現(xiàn)在與源不匹配的參數(shù)塊上例如 SQS 源攜帶KinesisStreamParameters.ParallelizationFactor→ValidationException非數(shù)字、非整數(shù)、超出 int 范圍的寬值如2^64 5也會被拒絕——源碼通過canConvertToInt()防止 BigInteger 溢出繞過邊界檢查。Enforcement status重要限制ParallelizationFactor目前僅持久化并在線格式上回顯輪詢器尚未按分片并發(fā)處理批次。Floci 對每個 Kinesis / DynamoDB Streams 管道只在單個分片shardId-000000000000上打開迭代器、每次只投遞一個批次無論配置值是多少見 initKinesisIterator 與 initDynamoDbIterator。多分片輪詢與真正的分片級并發(fā)被列為后續(xù)跟進項。Filter事件過濾與匹配語義管道支持 AWS 風(fēng)格的FilterCriteria過濾位于SourceParameters.FilterCriteria.Filters每個過濾條件包含PatternJSON 匹配模式。實現(xiàn)位于 PipesFilterMatcher多條 filter 之間是OR關(guān)系任一匹配即通過單條 filter 內(nèi)的多個字段是AND關(guān)系模式值必須是數(shù)組或?qū)ο髽肆磕Ｊ綍慌卸榉欠ú⒕芙^該記錄支持字符串精確匹配、數(shù)字精確匹配、null字段不存在、對象嵌套支持 AWS 特殊模式{prefix: ...}、{suffix: ...}、{equals-ignore-case: ...}、{anything-but: ...}含anything-butprefix、{exists: true|false}、{numeric: [, 5, , 1]}等。SQS 路徑下未匹配過濾條件的消息會被立即從源隊列刪除AWS 行為被過濾掉的消息視為已消費見 PipesPoller.pollSqs。Kafka 路徑下未匹配的記錄會推進提交偏移量相當于消費掉見 deliverKafkaRecords。EnrichmentLambda 富化與 DLQ 兜底管道的可選富化步驟source → filter → enrichment → target目前僅為 Lambda enrichment 提供模擬過濾后的批次被同步調(diào)用RequestResponse富化函數(shù)的響應(yīng)成為目標的輸入。實現(xiàn)在 PipesTargetInvoker.applyEnrichment。行為細則與 AWS 對齊空響應(yīng)跳過目標空 body、null、{}、[]都會消費源記錄而不調(diào)用目標非空數(shù)組如[{}]仍會調(diào)用目標攜帶一個空 payload 元素。注意源碼會先解析 JSON 以區(qū)分{ }、[ ]等空白變體Lambda 富化返回FunctionError會判定批次失敗源記錄被路由到管道的死信隊列DLQ而非靜默消費非 Lambda 富化類型API destinations、API Gateway、Step Functions Express在 AWS 上合法但 Floci不模擬配置了此類富化的管道會把批次判失敗并轉(zhuǎn)入 DLQ而不是投遞未富化的 payload富化目前只在 SQS 源路徑生效Kinesis、DynamoDB Streams 與 Kafka 源會把過濾后的記錄直接投遞給目標配置在這些源上的 enrichment 暫不生效見 PipesPoller.deliverRecords 的注釋。富化響應(yīng)到目標的形狀轉(zhuǎn)換富化返回的是單個 JSON 值非數(shù)組時代表一條事件會被包裝成單元素數(shù)組返回數(shù)組則視為已是批次、原樣轉(zhuǎn)發(fā)。該邏輯在 asEventArray。同時Lambda 目標接收的輸入保持{Records: [...]}批次形狀而 Step Functions、SQS、SNS、EventBridge 目標接收富化響應(yīng)的原始內(nèi)容數(shù)組包裝會破壞它們的輸入見 deliverEnrichedBatch。DLQ死信隊列配置管道的 DLQ 通過源參數(shù)塊中的DeadLetterConfig.Arn配置支持SqsQueueParameters、KinesisStreamParameters、DynamoDBStreamParameters三個塊getDlqArnaws pipes create-pipe \ --name my-pipe \ --source arn:aws:sqs:us-east-1:000000000000:source-queue \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters { SqsQueueParameters: { BatchSize: 10, DeadLetterConfig: {Arn: arn:aws:sqs:us-east-1:000000000000:dlq-queue} } } \ --endpoint-url $AWS_ENDPOINT_URL投遞失敗時sendToDeadLetterQueue 會把失敗 payload 發(fā)送到 DLQ若未配置 DLQ則返回投遞失敗計數(shù)并記錄告警日志。批次大小與輪詢行為默認BatchSize為 10可通過各源參數(shù)塊的BatchSize字段調(diào)整getBatchSize輪詢間隔固定為 1 秒POLL_INTERVAL_MS 1000通過 Vert.x 周期定時器驅(qū)動并使用獨立線程池執(zhí)行保證同一管道不會并發(fā)輪詢activePolls防重入見 PipesPollerKinesis 迭代器過期ExpiredIteratorException或 DynamoDB 流數(shù)據(jù)被裁剪TrimmedDataAccessException時會自動重置迭代器并重新拉取pollKinesis、pollDynamoDbStreamsKafka 路徑下REST Proxy 一次輪詢可能返回超過一個BatchSize的記錄因此一個輪詢周期會順序投遞多個批次一旦某分區(qū)投遞失敗該分區(qū)在后續(xù)批次中也不會提交偏移量保證“先失敗記錄重投”的順序語義deliverKafkaBatch。與 CloudFormation 的集成Pipes 服務(wù)實現(xiàn)了ResourceProvider接口支持 CloudFormation 對AWS::Pipes::Pipe資源的創(chuàng)建、更新與刪除提供器位于 cloudformation/provisioners 下的PipesCfnProvisioner并配套了回滾恢復(fù)測試 PipesCfnRollbackRestoresExactConfigurationTest.java 與 PipesCfnProvisionerTest.java。restorePipe路徑采用“null 即清空”的全量恢復(fù)語義確保 CloudFormation 回滾時能把管道精確還原到快照狀態(tài)PipesService.restorePipe。測試與驗證倉庫提供了多層測試覆蓋 pipes 服務(wù)單元測試PipesServiceTest.java、PipesPollerTest.java、PipesTargetInvokerTest.java、PipesFilterMatcherTest.java、KarapaceManagerTest.java、PipesKafkaRestClientTest.java集成測試PipesIntegrationTest.java、PipesPollerIntegrationTest.javaAWS CLI 兼容性測試pipes.bats 覆蓋 create/describe/list/update/start/stop 全流程SDK 各語言套件sdk-test-go、sdk-test-node、sdk-test-java、sdk-test-python中的pipes用例驗證多語言 SDK 兼容性。結(jié)語Floci 的 EventBridge Pipes 實現(xiàn)覆蓋了從創(chuàng)建、啟停、更新、刪除到列表過濾的完整生命周期并對 SQS/Kinesis/DynamoDB Streams/Kafka 四種源與 Lambda/SQS/SNS/Kinesis/Step Functions/EventBridge 多種目標提供事件搬運能力Kafka 源通過按需啟動的 Karapace REST 邊車實現(xiàn)零內(nèi)嵌依賴的輪詢。在使用時請?zhí)貏e注意當前仿真邊界ParallelizationFactor僅為線格式兼容單分片順序輪詢、enrichment 僅支持 Lambda 且僅作用于 SQS 源路徑、Kafka 源依賴 Docker。這些限制都以代碼注釋和文檔!!! note塊的形式明確標注便于你在搭建本地事件管道時做出準確預(yù)期?！久赓M下載鏈接】flociLight, fluffy, and always free - The AWS Local Emulator alternative項目地址: https://gitcode.com/gh_mirrors/fl/floci創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考