http2: migrates DATA frame payloads to the visitor API

changelogs/current.yaml

@@ -16,6 +16,10 @@ behavior_changes:
   change: |
     Changes the default value of ``envoy.reloadable_features.http2_use_oghttp2`` to true. This changes the codec used for HTTP/2
     requests and responses. This behavior can be reverted by setting the feature to false.
+- area: http2
+  change: |
+    Passes HTTP/2 DATA frames through a different codec API. This behavior can be temporarily disabled by setting the runtime
+    feature ``envoy.reloadable_features.http2_use_visitor_for_data`` to false.
 - area: proxy_protocol
   change: |
     Populate typed metadata by default in proxy protocol listener. Typed metadata can be consumed as

source/common/http/http2/codec_impl.cc

@@ -368,7 +368,7 @@ void ConnectionImpl::StreamImpl::processBufferedData() {
     // We only buffer the onStreamClose if we had no errors.
     if (Status status = parent_.onStreamClose(this, 0); !status.ok()) {
       ENVOY_CONN_LOG(debug, "error invoking onStreamClose: {}", parent_.connection_,
-                     status.message()); // GCOV_EXCL_LINE
+                     status.message()); // LCOV_EXCL_LINE
     }
   }
 }
@@ -663,8 +663,12 @@ bool ConnectionImpl::StreamDataFrameSource::Send(absl::string_view frame_header,
 
 void ConnectionImpl::ClientStreamImpl::submitHeaders(const HeaderMap& headers, bool end_stream) {
   ASSERT(stream_id_ == -1);
+  const bool skip_frame_source =
+      end_stream ||
+      Runtime::runtimeFeatureEnabled("envoy.reloadable_features.http2_use_visitor_for_data");
   stream_id_ = parent_.adapter_->SubmitRequest(
-      buildHeaders(headers), end_stream ? nullptr : std::make_unique<StreamDataFrameSource>(*this),
+      buildHeaders(headers),
+      skip_frame_source ? nullptr : std::make_unique<StreamDataFrameSource>(*this), end_stream,
       base());
   ASSERT(stream_id_ > 0);
 }
@@ -685,9 +689,12 @@ void ConnectionImpl::ClientStreamImpl::advanceHeadersState() {
 
 void ConnectionImpl::ServerStreamImpl::submitHeaders(const HeaderMap& headers, bool end_stream) {
   ASSERT(stream_id_ != -1);
-  parent_.adapter_->SubmitResponse(stream_id_, buildHeaders(headers),
-                                   end_stream ? nullptr
-                                              : std::make_unique<StreamDataFrameSource>(*this));
+  const bool skip_frame_source =
+      end_stream ||
+      Runtime::runtimeFeatureEnabled("envoy.reloadable_features.http2_use_visitor_for_data");
+  parent_.adapter_->SubmitResponse(
+      stream_id_, buildHeaders(headers),
+      skip_frame_source ? nullptr : std::make_unique<StreamDataFrameSource>(*this), end_stream);
 }
 
 Status ConnectionImpl::ServerStreamImpl::onBeginHeaders() {
@@ -780,7 +787,7 @@ void ConnectionImpl::StreamImpl::resetStream(StreamResetReason reason) {
     // its stream close.
     if (Status status = parent_.onStreamClose(this, 0); !status.ok()) {
       ENVOY_CONN_LOG(debug, "error invoking onStreamClose: {}", parent_.connection_,
-                     status.message()); // GCOV_EXCL_LINE
+                     status.message()); // LCOV_EXCL_LINE
     }
     return;
   }
@@ -978,7 +985,7 @@ Http::Status ConnectionImpl::dispatch(Buffer::Instance& data) {
     // protections that Envoy's codec does not have.
     if (rc == NGHTTP2_ERR_FLOODED) {
       return bufferFloodError(
-          "Flooding was detected in this HTTP/2 session, and it must be closed"); // GCOV_EXCL_LINE
+          "Flooding was detected in this HTTP/2 session, and it must be closed"); // LCOV_EXCL_LINE
     }
     if (rc != static_cast<ssize_t>(slice.len_)) {
       return codecProtocolError(nghttp2_strerror(rc));
@@ -1181,7 +1188,7 @@ Status ConnectionImpl::onHeaders(int32_t stream_id, size_t length, uint8_t flags
 
   default:
     // We do not currently support push.
-    ENVOY_BUG(false, "push not supported"); // GCOV_EXCL_LINE
+    ENVOY_BUG(false, "push not supported"); // LCOV_EXCL_LINE
   }
 
   stream->advanceHeadersState();
@@ -1305,7 +1312,7 @@ int ConnectionImpl::onInvalidFrame(int32_t stream_id, int error_code) {
 
   default:
     // Unknown error conditions. Trigger ENVOY_BUG and connection close.
-    ENVOY_BUG(false, absl::StrCat("Unexpected error_code: ", error_code)); // GCOV_EXCL_LINE
+    ENVOY_BUG(false, absl::StrCat("Unexpected error_code: ", error_code)); // LCOV_EXCL_LINE
     break;
   }
 
@@ -1692,6 +1699,58 @@ int64_t ConnectionImpl::Http2Visitor::OnReadyToSend(absl::string_view serialized
                              serialized.size());
 }
 
+ConnectionImpl::Http2Visitor::DataFrameHeaderInfo
+ConnectionImpl::Http2Visitor::OnReadyToSendDataForStream(Http2StreamId stream_id,
+                                                         size_t max_length) {
+  StreamImpl* stream = connection_->getStream(stream_id);
+  if (stream == nullptr) {
+    return {/*payload_length=*/-1, /*end_data=*/false, /*end_stream=*/false};
+  }
+  if (stream->pending_send_data_->length() == 0 && !stream->local_end_stream_) {
+    stream->data_deferred_ = true;
+    return {/*payload_length=*/0, /*end_data=*/false, /*end_stream=*/false};
+  }
+  const size_t length = std::min<size_t>(max_length, stream->pending_send_data_->length());
+  bool end_data = false;
+  bool end_stream = false;
+  if (stream->local_end_stream_ && length == stream->pending_send_data_->length()) {
+    end_data = true;
+    if (stream->pending_trailers_to_encode_) {
+      stream->submitTrailers(*stream->pending_trailers_to_encode_);
+      stream->pending_trailers_to_encode_.reset();
+    } else {
+      end_stream = true;
+    }
+  }
+  return {static_cast<int64_t>(length), end_data, end_stream};
+}
+
+bool ConnectionImpl::Http2Visitor::SendDataFrame(Http2StreamId stream_id,
+                                                 absl::string_view frame_header,
+                                                 size_t payload_length) {
+  connection_->protocol_constraints_.incrementOutboundDataFrameCount();
+
+  StreamImpl* stream = connection_->getStream(stream_id);
+  if (stream == nullptr) {
+    ENVOY_CONN_LOG(error, "error sending data frame: stream {} not found", connection_->connection_,
+                   stream_id);
+    return false;
+  }
+  Buffer::OwnedImpl output;
+  connection_->addOutboundFrameFragment(
+      output, reinterpret_cast<const uint8_t*>(frame_header.data()), frame_header.size());
+  if (!connection_->protocol_constraints_.checkOutboundFrameLimits().ok()) {
+    ENVOY_CONN_LOG(debug, "error sending data frame: Too many frames in the outbound queue",
+                   connection_->connection_);
+    stream->setDetails(Http2ResponseCodeDetails::get().outbound_frame_flood);
+  }
+
+  connection_->stats_.pending_send_bytes_.sub(payload_length);
+  output.move(*stream->pending_send_data_, payload_length);
+  connection_->connection_.write(output, false);
+  return true;
+}
+
 bool ConnectionImpl::Http2Visitor::OnFrameHeader(Http2StreamId stream_id, size_t length,
                                                  uint8_t type, uint8_t flags) {
   ENVOY_CONN_LOG(debug, "Http2Visitor::OnFrameHeader({}, {}, {}, {})", connection_->connection_,

source/common/http/http2/codec_impl.h

@@ -176,6 +176,10 @@ class ConnectionImpl : public virtual Connection,
       stream_close_listener_ = std::move(f);
     }
     int64_t OnReadyToSend(absl::string_view serialized) override;
+    DataFrameHeaderInfo OnReadyToSendDataForStream(Http2StreamId stream_id,
+                                                   size_t max_length) override;
+    bool SendDataFrame(Http2StreamId stream_id, absl::string_view frame_header,
+                       size_t payload_bytes) override;
     void OnConnectionError(ConnectionError /*error*/) override {}
     bool OnFrameHeader(Http2StreamId stream_id, size_t length, uint8_t type,
                        uint8_t flags) override;
@@ -466,6 +470,8 @@ class ConnectionImpl : public virtual Connection,
   };
 
   // Encapsulates the logic for sending DATA frames on a given stream.
+  // Deprecated. Remove when removing
+  // `envoy_reloadable_features_http2_use_visitor_for_data`.
   class StreamDataFrameSource : public http2::adapter::DataFrameSource {
   public:
     explicit StreamDataFrameSource(StreamImpl& stream) : stream_(stream) {}

source/common/runtime/runtime_features.cc

@@ -59,6 +59,7 @@ RUNTIME_GUARD(envoy_reloadable_features_http2_decode_metadata_with_quiche);
 RUNTIME_GUARD(envoy_reloadable_features_http2_discard_host_header);
 // Ignore the automated "remove this flag" issue: we should keep this for 1 year.
 RUNTIME_GUARD(envoy_reloadable_features_http2_use_oghttp2);
+RUNTIME_GUARD(envoy_reloadable_features_http2_use_visitor_for_data);
 RUNTIME_GUARD(envoy_reloadable_features_http2_validate_authority_with_quiche);
 RUNTIME_GUARD(envoy_reloadable_features_http_allow_partial_urls_in_referer);
 RUNTIME_GUARD(envoy_reloadable_features_http_filter_avoid_reentrant_local_reply);

test/common/http/http2/codec_impl_test.cc

@@ -597,6 +597,51 @@ TEST_P(Http2CodecImplTest, SimpleRequestResponse) {
   }
 }
 
+TEST_P(Http2CodecImplTest, SimpleRequestResponseOldApi) {
+  scoped_runtime_.mergeValues({{"envoy.reloadable_features.http2_use_visitor_for_data", "false"}});
+  initialize();
+
+  InSequence s;
+  TestRequestHeaderMapImpl request_headers;
+  HttpTestUtility::addDefaultHeaders(request_headers);
+  request_headers.setMethod("POST");
+
+  // Encode request headers.
+  EXPECT_CALL(request_decoder_, decodeHeaders_(_, false));
+  EXPECT_TRUE(request_encoder_->encodeHeaders(request_headers, false).ok());
+
+  // Queue request body.
+  Buffer::OwnedImpl request_body(std::string(1024, 'a'));
+  request_encoder_->encodeData(request_body, true);
+
+  // Flush request body.
+  EXPECT_CALL(request_decoder_, decodeData(_, true)).Times(AtLeast(1));
+  driveToCompletion();
+
+  TestResponseHeaderMapImpl response_headers{{":status", "200"}};
+
+  // Encode response headers.
+  EXPECT_CALL(response_decoder_, decodeHeaders_(_, false));
+  response_encoder_->encodeHeaders(response_headers, false);
+
+  // Queue response body.
+  Buffer::OwnedImpl response_body(std::string(1024, 'b'));
+  response_encoder_->encodeData(response_body, true);
+
+  // Flush response body.
+  EXPECT_CALL(response_decoder_, decodeData(_, true)).Times(AtLeast(1));
+  driveToCompletion();
+
+  EXPECT_TRUE(client_wrapper_->status_.ok());
+  EXPECT_TRUE(server_wrapper_->status_.ok());
+
+  if (http2_implementation_ == Http2Impl::Nghttp2) {
+    // Regression test for issue #19761.
+    EXPECT_EQ(0, getClientDataSourcesSize());
+    EXPECT_EQ(0, getServerDataSourcesSize());
+  }
+}
+
 TEST_P(Http2CodecImplTest, ShutdownNotice) {
   initialize();
   EXPECT_EQ(absl::nullopt, request_encoder_->http1StreamEncoderOptions());
@@ -3477,6 +3522,25 @@ TEST_P(Http2CodecImplTest, WindowUpdateFloodOverride) {
   EXPECT_NO_THROW(driveToCompletion());
 }
 
+TEST_P(Http2CodecImplTest, DataFrameWithPadding) {
+  initialize();
+  TestRequestHeaderMapImpl request_headers;
+  HttpTestUtility::addDefaultHeaders(request_headers);
+  request_headers.setMethod("POST");
+  EXPECT_CALL(request_decoder_, decodeHeaders_(_, false));
+  EXPECT_TRUE(request_encoder_->encodeHeaders(request_headers, false).ok());
+  driveToCompletion();
+  Http2Frame dataFrame = Http2Frame::makeDataFrameWithPadding(Http2Frame::makeClientStreamId(0),
+                                                              "some data with padding", 193);
+  Buffer::OwnedImpl data;
+  data.add(dataFrame.data(), dataFrame.size());
+  server_wrapper_->buffer_.add(std::move(data));
+  EXPECT_CALL(request_decoder_, decodeData(_, false));
+  driveToCompletion();
+  const Http::Status& status = server_wrapper_->status_;
+  EXPECT_TRUE(status.ok());
+}
+
 TEST_P(Http2CodecImplTest, EmptyDataFlood) {
   expect_buffered_data_on_teardown_ = true;
   Buffer::OwnedImpl data;

test/common/http/http2/http2_frame.cc

@@ -423,6 +423,19 @@ Http2Frame Http2Frame::makeDataFrame(uint32_t stream_index, absl::string_view da
   return frame;
 }
 
+Http2Frame Http2Frame::makeDataFrameWithPadding(uint32_t stream_index, absl::string_view data,
+                                                uint8_t padding_size) {
+  ASSERT(padding_size > 0);
+  Http2Frame frame;
+  frame.buildHeader(Type::Data, 0, 0x08, makeNetworkOrderStreamId(stream_index));
+  frame.appendData({static_cast<uint8_t>(padding_size - 1u)});
+  frame.appendData(data);
+  std::vector<uint8_t> padding(padding_size - 1);
+  frame.appendData(padding);
+  frame.adjustPayloadSize();
+  return frame;
+}
+
 } // namespace Http2
 } // namespace Http
 } // namespace Envoy

test/common/http/http2/http2_frame.h

@@ -184,6 +184,8 @@ class Http2Frame {
                                     const std::vector<Header> extra_headers);
   static Http2Frame makeDataFrame(uint32_t stream_index, absl::string_view data,
                                   DataFlags flags = DataFlags::None);
+  static Http2Frame makeDataFrameWithPadding(uint32_t stream_index, absl::string_view data,
+                                             uint8_t padding_size);
 
   /**
    * Creates a frame with the given contents. This frame can be

tools/spelling/spelling_dictionary.txt

@@ -258,6 +258,7 @@ KiB
 Kille
 LBs
 LC
+LCOV
 LDS
 LEDS
 LEV