ext_proc: send attributes #30781
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -8,6 +8,10 @@ | |
|
|
||
| #include "absl/strings/str_format.h" | ||
|
|
||
| #if defined(USE_CEL_PARSER) | ||
| #include "parser/parser.h" | ||
| #endif | ||
|
|
||
| namespace Envoy { | ||
| namespace Extensions { | ||
| namespace HttpFilters { | ||
|
|
@@ -113,6 +117,27 @@ ExtProcLoggingInfo::grpcCalls(envoy::config::core::v3::TrafficDirection traffic_ | |
| : encoding_processor_grpc_calls_; | ||
| } | ||
|
|
||
| absl::flat_hash_map<std::string, Extensions::Filters::Common::Expr::ExpressionPtr> | ||
| FilterConfig::initExpressions(const Protobuf::RepeatedPtrField<std::string>& matchers) const { | ||
| absl::flat_hash_map<std::string, Extensions::Filters::Common::Expr::ExpressionPtr> expressions; | ||
| #if defined(USE_CEL_PARSER) | ||
| for (const auto& matcher : matchers) { | ||
| auto parse_status = google::api::expr::parser::Parse(matcher); | ||
| if (!parse_status.ok()) { | ||
| throw EnvoyException("Unable to parse descriptor expression: " + | ||
| parse_status.status().ToString()); | ||
| } | ||
| expressions.emplace(matcher, Extensions::Filters::Common::Expr::createExpression( | ||
| builder_->builder(), parse_status.value().expr())); | ||
| } | ||
| #else | ||
| ENVOY_LOG(warn, "CEL expression parsing is not available for use in this environment." | ||
| " Attempted to parse " + | ||
| std::to_string(matchers.size()) + " expressions"); | ||
| #endif | ||
| return expressions; | ||
| } | ||
|
|
||
| FilterConfigPerRoute::FilterConfigPerRoute(const ExtProcPerRoute& config) | ||
| : disabled_(config.disabled()) { | ||
| if (config.has_overrides()) { | ||
|
|
@@ -201,7 +226,8 @@ void Filter::onDestroy() { | |
| } | ||
|
|
||
| FilterHeadersStatus Filter::onHeaders(ProcessorState& state, | ||
| Http::RequestOrResponseHeaderMap& headers, bool end_stream, | ||
| absl::optional<ProtobufWkt::Struct> proto) { | ||
| switch (openStream()) { | ||
| case StreamOpenState::Error: | ||
| return FilterHeadersStatus::StopIteration; | ||
|
|
@@ -219,6 +245,9 @@ FilterHeadersStatus Filter::onHeaders(ProcessorState& state, | |
| MutationUtils::headersToProto(headers, config_->allowedHeaders(), config_->disallowedHeaders(), | ||
| *headers_req->mutable_headers()); | ||
| headers_req->set_end_of_stream(end_stream); | ||
| if (proto.has_value()) { | ||
| (*headers_req->mutable_attributes())[FilterName] = proto.value(); | ||
| } | ||
| state.onStartProcessorCall(std::bind(&Filter::onMessageTimeout, this), config_->messageTimeout(), | ||
| ProcessorState::CallbackState::HeadersCallback); | ||
| ENVOY_LOG(debug, "Sending headers message"); | ||
|
|
@@ -228,6 +257,48 @@ FilterHeadersStatus Filter::onHeaders(ProcessorState& state, | |
| return FilterHeadersStatus::StopIteration; | ||
| } | ||
|
|
||
| const absl::optional<ProtobufWkt::Struct> Filter::evaluateAttributes( | ||
| Filters::Common::Expr::ActivationPtr activation, | ||
| const absl::flat_hash_map<std::string, Extensions::Filters::Common::Expr::ExpressionPtr>& | ||
| expr) { | ||
| absl::optional<ProtobufWkt::Struct> proto; | ||
| if (expr.size() > 0) { | ||
| proto.emplace(ProtobufWkt::Struct{}); | ||
| for (const auto& hash_entry : expr) { | ||
| ProtobufWkt::Arena arena; | ||
| const auto result = hash_entry.second.get()->Evaluate(*activation, &arena); | ||
| if (!result.ok()) { | ||
| // TODO: Stats? | ||
| continue; | ||
| } | ||
|
|
||
| if (result.value().IsError()) { | ||
| ENVOY_LOG(trace, "error parsing cel expression {}", hash_entry.first); | ||
| continue; | ||
| } | ||
|
|
||
| ProtobufWkt::Value value; | ||
| switch (result.value().type()) { | ||
| case google::api::expr::runtime::CelValue::Type::kBool: | ||
| value.set_bool_value(result.value().BoolOrDie()); | ||
| break; | ||
| case google::api::expr::runtime::CelValue::Type::kNullType: | ||
| value.set_null_value(ProtobufWkt::NullValue{}); | ||
| break; | ||
| case google::api::expr::runtime::CelValue::Type::kDouble: | ||
| value.set_number_value(result.value().DoubleOrDie()); | ||
| break; | ||
| default: | ||
| value.set_string_value(Filters::Common::Expr::print(result.value())); | ||
| } | ||
|
|
||
| (*(proto.value()).mutable_fields())[hash_entry.first] = value; | ||
| } | ||
| } | ||
|
|
||
| return proto; | ||
| } | ||
|
|
||
| FilterHeadersStatus Filter::decodeHeaders(RequestHeaderMap& headers, bool end_stream) { | ||
| ENVOY_LOG(trace, "decodeHeaders: end_stream = {}", end_stream); | ||
| mergePerRouteConfig(); | ||
|
|
@@ -237,7 +308,11 @@ FilterHeadersStatus Filter::decodeHeaders(RequestHeaderMap& headers, bool end_st | |
|
|
||
| FilterHeadersStatus status = FilterHeadersStatus::Continue; | ||
| if (decoding_state_.sendHeaders()) { | ||
| auto activation_ptr = Filters::Common::Expr::createActivation(decoding_state_.streamInfo(), | ||
| &headers, nullptr, nullptr); | ||
|
There was a problem hiding this comment. for my own education, why do we need to use CEL parser to forward the attributes? Arent these available as part of streamInfo readily? I think @tyxia 's concern was the speed and overhead when we drag the CEL parser into the picture There was a problem hiding this comment. Attributes, as documented here, are only implemented in CEL (as far as I can tell). I don't know why this very convenient mode of accessing this info was coupled to CEL in the first place, but because it is, CEL is used to access them here in lieu of re-implementing attributes in a more common/accessible location There was a problem hiding this comment. You may want to make conditional on presence of attributes in the configuration. Otherwise the CEL context is instantiated even if no attributes were needed. There was a problem hiding this comment. @yanavlasov I have made the change so the CEL context is only instantiated when needed |
||
| auto proto = evaluateAttributes(std::move(activation_ptr), config_->requestExpr()); | ||
|
|
||
| status = onHeaders(decoding_state_, headers, end_stream, proto); | ||
| ENVOY_LOG(trace, "onHeaders returning {}", static_cast<int>(status)); | ||
| } else { | ||
| ENVOY_LOG(trace, "decodeHeaders: Skipped header processing"); | ||
|
|
@@ -515,7 +590,11 @@ FilterHeadersStatus Filter::encodeHeaders(ResponseHeaderMap& headers, bool end_s | |
|
|
||
| FilterHeadersStatus status = FilterHeadersStatus::Continue; | ||
| if (!processing_complete_ && encoding_state_.sendHeaders()) { | ||
| auto activation_ptr = Filters::Common::Expr::createActivation(encoding_state_.streamInfo(), | ||
| nullptr, &headers, nullptr); | ||
| auto proto = evaluateAttributes(std::move(activation_ptr), config_->responseExpr()); | ||
|
|
||
| status = onHeaders(encoding_state_, headers, end_stream, proto); | ||
| ENVOY_LOG(trace, "onHeaders returns {}", static_cast<int>(status)); | ||
| } else { | ||
| ENVOY_LOG(trace, "encodeHeaders: Skipped header processing"); | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The CEL usage here is not correct.
We need to extend the lifetime of
parse_statushere because CEL lib specifically requires that parsed expression(return of parser::Parse() ) need to outlive the compiled expression (return ofExpr::createExpression)You can see some correct examples here and here
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
According to that comment, the
source_infomust also outlive the parsed expression, but we throw it away in this common method. Is this also a bug?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good observation but that is probably intentional because CHECK is not performed in Envoy:
source_info is optional for EVALUATION phase (where the return of this function is used) and is used for CHECK phase( see my comment here and working example).
So, it is fine for my CEL use case because only EVAL is performed in Envoy and both PARSE and CHECK are performed in control plane, as the split model I mentioned before. (i.e., source_info is not used)
Back to common method, I think it is fine so far and in your use case here, because CHECK has not been performed in Envoy data plane.
If we really want CHECK functionality in Envoy(probably should avoid as performance overhead), we can update that common method to use the source info in that parsed expression, i.e., return of
parser::Parse(matcher)which we want to save here.cced @kyessenov the author of common method.