network ext_proc: add forwarding filter metadata support to the side service (envoyproxy#39255)

Risk Level: low
Testing: unit and integration tests

---------

Signed-off-by: Boteng Yao <[email protected]>

Author: botengyao

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

@@ -34,7 +34,7 @@ option (xds.annotations.v3.file_status).work_in_progress = true;
 //
 // By using the filter's processing mode, you can selectively choose which data
 // directions to process (read, write or both), allowing for efficient processing.
-// [#next-free-field: 6]
+// [#next-free-field: 7]
 message NetworkExternalProcessor {
   // The gRPC service that will process network traffic.
   // This service must implement the NetworkExternalProcessor service
@@ -64,6 +64,9 @@ message NetworkExternalProcessor {
   }];
 
   string stat_prefix = 5 [(validate.rules).string = {min_len: 1}];
+
+  // Options related to the sending and receiving of dynamic metadata.
+  MetadataOptions metadata_options = 6;
 }
 
 // Options for controlling processing behavior.
@@ -86,3 +89,23 @@ message ProcessingMode {
   // Default: STREAMED
   DataSendMode process_write = 2;
 }
+
+// The MetadataOptions structure defines options for sending dynamic metadata. Specifically,
+// which namespaces to send to the server.
+message MetadataOptions {
+  message MetadataNamespaces {
+    // Specifies a list of metadata namespaces whose values, if present,
+    // will be passed to the ext_proc service as an opaque *protobuf::Struct*.
+    repeated string untyped = 1;
+
+    // Specifies a list of metadata namespaces whose values, if present,
+    // will be passed to the ext_proc service as a *protobuf::Any*. This allows
+    // envoy and the external processing server to share the protobuf message
+    // definition for safe parsing.
+    repeated string typed = 2;
+  }
+
+  // Describes which typed or untyped dynamic metadata namespaces to forward to
+  // the external processing server.
+  MetadataNamespaces forwarding_namespaces = 1;
+}

source/extensions/filters/network/ext_proc/ext_proc.cc

@@ -164,6 +164,7 @@ void NetworkExtProcFilter::sendRequest(Buffer::Instance& data, bool end_stream,
 
   // Prepare the request message
   ProcessingRequest request;
+  addDynamicMetadata(request);
 
   if (is_read) {
     auto* read_data = request.mutable_read_data();
@@ -270,6 +271,39 @@ void NetworkExtProcFilter::closeConnection(const std::string& reason) {
   stats_.connections_closed_.inc();
 }
 
+void NetworkExtProcFilter::addDynamicMetadata(ProcessingRequest& req) {
+  if (config_->untypedForwardingMetadataNamespaces().empty() &&
+      config_->typedForwardingMetadataNamespaces().empty()) {
+    return;
+  }
+
+  envoy::config::core::v3::Metadata forwarding_metadata;
+
+  const auto& dynamic_metadata = read_callbacks_->connection().streamInfo().dynamicMetadata();
+  const auto& connection_metadata = dynamic_metadata.filter_metadata();
+  const auto& connection_typed_metadata = dynamic_metadata.typed_filter_metadata();
+
+  for (const auto& context_key : config_->untypedForwardingMetadataNamespaces()) {
+    if (const auto metadata_it = connection_metadata.find(context_key);
+        metadata_it != connection_metadata.end()) {
+      (*forwarding_metadata.mutable_filter_metadata())[metadata_it->first] = metadata_it->second;
+    }
+  }
+
+  for (const auto& context_key : config_->typedForwardingMetadataNamespaces()) {
+    if (const auto metadata_it = connection_typed_metadata.find(context_key);
+        metadata_it != connection_typed_metadata.end()) {
+      (*forwarding_metadata.mutable_typed_filter_metadata())[metadata_it->first] =
+          metadata_it->second;
+    }
+  }
+
+  if (!forwarding_metadata.filter_metadata().empty() ||
+      !forwarding_metadata.typed_filter_metadata().empty()) {
+    *req.mutable_metadata() = std::move(forwarding_metadata);
+  }
+}
+
 } // namespace ExtProc
 } // namespace NetworkFilters
 } // namespace Extensions

source/extensions/filters/network/ext_proc/ext_proc.h

@@ -49,6 +49,12 @@ class Config {
          Stats::Scope& scope)
       : failure_mode_allow_(config.failure_mode_allow()),
         processing_mode_(config.processing_mode()), grpc_service_(config.grpc_service()),
+        untyped_forwarding_namespaces_(
+            config.metadata_options().forwarding_namespaces().untyped().begin(),
+            config.metadata_options().forwarding_namespaces().untyped().end()),
+        typed_forwarding_namespaces_(
+            config.metadata_options().forwarding_namespaces().typed().begin(),
+            config.metadata_options().forwarding_namespaces().typed().end()),
         stats_(generateStats(config.stat_prefix(), scope)) {};
 
   bool failureModeAllow() const { return failure_mode_allow_; }
@@ -59,6 +65,14 @@ class Config {
 
   const envoy::config::core::v3::GrpcService& grpcService() const { return grpc_service_; }
 
+  const std::vector<std::string>& untypedForwardingMetadataNamespaces() const {
+    return untyped_forwarding_namespaces_;
+  }
+
+  const std::vector<std::string>& typedForwardingMetadataNamespaces() const {
+    return typed_forwarding_namespaces_;
+  }
+
   const NetworkExtProcStats& stats() const { return stats_; }
 
 private:
@@ -70,6 +84,8 @@ class Config {
   const bool failure_mode_allow_;
   const envoy::extensions::filters::network::ext_proc::v3::ProcessingMode processing_mode_;
   const envoy::config::core::v3::GrpcService grpc_service_;
+  const std::vector<std::string> untyped_forwarding_namespaces_;
+  const std::vector<std::string> typed_forwarding_namespaces_;
   NetworkExtProcStats stats_;
 };
 
@@ -137,6 +153,7 @@ class NetworkExtProcFilter : public Envoy::Network::Filter,
   void closeStream();
 
   void sendRequest(Envoy::Buffer::Instance& data, bool end_stream, bool is_read);
+  void addDynamicMetadata(ProcessingRequest& req);
 
   Envoy::Network::FilterStatus handleStreamError();
   void closeConnection(const std::string& reason);