ext_proc: send attributes

api/envoy/extensions/filters/http/ext_proc/v3/ext_proc.proto

@@ -28,7 +28,6 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // **Current Implementation Status:**
 // All options and processing modes are implemented except for the following:
 //
-// * Request and response attributes are not sent and not processed.
 // * Dynamic metadata in responses from the external processor is ignored.
 // * "async mode" is not implemented.
 
@@ -125,15 +124,13 @@ message ExternalProcessor {
   // for a reply.
   bool async_mode = 4;
 
-  // [#not-implemented-hide:]
   // Envoy provides a number of :ref:`attributes <arch_overview_attributes>`
   // for expressive policies. Each attribute name provided in this field will be
   // matched against that list and populated in the request_headers message.
   // See the :ref:`attribute documentation <arch_overview_request_attributes>`
   // for the list of supported attributes and their types.
   repeated string request_attributes = 5;
 
-  // [#not-implemented-hide:]
   // Envoy provides a number of :ref:`attributes <arch_overview_attributes>`
   // for expressive policies. Each attribute name provided in this field will be
   // matched against that list and populated in the response_headers message.

bazel/repository_locations.bzl

@@ -1214,6 +1214,7 @@ REPOSITORY_LOCATIONS_SPEC = dict(
             "envoy.access_loggers.wasm",
             "envoy.bootstrap.wasm",
             "envoy.rate_limit_descriptors.expr",
+            "envoy.filters.http.ext_proc",
             "envoy.filters.http.rate_limit_quota",
             "envoy.filters.http.rbac",
             "envoy.filters.http.wasm",
@@ -1243,6 +1244,7 @@ REPOSITORY_LOCATIONS_SPEC = dict(
             "envoy.formatter.cel",
             "envoy.bootstrap.wasm",
             "envoy.rate_limit_descriptors.expr",
+            "envoy.filters.http.ext_proc",
             "envoy.filters.http.rate_limit_quota",
             "envoy.filters.http.rbac",
             "envoy.filters.http.wasm",

changelogs/current.yaml

@@ -13,5 +13,12 @@ removed_config_or_runtime:
 # *Normally occurs at the end of the* :ref:`deprecation period <deprecated>`
 
 new_features:
+- area: ext_proc
+  change: |
+    implemented
+    :ref:`request_attributes <envoy_v3_api_field_extensions.filters.http.ext_proc.v3.ExternalProcessor.request_attributes>`
+    and
+    :ref:`response_attributes <envoy_v3_api_field_extensions.filters.http.ext_proc.v3.ExternalProcessor.response_attributes>`
+    config APIs to enable sending and receiving attributes to/from the external processing server.
 
 deprecated:

source/extensions/filters/http/ext_proc/BUILD

@@ -19,8 +19,10 @@ envoy_cc_library(
         "ext_proc.h",
         "processor_state.h",
     ],
+    tags = ["skip_on_windows"],
     deps = [
         ":client_interface",
+        ":matching_utils_lib",
         ":mutation_utils_lib",
         "//envoy/event:timer_interface",
         "//envoy/http:filter_interface",
@@ -44,6 +46,7 @@ envoy_cc_extension(
     name = "config",
     srcs = ["config.cc"],
     hdrs = ["config.h"],
+    tags = ["skip_on_windows"],
     deps = [
         ":client_lib",
         ":ext_proc",
@@ -55,6 +58,7 @@ envoy_cc_extension(
 envoy_cc_library(
     name = "client_interface",
     hdrs = ["client.h"],
+    tags = ["skip_on_windows"],
     deps = [
         "//envoy/grpc:async_client_manager_interface",
         "//envoy/grpc:status",
@@ -80,10 +84,37 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "matching_utils_lib",
+    srcs = ["matching_utils.cc"],
+    hdrs = ["matching_utils.h"],
+    copts = select({
+        "//bazel:windows_x86_64": [],
+        "//conditions:default": [
+            "-DUSE_CEL_PARSER",
+        ],
+    }),
+    tags = ["skip_on_windows"],
+    deps = [
+        "//envoy/http:header_map_interface",
+        "//source/common/protobuf",
+        "//source/extensions/filters/common/expr:evaluator_lib",
+        "@com_google_cel_cpp//eval/public:cel_expr_builder_factory",
+    ] + select(
+        {
+            "//bazel:windows_x86_64": [],
+            "//conditions:default": [
+                "@com_google_cel_cpp//parser",
+            ],
+        },
+    ),
+)
+
 envoy_cc_library(
     name = "client_lib",
     srcs = ["client_impl.cc"],
     hdrs = ["client_impl.h"],
+    tags = ["skip_on_windows"],
     deps = [
         ":client_interface",
         "//envoy/grpc:async_client_interface",

source/extensions/filters/http/ext_proc/config.cc

@@ -1,5 +1,6 @@
 #include "source/extensions/filters/http/ext_proc/config.h"
 
+#include "source/extensions/filters/common/expr/evaluator.h"
 #include "source/extensions/filters/http/ext_proc/client_impl.h"
 #include "source/extensions/filters/http/ext_proc/ext_proc.h"
 
@@ -15,9 +16,11 @@ Http::FilterFactoryCb ExternalProcessingFilterConfig::createFilterFactoryFromPro
       PROTOBUF_GET_MS_OR_DEFAULT(proto_config, message_timeout, DefaultMessageTimeoutMs);
   const uint32_t max_message_timeout_ms =
       PROTOBUF_GET_MS_OR_DEFAULT(proto_config, max_message_timeout, DefaultMaxMessageTimeoutMs);
-  const auto filter_config =
-      std::make_shared<FilterConfig>(proto_config, std::chrono::milliseconds(message_timeout_ms),
-                                     max_message_timeout_ms, context.scope(), stats_prefix);
+  const auto filter_config = std::make_shared<FilterConfig>(
+      proto_config, std::chrono::milliseconds(message_timeout_ms), max_message_timeout_ms,
+      context.scope(), stats_prefix,
+      Envoy::Extensions::Filters::Common::Expr::getBuilder(context.serverFactoryContext()),
+      context.serverFactoryContext().localInfo());
 
   return [filter_config, grpc_service = proto_config.grpc_service(),
           &context](Http::FilterChainFactoryCallbacks& callbacks) {
@@ -44,9 +47,11 @@ ExternalProcessingFilterConfig::createFilterFactoryFromProtoWithServerContextTyp
       PROTOBUF_GET_MS_OR_DEFAULT(proto_config, message_timeout, DefaultMessageTimeoutMs);
   const uint32_t max_message_timeout_ms =
       PROTOBUF_GET_MS_OR_DEFAULT(proto_config, max_message_timeout, DefaultMaxMessageTimeoutMs);
-  const auto filter_config =
-      std::make_shared<FilterConfig>(proto_config, std::chrono::milliseconds(message_timeout_ms),
-                                     max_message_timeout_ms, server_context.scope(), stats_prefix);
+  const auto filter_config = std::make_shared<FilterConfig>(
+      proto_config, std::chrono::milliseconds(message_timeout_ms), max_message_timeout_ms,
+      server_context.scope(), stats_prefix,
+      Envoy::Extensions::Filters::Common::Expr::getBuilder(server_context),
+      server_context.localInfo());
 
   return [filter_config, grpc_service = proto_config.grpc_service(),
           &server_context](Http::FilterChainFactoryCallbacks& callbacks) {

source/extensions/filters/http/ext_proc/ext_proc.cc

@@ -221,7 +221,8 @@ void Filter::onDestroy() {
 }
 
 FilterHeadersStatus Filter::onHeaders(ProcessorState& state,
-                                      Http::RequestOrResponseHeaderMap& headers, bool end_stream) {
+                                      Http::RequestOrResponseHeaderMap& headers, bool end_stream,
+                                      ProtobufWkt::Struct* proto) {
   switch (openStream()) {
   case StreamOpenState::Error:
     return FilterHeadersStatus::StopIteration;
@@ -239,6 +240,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 != nullptr) {
+    (*headers_req->mutable_attributes())[FilterName] = *proto;
+  }
   state.onStartProcessorCall(std::bind(&Filter::onMessageTimeout, this), config_->messageTimeout(),
                              ProcessorState::CallbackState::HeadersCallback);
   ENVOY_LOG(debug, "Sending headers message");
@@ -257,7 +261,17 @@ FilterHeadersStatus Filter::decodeHeaders(RequestHeaderMap& headers, bool end_st
 
   FilterHeadersStatus status = FilterHeadersStatus::Continue;
   if (decoding_state_.sendHeaders()) {
-    status = onHeaders(decoding_state_, headers, end_stream);
+    ProtobufWkt::Struct proto;
+
+    if (config_->expressionManager().hasRequestExpr()) {
+      auto activation_ptr = Filters::Common::Expr::createActivation(
+          &config_->expressionManager().localInfo(), decoding_state_.streamInfo(), &headers,
+          nullptr, nullptr);
+      proto = config_->expressionManager().evaluateRequestAttributes(*activation_ptr);
+    }
+
+    status = onHeaders(decoding_state_, headers, end_stream,
+                       config_->expressionManager().hasRequestExpr() ? &proto : nullptr);
     ENVOY_LOG(trace, "onHeaders returning {}", static_cast<int>(status));
   } else {
     ENVOY_LOG(trace, "decodeHeaders: Skipped header processing");
@@ -535,7 +549,17 @@ FilterHeadersStatus Filter::encodeHeaders(ResponseHeaderMap& headers, bool end_s
 
   FilterHeadersStatus status = FilterHeadersStatus::Continue;
   if (!processing_complete_ && encoding_state_.sendHeaders()) {
-    status = onHeaders(encoding_state_, headers, end_stream);
+    ProtobufWkt::Struct proto;
+
+    if (config_->expressionManager().hasResponseExpr()) {
+      auto activation_ptr = Filters::Common::Expr::createActivation(
+          &config_->expressionManager().localInfo(), encoding_state_.streamInfo(), nullptr,
+          &headers, nullptr);
+      proto = config_->expressionManager().evaluateResponseAttributes(*activation_ptr);
+    }
+
+    status = onHeaders(encoding_state_, headers, end_stream,
+                       config_->expressionManager().hasResponseExpr() ? &proto : nullptr);
     ENVOY_LOG(trace, "onHeaders returns {}", static_cast<int>(status));
   } else {
     ENVOY_LOG(trace, "encodeHeaders: Skipped header processing");

source/extensions/filters/http/ext_proc/ext_proc.h

@@ -24,6 +24,7 @@
 #include "source/extensions/filters/common/mutation_rules/mutation_rules.h"
 #include "source/extensions/filters/http/common/pass_through_filter.h"
 #include "source/extensions/filters/http/ext_proc/client.h"
+#include "source/extensions/filters/http/ext_proc/matching_utils.h"
 #include "source/extensions/filters/http/ext_proc/processor_state.h"
 
 namespace Envoy {
@@ -126,7 +127,9 @@ class FilterConfig {
   FilterConfig(const envoy::extensions::filters::http::ext_proc::v3::ExternalProcessor& config,
                const std::chrono::milliseconds message_timeout,
                const uint32_t max_message_timeout_ms, Stats::Scope& scope,
-               const std::string& stats_prefix)
+               const std::string& stats_prefix,
+               Extensions::Filters::Common::Expr::BuilderInstanceSharedPtr builder,
+               const LocalInfo::LocalInfo& local_info)
       : failure_mode_allow_(config.failure_mode_allow()),
         disable_clear_route_cache_(config.disable_clear_route_cache()),
         message_timeout_(message_timeout), max_message_timeout_ms_(max_message_timeout_ms),
@@ -136,7 +139,9 @@ class FilterConfig {
         allow_mode_override_(config.allow_mode_override()),
         disable_immediate_response_(config.disable_immediate_response()),
         allowed_headers_(initHeaderMatchers(config.forward_rules().allowed_headers())),
-        disallowed_headers_(initHeaderMatchers(config.forward_rules().disallowed_headers())) {}
+        disallowed_headers_(initHeaderMatchers(config.forward_rules().disallowed_headers())),
+        expression_manager_(builder, local_info, config.request_attributes(),
+                            config.response_attributes()) {}
 
   bool failureModeAllow() const { return failure_mode_allow_; }
 
@@ -164,25 +169,16 @@ class FilterConfig {
     return disallowed_headers_;
   }
 
-  const Envoy::ProtobufWkt::Struct& filterMetadata() const { return filter_metadata_; }
+  const ProtobufWkt::Struct& filterMetadata() const { return filter_metadata_; }
+
+  const ExpressionManager& expressionManager() const { return expression_manager_; }
 
 private:
   ExtProcFilterStats generateStats(const std::string& prefix,
                                    const std::string& filter_stats_prefix, Stats::Scope& scope) {
     const std::string final_prefix = absl::StrCat(prefix, "ext_proc.", filter_stats_prefix);
     return {ALL_EXT_PROC_FILTER_STATS(POOL_COUNTER_PREFIX(scope, final_prefix))};
   }
-  const std::vector<Matchers::StringMatcherPtr>
-  initHeaderMatchers(const envoy::type::matcher::v3::ListStringMatcher& header_list) {
-    std::vector<Matchers::StringMatcherPtr> header_matchers;
-    for (const auto& matcher : header_list.patterns()) {
-      header_matchers.push_back(
-          std::make_unique<Matchers::StringMatcherImpl<envoy::type::matcher::v3::StringMatcher>>(
-              matcher));
-    }
-    return header_matchers;
-  }
-
   const bool failure_mode_allow_;
   const bool disable_clear_route_cache_;
   const std::chrono::milliseconds message_timeout_;
@@ -191,7 +187,7 @@ class FilterConfig {
   ExtProcFilterStats stats_;
   const envoy::extensions::filters::http::ext_proc::v3::ProcessingMode processing_mode_;
   const Filters::Common::MutationRules::Checker mutation_checker_;
-  const Envoy::ProtobufWkt::Struct filter_metadata_;
+  const ProtobufWkt::Struct filter_metadata_;
   // If set to true, allow the processing mode to be modified by the ext_proc response.
   const bool allow_mode_override_;
   // If set to true, disable the immediate response from the ext_proc server, which means
@@ -201,6 +197,8 @@ class FilterConfig {
   const std::vector<Matchers::StringMatcherPtr> allowed_headers_;
   // Empty disallowed_header_ means disallow nothing, i.e, allow all.
   const std::vector<Matchers::StringMatcherPtr> disallowed_headers_;
+
+  const ExpressionManager expression_manager_;
 };
 
 using FilterConfigSharedPtr = std::shared_ptr<FilterConfig>;
@@ -315,7 +313,9 @@ class Filter : public Logger::Loggable<Logger::Id::ext_proc>,
   void sendImmediateResponse(const envoy::service::ext_proc::v3::ImmediateResponse& response);
 
   Http::FilterHeadersStatus onHeaders(ProcessorState& state,
-                                      Http::RequestOrResponseHeaderMap& headers, bool end_stream);
+                                      Http::RequestOrResponseHeaderMap& headers, bool end_stream,
+                                      ProtobufWkt::Struct* proto);
+
   // Return a pair of whether to terminate returning the current result.
   std::pair<bool, Http::FilterDataStatus> sendStreamChunk(ProcessorState& state);
   Http::FilterDataStatus onData(ProcessorState& state, Buffer::Instance& data, bool end_stream);