http_filter: add addEncodedTrailers and addDecodedTrailers

include/envoy/http/filter.h

@@ -190,6 +190,18 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
    */
   virtual void addDecodedData(Buffer::Instance& data, bool streaming_filter) PURE;
 
+   /**
+    * Adds decoded trailers. May only be called in decodeData when end_stream is set to true or in
+    * decodeTrailers. If called in any other context, std::logic_error will be thrown.
+    *
+    * When called in decodeData, the trailers map will be initialized to an empty map and returned by
+    * reference. Calling it more than once is invalid and will result in std::logic_error being thrown.
+    * 
+    * When called in decodeTrailers it will simply return the trailers (same data as passed in the 
+    * parameter).
+    */
+  virtual HeaderMap& addDecodedTrailers() 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
@@ -395,6 +407,18 @@ class StreamEncoderFilterCallbacks : public virtual StreamFilterCallbacks {
    */
   virtual void addEncodedData(Buffer::Instance& data, bool streaming_filter) PURE;
 
+  /**
+   * Adds encoded trailers. May only be called in encodeData when end_stream is set to true or in
+   * encodeTrailers. If called in any other context, std::logic_error will be thrown.
+   *
+   * When called in encodeData, the trailers map will be initialized to an empty map and returned by
+   * reference. Calling it more than once is invalid and will result in std::logic_error being thrown.
+   * 
+   * When called in encodeTrailers it will simply return the trailers (same data as passed in the 
+   * parameter).
+   */
+  virtual HeaderMap& addEncodedTrailers() PURE;
+
   /**
    * Called when an encoder filter goes over its high watermark.
    */

source/common/http/async_client_impl.h

@@ -268,6 +268,7 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   Tracing::Span& activeSpan() override { return active_span_; }
   const Tracing::Config& tracingConfig() override { return tracing_config_; }
   void continueDecoding() override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
+  HeaderMap& addDecodedTrailers() override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
   void addDecodedData(Buffer::Instance&, bool) override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
   const Buffer::Instance* decodingBuffer() override { return buffered_body_.get(); }
   void sendLocalReply(Code code, const std::string& body,

source/common/http/conn_manager_impl.cc

@@ -775,15 +775,45 @@ void ConnectionManagerImpl::ActiveStream::decodeData(ActiveStreamDecoderFilter*
 
   for (; entry != decoder_filters_.end(); entry++) {
     ASSERT(!(state_.filter_call_state_ & FilterCallState::DecodeData));
+
+    // We check the request_trailers_ pointer here in case addDecodedTrailers
+    // is called in decodeData - at which point we communicate to the filter
+    // that the stream has not yet ended.
+    bool end_stream_no_trailers = end_stream && !request_trailers_;
+    if (end_stream_no_trailers) {
+      state_.filter_call_state_ |= FilterCallState::LastDataFrame;
+    }
     state_.filter_call_state_ |= FilterCallState::DecodeData;
-    FilterDataStatus status = (*entry)->handle_->decodeData(data, end_stream);
+    FilterDataStatus status = (*entry)->handle_->decodeData(data, end_stream_no_trailers);
     state_.filter_call_state_ &= ~FilterCallState::DecodeData;
+    if (end_stream_no_trailers) {
+      state_.filter_call_state_ &= ~FilterCallState::LastDataFrame;
+    }
     ENVOY_STREAM_LOG(trace, "decode data called: filter={} status={}", *this,
                      static_cast<const void*>((*entry).get()), static_cast<uint64_t>(status));
     if (!(*entry)->commonHandleAfterDataCallback(status, data, state_.decoder_filters_streaming_)) {
       return;
     }
   }
+
+  // If trailers were adding during decodeData we need to trigger decodeTrailers in order
+  // to allow filters to process the trailers.
+  if (end_stream && request_trailers_) {
+    decodeTrailers(filter, *request_trailers_);
+  }
+}
+
+HeaderMap& ConnectionManagerImpl::ActiveStream::addDecodedTrailers() {
+  if (state_.filter_call_state_ & FilterCallState::LastDataFrame) {
+    if (request_trailers_) {
+      throw std::logic_error("decodedTrailers added more than once");
+    }
+
+    request_trailers_ = std::make_unique<HeaderMapImpl>();
+    return *request_trailers_;
+  } else {
+    throw std::logic_error("addDecodedTrailers called in invalid context");
+  }
 }
 
 void ConnectionManagerImpl::ActiveStream::addDecodedData(ActiveStreamDecoderFilter& filter,
@@ -1058,6 +1088,24 @@ void ConnectionManagerImpl::ActiveStream::encodeHeaders(ActiveStreamEncoderFilte
   }
 }
 
+HeaderMap& ConnectionManagerImpl::ActiveStream::addEncodedTrailers() {
+  if (state_.filter_call_state_ & FilterCallState::LastDataFrame) {
+    if (response_trailers_) {
+      throw std::logic_error("encodedTrailers added more than once");
+    }
+
+    // local_complete_ is set to true at the start of encodeData(..., true), but since
+    // we've now added trailers we have to undo this to prevent encodeTrailers from
+    // blowing up when it assert on local_complete_
+    state_.local_complete_ = false;
+
+    response_trailers_ = std::make_unique<HeaderMapImpl>();
+    return *response_trailers_;
+  } else {
+    throw std::logic_error("encodedTrailers called in invalid context");
+  }
+}
+
 void ConnectionManagerImpl::ActiveStream::addEncodedData(ActiveStreamEncoderFilter& filter,
                                                          Buffer::Instance& data, bool streaming) {
   if (state_.filter_call_state_ == 0 ||
@@ -1085,9 +1133,20 @@ void ConnectionManagerImpl::ActiveStream::encodeData(ActiveStreamEncoderFilter*
   std::list<ActiveStreamEncoderFilterPtr>::iterator entry = commonEncodePrefix(filter, end_stream);
   for (; entry != encoder_filters_.end(); entry++) {
     ASSERT(!(state_.filter_call_state_ & FilterCallState::EncodeData));
+
+    // We check the request_trailers_ pointer here in case addEncodedTrailers
+    // is called in encodeData - at which point we communicate to the filter
+    // that the stream has not yet ended.
+    bool end_stream_no_trailers = end_stream && !response_trailers_;
     state_.filter_call_state_ |= FilterCallState::EncodeData;
-    FilterDataStatus status = (*entry)->handle_->encodeData(data, end_stream);
+    if (end_stream_no_trailers) {
+      state_.filter_call_state_ |= FilterCallState::LastDataFrame;
+    }
+    FilterDataStatus status = (*entry)->handle_->encodeData(data, end_stream_no_trailers);
     state_.filter_call_state_ &= ~FilterCallState::EncodeData;
+    if (end_stream_no_trailers) {
+      state_.filter_call_state_ &= ~FilterCallState::LastDataFrame;
+    }
     ENVOY_STREAM_LOG(trace, "encode data called: filter={} status={}", *this,
                      static_cast<const void*>((*entry).get()), static_cast<uint64_t>(status));
     if (!(*entry)->commonHandleAfterDataCallback(status, data, state_.encoder_filters_streaming_)) {
@@ -1099,8 +1158,16 @@ void ConnectionManagerImpl::ActiveStream::encodeData(ActiveStreamEncoderFilter*
                    end_stream);
 
   request_info_.addBytesSent(data.length());
-  response_encoder_->encodeData(data, end_stream);
-  maybeEndEncode(end_stream);
+
+  // If trailers were adding during encodeData we need to trigger decodeTrailers in order
+  // to allow filters to process the trailers.
+  if (end_stream && response_trailers_) {
+    response_encoder_->encodeData(data, false);
+    encodeTrailers(filter, *response_trailers_);
+  } else {
+    response_encoder_->encodeData(data, end_stream);
+    maybeEndEncode(end_stream);
+  }
 }
 
 void ConnectionManagerImpl::ActiveStream::encodeTrailers(ActiveStreamEncoderFilter* filter,
@@ -1380,6 +1447,10 @@ Buffer::WatermarkBufferPtr ConnectionManagerImpl::ActiveStreamDecoderFilter::cre
   return buffer;
 }
 
+HeaderMap& ConnectionManagerImpl::ActiveStreamDecoderFilter::addDecodedTrailers() {
+  return parent_.addDecodedTrailers();
+}
+
 void ConnectionManagerImpl::ActiveStreamDecoderFilter::addDecodedData(Buffer::Instance& data,
                                                                       bool streaming) {
   parent_.addDecodedData(*this, data, streaming);
@@ -1473,6 +1544,10 @@ void ConnectionManagerImpl::ActiveStreamEncoderFilter::addEncodedData(Buffer::In
   return parent_.addEncodedData(*this, data, streaming);
 }
 
+HeaderMap& ConnectionManagerImpl::ActiveStreamEncoderFilter::addEncodedTrailers() {
+  return parent_.addEncodedTrailers();
+}
+
 void ConnectionManagerImpl::ActiveStreamEncoderFilter::
     onEncoderFilterAboveWriteBufferHighWatermark() {
   ENVOY_STREAM_LOG(debug, "Disabling upstream stream due to filter callbacks.", parent_);

source/common/http/conn_manager_impl.h

@@ -162,6 +162,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
 
     // Http::StreamDecoderFilterCallbacks
     void addDecodedData(Buffer::Instance& data, bool streaming) override;
+    HeaderMap& addDecodedTrailers() override;
     void continueDecoding() override;
     const Buffer::Instance* decodingBuffer() override {
       return parent_.buffered_request_data_.get();
@@ -229,6 +230,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
 
     // Http::StreamEncoderFilterCallbacks
     void addEncodedData(Buffer::Instance& data, bool streaming) override;
+    HeaderMap& addEncodedTrailers() override;
     void onEncoderFilterAboveWriteBufferHighWatermark() override;
     void onEncoderFilterBelowWriteBufferLowWatermark() override;
     void setEncoderBufferLimit(uint32_t limit) override { parent_.setBufferLimit(limit); }
@@ -267,11 +269,13 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     commonEncodePrefix(ActiveStreamEncoderFilter* filter, bool end_stream);
     const Network::Connection* connection();
     void addDecodedData(ActiveStreamDecoderFilter& filter, Buffer::Instance& data, bool streaming);
+    HeaderMap& addDecodedTrailers();
     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 maybeEndDecode(bool end_stream);
     void addEncodedData(ActiveStreamEncoderFilter& filter, Buffer::Instance& data, bool streaming);
+    HeaderMap& addEncodedTrailers();
     void sendLocalReply(bool is_grpc_request, Code code, const std::string& body,
                         std::function<void(HeaderMap& headers)> modify_headers);
     void encode100ContinueHeaders(ActiveStreamEncoderFilter* filter, HeaderMap& headers);
@@ -339,6 +343,9 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
       // to verify we do not encode100Continue headers more than once per
       // filter.
       static constexpr uint32_t Encode100ContinueHeaders  = 0x40;
+      // Used to indicate that we're processing the final [en|de]Code frame,
+      // i.e. end_stream = true
+      static constexpr uint32_t LastDataFrame = 0x80;
     };
     // clang-format on