-
Notifications
You must be signed in to change notification settings - Fork 577
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.
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
Metrics Cardinality Reduction: PoC #4843
Closed
BenPope
wants to merge
1
commit into
redpanda-data:dev
from
BenPope:metrics-cardinality-reduction-poc
Closed
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -3,6 +3,8 @@ v_cc_library( | |
ssx | ||
HDRS | ||
"future-util.h" | ||
SRCS | ||
metrics.cc | ||
DEPS | ||
Seastar::seastar | ||
) | ||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,141 @@ | ||
/* | ||
* Copyright 2022 Redpanda Data, Inc. | ||
* | ||
* Use of this software is governed by the Business Source License | ||
* included in the file licenses/BSL.md | ||
* | ||
* As of the Change Date specified in that file, in accordance with | ||
* the Business Source License, use of this software will be governed | ||
* by the Apache License, Version 2.0 | ||
*/ | ||
|
||
#include "ssx/metrics.h" | ||
|
||
#include <seastar/core/metrics_api.hh> | ||
#include <seastar/core/metrics_registration.hh> | ||
|
||
#include <absl/algorithm/container.h> | ||
|
||
#include <numeric> | ||
#include <optional> | ||
#include <tuple> | ||
#include <vector> | ||
|
||
namespace ssx::metrics { | ||
|
||
namespace ssm = ss::metrics; | ||
namespace ssmi = ssm::impl; | ||
using agg_op = ssx::metrics::filter::config::agg_op; | ||
using registered_instances = std::vector<ssmi::register_ref>; | ||
// Store a mapping of allowed labels to aggregated registered_metrics | ||
using metric_multi_instance = std::map<ssmi::labels_type, registered_instances>; | ||
|
||
ssmi::metric_function get_op(agg_op op, registered_instances refs) { | ||
switch (op) { | ||
case agg_op::sum: | ||
return [refs{std::move(refs)}]() { | ||
auto type = refs.front()->get_function()().type(); | ||
return absl::c_accumulate( | ||
refs, ssmi::metric_value(0, type), [](auto lhs, auto rhs) { | ||
return lhs + rhs->get_function()(); | ||
}); | ||
}; | ||
case agg_op::count: | ||
return [s = static_cast<double>(refs.size())]() { | ||
return ssmi::metric_value(s, ssmi::data_type::ABSOLUTE); | ||
}; | ||
case agg_op::min: | ||
return [refs{std::move(refs)}]() { | ||
auto it = absl::c_min_element(refs, [](auto lhs, auto rhs) { | ||
return lhs->get_function()().d() < rhs->get_function()().d(); | ||
}); | ||
return (*it)->get_function()(); | ||
}; | ||
case agg_op::max: | ||
return [refs{std::move(refs)}]() { | ||
auto it = absl::c_max_element(refs, [](auto lhs, auto rhs) { | ||
return lhs->get_function()().d() < rhs->get_function()().d(); | ||
}); | ||
return (*it)->get_function()(); | ||
}; | ||
case agg_op::avg: | ||
return [refs{std::move(refs)}]() { | ||
auto type = refs.front()->get_function()().type(); | ||
auto sum = absl::c_accumulate( | ||
refs, ssmi::metric_value{0, type}, [](auto lhs, auto rhs) { | ||
return lhs + rhs->get_function()(); | ||
}); | ||
return ssmi::metric_value{ | ||
sum.d() / static_cast<double>(refs.size()), type}; | ||
}; | ||
}; | ||
} | ||
|
||
ss::metrics::impl::value_map get_filtered(const filter::config& cfg) { | ||
// The structure looks like this: | ||
// ssmi::value_map exemplar{ | ||
// {"test_fetch_metrics_counter", | ||
// ssmi::metric_family{ | ||
// ssmi::metric_instances{ | ||
// {ssmi::labels_type{ | ||
// {"aggregate", "0"}, {"leave", "0"}, {"shard", "0"}}, | ||
// ssmi::register_ref{}}, | ||
// {ssmi::labels_type{ | ||
// {"aggregate", "1"}, {"leave", "0"}, {"shard", "0"}}, | ||
// ssmi::register_ref{}}}, | ||
// ssmi::metric_family_info{ | ||
// ssmi::data_type::COUNTER, | ||
// ssm::metric_type_def{""}, | ||
// ssm::description{"Test a counter"}, | ||
// ss::sstring{"test_fetch_metrics_counter"}}}}}; | ||
|
||
// Collect metric families | ||
ssmi::value_map result; | ||
for (const auto& [name, family] : ssmi::get_value_map()) { | ||
auto m_it = absl::c_find_if( | ||
cfg.allow, [&n = name](const auto& m) { return m.name == n; }); | ||
|
||
// Ignore metrics that are not allowed. | ||
if (m_it == cfg.allow.end()) { | ||
continue; | ||
} | ||
|
||
const auto& metric_cfg{*m_it}; | ||
|
||
// Collect aggregated instances | ||
metric_multi_instance new_instances; | ||
for (const auto& [base_labels, base_metric] : family) { | ||
ssmi::labels_type new_labels; | ||
for (const auto& [l_name, l_value] : base_labels) { | ||
auto l_it = absl::c_find_if( | ||
metric_cfg.allow, [lhs = l_name](const auto& rhs) { | ||
return lhs == rhs.name(); | ||
}); | ||
if (l_it != metric_cfg.allow.end()) { | ||
new_labels.emplace(l_name, l_value); | ||
} | ||
} | ||
new_instances[new_labels].emplace_back(base_metric); | ||
} | ||
|
||
// Convert aggregated instances to registred metrics | ||
ssmi::metric_instances family_instances; | ||
for (auto& ni : new_instances) { | ||
auto id = ni.second.front()->get_id(); | ||
auto op = metric_cfg.op.value_or(agg_op::sum); | ||
|
||
family_instances.emplace( | ||
ni.first, | ||
ss::make_shared<ssmi::registered_metric>( | ||
std::move(id), get_op(op, std::move(ni.second)))); | ||
} | ||
auto family_info = family.info(); | ||
result.emplace( | ||
name, | ||
ssmi::metric_family{ | ||
std::move(family_instances), std::move(family_info)}); | ||
} | ||
return result; | ||
} | ||
|
||
} // namespace ssx::metrics |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,47 @@ | ||
/* | ||
* Copyright 2022 Redpanda Data, Inc. | ||
* | ||
* Use of this software is governed by the Business Source License | ||
* included in the file licenses/BSL.md | ||
* | ||
* As of the Change Date specified in that file, in accordance with | ||
* the Business Source License, use of this software will be governed | ||
* by the Apache License, Version 2.0 | ||
*/ | ||
|
||
#pragma once | ||
|
||
#include "seastarx.h" | ||
|
||
#include <seastar/core/metrics_api.hh> | ||
|
||
#include <optional> | ||
#include <vector> | ||
|
||
namespace ssx::metrics { | ||
|
||
namespace filter { | ||
|
||
struct config { | ||
enum class agg_op { | ||
sum, | ||
count, | ||
min, | ||
max, | ||
avg, | ||
}; | ||
|
||
struct metric { | ||
ss::sstring name; | ||
std::vector<ss::metrics::label> allow; | ||
std::optional<agg_op> op; | ||
}; | ||
|
||
std::vector<metric> allow; | ||
}; | ||
|
||
} // namespace filter | ||
|
||
ss::metrics::impl::value_map get_filtered(const filter::config& cfg); | ||
|
||
} // namespace ssx::metrics |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,136 @@ | ||
// Copyright 2022 Redpanda Data, Inc. | ||
// | ||
// Use of this software is governed by the Business Source License | ||
// included in the file licenses/BSL.md | ||
// | ||
// As of the Change Date specified in that file, in accordance with | ||
// the Business Source License, use of this software will be governed | ||
// by the Apache License, Version 2.0 | ||
|
||
#include "ssx/metrics.h" | ||
|
||
#include "prometheus/prometheus_sanitize.h" | ||
#include "seastarx.h" | ||
|
||
#include <seastar/core/metrics.hh> | ||
#include <seastar/testing/thread_test_case.hh> | ||
|
||
#include <ostream> | ||
|
||
namespace sm = ss::metrics; | ||
namespace ssmi = sm::impl; | ||
namespace ssxm = ssx::metrics; | ||
namespace ssxmf = ssxm::filter; | ||
using agg_op = ssx::metrics::filter::config::agg_op; | ||
|
||
SEASTAR_THREAD_TEST_CASE(test_fetch_metrics) { | ||
uint64_t counter_0{17}; | ||
uint64_t counter_1{19}; | ||
uint64_t counter_2{23}; | ||
sm::metric_groups metrics; | ||
|
||
sm::label label_ag{"aggregate"}; | ||
sm::label label_leave{"leave"}; | ||
|
||
metrics.add_group( | ||
prometheus_sanitize::metrics_name("test_fetch_metrics"), | ||
{ | ||
sm::make_counter( | ||
"counter", | ||
[&] { return counter_0; }, | ||
sm::description("Test a counter"), | ||
{ | ||
label_ag(0), | ||
label_leave(0), | ||
}), | ||
sm::make_counter( | ||
"counter", | ||
[&] { return counter_1; }, | ||
sm::description("Test a counter"), | ||
{ | ||
label_ag(1), | ||
label_leave(0), | ||
}), | ||
sm::make_counter( | ||
"counter", | ||
[&] { return counter_2; }, | ||
sm::description("Test a counter"), | ||
{ | ||
label_ag(1), | ||
label_leave(1), | ||
}), | ||
}); | ||
|
||
auto collect_values = [](const ssmi::value_map& metrics) { | ||
std::map<ssmi::labels_type, uint64_t> values; | ||
for (const auto& [n, m] : metrics) { | ||
for (const auto& [n, o] : m) { | ||
values[n] = o->get_function()().i(); | ||
} | ||
} | ||
return values; | ||
}; | ||
|
||
// Check registred metrics | ||
auto values = collect_values(ss::metrics::impl::get_value_map()); | ||
auto expected_0 = ssmi::labels_type{ | ||
{label_ag.name(), "0"}, | ||
{label_leave.name(), "0"}, | ||
{sm::shard_label.name(), "0"}}; | ||
|
||
auto expected_1 = ssmi::labels_type{ | ||
{label_ag.name(), "1"}, | ||
{label_leave.name(), "0"}, | ||
{sm::shard_label.name(), "0"}}; | ||
|
||
auto expected_2 = ssmi::labels_type{ | ||
{label_ag.name(), "1"}, | ||
{label_leave.name(), "1"}, | ||
{sm::shard_label.name(), "0"}}; | ||
|
||
BOOST_CHECK_EQUAL(values[expected_0], 17); | ||
BOOST_CHECK_EQUAL(values[expected_1], 19); | ||
BOOST_CHECK_EQUAL(values[expected_2], 23); | ||
|
||
// Check summed metrics | ||
ssxmf::config config{{{"test_fetch_metrics_counter", {label_leave}}}}; | ||
values = collect_values(ssxm::get_filtered(config)); | ||
|
||
BOOST_CHECK_EQUAL(values.size(), 2); | ||
auto leave_0 = ssmi::labels_type{{label_leave.name(), "0"}}; | ||
auto leave_1 = ssmi::labels_type{{label_leave.name(), "1"}}; | ||
BOOST_CHECK_EQUAL(values[leave_0], 36); | ||
BOOST_CHECK_EQUAL(values[leave_1], 23); | ||
|
||
// Check count metrics | ||
config = {{{"test_fetch_metrics_counter", {label_leave}, agg_op::count}}}; | ||
values = collect_values(ssxm::get_filtered(config)); | ||
|
||
BOOST_CHECK_EQUAL(values.size(), 2); | ||
BOOST_CHECK_EQUAL(values[leave_0], 2); | ||
BOOST_CHECK_EQUAL(values[leave_1], 1); | ||
|
||
// Check min metrics | ||
config = {{{"test_fetch_metrics_counter", {label_leave}, agg_op::min}}}; | ||
values = collect_values(ssxm::get_filtered(config)); | ||
|
||
BOOST_CHECK_EQUAL(values.size(), 2); | ||
BOOST_CHECK_EQUAL(values[leave_0], 17); | ||
BOOST_CHECK_EQUAL(values[leave_1], 23); | ||
|
||
// Check max metrics | ||
config = {{{"test_fetch_metrics_counter", {label_leave}, agg_op::max}}}; | ||
values = collect_values(ssxm::get_filtered(config)); | ||
|
||
BOOST_CHECK_EQUAL(values.size(), 2); | ||
BOOST_CHECK_EQUAL(values[leave_0], 19); | ||
BOOST_CHECK_EQUAL(values[leave_1], 23); | ||
|
||
// Check avg metrics | ||
config = {{{"test_fetch_metrics_counter", {label_leave}, agg_op::avg}}}; | ||
values = collect_values(ssxm::get_filtered(config)); | ||
|
||
BOOST_CHECK_EQUAL(values.size(), 2); | ||
BOOST_CHECK_EQUAL(values[leave_0], 18); | ||
BOOST_CHECK_EQUAL(values[leave_1], 23); | ||
} |
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There is an efficiency issue here for large partition counts: we'd be iterating over ~10E5 partition labels and summing them all up, plus the allocation overhead of building these intermediate vectors of metrics.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There are two phases: