ext_proc: send attributes

source/extensions/filters/http/ext_proc/BUILD

@@ -21,6 +21,7 @@ envoy_cc_library(
     ],
     deps = [
         ":client_interface",
+        ":matching_utils_lib",
         ":mutation_utils_lib",
         "//envoy/event:timer_interface",
         "//envoy/http:filter_interface",
@@ -80,6 +81,31 @@ 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",
+        ],
+    }),
+    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"],

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,9 @@ 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));
 
   return [filter_config, grpc_service = proto_config.grpc_service(),
           &context](Http::FilterChainFactoryCallbacks& callbacks) {
@@ -45,9 +46,10 @@ 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));
 
   return [filter_config, grpc_service = proto_config.grpc_service(),
           &server_context](Http::FilterChainFactoryCallbacks& callbacks) {

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

@@ -201,7 +201,8 @@ void Filter::onDestroy() {
 }
 
 FilterHeadersStatus Filter::onHeaders(ProcessorState& state,
-                                      Http::RequestOrResponseHeaderMap& headers, bool end_stream) {
+                                      Http::RequestOrResponseHeaderMap& headers, bool end_stream,
+                                      absl::optional<ProtobufWkt::Struct> proto) {
   switch (openStream()) {
   case StreamOpenState::Error:
     return FilterHeadersStatus::StopIteration;
@@ -219,6 +220,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");
@@ -237,7 +241,14 @@ FilterHeadersStatus Filter::decodeHeaders(RequestHeaderMap& headers, bool end_st
 
   FilterHeadersStatus status = FilterHeadersStatus::Continue;
   if (decoding_state_.sendHeaders()) {
-    status = onHeaders(decoding_state_, headers, end_stream);
+    absl::optional<Envoy::ProtobufWkt::Struct> proto;
+    if (config_->expressionManager().hasRequestExpr()) {
+      auto activation_ptr = Filters::Common::Expr::createActivation(decoding_state_.streamInfo(),
+                                                                    &headers, nullptr, nullptr);
+      proto = config_->expressionManager().evaluateRequestAttributes(std::move(activation_ptr));
+    }
+
+    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 +526,14 @@ 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);
+    absl::optional<Envoy::ProtobufWkt::Struct> proto;
+    if (config_->expressionManager().hasResponseExpr()) {
+      auto activation_ptr = Filters::Common::Expr::createActivation(encoding_state_.streamInfo(),
+                                                                    nullptr, &headers, nullptr);
+      proto = config_->expressionManager().evaluateResponseAttributes(std::move(activation_ptr));
+    }
+
+    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");

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,8 @@ 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)
       : 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),
@@ -135,8 +137,11 @@ class FilterConfig {
         filter_metadata_(config.filter_metadata()),
         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())) {}
+        allowed_headers_(
+            MatchingUtils::initHeaderMatchers(config.forward_rules().allowed_headers())),
+        disallowed_headers_(
+            MatchingUtils::initHeaderMatchers(config.forward_rules().disallowed_headers())),
+        expression_manager_(builder, config.request_attributes(), config.response_attributes()) {}
 
   bool failureModeAllow() const { return failure_mode_allow_; }
 
@@ -166,23 +171,14 @@ class FilterConfig {
 
   const Envoy::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_;
@@ -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>;
@@ -300,7 +298,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,
+                                      absl::optional<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);

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

@@ -0,0 +1,84 @@
+#include "source/extensions/filters/http/ext_proc/matching_utils.h"
+
+#include <memory>
+
+#if defined(USE_CEL_PARSER)
+#include "parser/parser.h"
+#endif
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace ExternalProcessing {
+
+absl::flat_hash_map<std::string, Filters::Common::Expr::ExpressionPtr>
+ExpressionManager::initExpressions(const Protobuf::RepeatedPtrField<std::string>& matchers) {
+  absl::flat_hash_map<std::string, 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());
+    }
+
+    auto iter = expr_list_.emplace_back(parse_status.value());
+
+    Filters::Common::Expr::ExpressionPtr expression =
+        Extensions::Filters::Common::Expr::createExpression(builder_->builder(), iter.expr());
+
+    expressions.try_emplace(matcher, std::move(expression));
+  }
+#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;
+}
+
+const absl::optional<ProtobufWkt::Struct> ExpressionManager::evaluateAttributes(
+    const Filters::Common::Expr::ActivationPtr& activation,
+    const absl::flat_hash_map<std::string, Filters::Common::Expr::ExpressionPtr>& expr) {
+  absl::optional<ProtobufWkt::Struct> proto;
+  if (!expr.empty()) {
+    proto.emplace(ProtobufWkt::Struct{});
+    for (const auto& hash_entry : expr) {
+      ProtobufWkt::Arena arena;
+      const auto result = hash_entry.second->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;
+}
+
+} // namespace ExternalProcessing
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

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

@@ -0,0 +1,69 @@
+#pragma once
+
+#include "source/common/common/logger.h"
+#include "source/common/common/matchers.h"
+#include "source/common/protobuf/protobuf.h"
+#include "source/extensions/filters/common/expr/evaluator.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace ExternalProcessing {
+
+class ExpressionManager : public Logger::Loggable<Logger::Id::ext_proc> {
+public:
+  ExpressionManager(Extensions::Filters::Common::Expr::BuilderInstanceSharedPtr builder,
+                    const Protobuf::RepeatedPtrField<std::string>& request_matchers,
+                    const Protobuf::RepeatedPtrField<std::string>& response_matchers)
+      : builder_(builder), request_expr_(initExpressions(request_matchers)),
+        response_expr_(initExpressions(response_matchers)){};
+
+  bool hasRequestExpr() const { return !request_expr_.empty(); };
+
+  bool hasResponseExpr() const { return !response_expr_.empty(); };
+
+  const absl::optional<ProtobufWkt::Struct>
+  evaluateRequestAttributes(const Filters::Common::Expr::ActivationPtr& activation) const {
+    return evaluateAttributes(activation, request_expr_);
+  }
+
+  const absl::optional<ProtobufWkt::Struct>
+  evaluateResponseAttributes(const Filters::Common::Expr::ActivationPtr& activation) const {
+    return evaluateAttributes(activation, response_expr_);
+  }
+
+  static const absl::optional<ProtobufWkt::Struct> evaluateAttributes(
+      const Filters::Common::Expr::ActivationPtr& activation,
+      const absl::flat_hash_map<std::string, Filters::Common::Expr::ExpressionPtr>& expr);
+
+private:
+  // This list is required to maintain the lifetimes of expr objects on which compiled expressions
+  // depend
+  std::list<google::api::expr::v1alpha1::ParsedExpr> expr_list_;
+  absl::flat_hash_map<std::string, Filters::Common::Expr::ExpressionPtr>
+  initExpressions(const Protobuf::RepeatedPtrField<std::string>& matchers);
+
+  Extensions::Filters::Common::Expr::BuilderInstanceSharedPtr builder_;
+
+  const absl::flat_hash_map<std::string, Filters::Common::Expr::ExpressionPtr> request_expr_;
+  const absl::flat_hash_map<std::string, Filters::Common::Expr::ExpressionPtr> response_expr_;
+};
+
+class MatchingUtils : public Logger::Loggable<Logger::Id::ext_proc> {
+public:
+  static 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;
+  }
+};
+
+} // namespace ExternalProcessing
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

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

@@ -172,6 +172,8 @@ class ProcessorState : public Logger::Loggable<Logger::Id::ext_proc> {
   virtual envoy::service::ext_proc::v3::HttpTrailers*
   mutableTrailers(envoy::service::ext_proc::v3::ProcessingRequest& request) const PURE;
 
+  virtual StreamInfo::StreamInfo& streamInfo() PURE;
+
 protected:
   void setBodyMode(
       envoy::extensions::filters::http::ext_proc::v3::ProcessingMode_BodySendMode body_mode);
@@ -283,6 +285,8 @@ class DecodingProcessorState : public ProcessorState {
   void requestWatermark() override;
   void clearWatermark() override;
 
+  StreamInfo::StreamInfo& streamInfo() override { return decoder_callbacks_->streamInfo(); }
+
 private:
   void setProcessingModeInternal(
       const envoy::extensions::filters::http::ext_proc::v3::ProcessingMode& mode);
@@ -356,6 +360,8 @@ class EncodingProcessorState : public ProcessorState {
   void requestWatermark() override;
   void clearWatermark() override;
 
+  StreamInfo::StreamInfo& streamInfo() override { return encoder_callbacks_->streamInfo(); }
+
 private:
   void setProcessingModeInternal(
       const envoy::extensions::filters::http::ext_proc::v3::ProcessingMode& mode);