diff --git a/docs/root/configuration/other_protocols/thrift_filters/router_filter.rst b/docs/root/configuration/other_protocols/thrift_filters/router_filter.rst index 171f6550bdb45..009c22e3e50a3 100644 --- a/docs/root/configuration/other_protocols/thrift_filters/router_filter.rst +++ b/docs/root/configuration/other_protocols/thrift_filters/router_filter.rst @@ -25,15 +25,19 @@ The filter outputs generic routing error statistics in the *thrift. no_healthy_upstream, Counter, Total requests with no healthy upstream endpoints available. -The filter also outputs MessageType statistics in the upstream cluster's stat scope. +The filter is also responsible for cluster-level statistics derived from routed upstream clusters. +Since these stats utilize the underlying cluster scope, we prefix with the `thrift` namespace. .. csv-table:: :header: Name, Type, Description :widths: 1, 1, 2 - 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. + thrift.upstream_rq_call, Counter, Total requests with the "Call" message type. + thrift.upstream_rq_oneway, Counter, Total requests with the "Oneway" message type. + thrift.upstream_rq_invalid_type, Counter, Total requests with an unsupported message type. + thrift.upstream_resp_reply, Counter, Total responses with the "Reply" message type. Sums both Successses and Errors. + thrift.upstream_resp_success, Counter, Total Replies that are considered "Successes". + thrift.upstream_resp_error, Counter, Total Replies that are considered "Errors". + 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. diff --git a/docs/root/version_history/current.rst b/docs/root/version_history/current.rst index a3c7b27d34e01..25dee82da9737 100644 --- a/docs/root/version_history/current.rst +++ b/docs/root/version_history/current.rst @@ -161,7 +161,8 @@ New Features * tcp_proxy: added a :ref:`use_post field ` for using HTTP POST to proxy TCP streams. * tcp_proxy: added a :ref:`headers_to_add field ` for setting additional headers to the HTTP requests for TCP proxing. * thrift_proxy: added a :ref:`max_requests_per_connection field ` for setting maximum requests for per downstream connection. -* thrift_proxy: added per upstream metrics within the :ref:`thrift router ` for messagetype in request/response. +* thrift_proxy: added per upstream metrics within the :ref:`thrift router ` for messagetype counters in request/response. +* thrift_proxy: added per upstream metrics within the :ref:`thrift router ` for request time histograms. * tls peer certificate validation: added :ref:`SPIFFE validator ` for supporting isolated multiple trust bundles in a single listener or cluster. * tracing: added the :ref:`pack_trace_reason ` field as well as explicit configuration for the built-in :ref:`UuidRequestIdConfig ` diff --git a/source/extensions/filters/network/thrift_proxy/conn_manager.h b/source/extensions/filters/network/thrift_proxy/conn_manager.h index 879b2d5e7be00..eb056aaed1221 100644 --- a/source/extensions/filters/network/thrift_proxy/conn_manager.h +++ b/source/extensions/filters/network/thrift_proxy/conn_manager.h @@ -127,6 +127,7 @@ class ConnectionManager : public Network::ReadFilter, // ThriftFilters::DecoderFilterCallbacks uint64_t streamId() const override { return parent_.stream_id_; } const Network::Connection* connection() const override { return parent_.connection(); } + Event::Dispatcher& dispatcher() override { return parent_.dispatcher(); } void continueDecoding() override; Router::RouteConstSharedPtr route() override { return parent_.route(); } TransportType downstreamTransportType() const override { @@ -206,6 +207,9 @@ class ConnectionManager : public Network::ReadFilter, // ThriftFilters::DecoderFilterCallbacks uint64_t streamId() const override { return stream_id_; } const Network::Connection* connection() const override; + Event::Dispatcher& dispatcher() override { + return parent_.read_callbacks_->connection().dispatcher(); + } void continueDecoding() override { parent_.continueDecoding(); } Router::RouteConstSharedPtr route() override; TransportType downstreamTransportType() const override { diff --git a/source/extensions/filters/network/thrift_proxy/filters/filter.h b/source/extensions/filters/network/thrift_proxy/filters/filter.h index a19ab8ef5351e..b8b73ddf517b5 100644 --- a/source/extensions/filters/network/thrift_proxy/filters/filter.h +++ b/source/extensions/filters/network/thrift_proxy/filters/filter.h @@ -43,6 +43,11 @@ class DecoderFilterCallbacks { */ virtual const Network::Connection* connection() const PURE; + /** + * @return Event::Dispatcher& the thread local dispatcher for allocating timers, etc. + */ + virtual Event::Dispatcher& dispatcher() PURE; + /** * Continue iterating through the filter chain with buffered data. This routine can only be * called if the filter has previously returned StopIteration from one of the DecoderFilter diff --git a/source/extensions/filters/network/thrift_proxy/router/router_impl.cc b/source/extensions/filters/network/thrift_proxy/router/router_impl.cc index 92ddc7eae09fb..f239463073bfb 100644 --- a/source/extensions/filters/network/thrift_proxy/router/router_impl.cc +++ b/source/extensions/filters/network/thrift_proxy/router/router_impl.cc @@ -241,15 +241,15 @@ FilterStatus Router::messageBegin(MessageMetadataSharedPtr metadata) { metadata->methodName()); switch (metadata->messageType()) { case MessageType::Call: - incClusterScopeCounter(request_call_); + incClusterScopeCounter({upstream_rq_call_}); break; case MessageType::Oneway: - incClusterScopeCounter(request_oneway_); + incClusterScopeCounter({upstream_rq_oneway_}); break; default: - incClusterScopeCounter(request_invalid_type_); + incClusterScopeCounter({upstream_rq_invalid_type_}); break; } @@ -353,20 +353,20 @@ void Router::onUpstreamData(Buffer::Instance& data, bool end_stream) { ENVOY_STREAM_LOG(debug, "response complete", *callbacks_); switch (callbacks_->responseMetadata()->messageType()) { case MessageType::Reply: - incClusterScopeCounter(response_reply_); + incClusterScopeCounter({upstream_resp_reply_}); if (callbacks_->responseSuccess()) { - incClusterScopeCounter(response_reply_success_); + incClusterScopeCounter({upstream_resp_reply_success_}); } else { - incClusterScopeCounter(response_reply_error_); + incClusterScopeCounter({upstream_resp_reply_error_}); } break; case MessageType::Exception: - incClusterScopeCounter(response_exception_); + incClusterScopeCounter({upstream_resp_exception_}); break; default: - incClusterScopeCounter(response_invalid_type_); + incClusterScopeCounter({upstream_resp_invalid_type_}); break; } upstream_request_->onResponseComplete(); @@ -522,9 +522,14 @@ void Router::UpstreamRequest::onRequestStart(bool continue_decoding) { } } -void Router::UpstreamRequest::onRequestComplete() { request_complete_ = true; } +void Router::UpstreamRequest::onRequestComplete() { + Event::Dispatcher& dispatcher = parent_.callbacks_->dispatcher(); + downstream_request_complete_time_ = dispatcher.timeSource().monotonicTime(); + request_complete_ = true; +} void Router::UpstreamRequest::onResponseComplete() { + chargeResponseTiming(); response_complete_ = true; conn_state_ = nullptr; conn_data_.reset(); @@ -542,6 +547,8 @@ void Router::UpstreamRequest::onResetStream(ConnectionPool::PoolFailureReason re return; } + chargeResponseTiming(); + switch (reason) { case ConnectionPool::PoolFailureReason::Overflow: parent_.callbacks_->sendLocalReply( @@ -576,6 +583,20 @@ void Router::UpstreamRequest::onResetStream(ConnectionPool::PoolFailureReason re } } +void Router::UpstreamRequest::chargeResponseTiming() { + if (charged_response_timing_) { + return; + } + charged_response_timing_ = true; + Event::Dispatcher& dispatcher = parent_.callbacks_->dispatcher(); + const std::chrono::milliseconds response_time = + std::chrono::duration_cast( + dispatcher.timeSource().monotonicTime() - downstream_request_complete_time_); + const uint64_t count = response_time.count(); + parent_.recordClusterScopeHistogram({parent_.upstream_rq_time_}, + Stats::Histogram::Unit::Milliseconds, count); +} + } // namespace Router } // namespace ThriftProxy } // namespace NetworkFilters diff --git a/source/extensions/filters/network/thrift_proxy/router/router_impl.h b/source/extensions/filters/network/thrift_proxy/router/router_impl.h index 99162191e5f0d..e125c69fce0b8 100644 --- a/source/extensions/filters/network/thrift_proxy/router/router_impl.h +++ b/source/extensions/filters/network/thrift_proxy/router/router_impl.h @@ -180,14 +180,16 @@ class Router : public Tcp::ConnectionPool::UpstreamCallbacks, 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_reply_success_(stat_name_set_->add("response_success")), - response_reply_error_(stat_name_set_->add("response_error")), - response_exception_(stat_name_set_->add("response_exception")), - response_invalid_type_(stat_name_set_->add("response_invalid_type")), + symbol_table_(scope.symbolTable()), + upstream_rq_call_(stat_name_set_->add("thrift.upstream_rq_call")), + upstream_rq_oneway_(stat_name_set_->add("thrift.upstream_rq_oneway")), + upstream_rq_invalid_type_(stat_name_set_->add("thrift.upstream_rq_invalid_type")), + upstream_resp_reply_(stat_name_set_->add("thrift.upstream_resp_reply")), + upstream_resp_reply_success_(stat_name_set_->add("thrift.upstream_resp_success")), + upstream_resp_reply_error_(stat_name_set_->add("thrift.upstream_resp_error")), + 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")), passthrough_supported_(false) {} ~Router() override = default; @@ -197,11 +199,6 @@ 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; @@ -245,6 +242,7 @@ class Router : public Tcp::ConnectionPool::UpstreamCallbacks, void onResponseComplete(); void onUpstreamHostSelected(Upstream::HostDescriptionConstSharedPtr host); void onResetStream(ConnectionPool::PoolFailureReason reason); + void chargeResponseTiming(); Router& parent_; Tcp::ConnectionPool::Instance& conn_pool_; @@ -261,8 +259,25 @@ class Router : public Tcp::ConnectionPool::UpstreamCallbacks, bool request_complete_ : 1; bool response_started_ : 1; bool response_complete_ : 1; + + bool charged_response_timing_{false}; + MonotonicTime downstream_request_complete_time_; }; + // Stats + void incClusterScopeCounter(const Stats::StatNameVec& names) const { + const Stats::SymbolTable::StoragePtr stat_name_storage = symbol_table_.join(names); + cluster_->statsScope().counterFromStatName(Stats::StatName(stat_name_storage.get())).inc(); + } + + void recordClusterScopeHistogram(const Stats::StatNameVec& names, Stats::Histogram::Unit unit, + uint64_t count) const { + const Stats::SymbolTable::StoragePtr stat_name_storage = symbol_table_.join(names); + cluster_->statsScope() + .histogramFromStatName(Stats::StatName(stat_name_storage.get()), unit) + .recordValue(count); + } + void convertMessageBegin(MessageMetadataSharedPtr metadata); void cleanup(); RouterStats generateStats(const std::string& prefix, Stats::Scope& scope) { @@ -274,14 +289,16 @@ 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_reply_success_; - const Stats::StatName response_reply_error_; - const Stats::StatName response_exception_; - const Stats::StatName response_invalid_type_; + Stats::SymbolTable& symbol_table_; + const Stats::StatName upstream_rq_call_; + const Stats::StatName upstream_rq_oneway_; + const Stats::StatName upstream_rq_invalid_type_; + const Stats::StatName upstream_resp_reply_; + const Stats::StatName upstream_resp_reply_success_; + const Stats::StatName upstream_resp_reply_error_; + const Stats::StatName upstream_resp_exception_; + const Stats::StatName upstream_resp_invalid_type_; + const Stats::StatName upstream_rq_time_; ThriftFilters::DecoderFilterCallbacks* callbacks_{}; RouteConstSharedPtr route_{}; diff --git a/test/extensions/filters/network/thrift_proxy/mocks.h b/test/extensions/filters/network/thrift_proxy/mocks.h index 5595bd92dd0f5..7db68172d259a 100644 --- a/test/extensions/filters/network/thrift_proxy/mocks.h +++ b/test/extensions/filters/network/thrift_proxy/mocks.h @@ -246,6 +246,7 @@ class MockDecoderFilterCallbacks : public DecoderFilterCallbacks { // ThriftProxy::ThriftFilters::DecoderFilterCallbacks MOCK_METHOD(uint64_t, streamId, (), (const)); MOCK_METHOD(const Network::Connection*, connection, (), (const)); + MOCK_METHOD(Event::Dispatcher&, dispatcher, ()); MOCK_METHOD(void, continueDecoding, ()); MOCK_METHOD(Router::RouteConstSharedPtr, route, ()); MOCK_METHOD(TransportType, downstreamTransportType, (), (const)); diff --git a/test/extensions/filters/network/thrift_proxy/router_test.cc b/test/extensions/filters/network/thrift_proxy/router_test.cc index f0af7824717f9..72410c685d493 100644 --- a/test/extensions/filters/network/thrift_proxy/router_test.cc +++ b/test/extensions/filters/network/thrift_proxy/router_test.cc @@ -132,6 +132,7 @@ class ThriftRouterTestBase { EXPECT_EQ(FilterStatus::StopIteration, router_->messageBegin(metadata_)); EXPECT_CALL(callbacks_, connection()).WillRepeatedly(Return(&connection_)); + EXPECT_CALL(callbacks_, dispatcher()).WillRepeatedly(ReturnRef(dispatcher_)); EXPECT_EQ(&connection_, router_->downstreamConnection()); // Not yet implemented: @@ -194,6 +195,7 @@ class ThriftRouterTestBase { Invoke([&]() -> Tcp::ConnectionPool::ConnectionState* { return conn_state_.get(); })); EXPECT_CALL(callbacks_, connection()).WillRepeatedly(Return(&connection_)); + EXPECT_CALL(callbacks_, dispatcher()).WillRepeatedly(ReturnRef(dispatcher_)); EXPECT_EQ(&connection_, router_->downstreamConnection()); // Not yet implemented: @@ -342,6 +344,8 @@ class ThriftRouterTestBase { NiceMock context_; NiceMock connection_; + NiceMock dispatcher_; + NiceMock time_source_; NiceMock callbacks_; NiceMock* transport_{}; NiceMock* protocol_{}; @@ -422,7 +426,7 @@ TEST_F(ThriftRouterTest, PoolRemoteConnectionFailure) { startRequest(MessageType::Call); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_CALL(callbacks_, sendLocalReply(_, _)) @@ -442,7 +446,7 @@ TEST_F(ThriftRouterTest, PoolLocalConnectionFailure) { startRequest(MessageType::Call); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); context_.cluster_manager_.thread_local_cluster_.tcp_conn_pool_.poolFailure( @@ -455,7 +459,7 @@ TEST_F(ThriftRouterTest, PoolTimeout) { startRequest(MessageType::Call); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_CALL(callbacks_, sendLocalReply(_, _)) @@ -475,7 +479,7 @@ TEST_F(ThriftRouterTest, PoolOverflowFailure) { startRequest(MessageType::Call); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_CALL(callbacks_, sendLocalReply(_, _)) @@ -494,7 +498,7 @@ TEST_F(ThriftRouterTest, PoolConnectionFailureWithOnewayMessage) { startRequest(MessageType::Oneway); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_oneway") + .counterFromString("thrift.upstream_rq_oneway") .value()); EXPECT_CALL(callbacks_, sendLocalReply(_, _)).Times(0); @@ -561,7 +565,7 @@ 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") + .counterFromString("thrift.upstream_rq_call") .value()); } @@ -586,7 +590,7 @@ 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") + .counterFromString("thrift.upstream_rq_call") .value()); } @@ -794,13 +798,13 @@ TEST_F(ThriftRouterTest, ProtocolUpgrade) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_reply") + .counterFromString("thrift.upstream_resp_reply") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_success") + .counterFromString("thrift.upstream_resp_success") .value()); } @@ -877,13 +881,13 @@ TEST_F(ThriftRouterTest, ProtocolUpgradeOnExistingUnusedConnection) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_reply") + .counterFromString("thrift.upstream_resp_reply") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_success") + .counterFromString("thrift.upstream_resp_success") .value()); } @@ -927,16 +931,43 @@ TEST_F(ThriftRouterTest, ProtocolUpgradeSkippedOnExistingConnection) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_reply") + .counterFromString("thrift.upstream_resp_reply") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_success") + .counterFromString("thrift.upstream_resp_success") .value()); } +TEST_F(ThriftRouterTest, PoolTimeoutUpstreamTimeMeasurement) { + 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")); + + startRequest(MessageType::Call); + + dispatcher_.time_system_.advanceTimeWait(std::chrono::milliseconds(500)); + 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"), 500)); + EXPECT_CALL(callbacks_, sendLocalReply(_, _)) + .WillOnce(Invoke([&](const DirectResponse& response, bool end_stream) -> void { + auto& app_ex = dynamic_cast(response); + EXPECT_EQ(AppExceptionType::InternalError, app_ex.type_); + EXPECT_THAT(app_ex.what(), ContainsRegex(".*connection failure.*")); + EXPECT_TRUE(end_stream); + })); + context_.cluster_manager_.thread_local_cluster_.tcp_conn_pool_.poolFailure( + ConnectionPool::PoolFailureReason::Timeout); +} + TEST_P(ThriftRouterFieldTypeTest, OneWay) { FieldType field_type = GetParam(); @@ -948,10 +979,10 @@ TEST_P(ThriftRouterFieldTypeTest, OneWay) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_oneway") + .counterFromString("thrift.upstream_rq_oneway") .value()); EXPECT_EQ(0UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_reply") + .counterFromString("thrift.upstream_resp_reply") .value()); } @@ -967,16 +998,43 @@ TEST_P(ThriftRouterFieldTypeTest, Call) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_reply") + .counterFromString("thrift.upstream_resp_reply") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_success") + .counterFromString("thrift.upstream_resp_success") .value()); } +TEST_P(ThriftRouterFieldTypeTest, CallWithUpstreamRqTime) { + FieldType field_type = GetParam(); + + 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")); + EXPECT_CALL(cluster_scope, counter("thrift.upstream_resp_reply")); + EXPECT_CALL(cluster_scope, counter("thrift.upstream_resp_success")); + + startRequest(MessageType::Call); + connectUpstream(); + sendTrivialStruct(field_type); + completeRequest(); + + dispatcher_.time_system_.advanceTimeWait(std::chrono::milliseconds(500)); + 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"), 500)); + returnResponse(); + destroyRouter(); +} + TEST_P(ThriftRouterFieldTypeTest, Call_Error) { FieldType field_type = GetParam(); @@ -989,16 +1047,16 @@ TEST_P(ThriftRouterFieldTypeTest, Call_Error) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_reply") + .counterFromString("thrift.upstream_resp_reply") .value()); EXPECT_EQ(0UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_success") + .counterFromString("thrift.upstream_resp_success") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_error") + .counterFromString("thrift.upstream_resp_error") .value()); } @@ -1014,10 +1072,10 @@ TEST_P(ThriftRouterFieldTypeTest, Exception) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_exception") + .counterFromString("thrift.upstream_resp_exception") .value()); } @@ -1033,10 +1091,10 @@ TEST_P(ThriftRouterFieldTypeTest, UnknownMessageTypes) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_invalid_type") + .counterFromString("thrift.upstream_rq_invalid_type") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_invalid_type") + .counterFromString("thrift.upstream_resp_invalid_type") .value()); } @@ -1056,13 +1114,13 @@ TEST_P(ThriftRouterFieldTypeTest, StripServiceNameEnabled) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_reply") + .counterFromString("thrift.upstream_resp_reply") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_success") + .counterFromString("thrift.upstream_resp_success") .value()); } @@ -1082,13 +1140,13 @@ TEST_P(ThriftRouterFieldTypeTest, StripServiceNameDisabled) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_reply") + .counterFromString("thrift.upstream_resp_reply") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_success") + .counterFromString("thrift.upstream_resp_success") .value()); } @@ -1108,13 +1166,13 @@ TEST_F(ThriftRouterTest, CallWithExistingConnection) { destroyRouter(); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("request_call") + .counterFromString("thrift.upstream_rq_call") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_reply") + .counterFromString("thrift.upstream_resp_reply") .value()); EXPECT_EQ(1UL, context_.cluster_manager_.thread_local_cluster_.cluster_.info_->statsScope() - .counterFromString("response_success") + .counterFromString("thrift.upstream_resp_success") .value()); }