thrift_proxy: add upstream cluster metrics for MessageType

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

@@ -23,3 +23,9 @@ The filter outputs statistics in the *thrift.<stat_prefix>.* namespace.
   unknown_cluster, Counter, Total requests with a route that has an unknown cluster.
   upstream_rq_maintenance_mode, Counter, Total requests with a destination cluster in maintenance mode.
   no_healthy_upstream, Counter, Total requests with no healthy upstream endpoints available.
+  request_call, Counter, Total requests with the "Call" message type.
+  request_oneway, Counter, Total requests with the "Oneway" message type.
+  request_invalid_type, Counter, Total requests with an unsupported message type.
+  response_reply, Counter, Total responses with the "Reply" message type. Includes both successes and errors.
+  response_exception, Counter, Total responses with the "Exception" message type.
+  response_invalid_type, Counter, Total responses with an unsupported message type.

docs/root/version_history/current.rst

@@ -157,6 +157,7 @@ New Features
 * tcp_proxy: added a :ref:`use_post field <envoy_v3_api_field_extensions.filters.network.tcp_proxy.v3.TcpProxy.TunnelingConfig.use_post>` for using HTTP POST to proxy TCP streams.
 * tcp_proxy: added a :ref:`headers_to_add field <envoy_v3_api_field_extensions.filters.network.tcp_proxy.v3.TcpProxy.TunnelingConfig.headers_to_add>` for setting additional headers to the HTTP requests for TCP proxing.
 * thrift_proxy: added a :ref:`max_requests_per_connection field <envoy_v3_api_field_extensions.filters.network.thrift_proxy.v3.ThriftProxy.max_requests_per_connection>` for setting maximum requests for per downstream connection.
+* thrift_proxy: added per upstream metrics within the :ref:`thrift router <envoy_v3_api_field_extensions.filters.network.thrift_proxy.router.v3.Router>` for messagetype in request/response.
 * tls peer certificate validation: added :ref:`SPIFFE validator <envoy_v3_api_msg_extensions.transport_sockets.tls.v3.SPIFFECertValidatorConfig>` for supporting isolated multiple trust bundles in a single listener or cluster.
 * tracing: added the :ref:`pack_trace_reason <envoy_v3_api_field_extensions.request_id.uuid.v3.UuidRequestIdConfig.pack_trace_reason>`
   field as well as explicit configuration for the built-in :ref:`UuidRequestIdConfig <envoy_v3_api_msg_extensions.request_id.uuid.v3.UuidRequestIdConfig>`

source/extensions/filters/network/thrift_proxy/conn_manager.h

@@ -146,6 +146,7 @@ class ConnectionManager : public Network::ReadFilter,
     }
     void resetDownstreamConnection() override { parent_.resetDownstreamConnection(); }
     StreamInfo::StreamInfo& streamInfo() override { return parent_.streamInfo(); }
+    MessageMetadataSharedPtr metadata() override { return parent_.metadata(); }
 
     ActiveRpc& parent_;
     ThriftFilters::DecoderFilterSharedPtr handle_;
@@ -217,6 +218,7 @@ class ConnectionManager : public Network::ReadFilter,
     ThriftFilters::ResponseStatus upstreamData(Buffer::Instance& buffer) override;
     void resetDownstreamConnection() override;
     StreamInfo::StreamInfo& streamInfo() override { return stream_info_; }
+    MessageMetadataSharedPtr metadata() override { return metadata_; }
 
     // Thrift::FilterChainFactoryCallbacks
     void addDecoderFilter(ThriftFilters::DecoderFilterSharedPtr filter) override {

source/extensions/filters/network/thrift_proxy/filters/filter.h

@@ -98,6 +98,11 @@ class DecoderFilterCallbacks {
    * @return StreamInfo for logging purposes.
    */
   virtual StreamInfo::StreamInfo& streamInfo() PURE;
+
+  /**
+   * @return Metadata contained in the callbacks.
+   */
+  virtual MessageMetadataSharedPtr metadata() PURE;
 };
 
 /**

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

@@ -239,6 +239,19 @@ FilterStatus Router::messageBegin(MessageMetadataSharedPtr metadata) {
   cluster_ = cluster->info();
   ENVOY_STREAM_LOG(debug, "cluster '{}' match for method '{}'", *callbacks_, cluster_name,
                    metadata->methodName());
+  switch (metadata->messageType()) {
+  case MessageType::Call:
+    incClusterScopeCounter(request_call_);
+    break;
+
+  case MessageType::Oneway:
+    incClusterScopeCounter(request_oneway_);
+    break;
+
+  default:
+    incClusterScopeCounter(request_invalid_type_);
+    break;
+  }
 
   if (cluster_->maintenanceMode()) {
     stats_.upstream_rq_maintenance_mode_.inc();
@@ -338,6 +351,19 @@ void Router::onUpstreamData(Buffer::Instance& data, bool end_stream) {
     ThriftFilters::ResponseStatus status = callbacks_->upstreamData(data);
     if (status == ThriftFilters::ResponseStatus::Complete) {
       ENVOY_STREAM_LOG(debug, "response complete", *callbacks_);
+      switch (callbacks_->metadata()->messageType()) {
+      case MessageType::Reply:
+        incClusterScopeCounter(response_reply_);
+        break;
+
+      case MessageType::Exception:
+        incClusterScopeCounter(response_exception_);
+        break;
+
+      default:
+        incClusterScopeCounter(response_invalid_type_);
+        break;
+      }
       upstream_request_->onResponseComplete();
       cleanup();
       return;

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

@@ -164,7 +164,13 @@ class RouteMatcher {
   COUNTER(route_missing)                                                                           \
   COUNTER(unknown_cluster)                                                                         \
   COUNTER(upstream_rq_maintenance_mode)                                                            \
-  COUNTER(no_healthy_upstream)
+  COUNTER(no_healthy_upstream)                                                                     \
+  COUNTER(request_call)                                                                            \
+  COUNTER(request_oneway)                                                                          \
+  COUNTER(request_invalid_type)                                                                    \
+  COUNTER(response_reply)                                                                          \
+  COUNTER(response_exception)                                                                      \
+  COUNTER(response_invalid_type)
 
 struct RouterStats {
   ALL_THRIFT_ROUTER_STATS(GENERATE_COUNTER_STRUCT, GENERATE_GAUGE_STRUCT, GENERATE_HISTOGRAM_STRUCT)
@@ -179,6 +185,13 @@ class Router : public Tcp::ConnectionPool::UpstreamCallbacks,
   Router(Upstream::ClusterManager& cluster_manager, const std::string& stat_prefix,
          Stats::Scope& scope)
       : cluster_manager_(cluster_manager), stats_(generateStats(stat_prefix, scope)),
+        stat_name_set_(scope.symbolTable().makeSet("thrift_proxy")),
+        request_call_(stat_name_set_->add("request_call")),
+        request_oneway_(stat_name_set_->add("request_oneway")),
+        request_invalid_type_(stat_name_set_->add("request_invalid_type")),
+        response_reply_(stat_name_set_->add("response_reply")),
+        response_exception_(stat_name_set_->add("response_exception")),
+        response_invalid_type_(stat_name_set_->add("response_invalid_type")),
         passthrough_supported_(false) {}
 
   ~Router() override = default;
@@ -188,6 +201,11 @@ class Router : public Tcp::ConnectionPool::UpstreamCallbacks,
   void setDecoderFilterCallbacks(ThriftFilters::DecoderFilterCallbacks& callbacks) override;
   bool passthroughSupported() const override { return passthrough_supported_; }
 
+  // Stats
+  void incClusterScopeCounter(Stats::StatName name) {
+    cluster_->statsScope().counterFromStatName(name).inc();
+  }
+
   // ProtocolConverter
   FilterStatus transportBegin(MessageMetadataSharedPtr metadata) override;
   FilterStatus transportEnd() override;
@@ -259,6 +277,13 @@ class Router : public Tcp::ConnectionPool::UpstreamCallbacks,
 
   Upstream::ClusterManager& cluster_manager_;
   RouterStats stats_;
+  Stats::StatNameSetPtr stat_name_set_;
+  const Stats::StatName request_call_;
+  const Stats::StatName request_oneway_;
+  const Stats::StatName request_invalid_type_;
+  const Stats::StatName response_reply_;
+  const Stats::StatName response_exception_;
+  const Stats::StatName response_invalid_type_;
 
   ThriftFilters::DecoderFilterCallbacks* callbacks_{};
   RouteConstSharedPtr route_{};

test/extensions/filters/network/thrift_proxy/mocks.h

@@ -255,10 +255,12 @@ class MockDecoderFilterCallbacks : public DecoderFilterCallbacks {
   MOCK_METHOD(ResponseStatus, upstreamData, (Buffer::Instance&));
   MOCK_METHOD(void, resetDownstreamConnection, ());
   MOCK_METHOD(StreamInfo::StreamInfo&, streamInfo, ());
+  MOCK_METHOD(MessageMetadataSharedPtr, metadata, ());
 
   uint64_t stream_id_{1};
   NiceMock<Network::MockConnection> connection_;
   NiceMock<StreamInfo::MockStreamInfo> stream_info_;
+  MessageMetadataSharedPtr metadata_;
   std::shared_ptr<Router::MockRoute> route_;
 };
 

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

@@ -305,11 +305,16 @@ class ThriftRouterTestBase {
     EXPECT_EQ(FilterStatus::Continue, router_->transportEnd());
   }
 
-  void returnResponse() {
+  void returnResponse(MessageType msg_type = MessageType::Reply) {
     Buffer::OwnedImpl buffer;
 
     EXPECT_CALL(callbacks_, startUpstreamResponse(_, _));
 
+    auto metadata = std::make_shared<MessageMetadata>();
+    metadata->setMessageType(msg_type);
+    metadata->setSequenceId(1);
+    ON_CALL(callbacks_, metadata()).WillByDefault(Return(metadata));
+
     EXPECT_CALL(callbacks_, upstreamData(Ref(buffer)))
         .WillOnce(Return(ThriftFilters::ResponseStatus::MoreData));
     upstream_callbacks_->onUpstreamData(buffer, false);
@@ -415,6 +420,10 @@ TEST_F(ThriftRouterTest, PoolRemoteConnectionFailure) {
 
   startRequest(MessageType::Call);
 
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+
   EXPECT_CALL(callbacks_, sendLocalReply(_, _))
       .WillOnce(Invoke([&](const DirectResponse& response, bool end_stream) -> void {
         auto& app_ex = dynamic_cast<const AppException&>(response);
@@ -431,6 +440,10 @@ TEST_F(ThriftRouterTest, PoolLocalConnectionFailure) {
 
   startRequest(MessageType::Call);
 
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+
   context_.cluster_manager_.thread_local_cluster_.tcp_conn_pool_.poolFailure(
       ConnectionPool::PoolFailureReason::LocalConnectionFailure);
 }
@@ -440,6 +453,10 @@ TEST_F(ThriftRouterTest, PoolTimeout) {
 
   startRequest(MessageType::Call);
 
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+
   EXPECT_CALL(callbacks_, sendLocalReply(_, _))
       .WillOnce(Invoke([&](const DirectResponse& response, bool end_stream) -> void {
         auto& app_ex = dynamic_cast<const AppException&>(response);
@@ -456,6 +473,10 @@ TEST_F(ThriftRouterTest, PoolOverflowFailure) {
 
   startRequest(MessageType::Call);
 
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+
   EXPECT_CALL(callbacks_, sendLocalReply(_, _))
       .WillOnce(Invoke([&](const DirectResponse& response, bool end_stream) -> void {
         auto& app_ex = dynamic_cast<const AppException&>(response);
@@ -471,6 +492,10 @@ TEST_F(ThriftRouterTest, PoolConnectionFailureWithOnewayMessage) {
   initializeRouter();
   startRequest(MessageType::Oneway);
 
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_oneway")
+                     .value());
+
   EXPECT_CALL(callbacks_, sendLocalReply(_, _)).Times(0);
   EXPECT_CALL(callbacks_, resetDownstreamConnection());
   context_.cluster_manager_.thread_local_cluster_.tcp_conn_pool_.poolFailure(
@@ -534,6 +559,9 @@ TEST_F(ThriftRouterTest, ClusterMaintenanceMode) {
       }));
   EXPECT_EQ(FilterStatus::StopIteration, router_->messageBegin(metadata_));
   EXPECT_EQ(1U, context_.scope().counterFromString("test.upstream_rq_maintenance_mode").value());
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
 }
 
 TEST_F(ThriftRouterTest, NoHealthyHosts) {
@@ -556,6 +584,9 @@ TEST_F(ThriftRouterTest, NoHealthyHosts) {
 
   EXPECT_EQ(FilterStatus::StopIteration, router_->messageBegin(metadata_));
   EXPECT_EQ(1U, context_.scope().counterFromString("test.no_healthy_upstream").value());
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
 }
 
 TEST_F(ThriftRouterTest, TruncatedResponse) {
@@ -760,6 +791,13 @@ TEST_F(ThriftRouterTest, ProtocolUpgrade) {
   completeRequest();
   returnResponse();
   destroyRouter();
+
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("response_reply")
+                     .value());
 }
 
 // Test the case where an upgrade will occur, but the conn pool
@@ -833,6 +871,13 @@ TEST_F(ThriftRouterTest, ProtocolUpgradeOnExistingUnusedConnection) {
   completeRequest();
   returnResponse();
   destroyRouter();
+
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("response_reply")
+                     .value());
 }
 
 TEST_F(ThriftRouterTest, ProtocolUpgradeSkippedOnExistingConnection) {
@@ -873,6 +918,13 @@ TEST_F(ThriftRouterTest, ProtocolUpgradeSkippedOnExistingConnection) {
   completeRequest();
   returnResponse();
   destroyRouter();
+
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("response_reply")
+                     .value());
 }
 
 TEST_P(ThriftRouterFieldTypeTest, OneWay) {
@@ -884,6 +936,13 @@ TEST_P(ThriftRouterFieldTypeTest, OneWay) {
   sendTrivialStruct(field_type);
   completeRequest();
   destroyRouter();
+
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_oneway")
+                     .value());
+  EXPECT_EQ(0UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("response_reply")
+                     .value());
 }
 
 TEST_P(ThriftRouterFieldTypeTest, Call) {
@@ -896,6 +955,13 @@ TEST_P(ThriftRouterFieldTypeTest, Call) {
   completeRequest();
   returnResponse();
   destroyRouter();
+
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("response_reply")
+                     .value());
 }
 
 // Ensure the service name gets stripped when strip_service_name = true.
@@ -912,6 +978,13 @@ TEST_P(ThriftRouterFieldTypeTest, StripServiceNameEnabled) {
 
   returnResponse();
   destroyRouter();
+
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("response_reply")
+                     .value());
 }
 
 // Ensure the service name prefix isn't stripped when strip_service_name = false.
@@ -928,6 +1001,13 @@ TEST_P(ThriftRouterFieldTypeTest, StripServiceNameDisabled) {
 
   returnResponse();
   destroyRouter();
+
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("response_reply")
+                     .value());
 }
 
 TEST_F(ThriftRouterTest, CallWithExistingConnection) {
@@ -944,6 +1024,13 @@ TEST_F(ThriftRouterTest, CallWithExistingConnection) {
 
   returnResponse();
   destroyRouter();
+
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("request_call")
+                     .value());
+  EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope()
+                     .counterFromString("response_reply")
+                     .value());
 }
 
 TEST_P(ThriftRouterContainerTest, DecoderFilterCallbacks) {