add aws_eventstream_parser filter for aws stream parsing

[cqi1217]

Commit Message: Add aws_eventstream_to_metadata filter
Additional Description:
This PR is adding a new HTTP filter for extracting values from AWS eventstream responses and writing them into dynamic metadata. This filter is similar to sse_to_metadata filter(#42762).

Key Architecture

• Filter Layer (envoy.filters.http.sse_to_metadata): Handles SSE protocol parsing(aws: Adds extension common util for parsing AWS eventstream #43544), buffering, and metadata writing

• Parser Interface (envoy.content_parser): Generic interface for parsing SSE event data fields

• JSON Parser (envoy.content_parsers.json): First implementation supporting JSON path-based extraction with on_present/on_missing/on_error semantics

Risk Level: Low (new http filter)
Testing: unit tests and integration tests
Docs Changes: Added documentation for aws_eventstream_to_matadata filter
Release Notes:
Fixes #43492

[repokitteh-read-only]

CC @envoyproxy/api-shepherds: Your approval is needed for changes made to (api/envoy/|docs/root/api-docs/).
envoyproxy/api-shepherds assignee is @markdroth
CC @envoyproxy/api-watchers: FYI only for changes made to (api/envoy/|docs/root/api-docs/).

🐱

Caused by: #44094 was opened by cqi1217.

see: more, trace.

[wbpcode]

What's the difference between this PR and the sse_to_metadata? And which maintainer will sponsor this PR? Note, any new extension will require a maintainer to sponsor. See https://github.com/envoyproxy/envoy/blob/main/EXTENSION_POLICY.md

[cqi1217]

@wbpcode the sse_to_metadata filter parse the standard SSE response, which is used by openAI, vertexAI and others. The aws_eventstream_to_metadata parse the AWS eventstream response, whicih is used by AWS bedrock. The maintainer could be @JuniorHsu @PeterL328

[JuniorHsu]

@wbpcode the sse_to_metadata filter parse the standard SSE response, which is used by openAI, vertexAI and others. The aws_eventstream_to_metadata parse the AWS eventstream response, whicih is used by AWS bedrock. The maintainer could be @JuniorHsu @PeterL328

It needs maintainer on the top two lists
https://github.com/envoyproxy/envoy/blob/main/OWNERS.md

Hello @tyxia, this is another AI related parser in HTTP filter chain. I noticed that you show great interest in SSE filter. Are we able to get your sponsor here so we could share the filter to open source? Peter and/or I could do the first pass to reduce the review effort.

[tyxia]

@wbpcode the sse_to_metadata filter parse the standard SSE response, which is used by openAI, vertexAI and others. The aws_eventstream_to_metadata parse the AWS eventstream response, whicih is used by AWS bedrock. The maintainer could be @JuniorHsu @PeterL328

It needs maintainer on the top two lists

https://github.com/envoyproxy/envoy/blob/main/OWNERS.md

Hello @tyxia, this is another AI related parser in HTTP filter chain. I noticed that you show great interest in SSE filter. Are we able to get your sponsor here so we could share the filter to open source? Peter and/or I could do the first pass to reduce the review effort.

Hi @JuniorHsu , thanks for reaching out. Yes I am happy to sponsor here . I will take a closer look at PR/proposal later today

[cqi1217]

seems ci fails due to some format checks, I'll take a look later today

[cqi1217]

/retest

[cqi1217]

looks like test coverage is low, I'll fix it

[wbpcode]

If possible, I will to see the possibility to merge this to see_to_metadata :)

[PeterL328]

@wbpcode
There was a discussion in this issue about whether merging with existing sse_to_metadata filter or dedicated filter.
#43492 (comment)
Think several maintainers leaning towards dedicated filter as this is quite different from sse_to_metadata.
I think perhaps makes sense not to add AWS eventstream logic in sse_to_metadata as name suggests coupled with SSE only.

[tyxia]

@wbpcode There was a discussion in this issue about whether merging with existing sse_to_metadata filter or dedicated filter. #43492 (comment) Think several maintainers leaning towards dedicated filter as this is quite different from sse_to_metadata. I think perhaps makes sense not to add AWS eventstream logic in sse_to_metadata as name suggests coupled with SSE only.

Sorry for delay here. Thank you for the contribution!

From high level design point view. I am hoping we don't lock in on metadata:
dynamic metadata is powerful. it works well with RBAC and ext_authz to enable access control but it is weakly typed and has slow performance. filter state is also preferred in certain cases as it is strongly typed and has better performance, which also could enable broader use case with native c++ filter.

With that said, i am think we can give this filter a bit generic name like AWS_eventstream_parser_filter and sse_parser_filter. This opens the flexibility and possibility for future development. What do you guys think?

[botengyao]

I agree with tianyu for a general name. metadata is just so widely and easily supported, and it should be fine if we only care about some attributes.

[cqi1217]

aws_eventstream_parser seems reasonable for me, but sse_to_metadata filter is merged already, not sure how hard to rename it. @PeterL328 what do you think?

[tyxia]

aws_eventstream_parser seems reasonable for me, but sse_to_metadata filter is merged already, not sure how hard to rename it. @PeterL328 what do you think?

With the power of AI, i would think renaming should be straightforward :) IMHO, given the status of sse_to_metadata (just merged and in alpha status) renaming should be fine. I actually left similar comment w.r.t. metadata vs filter state on that PR. @PeterL328 please share your thoughts here as well.

In the meantime, i will kick off the review with Gemini :) let see how it works

[tyxia]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces the aws_eventstream_to_metadata HTTP filter, which extracts values from AWS EventStream response bodies—commonly used by AWS Bedrock—and writes them to dynamic metadata. The implementation features a pluggable content parser architecture, currently supporting JSON, and includes comprehensive documentation and tests. Feedback suggests optimizing buffer handling by avoiding repeated linearization of the data buffer and adding a dedicated counter for metadata type conversion failures to improve observability.

[cqi1217]

@tyxia the parsing code is mainly in #43544. we can have one http filter called to_metadata, and in the future another http filter called to_filter_state. or do you think one http filter including both metadata and filter_state(in the future) is better?

[tyxia]

one filter to do both is better. Those are couple lines of code that should result in two filters.

You can use config to control whether to write to metadata or filter state

[PeterL328]

@tyxia @cqi1217
I am fine with changing existing filter for sse_to_metadata to not be coupled with dynamic metadata and have ways to configure whether to use dynamic metadata or use filter state 👍 . Something like sse_parser_filter could work; we can work out the naming.
+1 on also not coupling dynamic metadata with aws eventstream as I think we should still have a dedicated filter for handling AWS eventstream.

[cqi1217]

/retest

[gemini-code-assist]

Code Review

This pull request introduces the aws_eventstream_parser HTTP filter, which extracts values from AWS EventStream response bodies and headers into dynamic metadata. The implementation includes a pluggable content parser architecture, comprehensive documentation, and extensive unit and integration tests. Feedback highlights the need for a default case in the header value conversion logic to improve robustness and notes potential precision loss when mapping int64_t values to double in metadata.

[gemini-code-assist]

The switch statement in headerValueToProtobufValue does not have a default case. While it currently covers all types defined in the EventstreamParser, adding a default case would make the code more robust against future additions to the HeaderValueType enum. When implementing this error-handling block, use IS_ENVOY_BUG to report the unexpected state.

References

1. When reporting a bug within an explicit error-handling block in CPP, use IS_ENVOY_BUG instead of ENVOY_BUG(false, ...).

[markdroth]

Consider using google.protobuf.UInt32Value (or google.protobuf.BoolValue if you take my suggestion above) to allow individual data plane implementations to have their own defaults.

[cqi1217]

google.protobuf.UInt32Value makes sense

[markdroth]

/lgtm api

[cqi1217]

/retest

[KBaichoo]

PTAL @tyxia

[botengyao]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces the aws_eventstream_parser HTTP filter to Envoy, which extracts values from AWS EventStream response bodies (such as AWS Bedrock streaming responses) and writes them to dynamic metadata. The review feedback highlights a critical bug in writeMetadata() where dynamic metadata is completely overwritten rather than merged, which could wipe out other filters' metadata. Additionally, the feedback suggests a performance optimization to avoid double JSON parsing in unwrapBedrockEnvelope by performing a quick substring check, and recommends replacing Protobuf Map's contains() with count() > 0 to ensure backward compatibility with older Protobuf versions.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

writeMetadata() is called at the end of encodeData for every chunk. It calls setDynamicMetadata(entry.first, entry.second) for each namespace. However, setDynamicMetadata completely overwrites the existing metadata struct for that namespace. Since structs_by_namespace_ is cleared at the end of writeMetadata(), any subsequent chunk or other filters' metadata in the same namespace will be completely wiped out! To prevent this, we must retrieve the existing metadata struct from StreamInfo, merge our new fields into it, and then call setDynamicMetadata with the merged struct.

void Filter::writeMetadata() {
  for (const auto& entry : structs_by_namespace_) {
    const std::string& ns = entry.first;
    const Protobuf::Struct& new_struct = entry.second;

    // Get existing metadata for this namespace to avoid overwriting other keys.
    Protobuf::Struct merged_struct;
    const auto& filter_metadata =
        encoder_callbacks_->streamInfo().dynamicMetadata().filter_metadata();
    const auto existing_it = filter_metadata.find(ns);
    if (existing_it != filter_metadata.end()) {
      merged_struct = existing_it->second;
    }

    // Merge new fields into the existing struct.
    for (const auto& [key, value] : new_struct.fields()) {
      (*merged_struct.mutable_fields())[key] = value;
    }

    encoder_callbacks_->streamInfo().setDynamicMetadata(ns, merged_struct);
  }
  structs_by_namespace_.clear();
}

[gemini-code-assist]

For every single message in the EventStream, unwrapBedrockEnvelope parses the payload as JSON to check if it is a Bedrock envelope. If it is not (which is the case for normal JSON payloads or other EventStream messages), it is parsed again by the JSON content parser. This results in double JSON parsing for every message, which is highly inefficient. We can add a quick check using absl::StrContains to avoid parsing the JSON if it doesn't even contain the "bytes" key.

Suggested change

	absl::optional<std::string> Filter::unwrapBedrockEnvelope(absl::string_view payload) {
	auto json_or = Json::Factory::loadFromString(std::string(payload));
	if (!json_or.ok()) {
	return absl::nullopt;
	}
	absl::optional<std::string> Filter::unwrapBedrockEnvelope(absl::string_view payload) {
	// Quick check to avoid expensive JSON parsing for payloads that are not Bedrock envelopes.
	if (!absl::StrContains(payload, "\"bytes\"")) {
	return absl::nullopt;
	}
	auto json_or = Json::Factory::loadFromString(std::string(payload));
	if (!json_or.ok()) {
	return absl::nullopt;
	}

[gemini-code-assist]

The code uses contains() on google::protobuf::Map (via fields().contains(...)). However, contains() is only supported in newer versions of Protobuf (v22.0+). To ensure backward compatibility with older Protobuf versions that Envoy might be compiled with, we should use count(...) > 0 instead.

Suggested change

	const auto pending_it = structs_by_namespace_.find(namespace_str);
	if (pending_it != structs_by_namespace_.end() &&
	pending_it->second.fields().contains(action.key)) {
	const auto pending_it = structs_by_namespace_.find(namespace_str);
	if (pending_it != structs_by_namespace_.end() &&
	pending_it->second.fields().count(action.key) > 0) {

[gemini-code-assist]

The code uses contains() on google::protobuf::Map (via fields().contains(...)). However, contains() is only supported in newer versions of Protobuf (v22.0+). To ensure backward compatibility with older Protobuf versions that Envoy might be compiled with, we should use count(...) > 0 instead.

Suggested change

	const auto& filter_metadata =
	encoder_callbacks_->streamInfo().dynamicMetadata().filter_metadata();
	const auto entry_it = filter_metadata.find(namespace_str);
	if (entry_it != filter_metadata.end() && entry_it->second.fields().contains(action.key)) {
	const auto& filter_metadata =
	encoder_callbacks_->streamInfo().dynamicMetadata().filter_metadata();
	const auto entry_it = filter_metadata.find(namespace_str);
	if (entry_it != filter_metadata.end() && entry_it->second.fields().count(action.key) > 0) {

[botengyao]

on-call ping @tyxia for the review.

[tyxia]

@cqi1217 There are some open review feedback from gemini, please take a look at the open gemini review feedback. Also sorry for the delay I actually forgot the publish my pending comments.

Thanks for the nice work!

[tyxia]

This is free buffer instance that is unbounded? We should apply the configurable buffer limit.

[tyxia]

In Incomplete message - need more data case, the parsing is not completed but we will send forward the data to downstream client. Is this expected/designed behavior? Should we return something like stopIterationNoBuffer

[ggreenway]

Waiting on fixes from the author

/wait

[cqi1217]

will work on it early next week

[tyxia]

will work on it early next week

Sounds good Thank you for your contribution!