Envoy SSE-To-Metadata 过滤器实战从流式响应中提取元数据并写入动态元数据【免费下载链接】envoyCloud-native high-performance edge/middle/service proxy项目地址: https://gitcode.com/GitHub_Trending/en/envoy导读本篇文章围绕 Envoy 的SSE-To-Metadatasse_to_metadataHTTP 过滤器展开讲解如何从 Server-Sent EventsSSEtext/event-stream流式响应体中提取关键值并将其写入动态元数据dynamic metadata。该过滤器尤其适合 LLM 类 API如 OpenAI的 token 用量追踪、日志增强、自定义过滤器消费与链路追踪等可观测性场景。读完本文你将掌握该过滤器的完整配置语法content parser 类型化扩展、selectors、on_present/on_missing/on_error 规则、max_event_size理解其底层工作流程与 SSE 协议兼容性细节并学会通过统计指标与性能选项对长连接流式场景进行调优。关联文档sse_to_metadata_filter.rst一、过滤器是什么架构与定位SSE-To-Metadata 过滤器用于从流式 HTTP 响应体中提取值并写入动态元数据。当前版本仅处理响应体response body即作用于 Envoy 的 encoder响应链路。它对可观测性、访问日志以及需要访问“只出现在流式响应中的值”的自定义过滤器尤其有价值。其类型 URL 为type.googleapis.com/envoy.extensions.filters.http.sse_to_metadata.v3.SseToMetadata类型化扩展架构过滤器与内容解析器解耦过滤器采用类型化扩展typed extension架构过滤器本身只负责 SSE 协议层面的解析事件切分、data 字段提取而**内容解析器content parser**负责 payload 格式的解析与取值如 JSON、XML、protobuf。这一设计让新增数据格式只需注册新的 parser 扩展无需改动过滤器主体。过滤器由三部分配置组成配置项作用content parser指定如何解析并提取事件 payload 中的值如 JSON parserrulescontent parser 内部的规则定义 selector 路径与元数据写入动作SSE 协议配置允许的 Content-Type、单事件大小上限max_event_size当某条规则匹配时提取到的值会被写入配置指定的元数据 namespace 与 key。随后这些元数据可被访问日志消费、自定义过滤器读取、导出到指标系统或附加到 trace span。底层实现对应关系在源码中过滤器的骨架定义在 filter.h核心逻辑在 filter.cc。过滤器继承自Http::PassThroughEncoderFilter只覆写encodeHeaders、encodeData、encodeTrailers三个编码器回调并持有一个由ContentParser::ParserFactory创建的 parser 实例以及一个Buffer::OwnedImpl用于缓冲跨 chunk 的不完整事件。消息结构定义于 sse_to_metadata.proto。二、典型应用场景LLM API 的可观测性与成本追踪场景一提取 LLM 流式响应的 token 用量OpenAI 等 LLM API 会在流式响应末尾返回 token 用量信息。该过滤器可以把 token 计数与模型名等元数据提取出来供日志、指标与可观测系统使用。完整示例配置见 sse-to-metadata-filter.yaml下文第三、四节会完整展开。核心部分http_filters: - name: envoy.filters.http.sse_to_metadata typed_config: type: type.googleapis.com/envoy.extensions.filters.http.sse_to_metadata.v3.SseToMetadata response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.lb key: tokens type: NUMBER - rule: selectors: - key: model on_present: metadata_namespace: envoy.lb key: model_name type: STRING stop_processing_after_matches: 1该示例从 SSE 流中提取total_tokens与model写入envoy.lb元数据 namespace。随后这些元数据可用于日志访问日志通过%DYNAMIC_METADATA(envoy.lb:tokens)%引用动态元数据指标导出自定义 stats sink 可消费该元数据自定义过滤器消费下游downstream过滤器可读取并据此动作链路追踪元数据可附加到 trace span。注意标准的 Envoyrate_limit过滤器在请求阶段收到响应之前执行因此无法直接消费从响应体中提取的元数据。若要做基于 token 的限流需要自定义过滤器在响应后上报用量或引入跨请求统计用量的配额管理系统。场景二从流式响应中提取多个值用于日志与监控还可以一次提取多个字段response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.audit key: tokens type: NUMBER - rule: selectors: - key: model on_present: metadata_namespace: envoy.audit key: model_name type: STRING三、工作流程从 SSE 流到动态元数据以 SSE JSON content parser 为例过滤器的工作流程如下Content-Type 校验过滤器检查响应Content-Type头是否为允许的类型text/event-stream。匹配只针对 media typetype/subtype忽略charset等参数。SSE 流解析按 SSE 规范 解析兼容 CRLF、CR、LF 三种行尾正确处理跨多个数据 chunk 拆分的事件。事件分发对每个完整 SSE 事件从data字段提取值交给配置的content parser。JSON 解析与导航JSON parser 将 data 解析为 JSON并按配置的 selectors 导航对象。例如selectors: [{key: usage}, {key: total_tokens}]提取json[usage][total_tokens]。按规则写元数据根据 content parser 内定义的规则执行写入on_present任意事件中 selector 成功取值时立即执行on_missing延迟到流结束执行。仅当on_present从未执行、且 selector 路径至少在某个事件中未被找到时触发on_error延迟到流结束执行。仅当on_present从未执行、且发生 JSON 解析错误时触发。若两个条件同时满足on_error 优先于 on_missing。延迟执行的用意on_missing/on_error的延迟执行确保早期事件缺少目标字段LLM 流中的常见情况不会破坏后续成功的提取。整流处理控制默认每条规则处理整个流stop_processing_after_matches: 0。将某条规则的stop_processing_after_matches设为1可让该规则在首次匹配后停止求值。只有所有规则都设置了匹配上限且全部达到上限时过滤器才会提前停止处理整个流详见第七节性能考量。源码级印证Content-Type 匹配逻辑位于 filter.cc先StringUtil::cropRight(content_type, ;)去掉参数再absl::EqualsIgnoreCase与text/event-stream比较因此text/event-stream; charsetutf-8也能命中。事件切分使用Http::Sse::SseParser::findEventEnd(buffer_view, end_stream)与parseEvent(event)相关接口定义在 sse_parser.h。事件处理在processSseEvent中完成先检查事件是否有非空data字段无则累加no_data_field计数并跳过再调用parser_-parse()解析错误累加parse_error计数成功则遍历 parser 返回的immediate_actions逐一写元数据filter.cc。流结束时调用finalizeRules()通过parser_-getAllDeferredActions()取出所有延迟动作on_error/on_missing并执行filter.cc。四、完整配置示例下面是该过滤器的完整可运行配置来源sse-to-metadata-filter.yaml包含监听器、HTTP Connection Manager、过滤器链与上游集群static_resources: listeners: - address: socket_address: address: 0.0.0.0 port_value: 8080 filter_chains: - filters: - name: envoy.filters.network.http_connection_manager typed_config: type: type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager codec_type: AUTO stat_prefix: ingress_http route_config: name: local_route virtual_hosts: - name: llm_service domains: - * routes: - match: prefix: /chat route: cluster: llm-cluster http_filters: - name: envoy.filters.http.sse_to_metadata typed_config: type: type.googleapis.com/envoy.extensions.filters.http.sse_to_metadata.v3.SseToMetadata response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.lb key: tokens type: NUMBER - rule: selectors: - key: model on_present: metadata_namespace: envoy.lb key: model_name type: STRING stop_processing_after_matches: 1 - name: envoy.filters.http.router typed_config: type: type.googleapis.com/envoy.extensions.filters.http.router.v3.Router clusters: - name: llm-cluster type: STRICT_DNS lb_policy: ROUND_ROBIN load_assignment: cluster_name: llm-cluster endpoints: - lb_endpoints: - endpoint: address: socket_address: address: api.openai.com port_value: 443 transport_socket: name: envoy.transport_sockets.tls typed_config: type: type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext sni: api.openai.com要点过滤器必须放置在 HTTP 过滤器链中且通常位于envoy.filters.http.router之前。示例中第一条规则提取末尾的 token 用量默认stop_processing_after_matches: 0取最后出现的值第二条规则提取model并设置stop_processing_after_matches: 1model 在早期事件中即出现首次匹配后即停止该规则的求值。五、核心配置项详解5.1 response_rules处理 SSE 响应流的配置容器包含以下子项。5.2 response_rules.content_parser一个 TypedExtensionConfig指定如何解析并从事件 payload 提取值。当前可用的 parserenvoy.content_parsers.json解析 JSON 内容并使用类 JSONPath 的 selectors 提取值。其配置选项见 json_content_parser.proto。5.3 JSON Content Parser 的 rules使用envoy.content_parsers.json时在 typed_config 内配置 rules。每条规则包含ruleselectors指定如何从 JSON payload 提取值的 selector 列表。每个 selector 的key表示 JSON 对象的一层嵌套。例如selectors: [{key: usage}, {key: total_tokens}]提取json_object[usage][total_tokens]。至少必须指定一个 selector。on_presentselector 成功提取值时写入的元数据匹配时立即执行。字段包括metadata_namespace元数据 namespace如envoy.lb。为空时默认取envoy.content_parsers.json该默认逻辑实现在 json_content_parser_impl.cckey元数据 keyvalue可选硬编码值。若设置则写入该值而非提取值type值类型PROTOBUF_VALUE、STRING或NUMBERpreserve_existing_metadata_value为 true 时不覆盖已存在的元数据默认 false。on_missingJSON 中找不到 selector 路径时写入的元数据。在流结束时执行前提是on_present从未执行。必须设置value作为 fallback如-1以保证下游消费者总能拿到元数据——这覆盖了“JSON 合法但缺少期望字段”的情况。on_errorJSON 解析出错时写入的元数据。在流结束时执行前提是on_present从未执行且优先级高于 on_missing。必须设置value作为 fallback如0以应对畸形 JSON。stop_processing_after_matches每条规则可选的流处理控制字段设为0默认处理所有内容项。后面的匹配会覆盖前面的值除非设置了preserve_existing_metadata_value即提取最后一次出现的值设为1该规则在首次成功匹配后停止求值。适用于在流早期就出现的值设为N 1保留给未来使用如聚合多个值。proto 校验限制了lte: 1即当前版本大于 1 的值会被拒绝以避免未来特性落地时改变既有行为。注意每条规则中on_present、on_missing、on_error至少指定一个。on_missing与on_error是延迟动作只在流结束且on_present从未执行时触发。这防止早期错误/缺失内容覆盖后续成功的提取——这正是 LLM 流中 usage 数据出现在末尾内容里的典型情况。5.4 response_rules.max_event_size单个事件在视为无效并被丢弃前的最大字节数。防止恶意或畸形流从不发送空行分隔符导致内存无界增长。默认8192 字节8KB设为0可禁用限制生产环境不建议最大允许值 10485760 字节10MBproto 校验见 sse_to_metadata.proto。5.5 类型转换原理源码级补充JSON parser 将提取到的 JSON 值转换为Protobuf::Value时遵循 json_content_parser_impl.cc 中的规则STRING一律转成字符串数字、布尔也会被字符串化如布尔转成true/falseNUMBER整型/浮点直接转数字布尔转1.0/0.0字符串会尝试用absl::SimpleAtod解析为数字解析失败则值保持未设置写入时会被丢弃并告警PROTOBUF_VALUE默认保留原始类型字符串保持字符串、数字保持数字、布尔保持布尔。此外若 JSON 中目标 key 对应的是嵌套对象而非标量extractValueFromJson会将其序列化为 JSON 字符串返回见 json_content_parser_impl.cc因此即使取值目标是子对象也能得到可用的字符串值。六、高级配置模式6.1 写入多个 namespace将同一个值写入多个元数据 namespace迁移期常用response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: old.namespace key: tokens type: NUMBER - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: new.namespace key: tokens type: NUMBER6.2 保留已存在的元数据避免覆盖先前写入的元数据值response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.lb key: tokens type: NUMBER preserve_existing_metadata_value: true底层逻辑在 filter.cc写入前先检查该 namespace 的 filter_metadata 中是否已存在同名 key若存在则累加preserved_existing_metadata计数并跳过写入。6.3 on_present / on_missing / on_error 组合使用提取失败时写入 fallback 值确保下游处理总能拿到元数据response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.lb key: tokens type: NUMBER on_missing: metadata_namespace: envoy.lb key: tokens value: number_value: -1 on_error: metadata_namespace: envoy.lb key: tokens value: number_value: 0该配置的行为成功提取时值写入元数据所有事件中都不存在usage.total_tokens路径时流结束时写入-1作为哨兵值JSON 解析失败时流结束时写入0作为安全默认值延迟执行保证错误/缺失状态不会覆盖后续事件的成功提取。延迟动作的优先级实现在 json_content_parser_impl.cc仅对从未匹配过ever_matched_为 false的规则生效若发生过解析错误且配置了on_error则走 on_error否则若 selector 未找到且配置了on_missing则走 on_missing——允许用户只配 on_missing 也能兜底解析错误场景。七、性能考量内存占用过滤器会在内存中缓冲不完整的 SSE 事件直到事件完整一旦发现完整事件立即处理并从缓冲区移除max_event_size默认 8KB防止无界内存增长。流处理优化默认stop_processing_after_matches: 0会让规则处理整个流的所有事件将规则设为stop_processing_after_matches: 1可在首次匹配后停止该规则的求值对于出现在流末尾的值如 LLM token 用量应保持默认的0。何时会提前终止整个流只有当所有规则都设置了stop_processing_after_matches 0且所有上限都已达到时过滤器才会提前停止处理 SSE 流从而避免解析剩余事件带来显著性能收益rules: - rule: selectors: [{ key: request_id }] on_present: { ... } stop_processing_after_matches: 1 # Stop after first match - rule: selectors: [{ key: model }] on_present: { ... } stop_processing_after_matches: 1 # Stop after first match此例中当request_id与model都在第一个事件中被提取后过滤器整体停止处理该流对长连接流可节省大量 CPU 与内存。停止条件的判定实现在 json_content_parser_impl.cc每次 parse 都计算all_rules_have_limits是否存在stop_processing_after_matches 0的规则与all_limited_rules_satisfied所有设限规则是否均已达到匹配次数二者同时为 true 时返回stop_processing true过滤器随即置processing_complete_并停止后续处理见 filter.cc。混合策略——性能收益有限当规则策略混用部分设限、部分不设限时性能收益微乎其微rules: - rule: selectors: [{ key: model }] on_present: { ... } stop_processing_after_matches: 1 # Extract first occurrence - rule: selectors: [{ key: usage }, { key: total_tokens }] on_present: { ... } # Default: 0 - extract last occurrence结果过滤器必须处理整个流才能拿到最终 token 计数。唯一的节省只是首次匹配后跳过modelselector 的求值相比 JSON 解析CPU 开销可忽略。建议对常见的 LLM 流式场景既要末尾的 token 用量又要早期的元数据过滤器无论如何都必须处理整个流。真正的性能收益来自所有元数据都能在早期提取完成的场景例如纯请求/响应关联而无需流末尾的值。八、统计指标SSE-To-Metadata 过滤器的统计位于http.stat_prefix.sse_to_metadata.resp.parser_prefix*命名空间。其中stat_prefix来自所属 HTTP Connection Manager 的配置见 HttpConnectionManager.stat_prefix 对应的 HCM 配置字段parser_prefix来自 content parserJSON parser 为json.定义于 json_content_parser_impl.cc 的statsPrefix()。例如使用 JSON parser 时指标位于http.stat_prefix.sse_to_metadata.resp.json.*。所有指标均为 Counter定义于 filter.h指标名类型说明resp.parser_prefix.metadata_addedCounter成功写入的元数据条目总数含提取值与 fallback 值resp.parser_prefix.metadata_from_fallbackCounter使用 on_missing 或 on_error fallback 值写入的元数据条目数metadata_added 的子集resp.parser_prefix.mismatched_content_typeCounterContent-Type 与期望类型不匹配的响应总数resp.parser_prefix.no_data_fieldCounter无 data 字段的 SSE 事件总数resp.parser_prefix.parse_errorCountercontent parser 解析 data 字段失败的事件总数resp.parser_prefix.preserved_existing_metadataCounter因preserve_existing_metadata_value为 true 而未写入元数据的次数resp.parser_prefix.event_too_largeCounter因超过max_event_size而被丢弃的事件总数九、SSE 规范兼容性过滤器实现了完整的 SSE 规范行尾支持 CRLF\r\n、CR\r、LF\n包括混用注释以:开头的行按规范正确忽略字段解析同时处理data: value与data:value冒号后有无空格多个 data 字段按规范用换行符正确拼接多条data:行字段顺序无论字段顺序如何都能正确处理事件分块传输处理跨多个 TCP 包/HTTP chunk 拆分的事件正确缓冲不完整事件。以上行为均可在 filter_test.cc 中找到对应单测例如CRLFLineEndings、CRLineEndings、MixedLineEndings、NoSpaceAfterColon、CommentLines、MultipleDataFieldsConcatenated、EventSplitAcrossChunks、EventSplitAcrossThreeChunks等用例覆盖行尾混用、跨 chunk 切分、注释行、冒号后无空格等边界情况。十、安全考量max_event_size默认 8KB防止恶意流从不发送事件分隔符导致无界内存增长事件超过max_event_size时缓冲数据被丢弃并累加event_too_large计数见 filter.cc生产环境建议将max_event_size保持在合理值默认 8KB 对大多数合法 SSE 事件足够将max_event_size: 0会禁用限制但不推荐用于不可信的上游源。延伸阅读过滤器 API 定义sse_to_metadata.protoJSON content parser API 定义json_content_parser.proto过滤器实现源码filter.h / filter.ccJSON parser 实现源码json_content_parser_impl.cc单元测试filter_test.cc、integration_test.cc、config_test.cc完整示例配置sse-to-metadata-filter.yaml【免费下载链接】envoyCloud-native high-performance edge/middle/service proxy项目地址: https://gitcode.com/GitHub_Trending/en/envoy创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Envoy SSE-To-Metadata 过滤器实战：从流式响应中提取元数据并写入动态元数据
Envoy SSE-To-Metadata 过滤器实战从流式响应中提取元数据并写入动态元数据【免费下载链接】envoyCloud-native high-performance edge/middle/service proxy项目地址: https://gitcode.com/GitHub_Trending/en/envoy导读本篇文章围绕 Envoy 的SSE-To-Metadatasse_to_metadataHTTP 过滤器展开讲解如何从 Server-Sent EventsSSEtext/event-stream流式响应体中提取关键值并将其写入动态元数据dynamic metadata。该过滤器尤其适合 LLM 类 API如 OpenAI的 token 用量追踪、日志增强、自定义过滤器消费与链路追踪等可观测性场景。读完本文你将掌握该过滤器的完整配置语法content parser 类型化扩展、selectors、on_present/on_missing/on_error 规则、max_event_size理解其底层工作流程与 SSE 协议兼容性细节并学会通过统计指标与性能选项对长连接流式场景进行调优。关联文档sse_to_metadata_filter.rst一、过滤器是什么架构与定位SSE-To-Metadata 过滤器用于从流式 HTTP 响应体中提取值并写入动态元数据。当前版本仅处理响应体response body即作用于 Envoy 的 encoder响应链路。它对可观测性、访问日志以及需要访问“只出现在流式响应中的值”的自定义过滤器尤其有价值。其类型 URL 为type.googleapis.com/envoy.extensions.filters.http.sse_to_metadata.v3.SseToMetadata类型化扩展架构过滤器与内容解析器解耦过滤器采用类型化扩展typed extension架构过滤器本身只负责 SSE 协议层面的解析事件切分、data 字段提取而**内容解析器content parser**负责 payload 格式的解析与取值如 JSON、XML、protobuf。这一设计让新增数据格式只需注册新的 parser 扩展无需改动过滤器主体。过滤器由三部分配置组成配置项作用content parser指定如何解析并提取事件 payload 中的值如 JSON parserrulescontent parser 内部的规则定义 selector 路径与元数据写入动作SSE 协议配置允许的 Content-Type、单事件大小上限max_event_size当某条规则匹配时提取到的值会被写入配置指定的元数据 namespace 与 key。随后这些元数据可被访问日志消费、自定义过滤器读取、导出到指标系统或附加到 trace span。底层实现对应关系在源码中过滤器的骨架定义在 filter.h核心逻辑在 filter.cc。过滤器继承自Http::PassThroughEncoderFilter只覆写encodeHeaders、encodeData、encodeTrailers三个编码器回调并持有一个由ContentParser::ParserFactory创建的 parser 实例以及一个Buffer::OwnedImpl用于缓冲跨 chunk 的不完整事件。消息结构定义于 sse_to_metadata.proto。二、典型应用场景LLM API 的可观测性与成本追踪场景一提取 LLM 流式响应的 token 用量OpenAI 等 LLM API 会在流式响应末尾返回 token 用量信息。该过滤器可以把 token 计数与模型名等元数据提取出来供日志、指标与可观测系统使用。完整示例配置见 sse-to-metadata-filter.yaml下文第三、四节会完整展开。核心部分http_filters: - name: envoy.filters.http.sse_to_metadata typed_config: type: type.googleapis.com/envoy.extensions.filters.http.sse_to_metadata.v3.SseToMetadata response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.lb key: tokens type: NUMBER - rule: selectors: - key: model on_present: metadata_namespace: envoy.lb key: model_name type: STRING stop_processing_after_matches: 1该示例从 SSE 流中提取total_tokens与model写入envoy.lb元数据 namespace。随后这些元数据可用于日志访问日志通过%DYNAMIC_METADATA(envoy.lb:tokens)%引用动态元数据指标导出自定义 stats sink 可消费该元数据自定义过滤器消费下游downstream过滤器可读取并据此动作链路追踪元数据可附加到 trace span。注意标准的 Envoyrate_limit过滤器在请求阶段收到响应之前执行因此无法直接消费从响应体中提取的元数据。若要做基于 token 的限流需要自定义过滤器在响应后上报用量或引入跨请求统计用量的配额管理系统。场景二从流式响应中提取多个值用于日志与监控还可以一次提取多个字段response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.audit key: tokens type: NUMBER - rule: selectors: - key: model on_present: metadata_namespace: envoy.audit key: model_name type: STRING三、工作流程从 SSE 流到动态元数据以 SSE JSON content parser 为例过滤器的工作流程如下Content-Type 校验过滤器检查响应Content-Type头是否为允许的类型text/event-stream。匹配只针对 media typetype/subtype忽略charset等参数。SSE 流解析按 SSE 规范 解析兼容 CRLF、CR、LF 三种行尾正确处理跨多个数据 chunk 拆分的事件。事件分发对每个完整 SSE 事件从data字段提取值交给配置的content parser。JSON 解析与导航JSON parser 将 data 解析为 JSON并按配置的 selectors 导航对象。例如selectors: [{key: usage}, {key: total_tokens}]提取json[usage][total_tokens]。按规则写元数据根据 content parser 内定义的规则执行写入on_present任意事件中 selector 成功取值时立即执行on_missing延迟到流结束执行。仅当on_present从未执行、且 selector 路径至少在某个事件中未被找到时触发on_error延迟到流结束执行。仅当on_present从未执行、且发生 JSON 解析错误时触发。若两个条件同时满足on_error 优先于 on_missing。延迟执行的用意on_missing/on_error的延迟执行确保早期事件缺少目标字段LLM 流中的常见情况不会破坏后续成功的提取。整流处理控制默认每条规则处理整个流stop_processing_after_matches: 0。将某条规则的stop_processing_after_matches设为1可让该规则在首次匹配后停止求值。只有所有规则都设置了匹配上限且全部达到上限时过滤器才会提前停止处理整个流详见第七节性能考量。源码级印证Content-Type 匹配逻辑位于 filter.cc先StringUtil::cropRight(content_type, ;)去掉参数再absl::EqualsIgnoreCase与text/event-stream比较因此text/event-stream; charsetutf-8也能命中。事件切分使用Http::Sse::SseParser::findEventEnd(buffer_view, end_stream)与parseEvent(event)相关接口定义在 sse_parser.h。事件处理在processSseEvent中完成先检查事件是否有非空data字段无则累加no_data_field计数并跳过再调用parser_-parse()解析错误累加parse_error计数成功则遍历 parser 返回的immediate_actions逐一写元数据filter.cc。流结束时调用finalizeRules()通过parser_-getAllDeferredActions()取出所有延迟动作on_error/on_missing并执行filter.cc。四、完整配置示例下面是该过滤器的完整可运行配置来源sse-to-metadata-filter.yaml包含监听器、HTTP Connection Manager、过滤器链与上游集群static_resources: listeners: - address: socket_address: address: 0.0.0.0 port_value: 8080 filter_chains: - filters: - name: envoy.filters.network.http_connection_manager typed_config: type: type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager codec_type: AUTO stat_prefix: ingress_http route_config: name: local_route virtual_hosts: - name: llm_service domains: - * routes: - match: prefix: /chat route: cluster: llm-cluster http_filters: - name: envoy.filters.http.sse_to_metadata typed_config: type: type.googleapis.com/envoy.extensions.filters.http.sse_to_metadata.v3.SseToMetadata response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.lb key: tokens type: NUMBER - rule: selectors: - key: model on_present: metadata_namespace: envoy.lb key: model_name type: STRING stop_processing_after_matches: 1 - name: envoy.filters.http.router typed_config: type: type.googleapis.com/envoy.extensions.filters.http.router.v3.Router clusters: - name: llm-cluster type: STRICT_DNS lb_policy: ROUND_ROBIN load_assignment: cluster_name: llm-cluster endpoints: - lb_endpoints: - endpoint: address: socket_address: address: api.openai.com port_value: 443 transport_socket: name: envoy.transport_sockets.tls typed_config: type: type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext sni: api.openai.com要点过滤器必须放置在 HTTP 过滤器链中且通常位于envoy.filters.http.router之前。示例中第一条规则提取末尾的 token 用量默认stop_processing_after_matches: 0取最后出现的值第二条规则提取model并设置stop_processing_after_matches: 1model 在早期事件中即出现首次匹配后即停止该规则的求值。五、核心配置项详解5.1 response_rules处理 SSE 响应流的配置容器包含以下子项。5.2 response_rules.content_parser一个 TypedExtensionConfig指定如何解析并从事件 payload 提取值。当前可用的 parserenvoy.content_parsers.json解析 JSON 内容并使用类 JSONPath 的 selectors 提取值。其配置选项见 json_content_parser.proto。5.3 JSON Content Parser 的 rules使用envoy.content_parsers.json时在 typed_config 内配置 rules。每条规则包含ruleselectors指定如何从 JSON payload 提取值的 selector 列表。每个 selector 的key表示 JSON 对象的一层嵌套。例如selectors: [{key: usage}, {key: total_tokens}]提取json_object[usage][total_tokens]。至少必须指定一个 selector。on_presentselector 成功提取值时写入的元数据匹配时立即执行。字段包括metadata_namespace元数据 namespace如envoy.lb。为空时默认取envoy.content_parsers.json该默认逻辑实现在 json_content_parser_impl.cckey元数据 keyvalue可选硬编码值。若设置则写入该值而非提取值type值类型PROTOBUF_VALUE、STRING或NUMBERpreserve_existing_metadata_value为 true 时不覆盖已存在的元数据默认 false。on_missingJSON 中找不到 selector 路径时写入的元数据。在流结束时执行前提是on_present从未执行。必须设置value作为 fallback如-1以保证下游消费者总能拿到元数据——这覆盖了“JSON 合法但缺少期望字段”的情况。on_errorJSON 解析出错时写入的元数据。在流结束时执行前提是on_present从未执行且优先级高于 on_missing。必须设置value作为 fallback如0以应对畸形 JSON。stop_processing_after_matches每条规则可选的流处理控制字段设为0默认处理所有内容项。后面的匹配会覆盖前面的值除非设置了preserve_existing_metadata_value即提取最后一次出现的值设为1该规则在首次成功匹配后停止求值。适用于在流早期就出现的值设为N 1保留给未来使用如聚合多个值。proto 校验限制了lte: 1即当前版本大于 1 的值会被拒绝以避免未来特性落地时改变既有行为。注意每条规则中on_present、on_missing、on_error至少指定一个。on_missing与on_error是延迟动作只在流结束且on_present从未执行时触发。这防止早期错误/缺失内容覆盖后续成功的提取——这正是 LLM 流中 usage 数据出现在末尾内容里的典型情况。5.4 response_rules.max_event_size单个事件在视为无效并被丢弃前的最大字节数。防止恶意或畸形流从不发送空行分隔符导致内存无界增长。默认8192 字节8KB设为0可禁用限制生产环境不建议最大允许值 10485760 字节10MBproto 校验见 sse_to_metadata.proto。5.5 类型转换原理源码级补充JSON parser 将提取到的 JSON 值转换为Protobuf::Value时遵循 json_content_parser_impl.cc 中的规则STRING一律转成字符串数字、布尔也会被字符串化如布尔转成true/falseNUMBER整型/浮点直接转数字布尔转1.0/0.0字符串会尝试用absl::SimpleAtod解析为数字解析失败则值保持未设置写入时会被丢弃并告警PROTOBUF_VALUE默认保留原始类型字符串保持字符串、数字保持数字、布尔保持布尔。此外若 JSON 中目标 key 对应的是嵌套对象而非标量extractValueFromJson会将其序列化为 JSON 字符串返回见 json_content_parser_impl.cc因此即使取值目标是子对象也能得到可用的字符串值。六、高级配置模式6.1 写入多个 namespace将同一个值写入多个元数据 namespace迁移期常用response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: old.namespace key: tokens type: NUMBER - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: new.namespace key: tokens type: NUMBER6.2 保留已存在的元数据避免覆盖先前写入的元数据值response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.lb key: tokens type: NUMBER preserve_existing_metadata_value: true底层逻辑在 filter.cc写入前先检查该 namespace 的 filter_metadata 中是否已存在同名 key若存在则累加preserved_existing_metadata计数并跳过写入。6.3 on_present / on_missing / on_error 组合使用提取失败时写入 fallback 值确保下游处理总能拿到元数据response_rules: content_parser: name: envoy.content_parsers.json typed_config: type: type.googleapis.com/envoy.extensions.content_parsers.json.v3.JsonContentParser rules: - rule: selectors: - key: usage - key: total_tokens on_present: metadata_namespace: envoy.lb key: tokens type: NUMBER on_missing: metadata_namespace: envoy.lb key: tokens value: number_value: -1 on_error: metadata_namespace: envoy.lb key: tokens value: number_value: 0该配置的行为成功提取时值写入元数据所有事件中都不存在usage.total_tokens路径时流结束时写入-1作为哨兵值JSON 解析失败时流结束时写入0作为安全默认值延迟执行保证错误/缺失状态不会覆盖后续事件的成功提取。延迟动作的优先级实现在 json_content_parser_impl.cc仅对从未匹配过ever_matched_为 false的规则生效若发生过解析错误且配置了on_error则走 on_error否则若 selector 未找到且配置了on_missing则走 on_missing——允许用户只配 on_missing 也能兜底解析错误场景。七、性能考量内存占用过滤器会在内存中缓冲不完整的 SSE 事件直到事件完整一旦发现完整事件立即处理并从缓冲区移除max_event_size默认 8KB防止无界内存增长。流处理优化默认stop_processing_after_matches: 0会让规则处理整个流的所有事件将规则设为stop_processing_after_matches: 1可在首次匹配后停止该规则的求值对于出现在流末尾的值如 LLM token 用量应保持默认的0。何时会提前终止整个流只有当所有规则都设置了stop_processing_after_matches 0且所有上限都已达到时过滤器才会提前停止处理 SSE 流从而避免解析剩余事件带来显著性能收益rules: - rule: selectors: [{ key: request_id }] on_present: { ... } stop_processing_after_matches: 1 # Stop after first match - rule: selectors: [{ key: model }] on_present: { ... } stop_processing_after_matches: 1 # Stop after first match此例中当request_id与model都在第一个事件中被提取后过滤器整体停止处理该流对长连接流可节省大量 CPU 与内存。停止条件的判定实现在 json_content_parser_impl.cc每次 parse 都计算all_rules_have_limits是否存在stop_processing_after_matches 0的规则与all_limited_rules_satisfied所有设限规则是否均已达到匹配次数二者同时为 true 时返回stop_processing true过滤器随即置processing_complete_并停止后续处理见 filter.cc。混合策略——性能收益有限当规则策略混用部分设限、部分不设限时性能收益微乎其微rules: - rule: selectors: [{ key: model }] on_present: { ... } stop_processing_after_matches: 1 # Extract first occurrence - rule: selectors: [{ key: usage }, { key: total_tokens }] on_present: { ... } # Default: 0 - extract last occurrence结果过滤器必须处理整个流才能拿到最终 token 计数。唯一的节省只是首次匹配后跳过modelselector 的求值相比 JSON 解析CPU 开销可忽略。建议对常见的 LLM 流式场景既要末尾的 token 用量又要早期的元数据过滤器无论如何都必须处理整个流。真正的性能收益来自所有元数据都能在早期提取完成的场景例如纯请求/响应关联而无需流末尾的值。八、统计指标SSE-To-Metadata 过滤器的统计位于http.stat_prefix.sse_to_metadata.resp.parser_prefix*命名空间。其中stat_prefix来自所属 HTTP Connection Manager 的配置见 HttpConnectionManager.stat_prefix 对应的 HCM 配置字段parser_prefix来自 content parserJSON parser 为json.定义于 json_content_parser_impl.cc 的statsPrefix()。例如使用 JSON parser 时指标位于http.stat_prefix.sse_to_metadata.resp.json.*。所有指标均为 Counter定义于 filter.h指标名类型说明resp.parser_prefix.metadata_addedCounter成功写入的元数据条目总数含提取值与 fallback 值resp.parser_prefix.metadata_from_fallbackCounter使用 on_missing 或 on_error fallback 值写入的元数据条目数metadata_added 的子集resp.parser_prefix.mismatched_content_typeCounterContent-Type 与期望类型不匹配的响应总数resp.parser_prefix.no_data_fieldCounter无 data 字段的 SSE 事件总数resp.parser_prefix.parse_errorCountercontent parser 解析 data 字段失败的事件总数resp.parser_prefix.preserved_existing_metadataCounter因preserve_existing_metadata_value为 true 而未写入元数据的次数resp.parser_prefix.event_too_largeCounter因超过max_event_size而被丢弃的事件总数九、SSE 规范兼容性过滤器实现了完整的 SSE 规范行尾支持 CRLF\r\n、CR\r、LF\n包括混用注释以:开头的行按规范正确忽略字段解析同时处理data: value与data:value冒号后有无空格多个 data 字段按规范用换行符正确拼接多条data:行字段顺序无论字段顺序如何都能正确处理事件分块传输处理跨多个 TCP 包/HTTP chunk 拆分的事件正确缓冲不完整事件。以上行为均可在 filter_test.cc 中找到对应单测例如CRLFLineEndings、CRLineEndings、MixedLineEndings、NoSpaceAfterColon、CommentLines、MultipleDataFieldsConcatenated、EventSplitAcrossChunks、EventSplitAcrossThreeChunks等用例覆盖行尾混用、跨 chunk 切分、注释行、冒号后无空格等边界情况。十、安全考量max_event_size默认 8KB防止恶意流从不发送事件分隔符导致无界内存增长事件超过max_event_size时缓冲数据被丢弃并累加event_too_large计数见 filter.cc生产环境建议将max_event_size保持在合理值默认 8KB 对大多数合法 SSE 事件足够将max_event_size: 0会禁用限制但不推荐用于不可信的上游源。延伸阅读过滤器 API 定义sse_to_metadata.protoJSON content parser API 定义json_content_parser.proto过滤器实现源码filter.h / filter.ccJSON parser 实现源码json_content_parser_impl.cc单元测试filter_test.cc、integration_test.cc、config_test.cc完整示例配置sse-to-metadata-filter.yaml【免费下载链接】envoyCloud-native high-performance edge/middle/service proxy项目地址: https://gitcode.com/GitHub_Trending/en/envoy创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
