diff --git a/docs/root/version_history/current.rst b/docs/root/version_history/current.rst
index 0d67d7b220380..279b3db99f1ce 100644
--- a/docs/root/version_history/current.rst
+++ b/docs/root/version_history/current.rst
@@ -44,6 +44,8 @@ Bug Fixes
 *Changes expected to improve the state of the world and are unlikely to have negative effects*
 
 * aws_lambda: if `payload_passthrough` is set to ``false``, the downstream response content-type header will now be set from the content-type entry in the JSON response's headers map, if present.
+* cluster: fixed the :ref:`cluster stats <config_cluster_manager_cluster_stats_request_response_sizes>` histograms by moving the accounting into the router
+  filter. This means that we now properly compute the number of bytes sent as well as handling retries which were previously ignored.
 * hot_restart: fix double counting of `server.seconds_until_first_ocsp_response_expiring` and `server.days_until_first_cert_expiring` during hot-restart. This stat was only incorrect until the parent process terminated.
 * http: port stripping now works for CONNECT requests, though the port will be restored if the CONNECT request is sent upstream. This behavior can be temporarily reverted by setting ``envoy.reloadable_features.strip_port_from_connect`` to false.
 * http: raise max configurable max_request_headers_kb limit to 8192 KiB (8MiB) from 96 KiB in http connection manager.
diff --git a/source/common/http/conn_manager_impl.cc b/source/common/http/conn_manager_impl.cc
index 3f4a567f26e07..12c29a1e0cbda 100644
--- a/source/common/http/conn_manager_impl.cc
+++ b/source/common/http/conn_manager_impl.cc
@@ -694,19 +694,6 @@ ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connect
 
 void ConnectionManagerImpl::ActiveStream::completeRequest() {
   filter_manager_.streamInfo().onRequestComplete();
-  Upstream::HostDescriptionConstSharedPtr upstream_host =
-      connection_manager_.read_callbacks_->upstreamHost();
-
-  if (upstream_host != nullptr) {
-    Upstream::ClusterRequestResponseSizeStatsOptRef req_resp_stats =
-        upstream_host->cluster().requestResponseSizeStats();
-    if (req_resp_stats.has_value()) {
-      req_resp_stats->get().upstream_rq_body_size_.recordValue(
-          filter_manager_.streamInfo().bytesReceived());
-      req_resp_stats->get().upstream_rs_body_size_.recordValue(
-          filter_manager_.streamInfo().bytesSent());
-    }
-  }
 
   if (connection_manager_.remote_close_) {
     filter_manager_.streamInfo().setResponseCodeDetails(
@@ -794,17 +781,6 @@ void ConnectionManagerImpl::ActiveStream::chargeStats(const ResponseHeaderMap& h
     return;
   }
 
-  Upstream::HostDescriptionConstSharedPtr upstream_host =
-      connection_manager_.read_callbacks_->upstreamHost();
-
-  if (upstream_host != nullptr) {
-    Upstream::ClusterRequestResponseSizeStatsOptRef req_resp_stats =
-        upstream_host->cluster().requestResponseSizeStats();
-    if (req_resp_stats.has_value()) {
-      req_resp_stats->get().upstream_rs_headers_size_.recordValue(headers.byteSize());
-    }
-  }
-
   // No response is sent back downstream for internal redirects, so don't charge downstream stats.
   const absl::optional<std::string>& response_code_details =
       filter_manager_.streamInfo().responseCodeDetails();
@@ -866,17 +842,6 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(RequestHeaderMapPtr&& he
     request_header_timer_.reset();
   }
 
-  Upstream::HostDescriptionConstSharedPtr upstream_host =
-      connection_manager_.read_callbacks_->upstreamHost();
-
-  if (upstream_host != nullptr) {
-    Upstream::ClusterRequestResponseSizeStatsOptRef req_resp_stats =
-        upstream_host->cluster().requestResponseSizeStats();
-    if (req_resp_stats.has_value()) {
-      req_resp_stats->get().upstream_rq_headers_size_.recordValue(request_headers_->byteSize());
-    }
-  }
-
   // Both saw_connection_close_ and is_head_request_ affect local replies: set
   // them as early as possible.
   const Protocol protocol = connection_manager_.codec_->protocol();
diff --git a/source/common/router/upstream_request.cc b/source/common/router/upstream_request.cc
index 77b47ef443470..4a81e24d58354 100644
--- a/source/common/router/upstream_request.cc
+++ b/source/common/router/upstream_request.cc
@@ -92,6 +92,20 @@ UpstreamRequest::~UpstreamRequest() {
         FilterUtility::percentageOfTimeout(response_time, parent_.timeout().per_try_timeout_));
   }
 
+  // Ditto for request/response size histograms.
+  Upstream::ClusterRequestResponseSizeStatsOptRef req_resp_stats_opt =
+      parent_.cluster()->requestResponseSizeStats();
+  if (req_resp_stats_opt.has_value()) {
+    auto& req_resp_stats = req_resp_stats_opt->get();
+    req_resp_stats.upstream_rq_headers_size_.recordValue(parent_.downstreamHeaders()->byteSize());
+    req_resp_stats.upstream_rq_body_size_.recordValue(stream_info_.bytesSent());
+
+    if (response_headers_size_.has_value()) {
+      req_resp_stats.upstream_rs_headers_size_.recordValue(response_headers_size_.value());
+      req_resp_stats.upstream_rs_body_size_.recordValue(stream_info_.bytesReceived());
+    }
+  }
+
   stream_info_.setUpstreamTiming(upstream_timing_);
   stream_info_.onRequestComplete();
   for (const auto& upstream_log : parent_.config().upstream_logs_) {
@@ -110,12 +124,15 @@ void UpstreamRequest::decode100ContinueHeaders(Http::ResponseHeaderMapPtr&& head
   ScopeTrackerScopeState scope(&parent_.callbacks()->scope(), parent_.callbacks()->dispatcher());
 
   ASSERT(100 == Http::Utility::getResponseStatus(*headers));
+  addResponseHeadersSize(headers->byteSize());
   parent_.onUpstream100ContinueHeaders(std::move(headers), *this);
 }
 
 void UpstreamRequest::decodeHeaders(Http::ResponseHeaderMapPtr&& headers, bool end_stream) {
   ScopeTrackerScopeState scope(&parent_.callbacks()->scope(), parent_.callbacks()->dispatcher());
 
+  addResponseHeadersSize(headers->byteSize());
+
   // We drop 1xx other than 101 on the floor; 101 upgrade headers need to be passed to the client as
   // part of the final response. 100-continue headers are handled in onUpstream100ContinueHeaders.
   //
diff --git a/source/common/router/upstream_request.h b/source/common/router/upstream_request.h
index 33edf91bf0df2..8776089fe0434 100644
--- a/source/common/router/upstream_request.h
+++ b/source/common/router/upstream_request.h
@@ -127,6 +127,9 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
     return encode_complete_ && !buffered_request_body_ && !encode_trailers_ &&
            downstream_metadata_map_vector_.empty();
   }
+  void addResponseHeadersSize(uint64_t size) {
+    response_headers_size_ = response_headers_size_.value_or(0) + size;
+  }
 
   RouterFilterInterface& parent_;
   std::unique_ptr<GenericConnPool> conn_pool_;
@@ -141,6 +144,9 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
   StreamInfo::StreamInfoImpl stream_info_;
   StreamInfo::UpstreamTiming upstream_timing_;
   const MonotonicTime start_time_;
+  // This is wrapped in an optional, since we want to avoid computing zero size headers when in
+  // reality we just didn't get a response back.
+  absl::optional<uint64_t> response_headers_size_{};
   // Copies of upstream headers/trailers. These are only set if upstream
   // access logging is configured.
   Http::ResponseHeaderMapPtr upstream_headers_;
diff --git a/test/common/http/async_client_impl_test.cc b/test/common/http/async_client_impl_test.cc
index ea82b2993aad9..ea81164f1cc42 100644
--- a/test/common/http/async_client_impl_test.cc
+++ b/test/common/http/async_client_impl_test.cc
@@ -53,6 +53,7 @@ class AsyncClientImplTest : public testing::Test {
     ON_CALL(*cm_.thread_local_cluster_.conn_pool_.host_, locality())
         .WillByDefault(ReturnRef(envoy::config::core::v3::Locality().default_instance()));
     cm_.initializeThreadLocalClusters({"fake_cluster"});
+    HttpTestUtility::addDefaultHeaders(headers_);
   }
 
   virtual void expectSuccess(AsyncClient::Request* sent_request, uint64_t code) {
@@ -75,6 +76,7 @@ class AsyncClientImplTest : public testing::Test {
         }));
   }
 
+  TestRequestHeaderMapImpl headers_{};
   RequestMessagePtr message_{new RequestMessageImpl()};
   Stats::MockIsolatedStatsStore stats_store_;
   MockAsyncClientCallbacks callbacks_;
@@ -1488,10 +1490,8 @@ TEST_F(AsyncClientImplTest, MultipleDataStream) {
 }
 
 TEST_F(AsyncClientImplTest, WatermarkCallbacks) {
-  TestRequestHeaderMapImpl headers;
-  HttpTestUtility::addDefaultHeaders(headers);
   AsyncClient::Stream* stream = client_.start(stream_callbacks_, AsyncClient::StreamOptions());
-  stream->sendHeaders(headers, false);
+  stream->sendHeaders(headers_, false);
   Http::StreamDecoderFilterCallbacks* filter_callbacks =
       static_cast<Http::AsyncStreamImpl*>(stream);
   filter_callbacks->onDecoderFilterAboveWriteBufferHighWatermark();
@@ -1506,10 +1506,8 @@ TEST_F(AsyncClientImplTest, WatermarkCallbacks) {
 }
 
 TEST_F(AsyncClientImplTest, RdsGettersTest) {
-  TestRequestHeaderMapImpl headers;
-  HttpTestUtility::addDefaultHeaders(headers);
   AsyncClient::Stream* stream = client_.start(stream_callbacks_, AsyncClient::StreamOptions());
-  stream->sendHeaders(headers, false);
+  stream->sendHeaders(headers_, false);
   Http::StreamDecoderFilterCallbacks* filter_callbacks =
       static_cast<Http::AsyncStreamImpl*>(stream);
   auto route = filter_callbacks->route();
@@ -1522,7 +1520,7 @@ TEST_F(AsyncClientImplTest, RdsGettersTest) {
   const auto& route_config = route_entry->virtualHost().routeConfig();
   EXPECT_EQ("", route_config.name());
   EXPECT_EQ(0, route_config.internalOnlyHeaders().size());
-  EXPECT_EQ(nullptr, route_config.route(headers, stream_info_, 0));
+  EXPECT_EQ(nullptr, route_config.route(headers_, stream_info_, 0));
   auto cluster_info = filter_callbacks->clusterInfo();
   ASSERT_NE(nullptr, cluster_info);
   EXPECT_EQ(cm_.thread_local_cluster_.cluster_.info_, cluster_info);
@@ -1530,8 +1528,6 @@ TEST_F(AsyncClientImplTest, RdsGettersTest) {
 }
 
 TEST_F(AsyncClientImplTest, DumpState) {
-  TestRequestHeaderMapImpl headers;
-  HttpTestUtility::addDefaultHeaders(headers);
   AsyncClient::Stream* stream = client_.start(stream_callbacks_, AsyncClient::StreamOptions());
   Http::StreamDecoderFilterCallbacks* filter_callbacks =
       static_cast<Http::AsyncStreamImpl*>(stream);
diff --git a/test/common/http/conn_manager_impl_test_2.cc b/test/common/http/conn_manager_impl_test_2.cc
index 3c71e80667695..82e09cee54e91 100644
--- a/test/common/http/conn_manager_impl_test_2.cc
+++ b/test/common/http/conn_manager_impl_test_2.cc
@@ -2630,196 +2630,6 @@ TEST_F(HttpConnectionManagerImplTest, NewConnection) {
   EXPECT_EQ(1U, stats_.named_.downstream_cx_http3_active_.value());
 }
 
-TEST_F(HttpConnectionManagerImplTest, TestUpstreamRequestHeadersSize) {
-  // Test with Headers only request, No Data, No response.
-  setup(false, "");
-  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> Http::Status {
-    decoder_ = &conn_manager_->newStream(response_encoder_);
-    RequestHeaderMapPtr headers{
-        new TestRequestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
-    decoder_->decodeHeaders(std::move(headers), true);
-    return Http::okStatus();
-  }));
-
-  setupFilterChain(1, 0);
-
-  EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, true))
-      .WillOnce(Return(FilterHeadersStatus::StopIteration));
-  EXPECT_CALL(*decoder_filters_[0], decodeComplete());
-
-  std::shared_ptr<NiceMock<Upstream::MockHostDescription>> host_{
-      new NiceMock<Upstream::MockHostDescription>()};
-  filter_callbacks_.upstreamHost(host_);
-
-  EXPECT_CALL(
-      host_->cluster_.request_response_size_stats_store_,
-      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_headers_size"), 30));
-  EXPECT_CALL(host_->cluster_.request_response_size_stats_store_,
-              deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_body_size"), 0));
-  EXPECT_CALL(host_->cluster_.request_response_size_stats_store_,
-              deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_body_size"), 0));
-
-  Buffer::OwnedImpl fake_input("1234");
-  conn_manager_->onData(fake_input, false);
-
-  expectOnDestroy();
-  filter_callbacks_.connection_.raiseEvent(Network::ConnectionEvent::RemoteClose);
-}
-
-TEST_F(HttpConnectionManagerImplTest, TestUpstreamRequestBodySize) {
-  // Test Request with Headers and Data, No response.
-  setup(false, "");
-  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> Http::Status {
-    decoder_ = &conn_manager_->newStream(response_encoder_);
-    RequestHeaderMapPtr headers{
-        new TestRequestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
-    decoder_->decodeHeaders(std::move(headers), false);
-
-    Buffer::OwnedImpl fake_data("12345");
-    decoder_->decodeData(fake_data, true);
-    return Http::okStatus();
-  }));
-
-  setupFilterChain(1, 0);
-
-  EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, false))
-      .WillOnce(Return(FilterHeadersStatus::StopIteration));
-  EXPECT_CALL(*decoder_filters_[0], decodeData(_, true))
-      .WillOnce(Return(FilterDataStatus::StopIterationNoBuffer));
-
-  EXPECT_CALL(*decoder_filters_[0], decodeComplete());
-
-  std::shared_ptr<NiceMock<Upstream::MockHostDescription>> host_{
-      new NiceMock<Upstream::MockHostDescription>()};
-  filter_callbacks_.upstreamHost(host_);
-
-  EXPECT_CALL(
-      host_->cluster_.request_response_size_stats_store_,
-      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_headers_size"), 30));
-  EXPECT_CALL(host_->cluster_.request_response_size_stats_store_,
-              deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_body_size"), 5));
-  EXPECT_CALL(host_->cluster_.request_response_size_stats_store_,
-              deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_body_size"), 0));
-
-  Buffer::OwnedImpl fake_input("1234");
-  conn_manager_->onData(fake_input, false);
-
-  expectOnDestroy();
-  filter_callbacks_.connection_.raiseEvent(Network::ConnectionEvent::RemoteClose);
-}
-
-TEST_F(HttpConnectionManagerImplTest, TestUpstreamResponseHeadersSize) {
-  // Test with Header only response.
-  setup(false, "");
-
-  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> Http::Status {
-    decoder_ = &conn_manager_->newStream(response_encoder_);
-    RequestHeaderMapPtr headers{
-        new TestRequestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
-    decoder_->decodeHeaders(std::move(headers), false);
-
-    Buffer::OwnedImpl fake_data("1234");
-    decoder_->decodeData(fake_data, true);
-
-    return Http::okStatus();
-  }));
-
-  setupFilterChain(1, 0);
-
-  EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, false))
-      .WillOnce(Return(FilterHeadersStatus::StopIteration));
-  EXPECT_CALL(*decoder_filters_[0], decodeData(_, true))
-      .WillOnce(Return(FilterDataStatus::StopIterationNoBuffer));
-
-  EXPECT_CALL(*decoder_filters_[0], decodeComplete());
-
-  std::shared_ptr<NiceMock<Upstream::MockHostDescription>> host_{
-      new NiceMock<Upstream::MockHostDescription>()};
-  filter_callbacks_.upstreamHost(host_);
-
-  EXPECT_CALL(
-      host_->cluster_.request_response_size_stats_store_,
-      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_headers_size"), 30));
-
-  // Response headers are internally mutated and we record final response headers.
-  // for example in the below test case, response headers are modified as
-  // {':status', '200' 'date', 'Mon, 06 Jul 2020 06:08:55 GMT' 'server', ''}
-  // whose size is 49 instead of original response headers size 10({":status", "200"}).
-  EXPECT_CALL(
-      host_->cluster_.request_response_size_stats_store_,
-      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_headers_size"), 49));
-  EXPECT_CALL(host_->cluster_.request_response_size_stats_store_,
-              deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_body_size"), 4));
-  EXPECT_CALL(host_->cluster_.request_response_size_stats_store_,
-              deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_body_size"), 0));
-
-  Buffer::OwnedImpl fake_input("1234");
-  conn_manager_->onData(fake_input, false);
-
-  EXPECT_CALL(response_encoder_, encodeHeaders(_, true));
-  expectOnDestroy();
-
-  decoder_filters_[0]->callbacks_->streamInfo().setResponseCodeDetails("");
-  decoder_filters_[0]->callbacks_->encodeHeaders(
-      ResponseHeaderMapPtr{new TestResponseHeaderMapImpl{{":status", "200"}}}, true, "details");
-}
-
-TEST_F(HttpConnectionManagerImplTest, TestUpstreamResponseBodySize) {
-  // Test with response headers and body.
-  setup(false, "");
-
-  EXPECT_CALL(*codec_, dispatch(_)).WillOnce(Invoke([&](Buffer::Instance&) -> Http::Status {
-    decoder_ = &conn_manager_->newStream(response_encoder_);
-    RequestHeaderMapPtr headers{
-        new TestRequestHeaderMapImpl{{":authority", "host"}, {":path", "/"}, {":method", "GET"}}};
-    decoder_->decodeHeaders(std::move(headers), false);
-
-    Buffer::OwnedImpl fake_data("1234");
-    decoder_->decodeData(fake_data, true);
-
-    return Http::okStatus();
-  }));
-
-  setupFilterChain(1, 0);
-
-  EXPECT_CALL(*decoder_filters_[0], decodeHeaders(_, false))
-      .WillOnce(Return(FilterHeadersStatus::StopIteration));
-  EXPECT_CALL(*decoder_filters_[0], decodeData(_, true))
-      .WillOnce(Return(FilterDataStatus::StopIterationNoBuffer));
-
-  EXPECT_CALL(*decoder_filters_[0], decodeComplete());
-
-  std::shared_ptr<NiceMock<Upstream::MockHostDescription>> host_{
-      new NiceMock<Upstream::MockHostDescription>()};
-  filter_callbacks_.upstreamHost(host_);
-
-  EXPECT_CALL(
-      host_->cluster_.request_response_size_stats_store_,
-      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_headers_size"), 30));
-  EXPECT_CALL(
-      host_->cluster_.request_response_size_stats_store_,
-      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_headers_size"), 49));
-  EXPECT_CALL(host_->cluster_.request_response_size_stats_store_,
-              deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_body_size"), 4));
-  EXPECT_CALL(host_->cluster_.request_response_size_stats_store_,
-              deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_body_size"), 11));
-
-  Buffer::OwnedImpl fake_input("1234");
-  conn_manager_->onData(fake_input, false);
-
-  EXPECT_CALL(response_encoder_, encodeHeaders(_, false));
-
-  decoder_filters_[0]->callbacks_->streamInfo().setResponseCodeDetails("");
-  decoder_filters_[0]->callbacks_->encodeHeaders(
-      ResponseHeaderMapPtr{new TestResponseHeaderMapImpl{{":status", "200"}}}, false, "details");
-
-  EXPECT_CALL(response_encoder_, encodeData(_, true));
-  expectOnDestroy();
-
-  Buffer::OwnedImpl fake_response("hello-world");
-  decoder_filters_[0]->callbacks_->encodeData(fake_response, true);
-}
-
 TEST_F(HttpConnectionManagerImplTest, HeaderOnlyRequestAndResponseUsingHttp3) {
   setup(false, "envoy-custom-server", false);
 
diff --git a/test/common/router/router_test.cc b/test/common/router/router_test.cc
index 1b58a3dfdecac..ded9d4c300bd5 100644
--- a/test/common/router/router_test.cc
+++ b/test/common/router/router_test.cc
@@ -69,6 +69,78 @@ class RouterTest : public RouterTestBase {
   RouterTest() : RouterTestBase(false, false, Protobuf::RepeatedPtrField<std::string>{}) {
     EXPECT_CALL(callbacks_, activeSpan()).WillRepeatedly(ReturnRef(span_));
   };
+
+  void testRequestResponseSize(bool with_trailers) {
+    NiceMock<Http::MockRequestEncoder> encoder;
+    Http::ResponseDecoder* response_decoder = nullptr;
+
+    EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_, newStream(_, _))
+        .WillOnce(Invoke(
+            [&](Http::ResponseDecoder& decoder,
+                Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+              response_decoder = &decoder;
+              callbacks.onPoolReady(encoder, cm_.thread_local_cluster_.conn_pool_.host_,
+                                    upstream_stream_info_, Http::Protocol::Http10);
+              return nullptr;
+            }));
+
+    cm_.thread_local_cluster_.cluster_.info_->upstream_config_ =
+        absl::make_optional<envoy::config::core::v3::TypedExtensionConfig>();
+    envoy::extensions::upstreams::http::generic::v3::GenericConnectionPoolProto generic_config;
+    cm_.thread_local_cluster_.cluster_.info_->upstream_config_.value()
+        .mutable_typed_config()
+        ->PackFrom(generic_config);
+    callbacks_.route_->route_entry_.connect_config_ =
+        absl::make_optional<RouteEntry::ConnectConfig>();
+
+    EXPECT_CALL(cm_.thread_local_cluster_, httpConnPool(_, _, _));
+
+    Http::TestRequestHeaderMapImpl headers;
+    HttpTestUtility::addDefaultHeaders(headers);
+    headers.setMethod("POST");
+
+    EXPECT_CALL(
+        cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
+        deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_headers_size"), 74ull));
+    router_.decodeHeaders(headers, false);
+
+    EXPECT_CALL(callbacks_.dispatcher_, createTimer_);
+    EXPECT_CALL(
+        cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
+        deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_body_size"), 5ull));
+    Buffer::InstancePtr body_data(new Buffer::OwnedImpl("hello"));
+    EXPECT_EQ(Http::FilterDataStatus::StopIterationNoBuffer,
+              router_.decodeData(*body_data, !with_trailers));
+
+    if (with_trailers) {
+      Http::TestRequestTrailerMapImpl trailers{{"some", "trailer"}};
+      router_.decodeTrailers(trailers);
+    }
+
+    EXPECT_CALL(
+        cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
+        deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_headers_size"), 10ull));
+    Http::ResponseHeaderMapPtr response_headers(
+        new Http::TestResponseHeaderMapImpl{{":status", "200"}});
+    // NOLINTNEXTLINE: Silence null pointer access warning
+    response_decoder->decodeHeaders(std::move(response_headers), false);
+
+    EXPECT_CALL(
+        cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
+        deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_body_size"), 7ull));
+    Buffer::OwnedImpl response_data("goodbye");
+    // NOLINTNEXTLINE: Silence null pointer access warning
+    response_decoder->decodeData(response_data, !with_trailers);
+
+    if (with_trailers) {
+      Http::ResponseTrailerMapPtr response_trailers(
+          new Http::TestResponseTrailerMapImpl{{"some-trailer", "13"}});
+      // NOLINTNEXTLINE: Silence null pointer access warning
+      response_decoder->decodeTrailers(std::move(response_trailers));
+    }
+
+    router_.onDestroy();
+  }
 };
 
 TEST_F(RouterTest, UpdateServerNameFilterState) {
@@ -2148,6 +2220,24 @@ TEST_F(RouterTest, HedgedPerTryTimeoutResetsOnBadHeaders) {
 TEST_F(RouterTest, HedgedPerTryTimeoutThirdRequestSucceeds) {
   enableHedgeOnPerTryTimeout();
 
+  // Verify cluster request/response sizes are accounted for all requests/responses.
+  EXPECT_CALL(
+      cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
+      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_headers_size"), 73ull))
+      .Times(3);
+  EXPECT_CALL(
+      cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
+      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rq_body_size"), 0ull))
+      .Times(3);
+  EXPECT_CALL(
+      cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
+      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_headers_size"), 10ull))
+      .Times(2);
+  EXPECT_CALL(
+      cm_.thread_local_cluster_.cluster_.info_->request_response_size_stats_store_,
+      deliverHistogramToSinks(Property(&Stats::Metric::name, "upstream_rs_body_size"), 0ull))
+      .Times(2);
+
   NiceMock<Http::MockRequestEncoder> encoder1;
   Http::ResponseDecoder* response_decoder1 = nullptr;
   EXPECT_CALL(cm_.thread_local_cluster_.conn_pool_, newStream(_, _))
@@ -2183,6 +2273,7 @@ TEST_F(RouterTest, HedgedPerTryTimeoutThirdRequestSucceeds) {
   EXPECT_CALL(encoder1.stream_, resetStream(_)).Times(0);
   EXPECT_CALL(callbacks_, encodeHeaders_(_, _)).Times(0);
   router_.retry_state_->expectHeadersRetry();
+  // NOLINTNEXTLINE: Silence null pointer access warning
   response_decoder1->decodeHeaders(std::move(response_headers1), true);
 
   NiceMock<Http::MockRequestEncoder> encoder2;
@@ -2246,6 +2337,7 @@ TEST_F(RouterTest, HedgedPerTryTimeoutThirdRequestSucceeds) {
         EXPECT_TRUE(end_stream);
       }));
   EXPECT_CALL(*router_.retry_state_, shouldRetryHeaders(_, _)).WillOnce(Return(RetryStatus::No));
+  // NOLINTNEXTLINE: Silence null pointer access warning
   response_decoder3->decodeHeaders(std::move(response_headers2), true);
   EXPECT_TRUE(verifyHostUpstreamStats(1, 1));
 
@@ -5968,5 +6060,13 @@ TEST_F(RouterTest, PostHttpUpstream) {
 
   router_.onDestroy();
 }
+
+// Test that request/response header/body sizes are properly recorded.
+TEST_F(RouterTest, RequestResponseSize) { testRequestResponseSize(false); }
+
+// Test that request/response header/body sizes are properly recorded
+// when there are trailers in both the request/response.
+TEST_F(RouterTest, RequestResponseSizeWithTrailers) { testRequestResponseSize(true); }
+
 } // namespace Router
 } // namespace Envoy
diff --git a/test/common/router/upstream_request_test.cc b/test/common/router/upstream_request_test.cc
index 4b38122ed075f..312d5f0c99d69 100644
--- a/test/common/router/upstream_request_test.cc
+++ b/test/common/router/upstream_request_test.cc
@@ -19,6 +19,13 @@ namespace {
 
 class UpstreamRequestTest : public testing::Test {
 public:
+  UpstreamRequestTest() {
+    HttpTestUtility::addDefaultHeaders(downstream_request_header_map_);
+    ON_CALL(router_filter_interface_, downstreamHeaders())
+        .WillByDefault(Return(&downstream_request_header_map_));
+  }
+
+  Http::TestRequestHeaderMapImpl downstream_request_header_map_{};
   NiceMock<MockRouterFilterInterface> router_filter_interface_;
   UpstreamRequest upstream_request_{router_filter_interface_,
                                     std::make_unique<NiceMock<Router::MockGenericConnPool>>()};
@@ -57,11 +64,6 @@ TEST_F(UpstreamRequestTest, DumpsStateWithoutAllocatingMemory) {
   address_provider->setLocalAddress(Network::Utility::parseInternetAddressAndPort("5.6.7.8:5678"));
   address_provider->setDirectRemoteAddressForTest(
       Network::Utility::parseInternetAddressAndPort("1.2.3.4:5678"));
-  Http::TestRequestHeaderMapImpl downstream_request_header_map;
-  HttpTestUtility::addDefaultHeaders(downstream_request_header_map);
-
-  EXPECT_CALL(router_filter_interface_, downstreamHeaders())
-      .WillOnce(Return(&downstream_request_header_map));
 
   // Dump State
   std::array<char, 1024> buffer;
diff --git a/test/extensions/upstreams/http/tcp/upstream_request_test.cc b/test/extensions/upstreams/http/tcp/upstream_request_test.cc
index fad02b44fe118..1c18cb0c0770e 100644
--- a/test/extensions/upstreams/http/tcp/upstream_request_test.cc
+++ b/test/extensions/upstreams/http/tcp/upstream_request_test.cc
@@ -93,6 +93,10 @@ TEST_F(TcpConnPoolTest, Cancel) {
 class TcpUpstreamTest : public ::testing::Test {
 public:
   TcpUpstreamTest() {
+    EXPECT_CALL(mock_router_filter_, downstreamHeaders())
+        .Times(AnyNumber())
+        .WillRepeatedly(Return(&request_));
+    EXPECT_CALL(mock_router_filter_, cluster()).Times(AnyNumber());
     mock_router_filter_.requests_.push_back(std::make_unique<UpstreamRequest>(
         mock_router_filter_, std::make_unique<NiceMock<Router::MockGenericConnPool>>()));
     auto data = std::make_unique<NiceMock<Envoy::Tcp::ConnectionPool::MockConnectionData>>();
@@ -103,15 +107,15 @@ class TcpUpstreamTest : public ::testing::Test {
   ~TcpUpstreamTest() override { EXPECT_CALL(mock_router_filter_, config()).Times(AnyNumber()); }
 
 protected:
-  NiceMock<Network::MockClientConnection> connection_;
-  NiceMock<Router::MockRouterFilterInterface> mock_router_filter_;
-  Envoy::Tcp::ConnectionPool::MockConnectionData* mock_connection_data_;
-  std::unique_ptr<TcpUpstream> tcp_upstream_;
   TestRequestHeaderMapImpl request_{{":method", "CONNECT"},
                                     {":path", "/"},
                                     {":protocol", "bytestream"},
                                     {":scheme", "https"},
                                     {":authority", "host"}};
+  NiceMock<Network::MockClientConnection> connection_;
+  NiceMock<Router::MockRouterFilterInterface> mock_router_filter_;
+  Envoy::Tcp::ConnectionPool::MockConnectionData* mock_connection_data_;
+  std::unique_ptr<TcpUpstream> tcp_upstream_;
 };
 
 TEST_F(TcpUpstreamTest, Basic) {
diff --git a/test/integration/protocol_integration_test.cc b/test/integration/protocol_integration_test.cc
index ca7985b6d567d..441822786a2e4 100644
--- a/test/integration/protocol_integration_test.cc
+++ b/test/integration/protocol_integration_test.cc
@@ -703,6 +703,11 @@ TEST_P(ProtocolIntegrationTest, LongHeaderValueWithSpaces) {
 }
 
 TEST_P(ProtocolIntegrationTest, Retry) {
+  config_helper_.addConfigModifier([&](envoy::config::bootstrap::v3::Bootstrap& bootstrap) -> void {
+    RELEASE_ASSERT(bootstrap.mutable_static_resources()->clusters_size() == 1, "");
+    auto& cluster = *bootstrap.mutable_static_resources()->mutable_clusters(0);
+    cluster.mutable_track_cluster_stats()->set_request_response_sizes(true);
+  });
   initialize();
   codec_client_ = makeHttpConnection(lookupPort("http"));
   auto response = codec_client_->makeRequestWithBody(
@@ -744,6 +749,22 @@ TEST_P(ProtocolIntegrationTest, Retry) {
   EXPECT_NE(nullptr,
             test_server_->counter(absl::StrCat("cluster.cluster_0.", upstreamProtocolStatsRoot(),
                                                ".dropped_headers_with_underscores")));
+
+  test_server_->waitUntilHistogramHasSamples("cluster.cluster_0.upstream_rq_headers_size");
+  test_server_->waitUntilHistogramHasSamples("cluster.cluster_0.upstream_rs_headers_size");
+
+  auto find_histo_sample_count = [&](const std::string& name) -> uint64_t {
+    for (auto& histogram : test_server_->histograms()) {
+      if (histogram->name() == name) {
+        const auto& stats = histogram->cumulativeStatistics();
+        return stats.sampleCount();
+      }
+    }
+    return 0;
+  };
+
+  EXPECT_EQ(find_histo_sample_count("cluster.cluster_0.upstream_rq_headers_size"), 2);
+  EXPECT_EQ(find_histo_sample_count("cluster.cluster_0.upstream_rs_headers_size"), 2);
 }
 
 TEST_P(ProtocolIntegrationTest, RetryStreaming) {
@@ -2645,4 +2666,26 @@ TEST_P(ProtocolIntegrationTest, RemoveResponseHeadersFilter) {
   EXPECT_THAT(response->body(), HasSubstr("missing required header: :status"));
 }
 
+TEST_P(ProtocolIntegrationTest, ReqRespSizeStats) {
+  config_helper_.addConfigModifier([&](envoy::config::bootstrap::v3::Bootstrap& bootstrap) -> void {
+    RELEASE_ASSERT(bootstrap.mutable_static_resources()->clusters_size() == 1, "");
+    auto& cluster = *bootstrap.mutable_static_resources()->mutable_clusters(0);
+    cluster.mutable_track_cluster_stats()->set_request_response_sizes(true);
+  });
+  initialize();
+
+  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
+  Http::TestRequestHeaderMapImpl request_headers{
+      {":method", "GET"}, {":path", "/found"}, {":scheme", "http"}, {":authority", "foo.com"}};
+  Http::TestResponseHeaderMapImpl response_headers{{":status", "200"}};
+  auto response = sendRequestAndWaitForResponse(request_headers, 0, response_headers, 0, 0,
+                                                TestUtility::DefaultTimeout);
+  EXPECT_TRUE(upstream_request_->complete());
+  ASSERT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().getStatusValue());
+
+  test_server_->waitUntilHistogramHasSamples("cluster.cluster_0.upstream_rq_headers_size");
+  test_server_->waitUntilHistogramHasSamples("cluster.cluster_0.upstream_rs_headers_size");
+}
+
 } // namespace Envoy
diff --git a/test/integration/server.h b/test/integration/server.h
index a192058fa6e0b..e520dc1f6447f 100644
--- a/test/integration/server.h
+++ b/test/integration/server.h
@@ -466,6 +466,13 @@ class IntegrationTestServer : public Logger::Loggable<Logger::Id::testing>,
     notifyingStatsAllocator().waitForCounterExists(name);
   }
 
+  void waitUntilHistogramHasSamples(
+      const std::string& name,
+      std::chrono::milliseconds timeout = std::chrono::milliseconds::zero()) override {
+    ASSERT_TRUE(
+        TestUtility::waitUntilHistogramHasSamples(statStore(), name, time_system_, timeout));
+  }
+
   Stats::CounterSharedPtr counter(const std::string& name) override {
     // When using the thread local store, only counters() is thread safe. This also allows us
     // to test if a counter exists at all versus just defaulting to zero.
@@ -482,6 +489,10 @@ class IntegrationTestServer : public Logger::Loggable<Logger::Id::testing>,
 
   std::vector<Stats::GaugeSharedPtr> gauges() override { return statStore().gauges(); }
 
+  std::vector<Stats::ParentHistogramSharedPtr> histograms() override {
+    return statStore().histograms();
+  }
+
   // ListenerHooks
   void onWorkerListenerAdded() override;
   void onWorkerListenerRemoved() override;
diff --git a/test/integration/server_stats.h b/test/integration/server_stats.h
index 6c169ed68dd46..66cb7e07e7d27 100644
--- a/test/integration/server_stats.h
+++ b/test/integration/server_stats.h
@@ -38,6 +38,14 @@ class IntegrationTestServerStats {
    */
   virtual void waitForCounterExists(const std::string& name) PURE;
 
+  /**
+   * Wait until a histogram has samples.
+   * @param name histogram name.
+   */
+  virtual void waitUntilHistogramHasSamples(
+      const std::string& name,
+      std::chrono::milliseconds timeout = std::chrono::milliseconds::zero()) PURE;
+
   /**
    * Wait for a gauge to >= a given value.
    * @param name gauge name.
@@ -83,6 +91,11 @@ class IntegrationTestServerStats {
    * @return std::vector<Stats::GaugeSharedPtr> snapshot of server counters.
    */
   virtual std::vector<Stats::GaugeSharedPtr> gauges() PURE;
+
+  /**
+   * @return std::vector<Stats::ParentHistogramSharedPtr> snapshot of server histograms.
+   */
+  virtual std::vector<Stats::ParentHistogramSharedPtr> histograms() PURE;
 };
 
 } // namespace Envoy
diff --git a/test/test_common/utility.cc b/test/test_common/utility.cc
index accd549af2031..ab13448ecc794 100644
--- a/test/test_common/utility.cc
+++ b/test/test_common/utility.cc
@@ -222,6 +222,28 @@ AssertionResult TestUtility::waitForGaugeEq(Stats::Store& store, const std::stri
   return AssertionSuccess();
 }
 
+AssertionResult TestUtility::waitUntilHistogramHasSamples(Stats::Store& store,
+                                                          const std::string& name,
+                                                          Event::TestTimeSystem& time_system,
+                                                          std::chrono::milliseconds timeout) {
+  Event::TestTimeSystem::RealTimeBound bound(timeout);
+  while (true) {
+    auto histo = findByName<Stats::ParentHistogramSharedPtr>(store.histograms(), name);
+    if (histo) {
+      if (histo->cumulativeStatistics().sampleCount()) {
+        break;
+      }
+    }
+
+    time_system.advanceTimeWait(std::chrono::milliseconds(10));
+
+    if (timeout != std::chrono::milliseconds::zero() && !bound.withinBound()) {
+      return AssertionFailure() << fmt::format("timed out waiting for {} to have samples", name);
+    }
+  }
+  return AssertionSuccess();
+}
+
 std::list<Network::DnsResponse>
 TestUtility::makeDnsResponse(const std::list<std::string>& addresses, std::chrono::seconds ttl) {
   std::list<Network::DnsResponse> ret;
diff --git a/test/test_common/utility.h b/test/test_common/utility.h
index 692b7248b768e..ce92158009fd2 100644
--- a/test/test_common/utility.h
+++ b/test/test_common/utility.h
@@ -278,6 +278,19 @@ class TestUtility {
                  Event::TestTimeSystem& time_system,
                  std::chrono::milliseconds timeout = std::chrono::milliseconds::zero());
 
+  /**
+   * Wait for a histogram to have samples.
+   * @param store supplies the stats store.
+   * @param name histogram name.
+   * @param time_system the time system to use for waiting.
+   * @param timeout the maximum time to wait before timing out, or 0 for no timeout.
+   * @return AssertionSuccess() if the histogram was populated within the timeout, else
+   * AssertionFailure().
+   */
+  static AssertionResult waitUntilHistogramHasSamples(
+      Stats::Store& store, const std::string& name, Event::TestTimeSystem& time_system,
+      std::chrono::milliseconds timeout = std::chrono::milliseconds::zero());
+
   /**
    * Find a readout in a stats store.
    * @param store supplies the stats store.