Common: Consume, proxy and insert metadata from downstream to upstream

include/envoy/http/filter.h

@@ -227,6 +227,15 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
    */
   virtual HeaderMap& addDecodedTrailers() PURE;
 
+  /**
+   * Adds decoded metadata. This function can only be called in
+   * StreamDecoderFilter::decodeHeaders/Data/Trailers(). Do not call in
+   * StreamDecoderFilter::decodeMetadata().
+   *
+   * @return a reference to metadata map vector, where new metadata map can be added.
+   */
+  virtual MetadataMapVector& addDecodedMetadata() PURE;
+
   /**
    * Create a locally generated response using the provided response_code and body_text parameters.
    * If the request was a gRPC request the local reply will be encoded as a gRPC response with a 200
@@ -396,6 +405,14 @@ class StreamDecoderFilter : public StreamFilterBase {
    */
   virtual FilterTrailersStatus decodeTrailers(HeaderMap& trailers) PURE;
 
+  /**
+   * Called with decoded metadata. Add new metadata to metadata_map directly. Do not call
+   * StreamDecoderFilterCallbacks::addDecodedMetadata() to add new metadata.
+   *
+   * @param metadata supplies the decoded metadata.
+   */
+  virtual FilterMetadataStatus decodeMetadata(MetadataMap& metadata_map) PURE;
+
   /**
    * Called by the filter manager once to initialize the filter decoder callbacks that the
    * filter should use. Callbacks will not be invoked by the filter after onDestroy() is called.

source/common/http/async_client_impl.h

@@ -298,6 +298,7 @@ class AsyncStreamImpl : public AsyncClient::Stream,
     // filter which uses this function for buffering.
     ASSERT(buffered_body_ != nullptr);
   }
+  MetadataMapVector& addDecodedMetadata() override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
   const Buffer::Instance* decodingBuffer() override { return buffered_body_.get(); }
   void sendLocalReply(Code code, absl::string_view body,
                       std::function<void(HeaderMap& headers)> modify_headers,

source/common/http/conn_manager_impl.cc

@@ -774,6 +774,19 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(ActiveStreamDecoderFilte
     ENVOY_STREAM_LOG(trace, "decode headers called: filter={} status={}", *this,
                      static_cast<const void*>((*entry).get()), static_cast<uint64_t>(status));
 
+    bool new_metadata_added = processNewlyAddedMetadata();
+
+    // If end_stream is set in headers, and a filter adds new metadata, we need to delay end_stream
+    // in headers by inserting an empty data frame with end_stream set. The empty data frame is sent
+    // after the new metadata.
+    if ((*entry)->end_stream_ && new_metadata_added && !empty_data_end_stream_sent_) {
+      Buffer::OwnedImpl empty_data("");
+      ENVOY_STREAM_LOG(
+          trace, "inserting an empty data frame for end_stream due metadata being added.", *this);
+      addDecodedData(*((*entry).get()), empty_data, true);
+      empty_data_end_stream_sent_ = true;
+    }
+
     if (!(*entry)->commonHandleAfterHeadersCallback(status, decoding_headers_only_) &&
         std::next(entry) != decoder_filters_.end()) {
       // Stop iteration IFF this is not the last filter. If it is the last filter, continue with
@@ -885,6 +898,8 @@ void ConnectionManagerImpl::ActiveStream::decodeData(ActiveStreamDecoderFilter*
     ENVOY_STREAM_LOG(trace, "decode data called: filter={} status={}", *this,
                      static_cast<const void*>((*entry).get()), static_cast<uint64_t>(status));
 
+    processNewlyAddedMetadata();
+
     if (!trailers_exists_at_start && request_trailers_ &&
         trailers_added_entry == decoder_filters_.end()) {
       trailers_added_entry = entry;
@@ -942,6 +957,10 @@ void ConnectionManagerImpl::ActiveStream::addDecodedData(ActiveStreamDecoderFilt
   }
 }
 
+MetadataMapVector& ConnectionManagerImpl::ActiveStream::addDecodedMetadata() {
+  return request_metadata_map_vector_;
+}
+
 void ConnectionManagerImpl::ActiveStream::decodeTrailers(HeaderMapPtr&& trailers) {
   resetIdleTimer();
   maybeEndDecode(true);
@@ -976,13 +995,41 @@ void ConnectionManagerImpl::ActiveStream::decodeTrailers(ActiveStreamDecoderFilt
     state_.filter_call_state_ &= ~FilterCallState::DecodeTrailers;
     ENVOY_STREAM_LOG(trace, "decode trailers called: filter={} status={}", *this,
                      static_cast<const void*>((*entry).get()), static_cast<uint64_t>(status));
+
+    processNewlyAddedMetadata();
+
     if (!(*entry)->commonHandleAfterTrailersCallback(status)) {
       return;
     }
   }
   disarmRequestTimeout();
 }
 
+void ConnectionManagerImpl::ActiveStream::decodeMetadata(MetadataMapPtr&& metadata_map) {
+  resetIdleTimer();
+  // After going through filters, the ownership of metadata_map will be passed to router filter.
+  // Router filter may encode metadata_map to the next hop immediately or store metadata_map and
+  // encode later when connection pool is ready.
+  decodeMetadata(nullptr, *metadata_map);
+}
+
+void ConnectionManagerImpl::ActiveStream::decodeMetadata(ActiveStreamDecoderFilter* filter,
+                                                         MetadataMap& metadata_map) {
+  std::list<ActiveStreamDecoderFilterPtr>::iterator entry;
+  if (!filter) {
+    entry = decoder_filters_.begin();
+  } else {
+    entry = std::next(filter->entry());
+  }
+
+  for (; entry != decoder_filters_.end(); entry++) {
+    FilterMetadataStatus status = (*entry)->handle_->decodeMetadata(metadata_map);
+    ENVOY_STREAM_LOG(trace, "decode metadata called: filter={} status={}, metadata: {}", *this,
+                     static_cast<const void*>((*entry).get()), static_cast<uint64_t>(status),
+                     metadata_map);
+  }
+}
+
 void ConnectionManagerImpl::ActiveStream::maybeEndDecode(bool end_stream) {
   ASSERT(!state_.remote_complete_);
   state_.remote_complete_ = end_stream;
@@ -1398,6 +1445,17 @@ void ConnectionManagerImpl::ActiveStream::maybeEndEncode(bool end_stream) {
   }
 }
 
+bool ConnectionManagerImpl::ActiveStream::processNewlyAddedMetadata() {
+  if (request_metadata_map_vector_.empty()) {
+    return false;
+  }
+  for (const auto& metadata_map : request_metadata_map_vector_) {
+    decodeMetadata(nullptr, *metadata_map);
+  }
+  request_metadata_map_vector_.clear();
+  return true;
+}
+
 void ConnectionManagerImpl::ActiveStream::onResetStream(StreamResetReason) {
   // NOTE: This function gets called in all of the following cases:
   //       1) We TX an app level reset
@@ -1686,6 +1744,10 @@ void ConnectionManagerImpl::ActiveStreamDecoderFilter::addDecodedData(Buffer::In
   parent_.addDecodedData(*this, data, streaming);
 }
 
+MetadataMapVector& ConnectionManagerImpl::ActiveStreamDecoderFilter::addDecodedMetadata() {
+  return parent_.addDecodedMetadata();
+}
+
 void ConnectionManagerImpl::ActiveStreamDecoderFilter::continueDecoding() { commonContinue(); }
 
 void ConnectionManagerImpl::ActiveStreamDecoderFilter::encode100ContinueHeaders(

source/common/http/conn_manager_impl.h

@@ -172,6 +172,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     // Http::StreamDecoderFilterCallbacks
     void addDecodedData(Buffer::Instance& data, bool streaming) override;
     HeaderMap& addDecodedTrailers() override;
+    MetadataMapVector& addDecodedMetadata() override;
     void continueDecoding() override;
     const Buffer::Instance* decodingBuffer() override {
       return parent_.buffered_request_data_.get();
@@ -282,9 +283,11 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     const Network::Connection* connection();
     void addDecodedData(ActiveStreamDecoderFilter& filter, Buffer::Instance& data, bool streaming);
     HeaderMap& addDecodedTrailers();
+    MetadataMapVector& addDecodedMetadata();
     void decodeHeaders(ActiveStreamDecoderFilter* filter, HeaderMap& headers, bool end_stream);
     void decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instance& data, bool end_stream);
     void decodeTrailers(ActiveStreamDecoderFilter* filter, HeaderMap& trailers);
+    void decodeMetadata(ActiveStreamDecoderFilter* filter, MetadataMap& metadata_map);
     void disarmRequestTimeout();
     void maybeEndDecode(bool end_stream);
     void addEncodedData(ActiveStreamEncoderFilter& filter, Buffer::Instance& data, bool streaming);
@@ -299,6 +302,8 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     void encodeTrailers(ActiveStreamEncoderFilter* filter, HeaderMap& trailers);
     void encodeMetadata(ActiveStreamEncoderFilter* filter, MetadataMapPtr&& metadata_map_ptr);
     void maybeEndEncode(bool end_stream);
+    // Returns true if new metadata is inserted. Otherwise, returns false.
+    bool processNewlyAddedMetadata();
     uint64_t streamId() { return stream_id_; }
 
     // Http::StreamCallbacks
@@ -311,7 +316,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     void decodeHeaders(HeaderMapPtr&& headers, bool end_stream) override;
     void decodeData(Buffer::Instance& data, bool end_stream) override;
     void decodeTrailers(HeaderMapPtr&& trailers) override;
-    void decodeMetadata(MetadataMapPtr&&) override { NOT_REACHED_GCOVR_EXCL_LINE; }
+    void decodeMetadata(MetadataMapPtr&&) override;
 
     // Http::FilterChainFactoryCallbacks
     void addStreamDecoderFilter(StreamDecoderFilterSharedPtr filter) override {
@@ -423,6 +428,9 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     absl::optional<Router::RouteConstSharedPtr> cached_route_;
     absl::optional<Upstream::ClusterInfoConstSharedPtr> cached_cluster_info_;
     DownstreamWatermarkCallbacks* watermark_callbacks_{nullptr};
+    // Stores metadata added in the decoding filter that is being processed. Will be cleared before
+    // processing the next filter.
+    MetadataMapVector request_metadata_map_vector_;
     uint32_t buffer_limit_{0};
     uint32_t high_watermark_count_{0};
     const std::string* decorated_operation_{nullptr};
@@ -435,6 +443,8 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     // Whether a filter has indicated that the response should be treated as a headers only
     // response.
     bool encoding_headers_only_{};
+    // If an empty data frame with end_stream true is sent.
+    bool empty_data_end_stream_sent_{false};
   };
 
   typedef std::unique_ptr<ActiveStream> ActiveStreamPtr;

source/common/router/router.cc

@@ -453,6 +453,22 @@ Http::FilterTrailersStatus Filter::decodeTrailers(Http::HeaderMap& trailers) {
   return Http::FilterTrailersStatus::StopIteration;
 }
 
+Http::FilterMetadataStatus Filter::decodeMetadata(Http::MetadataMap& metadata_map) {
+  Http::MetadataMap metadata;
+  metadata.insert(metadata_map.begin(), metadata_map.end());
+  Http::MetadataMapPtr metadata_map_ptr = std::make_unique<Http::MetadataMap>(metadata);
+
+  if (upstream_request_ == nullptr) {
+    ENVOY_STREAM_LOG(trace, "upstream_request_ not ready. Storing metadata_map to encode later: {}",
+                     *callbacks_, metadata_map);
+    downstream_metadata_map_vector_.emplace_back(std::move(metadata_map_ptr));
+    return Http::FilterMetadataStatus::Continue;
+  }
+
+  upstream_request_->encodeMetadata(std::move(metadata_map_ptr));
+  return Http::FilterMetadataStatus::Continue;
+}
+
 void Filter::setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) {
   callbacks_ = &callbacks;
   // As the decoder filter only pushes back via watermarks once data has reached
@@ -1024,6 +1040,14 @@ void Filter::UpstreamRequest::encodeData(Buffer::Instance& data, bool end_stream
 
     buffered_request_body_->move(data);
   } else {
+    // Encodes metadata if exists.
+    if (!parent_.downstream_metadata_map_vector_.empty()) {
+      ENVOY_STREAM_LOG(trace, "Send metadata before sending data. {}", *parent_.callbacks_,
+                       parent_.downstream_metadata_map_vector_);
+      request_encoder_->encodeMetadata(parent_.downstream_metadata_map_vector_);
+      parent_.downstream_metadata_map_vector_.clear();
+    }
+
     ENVOY_STREAM_LOG(trace, "proxying {} bytes", *parent_.callbacks_, data.length());
     stream_info_.addBytesSent(data.length());
     request_encoder_->encodeData(data, end_stream);
@@ -1042,13 +1066,34 @@ void Filter::UpstreamRequest::encodeTrailers(const Http::HeaderMap& trailers) {
   if (!request_encoder_) {
     ENVOY_STREAM_LOG(trace, "buffering trailers", *parent_.callbacks_);
   } else {
+    // Encodes metadata if exists.
+    if (!parent_.downstream_metadata_map_vector_.empty()) {
+      ENVOY_STREAM_LOG(trace, "Send metadata before sending trailers. {}", *parent_.callbacks_,
+                       parent_.downstream_metadata_map_vector_);
+      request_encoder_->encodeMetadata(parent_.downstream_metadata_map_vector_);
+      parent_.downstream_metadata_map_vector_.clear();
+    }
+
     ENVOY_STREAM_LOG(trace, "proxying trailers", *parent_.callbacks_);
     request_encoder_->encodeTrailers(trailers);
     stream_info_.onLastUpstreamTxByteSent();
     parent_.callbacks_->streamInfo().onLastUpstreamTxByteSent();
   }
 }
 
+void Filter::UpstreamRequest::encodeMetadata(Http::MetadataMapPtr&& metadata_map_ptr) {
+  if (!request_encoder_) {
+    ENVOY_STREAM_LOG(trace, "request_encoder_ not ready. Store metadata_map to encode later: {}",
+                     *parent_.callbacks_, *metadata_map_ptr);
+    parent_.downstream_metadata_map_vector_.emplace_back(std::move(metadata_map_ptr));
+  } else {
+    ENVOY_STREAM_LOG(trace, "Encode metadata: {}", *parent_.callbacks_, *metadata_map_ptr);
+    Http::MetadataMapVector metadata_map_vector;
+    metadata_map_vector.emplace_back(std::move(metadata_map_ptr));
+    request_encoder_->encodeMetadata(metadata_map_vector);
+  }
+}
+
 void Filter::UpstreamRequest::onResetStream(Http::StreamResetReason reason) {
   clearRequestEncoder();
   if (!calling_encode_headers_) {
@@ -1145,8 +1190,12 @@ void Filter::UpstreamRequest::onPoolReady(Http::StreamEncoder& request_encoder,
 
   stream_info_.onFirstUpstreamTxByteSent();
   parent_.callbacks_->streamInfo().onFirstUpstreamTxByteSent();
+  bool end_stream = !buffered_request_body_ && encode_complete_ && !encode_trailers_;
+  // If end_stream is set in headers, and there are metadata to send, delays end_stream. The case
+  // only happens when decoding headers filters return ContinueAndEndStream.
+  bool delay_headers_end_stream = end_stream && !parent_.downstream_metadata_map_vector_.empty();
   request_encoder.encodeHeaders(*parent_.downstream_headers_,
-                                !buffered_request_body_ && encode_complete_ && !encode_trailers_);
+                                end_stream && !delay_headers_end_stream);
   calling_encode_headers_ = false;
 
   // It is possible to get reset in the middle of an encodeHeaders() call. This happens for example
@@ -1157,6 +1206,18 @@ void Filter::UpstreamRequest::onPoolReady(Http::StreamEncoder& request_encoder,
   if (deferred_reset_reason_) {
     onResetStream(deferred_reset_reason_.value());
   } else {
+    // Encode metadata after headers and before any other frame type.
+    if (!parent_.downstream_metadata_map_vector_.empty()) {
+      ENVOY_STREAM_LOG(trace, "Send metadata onPoolReady. {}", *parent_.callbacks_,
+                       parent_.downstream_metadata_map_vector_);
+      request_encoder.encodeMetadata(parent_.downstream_metadata_map_vector_);
+      parent_.downstream_metadata_map_vector_.clear();
+      if (delay_headers_end_stream) {
+        Buffer::OwnedImpl empty_data("");
+        request_encoder.encodeData(empty_data, true);
+      }
+    }
+
     if (buffered_request_body_) {
       stream_info_.addBytesSent(buffered_request_body_->length());
       request_encoder.encodeData(*buffered_request_body_, encode_complete_ && !encode_trailers_);

source/common/router/router.h

@@ -161,6 +161,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
   Http::FilterDataStatus decodeData(Buffer::Instance& data, bool end_stream) override;
   Http::FilterTrailersStatus decodeTrailers(Http::HeaderMap& trailers) override;
+  Http::FilterMetadataStatus decodeMetadata(Http::MetadataMap& metadata_map) override;
   void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override;
 
   // Upstream::LoadBalancerContext
@@ -274,6 +275,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
     void encodeHeaders(bool end_stream);
     void encodeData(Buffer::Instance& data, bool end_stream);
     void encodeTrailers(const Http::HeaderMap& trailers);
+    void encodeMetadata(Http::MetadataMapPtr&& metadata_map_ptr);
 
     void resetStream();
     void setupPerTryTimeout();
@@ -403,6 +405,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   bool grpc_request_{};
   Http::HeaderMap* downstream_headers_{};
   Http::HeaderMap* downstream_trailers_{};
+  Http::MetadataMapVector downstream_metadata_map_vector_;
   MonotonicTime downstream_request_complete_time_;
   uint32_t buffer_limit_{0};
   MetadataMatchCriteriaConstPtr metadata_match_;