http1 encode trailers in chunk encoding

api/envoy/api/v2/core/protocol.proto

@@ -46,6 +46,7 @@ message HttpProtocolOptions {
   google.protobuf.UInt32Value max_headers_count = 2 [(validate.rules).uint32 = {gte: 1}];
 }
 
+// [#next-free-field: 6]
 message Http1ProtocolOptions {
   message HeaderKeyFormat {
     message ProperCaseWords {
@@ -83,6 +84,9 @@ message Http1ProtocolOptions {
   // Describes how the keys for response headers should be formatted. By default, all header keys
   // are lower cased.
   HeaderKeyFormat header_key_format = 4;
+
+  // Enables encoding of trailers for HTTP1.
+  bool enable_trailers = 5;
 }
 
 // [#next-free-field: 13]

api/envoy/api/v3alpha/core/protocol.proto

@@ -46,6 +46,7 @@ message HttpProtocolOptions {
   google.protobuf.UInt32Value max_headers_count = 2 [(validate.rules).uint32 = {gte: 1}];
 }
 
+// [#next-free-field: 6]
 message Http1ProtocolOptions {
   message HeaderKeyFormat {
     message ProperCaseWords {
@@ -83,6 +84,9 @@ message Http1ProtocolOptions {
   // Describes how the keys for response headers should be formatted. By default, all header keys
   // are lower cased.
   HeaderKeyFormat header_key_format = 4;
+
+  // Enables encoding of trailers for HTTP1.
+  bool enable_trailers = 5;
 }
 
 // [#next-free-field: 13]

docs/root/intro/version_history.rst

@@ -17,6 +17,7 @@ Version history
 * tcp_proxy: added :ref:`ClusterWeight.metadata_match<envoy_api_field_config.filter.network.tcp_proxy.v2.TcpProxy.WeightedCluster.ClusterWeight.metadata_match>`
 * tcp_proxy: added :ref:`hash_policy<envoy_api_field_config.filter.network.tcp_proxy.v2.TcpProxy.hash_policy>`
 * tls: remove TLS 1.0 and 1.1 from client defaults
+* http: add support for encoding http1 trailers, to enable use :ref:`enable_trailers <envoy_api_field_core.Http1ProtocolOptions.enable_trailers>`.
 * router: exposed DOWNSTREAM_REMOTE_ADDRESS as custom HTTP request/response headers.
 
 1.12.0 (October 31, 2019)

include/envoy/http/codec.h

@@ -229,6 +229,8 @@ struct Http1Settings {
   bool accept_http_10_{false};
   // Set a default host if no Host: header is present for HTTP/1.0 requests.`
   std::string default_host_for_http_10_;
+  // Encode trailers in Http1
+  bool enable_trailers_{false};
 
   enum class HeaderKeyFormat {
     // By default no formatting is performed, presenting all headers in lowercase (as Envoy

source/common/http/http1/codec_impl.cc

@@ -40,7 +40,8 @@ HeaderKeyFormatterPtr formatter(const Http::Http1Settings& settings) {
 } // namespace
 
 const std::string StreamEncoderImpl::CRLF = "\r\n";
-const std::string StreamEncoderImpl::LAST_CHUNK = "0\r\n\r\n";
+// Last chunk as defined here https://tools.ietf.org/html/rfc7230#section-4.1
+const std::string StreamEncoderImpl::LAST_CHUNK = "0\r\n";
 
 StreamEncoderImpl::StreamEncoderImpl(ConnectionImpl& connection,
                                      HeaderKeyFormatter* header_key_formatter)
@@ -187,7 +188,31 @@ void StreamEncoderImpl::encodeData(Buffer::Instance& data, bool end_stream) {
   }
 }
 
-void StreamEncoderImpl::encodeTrailers(const HeaderMap&) { endEncode(); }
+void StreamEncoderImpl::encodeTrailers(const HeaderMap& trailers) {
+  if (!connection_.enable_trailers_) {
+    return endEncode();
+  }
+  // Trailers only matter if it is a chunk transfer encoding
+  // https://tools.ietf.org/html/rfc7230#section-4.4
+  if (chunk_encoding_) {
+    // Finalize the body
+    connection_.buffer().add(LAST_CHUNK);
+
+    trailers.iterate(
+        [](const HeaderEntry& header, void* context) -> HeaderMap::Iterate {
+          static_cast<StreamEncoderImpl*>(context)->encodeFormattedHeader(
+              header.key().getStringView(), header.value().getStringView());
+          return HeaderMap::Iterate::Continue;
+        },
+        this);
+
+    connection_.flushOutput();
+    connection_.buffer().add(CRLF);
+  }
+
+  connection_.flushOutput();
+  connection_.onEncodeComplete();
+}
 
 void StreamEncoderImpl::encodeMetadata(const MetadataMapVector&) {
   connection_.stats().metadata_not_supported_error_.inc();
@@ -196,6 +221,7 @@ void StreamEncoderImpl::encodeMetadata(const MetadataMapVector&) {
 void StreamEncoderImpl::endEncode() {
   if (chunk_encoding_) {
     connection_.buffer().add(LAST_CHUNK);
+    connection_.buffer().add(CRLF);
   }
 
   connection_.flushOutput();
@@ -353,8 +379,9 @@ const ToLowerTable& ConnectionImpl::toLowerTable() {
 ConnectionImpl::ConnectionImpl(Network::Connection& connection, Stats::Scope& stats,
                                http_parser_type type, uint32_t max_headers_kb,
                                const uint32_t max_headers_count,
-                               HeaderKeyFormatterPtr&& header_key_formatter)
-    : connection_(connection), stats_{ALL_HTTP1_CODEC_STATS(POOL_COUNTER_PREFIX(stats, "http1."))},
+                               HeaderKeyFormatterPtr&& header_key_formatter, bool enable_trailers)
+    : enable_trailers_(enable_trailers),
+      connection_(connection), stats_{ALL_HTTP1_CODEC_STATS(POOL_COUNTER_PREFIX(stats, "http1."))},
       header_key_formatter_(std::move(header_key_formatter)),
       output_buffer_([&]() -> void { this->onBelowLowWatermark(); },
                      [&]() -> void { this->onAboveHighWatermark(); }),
@@ -369,11 +396,13 @@ ConnectionImpl::ConnectionImpl(Network::Connection& connection, Stats::Scope& st
 void ConnectionImpl::completeLastHeader() {
   ENVOY_CONN_LOG(trace, "completed header: key={} value={}", connection_,
                  current_header_field_.getStringView(), current_header_value_.getStringView());
+
   if (!current_header_field_.empty()) {
     toLowerTable().toLowerCase(current_header_field_.buffer(), current_header_field_.size());
     current_header_map_->addViaMove(std::move(current_header_field_),
                                     std::move(current_header_value_));
   }
+
   // Check if the number of headers exceeds the limit.
   if (current_header_map_->size() > max_headers_count_) {
     error_code_ = Http::Code::RequestHeaderFieldsTooLarge;
@@ -447,11 +476,13 @@ size_t ConnectionImpl::dispatchSlice(const char* slice, size_t len) {
 }
 
 void ConnectionImpl::onHeaderField(const char* data, size_t length) {
+  ENVOY_CONN_LOG(trace, "onHeaderField: data={}", connection_, absl::string_view(data, length));
+  // We previously already finished up the headers, these headers are
+  // now trailers
   if (header_parsing_state_ == HeaderParsingState::Done) {
-    // Ignore trailers.
-    return;
+    processing_trailers_ = true;
+    header_parsing_state_ = HeaderParsingState::Field;
   }
-
   if (header_parsing_state_ == HeaderParsingState::Value) {
     completeLastHeader();
   }
@@ -460,12 +491,11 @@ void ConnectionImpl::onHeaderField(const char* data, size_t length) {
 }
 
 void ConnectionImpl::onHeaderValue(const char* data, size_t length) {
-  if (header_parsing_state_ == HeaderParsingState::Done) {
-    // Ignore trailers.
-    return;
-  }
-
   const absl::string_view header_value = absl::string_view(data, length);
+  ENVOY_CONN_LOG(trace, "onHeaderValue: data={}", connection_, header_value);
+  if (!current_header_map_) {
+    current_header_map_ = std::make_unique<HeaderMapImpl>();
+  }
 
   if (strict_header_validation_) {
     if (!Http::HeaderUtility::headerIsValid(header_value)) {
@@ -498,8 +528,9 @@ void ConnectionImpl::onHeaderValue(const char* data, size_t length) {
 }
 
 int ConnectionImpl::onHeadersCompleteBase() {
-  ENVOY_CONN_LOG(trace, "headers complete", connection_);
+  ENVOY_CONN_LOG(trace, "onHeadersCompleteBase complete", connection_);
   completeLastHeader();
+
   // Validate that the completed HeaderMap's cached byte size exists and is correct.
   // This assert iterates over the HeaderMap.
   ASSERT(current_header_map_->byteSize().has_value() &&
@@ -536,6 +567,7 @@ int ConnectionImpl::onHeadersCompleteBase() {
 
   int rc = onHeadersComplete(std::move(current_header_map_));
   current_header_map_.reset();
+
   header_parsing_state_ = HeaderParsingState::Done;
 
   // Returning 2 informs http_parser to not expect a body or further data on this connection.
@@ -544,6 +576,7 @@ int ConnectionImpl::onHeadersCompleteBase() {
 
 void ConnectionImpl::onMessageCompleteBase() {
   ENVOY_CONN_LOG(trace, "message complete", connection_);
+
   if (handling_upgrade_) {
     // If this is an upgrade request, swallow the onMessageComplete. The
     // upgrade payload will be treated as stream body.
@@ -552,7 +585,14 @@ void ConnectionImpl::onMessageCompleteBase() {
     http_parser_pause(&parser_, 1);
     return;
   }
-  onMessageComplete();
+
+  // If true, this indicates we were processing trailers and must
+  // move the last header into current_header_map_
+  if (header_parsing_state_ == HeaderParsingState::Value) {
+    completeLastHeader();
+  }
+
+  onMessageComplete(std::move(current_header_map_));
 }
 
 void ConnectionImpl::onMessageBeginBase() {
@@ -578,7 +618,7 @@ ServerConnectionImpl::ServerConnectionImpl(Network::Connection& connection, Stat
                                            Http1Settings settings, uint32_t max_request_headers_kb,
                                            const uint32_t max_request_headers_count)
     : ConnectionImpl(connection, stats, HTTP_REQUEST, max_request_headers_kb,
-                     max_request_headers_count, formatter(settings)),
+                     max_request_headers_count, formatter(settings), settings.enable_trailers_),
       callbacks_(callbacks), codec_settings_(settings) {}
 
 void ServerConnectionImpl::onEncodeComplete() {
@@ -635,6 +675,8 @@ void ServerConnectionImpl::handlePath(HeaderMapImpl& headers, unsigned int metho
 }
 
 int ServerConnectionImpl::onHeadersComplete(HeaderMapImplPtr&& headers) {
+  ENVOY_CONN_LOG(trace, "Server: onHeadersComplete size={}", connection_, headers->size());
+
   // Handle the case where response happens prior to request complete. It's up to upper layer code
   // to disconnect the connection but we shouldn't fire any more events since it doesn't make
   // sense.
@@ -699,7 +741,7 @@ void ServerConnectionImpl::onBody(const char* data, size_t length) {
   }
 }
 
-void ServerConnectionImpl::onMessageComplete() {
+void ServerConnectionImpl::onMessageComplete(HeaderMapImplPtr&& trailers) {
   if (active_request_) {
     Buffer::OwnedImpl buffer;
     active_request_->remote_complete_ = true;
@@ -708,6 +750,8 @@ void ServerConnectionImpl::onMessageComplete() {
       active_request_->request_decoder_->decodeHeaders(std::move(deferred_end_stream_headers_),
                                                        true);
       deferred_end_stream_headers_.reset();
+    } else if (processing_trailers_) {
+      active_request_->request_decoder_->decodeTrailers(std::move(trailers));
     } else {
       active_request_->request_decoder_->decodeData(buffer, true);
     }
@@ -754,7 +798,7 @@ ClientConnectionImpl::ClientConnectionImpl(Network::Connection& connection, Stat
                                            ConnectionCallbacks&, const Http1Settings& settings,
                                            const uint32_t max_response_headers_count)
     : ConnectionImpl(connection, stats, HTTP_RESPONSE, MAX_RESPONSE_HEADERS_KB,
-                     max_response_headers_count, formatter(settings)) {}
+                     max_response_headers_count, formatter(settings), settings.enable_trailers_) {}
 
 bool ClientConnectionImpl::cannotHaveBody() {
   if ((!pending_responses_.empty() && pending_responses_.front().head_request_) ||
@@ -787,6 +831,7 @@ void ClientConnectionImpl::onEncodeHeaders(const HeaderMap& headers) {
 }
 
 int ClientConnectionImpl::onHeadersComplete(HeaderMapImplPtr&& headers) {
+  ENVOY_CONN_LOG(trace, "Client: onHeadersComplete size={}", connection_, headers->size());
   headers->setStatus(parser_.status_code);
 
   // Handle the case where the client is closing a kept alive connection (by sending a 408
@@ -821,7 +866,7 @@ void ClientConnectionImpl::onBody(const char* data, size_t length) {
   }
 }
 
-void ClientConnectionImpl::onMessageComplete() {
+void ClientConnectionImpl::onMessageComplete(HeaderMapImplPtr&& trailers) {
   ENVOY_CONN_LOG(trace, "message complete", connection_);
   if (ignore_message_complete_for_100_continue_) {
     ignore_message_complete_for_100_continue_ = false;
@@ -847,6 +892,8 @@ void ClientConnectionImpl::onMessageComplete() {
     if (deferred_end_stream_headers_) {
       response.decoder_->decodeHeaders(std::move(deferred_end_stream_headers_), true);
       deferred_end_stream_headers_.reset();
+    } else if (processing_trailers_) {
+      response.decoder_->decodeTrailers(std::move(trailers));
     } else {
       Buffer::OwnedImpl buffer;
       response.decoder_->decodeData(buffer, true);

source/common/http/http1/codec_impl.h

@@ -192,10 +192,12 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
 
   CodecStats& stats() { return stats_; }
 
+  const bool enable_trailers_;
+
 protected:
   ConnectionImpl(Network::Connection& connection, Stats::Scope& stats, http_parser_type type,
                  uint32_t max_headers_kb, const uint32_t max_headers_count,
-                 HeaderKeyFormatterPtr&& header_key_formatter);
+                 HeaderKeyFormatterPtr&& header_key_formatter, bool enable_trailers);
 
   bool resetStreamCalled() { return reset_stream_called_; }
 
@@ -204,6 +206,7 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
   http_parser parser_;
   HeaderMapPtr deferred_end_stream_headers_;
   Http::Code error_code_{Http::Code::BadRequest};
+  bool processing_trailers_{};
   bool handling_upgrade_{};
   const HeaderKeyFormatterPtr header_key_formatter_;
 
@@ -269,7 +272,7 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
    * Called when the request/response is complete.
    */
   void onMessageCompleteBase();
-  virtual void onMessageComplete() PURE;
+  virtual void onMessageComplete(HeaderMapImplPtr&& trailers) PURE;
 
   /**
    * @see onResetStreamBase().
@@ -353,7 +356,7 @@ class ServerConnectionImpl : public ServerConnection, public ConnectionImpl {
   void onUrl(const char* data, size_t length) override;
   int onHeadersComplete(HeaderMapImplPtr&& headers) override;
   void onBody(const char* data, size_t length) override;
-  void onMessageComplete() override;
+  void onMessageComplete(HeaderMapImplPtr&& trailers) override;
   void onResetStream(StreamResetReason reason) override;
   void sendProtocolError() override;
   void onAboveHighWatermark() override;
@@ -393,7 +396,7 @@ class ClientConnectionImpl : public ClientConnection, public ConnectionImpl {
   void onUrl(const char*, size_t) override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
   int onHeadersComplete(HeaderMapImplPtr&& headers) override;
   void onBody(const char* data, size_t length) override;
-  void onMessageComplete() override;
+  void onMessageComplete(HeaderMapImplPtr&& trailers) override;
   void onResetStream(StreamResetReason reason) override;
   void sendProtocolError() override {}
   void onAboveHighWatermark() override;

source/common/http/utility.cc

@@ -276,6 +276,7 @@ Utility::parseHttp1Settings(const envoy::api::v2::core::Http1ProtocolOptions& co
   ret.allow_absolute_url_ = PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, allow_absolute_url, true);
   ret.accept_http_10_ = config.accept_http_10();
   ret.default_host_for_http_10_ = config.default_host_for_http_10();
+  ret.enable_trailers_ = config.enable_trailers();
 
   if (config.header_key_format().has_proper_case_words()) {
     ret.header_key_format_ = Http1Settings::HeaderKeyFormat::ProperCase;

source/extensions/filters/http/grpc_http1_reverse_bridge/filter.cc

@@ -172,21 +172,8 @@ Http::FilterDataStatus Filter::encodeData(Buffer::Instance& buffer, bool end_str
     trailers.setGrpcStatus(grpc_status_);
 
     if (withhold_grpc_frames_) {
-      // Compute the size of the payload and construct the length prefix.
-      //
-      // We do this even if the upstream failed: If the response returned non-200,
-      // we'll respond with a grpc-status with an error, so clients will know that the request
-      // was unsuccessful. Since we're guaranteed at this point to have a valid response
-      // (unless upstream lied in content-type) we attempt to return a well-formed gRPC
-      // response body.
-      const auto length = buffer.length() + buffer_.length();
-
-      std::array<uint8_t, Grpc::GRPC_FRAME_HEADER_SIZE> frame;
-      Grpc::Encoder().newFrame(Grpc::GRPC_FH_DEFAULT, length, frame);
-
       buffer.prepend(buffer_);
-      Buffer::OwnedImpl frame_buffer(frame.data(), frame.size());
-      buffer.prepend(frame_buffer);
+      buildGrpcFrameHeader(buffer);
     }
 
     return Http::FilterDataStatus::Continue;
@@ -203,6 +190,32 @@ Http::FilterDataStatus Filter::encodeData(Buffer::Instance& buffer, bool end_str
   }
 }
 
+Http::FilterTrailersStatus Filter::encodeTrailers(Http::HeaderMap& trailers) {
+  trailers.insertGrpcStatus().value(grpc_status_);
+
+  if (withhold_grpc_frames_) {
+    buildGrpcFrameHeader(buffer_);
+    encoder_callbacks_->addEncodedData(buffer_, false);
+  }
+
+  return Http::FilterTrailersStatus::Continue;
+}
+
+void Filter::buildGrpcFrameHeader(Buffer::Instance& buffer) {
+  // Compute the size of the payload and construct the length prefix.
+  //
+  // We do this even if the upstream failed: If the response returned non-200,
+  // we'll respond with a grpc-status with an error, so clients will know that the request
+  // was unsuccessful. Since we're guaranteed at this point to have a valid response
+  // (unless upstream lied in content-type) we attempt to return a well-formed gRPC
+  // response body.
+  const auto length = buffer.length();
+  std::array<uint8_t, Grpc::GRPC_FRAME_HEADER_SIZE> frame;
+  Grpc::Encoder().newFrame(Grpc::GRPC_FH_DEFAULT, length, frame);
+  Buffer::OwnedImpl frame_buffer(frame.data(), frame.size());
+  buffer.prepend(frame_buffer);
+}
+
 } // namespace GrpcHttp1ReverseBridge
 } // namespace HttpFilters
 } // namespace Extensions