QUIC deferred logging: add retransmission rate

changelogs/current.yaml

@@ -160,6 +160,9 @@ new_features:
 - area: access_log
   change: |
     added :ref:`CEL <envoy_v3_api_msg_extensions.formatter.cel.v3.Cel>` access log formatter to print CEL expression.
+- area: access_log
+  change: |
+    (QUIC only) Added support for %BYTES_RETRANSMITTED% and %PACKETS_RETRANSMITTED%.
 - area: dynamic_forward_proxy
   change: |
     added :ref:`sub_clusters_config

envoy/stream_info/stream_info.h

@@ -580,6 +580,26 @@ class StreamInfo {
    */
   virtual uint64_t bytesReceived() const PURE;
 
+  /**
+   * @param bytes_retransmitted denotes number of bytes to add to total retransmitted bytes.
+   */
+  virtual void addBytesRetransmitted(uint64_t bytes_retransmitted) PURE;
+
+  /**
+   * @return the number of body bytes retransmitted by the stream.
+   */
+  virtual uint64_t bytesRetransmitted() const PURE;
+
+  /**
+   * @param packets_retransmitted denotes number of packets to add to total retransmitted packets.
+   */
+  virtual void addPacketsRetransmitted(uint64_t packets_retransmitted) PURE;
+
+  /**
+   * @return the number of packets retransmitted by the stream.
+   */
+  virtual uint64_t packetsRetransmitted() const PURE;
+
   /**
    * @return the protocol of the request.
    */

source/common/formatter/substitution_formatter.cc

@@ -969,6 +969,22 @@ const StreamInfoFormatter::FieldExtractorLookupTbl& StreamInfoFormatter::getKnow
                                     return stream_info.bytesReceived();
                                   });
                             }}},
+                          {"BYTES_RETRANSMITTED",
+                           {CommandSyntaxChecker::COMMAND_ONLY,
+                            [](const std::string&, const absl::optional<size_t>&) {
+                              return std::make_unique<StreamInfoUInt64FieldExtractor>(
+                                  [](const StreamInfo::StreamInfo& stream_info) {
+                                    return stream_info.bytesRetransmitted();
+                                  });
+                            }}},
+                          {"PACKETS_RETRANSMITTED",
+                           {CommandSyntaxChecker::COMMAND_ONLY,
+                            [](const std::string&, const absl::optional<size_t>&) {
+                              return std::make_unique<StreamInfoUInt64FieldExtractor>(
+                                  [](const StreamInfo::StreamInfo& stream_info) {
+                                    return stream_info.packetsRetransmitted();
+                                  });
+                            }}},
                           {"UPSTREAM_WIRE_BYTES_RECEIVED",
                            {CommandSyntaxChecker::COMMAND_ONLY,
                             [](const std::string&, const absl::optional<size_t>&) {

source/common/quic/quic_stats_gatherer.cc

@@ -13,6 +13,11 @@ void QuicStatsGatherer::OnPacketAcked(int acked_bytes,
   }
 }
 
+void QuicStatsGatherer::OnPacketRetransmitted(int retransmitted_bytes) {
+  retransmitted_packets_++;
+  retransmitted_bytes_ += retransmitted_bytes;
+}
+
 void QuicStatsGatherer::maybeDoDeferredLog(bool record_ack_timing) {
   logging_done_ = true;
   if (stream_info_ == nullptr) {
@@ -21,6 +26,8 @@ void QuicStatsGatherer::maybeDoDeferredLog(bool record_ack_timing) {
   if (time_source_ != nullptr && record_ack_timing) {
     stream_info_->downstreamTiming().onLastDownstreamAckReceived(*time_source_);
   }
+  stream_info_->addBytesRetransmitted(retransmitted_bytes_);
+  stream_info_->addPacketsRetransmitted(retransmitted_packets_);
   const Http::RequestHeaderMap* request_headers = request_header_map_.get();
   const Http::ResponseHeaderMap* response_headers = response_header_map_.get();
   const Http::ResponseTrailerMap* response_trailers = response_trailer_map_.get();

source/common/quic/quic_stats_gatherer.h

@@ -20,7 +20,7 @@ class QuicStatsGatherer : public quic::QuicAckListenerInterface {
 
   // QuicAckListenerInterface
   void OnPacketAcked(int acked_bytes, quic::QuicTime::Delta delta_largest_observed) override;
-  void OnPacketRetransmitted(int /* retransmitted_bytes */) override {}
+  void OnPacketRetransmitted(int retransmitted_bytes) override;
 
   // Add bytes sent for this stream, for internal tracking of bytes acked.
   void addBytesSent(uint64_t bytes_sent, bool end_stream) {
@@ -58,6 +58,8 @@ class QuicStatsGatherer : public quic::QuicAckListenerInterface {
   std::unique_ptr<StreamInfo::StreamInfo> stream_info_;
   Envoy::TimeSource* time_source_ = nullptr;
   bool logging_done_ = false;
+  uint64_t retransmitted_packets_ = 0;
+  uint64_t retransmitted_bytes_ = 0;
 };
 
 } // namespace Quic

source/common/stream_info/stream_info_impl.h

@@ -185,6 +185,18 @@ struct StreamInfoImpl : public StreamInfo {
 
   uint64_t bytesReceived() const override { return bytes_received_; }
 
+  void addBytesRetransmitted(uint64_t bytes_retransmitted) override {
+    bytes_retransmitted_ += bytes_retransmitted;
+  }
+
+  uint64_t bytesRetransmitted() const override { return bytes_retransmitted_; }
+
+  void addPacketsRetransmitted(uint64_t packets_retransmitted) override {
+    packets_retransmitted_ += packets_retransmitted;
+  }
+
+  uint64_t packetsRetransmitted() const override { return packets_retransmitted_; }
+
   absl::optional<Http::Protocol> protocol() const override { return protocol_; }
 
   void protocol(Http::Protocol protocol) override { protocol_ = protocol; }
@@ -344,6 +356,8 @@ struct StreamInfoImpl : public StreamInfo {
     start_time_ = info.startTime();
     start_time_monotonic_ = info.startTimeMonotonic();
     downstream_transport_failure_reason_ = std::string(info.downstreamTransportFailureReason());
+    bytes_retransmitted_ = info.bytesRetransmitted();
+    packets_retransmitted_ = info.packetsRetransmitted();
   }
 
   // This function is used to copy over every field exposed in the StreamInfo interface, with a
@@ -435,6 +449,8 @@ struct StreamInfoImpl : public StreamInfo {
 
   std::shared_ptr<UpstreamInfo> upstream_info_;
   uint64_t bytes_received_{};
+  uint64_t bytes_retransmitted_{};
+  uint64_t packets_retransmitted_{};
   uint64_t bytes_sent_{};
   const Network::ConnectionInfoProviderSharedPtr downstream_connection_info_provider_;
   const Http::RequestHeaderMap* request_headers_{};

test/common/formatter/substitution_formatter_test.cc

@@ -402,6 +402,30 @@ TEST(SubstitutionFormatterTest, streamInfoFormatter) {
                 ProtoEq(ValueUtil::numberValue(25.0)));
   }
 
+  {
+    StreamInfoFormatter bytes_retransmitted_format("BYTES_RETRANSMITTED");
+    EXPECT_CALL(stream_info, bytesRetransmitted()).WillRepeatedly(Return(1));
+    EXPECT_EQ("1", bytes_retransmitted_format.format(request_headers, response_headers,
+                                                     response_trailers, stream_info, body,
+                                                     AccessLog::AccessLogType::NotSet));
+    EXPECT_THAT(bytes_retransmitted_format.formatValue(request_headers, response_headers,
+                                                       response_trailers, stream_info, body,
+                                                       AccessLog::AccessLogType::NotSet),
+                ProtoEq(ValueUtil::numberValue(1.0)));
+  }
+
+  {
+    StreamInfoFormatter packets_retransmitted_format("PACKETS_RETRANSMITTED");
+    EXPECT_CALL(stream_info, packetsRetransmitted()).WillRepeatedly(Return(1));
+    EXPECT_EQ("1", packets_retransmitted_format.format(request_headers, response_headers,
+                                                       response_trailers, stream_info, body,
+                                                       AccessLog::AccessLogType::NotSet));
+    EXPECT_THAT(packets_retransmitted_format.formatValue(request_headers, response_headers,
+                                                         response_trailers, stream_info, body,
+                                                         AccessLog::AccessLogType::NotSet),
+                ProtoEq(ValueUtil::numberValue(1.0)));
+  }
+
   {
     StreamInfoFormatter bytes_received_format("BYTES_RECEIVED");
     EXPECT_CALL(stream_info, bytesReceived()).WillRepeatedly(Return(1));

test/common/stream_info/stream_info_impl_test.cc

@@ -37,8 +37,8 @@ std::chrono::nanoseconds checkDuration(std::chrono::nanoseconds last,
 class StreamInfoImplTest : public testing::Test {
 protected:
   void assertStreamInfoSize(StreamInfoImpl stream_info) {
-    ASSERT_TRUE(sizeof(stream_info) == 824 || sizeof(stream_info) == 840 ||
-                sizeof(stream_info) == 872)
+    ASSERT_TRUE(sizeof(stream_info) == 840 || sizeof(stream_info) == 856 ||
+                sizeof(stream_info) == 888)
         << "If adding fields to StreamInfoImpl, please check to see if you "
            "need to add them to setFromForRecreateStream or setFrom! Current size "
         << sizeof(stream_info);
@@ -245,6 +245,8 @@ TEST_F(StreamInfoImplTest, SetFromForRecreateStream) {
 
   s1.addBytesReceived(1);
   s1.downstreamTiming().onLastDownstreamRxByteReceived(test_time_.timeSystem());
+  s1.addBytesRetransmitted(1);
+  s1.addPacketsRetransmitted(1);
 
 #ifdef __clang__
 #if defined(__linux__)
@@ -264,6 +266,8 @@ TEST_F(StreamInfoImplTest, SetFromForRecreateStream) {
   EXPECT_EQ(s1.bytesReceived(), s2.bytesReceived());
   EXPECT_EQ(s1.getDownstreamBytesMeter(), s2.getDownstreamBytesMeter());
   EXPECT_EQ(s1.downstreamTransportFailureReason(), s2.downstreamTransportFailureReason());
+  EXPECT_EQ(s1.bytesRetransmitted(), s2.bytesRetransmitted());
+  EXPECT_EQ(s1.packetsRetransmitted(), s2.packetsRetransmitted());
 }
 
 TEST_F(StreamInfoImplTest, SetFrom) {
@@ -272,6 +276,8 @@ TEST_F(StreamInfoImplTest, SetFrom) {
   // setFromForRecreateStream
   s1.addBytesReceived(1);
   s1.downstreamTiming().onLastDownstreamRxByteReceived(test_time_.timeSystem());
+  s1.addBytesRetransmitted(1);
+  s1.addPacketsRetransmitted(1);
 
   // setFrom
   s1.setRouteName("foo");
@@ -322,6 +328,8 @@ TEST_F(StreamInfoImplTest, SetFrom) {
   EXPECT_EQ(s1.bytesReceived(), s2.bytesReceived());
   EXPECT_EQ(s1.getDownstreamBytesMeter(), s2.getDownstreamBytesMeter());
   EXPECT_EQ(s1.downstreamTransportFailureReason(), s2.downstreamTransportFailureReason());
+  EXPECT_EQ(s1.bytesRetransmitted(), s2.bytesRetransmitted());
+  EXPECT_EQ(s1.packetsRetransmitted(), s2.packetsRetransmitted());
 
   // Copied by setFrom
   EXPECT_EQ(s1.getRouteName(), s2.getRouteName());

test/integration/BUILD

@@ -2273,6 +2273,7 @@ envoy_cc_test(
         "//test/common/config:dummy_config_proto_cc_proto",
         "//test/extensions/transport_sockets/tls/cert_validator:timed_cert_validator",
         ":http_integration_lib",
+        ":socket_interface_swap_lib",
         "//source/common/quic:client_connection_factory_lib",
         "//source/common/quic:envoy_quic_client_connection_lib",
         "//source/common/quic:envoy_quic_client_session_lib",

test/integration/quic_http_integration_test.cc

@@ -28,6 +28,7 @@
 #include "test/config/utility.h"
 #include "test/extensions/transport_sockets/tls/cert_validator/timed_cert_validator.h"
 #include "test/integration/http_integration.h"
+#include "test/integration/socket_interface_swap.h"
 #include "test/integration/ssl_utility.h"
 #include "test/test_common/registry.h"
 #include "test/test_common/test_runtime.h"
@@ -1392,6 +1393,41 @@ TEST_P(QuicHttpIntegrationTest, DeferredLoggingWithInternalRedirect) {
   EXPECT_EQ(/* RESP(test-header) */ metrics.at(21), "-");
 }
 
+TEST_P(QuicHttpIntegrationTest, DeferredLoggingWithRetransmission) {
+  useAccessLog("%BYTES_RETRANSMITTED%,%PACKETS_RETRANSMITTED%");
+  initialize();
+
+  codec_client_ = makeHttpConnection(makeClientConnection(lookupPort("http")));
+  IntegrationStreamDecoderPtr response =
+      codec_client_->makeHeaderOnlyRequest(default_request_headers_);
+  waitForNextUpstreamRequest(0, TestUtility::DefaultTimeout);
+
+  // Temporarily prevent server from writing packets (i.e. to respond to downstream)
+  // to simulate packet loss and trigger retransmissions.
+  SocketInterfaceSwap socket_swap(upstreamProtocol() == Http::CodecType::HTTP3
+                                      ? Network::Socket::Type::Datagram
+                                      : Network::Socket::Type::Stream);
+  Network::IoSocketError* ebadf = Network::IoSocketError::getIoSocketEbadfInstance();
+  socket_swap.write_matcher_->setDestinationPort(lookupPort("http"));
+  socket_swap.write_matcher_->setWriteOverride(ebadf);
+  upstream_request_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "200"}}, true);
+  absl::SleepFor(absl::Milliseconds(500 * TSAN_TIMEOUT_FACTOR));
+  // Allow the response to be sent downstream again.
+  socket_swap.write_matcher_->setWriteOverride(nullptr);
+
+  ASSERT_TRUE(response->waitForEndStream());
+  codec_client_->close();
+  ASSERT_TRUE(response->complete());
+
+  // Confirm that retransmissions are logged.
+  std::string log = waitForAccessLog(access_log_name_);
+  std::vector<std::string> metrics = absl::StrSplit(log, ',');
+  ASSERT_EQ(metrics.size(), 2);
+  EXPECT_GT(/* BYTES_RETRANSMITTED */ std::stoi(metrics.at(0)), 0);
+  EXPECT_GT(/* PACKETS_RETRANSMITTED */ std::stoi(metrics.at(1)), 0);
+  EXPECT_GE(std::stoi(metrics.at(0)), std::stoi(metrics.at(1)));
+}
+
 TEST_P(QuicHttpIntegrationTest, InvalidTrailer) {
   initialize();
   // Empty string in trailer key is invalid.

test/mocks/stream_info/mocks.h

@@ -85,6 +85,10 @@ class MockStreamInfo : public StreamInfo {
   MOCK_METHOD(OptRef<const DownstreamTiming>, downstreamTiming, (), (const));
   MOCK_METHOD(void, addBytesReceived, (uint64_t));
   MOCK_METHOD(uint64_t, bytesReceived, (), (const));
+  MOCK_METHOD(void, addBytesRetransmitted, (uint64_t));
+  MOCK_METHOD(uint64_t, bytesRetransmitted, (), (const));
+  MOCK_METHOD(void, addPacketsRetransmitted, (uint64_t));
+  MOCK_METHOD(uint64_t, packetsRetransmitted, (), (const));
   MOCK_METHOD(void, addWireBytesReceived, (uint64_t));
   MOCK_METHOD(uint64_t, wireBytesReceived, (), (const));
   MOCK_METHOD(void, setRouteName, (absl::string_view route_name));

tools/spelling/spelling_dictionary.txt

@@ -1160,6 +1160,7 @@ responder
 restarter
 resync
 ret
+retransmissions
 retransmitted
 retransmitting
 retriable