thrift: add req/resp size histograms

docs/root/configuration/other_protocols/thrift_filters/router_filter.rst

@@ -41,3 +41,5 @@ Since these stats utilize the underlying cluster scope, we prefix with the ``thr
   thrift.upstream_resp_exception, Counter, Total responses with the "Exception" message type.
   thrift.upstream_resp_invalid_type, Counter, Total responses with an unsupported message type.
   thrift.upstream_rq_time, Histogram, total rq time from rq complete to resp complete; includes oneway messages.
+  thrift.upstream_rq_size, Histogram, Request message size in bytes per upstream
+  thrift.upstream_rs_size, Histogram, Response message size in bytes per upstream

docs/root/version_history/current.rst

@@ -69,6 +69,7 @@ New Features
 * http: added upstream and downstream alpha HTTP/3 support! See :ref:`quic_options <envoy_v3_api_field_config.listener.v3.UdpListenerConfig.quic_options>` for downstream and the new http3_protocol_options in :ref:`http_protocol_options <envoy_v3_api_msg_extensions.upstreams.http.v3.HttpProtocolOptions>` for upstream HTTP/3.
 * listener: added ability to change an existing listener's address.
 * metric service: added support for sending metric tags as labels. This can be enabled by setting the :ref:`emit_tags_as_labels <envoy_v3_api_field_config.metrics.v3.MetricsServiceConfig.emit_tags_as_labels>` field to true.
+* thrift_proxy: added per upstream metrics within the :ref:`thrift router <envoy_v3_api_msg_extensions.filters.network.thrift_proxy.router.v3.Router>` for request and response size histograms.
 * tcp: added support for :ref:`preconnecting <v1.18.0:envoy_v3_api_msg_config.cluster.v3.Cluster.PreconnectPolicy>`. Preconnecting is off by default, but recommended for clusters serving latency-sensitive traffic.
 * udp_proxy: added :ref:`key <envoy_v3_api_msg_extensions.filters.udp.udp_proxy.v3.UdpProxyConfig.HashPolicy>` as another hash policy to support hash based routing on any given key.
 

source/extensions/filters/network/thrift_proxy/router/router_impl.cc

@@ -316,6 +316,9 @@ FilterStatus Router::messageEnd() {
 
   upstream_request_->transport_->encodeFrame(transport_buffer, *upstream_request_->metadata_,
                                              upstream_request_buffer_);
+
+  recordClusterScopeHistogram({upstream_rq_size_}, Stats::Histogram::Unit::Bytes,
+                              transport_buffer.length());
   upstream_request_->conn_data_->connection().write(transport_buffer, false);
   upstream_request_->onRequestComplete();
   return FilterStatus::Continue;
@@ -324,6 +327,8 @@ FilterStatus Router::messageEnd() {
 void Router::onUpstreamData(Buffer::Instance& data, bool end_stream) {
   ASSERT(!upstream_request_->response_complete_);
 
+  recordClusterScopeHistogram({upstream_rs_size_}, Stats::Histogram::Unit::Bytes, data.length());
+
   if (upstream_request_->upgrade_response_ != nullptr) {
     ENVOY_STREAM_LOG(trace, "reading upgrade response: {} bytes", *callbacks_, data.length());
     // Handle upgrade response.
@@ -503,6 +508,9 @@ void Router::UpstreamRequest::onPoolReady(Tcp::ConnectionPool::ConnectionDataPtr
     upgrade_response_ =
         protocol_->attemptUpgrade(*transport_, *conn_state_, parent_.upstream_request_buffer_);
     if (upgrade_response_ != nullptr) {
+      parent_.recordClusterScopeHistogram({parent_.upstream_rq_size_},
+                                          Stats::Histogram::Unit::Bytes,
+                                          parent_.upstream_request_buffer_.length());
       conn_data_->connection().write(parent_.upstream_request_buffer_, false);
       return;
     }

source/extensions/filters/network/thrift_proxy/router/router_impl.h

@@ -190,6 +190,8 @@ class Router : public Tcp::ConnectionPool::UpstreamCallbacks,
         upstream_resp_exception_(stat_name_set_->add("thrift.upstream_resp_exception")),
         upstream_resp_invalid_type_(stat_name_set_->add("thrift.upstream_resp_invalid_type")),
         upstream_rq_time_(stat_name_set_->add("thrift.upstream_rq_time")),
+        upstream_rq_size_(stat_name_set_->add("thrift.upstream_rq_size")),
+        upstream_rs_size_(stat_name_set_->add("thrift.upstream_rs_size")),
         passthrough_supported_(false) {}
 
   ~Router() override = default;
@@ -299,6 +301,8 @@ class Router : public Tcp::ConnectionPool::UpstreamCallbacks,
   const Stats::StatName upstream_resp_exception_;
   const Stats::StatName upstream_resp_invalid_type_;
   const Stats::StatName upstream_rq_time_;
+  const Stats::StatName upstream_rq_size_;
+  const Stats::StatName upstream_rs_size_;
 
   ThriftFilters::DecoderFilterCallbacks* callbacks_{};
   RouteConstSharedPtr route_{};

test/extensions/filters/network/thrift_proxy/router_test.cc

@@ -1022,6 +1022,17 @@ TEST_P(ThriftRouterFieldTypeTest, CallWithUpstreamRqTime) {
   EXPECT_CALL(cluster_scope, counter("thrift.upstream_resp_reply"));
   EXPECT_CALL(cluster_scope, counter("thrift.upstream_resp_success"));
 
+  EXPECT_CALL(cluster_scope, histogram("thrift.upstream_rq_size", Stats::Histogram::Unit::Bytes));
+  EXPECT_CALL(cluster_scope,
+              deliverHistogramToSinks(
+                  testing::Property(&Stats::Metric::name, "thrift.upstream_rq_size"), _));
+  EXPECT_CALL(cluster_scope, histogram("thrift.upstream_rs_size", Stats::Histogram::Unit::Bytes))
+      .Times(2);
+  EXPECT_CALL(cluster_scope,
+              deliverHistogramToSinks(
+                  testing::Property(&Stats::Metric::name, "thrift.upstream_rs_size"), _))
+      .Times(2);
+
   startRequest(MessageType::Call);
   connectUpstream();
   sendTrivialStruct(field_type);
@@ -1297,6 +1308,41 @@ TEST_P(ThriftRouterPassthroughTest, PassthroughEnable) {
       ConnectionPool::PoolFailureReason::RemoteConnectionFailure);
 }
 
+TEST_F(ThriftRouterTest, RequestResponseSize) {
+  initializeRouter();
+
+  Stats::MockStore cluster_scope;
+  ON_CALL(*context_.cluster_manager_.thread_local_cluster_.cluster_.info_, statsScope())
+      .WillByDefault(ReturnRef(cluster_scope));
+
+  EXPECT_CALL(cluster_scope, counter("thrift.upstream_rq_call")).Times(AtLeast(1));
+  EXPECT_CALL(cluster_scope, counter("thrift.upstream_resp_reply")).Times(AtLeast(1));
+  EXPECT_CALL(cluster_scope, counter("thrift.upstream_resp_success")).Times(AtLeast(1));
+
+  EXPECT_CALL(cluster_scope,
+              histogram("thrift.upstream_rq_time", Stats::Histogram::Unit::Milliseconds));
+  EXPECT_CALL(cluster_scope,
+              deliverHistogramToSinks(
+                  testing::Property(&Stats::Metric::name, "thrift.upstream_rq_time"), _));
+
+  EXPECT_CALL(cluster_scope, histogram("thrift.upstream_rq_size", Stats::Histogram::Unit::Bytes));
+  EXPECT_CALL(cluster_scope,
+              deliverHistogramToSinks(
+                  testing::Property(&Stats::Metric::name, "thrift.upstream_rq_size"), _));
+  EXPECT_CALL(cluster_scope, histogram("thrift.upstream_rs_size", Stats::Histogram::Unit::Bytes))
+      .Times(2);
+  EXPECT_CALL(cluster_scope,
+              deliverHistogramToSinks(
+                  testing::Property(&Stats::Metric::name, "thrift.upstream_rs_size"), _))
+      .Times(2);
+
+  startRequestWithExistingConnection(MessageType::Call);
+  sendTrivialStruct(FieldType::I32);
+  completeRequest();
+  returnResponse();
+  destroyRouter();
+}
+
 } // namespace Router
 } // namespace ThriftProxy
 } // namespace NetworkFilters