Skip to content

WiP admin: get basic streaming working for /stats #19649

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

Closed
jmarantz wants to merge 1 commit into from
Closed

WiP admin: get basic streaming working for /stats #19649

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
20 changes: 12 additions & 8 deletions source/server/admin/admin_filter.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,14 +68,18 @@ void AdminFilter::onComplete() {
Buffer::OwnedImpl response;
auto header_map = Http::ResponseHeaderMapImpl::create();
RELEASE_ASSERT(request_headers_, "");
Http::Code code = admin_server_callback_func_(path, *header_map, response, *this);
Utility::populateFallbackResponseHeaders(code, *header_map);
decoder_callbacks_->encodeHeaders(std::move(header_map),
end_stream_on_complete_ && response.length() == 0,
StreamInfo::ResponseCodeDetails::get().AdminFilterResponse);

if (response.length() > 0) {
decoder_callbacks_->encodeData(response, end_stream_on_complete_);
for (bool cont = true, first = true; cont; first = false) {
Copy link
Contributor Author

@jmarantz jmarantz Jan 21, 2022

Choose a reason for hiding this comment

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

This pattern here is deeply flawed on a couple of levels. This loops over all the data, streaming it out to the client. Consider the data could be 1G. That will block the main thread until all the data gets to the client.

It also will stuff this into the network without any notion of client flow-control.

Instead we should use the exiting Envoy filter state-machine stuff to stream out a chunk at a time and return control.

See

envoy/source/extensions/filters/http/decompressor/decompressor_filter.cc

Line 96 in 3d5f637

trailers = HeaderMapOptRef(std::ref(decoder_callbacks_->addDecodedTrailers()));
(gzip decompress filter) for something that I think is doing this right.

OTOH I tested this manually and it seems to "work" in the sense that all the data comes out and is received by 'wget' via chunked encoding.

Also what we really want here is create a Handler from the callback and save it in the filter, and then call the handler for each chunk whenever encodeData is called (I think). I am not too fluent in Envoy filter streaming anymore.

Http::Code code = admin_server_callback_func_(path, *header_map, response, *this);
cont = code == Http::Code::Continue;
if (first) {
Utility::populateFallbackResponseHeaders(cont ? Http::Code::OK : code, *header_map);
decoder_callbacks_->encodeHeaders(std::move(header_map),
end_stream_on_complete_ && response.length() == 0,
StreamInfo::ResponseCodeDetails::get().AdminFilterResponse);
}
if (response.length() > 0) {
decoder_callbacks_->encodeData(response, !cont && end_stream_on_complete_);
}
}
}

Expand Down
121 changes: 74 additions & 47 deletions source/server/admin/stats_handler.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,43 +90,65 @@ Http::Code StatsHandler::handlerStats(absl::string_view url,
return handlerPrometheusStats(url, response_headers, response, admin_stream);
}

std::map<std::string, uint64_t> all_stats;
for (const Stats::CounterSharedPtr& counter : server_.stats().counters()) {
if (shouldShowMetric(*counter, used_only, regex)) {
all_stats.emplace(counter->name(), counter->value());
auto iter = context_map_.find(&admin_stream);
if (iter == context_map_.end()) {
auto insertion = context_map_.emplace(&admin_stream, std::make_unique<Context>(
server_, used_only, regex, format_value));
iter = insertion.first;
}
Context& context = *iter->second;

Http::Code code = Http::Code::NotFound;
if (!format_value.has_value()) {
// Display plain stats if format query param is not there.
code = context.statsAsText(response_headers, response);
} else if (format_value.value() == "json") {
code = context.statsAsJson(response_headers, response, params.find("pretty") != params.end());
} else {
response.add("usage: /stats?format=json or /stats?format=prometheus \n");
response.add("\n");
}
if (code != Http::Code::Continue) {
context_map_.erase(iter);
}

return code;
}

StatsHandler::Context::Context(Server::Instance& server,
bool used_only, absl::optional<std::regex> regex,
absl::optional<std::string> format_value)
: used_only_(used_only), regex_(regex), format_value_(format_value) {
for (const Stats::CounterSharedPtr& counter : server.stats().counters()) {
if (shouldShowMetric(*counter)) {
all_stats_.emplace(counter->name(), counter->value());
}
}

for (const Stats::GaugeSharedPtr& gauge : server_.stats().gauges()) {
if (shouldShowMetric(*gauge, used_only, regex)) {
for (const Stats::GaugeSharedPtr& gauge : server.stats().gauges()) {
if (shouldShowMetric(*gauge)) {
ASSERT(gauge->importMode() != Stats::Gauge::ImportMode::Uninitialized);
all_stats.emplace(gauge->name(), gauge->value());
all_stats_.emplace(gauge->name(), gauge->value());
}
}
all_stats_iter_ = all_stats_.begin();

std::map<std::string, std::string> text_readouts;
for (const auto& text_readout : server_.stats().textReadouts()) {
if (shouldShowMetric(*text_readout, used_only, regex)) {
text_readouts.emplace(text_readout->name(), text_readout->value());
for (const auto& text_readout : server.stats().textReadouts()) {
if (shouldShowMetric(*text_readout)) {
text_readouts_.emplace(text_readout->name(), text_readout->value());
}
}
text_readouts_iter_ = text_readouts_.begin();

if (!format_value.has_value()) {
// Display plain stats if format query param is not there.
statsAsText(all_stats, text_readouts, server_.stats().histograms(), used_only, regex, response);
return Http::Code::OK;
}
histograms_ = server.stats().histograms();
}

if (format_value.value() == "json") {
response_headers.setReferenceContentType(Http::Headers::get().ContentTypeValues.Json);
response.add(
statsAsJson(all_stats, text_readouts, server_.stats().histograms(), used_only, regex));
return Http::Code::OK;
bool StatsHandler::Context::Context::nextChunk() {
if (++chunk_index_ == chunk_size_) {
chunk_index_= 0;
return true;
}

response.add("usage: /stats?format=json or /stats?format=prometheus \n");
response.add("\n");
return Http::Code::NotFound;
return false;
}

Http::Code StatsHandler::handlerPrometheusStats(absl::string_view path_and_query,
Expand Down Expand Up @@ -171,48 +193,52 @@ Http::Code StatsHandler::handlerContention(absl::string_view,
return Http::Code::OK;
}

void StatsHandler::statsAsText(const std::map<std::string, uint64_t>& all_stats,
const std::map<std::string, std::string>& text_readouts,
const std::vector<Stats::ParentHistogramSharedPtr>& histograms,
bool used_only, const absl::optional<std::regex>& regex,
Buffer::Instance& response) {
Http::Code StatsHandler::Context::statsAsText(Http::ResponseHeaderMap& /*response_headers*/,
Buffer::Instance& response) {
// Display plain stats if format query param is not there.
for (const auto& text_readout : text_readouts) {
for (; text_readouts_iter_ != text_readouts_.end(); ++text_readouts_iter_) {
if (nextChunk()) {
return Http::Code::Continue;
}
response.addFragments(
{text_readout.first, ": \"", Html::Utility::sanitize(text_readout.second), "\"\n"});
{text_readouts_iter_->first, ": \"", Html::Utility::sanitize(text_readouts_iter_->second),
"\"\n"});
}
for (const auto& stat : all_stats) {
response.addFragments({stat.first, ": ", absl::StrCat(stat.second), "\n"});
for (; all_stats_iter_ != all_stats_.end(); ++all_stats_iter_) {
if (nextChunk()) {
return Http::Code::Continue;
}
response.addFragments({all_stats_iter_->first, ": ", absl::StrCat(all_stats_iter_->second), "\n"});
}
std::map<std::string, std::string> all_histograms;
for (const Stats::ParentHistogramSharedPtr& histogram : histograms) {
if (shouldShowMetric(*histogram, used_only, regex)) {
for (const Stats::ParentHistogramSharedPtr& histogram : histograms_) {
if (shouldShowMetric(*histogram)) {
auto insert = all_histograms.emplace(histogram->name(), histogram->quantileSummary());
ASSERT(insert.second); // No duplicates expected.
}
}
for (const auto& histogram : all_histograms) {
response.addFragments({histogram.first, ": ", histogram.second, "\n"});
}
return Http::Code::OK;
}

std::string
StatsHandler::statsAsJson(const std::map<std::string, uint64_t>& all_stats,
const std::map<std::string, std::string>& text_readouts,
const std::vector<Stats::ParentHistogramSharedPtr>& all_histograms,
const bool used_only, const absl::optional<std::regex>& regex,
const bool pretty_print) {
Http::Code StatsHandler::Context::statsAsJson(Http::ResponseHeaderMap& response_headers,
Buffer::Instance& response,
const bool pretty_print) {

response_headers.setReferenceContentType(Http::Headers::get().ContentTypeValues.Json);

ProtobufWkt::Struct document;
std::vector<ProtobufWkt::Value> stats_array;
for (const auto& text_readout : text_readouts) {
for (const auto& text_readout : text_readouts_) {
ProtobufWkt::Struct stat_obj;
auto* stat_obj_fields = stat_obj.mutable_fields();
(*stat_obj_fields)["name"] = ValueUtil::stringValue(text_readout.first);
(*stat_obj_fields)["value"] = ValueUtil::stringValue(text_readout.second);
stats_array.push_back(ValueUtil::structValue(stat_obj));
}
for (const auto& stat : all_stats) {
for (const auto& stat : all_stats_) {
ProtobufWkt::Struct stat_obj;
auto* stat_obj_fields = stat_obj.mutable_fields();
(*stat_obj_fields)["name"] = ValueUtil::stringValue(stat.first);
Expand All @@ -228,8 +254,8 @@ StatsHandler::statsAsJson(const std::map<std::string, uint64_t>& all_stats,
std::vector<ProtobufWkt::Value> computed_quantile_array;

bool found_used_histogram = false;
for (const Stats::ParentHistogramSharedPtr& histogram : all_histograms) {
if (shouldShowMetric(*histogram, used_only, regex)) {
for (const Stats::ParentHistogramSharedPtr& histogram : histograms_) {
if (shouldShowMetric(*histogram)) {
if (!found_used_histogram) {
// It is not possible for the supported quantiles to differ across histograms, so it is ok
// to send them once.
Expand Down Expand Up @@ -274,7 +300,8 @@ StatsHandler::statsAsJson(const std::map<std::string, uint64_t>& all_stats,
auto* document_fields = document.mutable_fields();
(*document_fields)["stats"] = ValueUtil::listValue(stats_array);

return MessageUtil::getJsonStringFromMessageOrDie(document, pretty_print, true);
response.add(MessageUtil::getJsonStringFromMessageOrDie(document, pretty_print, true));
return Http::Code::OK;
}

} // namespace Server
Expand Down
42 changes: 31 additions & 11 deletions source/server/admin/stats_handler.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
#include "source/common/stats/histogram_impl.h"
#include "source/server/admin/handler_ctx.h"

#include "absl/container/flat_hash_map.h"
#include "absl/strings/string_view.h"

namespace Envoy {
Expand Down Expand Up @@ -47,6 +48,35 @@ class StatsHandler : public HandlerContextBase {
Http::ResponseHeaderMap& response_headers,
Buffer::Instance& response, AdminStream&);

class Context {
public:
Context(Server::Instance& server,
bool used_only, absl::optional<std::regex> regex,
absl::optional<std::string> format_value);

bool nextChunk();

template<class StatType> bool shouldShowMetric(const StatType& stat) {
return StatsHandler::shouldShowMetric(stat, used_only_, regex_);
}

Http::Code statsAsText(Http::ResponseHeaderMap& response_headers, Buffer::Instance& response);
Http::Code statsAsJson(Http::ResponseHeaderMap& response_headers, Buffer::Instance& response,
bool pretty);

const bool used_only_;
absl::optional<std::regex> regex_;
absl::optional<std::string> format_value_;
std::map<std::string, uint64_t> all_stats_;
std::map<std::string, uint64_t>::iterator all_stats_iter_;
std::map<std::string, std::string> text_readouts_;
std::map<std::string, std::string>::iterator text_readouts_iter_;
std::vector<Stats::ParentHistogramSharedPtr> histograms_;
static constexpr uint32_t chunk_size_ = 10;
uint32_t chunk_index_{0};;
};
using ContextPtr = std::unique_ptr<Context>;

private:
template <class StatType>
static bool shouldShowMetric(const StatType& metric, const bool used_only,
Expand All @@ -57,17 +87,7 @@ class StatsHandler : public HandlerContextBase {

friend class StatsHandlerTest;

static std::string statsAsJson(const std::map<std::string, uint64_t>& all_stats,
const std::map<std::string, std::string>& text_readouts,
const std::vector<Stats::ParentHistogramSharedPtr>& all_histograms,
bool used_only, const absl::optional<std::regex>& regex,
bool pretty_print = false);

void statsAsText(const std::map<std::string, uint64_t>& all_stats,
const std::map<std::string, std::string>& text_readouts,
const std::vector<Stats::ParentHistogramSharedPtr>& all_histograms,
bool used_only, const absl::optional<std::regex>& regex,
Buffer::Instance& response);
absl::flat_hash_map<AdminStream*, std::unique_ptr<Context>> context_map_;
};

} // namespace Server
Expand Down