quic: Populate counter downstream_cx_tx_bytes_total for HTTP/3. (envoyproxy#39005)

Commit Message: Populate counter downstream_cx_tx_bytes_total for
HTTP/3.
Additional Description:
Risk Level: Low
Testing: Added DownstreamProtocolIntegrationTest.DownstreamCxStats
Docs Changes:
Release Notes:
Platform Specific Features:
[Optional Runtime guard:]
[Optional Fixes #Issue]
[Optional Fixes commit #PR or SHA]
[Optional Deprecated:]
[Optional [API
Considerations](https://github.com/envoyproxy/envoy/blob/main/api/review_checklist.md):]

---------

Signed-off-by: Bin Wu <[email protected]>

Author: wu-bin

source/common/quic/BUILD

@@ -383,6 +383,7 @@ envoy_cc_library(
         "//source/common/network:listen_socket_lib",
         "//source/common/quic:envoy_quic_utils_lib",
         "@com_github_google_quiche//:quic_core_connection_lib",
+        "@com_github_google_quiche//:quic_core_packet_writer_lib",
         "@com_github_google_quiche//:quic_core_packets_lib",
     ],
 )

source/common/quic/envoy_quic_server_connection.cc

@@ -1,14 +1,32 @@
 #include "source/common/quic/envoy_quic_server_connection.h"
 
+#include <memory>
+
 #include "source/common/network/listen_socket_impl.h"
 #include "source/common/quic/envoy_quic_utils.h"
 #include "source/common/quic/quic_io_handle_wrapper.h"
 
+#include "quiche/quic/core/quic_packet_writer_wrapper.h"
 #include "quiche/quic/core/quic_packets.h"
 
 namespace Envoy {
 namespace Quic {
 
+namespace {
+std::unique_ptr<quic::QuicPacketWriterWrapper>
+wrapWriter(quic::QuicPacketWriter* writer, bool owns_writer,
+           quic::QuicPacketWriterWrapper::OnWriteDoneCallback on_write_done) {
+  auto wrapper = std::make_unique<quic::QuicPacketWriterWrapper>();
+  if (owns_writer) {
+    wrapper->set_writer(writer);
+  } else {
+    wrapper->set_non_owning_writer(writer);
+  }
+  wrapper->set_on_write_done(std::move(on_write_done));
+  return wrapper;
+}
+} // namespace
+
 EnvoyQuicServerConnection::EnvoyQuicServerConnection(
     const quic::QuicConnectionId& server_connection_id,
     quic::QuicSocketAddress initial_self_address, quic::QuicSocketAddress initial_peer_address,
@@ -17,9 +35,15 @@ EnvoyQuicServerConnection::EnvoyQuicServerConnection(
     const quic::ParsedQuicVersionVector& supported_versions,
     Network::ConnectionSocketPtr connection_socket, quic::ConnectionIdGeneratorInterface& generator,
     std::unique_ptr<QuicListenerFilterManagerImpl> listener_filter_manager)
-    : quic::QuicConnection(server_connection_id, initial_self_address, initial_peer_address,
-                           &helper, &alarm_factory, writer, owns_writer,
-                           quic::Perspective::IS_SERVER, supported_versions, generator),
+    : quic::QuicConnection(
+          server_connection_id, initial_self_address, initial_peer_address, &helper, &alarm_factory,
+          // Wrap the packet writer to get notified when a packet is written.
+          wrapWriter(writer, owns_writer,
+                     [this](size_t packet_size, const quic::WriteResult& result) {
+                       OnWritePacketDone(packet_size, result);
+                     })
+              .release(),
+          /*owns_writer=*/true, quic::Perspective::IS_SERVER, supported_versions, generator),
       QuicNetworkConnection(std::move(connection_socket)),
       listener_filter_manager_(std::move(listener_filter_manager)) {
 #ifndef WIN32
@@ -47,6 +71,13 @@ bool EnvoyQuicServerConnection::OnPacketHeader(const quic::QuicPacketHeader& hea
   return true;
 }
 
+void EnvoyQuicServerConnection::OnWritePacketDone(size_t packet_size,
+                                                  const quic::WriteResult& /*result*/) {
+  if (hasConnectionStats()) {
+    connectionStats().write_total_.add(packet_size);
+  }
+}
+
 void EnvoyQuicServerConnection::OnCanWrite() {
   quic::QuicConnection::OnCanWrite();
   onWriteEventDone();

source/common/quic/envoy_quic_server_connection.h

@@ -157,6 +157,8 @@ class EnvoyQuicServerConnection : public quic::QuicConnection, public QuicNetwor
   void OnEffectivePeerMigrationValidated(bool is_migration_linkable) override;
 
 private:
+  // Called when a packet is written to the packet writer.
+  void OnWritePacketDone(size_t packet_size, const quic::WriteResult& result);
   std::unique_ptr<QuicListenerFilterManagerImpl> listener_filter_manager_;
   bool first_packet_received_ = false;
 };

source/common/quic/quic_network_connection.h

@@ -43,8 +43,11 @@ class QuicNetworkConnection : protected Logger::Loggable<Logger::Id::connection>
   uint64_t id() const;
 
 protected:
+  // REQUIRES: hasConnectionStats() == true.
   Network::Connection::ConnectionStats& connectionStats() const { return *connection_stats_; }
 
+  bool hasConnectionStats() const { return connection_stats_ != nullptr; }
+
   void setConnectionSocket(Network::ConnectionSocketPtr&& connection_socket) {
     connection_sockets_.push_back(std::move(connection_socket));
   }

test/common/quic/envoy_quic_dispatcher_test.cc

@@ -33,6 +33,7 @@
 #include "quiche/quic/test_tools/quic_dispatcher_peer.h"
 #include "quiche/quic/test_tools/quic_test_utils.h"
 
+using testing::AnyNumber;
 using testing::Invoke;
 using testing::Return;
 using testing::ReturnRef;
@@ -211,6 +212,7 @@ class EnvoyQuicDispatcherTest : public testing::TestWithParam<Network::Address::
           // Stop iteration to avoid calling getRead/WriteBuffer().
           .WillOnce(Return(Network::FilterStatus::StopIteration));
       EXPECT_CALL(network_connection_callbacks_, onEvent(Network::ConnectionEvent::LocalClose));
+      EXPECT_CALL(write_total_, add(_)).Times(AnyNumber());
     }
 
   private:
@@ -219,7 +221,8 @@ class EnvoyQuicDispatcherTest : public testing::TestWithParam<Network::Address::
     Network::MockConnectionCallbacks network_connection_callbacks_;
     testing::StrictMock<Stats::MockCounter> read_total_;
     testing::StrictMock<Stats::MockGauge> read_current_;
-    testing::StrictMock<Stats::MockCounter> write_total_;
+    // Currently QUIC only populates write_total_.
+    testing::NiceMock<Stats::MockCounter> write_total_;
     testing::StrictMock<Stats::MockGauge> write_current_;
 
     Filter::NetworkFilterFactoriesList filter_factory_;
@@ -271,7 +274,8 @@ TEST_P(EnvoyQuicDispatcherTest, CloseConnectionDuringNetworkFilterInstallation)
   Network::MockConnectionCallbacks network_connection_callbacks;
   testing::StrictMock<Stats::MockCounter> read_total;
   testing::StrictMock<Stats::MockGauge> read_current;
-  testing::StrictMock<Stats::MockCounter> write_total;
+  testing::NiceMock<Stats::MockCounter> write_total;
+  EXPECT_CALL(write_total, add(_)).Times(AnyNumber());
   testing::StrictMock<Stats::MockGauge> write_current;
 
   Filter::NetworkFilterFactoriesList filter_factory;
@@ -433,7 +437,8 @@ TEST_P(EnvoyQuicDispatcherTest, CloseWithGivenFilterChain) {
   Network::MockConnectionCallbacks network_connection_callbacks;
   testing::StrictMock<Stats::MockCounter> read_total;
   testing::StrictMock<Stats::MockGauge> read_current;
-  testing::StrictMock<Stats::MockCounter> write_total;
+  testing::NiceMock<Stats::MockCounter> write_total;
+  EXPECT_CALL(write_total, add(_)).Times(AnyNumber());
   testing::StrictMock<Stats::MockGauge> write_current;
 
   Filter::NetworkFilterFactoriesList filter_factory;

test/common/quic/envoy_quic_server_session_test.cc

@@ -185,6 +185,7 @@ class EnvoyQuicServerSessionTest : public testing::Test {
           return quic::WriteResult{quic::WRITE_STATUS_OK, static_cast<int>(buf_len)};
         }));
     ON_CALL(crypto_stream_helper_, CanAcceptClientHello(_, _, _, _, _)).WillByDefault(Return(true));
+    EXPECT_CALL(write_total_, add(_)).Times(AnyNumber());
     EXPECT_CALL(*debug_visitor_factory_.mock_debug_visitor_, OnConnectionClosed(_, _));
   }
 
@@ -273,7 +274,8 @@ class EnvoyQuicServerSessionTest : public testing::Test {
   Http::MockServerConnectionCallbacks http_connection_callbacks_;
   testing::StrictMock<Stats::MockCounter> read_total_;
   testing::StrictMock<Stats::MockGauge> read_current_;
-  testing::StrictMock<Stats::MockCounter> write_total_;
+  // Currently QUIC only populates write_total_.
+  testing::NiceMock<Stats::MockCounter> write_total_;
   testing::StrictMock<Stats::MockGauge> write_current_;
   Http::ServerConnectionPtr http_connection_;
   Http::Http3::CodecStats stats_;

test/integration/protocol_integration_test.cc

@@ -5720,4 +5720,25 @@ TEST_P(DownstreamProtocolIntegrationTest, EmptyCookieHeader) {
   codec_client_->close();
 }
 
+TEST_P(DownstreamProtocolIntegrationTest, DownstreamCxStats) {
+  initialize();
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  auto response = codec_client_->makeRequestWithBody(default_request_headers_, 1024);
+  waitForNextUpstreamRequest();
+
+  upstream_request_->encodeHeaders(default_response_headers_, false);
+  upstream_request_->encodeData(512, true);
+  ASSERT_TRUE(response->waitForEndStream());
+
+  EXPECT_TRUE(upstream_request_->complete());
+  EXPECT_EQ(1024U, upstream_request_->bodyLength());
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().getStatusValue());
+  EXPECT_EQ(512U, response->body().size());
+
+  test_server_->waitForCounterGe("http.config_test.downstream_cx_tx_bytes_total", 512);
+}
+
 } // namespace Envoy