Skip to content

thrift proxy: add upstream_rq_time histograms to cluster metrics #15884

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zuercher merged 5 commits into from
Apr 14, 2021
Merged

thrift proxy: add upstream_rq_time histograms to cluster metrics #15884

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
684469b
thrift proxy: add upstream_rq_time histograms to cluster metrics
Apr 8, 2021
cd80842
Update docs and add boolean to avoid double reporting
Apr 8, 2021
6a16b93
Fix glint
Apr 9, 2021
2ed8cb9
Avoid concatenation in stat, move around chargeResponseTiming into
Apr 12, 2021
3f4a6d7
Add line to version log rst
Apr 13, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 11 additions & 7 deletions docs/root/configuration/other_protocols/thrift_filters/router_filter.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,15 +25,19 @@ The filter outputs generic routing error statistics in the *thrift.<stat_prefix>
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 clusters.

@rgs1 rgs1 Apr 8, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: derived from 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.
4 changes: 4 additions & 0 deletions source/extensions/filters/network/thrift_proxy/conn_manager.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(); }
Comment thread
williamsfu99 marked this conversation as resolved.
void continueDecoding() override;
Router::RouteConstSharedPtr route() override { return parent_.route(); }
TransportType downstreamTransportType() const override {
Expand Down Expand Up @@ -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 {
Expand Down
5 changes: 5 additions & 0 deletions source/extensions/filters/network/thrift_proxy/filters/filter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
40 changes: 31 additions & 9 deletions source/extensions/filters/network/thrift_proxy/router/router_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,15 +241,15 @@ FilterStatus Router::messageBegin(MessageMetadataSharedPtr metadata) {
metadata->methodName());
switch (metadata->messageType()) {
case MessageType::Call:
incClusterScopeCounter(request_call_);
incClusterScopeCounter({thrift_, upstream_rq_call_});
break;

case MessageType::Oneway:
incClusterScopeCounter(request_oneway_);
incClusterScopeCounter({thrift_, upstream_rq_oneway_});
break;

default:
incClusterScopeCounter(request_invalid_type_);
incClusterScopeCounter({thrift_, upstream_rq_invalid_type_});
break;
}

Expand Down Expand Up @@ -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({thrift_, upstream_resp_reply_});
if (callbacks_->responseSuccess()) {
incClusterScopeCounter(response_reply_success_);
incClusterScopeCounter({thrift_, upstream_resp_reply_success_});
} else {
incClusterScopeCounter(response_reply_error_);
incClusterScopeCounter({thrift_, upstream_resp_reply_error_});
}
break;

case MessageType::Exception:
incClusterScopeCounter(response_exception_);
incClusterScopeCounter({thrift_, upstream_resp_exception_});
break;

default:
incClusterScopeCounter(response_invalid_type_);
incClusterScopeCounter({thrift_, upstream_resp_invalid_type_});
break;
}
upstream_request_->onResponseComplete();
Expand Down Expand Up @@ -522,9 +522,22 @@ 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() {
if (!charged_response_timing_) {
Event::Dispatcher& dispatcher = parent_.callbacks_->dispatcher();
const std::chrono::milliseconds response_time =
std::chrono::duration_cast<std::chrono::milliseconds>(
dispatcher.timeSource().monotonicTime() - downstream_request_complete_time_);
parent_.chargeResponseTiming(response_time);
charged_response_timing_ = true;
}
Comment thread
williamsfu99 marked this conversation as resolved.
Outdated

response_complete_ = true;
conn_state_ = nullptr;
conn_data_.reset();
Expand All @@ -542,6 +555,15 @@ void Router::UpstreamRequest::onResetStream(ConnectionPool::PoolFailureReason re
return;
}

if (!charged_response_timing_) {
Event::Dispatcher& dispatcher = parent_.callbacks_->dispatcher();
const std::chrono::milliseconds response_time =
std::chrono::duration_cast<std::chrono::milliseconds>(
dispatcher.timeSource().monotonicTime() - downstream_request_complete_time_);
parent_.chargeResponseTiming(response_time);
charged_response_timing_ = true;
}

switch (reason) {
case ConnectionPool::PoolFailureReason::Overflow:
parent_.callbacks_->sendLocalReply(
Expand Down
60 changes: 41 additions & 19 deletions source/extensions/filters/network/thrift_proxy/router/router_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,15 +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")),
passthrough_supported_(false) {}
symbol_table_(scope.symbolTable()), thrift_(stat_name_set_->add("thrift")),
upstream_rq_call_(stat_name_set_->add("upstream_rq_call")),
upstream_rq_oneway_(stat_name_set_->add("upstream_rq_oneway")),
upstream_rq_invalid_type_(stat_name_set_->add("upstream_rq_invalid_type")),
upstream_resp_reply_(stat_name_set_->add("upstream_resp_reply")),
upstream_resp_reply_success_(stat_name_set_->add("upstream_resp_success")),
upstream_resp_reply_error_(stat_name_set_->add("upstream_resp_error")),
upstream_resp_exception_(stat_name_set_->add("upstream_resp_exception")),
upstream_resp_invalid_type_(stat_name_set_->add("upstream_resp_invalid_type")),
upstream_rq_time_(stat_name_set_->add("upstream_rq_time")), passthrough_supported_(false) {}

@rgs1 rgs1 Apr 8, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if we should just add the thrift. prefix to every one of these stats here, to avoid the runtime concat of these two stats... [ I see it done for some of the TLS code... ]

@williamsfu99 williamsfu99 Apr 12, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see both ways being done in the repo

@rgs1 rgs1 Apr 12, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah ok, np then.


~Router() override = default;

Expand All @@ -198,8 +199,23 @@ class Router : public Tcp::ConnectionPool::UpstreamCallbacks,
bool passthroughSupported() const override { return passthrough_supported_; }

// Stats
void incClusterScopeCounter(Stats::StatName name) {
cluster_->statsScope().counterFromStatName(name).inc();
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 chargeResponseTiming(const std::chrono::milliseconds response_time) const {
const uint64_t count = response_time.count();
recordClusterScopeHistogram({thrift_, upstream_rq_time_}, Stats::Histogram::Unit::Milliseconds,
count);
}

// ProtocolConverter
Expand Down Expand Up @@ -261,6 +277,9 @@ 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_;
};

void convertMessageBegin(MessageMetadataSharedPtr metadata);
Expand All @@ -274,14 +293,17 @@ 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 thrift_;
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_{};
Expand Down
1 change: 1 addition & 0 deletions test/extensions/filters/network/thrift_proxy/mocks.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));
Expand Down
Loading