diff --git a/source/common/http/conn_manager_impl.cc b/source/common/http/conn_manager_impl.cc index 86e6bbfe7f27..92f7a5cdd946 100644 --- a/source/common/http/conn_manager_impl.cc +++ b/source/common/http/conn_manager_impl.cc @@ -296,53 +296,40 @@ Network::FilterStatus ConnectionManagerImpl::onData(Buffer::Instance& data, bool } } - // TODO(alyssawilk) clean this up after #8352 is well vetted. - bool redispatch; - do { - redispatch = false; - - try { - codec_->dispatch(data); - } catch (const FrameFloodException& e) { - // TODO(mattklein123): This is an emergency substitute for the lack of connection level - // logging in the HCM. In a public follow up change we will add full support for connection - // level logging in the HCM, similar to what we have in tcp_proxy. This will allow abuse - // indicators to be stored in the connection level stream info, and then matched, sampled, - // etc. when logged. - const envoy::type::FractionalPercent default_value; // 0 - if (runtime_.snapshot().featureEnabled("http.connection_manager.log_flood_exception", - default_value)) { - ENVOY_CONN_LOG(warn, "downstream HTTP flood from IP '{}': {}", - read_callbacks_->connection(), - read_callbacks_->connection().remoteAddress()->asString(), e.what()); - } - - handleCodecException(e.what()); - return Network::FilterStatus::StopIteration; - } catch (const CodecProtocolException& e) { - stats_.named_.downstream_cx_protocol_error_.inc(); - handleCodecException(e.what()); - return Network::FilterStatus::StopIteration; + try { + codec_->dispatch(data); + } catch (const FrameFloodException& e) { + // TODO(mattklein123): This is an emergency substitute for the lack of connection level + // logging in the HCM. In a public follow up change we will add full support for connection + // level logging in the HCM, similar to what we have in tcp_proxy. This will allow abuse + // indicators to be stored in the connection level stream info, and then matched, sampled, + // etc. when logged. + const envoy::type::FractionalPercent default_value; // 0 + if (runtime_.snapshot().featureEnabled("http.connection_manager.log_flood_exception", + default_value)) { + ENVOY_CONN_LOG(warn, "downstream HTTP flood from IP '{}': {}", read_callbacks_->connection(), + read_callbacks_->connection().remoteAddress()->asString(), e.what()); } - // Processing incoming data may release outbound data so check for closure here as well. - checkForDeferredClose(); - - // The HTTP/1 codec will pause dispatch after a single message is complete. We want to - // either redispatch if there are no streams and we have more data. If we have a single - // complete non-WebSocket stream but have not responded yet we will pause socket reads - // to apply back pressure. - if (codec_->protocol() < Protocol::Http2) { - if (read_callbacks_->connection().state() == Network::Connection::State::Open && - data.length() > 0 && streams_.empty()) { - redispatch = true; - } + handleCodecException(e.what()); + return Network::FilterStatus::StopIteration; + } catch (const CodecProtocolException& e) { + stats_.named_.downstream_cx_protocol_error_.inc(); + handleCodecException(e.what()); + return Network::FilterStatus::StopIteration; + } - if (!streams_.empty() && streams_.front()->state_.remote_complete_) { - read_callbacks_->connection().readDisable(true); - } + // Processing incoming data may release outbound data so check for closure here as well. + checkForDeferredClose(); + + // The HTTP/1 codec will pause parsing after a single message is complete. If we have a single + // complete non-WebSocket stream but have not responded yet we will pause socket reads + // to apply back pressure. + if (codec_->protocol() < Protocol::Http2) { + if (!streams_.empty() && streams_.front()->state_.remote_complete_) { + read_callbacks_->connection().readDisable(true); } - } while (redispatch); + } return Network::FilterStatus::StopIteration; } diff --git a/test/common/http/conn_manager_impl_test.cc b/test/common/http/conn_manager_impl_test.cc index 3dcc014b9f79..734769109097 100644 --- a/test/common/http/conn_manager_impl_test.cc +++ b/test/common/http/conn_manager_impl_test.cc @@ -442,6 +442,10 @@ TEST_F(HttpConnectionManagerImplTest, HeaderOnlyRequestAndResponse) { // Kick off the incoming data. Use extra data which should cause a redispatch. Buffer::OwnedImpl fake_input("1234"); conn_manager_->onData(fake_input, false); + // If this were actual HTTP input, the codec would kick off a fake event after + // the first request was finished, triggering the second onData. We do this + // manually. + conn_manager_->onData(fake_input, false); EXPECT_EQ(1U, stats_.named_.downstream_rq_2xx_.value()); EXPECT_EQ(1U, listener_stats_.downstream_rq_2xx_.value()); diff --git a/test/integration/integration_test.cc b/test/integration/integration_test.cc index 78134daffed4..e891587b87b2 100644 --- a/test/integration/integration_test.cc +++ b/test/integration/integration_test.cc @@ -490,6 +490,35 @@ TEST_P(IntegrationTest, Pipeline) { connection.close(); } +// Add a pipeline test where complete request headers in the first request merit +// an inline sendLocalReply to make sure the "kick" works under the call stack +// of dispatch as well as when a response is proxied from upstream. +TEST_P(IntegrationTest, PipelineInline) { + autonomous_upstream_ = true; + initialize(); + std::string response; + + Buffer::OwnedImpl buffer("GET / HTTP/1.1\r\n\r\nGET / HTTP/1.1\r\n\r\n"); + RawConnectionDriver connection( + lookupPort("http"), buffer, + [&](Network::ClientConnection&, const Buffer::Instance& data) -> void { + response.append(data.toString()); + }, + version_); + // First is an error: no host. + while (response.find("400") == std::string::npos) { + connection.run(Event::Dispatcher::RunType::NonBlock); + } + EXPECT_THAT(response, HasSubstr("HTTP/1.1 400 Bad Request\r\n")); + + // Second response should be 400 (no host) + while (response.find("400") == std::string::npos) { + connection.run(Event::Dispatcher::RunType::NonBlock); + } + EXPECT_THAT(response, HasSubstr("HTTP/1.1 400 Bad Request\r\n")); + connection.close(); +} + TEST_P(IntegrationTest, NoHost) { initialize(); codec_client_ = makeHttpConnection(lookupPort("http"));