You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pegasus.apache.org by wa...@apache.org on 2023/04/06 04:08:59 UTC
[incubator-pegasus] branch migrate-metrics-dev updated (b6cc2ac50 -> a8cd54564)
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a change to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
discard b6cc2ac50 feat(new_metrics): add disk-level metric entity and migrate disk-level metrics for fs_manager (#1427)
omit dcecbe119 feat(new_metrics): migrate server-level metrics for nfs (#1421)
omit 9023f41d4 feat(new_metrics): migrate built-in server-level metrics (#1418)
omit e3bd72219 feat(new_metrics): add server-level metric entity (#1415)
omit a87b36b32 feat(new_metrics): migrate replica-level metrics for pegasus_mutation_duplicator (#1413)
omit fce08466b feat(new_metrics): migrate replica-level metrics for pegasus_event_listener (#1407)
omit 2836f56ea feat(new_metrics): migrate replica-level metrics for replica class (#1397)
omit bc7cb0511 feat(new_metrics): migrate replica-level metrics for capacity_unit_calculator (#1387)
omit 1109fd04c feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 2) (#1386)
omit 91431d704 feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1) (#1374)
omit e61a13ae7 feat(new_metrics): migrate replica-level metrics for write service (#1351)
omit 5b3d7d4b9 feat(new_metrics): add replica-level metric entity (#1345)
add 039e2fd3f refactor: minor refactor on replica module (#1423)
add 87f169a80 feat(Ranger): Use Apache Ranger for meta access controller (#1402)
add f630ce148 feat: add new macro definitions (#1429)
add 0199edee8 fix: log error but not crash if found an imcomplete replica path (#1428)
add ababcc08a fix(java-client): Fix java spotbug (#1416)
new 7bf18e6c5 feat(new_metrics): add replica-level metric entity (#1345)
new 7dfb98825 feat(new_metrics): migrate replica-level metrics for write service (#1351)
new fcae8db24 feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1) (#1374)
new 50608de53 feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 2) (#1386)
new 9d1a5665e feat(new_metrics): migrate replica-level metrics for capacity_unit_calculator (#1387)
new 0848525a6 feat(new_metrics): migrate replica-level metrics for replica class (#1397)
new de20d0f8f feat(new_metrics): migrate replica-level metrics for pegasus_event_listener (#1407)
new 121da7c51 feat(new_metrics): migrate replica-level metrics for pegasus_mutation_duplicator (#1413)
new 380ede5d9 feat(new_metrics): add server-level metric entity (#1415)
new a6664d6e4 feat(new_metrics): migrate built-in server-level metrics (#1418)
new 6d17dc4f1 feat(new_metrics): migrate server-level metrics for nfs (#1421)
new a8cd54564 feat(new_metrics): add disk-level metric entity and migrate disk-level metrics for fs_manager (#1427)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (b6cc2ac50)
\
N -- N -- N refs/heads/migrate-metrics-dev (a8cd54564)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 12 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docker/pegasus-build-env/centos6/Dockerfile | 2 +-
java-client/dev-support/findbugs-exclude.xml | 2 +
java-client/pom.xml | 14 +-
src/common/fs_manager.cpp | 3 +-
src/common/fs_manager.h | 2 +-
src/geo/lib/geo_client.cpp | 5 +-
src/meta/meta_backup_service.cpp | 32 ++++-
src/meta/meta_service.cpp | 153 +++++++++++++--------
src/meta/meta_service.h | 99 ++++++++++---
src/meta/server_state.cpp | 10 +-
src/meta/server_state.h | 3 +-
src/meta/server_state_restore.cpp | 43 +++---
src/meta/test/meta_service_test.cpp | 9 +-
src/meta/test/meta_service_test_app.h | 5 +-
src/redis_protocol/proxy_lib/redis_parser.cpp | 21 +--
src/replica/CMakeLists.txt | 1 +
src/replica/backup/test/CMakeLists.txt | 2 +-
src/replica/duplication/mutation_batch.h | 2 +-
src/replica/prepare_list.h | 3 +-
src/replica/replica_learn.cpp | 26 ++--
src/replica/replica_stub.cpp | 52 ++++---
src/replica/replica_stub.h | 11 +-
src/replica/replication_app_base.cpp | 5 +-
src/replica/replication_app_base.h | 7 +-
src/replica/split/replica_split_manager.cpp | 6 +-
src/replica/test/mock_utils.h | 9 +-
src/replica/test/replica_test.cpp | 9 +-
.../ranger/ranger_resource_policy_manager.cpp | 113 +++++++++++----
.../ranger/ranger_resource_policy_manager.h | 28 ++--
src/runtime/security/access_controller.cpp | 5 +-
src/runtime/security/access_controller.h | 7 +-
src/runtime/security/meta_access_controller.cpp | 99 +++++++++++--
src/runtime/security/meta_access_controller.h | 16 ++-
src/runtime/test/meta_access_controller_test.cpp | 15 +-
.../test/ranger_resource_policy_manager_test.cpp | 126 +++++++++++++++--
src/shell/CMakeLists.txt | 1 +
src/utils/errors.h | 12 +-
src/utils/fmt_logging.h | 14 ++
38 files changed, 698 insertions(+), 274 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 08/12: feat(new_metrics): migrate replica-level metrics for pegasus_mutation_duplicator (#1413)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit 121da7c5137af002de5d38e3e9851df81ade7387
Author: Dan Wang <wa...@apache.org>
AuthorDate: Fri Mar 24 16:06:02 2023 +0800
feat(new_metrics): migrate replica-level metrics for pegasus_mutation_duplicator (#1413)
https://github.com/apache/incubator-pegasus/issues/1412
Migrate replica-level metrics in pegasus_mutation_duplicator class to new
framework, including the numbers of successful/failed DUPLICATE requests
sent from mutation duplicator.
---
src/server/pegasus_mutation_duplicator.cpp | 36 ++++++++++++++----------------
src/server/pegasus_mutation_duplicator.h | 6 ++---
2 files changed, 20 insertions(+), 22 deletions(-)
diff --git a/src/server/pegasus_mutation_duplicator.cpp b/src/server/pegasus_mutation_duplicator.cpp
index e9295af35..8d87ce60b 100644
--- a/src/server/pegasus_mutation_duplicator.cpp
+++ b/src/server/pegasus_mutation_duplicator.cpp
@@ -19,14 +19,11 @@
#include "pegasus_mutation_duplicator.h"
-#include <fmt/core.h>
-#include <fmt/ostream.h>
#include <pegasus/error.h>
#include <sys/types.h>
#include <chrono>
#include <cstdint>
#include <functional>
-#include <iosfwd>
#include <memory>
#include <tuple>
#include <utility>
@@ -34,16 +31,15 @@
#include "client_lib/pegasus_client_impl.h"
#include "common/duplication_common.h"
-#include "common/gpid.h"
#include "duplication_internal_types.h"
#include "pegasus/client.h"
#include "pegasus_key_schema.h"
-#include "perf_counter/perf_counter.h"
#include "rrdb/rrdb.code.definition.h"
#include "rrdb/rrdb_types.h"
#include "runtime/message_utils.h"
#include "runtime/rpc/rpc_message.h"
#include "server/pegasus_write_service.h"
+#include "utils/autoref_ptr.h"
#include "utils/blob.h"
#include "utils/chrono_literals.h"
#include "utils/error_code.h"
@@ -51,6 +47,16 @@
#include "utils/fmt_logging.h"
#include "utils/rand.h"
+METRIC_DEFINE_counter(replica,
+ successful_mutation_dup_requests,
+ dsn::metric_unit::kRequests,
+ "The number of successful DUPLICATE requests sent from mutation duplicator");
+
+METRIC_DEFINE_counter(replica,
+ failed_mutation_dup_requests,
+ dsn::metric_unit::kRequests,
+ "The number of failed DUPLICATE requests sent from mutation duplicator");
+
namespace dsn {
namespace replication {
struct replica_base;
@@ -99,7 +105,10 @@ using namespace dsn::literals::chrono_literals;
pegasus_mutation_duplicator::pegasus_mutation_duplicator(dsn::replication::replica_base *r,
dsn::string_view remote_cluster,
dsn::string_view app)
- : mutation_duplicator(r), _remote_cluster(remote_cluster)
+ : mutation_duplicator(r),
+ _remote_cluster(remote_cluster),
+ METRIC_VAR_INIT_replica(successful_mutation_dup_requests),
+ METRIC_VAR_INIT_replica(failed_mutation_dup_requests)
{
// initialize pegasus-client when this class is first time used.
static __attribute__((unused)) bool _dummy = pegasus_client_factory::initialize(nullptr);
@@ -123,17 +132,6 @@ pegasus_mutation_duplicator::pegasus_mutation_duplicator(dsn::replication::repli
// never possible to duplicate data to itself
CHECK_NE_PREFIX_MSG(
get_current_cluster_id(), _remote_cluster_id, "invalid remote cluster: {}", remote_cluster);
-
- std::string str_gpid = fmt::format("{}", get_gpid());
- _shipped_ops.init_app_counter("app.pegasus",
- fmt::format("dup_shipped_ops@{}", str_gpid).c_str(),
- COUNTER_TYPE_RATE,
- "the total ops of DUPLICATE requests sent from this app");
- _failed_shipping_ops.init_app_counter(
- "app.pegasus",
- fmt::format("dup_failed_shipping_ops@{}", str_gpid).c_str(),
- COUNTER_TYPE_RATE,
- "the qps of failed DUPLICATE requests sent from this app");
}
void pegasus_mutation_duplicator::send(uint64_t hash, callback cb)
@@ -164,7 +162,7 @@ void pegasus_mutation_duplicator::on_duplicate_reply(uint64_t hash,
}
if (perr != PERR_OK || err != dsn::ERR_OK) {
- _failed_shipping_ops->increment();
+ METRIC_VAR_INCREMENT(failed_mutation_dup_requests);
// randomly log the 1% of the failed duplicate rpc, because minor number of
// errors are acceptable.
@@ -177,7 +175,7 @@ void pegasus_mutation_duplicator::on_duplicate_reply(uint64_t hash,
// duplicating an illegal write to server is unacceptable, fail fast.
CHECK_NE_PREFIX_MSG(perr, PERR_INVALID_ARGUMENT, rpc.response().error_hint);
} else {
- _shipped_ops->increment();
+ METRIC_VAR_INCREMENT(successful_mutation_dup_requests);
_total_shipped_size +=
rpc.dsn_request()->header->body_length + rpc.dsn_request()->header->hdr_length;
}
diff --git a/src/server/pegasus_mutation_duplicator.h b/src/server/pegasus_mutation_duplicator.h
index 2107cda5b..9a5aa086c 100644
--- a/src/server/pegasus_mutation_duplicator.h
+++ b/src/server/pegasus_mutation_duplicator.h
@@ -25,13 +25,13 @@
#include <map>
#include <string>
-#include "perf_counter/perf_counter_wrapper.h"
#include "replica/duplication/mutation_duplicator.h"
#include "rrdb/rrdb.client.h"
#include "runtime/pipeline.h"
#include "runtime/task/task_code.h"
#include "runtime/task/task_tracker.h"
#include "utils/chrono_literals.h"
+#include "utils/metrics.h"
#include "utils/string_view.h"
#include "utils/zlocks.h"
@@ -89,8 +89,8 @@ private:
size_t _total_shipped_size{0};
- dsn::perf_counter_wrapper _shipped_ops;
- dsn::perf_counter_wrapper _failed_shipping_ops;
+ METRIC_VAR_DECLARE_counter(successful_mutation_dup_requests);
+ METRIC_VAR_DECLARE_counter(failed_mutation_dup_requests);
};
// Decodes the binary `request_data` into write request in thrift struct, and
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 04/12: feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 2) (#1386)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit 50608de5388241bab7ecd16744bfbc24c504fb05
Author: Dan Wang <wa...@apache.org>
AuthorDate: Fri Mar 10 17:05:59 2023 +0800
feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 2) (#1386)
This PR is to migrate replica-level metrics of pegasus_server_impl to new framework,
2nd part, for #1333.
This PR focuses on migrating all rocksdb-related metrics for each replica, including
total number and size of sst files, estimated number of keys, memory usage and hit
rate, write/read amplification, negatives/positives of bloom filters.
---
src/server/pegasus_server_impl.cpp | 150 ++++++-----------
src/server/pegasus_server_impl.h | 48 +++---
src/server/pegasus_server_impl_init.cpp | 275 ++++++++++++++++++--------------
src/utils/metrics.h | 12 +-
4 files changed, 238 insertions(+), 247 deletions(-)
diff --git a/src/server/pegasus_server_impl.cpp b/src/server/pegasus_server_impl.cpp
index c9bd093b3..38642c5f9 100644
--- a/src/server/pegasus_server_impl.cpp
+++ b/src/server/pegasus_server_impl.cpp
@@ -354,6 +354,15 @@ void pegasus_server_impl::log_expired_data(const char *op,
} \
} while (0)
+#define CHECK_READ_THROTTLING() \
+ do { \
+ if (dsn_unlikely(!_read_size_throttling_controller->available())) { \
+ rpc.error() = dsn::ERR_BUSY; \
+ METRIC_VAR_INCREMENT(throttling_rejected_read_requests); \
+ return; \
+ } \
+ } while (0)
+
void pegasus_server_impl::on_get(get_rpc rpc)
{
CHECK_TRUE(_is_open);
@@ -365,11 +374,7 @@ void pegasus_server_impl::on_get(get_rpc rpc)
resp.partition_index = _gpid.get_partition_index();
resp.server = _primary_address;
- if (!_read_size_throttling_controller->available()) {
- rpc.error() = dsn::ERR_BUSY;
- _counter_recent_read_throttling_reject_count->increment();
- return;
- }
+ CHECK_READ_THROTTLING();
METRIC_VAR_AUTO_LATENCY(get_latency_ns);
@@ -442,11 +447,7 @@ void pegasus_server_impl::on_multi_get(multi_get_rpc rpc)
resp.partition_index = _gpid.get_partition_index();
resp.server = _primary_address;
- if (!_read_size_throttling_controller->available()) {
- rpc.error() = dsn::ERR_BUSY;
- _counter_recent_read_throttling_reject_count->increment();
- return;
- }
+ CHECK_READ_THROTTLING();
METRIC_VAR_AUTO_LATENCY(multi_get_latency_ns);
@@ -855,11 +856,7 @@ void pegasus_server_impl::on_batch_get(batch_get_rpc rpc)
response.partition_index = _gpid.get_partition_index();
response.server = _primary_address;
- if (!_read_size_throttling_controller->available()) {
- rpc.error() = dsn::ERR_BUSY;
- _counter_recent_read_throttling_reject_count->increment();
- return;
- }
+ CHECK_READ_THROTTLING();
METRIC_VAR_AUTO_LATENCY(batch_get_latency_ns);
@@ -971,11 +968,7 @@ void pegasus_server_impl::on_sortkey_count(sortkey_count_rpc rpc)
resp.partition_index = _gpid.get_partition_index();
resp.server = _primary_address;
- if (!_read_size_throttling_controller->available()) {
- rpc.error() = dsn::ERR_BUSY;
- _counter_recent_read_throttling_reject_count->increment();
- return;
- }
+ CHECK_READ_THROTTLING();
METRIC_VAR_AUTO_LATENCY(scan_latency_ns);
@@ -1048,11 +1041,7 @@ void pegasus_server_impl::on_ttl(ttl_rpc rpc)
resp.partition_index = _gpid.get_partition_index();
resp.server = _primary_address;
- if (!_read_size_throttling_controller->available()) {
- rpc.error() = dsn::ERR_BUSY;
- _counter_recent_read_throttling_reject_count->increment();
- return;
- }
+ CHECK_READ_THROTTLING();
rocksdb::Slice skey(key.data(), key.length());
std::string value;
@@ -1112,11 +1101,7 @@ void pegasus_server_impl::on_get_scanner(get_scanner_rpc rpc)
resp.partition_index = _gpid.get_partition_index();
resp.server = _primary_address;
- if (!_read_size_throttling_controller->available()) {
- rpc.error() = dsn::ERR_BUSY;
- _counter_recent_read_throttling_reject_count->increment();
- return;
- }
+ CHECK_READ_THROTTLING();
METRIC_VAR_AUTO_LATENCY(scan_latency_ns);
@@ -1363,11 +1348,7 @@ void pegasus_server_impl::on_scan(scan_rpc rpc)
resp.partition_index = _gpid.get_partition_index();
resp.server = _primary_address;
- if (!_read_size_throttling_controller->available()) {
- rpc.error() = dsn::ERR_BUSY;
- _counter_recent_read_throttling_reject_count->increment();
- return;
- }
+ CHECK_READ_THROTTLING();
METRIC_VAR_AUTO_LATENCY(scan_latency_ns);
@@ -1855,13 +1836,13 @@ void pegasus_server_impl::cancel_background_work(bool wait)
LOG_ERROR_PREFIX("rmdir {} failed when stop app", data_dir());
return ::dsn::ERR_FILE_OPERATION_FAILED;
}
- _pfc_rdb_sst_count->set(0);
- _pfc_rdb_sst_size->set(0);
- _pfc_rdb_block_cache_hit_count->set(0);
- _pfc_rdb_block_cache_total_count->set(0);
+ METRIC_VAR_SET(rdb_total_sst_files, 0);
+ METRIC_VAR_SET(rdb_total_sst_size_mb, 0);
+ METRIC_VAR_SET(rdb_index_and_filter_blocks_mem_usage_bytes, 0);
+ METRIC_VAR_SET(rdb_memtable_mem_usage_bytes, 0);
+ METRIC_VAR_SET(rdb_block_cache_hit_count, 0);
+ METRIC_VAR_SET(rdb_block_cache_total_count, 0);
_pfc_rdb_block_cache_mem_usage->set(0);
- _pfc_rdb_index_and_filter_blocks_mem_usage->set(0);
- _pfc_rdb_memtable_mem_usage->set(0);
}
LOG_INFO_PREFIX("close app succeed, clear_state = {}", clear_state ? "true" : "false");
@@ -2423,12 +2404,16 @@ range_iteration_state pegasus_server_impl::append_key_value_for_multi_get(
return range_iteration_state::kNormal;
}
+#define GET_TICKER_COUNT_AND_SET_METRIC(ticker_name, metric_name) \
+ do { \
+ METRIC_VAR_SET(metric_name, _statistics->getTickerCount(rocksdb::ticker_name)); \
+ } while (0)
+
void pegasus_server_impl::update_replica_rocksdb_statistics()
{
std::string str_val;
uint64_t val = 0;
- // Update _pfc_rdb_sst_count
for (int i = 0; i < _data_cf_opts.num_levels; ++i) {
int cur_level_count = 0;
if (_db->GetProperty(rocksdb::DB::Properties::kNumFilesAtLevelPrefix + std::to_string(i),
@@ -2437,49 +2422,38 @@ void pegasus_server_impl::update_replica_rocksdb_statistics()
val += cur_level_count;
}
}
- _pfc_rdb_sst_count->set(val);
- LOG_DEBUG_PREFIX("_pfc_rdb_sst_count: {}", val);
+ METRIC_VAR_SET(rdb_total_sst_files, val);
- // Update _pfc_rdb_sst_size
if (_db->GetProperty(_data_cf, rocksdb::DB::Properties::kTotalSstFilesSize, &str_val) &&
dsn::buf2uint64(str_val, val)) {
static uint64_t bytes_per_mb = 1U << 20U;
- _pfc_rdb_sst_size->set(val / bytes_per_mb);
- LOG_DEBUG_PREFIX("_pfc_rdb_sst_size: {} bytes", val);
+ METRIC_VAR_SET(rdb_total_sst_size_mb, val / bytes_per_mb);
}
- // Update _pfc_rdb_write_amplification
std::map<std::string, std::string> props;
if (_db->GetMapProperty(_data_cf, "rocksdb.cfstats", &props)) {
auto write_amplification_iter = props.find("compaction.Sum.WriteAmp");
auto write_amplification = write_amplification_iter == props.end()
? 1
: std::stod(write_amplification_iter->second);
- _pfc_rdb_write_amplification->set(write_amplification);
- LOG_DEBUG_PREFIX("_pfc_rdb_write_amplification: {}", write_amplification);
+ METRIC_VAR_SET(rdb_write_amplification, write_amplification);
}
- // Update _pfc_rdb_index_and_filter_blocks_mem_usage
if (_db->GetProperty(_data_cf, rocksdb::DB::Properties::kEstimateTableReadersMem, &str_val) &&
dsn::buf2uint64(str_val, val)) {
- _pfc_rdb_index_and_filter_blocks_mem_usage->set(val);
- LOG_DEBUG_PREFIX("_pfc_rdb_index_and_filter_blocks_mem_usage: {} bytes", val);
+ METRIC_VAR_SET(rdb_index_and_filter_blocks_mem_usage_bytes, val);
}
- // Update _pfc_rdb_memtable_mem_usage
if (_db->GetProperty(_data_cf, rocksdb::DB::Properties::kCurSizeAllMemTables, &str_val) &&
dsn::buf2uint64(str_val, val)) {
- _pfc_rdb_memtable_mem_usage->set(val);
- LOG_DEBUG_PREFIX("_pfc_rdb_memtable_mem_usage: {} bytes", val);
+ METRIC_VAR_SET(rdb_memtable_mem_usage_bytes, val);
}
- // Update _pfc_rdb_estimate_num_keys
// NOTE: for the same n kv pairs, kEstimateNumKeys will be counted n times, you need compaction
// to remove duplicate
if (_db->GetProperty(_data_cf, rocksdb::DB::Properties::kEstimateNumKeys, &str_val) &&
dsn::buf2uint64(str_val, val)) {
- _pfc_rdb_estimate_num_keys->set(val);
- LOG_DEBUG_PREFIX("_pfc_rdb_estimate_num_keys: {}", val);
+ METRIC_VAR_SET(rdb_estimated_keys, val);
}
// the follow stats is related to `read`, so only primary need update it，ignore
@@ -2488,7 +2462,6 @@ void pegasus_server_impl::update_replica_rocksdb_statistics()
return;
}
- // Update _pfc_rdb_read_amplification
if (FLAGS_read_amp_bytes_per_bit > 0) {
auto estimate_useful_bytes =
_statistics->getTickerCount(rocksdb::READ_AMP_ESTIMATE_USEFUL_BYTES);
@@ -2496,68 +2469,41 @@ void pegasus_server_impl::update_replica_rocksdb_statistics()
auto read_amplification =
_statistics->getTickerCount(rocksdb::READ_AMP_TOTAL_READ_BYTES) /
estimate_useful_bytes;
- _pfc_rdb_read_amplification->set(read_amplification);
- LOG_DEBUG_PREFIX("_pfc_rdb_read_amplification: {}", read_amplification);
+ METRIC_VAR_SET(rdb_read_amplification, read_amplification);
}
}
- // Update _pfc_rdb_bf_seek_negatives
- auto bf_seek_negatives = _statistics->getTickerCount(rocksdb::BLOOM_FILTER_PREFIX_USEFUL);
- _pfc_rdb_bf_seek_negatives->set(bf_seek_negatives);
- LOG_DEBUG_PREFIX("_pfc_rdb_bf_seek_negatives: {}", bf_seek_negatives);
+ GET_TICKER_COUNT_AND_SET_METRIC(BLOOM_FILTER_PREFIX_USEFUL, rdb_bloom_filter_seek_negatives);
- // Update _pfc_rdb_bf_seek_total
- auto bf_seek_total = _statistics->getTickerCount(rocksdb::BLOOM_FILTER_PREFIX_CHECKED);
- _pfc_rdb_bf_seek_total->set(bf_seek_total);
- LOG_DEBUG_PREFIX("_pfc_rdb_bf_seek_total: {}", bf_seek_total);
+ GET_TICKER_COUNT_AND_SET_METRIC(BLOOM_FILTER_PREFIX_CHECKED, rdb_bloom_filter_seek_total);
- // Update _pfc_rdb_bf_point_positive_true
- auto bf_point_positive_true =
- _statistics->getTickerCount(rocksdb::BLOOM_FILTER_FULL_TRUE_POSITIVE);
- _pfc_rdb_bf_point_positive_true->set(bf_point_positive_true);
- LOG_DEBUG_PREFIX("_pfc_rdb_bf_point_positive_true: {}", bf_point_positive_true);
+ GET_TICKER_COUNT_AND_SET_METRIC(BLOOM_FILTER_USEFUL, rdb_bloom_filter_point_lookup_negatives);
- // Update _pfc_rdb_bf_point_positive_total
- auto bf_point_positive_total = _statistics->getTickerCount(rocksdb::BLOOM_FILTER_FULL_POSITIVE);
- _pfc_rdb_bf_point_positive_total->set(bf_point_positive_total);
- LOG_DEBUG_PREFIX("_pfc_rdb_bf_point_positive_total: {}", bf_point_positive_total);
+ GET_TICKER_COUNT_AND_SET_METRIC(BLOOM_FILTER_FULL_POSITIVE,
+ rdb_bloom_filter_point_lookup_positives);
- // Update _pfc_rdb_bf_point_negatives
- auto bf_point_negatives = _statistics->getTickerCount(rocksdb::BLOOM_FILTER_USEFUL);
- _pfc_rdb_bf_point_negatives->set(bf_point_negatives);
- LOG_DEBUG_PREFIX("_pfc_rdb_bf_point_negatives: {}", bf_point_negatives);
+ GET_TICKER_COUNT_AND_SET_METRIC(BLOOM_FILTER_FULL_TRUE_POSITIVE,
+ rdb_bloom_filter_point_lookup_true_positives);
- // Update _pfc_rdb_block_cache_hit_count and _pfc_rdb_block_cache_total_count
auto block_cache_hit = _statistics->getTickerCount(rocksdb::BLOCK_CACHE_HIT);
- _pfc_rdb_block_cache_hit_count->set(block_cache_hit);
- LOG_DEBUG_PREFIX("_pfc_rdb_block_cache_hit_count: {}", block_cache_hit);
+ METRIC_VAR_SET(rdb_block_cache_hit_count, block_cache_hit);
auto block_cache_miss = _statistics->getTickerCount(rocksdb::BLOCK_CACHE_MISS);
auto block_cache_total = block_cache_hit + block_cache_miss;
- _pfc_rdb_block_cache_total_count->set(block_cache_total);
- LOG_DEBUG_PREFIX("_pfc_rdb_block_cache_total_count: {}", block_cache_total);
+ METRIC_VAR_SET(rdb_block_cache_total_count, block_cache_total);
- // update block memtable/l0/l1/l2andup hit rate under block cache up level
auto memtable_hit_count = _statistics->getTickerCount(rocksdb::MEMTABLE_HIT);
- _pfc_rdb_memtable_hit_count->set(memtable_hit_count);
- LOG_DEBUG_PREFIX("_pfc_rdb_memtable_hit_count: {}", memtable_hit_count);
+ METRIC_VAR_SET(rdb_memtable_hit_count, memtable_hit_count);
auto memtable_miss_count = _statistics->getTickerCount(rocksdb::MEMTABLE_MISS);
auto memtable_total = memtable_hit_count + memtable_miss_count;
- _pfc_rdb_memtable_total_count->set(memtable_total);
- LOG_DEBUG_PREFIX("_pfc_rdb_memtable_total_count: {}", memtable_total);
+ METRIC_VAR_SET(rdb_memtable_total_count, memtable_total);
- auto l0_hit_count = _statistics->getTickerCount(rocksdb::GET_HIT_L0);
- _pfc_rdb_l0_hit_count->set(l0_hit_count);
- LOG_DEBUG_PREFIX("_pfc_rdb_l0_hit_count: {}", l0_hit_count);
+ GET_TICKER_COUNT_AND_SET_METRIC(GET_HIT_L0, rdb_l0_hit_count);
- auto l1_hit_count = _statistics->getTickerCount(rocksdb::GET_HIT_L1);
- _pfc_rdb_l1_hit_count->set(l1_hit_count);
- LOG_DEBUG_PREFIX("_pfc_rdb_l1_hit_count: {}", l1_hit_count);
+ GET_TICKER_COUNT_AND_SET_METRIC(GET_HIT_L1, rdb_l1_hit_count);
- auto l2andup_hit_count = _statistics->getTickerCount(rocksdb::GET_HIT_L2_AND_UP);
- _pfc_rdb_l2andup_hit_count->set(l2andup_hit_count);
- LOG_DEBUG_PREFIX("_pfc_rdb_l2andup_hit_count: {}", l2andup_hit_count);
+ GET_TICKER_COUNT_AND_SET_METRIC(GET_HIT_L2_AND_UP, rdb_l2_and_up_hit_count);
}
void pegasus_server_impl::update_server_rocksdb_statistics()
diff --git a/src/server/pegasus_server_impl.h b/src/server/pegasus_server_impl.h
index 8470b1e53..7f313cf68 100644
--- a/src/server/pegasus_server_impl.h
+++ b/src/server/pegasus_server_impl.h
@@ -541,34 +541,36 @@ private:
METRIC_VAR_DECLARE_counter(read_expired_values);
METRIC_VAR_DECLARE_counter(read_filtered_values);
METRIC_VAR_DECLARE_counter(abnormal_read_requests);
+ METRIC_VAR_DECLARE_counter(throttling_rejected_read_requests);
// rocksdb internal statistics
// server level
static ::dsn::perf_counter_wrapper _pfc_rdb_write_limiter_rate_bytes;
static ::dsn::perf_counter_wrapper _pfc_rdb_block_cache_mem_usage;
- // replica level
- dsn::perf_counter_wrapper _pfc_rdb_sst_count;
- dsn::perf_counter_wrapper _pfc_rdb_sst_size;
- dsn::perf_counter_wrapper _pfc_rdb_index_and_filter_blocks_mem_usage;
- dsn::perf_counter_wrapper _pfc_rdb_memtable_mem_usage;
- dsn::perf_counter_wrapper _pfc_rdb_estimate_num_keys;
-
- dsn::perf_counter_wrapper _pfc_rdb_bf_seek_negatives;
- dsn::perf_counter_wrapper _pfc_rdb_bf_seek_total;
- dsn::perf_counter_wrapper _pfc_rdb_bf_point_positive_true;
- dsn::perf_counter_wrapper _pfc_rdb_bf_point_positive_total;
- dsn::perf_counter_wrapper _pfc_rdb_bf_point_negatives;
- dsn::perf_counter_wrapper _pfc_rdb_block_cache_hit_count;
- dsn::perf_counter_wrapper _pfc_rdb_block_cache_total_count;
- dsn::perf_counter_wrapper _pfc_rdb_write_amplification;
- dsn::perf_counter_wrapper _pfc_rdb_read_amplification;
- dsn::perf_counter_wrapper _pfc_rdb_memtable_hit_count;
- dsn::perf_counter_wrapper _pfc_rdb_memtable_total_count;
- dsn::perf_counter_wrapper _pfc_rdb_l0_hit_count;
- dsn::perf_counter_wrapper _pfc_rdb_l1_hit_count;
- dsn::perf_counter_wrapper _pfc_rdb_l2andup_hit_count;
-
- dsn::perf_counter_wrapper _counter_recent_read_throttling_reject_count;
+
+ // Replica-level metrics for rocksdb.
+ METRIC_VAR_DECLARE_gauge_int64(rdb_total_sst_files);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_total_sst_size_mb);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_estimated_keys);
+
+ METRIC_VAR_DECLARE_gauge_int64(rdb_index_and_filter_blocks_mem_usage_bytes);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_memtable_mem_usage_bytes);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_block_cache_hit_count);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_block_cache_total_count);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_memtable_hit_count);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_memtable_total_count);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_l0_hit_count);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_l1_hit_count);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_l2_and_up_hit_count);
+
+ METRIC_VAR_DECLARE_gauge_int64(rdb_write_amplification);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_read_amplification);
+
+ METRIC_VAR_DECLARE_gauge_int64(rdb_bloom_filter_seek_negatives);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_bloom_filter_seek_total);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_bloom_filter_point_lookup_negatives);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_bloom_filter_point_lookup_positives);
+ METRIC_VAR_DECLARE_gauge_int64(rdb_bloom_filter_point_lookup_true_positives);
};
} // namespace server
diff --git a/src/server/pegasus_server_impl_init.cpp b/src/server/pegasus_server_impl_init.cpp
index 27aebdfab..0840c094b 100644
--- a/src/server/pegasus_server_impl_init.cpp
+++ b/src/server/pegasus_server_impl_init.cpp
@@ -65,57 +65,179 @@ class replica;
METRIC_DEFINE_counter(replica,
get_requests,
dsn::metric_unit::kRequests,
- "The number of GET requests for each replica");
+ "The number of GET requests");
METRIC_DEFINE_counter(replica,
multi_get_requests,
dsn::metric_unit::kRequests,
- "The number of MULTI_GET requests for each replica");
+ "The number of MULTI_GET requests");
METRIC_DEFINE_counter(replica,
batch_get_requests,
dsn::metric_unit::kRequests,
- "The number of BATCH_GET requests for each replica");
+ "The number of BATCH_GET requests");
METRIC_DEFINE_counter(replica,
scan_requests,
dsn::metric_unit::kRequests,
- "The number of SCAN requests for each replica");
+ "The number of SCAN requests");
METRIC_DEFINE_percentile_int64(replica,
get_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of GET requests for each replica");
+ "The latency of GET requests");
METRIC_DEFINE_percentile_int64(replica,
multi_get_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of MULTI_GET requests for each replica");
+ "The latency of MULTI_GET requests");
METRIC_DEFINE_percentile_int64(replica,
batch_get_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of BATCH_GET requests for each replica");
+ "The latency of BATCH_GET requests");
METRIC_DEFINE_percentile_int64(replica,
scan_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of SCAN requests for each replica");
+ "The latency of SCAN requests");
METRIC_DEFINE_counter(replica,
read_expired_values,
dsn::metric_unit::kValues,
- "The number of expired values read for each replica");
+ "The number of expired values read");
METRIC_DEFINE_counter(replica,
read_filtered_values,
dsn::metric_unit::kValues,
- "The number of filtered values read for each replica");
+ "The number of filtered values read");
METRIC_DEFINE_counter(replica,
abnormal_read_requests,
dsn::metric_unit::kRequests,
- "The number of abnormal read requests for each replica");
+ "The number of abnormal read requests");
+
+METRIC_DEFINE_counter(replica,
+ throttling_rejected_read_requests,
+ dsn::metric_unit::kRequests,
+ "The number of rejected read requests by throttling");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_total_sst_files,
+ dsn::metric_unit::kFiles,
+ "The total number of rocksdb sst files");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_total_sst_size_mb,
+ dsn::metric_unit::kMegaBytes,
+ "The total size of rocksdb sst files in MB");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_estimated_keys,
+ dsn::metric_unit::kKeys,
+ "The estimated number of rocksdb keys");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_index_and_filter_blocks_mem_usage_bytes,
+ dsn::metric_unit::kBytes,
+ "The memory usage of rocksdb index and filter blocks in bytes");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_memtable_mem_usage_bytes,
+ dsn::metric_unit::kBytes,
+ "The memory usage of rocksdb memtables in bytes");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_block_cache_hit_count,
+ dsn::metric_unit::kPointLookups,
+ "The hit number of lookups on rocksdb block cache");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_block_cache_total_count,
+ dsn::metric_unit::kPointLookups,
+ "The total number of lookups on rocksdb block cache");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_memtable_hit_count,
+ dsn::metric_unit::kPointLookups,
+ "The hit number of lookups on rocksdb memtable");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_memtable_total_count,
+ dsn::metric_unit::kPointLookups,
+ "The total number of lookups on rocksdb memtable");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_l0_hit_count,
+ dsn::metric_unit::kPointLookups,
+ "The number of lookups served by rocksdb L0");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_l1_hit_count,
+ dsn::metric_unit::kPointLookups,
+ "The number of lookups served by rocksdb L1");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_l2_and_up_hit_count,
+ dsn::metric_unit::kPointLookups,
+ "The number of lookups served by rocksdb L2 and up");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_write_amplification,
+ dsn::metric_unit::kAmplification,
+ "The write amplification of rocksdb");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_read_amplification,
+ dsn::metric_unit::kAmplification,
+ "The read amplification of rocksdb");
+
+// Following metrics are rocksdb statistics that are related to bloom filters.
+//
+// To measure prefix bloom filters, these metrics are updated after each ::Seek and ::SeekForPrev if
+// prefix is enabled and check_filter is set:
+// * rdb_bloom_filter_seek_negatives: seek_negatives
+// * rdb_bloom_filter_seek_total: seek_negatives + seek_positives
+//
+// To measure full bloom filters, these metrics are updated after each point lookup. If
+// whole_key_filtering is set, this is the result of checking the bloom of the whole key, otherwise
+// this is the result of checking the bloom of the prefix:
+// * rdb_bloom_filter_point_lookup_negatives: [true] negatives
+// * rdb_bloom_filter_point_lookup_positives: positives
+// * rdb_bloom_filter_point_lookup_true_positives: true positives
+//
+// For details please see https://github.com/facebook/rocksdb/wiki/RocksDB-Bloom-Filter#statistic.
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_bloom_filter_seek_negatives,
+ dsn::metric_unit::kSeeks,
+ "The number of times the check for prefix bloom filter was useful in "
+ "avoiding iterator creation (and thus likely IOPs), used by rocksdb for "
+ "each replica");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_bloom_filter_seek_total,
+ dsn::metric_unit::kSeeks,
+ "The number of times prefix bloom filter was checked before creating "
+ "iterator on a file, used by rocksdb");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_bloom_filter_point_lookup_negatives,
+ dsn::metric_unit::kPointLookups,
+ "The number of times full bloom filter has avoided file reads (i.e., "
+ "negatives), used by rocksdb");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_bloom_filter_point_lookup_positives,
+ dsn::metric_unit::kPointLookups,
+ "The number of times full bloom filter has not avoided the reads, used "
+ "by rocksdb");
+
+METRIC_DEFINE_gauge_int64(replica,
+ rdb_bloom_filter_point_lookup_true_positives,
+ dsn::metric_unit::kPointLookups,
+ "The number of times full bloom filter has not avoided the reads and "
+ "data actually exist, used by rocksdb");
namespace pegasus {
namespace server {
@@ -469,7 +591,27 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
METRIC_VAR_INIT_replica(scan_latency_ns),
METRIC_VAR_INIT_replica(read_expired_values),
METRIC_VAR_INIT_replica(read_filtered_values),
- METRIC_VAR_INIT_replica(abnormal_read_requests)
+ METRIC_VAR_INIT_replica(abnormal_read_requests),
+ METRIC_VAR_INIT_replica(throttling_rejected_read_requests),
+ METRIC_VAR_INIT_replica(rdb_total_sst_files),
+ METRIC_VAR_INIT_replica(rdb_total_sst_size_mb),
+ METRIC_VAR_INIT_replica(rdb_estimated_keys),
+ METRIC_VAR_INIT_replica(rdb_index_and_filter_blocks_mem_usage_bytes),
+ METRIC_VAR_INIT_replica(rdb_memtable_mem_usage_bytes),
+ METRIC_VAR_INIT_replica(rdb_block_cache_hit_count),
+ METRIC_VAR_INIT_replica(rdb_block_cache_total_count),
+ METRIC_VAR_INIT_replica(rdb_memtable_hit_count),
+ METRIC_VAR_INIT_replica(rdb_memtable_total_count),
+ METRIC_VAR_INIT_replica(rdb_l0_hit_count),
+ METRIC_VAR_INIT_replica(rdb_l1_hit_count),
+ METRIC_VAR_INIT_replica(rdb_l2_and_up_hit_count),
+ METRIC_VAR_INIT_replica(rdb_write_amplification),
+ METRIC_VAR_INIT_replica(rdb_read_amplification),
+ METRIC_VAR_INIT_replica(rdb_bloom_filter_seek_negatives),
+ METRIC_VAR_INIT_replica(rdb_bloom_filter_seek_total),
+ METRIC_VAR_INIT_replica(rdb_bloom_filter_point_lookup_negatives),
+ METRIC_VAR_INIT_replica(rdb_bloom_filter_point_lookup_positives),
+ METRIC_VAR_INIT_replica(rdb_bloom_filter_point_lookup_true_positives)
{
_primary_address = dsn::rpc_address(dsn_primary_address()).to_string();
_gpid = get_gpid();
@@ -675,54 +817,6 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
std::string str_gpid = _gpid.to_string();
char name[256];
- // register the perf counters
- snprintf(name, 255, "disk.storage.sst.count@%s", str_gpid.c_str());
- _pfc_rdb_sst_count.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistic the count of sstable files");
-
- snprintf(name, 255, "disk.storage.sst(MB)@%s", str_gpid.c_str());
- _pfc_rdb_sst_size.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistic the size of sstable files");
-
- snprintf(name, 255, "rdb.block_cache.hit_count@%s", str_gpid.c_str());
- _pfc_rdb_block_cache_hit_count.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistic the hit count of rocksdb block cache");
-
- snprintf(name, 255, "rdb.block_cache.total_count@%s", str_gpid.c_str());
- _pfc_rdb_block_cache_total_count.init_app_counter(
- "app.pegasus",
- name,
- COUNTER_TYPE_NUMBER,
- "statistic the total count of rocksdb block cache");
-
- snprintf(name, 255, "rdb.write_amplification@%s", str_gpid.c_str());
- _pfc_rdb_write_amplification.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistics the write amplification of rocksdb");
-
- snprintf(name, 255, "rdb.read_amplification@%s", str_gpid.c_str());
- _pfc_rdb_read_amplification.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistics the read amplification of rocksdb");
-
- snprintf(name, 255, "rdb.read_memtable_hit_count@%s", str_gpid.c_str());
- _pfc_rdb_memtable_hit_count.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistics the read memtable hit count");
-
- snprintf(name, 255, "rdb.read_memtable_total_count@%s", str_gpid.c_str());
- _pfc_rdb_memtable_total_count.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistics the read memtable total count");
-
- snprintf(name, 255, "rdb.read_l0_hit_count@%s", str_gpid.c_str());
- _pfc_rdb_l0_hit_count.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistics the read l0 hit count");
-
- snprintf(name, 255, "rdb.read_l1_hit_count@%s", str_gpid.c_str());
- _pfc_rdb_l1_hit_count.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistics the read l1 hit count");
-
- snprintf(name, 255, "rdb.read_l2andup_hit_count@%s", str_gpid.c_str());
- _pfc_rdb_l2andup_hit_count.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistics the read l2andup hit count");
-
// These counters are singletons on this server shared by all replicas, so we initialize
// them only once.
static std::once_flag flag;
@@ -741,66 +835,7 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
COUNTER_TYPE_NUMBER,
"statistic the through bytes of rocksdb write rate limiter");
});
-
- snprintf(name, 255, "rdb.index_and_filter_blocks.memory_usage@%s", str_gpid.c_str());
- _pfc_rdb_index_and_filter_blocks_mem_usage.init_app_counter(
- "app.pegasus",
- name,
- COUNTER_TYPE_NUMBER,
- "statistic the memory usage of rocksdb index and filter blocks");
-
- snprintf(name, 255, "rdb.memtable.memory_usage@%s", str_gpid.c_str());
- _pfc_rdb_memtable_mem_usage.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_NUMBER, "statistic the memory usage of rocksdb memtable");
-
- snprintf(name, 255, "rdb.estimate_num_keys@%s", str_gpid.c_str());
- _pfc_rdb_estimate_num_keys.init_app_counter(
- "app.pegasus",
- name,
- COUNTER_TYPE_NUMBER,
- "statistics the estimated number of keys inside the rocksdb");
-
- snprintf(name, 255, "rdb.bf_seek_negatives@%s", str_gpid.c_str());
- _pfc_rdb_bf_seek_negatives.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_NUMBER,
- "statistics the number of times bloom filter was "
- "checked before creating iterator on a file and "
- "useful in avoiding iterator creation (and thus "
- "likely IOPs)");
-
- snprintf(name, 255, "rdb.bf_seek_total@%s", str_gpid.c_str());
- _pfc_rdb_bf_seek_total.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_NUMBER,
- "statistics the number of times bloom filter was "
- "checked before creating iterator on a file");
-
- snprintf(name, 255, "rdb.bf_point_positive_true@%s", str_gpid.c_str());
- _pfc_rdb_bf_point_positive_true.init_app_counter(
- "app.pegasus",
- name,
- COUNTER_TYPE_NUMBER,
- "statistics the number of times bloom filter has avoided file reads, i.e., negatives");
-
- snprintf(name, 255, "rdb.bf_point_positive_total@%s", str_gpid.c_str());
- _pfc_rdb_bf_point_positive_total.init_app_counter(
- "app.pegasus",
- name,
- COUNTER_TYPE_NUMBER,
- "statistics the number of times bloom FullFilter has not avoided the reads");
-
- snprintf(name, 255, "rdb.bf_point_negatives@%s", str_gpid.c_str());
- _pfc_rdb_bf_point_negatives.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_NUMBER,
- "statistics the number of times bloom FullFilter "
- "has not avoided the reads and data actually "
- "exist");
-
- auto counter_str = fmt::format("recent.read.throttling.reject.count@{}", str_gpid.c_str());
- _counter_recent_read_throttling_reject_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
}
+
} // namespace server
} // namespace pegasus
diff --git a/src/utils/metrics.h b/src/utils/metrics.h
index da1d056d8..facc83f26 100644
--- a/src/utils/metrics.h
+++ b/src/utils/metrics.h
@@ -161,8 +161,9 @@ class error_code;
METRIC_VAR_DECLARE(name, dsn::percentile_ptr<int64_t>)
// Initialize a metric variable in user class.
-#define METRIC_VAR_INIT(name, entity) _##name(METRIC_##name.instantiate(entity##_metric_entity()))
-#define METRIC_VAR_INIT_replica(name) METRIC_VAR_INIT(name, replica)
+#define METRIC_VAR_INIT(name, entity, ...) \
+ _##name(METRIC_##name.instantiate(entity##_metric_entity(), ##__VA_ARGS__))
+#define METRIC_VAR_INIT_replica(name, ...) METRIC_VAR_INIT(name, replica, ##__VA_ARGS__)
// Perform increment-related operations on metrics including gauge and counter.
#define METRIC_VAR_INCREMENT_BY(name, x) \
@@ -608,8 +609,15 @@ enum class metric_unit : size_t
kMicroSeconds,
kMilliSeconds,
kSeconds,
+ kBytes,
+ kMegaBytes,
kRequests,
+ kSeeks,
+ kPointLookups,
kValues,
+ kKeys,
+ kFiles,
+ kAmplification,
kInvalidUnit,
};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 07/12: feat(new_metrics): migrate replica-level metrics for pegasus_event_listener (#1407)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit de20d0f8f4c10e2e09be31115a1ec8f243a83e53
Author: Dan Wang <wa...@apache.org>
AuthorDate: Thu Mar 23 18:06:01 2023 +0800
feat(new_metrics): migrate replica-level metrics for pegasus_event_listener (#1407)
https://github.com/apache/incubator-pegasus/issues/1343
Migrate replica-level metrics in pegasus_event_listener class to new framework, all of which
are rocksdb-related, including the number of completed flushes/compactions, the size of
flush output in bytes, the size of compaction input/output in bytes.
Note that in old perf counters there are just 2 replica-level metrics for pegasus_event_listener
while all of others are server-level. Migrated to new framework all of the metrics have become
replica-level; once server-level metrics are needed, just aggregate on replica-level ones.
---
src/server/pegasus_event_listener.cpp | 123 ++++++++++++++++------------------
src/server/pegasus_event_listener.h | 26 ++++---
src/utils/metrics.h | 3 +
3 files changed, 72 insertions(+), 80 deletions(-)
diff --git a/src/server/pegasus_event_listener.cpp b/src/server/pegasus_event_listener.cpp
index e5414d8b7..330f93af9 100644
--- a/src/server/pegasus_event_listener.cpp
+++ b/src/server/pegasus_event_listener.cpp
@@ -19,100 +19,91 @@
#include "pegasus_event_listener.h"
-#include <fmt/core.h>
-#include <fmt/ostream.h>
#include <rocksdb/compaction_job_stats.h>
#include <rocksdb/table_properties.h>
-#include <iosfwd>
-#include <string>
-#include "common/gpid.h"
-#include "perf_counter/perf_counter.h"
+#include "utils/autoref_ptr.h"
#include "utils/fmt_logging.h"
+#include "utils/string_view.h"
namespace rocksdb {
class DB;
} // namespace rocksdb
+METRIC_DEFINE_counter(replica,
+ rdb_flush_completed_count,
+ dsn::metric_unit::kFlushes,
+ "The number of completed rocksdb flushes");
+
+METRIC_DEFINE_counter(replica,
+ rdb_flush_output_bytes,
+ dsn::metric_unit::kBytes,
+ "The size of rocksdb flush output in bytes");
+
+METRIC_DEFINE_counter(replica,
+ rdb_compaction_completed_count,
+ dsn::metric_unit::kCompactions,
+ "The number of completed rocksdb compactions");
+
+METRIC_DEFINE_counter(replica,
+ rdb_compaction_input_bytes,
+ dsn::metric_unit::kBytes,
+ "The size of rocksdb compaction input in bytes");
+
+METRIC_DEFINE_counter(replica,
+ rdb_compaction_output_bytes,
+ dsn::metric_unit::kBytes,
+ "The size of rocksdb compaction output in bytes");
+
+METRIC_DEFINE_counter(
+ replica,
+ rdb_changed_delayed_writes,
+ dsn::metric_unit::kWrites,
+ "The number of rocksdb delayed writes changed from another write stall condition");
+
+METRIC_DEFINE_counter(
+ replica,
+ rdb_changed_stopped_writes,
+ dsn::metric_unit::kWrites,
+ "The number of rocksdb stopped writes changed from another write stall condition");
+
namespace pegasus {
namespace server {
-pegasus_event_listener::pegasus_event_listener(replica_base *r) : replica_base(r)
+pegasus_event_listener::pegasus_event_listener(replica_base *r)
+ : replica_base(r),
+ METRIC_VAR_INIT_replica(rdb_flush_completed_count),
+ METRIC_VAR_INIT_replica(rdb_flush_output_bytes),
+ METRIC_VAR_INIT_replica(rdb_compaction_completed_count),
+ METRIC_VAR_INIT_replica(rdb_compaction_input_bytes),
+ METRIC_VAR_INIT_replica(rdb_compaction_output_bytes),
+ METRIC_VAR_INIT_replica(rdb_changed_delayed_writes),
+ METRIC_VAR_INIT_replica(rdb_changed_stopped_writes)
{
- _pfc_recent_flush_completed_count.init_app_counter("app.pegasus",
- "recent.flush.completed.count",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "rocksdb recent flush completed count");
- _pfc_recent_flush_output_bytes.init_app_counter("app.pegasus",
- "recent.flush.output.bytes",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "rocksdb recent flush output bytes");
- _pfc_recent_compaction_completed_count.init_app_counter(
- "app.pegasus",
- "recent.compaction.completed.count",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "rocksdb recent compaction completed count");
- _pfc_recent_compaction_input_bytes.init_app_counter("app.pegasus",
- "recent.compaction.input.bytes",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "rocksdb recent compaction input bytes");
- _pfc_recent_compaction_output_bytes.init_app_counter("app.pegasus",
- "recent.compaction.output.bytes",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "rocksdb recent compaction output bytes");
- _pfc_recent_write_change_delayed_count.init_app_counter(
- "app.pegasus",
- "recent.write.change.delayed.count",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "rocksdb recent write change delayed count");
- _pfc_recent_write_change_stopped_count.init_app_counter(
- "app.pegasus",
- "recent.write.change.stopped.count",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "rocksdb recent write change stopped count");
-
- // replica-level perfcounter
- std::string counter_str = fmt::format("recent_rdb_compaction_input_bytes@{}", r->get_gpid());
- _pfc_recent_rdb_compaction_input_bytes.init_app_counter(
- "app.pegasus",
- counter_str.c_str(),
- COUNTER_TYPE_VOLATILE_NUMBER,
- "rocksdb recent compaction input bytes");
-
- counter_str = fmt::format("recent_rdb_compaction_output_bytes@{}", r->get_gpid());
- _pfc_recent_rdb_compaction_output_bytes.init_app_counter(
- "app.pegasus",
- counter_str.c_str(),
- COUNTER_TYPE_VOLATILE_NUMBER,
- "rocksdb recent compaction output bytes");
}
-void pegasus_event_listener::OnFlushCompleted(rocksdb::DB *db,
- const rocksdb::FlushJobInfo &flush_job_info)
+void pegasus_event_listener::OnFlushCompleted(rocksdb::DB *db, const rocksdb::FlushJobInfo &info)
{
- _pfc_recent_flush_completed_count->increment();
- _pfc_recent_flush_output_bytes->add(flush_job_info.table_properties.data_size);
+ METRIC_VAR_INCREMENT(rdb_flush_completed_count);
+ METRIC_VAR_INCREMENT_BY(rdb_flush_output_bytes, info.table_properties.data_size);
}
void pegasus_event_listener::OnCompactionCompleted(rocksdb::DB *db,
- const rocksdb::CompactionJobInfo &ci)
+ const rocksdb::CompactionJobInfo &info)
{
- _pfc_recent_compaction_completed_count->increment();
- _pfc_recent_compaction_input_bytes->add(ci.stats.total_input_bytes);
- _pfc_recent_compaction_output_bytes->add(ci.stats.total_output_bytes);
-
- _pfc_recent_rdb_compaction_input_bytes->add(ci.stats.total_input_bytes);
- _pfc_recent_rdb_compaction_output_bytes->add(ci.stats.total_output_bytes);
+ METRIC_VAR_INCREMENT(rdb_compaction_completed_count);
+ METRIC_VAR_INCREMENT_BY(rdb_compaction_input_bytes, info.stats.total_input_bytes);
+ METRIC_VAR_INCREMENT_BY(rdb_compaction_output_bytes, info.stats.total_output_bytes);
}
void pegasus_event_listener::OnStallConditionsChanged(const rocksdb::WriteStallInfo &info)
{
if (info.condition.cur == rocksdb::WriteStallCondition::kDelayed) {
LOG_ERROR_PREFIX("rocksdb write delayed");
- _pfc_recent_write_change_delayed_count->increment();
+ METRIC_VAR_INCREMENT(rdb_changed_delayed_writes);
} else if (info.condition.cur == rocksdb::WriteStallCondition::kStopped) {
LOG_ERROR_PREFIX("rocksdb write stopped");
- _pfc_recent_write_change_stopped_count->increment();
+ METRIC_VAR_INCREMENT(rdb_changed_stopped_writes);
}
}
diff --git a/src/server/pegasus_event_listener.h b/src/server/pegasus_event_listener.h
index 5e6ab7e1a..900fabdc8 100644
--- a/src/server/pegasus_event_listener.h
+++ b/src/server/pegasus_event_listener.h
@@ -21,8 +21,8 @@
#include <rocksdb/listener.h>
-#include "perf_counter/perf_counter_wrapper.h"
#include "replica/replica_base.h"
+#include "utils/metrics.h"
namespace rocksdb {
class DB;
@@ -37,24 +37,22 @@ public:
explicit pegasus_event_listener(replica_base *r);
~pegasus_event_listener() override = default;
- void OnFlushCompleted(rocksdb::DB *db, const rocksdb::FlushJobInfo &flush_job_info) override;
+ void OnFlushCompleted(rocksdb::DB *db, const rocksdb::FlushJobInfo &info) override;
- void OnCompactionCompleted(rocksdb::DB *db, const rocksdb::CompactionJobInfo &ci) override;
+ void OnCompactionCompleted(rocksdb::DB *db, const rocksdb::CompactionJobInfo &info) override;
void OnStallConditionsChanged(const rocksdb::WriteStallInfo &info) override;
private:
- ::dsn::perf_counter_wrapper _pfc_recent_flush_completed_count;
- ::dsn::perf_counter_wrapper _pfc_recent_flush_output_bytes;
- ::dsn::perf_counter_wrapper _pfc_recent_compaction_completed_count;
- ::dsn::perf_counter_wrapper _pfc_recent_compaction_input_bytes;
- ::dsn::perf_counter_wrapper _pfc_recent_compaction_output_bytes;
- ::dsn::perf_counter_wrapper _pfc_recent_write_change_delayed_count;
- ::dsn::perf_counter_wrapper _pfc_recent_write_change_stopped_count;
-
- // replica-level perfcounter
- ::dsn::perf_counter_wrapper _pfc_recent_rdb_compaction_input_bytes;
- ::dsn::perf_counter_wrapper _pfc_recent_rdb_compaction_output_bytes;
+ METRIC_VAR_DECLARE_counter(rdb_flush_completed_count);
+ METRIC_VAR_DECLARE_counter(rdb_flush_output_bytes);
+
+ METRIC_VAR_DECLARE_counter(rdb_compaction_completed_count);
+ METRIC_VAR_DECLARE_counter(rdb_compaction_input_bytes);
+ METRIC_VAR_DECLARE_counter(rdb_compaction_output_bytes);
+
+ METRIC_VAR_DECLARE_counter(rdb_changed_delayed_writes);
+ METRIC_VAR_DECLARE_counter(rdb_changed_stopped_writes);
};
} // namespace server
diff --git a/src/utils/metrics.h b/src/utils/metrics.h
index 333534703..9087b78e5 100644
--- a/src/utils/metrics.h
+++ b/src/utils/metrics.h
@@ -620,6 +620,9 @@ enum class metric_unit : size_t
kKeys,
kFiles,
kAmplification,
+ kFlushes,
+ kCompactions,
+ kWrites,
kInvalidUnit,
};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 12/12: feat(new_metrics): add disk-level metric entity and migrate disk-level metrics for fs_manager (#1427)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit a8cd54564575aa77aa149ec7a2390a6ffd35381d
Author: Dan Wang <wa...@apache.org>
AuthorDate: Thu Apr 6 11:41:34 2023 +0800
feat(new_metrics): add disk-level metric entity and migrate disk-level metrics for fs_manager (#1427)
https://github.com/apache/incubator-pegasus/issues/1425
In perf counters, all metrics of `fs_manager` are server-level. For example,
the total capacity and the available capacity of all disks where there are
data of pegasus.
However, sometimes the capacity and the available capacity of each disk
seem more important: no space left on the disk will lead to serious problems.
Therefore, after being migrated to new framework, the server-level metrics
of perf counters become disk-level, including the capacity and the available
capacity of a disk. As for another disk-level metric -- the available percentage
of each disk used by a replica server, just use division operator.
Once server-level metrics are needed, just aggregate on the disk-level ones.
To compute another 2 server-level metrics -- the minimal/maximal available
percentage among all disks used by a replica server in a node, for example,
just use min/max operators over disk-level ones for Prometheus.
To implement disk-level metrics, disk-level metric entity are also added.
---
src/common/fs_manager.cpp | 102 +++++++++++++++++++++++++----------------
src/common/fs_manager.h | 58 +++++++++++++----------
src/common/test/CMakeLists.txt | 1 +
src/meta/test/misc/misc.cpp | 2 +-
src/replica/replica_stub.cpp | 1 -
src/utils/metrics.h | 7 +++
6 files changed, 105 insertions(+), 66 deletions(-)
diff --git a/src/common/fs_manager.cpp b/src/common/fs_manager.cpp
index 5a460a896..61ca99411 100644
--- a/src/common/fs_manager.cpp
+++ b/src/common/fs_manager.cpp
@@ -41,7 +41,7 @@
#include "common/gpid.h"
#include "common/replication_enums.h"
-#include "perf_counter/perf_counter.h"
+#include "fmt/core.h"
#include "runtime/api_layer1.h"
#include "runtime/rpc/rpc_address.h"
#include "utils/fail_point.h"
@@ -49,6 +49,18 @@
#include "utils/fmt_logging.h"
#include "utils/string_view.h"
+METRIC_DEFINE_entity(disk);
+
+METRIC_DEFINE_gauge_int64(disk,
+ total_disk_capacity_mb,
+ dsn::metric_unit::kMegaBytes,
+ "The total disk capacity");
+
+METRIC_DEFINE_gauge_int64(disk,
+ avail_disk_capacity_mb,
+ dsn::metric_unit::kMegaBytes,
+ "The available disk capacity");
+
namespace dsn {
namespace replication {
@@ -61,6 +73,34 @@ DSN_DEFINE_int32(replication,
"space insufficient");
DSN_TAG_VARIABLE(disk_min_available_space_ratio, FT_MUTABLE);
+namespace {
+
+metric_entity_ptr instantiate_disk_metric_entity(const std::string &tag,
+ const std::string &data_dir)
+{
+ auto entity_id = fmt::format("disk_{}", tag);
+
+ return METRIC_ENTITY_disk.instantiate(entity_id, {{"tag", tag}, {"data_dir", data_dir}});
+}
+
+} // anonymous namespace
+
+disk_capacity_metrics::disk_capacity_metrics(const std::string &tag, const std::string &data_dir)
+ : _disk_metric_entity(instantiate_disk_metric_entity(tag, data_dir)),
+ METRIC_VAR_INIT_disk(total_disk_capacity_mb),
+ METRIC_VAR_INIT_disk(avail_disk_capacity_mb)
+{
+}
+
+const metric_entity_ptr &disk_capacity_metrics::disk_metric_entity() const
+{
+ CHECK_NOTNULL(_disk_metric_entity,
+ "disk metric entity should has been instantiated: "
+ "uninitialized entity cannot be used to instantiate "
+ "metric");
+ return _disk_metric_entity;
+}
+
unsigned dir_node::replicas_count() const
{
unsigned sum = 0;
@@ -108,6 +148,9 @@ bool dir_node::update_disk_stat(const bool update_disk_status)
disk_available_ratio = static_cast<int>(
disk_capacity_mb == 0 ? 0 : std::round(disk_available_mb * 100.0 / disk_capacity_mb));
+ METRIC_CALL_SET_METHOD(disk_capacity, total_disk_capacity_mb, disk_capacity_mb);
+ METRIC_CALL_SET_METHOD(disk_capacity, avail_disk_capacity_mb, disk_available_mb);
+
if (!update_disk_status) {
LOG_INFO("update disk space succeed: dir = {}, capacity_mb = {}, available_mb = {}, "
"available_ratio = {}%",
@@ -134,32 +177,6 @@ bool dir_node::update_disk_stat(const bool update_disk_status)
return (old_status != new_status);
}
-fs_manager::fs_manager(bool for_test)
-{
- if (!for_test) {
- _counter_total_capacity_mb.init_app_counter("eon.replica_stub",
- "disk.capacity.total(MB)",
- COUNTER_TYPE_NUMBER,
- "total disk capacity in MB");
- _counter_total_available_mb.init_app_counter("eon.replica_stub",
- "disk.available.total(MB)",
- COUNTER_TYPE_NUMBER,
- "total disk available in MB");
- _counter_total_available_ratio.init_app_counter("eon.replica_stub",
- "disk.available.total.ratio",
- COUNTER_TYPE_NUMBER,
- "total disk available ratio");
- _counter_min_available_ratio.init_app_counter("eon.replica_stub",
- "disk.available.min.ratio",
- COUNTER_TYPE_NUMBER,
- "minimal disk available ratio in all disks");
- _counter_max_available_ratio.init_app_counter("eon.replica_stub",
- "disk.available.max.ratio",
- COUNTER_TYPE_NUMBER,
- "maximal disk available ratio in all disks");
- }
-}
-
dir_node *fs_manager::get_dir_node(const std::string &subdir) const
{
zauto_read_lock l(_lock);
@@ -298,17 +315,29 @@ bool fs_manager::for_each_dir_node(const std::function<bool(const dir_node &)> &
void fs_manager::update_disk_stat(bool check_status_changed)
{
- reset_disk_stat();
+ _total_capacity_mb = 0;
+ _total_available_mb = 0;
+ int total_available_ratio = 0;
+ int min_available_ratio = 100;
+ int max_available_ratio = 0;
+
+ // _status_updated_dir_nodes is accessed sequentially in update_disk_stat() and
+ // replica_stub::update_disks_status() during replica_stub::on_disk_stat(), thus
+ // no need to protect it by lock.
+ _status_updated_dir_nodes.clear();
+
+ zauto_read_lock l(_lock);
+
for (auto &dir_node : _dir_nodes) {
if (dir_node->update_disk_stat(check_status_changed)) {
_status_updated_dir_nodes.emplace_back(dir_node);
}
_total_capacity_mb += dir_node->disk_capacity_mb;
_total_available_mb += dir_node->disk_available_mb;
- _min_available_ratio = std::min(dir_node->disk_available_ratio, _min_available_ratio);
- _max_available_ratio = std::max(dir_node->disk_available_ratio, _max_available_ratio);
+ min_available_ratio = std::min(dir_node->disk_available_ratio, min_available_ratio);
+ max_available_ratio = std::max(dir_node->disk_available_ratio, max_available_ratio);
}
- _total_available_ratio = static_cast<int>(
+ total_available_ratio = static_cast<int>(
_total_capacity_mb == 0 ? 0 : std::round(_total_available_mb * 100.0 / _total_capacity_mb));
LOG_INFO("update disk space succeed: disk_count = {}, total_capacity_mb = {}, "
@@ -317,14 +346,9 @@ void fs_manager::update_disk_stat(bool check_status_changed)
_dir_nodes.size(),
_total_capacity_mb,
_total_available_mb,
- _total_available_ratio,
- _min_available_ratio,
- _max_available_ratio);
- _counter_total_capacity_mb->set(_total_capacity_mb);
- _counter_total_available_mb->set(_total_available_mb);
- _counter_total_available_ratio->set(_total_available_ratio);
- _counter_min_available_ratio->set(_min_available_ratio);
- _counter_max_available_ratio->set(_max_available_ratio);
+ total_available_ratio,
+ min_available_ratio,
+ max_available_ratio);
}
void fs_manager::add_new_dir_node(const std::string &data_dir, const std::string &tag)
diff --git a/src/common/fs_manager.h b/src/common/fs_manager.h
index 3c29973c2..fbefc64b4 100644
--- a/src/common/fs_manager.h
+++ b/src/common/fs_manager.h
@@ -27,9 +27,11 @@
#include "common/replication_other_types.h"
#include "metadata_types.h"
-#include "perf_counter/perf_counter_wrapper.h"
+#include "utils/autoref_ptr.h"
#include "utils/error_code.h"
#include "utils/flags.h"
+#include "utils/metrics.h"
+#include "utils/ports.h"
#include "utils/zlocks.h"
namespace dsn {
@@ -40,6 +42,25 @@ class replication_options;
DSN_DECLARE_int32(disk_min_available_space_ratio);
+class disk_capacity_metrics
+{
+public:
+ disk_capacity_metrics(const std::string &tag, const std::string &data_dir);
+ ~disk_capacity_metrics() = default;
+
+ const metric_entity_ptr &disk_metric_entity() const;
+
+ METRIC_DEFINE_SET_METHOD(total_disk_capacity_mb, int64_t)
+ METRIC_DEFINE_SET_METHOD(avail_disk_capacity_mb, int64_t)
+
+private:
+ const metric_entity_ptr _disk_metric_entity;
+ METRIC_VAR_DECLARE_gauge_int64(total_disk_capacity_mb);
+ METRIC_VAR_DECLARE_gauge_int64(avail_disk_capacity_mb);
+
+ DISALLOW_COPY_AND_ASSIGN(disk_capacity_metrics);
+};
+
struct dir_node
{
public:
@@ -53,6 +74,9 @@ public:
std::map<app_id, std::set<gpid>> holding_primary_replicas;
std::map<app_id, std::set<gpid>> holding_secondary_replicas;
+private:
+ disk_capacity_metrics disk_capacity;
+
public:
dir_node(const std::string &tag_,
const std::string &dir_,
@@ -65,7 +89,8 @@ public:
disk_capacity_mb(disk_capacity_mb_),
disk_available_mb(disk_available_mb_),
disk_available_ratio(disk_available_ratio_),
- status(status_)
+ status(status_),
+ disk_capacity(tag_, dir_)
{
}
unsigned replicas_count(app_id id) const;
@@ -78,8 +103,8 @@ public:
class fs_manager
{
public:
- fs_manager(bool for_test);
- ~fs_manager() {}
+ fs_manager() = default;
+ ~fs_manager() = default;
// this should be called before open/load any replicas
dsn::error_code initialize(const replication_options &opts);
@@ -105,27 +130,16 @@ public:
}
private:
- void reset_disk_stat()
- {
- _total_capacity_mb = 0;
- _total_available_mb = 0;
- _total_available_ratio = 0;
- _min_available_ratio = 100;
- _max_available_ratio = 0;
- _status_updated_dir_nodes.clear();
- }
-
dir_node *get_dir_node(const std::string &subdir) const;
- // when visit the tag/storage of the _dir_nodes map, there's no need to protect by the lock.
- // but when visit the holding_replicas, you must take care.
+ // TODO(wangdan): _dir_nodes should be protected by lock since add_new_disk are supported:
+ // it might be updated arbitrarily at any time.
+ //
+ // Especially when visiting the holding_replicas, you must take care.
mutable zrwlock_nr _lock;
int64_t _total_capacity_mb = 0;
int64_t _total_available_mb = 0;
- int _total_available_ratio = 0;
- int _min_available_ratio = 100;
- int _max_available_ratio = 0;
std::vector<std::shared_ptr<dir_node>> _dir_nodes;
std::vector<std::string> _available_data_dirs;
@@ -135,12 +149,6 @@ private:
// in this round
std::vector<std::shared_ptr<dir_node>> _status_updated_dir_nodes;
- perf_counter_wrapper _counter_total_capacity_mb;
- perf_counter_wrapper _counter_total_available_mb;
- perf_counter_wrapper _counter_total_available_ratio;
- perf_counter_wrapper _counter_min_available_ratio;
- perf_counter_wrapper _counter_max_available_ratio;
-
friend class replica_test;
friend class replica_stub;
friend class mock_replica_stub;
diff --git a/src/common/test/CMakeLists.txt b/src/common/test/CMakeLists.txt
index 78d94000c..74a9cdf3e 100644
--- a/src/common/test/CMakeLists.txt
+++ b/src/common/test/CMakeLists.txt
@@ -27,6 +27,7 @@ set(MY_PROJ_NAME dsn_replication_common_test)
set(MY_SRC_SEARCH_MODE "GLOB")
set(MY_PROJ_LIBS
+ dsn_http
dsn_replication_common
dsn_runtime
gtest
diff --git a/src/meta/test/misc/misc.cpp b/src/meta/test/misc/misc.cpp
index 5733dd651..36a20806d 100644
--- a/src/meta/test/misc/misc.cpp
+++ b/src/meta/test/misc/misc.cpp
@@ -203,7 +203,7 @@ void generate_node_fs_manager(const app_mapper &apps,
for (const auto &kv : nodes) {
const node_state &ns = kv.second;
if (nfm.find(ns.addr()) == nfm.end()) {
- nfm.emplace(ns.addr(), std::make_shared<fs_manager>(true));
+ nfm.emplace(ns.addr(), std::make_shared<fs_manager>());
}
fs_manager &manager = *(nfm.find(ns.addr())->second);
manager.initialize(data_dirs, tags, true);
diff --git a/src/replica/replica_stub.cpp b/src/replica/replica_stub.cpp
index 2b3030295..990ab1bd1 100644
--- a/src/replica/replica_stub.cpp
+++ b/src/replica/replica_stub.cpp
@@ -194,7 +194,6 @@ replica_stub::replica_stub(replica_state_subscriber subscriber /*= nullptr*/,
_mem_release_max_reserved_mem_percentage(10),
_max_concurrent_bulk_load_downloading_count(5),
_learn_app_concurrent_count(0),
- _fs_manager(false),
_bulk_load_downloading_count(0),
_manual_emergency_checkpointing_count(0),
_is_running(false)
diff --git a/src/utils/metrics.h b/src/utils/metrics.h
index e69268006..a230aa1f9 100644
--- a/src/utils/metrics.h
+++ b/src/utils/metrics.h
@@ -165,6 +165,7 @@ class error_code;
_##name(METRIC_##name.instantiate(entity##_metric_entity(), ##__VA_ARGS__))
#define METRIC_VAR_INIT_replica(name, ...) METRIC_VAR_INIT(name, replica, ##__VA_ARGS__)
#define METRIC_VAR_INIT_server(name, ...) METRIC_VAR_INIT(name, server, ##__VA_ARGS__)
+#define METRIC_VAR_INIT_disk(name, ...) METRIC_VAR_INIT(name, disk, ##__VA_ARGS__)
// Perform increment-related operations on metrics including gauge and counter.
#define METRIC_VAR_INCREMENT_BY(name, x) \
@@ -194,6 +195,11 @@ class error_code;
#define METRIC_VAR_AUTO_LATENCY_DURATION_NS(name) __##name##_auto_latency.duration_ns()
+#define METRIC_DEFINE_SET_METHOD(name, value_type) \
+ void set_##name(value_type value) { METRIC_VAR_SET(name, value); }
+
+#define METRIC_CALL_SET_METHOD(obj, name, value) obj.set_##name(value)
+
namespace dsn {
class metric; // IWYU pragma: keep
class metric_entity_prototype; // IWYU pragma: keep
@@ -614,6 +620,7 @@ enum class metric_unit : size_t
kBytes,
kMegaBytes,
kCapacityUnits,
+ kPercent,
kRequests,
kSeeks,
kPointLookups,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 10/12: feat(new_metrics): migrate built-in server-level metrics (#1418)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit a6664d6e44b7c4a9c202e451ca2e5e5c6b666267
Author: Dan Wang <wa...@apache.org>
AuthorDate: Wed Mar 29 16:17:54 2023 +0800
feat(new_metrics): migrate built-in server-level metrics (#1418)
https://github.com/apache/incubator-pegasus/issues/1321
Migrate built-in server-level metrics to new framework, including the total
amount of virtual/physical memory usage in MB. Both metrics are wrapped
in `builtin_metrics` and updated periodically by a timer, which is started
or stopped along with meta/replica servers.
Since `dsn_perf_counter_test` involves both metrics which have been
removed from perf-counters, just disable it.
---
.github/workflows/lint_and_test_cpp.yaml | 18 ++++-
run.sh | 4 +-
src/perf_counter/builtin_counters.cpp | 57 --------------
src/perf_counter/perf_counters.cpp | 3 -
src/server/pegasus_service_app.h | 58 ++++++++------
src/utils/builtin_metrics.cpp | 92 ++++++++++++++++++++++
.../builtin_counters.h => utils/builtin_metrics.h} | 33 +++++---
src/utils/metrics.cpp | 2 +-
8 files changed, 169 insertions(+), 98 deletions(-)
diff --git a/.github/workflows/lint_and_test_cpp.yaml b/.github/workflows/lint_and_test_cpp.yaml
index 77b5a00bd..3683dd86c 100644
--- a/.github/workflows/lint_and_test_cpp.yaml
+++ b/.github/workflows/lint_and_test_cpp.yaml
@@ -184,7 +184,9 @@ jobs:
- dsn_meta_state_tests
- dsn.meta.test
- dsn_nfs_test
- - dsn_perf_counter_test
+ # TODO(wangdan): Since builtin_counters (memused.virt and memused.res) for perf-counters
+ # have been removed and dsn_perf_counter_test depends on them, disable it.
+ # - dsn_perf_counter_test
- dsn_replica_backup_test
- dsn_replica_bulk_load_test
- dsn_replica_dup_test
@@ -310,7 +312,9 @@ jobs:
- dsn_meta_state_tests
- dsn.meta.test
- dsn_nfs_test
- - dsn_perf_counter_test
+ # TODO(wangdan): Since builtin_counters (memused.virt and memused.res) for perf-counters
+ # have been removed and dsn_perf_counter_test depends on them, disable it.
+ # - dsn_perf_counter_test
- dsn_replica_backup_test
- dsn_replica_bulk_load_test
- dsn_replica_dup_test
@@ -429,7 +433,11 @@ jobs:
# - base_api_test
# - base_test
# - bulk_load_test
-# - detect_hotspot_test
+# # TODO(wangdan): Since the hotspot detection depends on the perf-counters system which
+# # is being replaced with the new metrics system, its test will fail. Temporarily disable
+# # the test and re-enable it after the hotspot detection is migrated to the new metrics
+# # system.
+# # - detect_hotspot_test
# - dsn_aio_test
# - dsn_block_service_test
# - dsn.failure_detector.tests
@@ -437,7 +445,9 @@ jobs:
# - dsn_meta_state_tests
# - dsn.meta.test
# - dsn_nfs_test
-# - dsn_perf_counter_test
+# # TODO(wangdan): Since builtin_counters (memused.virt and memused.res) for perf-counters
+# # have been removed and dsn_perf_counter_test depends on them, disable it.
+# # - dsn_perf_counter_test
# - dsn_replica_backup_test
# - dsn_replica_bulk_load_test
# - dsn_replica_dup_test
diff --git a/run.sh b/run.sh
index ecf64a616..46ce47141 100755
--- a/run.sh
+++ b/run.sh
@@ -354,7 +354,9 @@ function run_test()
dsn_meta_state_tests
dsn.meta.test
dsn_nfs_test
- dsn_perf_counter_test
+ # TODO(wangdan): Since builtin_counters (memused.virt and memused.res) for perf-counters
+ # have been removed and dsn_perf_counter_test depends on them, disable it.
+ # dsn_perf_counter_test
dsn_replica_backup_test
dsn_replica_bulk_load_test
dsn_replica_dup_test
diff --git a/src/perf_counter/builtin_counters.cpp b/src/perf_counter/builtin_counters.cpp
deleted file mode 100644
index 26f57c566..000000000
--- a/src/perf_counter/builtin_counters.cpp
+++ /dev/null
@@ -1,57 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#include "builtin_counters.h"
-
-#include <stdint.h>
-
-#include "perf_counter/perf_counter.h"
-#include "perf_counter/perf_counter_wrapper.h"
-#include "utils/fmt_logging.h"
-#include "utils/process_utils.h"
-
-namespace dsn {
-
-builtin_counters::builtin_counters()
-{
- _memused_virt.init_global_counter("replica",
- "server",
- "memused.virt(MB)",
- COUNTER_TYPE_NUMBER,
- "virtual memory usages in MB");
- _memused_res.init_global_counter("replica",
- "server",
- "memused.res(MB)",
- COUNTER_TYPE_NUMBER,
- "physically memory usages in MB");
-}
-
-builtin_counters::~builtin_counters() {}
-
-void builtin_counters::update_counters()
-{
- double vm_usage;
- double resident_set;
- utils::process_mem_usage(vm_usage, resident_set);
- uint64_t memused_virt = (uint64_t)vm_usage / 1024;
- uint64_t memused_res = (uint64_t)resident_set / 1024;
- _memused_virt->set(memused_virt);
- _memused_res->set(memused_res);
- LOG_INFO("memused_virt = {} MB, memused_res = {} MB", memused_virt, memused_res);
-}
-
-} // namespace dsn
diff --git a/src/perf_counter/perf_counters.cpp b/src/perf_counter/perf_counters.cpp
index 1f2d4574b..49482fbd2 100644
--- a/src/perf_counter/perf_counters.cpp
+++ b/src/perf_counter/perf_counters.cpp
@@ -32,7 +32,6 @@
#include <sstream>
#include <utility>
-#include "builtin_counters.h"
#include "perf_counter/perf_counter.h"
#include "perf_counter/perf_counter_atomic.h"
#include "perf_counter/perf_counter_utils.h"
@@ -367,8 +366,6 @@ std::string perf_counters::list_snapshot_by_literal(
void perf_counters::take_snapshot()
{
- builtin_counters::instance().update_counters();
-
std::vector<perf_counter_ptr> all_counters;
get_all_counters(&all_counters);
diff --git a/src/server/pegasus_service_app.h b/src/server/pegasus_service_app.h
index bae690114..c581ecae0 100644
--- a/src/server/pegasus_service_app.h
+++ b/src/server/pegasus_service_app.h
@@ -24,6 +24,7 @@
#include <pegasus/version.h>
#include <pegasus/git_commit.h>
#include "reporter/pegasus_counter_reporter.h"
+#include "utils/builtin_metrics.h"
namespace pegasus {
namespace server {
@@ -32,8 +33,7 @@ class pegasus_replication_service_app : public ::dsn::replication::replication_s
{
public:
pegasus_replication_service_app(const dsn::service_app_info *info)
- : ::dsn::replication::replication_service_app::replication_service_app(info),
- _updater_started(false)
+ : ::dsn::replication::replication_service_app::replication_service_app(info)
{
}
@@ -43,33 +43,39 @@ public:
std::vector<std::string> args_new(args);
args_new.emplace_back(PEGASUS_VERSION);
args_new.emplace_back(PEGASUS_GIT_COMMIT);
- ::dsn::error_code ret = ::dsn::replication::replication_service_app::start(args_new);
- if (ret == ::dsn::ERR_OK) {
- pegasus_counter_reporter::instance().start();
- _updater_started = true;
- }
- return ret;
+ // Actually the root caller, start_app() in service_control_task::exec() will also do
+ // CHECK for ERR_OK. Do CHECK here to guarantee that all following services (such as
+ // built-in metrics) are started.
+ CHECK_EQ(::dsn::replication::replication_service_app::start(args_new), ::dsn::ERR_OK);
+
+ // TODO(wangdan): remove after all metrics have been migrated.
+ pegasus_counter_reporter::instance().start();
+
+ _builtin_metrics.start();
+ return ::dsn::ERR_OK;
}
virtual ::dsn::error_code stop(bool cleanup = false) override
{
::dsn::error_code ret = ::dsn::replication::replication_service_app::stop();
- if (_updater_started) {
- pegasus_counter_reporter::instance().stop();
- }
+
+ // TODO(wangdan): remove after all metrics have been migrated.
+ pegasus_counter_reporter::instance().stop();
+
+ _builtin_metrics.stop();
return ret;
}
private:
- bool _updater_started;
+ dsn::builtin_metrics _builtin_metrics;
};
class pegasus_meta_service_app : public ::dsn::service::meta_service_app
{
public:
pegasus_meta_service_app(const dsn::service_app_info *info)
- : ::dsn::service::meta_service_app::meta_service_app(info), _updater_started(false)
+ : ::dsn::service::meta_service_app::meta_service_app(info)
{
}
@@ -79,26 +85,32 @@ public:
std::vector<std::string> args_new(args);
args_new.emplace_back(PEGASUS_VERSION);
args_new.emplace_back(PEGASUS_GIT_COMMIT);
- ::dsn::error_code ret = ::dsn::service::meta_service_app::start(args_new);
- if (ret == ::dsn::ERR_OK) {
- pegasus_counter_reporter::instance().start();
- _updater_started = true;
- }
- return ret;
+ // Actually the root caller, start_app() in service_control_task::exec() will also do
+ // CHECK for ERR_OK. Do CHECK here to guarantee that all following services (such as
+ // built-in metrics) are started.
+ CHECK_EQ(::dsn::service::meta_service_app::start(args_new), ::dsn::ERR_OK);
+
+ // TODO(wangdan): remove after all metrics have been migrated.
+ pegasus_counter_reporter::instance().start();
+
+ _builtin_metrics.start();
+ return ::dsn::ERR_OK;
}
virtual ::dsn::error_code stop(bool cleanup = false) override
{
::dsn::error_code ret = ::dsn::service::meta_service_app::stop();
- if (_updater_started) {
- pegasus_counter_reporter::instance().stop();
- }
+
+ // TODO(wangdan): remove after all metrics have been migrated.
+ pegasus_counter_reporter::instance().stop();
+
+ _builtin_metrics.stop();
return ret;
}
private:
- bool _updater_started;
+ dsn::builtin_metrics _builtin_metrics;
};
} // namespace server
diff --git a/src/utils/builtin_metrics.cpp b/src/utils/builtin_metrics.cpp
new file mode 100644
index 000000000..af79fc8b4
--- /dev/null
+++ b/src/utils/builtin_metrics.cpp
@@ -0,0 +1,92 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "utils/builtin_metrics.h"
+
+#include <stdint.h>
+#include <functional>
+
+#include "utils/autoref_ptr.h"
+#include "utils/flags.h"
+#include "utils/fmt_logging.h"
+#include "utils/process_utils.h"
+#include "utils/string_view.h"
+
+METRIC_DEFINE_gauge_int64(server,
+ virtual_mem_usage_mb,
+ dsn::metric_unit::kMegaBytes,
+ "The total amount of virtual memory usage in MB");
+
+METRIC_DEFINE_gauge_int64(server,
+ resident_mem_usage_mb,
+ dsn::metric_unit::kMegaBytes,
+ "The total amount of physical memory usage in MB");
+
+namespace dsn {
+
+DSN_DEFINE_uint64(metrics,
+ builtin_metrics_update_interval_ms,
+ 10 * 1000,
+ "The interval (milliseconds) at which builtin metrics are updated.");
+
+builtin_metrics::builtin_metrics()
+ : METRIC_VAR_INIT_server(virtual_mem_usage_mb), METRIC_VAR_INIT_server(resident_mem_usage_mb)
+{
+}
+
+builtin_metrics::~builtin_metrics()
+{
+ CHECK(!_timer, "timer should have been destroyed by stop()");
+}
+
+void builtin_metrics::on_close() {}
+
+void builtin_metrics::start()
+{
+ CHECK(!_timer, "timer should not have been initialized before start()");
+
+ _timer.reset(new metric_timer(FLAGS_builtin_metrics_update_interval_ms,
+ std::bind(&builtin_metrics::update, this),
+ std::bind(&builtin_metrics::on_close, this)));
+}
+
+void builtin_metrics::stop()
+{
+ CHECK(_timer, "timer should have been initialized before stop()");
+
+ // Close the timer synchronously.
+ _timer->close();
+ _timer->wait();
+
+ // Reset the timer to mark that it has been stopped, now it could be started.
+ _timer.reset();
+}
+
+void builtin_metrics::update()
+{
+ double vm_usage;
+ double resident_set;
+ utils::process_mem_usage(vm_usage, resident_set);
+
+ auto virt_mb = static_cast<uint64_t>(vm_usage) >> 10;
+ auto res_mb = static_cast<uint64_t>(resident_set) >> 10;
+ METRIC_VAR_SET(virtual_mem_usage_mb, virt_mb);
+ METRIC_VAR_SET(resident_mem_usage_mb, res_mb);
+ LOG_INFO("virt = {} MB, res = {} MB", virt_mb, res_mb);
+}
+
+} // namespace dsn
diff --git a/src/perf_counter/builtin_counters.h b/src/utils/builtin_metrics.h
similarity index 64%
rename from src/perf_counter/builtin_counters.h
rename to src/utils/builtin_metrics.h
index 58ca9f778..2a0094d70 100644
--- a/src/perf_counter/builtin_counters.h
+++ b/src/utils/builtin_metrics.h
@@ -15,19 +15,34 @@
// specific language governing permissions and limitations
// under the License.
-#include "perf_counter_wrapper.h"
-#include "utils/singleton.h"
+#pragma once
+
+#include <memory>
+
+#include "utils/metrics.h"
+#include "utils/ports.h"
namespace dsn {
-class builtin_counters : public dsn::utils::singleton<builtin_counters>
+
+class builtin_metrics
{
public:
- builtin_counters();
- ~builtin_counters();
- void update_counters();
+ builtin_metrics();
+ ~builtin_metrics();
+
+ void start();
+ void stop();
private:
- dsn::perf_counter_wrapper _memused_virt;
- dsn::perf_counter_wrapper _memused_res;
+ void on_close();
+ void update();
+
+ METRIC_VAR_DECLARE_gauge_int64(virtual_mem_usage_mb);
+ METRIC_VAR_DECLARE_gauge_int64(resident_mem_usage_mb);
+
+ std::unique_ptr<metric_timer> _timer;
+
+ DISALLOW_COPY_AND_ASSIGN(builtin_metrics);
};
-}
+
+} // namespace dsn
diff --git a/src/utils/metrics.cpp b/src/utils/metrics.cpp
index b32887c55..dd4c5dd56 100644
--- a/src/utils/metrics.cpp
+++ b/src/utils/metrics.cpp
@@ -43,7 +43,7 @@ namespace dsn {
DSN_DEFINE_uint64(metrics,
entity_retirement_delay_ms,
10 * 60 * 1000,
- "The retention internal (milliseconds) for an entity after it becomes stale.");
+ "The retention interval (milliseconds) for an entity after it becomes stale.");
metric_entity::metric_entity(const metric_entity_prototype *prototype,
const std::string &id,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 01/12: feat(new_metrics): add replica-level metric entity (#1345)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit 7bf18e6c5b3e1645984d158fc18bf0063d069b90
Author: Dan Wang <wa...@apache.org>
AuthorDate: Mon Feb 13 15:45:40 2023 +0800
feat(new_metrics): add replica-level metric entity (#1345)
---
src/replica/replica_base.cpp | 52 ++++++++++++++++++++++++++++++++++++++++++++
src/replica/replica_base.h | 10 +++++----
2 files changed, 58 insertions(+), 4 deletions(-)
diff --git a/src/replica/replica_base.cpp b/src/replica/replica_base.cpp
new file mode 100644
index 000000000..11e08ae05
--- /dev/null
+++ b/src/replica/replica_base.cpp
@@ -0,0 +1,52 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "replica_base.h"
+
+#include <fmt/core.h>
+
+METRIC_DEFINE_entity(replica);
+
+namespace dsn {
+namespace replication {
+
+namespace {
+
+metric_entity_ptr instantiate_replica_metric_entity(const gpid &id)
+{
+ auto entity_id = fmt::format("replica_{}", id);
+
+ // Do NOT add `replica_base._app_name` as the table name to the attributes of entity, since
+ // it is read-only and will never be updated even if the table is renamed.
+ return METRIC_ENTITY_replica.instantiate(
+ entity_id,
+ {{"table_id", std::to_string(id.get_app_id())},
+ {"partition_id", std::to_string(id.get_partition_index())}});
+}
+
+} // anonymous namespace
+
+replica_base::replica_base(gpid id, string_view name, string_view app_name)
+ : _gpid(id),
+ _name(name),
+ _app_name(app_name),
+ _replica_metric_entity(instantiate_replica_metric_entity(id))
+{
+}
+
+} // namespace replication
+} // namespace dsn
diff --git a/src/replica/replica_base.h b/src/replica/replica_base.h
index 64f294e87..88202d055 100644
--- a/src/replica/replica_base.h
+++ b/src/replica/replica_base.h
@@ -27,6 +27,7 @@
#pragma once
#include "common/gpid.h"
+#include "utils/metrics.h"
#include "utils/string_view.h"
namespace dsn {
@@ -35,10 +36,7 @@ namespace replication {
/// Base class for types that are one-instance-per-replica.
struct replica_base
{
- replica_base(gpid id, string_view name, string_view app_name)
- : _gpid(id), _name(name), _app_name(app_name)
- {
- }
+ replica_base(gpid id, string_view name, string_view app_name);
explicit replica_base(replica_base *rhs)
: replica_base(rhs->get_gpid(), rhs->replica_name(), rhs->_app_name)
@@ -53,10 +51,14 @@ struct replica_base
const char *log_prefix() const { return _name.c_str(); }
+ const metric_entity_ptr &replica_metric_entity() const { return _replica_metric_entity; }
+
private:
const gpid _gpid;
const std::string _name;
+ // TODO(wangdan): drop `_app_name` or make it changeable, since a table could be renamed.
const std::string _app_name;
+ const metric_entity_ptr _replica_metric_entity;
};
} // namespace replication
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 02/12: feat(new_metrics): migrate replica-level metrics for write service (#1351)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit 7dfb988259269e3aaeee69b18eca7818ce8e48e9
Author: Dan Wang <wa...@apache.org>
AuthorDate: Thu Feb 23 23:18:20 2023 +0800
feat(new_metrics): migrate replica-level metrics for write service (#1351)
---
.github/workflows/lint_and_test_cpp.yaml | 12 +-
run.sh | 6 +-
src/replica/replica_base.h | 9 +-
src/server/pegasus_server_write.cpp | 21 +-
src/server/pegasus_server_write.h | 3 +-
src/server/pegasus_write_service.cpp | 284 ++++++++++++++------------
src/server/pegasus_write_service.h | 47 +++--
src/server/test/pegasus_server_write_test.cpp | 4 +-
src/utils/metrics.h | 135 +++++++++++-
src/utils/test/metrics_test.cpp | 147 ++++++++++++-
src/utils/time_utils.h | 28 ++-
11 files changed, 513 insertions(+), 183 deletions(-)
diff --git a/.github/workflows/lint_and_test_cpp.yaml b/.github/workflows/lint_and_test_cpp.yaml
index afc56ebfc..77b5a00bd 100644
--- a/.github/workflows/lint_and_test_cpp.yaml
+++ b/.github/workflows/lint_and_test_cpp.yaml
@@ -172,7 +172,11 @@ jobs:
- base_api_test
- base_test
- bulk_load_test
- - detect_hotspot_test
+ # TODO(wangdan): Since the hotspot detection depends on the perf-counters system which
+ # is being replaced with the new metrics system, its test will fail. Temporarily disable
+ # the test and re-enable it after the hotspot detection is migrated to the new metrics
+ # system.
+ # - detect_hotspot_test
- dsn_aio_test
- dsn_block_service_test
- dsn.failure_detector.tests
@@ -294,7 +298,11 @@ jobs:
- base_api_test
- base_test
- bulk_load_test
- - detect_hotspot_test
+ # TODO(wangdan): Since the hotspot detection depends on the perf-counters system which
+ # is being replaced with the new metrics system, its test will fail. Temporarily disable
+ # the test and re-enable it after the hotspot detection is migrated to the new metrics
+ # system.
+ # - detect_hotspot_test
- dsn_aio_test
- dsn_block_service_test
- dsn.failure_detector.tests
diff --git a/run.sh b/run.sh
index ff14a569c..ecf64a616 100755
--- a/run.sh
+++ b/run.sh
@@ -342,7 +342,11 @@ function run_test()
base_api_test
base_test
bulk_load_test
- detect_hotspot_test
+ # TODO(wangdan): Since the hotspot detection depends on the perf-counters system which
+ # is being replaced with the new metrics system, its test will fail. Temporarily disable
+ # the test and re-enable it after the hotspot detection is migrated to the new metrics
+ # system.
+ # detect_hotspot_test
dsn_aio_test
dsn_block_service_test
dsn.failure_detector.tests
diff --git a/src/replica/replica_base.h b/src/replica/replica_base.h
index 88202d055..7c5b7747e 100644
--- a/src/replica/replica_base.h
+++ b/src/replica/replica_base.h
@@ -51,7 +51,14 @@ struct replica_base
const char *log_prefix() const { return _name.c_str(); }
- const metric_entity_ptr &replica_metric_entity() const { return _replica_metric_entity; }
+ const metric_entity_ptr &replica_metric_entity() const
+ {
+ CHECK_NOTNULL(_replica_metric_entity,
+ "replica metric entity should has been instantiated: "
+ "uninitialized entity cannot be used to instantiate "
+ "metric");
+ return _replica_metric_entity;
+ }
private:
const gpid _gpid;
diff --git a/src/server/pegasus_server_write.cpp b/src/server/pegasus_server_write.cpp
index 0147050a6..e6cb5331d 100644
--- a/src/server/pegasus_server_write.cpp
+++ b/src/server/pegasus_server_write.cpp
@@ -31,7 +31,6 @@
#include "pegasus_server_impl.h"
#include "pegasus_server_write.h"
#include "pegasus_utils.h"
-#include "perf_counter/perf_counter.h"
#include "rrdb/rrdb.code.definition.h"
#include "runtime/rpc/rpc_address.h"
#include "runtime/rpc/rpc_holder.h"
@@ -42,20 +41,20 @@
#include "utils/fmt_logging.h"
#include "utils/ports.h"
+METRIC_DEFINE_counter(replica,
+ corrupt_writes,
+ dsn::metric_unit::kRequests,
+ "The number of corrupt writes for each replica");
+
namespace pegasus {
namespace server {
DSN_DECLARE_bool(rocksdb_verbose_log);
pegasus_server_write::pegasus_server_write(pegasus_server_impl *server)
- : replica_base(server), _write_svc(new pegasus_write_service(server))
+ : replica_base(server),
+ _write_svc(new pegasus_write_service(server)),
+ METRIC_VAR_INIT_replica(corrupt_writes)
{
- char name[256];
- snprintf(name, 255, "recent_corrupt_write_count@%s", get_gpid().to_string());
- _pfc_recent_corrupt_write_count.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_VOLATILE_NUMBER,
- "statistic the recent corrupt write count");
-
init_non_batch_write_handlers();
}
@@ -81,7 +80,7 @@ int pegasus_server_write::on_batched_write_requests(dsn::message_ex **requests,
return iter->second(requests[0]);
}
} catch (TTransportException &ex) {
- _pfc_recent_corrupt_write_count->increment();
+ METRIC_VAR_INCREMENT(corrupt_writes);
LOG_ERROR_PREFIX("pegasus not batch write handler failed, from = {}, exception = {}",
requests[0]->header->from_address.to_string(),
ex.what());
@@ -125,7 +124,7 @@ int pegasus_server_write::on_batched_writes(dsn::message_ex **requests, int coun
}
}
} catch (TTransportException &ex) {
- _pfc_recent_corrupt_write_count->increment();
+ METRIC_VAR_INCREMENT(corrupt_writes);
LOG_ERROR_PREFIX("pegasus batch writes handler failed, from = {}, exception = {}",
requests[i]->header->from_address.to_string(),
ex.what());
diff --git a/src/server/pegasus_server_write.h b/src/server/pegasus_server_write.h
index 6a002c299..d8a358164 100644
--- a/src/server/pegasus_server_write.h
+++ b/src/server/pegasus_server_write.h
@@ -27,7 +27,6 @@
#include "base/pegasus_rpc_types.h"
#include "pegasus_write_service.h"
-#include "perf_counter/perf_counter_wrapper.h"
#include "replica/replica_base.h"
#include "rrdb/rrdb_types.h"
#include "runtime/task/task_code.h"
@@ -102,7 +101,7 @@ private:
typedef std::map<dsn::task_code, std::function<int(dsn::message_ex *)>> non_batch_writes_map;
non_batch_writes_map _non_batch_write_handlers;
- ::dsn::perf_counter_wrapper _pfc_recent_corrupt_write_count;
+ METRIC_VAR_DECLARE_counter(corrupt_writes);
};
} // namespace server
diff --git a/src/server/pegasus_write_service.cpp b/src/server/pegasus_write_service.cpp
index 73f6cc8d6..4889329d9 100644
--- a/src/server/pegasus_write_service.cpp
+++ b/src/server/pegasus_write_service.cpp
@@ -52,6 +52,93 @@ class blob;
class message_ex;
} // namespace dsn
+METRIC_DEFINE_counter(replica,
+ put_requests,
+ dsn::metric_unit::kRequests,
+ "The number of PUT requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ multi_put_requests,
+ dsn::metric_unit::kRequests,
+ "The number of MULTI_PUT requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ remove_requests,
+ dsn::metric_unit::kRequests,
+ "The number of REMOVE requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ multi_remove_requests,
+ dsn::metric_unit::kRequests,
+ "The number of MULTI_REMOVE requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ incr_requests,
+ dsn::metric_unit::kRequests,
+ "The number of INCR requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ check_and_set_requests,
+ dsn::metric_unit::kRequests,
+ "The number of CHECK_AND_SET requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ check_and_mutate_requests,
+ dsn::metric_unit::kRequests,
+ "The number of CHECK_AND_MUTATE requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ put_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of PUT requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ multi_put_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of MULTI_PUT requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ remove_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of REMOVE requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ multi_remove_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of MULTI_REMOVE requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ incr_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of INCR requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ check_and_set_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of CHECK_AND_SET requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ check_and_mutate_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of CHECK_AND_MUTATE requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ dup_requests,
+ dsn::metric_unit::kRequests,
+ "The number of DUPLICATE requests for each replica");
+
+METRIC_DEFINE_percentile_int64(
+ replica,
+ dup_time_lag_ms,
+ dsn::metric_unit::kMilliSeconds,
+ "the time lag (in ms) between master and slave in the duplication for each replica");
+
+METRIC_DEFINE_counter(
+ replica,
+ dup_lagging_writes,
+ dsn::metric_unit::kRequests,
+ "the number of lagging writes (time lag larger than `dup_lagging_write_threshold_ms`)");
+
namespace pegasus {
namespace server {
@@ -68,105 +155,33 @@ pegasus_write_service::pegasus_write_service(pegasus_server_impl *server)
_server(server),
_impl(new impl(server)),
_batch_start_time(0),
- _cu_calculator(server->_cu_calculator.get())
+ _cu_calculator(server->_cu_calculator.get()),
+ METRIC_VAR_INIT_replica(put_requests),
+ METRIC_VAR_INIT_replica(multi_put_requests),
+ METRIC_VAR_INIT_replica(remove_requests),
+ METRIC_VAR_INIT_replica(multi_remove_requests),
+ METRIC_VAR_INIT_replica(incr_requests),
+ METRIC_VAR_INIT_replica(check_and_set_requests),
+ METRIC_VAR_INIT_replica(check_and_mutate_requests),
+ METRIC_VAR_INIT_replica(put_latency_ns),
+ METRIC_VAR_INIT_replica(multi_put_latency_ns),
+ METRIC_VAR_INIT_replica(remove_latency_ns),
+ METRIC_VAR_INIT_replica(multi_remove_latency_ns),
+ METRIC_VAR_INIT_replica(incr_latency_ns),
+ METRIC_VAR_INIT_replica(check_and_set_latency_ns),
+ METRIC_VAR_INIT_replica(check_and_mutate_latency_ns),
+ METRIC_VAR_INIT_replica(dup_requests),
+ METRIC_VAR_INIT_replica(dup_time_lag_ms),
+ METRIC_VAR_INIT_replica(dup_lagging_writes),
+ _put_batch_size(0),
+ _remove_batch_size(0)
{
- std::string str_gpid = fmt::format("{}", server->get_gpid());
-
- std::string name;
-
- name = fmt::format("put_qps@{}", str_gpid);
- _pfc_put_qps.init_app_counter(
- "app.pegasus", name.c_str(), COUNTER_TYPE_RATE, "statistic the qps of PUT request");
-
- name = fmt::format("multi_put_qps@{}", str_gpid);
- _pfc_multi_put_qps.init_app_counter(
- "app.pegasus", name.c_str(), COUNTER_TYPE_RATE, "statistic the qps of MULTI_PUT request");
-
- name = fmt::format("remove_qps@{}", str_gpid);
- _pfc_remove_qps.init_app_counter(
- "app.pegasus", name.c_str(), COUNTER_TYPE_RATE, "statistic the qps of REMOVE request");
-
- name = fmt::format("multi_remove_qps@{}", str_gpid);
- _pfc_multi_remove_qps.init_app_counter("app.pegasus",
- name.c_str(),
- COUNTER_TYPE_RATE,
- "statistic the qps of MULTI_REMOVE request");
-
- name = fmt::format("incr_qps@{}", str_gpid);
- _pfc_incr_qps.init_app_counter(
- "app.pegasus", name.c_str(), COUNTER_TYPE_RATE, "statistic the qps of INCR request");
-
- name = fmt::format("check_and_set_qps@{}", str_gpid);
- _pfc_check_and_set_qps.init_app_counter("app.pegasus",
- name.c_str(),
- COUNTER_TYPE_RATE,
- "statistic the qps of CHECK_AND_SET request");
-
- name = fmt::format("check_and_mutate_qps@{}", str_gpid);
- _pfc_check_and_mutate_qps.init_app_counter("app.pegasus",
- name.c_str(),
- COUNTER_TYPE_RATE,
- "statistic the qps of CHECK_AND_MUTATE request");
-
- name = fmt::format("put_latency@{}", str_gpid);
- _pfc_put_latency.init_app_counter("app.pegasus",
- name.c_str(),
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of PUT request");
-
- name = fmt::format("multi_put_latency@{}", str_gpid);
- _pfc_multi_put_latency.init_app_counter("app.pegasus",
- name.c_str(),
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of MULTI_PUT request");
-
- name = fmt::format("remove_latency@{}", str_gpid);
- _pfc_remove_latency.init_app_counter("app.pegasus",
- name.c_str(),
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of REMOVE request");
-
- name = fmt::format("multi_remove_latency@{}", str_gpid);
- _pfc_multi_remove_latency.init_app_counter("app.pegasus",
- name.c_str(),
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of MULTI_REMOVE request");
-
- name = fmt::format("incr_latency@{}", str_gpid);
- _pfc_incr_latency.init_app_counter("app.pegasus",
- name.c_str(),
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of INCR request");
-
- name = fmt::format("check_and_set_latency@{}", str_gpid);
- _pfc_check_and_set_latency.init_app_counter("app.pegasus",
- name.c_str(),
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of CHECK_AND_SET request");
-
- name = fmt::format("check_and_mutate_latency@{}", str_gpid);
- _pfc_check_and_mutate_latency.init_app_counter(
- "app.pegasus",
- name.c_str(),
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of CHECK_AND_MUTATE request");
-
- _pfc_duplicate_qps.init_app_counter("app.pegasus",
- fmt::format("duplicate_qps@{}", str_gpid).c_str(),
- COUNTER_TYPE_RATE,
- "statistic the qps of DUPLICATE requests");
-
- _pfc_dup_time_lag.init_app_counter(
- "app.pegasus",
- fmt::format("dup.time_lag_ms@{}", app_name()).c_str(),
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "the time (in ms) lag between master and slave in the duplication");
-
- _pfc_dup_lagging_writes.init_app_counter(
- "app.pegasus",
- fmt::format("dup.lagging_writes@{}", app_name()).c_str(),
- COUNTER_TYPE_VOLATILE_NUMBER,
- "the number of lagging writes (time lag larger than `dup_lagging_write_threshold_ms`)");
+ _dup_lagging_write_threshold_ms = dsn_config_get_value_int64(
+ "pegasus.server",
+ "dup_lagging_write_threshold_ms",
+ 10 * 1000,
+ "If the duration that a write flows from master to slave is larger than this threshold, "
+ "the write is defined a lagging write.");
}
pegasus_write_service::~pegasus_write_service() {}
@@ -177,15 +192,15 @@ int pegasus_write_service::multi_put(const db_write_context &ctx,
const dsn::apps::multi_put_request &update,
dsn::apps::update_response &resp)
{
- uint64_t start_time = dsn_now_ns();
- _pfc_multi_put_qps->increment();
+ METRIC_VAR_AUTO_LATENCY(multi_put_latency_ns);
+ METRIC_VAR_INCREMENT(multi_put_requests);
+
int err = _impl->multi_put(ctx, update, resp);
if (_server->is_primary()) {
_cu_calculator->add_multi_put_cu(resp.error, update.hash_key, update.kvs);
}
- _pfc_multi_put_latency->set(dsn_now_ns() - start_time);
return err;
}
@@ -193,15 +208,15 @@ int pegasus_write_service::multi_remove(int64_t decree,
const dsn::apps::multi_remove_request &update,
dsn::apps::multi_remove_response &resp)
{
- uint64_t start_time = dsn_now_ns();
- _pfc_multi_remove_qps->increment();
+ METRIC_VAR_AUTO_LATENCY(multi_remove_latency_ns);
+ METRIC_VAR_INCREMENT(multi_remove_requests);
+
int err = _impl->multi_remove(decree, update, resp);
if (_server->is_primary()) {
_cu_calculator->add_multi_remove_cu(resp.error, update.hash_key, update.sort_keys);
}
- _pfc_multi_remove_latency->set(dsn_now_ns() - start_time);
return err;
}
@@ -209,15 +224,15 @@ int pegasus_write_service::incr(int64_t decree,
const dsn::apps::incr_request &update,
dsn::apps::incr_response &resp)
{
- uint64_t start_time = dsn_now_ns();
- _pfc_incr_qps->increment();
+ METRIC_VAR_AUTO_LATENCY(incr_latency_ns);
+ METRIC_VAR_INCREMENT(incr_requests);
+
int err = _impl->incr(decree, update, resp);
if (_server->is_primary()) {
_cu_calculator->add_incr_cu(resp.error, update.key);
}
- _pfc_incr_latency->set(dsn_now_ns() - start_time);
return err;
}
@@ -225,8 +240,9 @@ int pegasus_write_service::check_and_set(int64_t decree,
const dsn::apps::check_and_set_request &update,
dsn::apps::check_and_set_response &resp)
{
- uint64_t start_time = dsn_now_ns();
- _pfc_check_and_set_qps->increment();
+ METRIC_VAR_AUTO_LATENCY(check_and_set_latency_ns);
+ METRIC_VAR_INCREMENT(check_and_set_requests);
+
int err = _impl->check_and_set(decree, update, resp);
if (_server->is_primary()) {
@@ -237,7 +253,6 @@ int pegasus_write_service::check_and_set(int64_t decree,
update.set_value);
}
- _pfc_check_and_set_latency->set(dsn_now_ns() - start_time);
return err;
}
@@ -245,8 +260,9 @@ int pegasus_write_service::check_and_mutate(int64_t decree,
const dsn::apps::check_and_mutate_request &update,
dsn::apps::check_and_mutate_response &resp)
{
- uint64_t start_time = dsn_now_ns();
- _pfc_check_and_mutate_qps->increment();
+ METRIC_VAR_AUTO_LATENCY(check_and_mutate_latency_ns);
+ METRIC_VAR_INCREMENT(check_and_mutate_requests);
+
int err = _impl->check_and_mutate(decree, update, resp);
if (_server->is_primary()) {
@@ -254,7 +270,6 @@ int pegasus_write_service::check_and_mutate(int64_t decree,
resp.error, update.hash_key, update.check_sort_key, update.mutate_list);
}
- _pfc_check_and_mutate_latency->set(dsn_now_ns() - start_time);
return err;
}
@@ -272,8 +287,7 @@ int pegasus_write_service::batch_put(const db_write_context &ctx,
{
CHECK_GT_MSG(_batch_start_time, 0, "batch_put must be called after batch_prepare");
- _batch_qps_perfcounters.push_back(_pfc_put_qps.get());
- _batch_latency_perfcounters.push_back(_pfc_put_latency.get());
+ ++_put_batch_size;
int err = _impl->batch_put(ctx, update, resp);
if (_server->is_primary()) {
@@ -289,8 +303,7 @@ int pegasus_write_service::batch_remove(int64_t decree,
{
CHECK_GT_MSG(_batch_start_time, 0, "batch_remove must be called after batch_prepare");
- _batch_qps_perfcounters.push_back(_pfc_remove_qps.get());
- _batch_latency_perfcounters.push_back(_pfc_remove_latency.get());
+ ++_remove_batch_size;
int err = _impl->batch_remove(decree, key, resp);
if (_server->is_primary()) {
@@ -322,15 +335,21 @@ void pegasus_write_service::set_default_ttl(uint32_t ttl) { _impl->set_default_t
void pegasus_write_service::clear_up_batch_states()
{
- uint64_t latency = dsn_now_ns() - _batch_start_time;
- for (dsn::perf_counter *pfc : _batch_qps_perfcounters)
- pfc->increment();
- for (dsn::perf_counter *pfc : _batch_latency_perfcounters)
- pfc->set(latency);
-
- _batch_qps_perfcounters.clear();
- _batch_latency_perfcounters.clear();
+#define PROCESS_WRITE_BATCH(op) \
+ do { \
+ METRIC_VAR_INCREMENT_BY(op##_requests, static_cast<int64_t>(_##op##_batch_size)); \
+ METRIC_VAR_SET(op##_latency_ns, static_cast<size_t>(_##op##_batch_size), latency_ns); \
+ _##op##_batch_size = 0; \
+ } while (0)
+
+ auto latency_ns = static_cast<int64_t>(dsn_now_ns() - _batch_start_time);
+
+ PROCESS_WRITE_BATCH(put);
+ PROCESS_WRITE_BATCH(remove);
+
_batch_start_time = 0;
+
+#undef PROCESS_WRITE_BATCH
}
int pegasus_write_service::duplicate(int64_t decree,
@@ -350,14 +369,13 @@ int pegasus_write_service::duplicate(int64_t decree,
return empty_put(decree);
}
- _pfc_duplicate_qps->increment();
- auto cleanup = dsn::defer([this, &request]() {
- uint64_t latency_ms = (dsn_now_us() - request.timestamp) / 1000;
- if (latency_ms > FLAGS_dup_lagging_write_threshold_ms) {
- _pfc_dup_lagging_writes->increment();
- }
- _pfc_dup_time_lag->set(latency_ms);
- });
+ METRIC_VAR_INCREMENT(dup_requests);
+ METRIC_VAR_AUTO_LATENCY(
+ dup_time_lag_ms, request.timestamp * 1000, [this](uint64_t latency) {
+ if (latency > _dup_lagging_write_threshold_ms) {
+ METRIC_VAR_INCREMENT(dup_lagging_writes);
+ }
+ });
dsn::message_ex *write =
dsn::from_blob_to_received_msg(request.task_code, request.raw_message);
bool is_delete = request.task_code == dsn::apps::RPC_RRDB_RRDB_MULTI_REMOVE ||
diff --git a/src/server/pegasus_write_service.h b/src/server/pegasus_write_service.h
index 9fb854ffd..9e79f9122 100644
--- a/src/server/pegasus_write_service.h
+++ b/src/server/pegasus_write_service.h
@@ -25,13 +25,11 @@
#include "common//duplication_common.h"
#include "common/common.h"
-#include "perf_counter/perf_counter_wrapper.h"
#include "replica/replica_base.h"
#include "utils/errors.h"
namespace dsn {
class blob;
-class perf_counter;
namespace apps {
class check_and_mutate_request;
class check_and_mutate_response;
@@ -216,28 +214,29 @@ private:
capacity_unit_calculator *_cu_calculator;
- ::dsn::perf_counter_wrapper _pfc_put_qps;
- ::dsn::perf_counter_wrapper _pfc_multi_put_qps;
- ::dsn::perf_counter_wrapper _pfc_remove_qps;
- ::dsn::perf_counter_wrapper _pfc_multi_remove_qps;
- ::dsn::perf_counter_wrapper _pfc_incr_qps;
- ::dsn::perf_counter_wrapper _pfc_check_and_set_qps;
- ::dsn::perf_counter_wrapper _pfc_check_and_mutate_qps;
- ::dsn::perf_counter_wrapper _pfc_duplicate_qps;
- ::dsn::perf_counter_wrapper _pfc_dup_time_lag;
- ::dsn::perf_counter_wrapper _pfc_dup_lagging_writes;
-
- ::dsn::perf_counter_wrapper _pfc_put_latency;
- ::dsn::perf_counter_wrapper _pfc_multi_put_latency;
- ::dsn::perf_counter_wrapper _pfc_remove_latency;
- ::dsn::perf_counter_wrapper _pfc_multi_remove_latency;
- ::dsn::perf_counter_wrapper _pfc_incr_latency;
- ::dsn::perf_counter_wrapper _pfc_check_and_set_latency;
- ::dsn::perf_counter_wrapper _pfc_check_and_mutate_latency;
-
- // Records all requests.
- std::vector<::dsn::perf_counter *> _batch_qps_perfcounters;
- std::vector<::dsn::perf_counter *> _batch_latency_perfcounters;
+ METRIC_VAR_DECLARE_counter(put_requests);
+ METRIC_VAR_DECLARE_counter(multi_put_requests);
+ METRIC_VAR_DECLARE_counter(remove_requests);
+ METRIC_VAR_DECLARE_counter(multi_remove_requests);
+ METRIC_VAR_DECLARE_counter(incr_requests);
+ METRIC_VAR_DECLARE_counter(check_and_set_requests);
+ METRIC_VAR_DECLARE_counter(check_and_mutate_requests);
+
+ METRIC_VAR_DECLARE_percentile_int64(put_latency_ns);
+ METRIC_VAR_DECLARE_percentile_int64(multi_put_latency_ns);
+ METRIC_VAR_DECLARE_percentile_int64(remove_latency_ns);
+ METRIC_VAR_DECLARE_percentile_int64(multi_remove_latency_ns);
+ METRIC_VAR_DECLARE_percentile_int64(incr_latency_ns);
+ METRIC_VAR_DECLARE_percentile_int64(check_and_set_latency_ns);
+ METRIC_VAR_DECLARE_percentile_int64(check_and_mutate_latency_ns);
+
+ METRIC_VAR_DECLARE_counter(dup_requests);
+ METRIC_VAR_DECLARE_percentile_int64(dup_time_lag_ms);
+ METRIC_VAR_DECLARE_counter(dup_lagging_writes);
+
+ // Record batch size for put and remove requests.
+ uint32_t _put_batch_size;
+ uint32_t _remove_batch_size;
// TODO(wutao1): add perf counters for failed rpc.
};
diff --git a/src/server/test/pegasus_server_write_test.cpp b/src/server/test/pegasus_server_write_test.cpp
index 0e8c57114..b771e67bf 100644
--- a/src/server/test/pegasus_server_write_test.cpp
+++ b/src/server/test/pegasus_server_write_test.cpp
@@ -108,8 +108,8 @@ public:
// make sure everything is cleanup after batch write.
ASSERT_TRUE(_server_write->_put_rpc_batch.empty());
ASSERT_TRUE(_server_write->_remove_rpc_batch.empty());
- ASSERT_TRUE(_server_write->_write_svc->_batch_qps_perfcounters.empty());
- ASSERT_TRUE(_server_write->_write_svc->_batch_latency_perfcounters.empty());
+ ASSERT_EQ(_server_write->_write_svc->_put_batch_size, 0);
+ ASSERT_EQ(_server_write->_write_svc->_remove_batch_size, 0);
ASSERT_EQ(_server_write->_write_svc->_batch_start_time, 0);
ASSERT_EQ(_server_write->_write_svc->_impl->_rocksdb_wrapper->_write_batch->Count(),
0);
diff --git a/src/utils/metrics.h b/src/utils/metrics.h
index 27c6355f3..2d6da6c0f 100644
--- a/src/utils/metrics.h
+++ b/src/utils/metrics.h
@@ -51,6 +51,7 @@
#include "utils/singleton.h"
#include "utils/string_view.h"
#include "utils/synchronize.h"
+#include "utils/time_utils.h"
namespace boost {
namespace system {
@@ -89,7 +90,8 @@ class error_code;
// Instantiating the metric in whatever class represents it with some initial arguments, if any:
// metric_instance = METRIC_my_gauge_name.instantiate(entity_instance, ...);
-// Convenient macros are provided to define entity types and metric prototypes.
+// The following are convenient macros provided to define entity types and metric prototypes.
+
#define METRIC_DEFINE_entity(name) ::dsn::metric_entity_prototype METRIC_ENTITY_##name(#name)
#define METRIC_DEFINE_gauge_int64(entity_type, name, unit, desc, ...) \
::dsn::gauge_prototype<int64_t> METRIC_##name( \
@@ -97,6 +99,7 @@ class error_code;
#define METRIC_DEFINE_gauge_double(entity_type, name, unit, desc, ...) \
::dsn::gauge_prototype<double> METRIC_##name( \
{#entity_type, dsn::metric_type::kGauge, #name, unit, desc, ##__VA_ARGS__})
+
// There are 2 kinds of counters:
// - `counter` is the general type of counter that is implemented by striped_long_adder, which can
// achieve high performance while consuming less memory if it's not updated very frequently.
@@ -141,6 +144,42 @@ class error_code;
#define METRIC_DECLARE_percentile_double(name) \
extern dsn::floating_percentile_prototype<double> METRIC_##name
+// Following METRIC_*VAR* macros are introduced so that:
+// * only need to use prototype name to operate each metric variable;
+// * uniformly name each variable in user class;
+// * differentiate operations on metrics significantly from main logic, improving code readability.
+
+// Declare a metric variable in user class.
+//
+// Since a type tends to be a class template where there might be commas, use variadic arguments
+// instead of a single fixed argument to represent a type.
+#define METRIC_VAR_DECLARE(name, ...) __VA_ARGS__ _##name
+#define METRIC_VAR_DECLARE_gauge_int64(name) METRIC_VAR_DECLARE(name, dsn::gauge_ptr<int64_t>)
+#define METRIC_VAR_DECLARE_counter(name) \
+ METRIC_VAR_DECLARE(name, dsn::counter_ptr<dsn::striped_long_adder, false>)
+#define METRIC_VAR_DECLARE_percentile_int64(name) \
+ METRIC_VAR_DECLARE(name, dsn::percentile_ptr<int64_t>)
+
+// Initialize a metric variable in user class.
+#define METRIC_VAR_INIT(name, entity) _##name(METRIC_##name.instantiate(entity##_metric_entity()))
+#define METRIC_VAR_INIT_replica(name) METRIC_VAR_INIT(name, replica)
+
+// Perform increment-related operations on metrics including gauge and counter.
+#define METRIC_VAR_INCREMENT_BY(name, x) _##name->increment_by(x)
+#define METRIC_VAR_INCREMENT(name) _##name->increment()
+
+// Perform set() operations on metrics including gauge and percentile.
+//
+// There are 2 kinds of invocations of set() for a metric:
+// * set(val): set a single value for a metric, such as gauge, percentile;
+// * set(n, val): set multiple repeated values (the number of duplicates is n) for a metric,
+// such as percentile.
+#define METRIC_VAR_SET(name, ...) _##name->set(__VA_ARGS__)
+
+// Convenient macro that is used to compute latency automatically, which is dedicated to percentile.
+#define METRIC_VAR_AUTO_LATENCY(name, ...) \
+ dsn::auto_latency __##name##_auto_latency(_##name, ##__VA_ARGS__)
+
namespace dsn {
class metric; // IWYU pragma: keep
class metric_entity_prototype; // IWYU pragma: keep
@@ -552,7 +591,7 @@ ENUM_REG_WITH_CUSTOM_NAME(metric_type::kVolatileCounter, volatile_counter)
ENUM_REG_WITH_CUSTOM_NAME(metric_type::kPercentile, percentile)
ENUM_END(metric_type)
-enum class metric_unit
+enum class metric_unit : size_t
{
kNanoSeconds,
kMicroSeconds,
@@ -562,6 +601,31 @@ enum class metric_unit
kInvalidUnit,
};
+#define METRIC_ASSERT_UNIT_LATENCY(unit, index) \
+ static_assert(static_cast<size_t>(metric_unit::unit) == index, \
+ #unit " should be at index " #index)
+
+METRIC_ASSERT_UNIT_LATENCY(kNanoSeconds, 0);
+METRIC_ASSERT_UNIT_LATENCY(kMicroSeconds, 1);
+METRIC_ASSERT_UNIT_LATENCY(kMilliSeconds, 2);
+METRIC_ASSERT_UNIT_LATENCY(kSeconds, 3);
+
+const std::vector<uint64_t> kMetricLatencyConverterFromNS = {
+ 1, 1000, 1000 * 1000, 1000 * 1000 * 1000};
+
+inline uint64_t convert_metric_latency_from_ns(uint64_t latency_ns, metric_unit target_unit)
+{
+ if (dsn_likely(target_unit == metric_unit::kNanoSeconds)) {
+ // Since nanoseconds are used as the latency unit more frequently, eliminate unnecessary
+ // conversion by branch prediction.
+ return latency_ns;
+ }
+
+ auto index = static_cast<size_t>(target_unit);
+ CHECK_LT(index, kMetricLatencyConverterFromNS.size());
+ return latency_ns / kMetricLatencyConverterFromNS[index];
+}
+
ENUM_BEGIN(metric_unit, metric_unit::kInvalidUnit)
ENUM_REG_WITH_CUSTOM_NAME(metric_unit::kNanoSeconds, nanoseconds)
ENUM_REG_WITH_CUSTOM_NAME(metric_unit::kMicroSeconds, microseconds)
@@ -1066,6 +1130,13 @@ public:
_samples.get()[index & (_sample_size - 1)] = val;
}
+ void set(size_t n, const value_type &val)
+ {
+ for (size_t i = 0; i < n; ++i) {
+ set(val);
+ }
+ }
+
// If `type` is not configured, it will return false with zero value stored in `val`;
// otherwise, it will always return true with the value corresponding to `type`.
bool get(kth_percentile_type type, value_type &val) const
@@ -1177,6 +1248,7 @@ private:
friend class metric_entity;
friend class ref_ptr<percentile<value_type, NthElementFinder>>;
+ friend class MetricVarTest;
virtual void close() override
{
@@ -1199,6 +1271,20 @@ private:
release_ref();
}
+ std::vector<value_type> samples_for_test()
+ {
+ size_type real_sample_size = std::min(static_cast<size_type>(_tail.load()), _sample_size);
+ if (real_sample_size == 0) {
+ return std::vector<value_type>();
+ }
+
+ std::vector<value_type> real_samples(real_sample_size);
+ std::copy(_samples.get(), _samples.get() + real_sample_size, real_samples.begin());
+ return real_samples;
+ }
+
+ void reset_tail_for_test() { _tail.store(0); }
+
value_type value(size_t index) const
{
return _full_nth_elements[index].load(std::memory_order_relaxed);
@@ -1219,7 +1305,7 @@ private:
}
// Find nth elements.
- std::vector<T> array(real_sample_size);
+ std::vector<value_type> array(real_sample_size);
std::copy(_samples.get(), _samples.get() + real_sample_size, array.begin());
_nth_element_finder(array.begin(), array.begin(), array.end());
@@ -1288,4 +1374,47 @@ template <typename T,
using floating_percentile_prototype =
metric_prototype_with<floating_percentile<T, NthElementFinder>>;
+// Compute latency automatically at the end of the scope, which is set to percentile which it has
+// bound to.
+class auto_latency
+{
+public:
+ auto_latency(const percentile_ptr<int64_t> &percentile) : _percentile(percentile) {}
+
+ auto_latency(const percentile_ptr<int64_t> &percentile, std::function<void(uint64_t)> callback)
+ : _percentile(percentile), _callback(std::move(callback))
+ {
+ }
+
+ auto_latency(const percentile_ptr<int64_t> &percentile, uint64_t start_time_ns)
+ : _percentile(percentile), _chrono(start_time_ns)
+ {
+ }
+
+ auto_latency(const percentile_ptr<int64_t> &percentile,
+ uint64_t start_time_ns,
+ std::function<void(uint64_t)> callback)
+ : _percentile(percentile), _chrono(start_time_ns), _callback(std::move(callback))
+ {
+ }
+
+ ~auto_latency()
+ {
+ auto latency =
+ convert_metric_latency_from_ns(_chrono.duration_ns(), _percentile->prototype()->unit());
+ _percentile->set(static_cast<int64_t>(latency));
+
+ if (_callback) {
+ _callback(latency);
+ }
+ }
+
+private:
+ percentile_ptr<int64_t> _percentile;
+ utils::chronograph _chrono;
+ std::function<void(uint64_t)> _callback;
+
+ DISALLOW_COPY_AND_ASSIGN(auto_latency);
+};
+
} // namespace dsn
diff --git a/src/utils/test/metrics_test.cpp b/src/utils/test/metrics_test.cpp
index b81a0ac8e..670af36b7 100644
--- a/src/utils/test/metrics_test.cpp
+++ b/src/utils/test/metrics_test.cpp
@@ -83,6 +83,8 @@ METRIC_DEFINE_entity(my_server);
METRIC_DEFINE_entity(my_table);
METRIC_DEFINE_entity(my_replica);
+#define METRIC_VAR_INIT_my_replica(name) METRIC_VAR_INIT(name, my_replica)
+
// Dedicated entity for getting metrics by http service.
METRIC_DEFINE_entity(my_app);
@@ -174,6 +176,26 @@ METRIC_DEFINE_percentile_double(my_server,
dsn::metric_unit::kNanoSeconds,
"a server-level percentile of double type for test");
+METRIC_DEFINE_percentile_int64(my_replica,
+ test_replica_percentile_int64_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "a replica-level percentile of int64 type in nanoseconds for test");
+
+METRIC_DEFINE_percentile_int64(my_replica,
+ test_replica_percentile_int64_us,
+ dsn::metric_unit::kMicroSeconds,
+ "a replica-level percentile of int64 type in microseconds for test");
+
+METRIC_DEFINE_percentile_int64(my_replica,
+ test_replica_percentile_int64_ms,
+ dsn::metric_unit::kMilliSeconds,
+ "a replica-level percentile of int64 type in milliseconds for test");
+
+METRIC_DEFINE_percentile_int64(my_replica,
+ test_replica_percentile_int64_s,
+ dsn::metric_unit::kSeconds,
+ "a replica-level percentile of int64 type in seconds for test");
+
namespace dsn {
TEST(metrics_test, create_entity)
@@ -737,9 +759,7 @@ void run_percentile(const metric_entity_ptr &my_entity,
auto my_metric = prototype.instantiate(my_entity, interval_ms, kth_percentiles, sample_size);
// Preload zero in current thread.
- for (size_t i = 0; i < num_preload; ++i) {
- my_metric->set(0);
- }
+ my_metric->set(num_preload, 0);
// Load other data in each spawned thread evenly.
const size_t num_operations = data.size() / num_threads;
@@ -3056,4 +3076,125 @@ INSTANTIATE_TEST_CASE_P(MetricsTest,
MetricsRetirementTest,
testing::ValuesIn(metrics_retirement_tests));
+class MetricVarTest : public testing::Test
+{
+protected:
+ MetricVarTest();
+
+ void SetUp() override
+ {
+ _test_replica_gauge_int64->set(0);
+ _test_replica_counter->reset();
+ _test_replica_percentile_int64_ns->reset_tail_for_test();
+ _test_replica_percentile_int64_us->reset_tail_for_test();
+ _test_replica_percentile_int64_ms->reset_tail_for_test();
+ _test_replica_percentile_int64_s->reset_tail_for_test();
+ }
+
+ const metric_entity_ptr &my_replica_metric_entity() const { return _my_replica_metric_entity; }
+
+ void test_set_percentile(const std::vector<int64_t> &expected_samples);
+ void test_set_percentile(size_t n, int64_t val);
+
+ const metric_entity_ptr _my_replica_metric_entity;
+ METRIC_VAR_DECLARE_gauge_int64(test_replica_gauge_int64);
+ METRIC_VAR_DECLARE_counter(test_replica_counter);
+ METRIC_VAR_DECLARE_percentile_int64(test_replica_percentile_int64_ns);
+ METRIC_VAR_DECLARE_percentile_int64(test_replica_percentile_int64_us);
+ METRIC_VAR_DECLARE_percentile_int64(test_replica_percentile_int64_ms);
+ METRIC_VAR_DECLARE_percentile_int64(test_replica_percentile_int64_s);
+
+ DISALLOW_COPY_AND_ASSIGN(MetricVarTest);
+};
+
+MetricVarTest::MetricVarTest()
+ : _my_replica_metric_entity(METRIC_ENTITY_my_replica.instantiate("replica_var_test")),
+ METRIC_VAR_INIT_my_replica(test_replica_gauge_int64),
+ METRIC_VAR_INIT_my_replica(test_replica_counter),
+ METRIC_VAR_INIT_my_replica(test_replica_percentile_int64_ns),
+ METRIC_VAR_INIT_my_replica(test_replica_percentile_int64_us),
+ METRIC_VAR_INIT_my_replica(test_replica_percentile_int64_ms),
+ METRIC_VAR_INIT_my_replica(test_replica_percentile_int64_s)
+{
+}
+
+#define METRIC_VAR_SAMPLES(name) _##name->samples_for_test()
+
+void MetricVarTest::test_set_percentile(const std::vector<int64_t> &expected_samples)
+{
+ for (const auto &val : expected_samples) {
+ METRIC_VAR_SET(test_replica_percentile_int64_ns, val);
+ }
+ EXPECT_EQ(expected_samples, METRIC_VAR_SAMPLES(test_replica_percentile_int64_ns));
+}
+
+void MetricVarTest::test_set_percentile(size_t n, int64_t val)
+{
+ METRIC_VAR_SET(test_replica_percentile_int64_ns, n, val);
+ EXPECT_EQ(std::vector<int64_t>(n, val), METRIC_VAR_SAMPLES(test_replica_percentile_int64_ns));
+}
+
+#define METRIC_VAR_VALUE(name) _##name->value()
+
+#define TEST_METRIC_VAR_INCREMENT(name) \
+ do { \
+ ASSERT_EQ(0, METRIC_VAR_VALUE(name)); \
+ \
+ METRIC_VAR_INCREMENT(name); \
+ ASSERT_EQ(1, METRIC_VAR_VALUE(name)); \
+ \
+ METRIC_VAR_INCREMENT(name); \
+ ASSERT_EQ(2, METRIC_VAR_VALUE(name)); \
+ \
+ METRIC_VAR_INCREMENT_BY(name, 5); \
+ ASSERT_EQ(7, METRIC_VAR_VALUE(name)); \
+ \
+ METRIC_VAR_INCREMENT_BY(name, 18); \
+ ASSERT_EQ(25, METRIC_VAR_VALUE(name)); \
+ } while (0);
+
+TEST_F(MetricVarTest, IncrementGauge) { TEST_METRIC_VAR_INCREMENT(test_replica_gauge_int64); }
+
+TEST_F(MetricVarTest, IncrementCounter) { TEST_METRIC_VAR_INCREMENT(test_replica_counter); }
+
+TEST_F(MetricVarTest, SetGauge)
+{
+ ASSERT_EQ(0, METRIC_VAR_VALUE(test_replica_gauge_int64));
+
+ METRIC_VAR_SET(test_replica_gauge_int64, 5);
+ ASSERT_EQ(5, METRIC_VAR_VALUE(test_replica_gauge_int64));
+
+ METRIC_VAR_SET(test_replica_gauge_int64, 18);
+ ASSERT_EQ(18, METRIC_VAR_VALUE(test_replica_gauge_int64));
+}
+
+TEST_F(MetricVarTest, SetPercentileIndividually) { test_set_percentile({20, 50, 10, 25, 16}); }
+
+TEST_F(MetricVarTest, SetPercentileRepeatedly) { test_set_percentile(5, 100); }
+
+#define TEST_METRIC_VAR_AUTO_LATENCY(unit_abbr, factor) \
+ do { \
+ auto start_time_ns = dsn_now_ns(); \
+ uint64_t actual_latency_ns = 0; \
+ { \
+ METRIC_VAR_AUTO_LATENCY(test_replica_percentile_int64_##unit_abbr, \
+ start_time_ns, \
+ [&actual_latency_ns](uint64_t latency) mutable { \
+ actual_latency_ns = latency * factor; \
+ }); \
+ } \
+ \
+ uint64_t expected_latency_ns = dsn_now_ns() - start_time_ns; \
+ ASSERT_GE(expected_latency_ns, actual_latency_ns); \
+ EXPECT_LT(expected_latency_ns - actual_latency_ns, 1000 * 1000); \
+ } while (0)
+
+TEST_F(MetricVarTest, AutoLatencyNanoSeconds) { TEST_METRIC_VAR_AUTO_LATENCY(ns, 1); }
+
+TEST_F(MetricVarTest, AutoLatencyMicroSeconds) { TEST_METRIC_VAR_AUTO_LATENCY(us, 1000); }
+
+TEST_F(MetricVarTest, AutoLatencyMilliSeconds) { TEST_METRIC_VAR_AUTO_LATENCY(ms, 1000 * 1000); }
+
+TEST_F(MetricVarTest, AutoLatencySeconds) { TEST_METRIC_VAR_AUTO_LATENCY(s, 1000 * 1000 * 1000); }
+
} // namespace dsn
diff --git a/src/utils/time_utils.h b/src/utils/time_utils.h
index 50df0b785..1db1676b0 100644
--- a/src/utils/time_utils.h
+++ b/src/utils/time_utils.h
@@ -32,7 +32,10 @@
#include <cstdio>
#include <string>
-#include "string_view.h"
+#include "runtime/api_layer1.h"
+#include "utils/fmt_logging.h"
+#include "utils/ports.h"
+#include "utils/string_view.h"
namespace dsn {
namespace utils {
@@ -130,5 +133,28 @@ inline int64_t hh_mm_today_to_unix_sec(string_view hhmm_of_day)
return get_unix_sec_today_midnight() + sec_of_day;
}
+class chronograph
+{
+public:
+ chronograph() : chronograph(dsn_now_ns()) {}
+ chronograph(uint64_t start_time_ns) : _start_time_ns(start_time_ns) {}
+ ~chronograph() = default;
+
+ inline void reset_start_time() { _start_time_ns = dsn_now_ns(); }
+
+ inline uint64_t duration_ns()
+ {
+ auto now = dsn_now_ns();
+ CHECK_GE(now, _start_time_ns);
+
+ return now - _start_time_ns;
+ }
+
+private:
+ uint64_t _start_time_ns;
+
+ DISALLOW_COPY_AND_ASSIGN(chronograph);
+};
+
} // namespace utils
} // namespace dsn
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 11/12: feat(new_metrics): migrate server-level metrics for nfs (#1421)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit 6d17dc4f1b381aae14fbaa577c43c6c4d16dfd0b
Author: Dan Wang <wa...@apache.org>
AuthorDate: Thu Mar 30 15:14:55 2023 +0800
feat(new_metrics): migrate server-level metrics for nfs (#1421)
https://github.com/apache/incubator-pegasus/issues/1329
Migrate server-level metrics to new framework for both server/client sides
of nfs, including the data size in bytes that are requested by client or read
from local file in server, the number of nfs copy requests that fail for client
or server, the data size in bytes that are written to local file in client, the
number of failed writes to local file in client.
The old type in perf counters of all these metrics are volatile counter, while
all of them become accumulated counter for new metrics.
---
src/nfs/nfs_client_impl.cpp | 56 ++++++++++++++++++++-----------------
src/nfs/nfs_client_impl.h | 10 +++----
src/nfs/nfs_server_impl.cpp | 34 ++++++++++++----------
src/nfs/nfs_server_impl.h | 6 ++--
src/nfs/test/CMakeLists.txt | 2 +-
src/server/pegasus_server_write.cpp | 1 -
6 files changed, 60 insertions(+), 49 deletions(-)
diff --git a/src/nfs/nfs_client_impl.cpp b/src/nfs/nfs_client_impl.cpp
index 9daefecd5..e5614b343 100644
--- a/src/nfs/nfs_client_impl.cpp
+++ b/src/nfs/nfs_client_impl.cpp
@@ -32,7 +32,6 @@
#include "nfs/nfs_code_definition.h"
#include "nfs/nfs_node.h"
-#include "perf_counter/perf_counter.h"
#include "utils/blob.h"
#include "utils/command_manager.h"
#include "utils/filesystem.h"
@@ -40,8 +39,30 @@
#include "utils/fmt_logging.h"
#include "utils/ports.h"
#include "utils/string_conv.h"
+#include "utils/string_view.h"
#include "utils/token_buckets.h"
+METRIC_DEFINE_counter(server,
+ nfs_client_copy_bytes,
+ dsn::metric_unit::kBytes,
+ "The accumulated data size in bytes requested by client during nfs copy");
+
+METRIC_DEFINE_counter(server,
+ nfs_client_failed_copy_requests,
+ dsn::metric_unit::kRequests,
+ "The number of failed nfs copy requests (requested by client)");
+
+METRIC_DEFINE_counter(
+ server,
+ nfs_client_write_bytes,
+ dsn::metric_unit::kBytes,
+ "The accumulated data size in bytes that are written to local file in client");
+
+METRIC_DEFINE_counter(server,
+ nfs_client_failed_writes,
+ dsn::metric_unit::kWrites,
+ "The number of failed writes to local file in client");
+
namespace dsn {
namespace service {
static uint32_t current_max_copy_rate_megabytes = 0;
@@ -98,27 +119,12 @@ nfs_client_impl::nfs_client_impl()
_concurrent_local_write_count(0),
_buffered_local_write_count(0),
_copy_requests_low(FLAGS_max_file_copy_request_count_per_file),
- _high_priority_remaining_time(FLAGS_high_priority_speed_rate)
+ _high_priority_remaining_time(FLAGS_high_priority_speed_rate),
+ METRIC_VAR_INIT_server(nfs_client_copy_bytes),
+ METRIC_VAR_INIT_server(nfs_client_failed_copy_requests),
+ METRIC_VAR_INIT_server(nfs_client_write_bytes),
+ METRIC_VAR_INIT_server(nfs_client_failed_writes)
{
- _recent_copy_data_size.init_app_counter("eon.nfs_client",
- "recent_copy_data_size",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "nfs client copy data size in the recent period");
- _recent_copy_fail_count.init_app_counter(
- "eon.nfs_client",
- "recent_copy_fail_count",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "nfs client copy fail count count in the recent period");
- _recent_write_data_size.init_app_counter("eon.nfs_client",
- "recent_write_data_size",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "nfs client write data size in the recent period");
- _recent_write_fail_count.init_app_counter(
- "eon.nfs_client",
- "recent_write_fail_count",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "nfs client write fail count count in the recent period");
-
_copy_token_buckets = std::make_unique<utils::token_buckets>();
register_cli_commands();
@@ -337,7 +343,7 @@ void nfs_client_impl::end_copy(::dsn::error_code err,
}
if (err != ::dsn::ERR_OK) {
- _recent_copy_fail_count->increment();
+ METRIC_VAR_INCREMENT(nfs_client_failed_copy_requests);
if (!fc->user_req->is_finished) {
if (reqc->retry_count > 0) {
@@ -373,7 +379,7 @@ void nfs_client_impl::end_copy(::dsn::error_code err,
}
else {
- _recent_copy_data_size->add(resp.size);
+ METRIC_VAR_INCREMENT_BY(nfs_client_copy_bytes, resp.size);
reqc->response = resp;
reqc->is_ready_for_write = true;
@@ -504,7 +510,7 @@ void nfs_client_impl::end_write(error_code err, size_t sz, const copy_request_ex
bool completed = false;
if (err != ERR_OK) {
- _recent_write_fail_count->increment();
+ METRIC_VAR_INCREMENT(nfs_client_failed_writes);
LOG_ERROR("[nfs_service] local write failed, dir = {}, file = {}, err = {}",
fc->user_req->file_size_req.dst_dir,
@@ -512,7 +518,7 @@ void nfs_client_impl::end_write(error_code err, size_t sz, const copy_request_ex
err);
completed = true;
} else {
- _recent_write_data_size->add(sz);
+ METRIC_VAR_INCREMENT_BY(nfs_client_write_bytes, sz);
file_wrapper_ptr temp_holder;
zauto_lock l(fc->user_req->user_req_lock);
diff --git a/src/nfs/nfs_client_impl.h b/src/nfs/nfs_client_impl.h
index 88f70aad4..0c15fc8b3 100644
--- a/src/nfs/nfs_client_impl.h
+++ b/src/nfs/nfs_client_impl.h
@@ -41,7 +41,6 @@
#include "aio/file_io.h"
#include "nfs_code_definition.h"
#include "nfs_types.h"
-#include "perf_counter/perf_counter_wrapper.h"
#include "runtime/rpc/rpc_address.h"
#include "runtime/task/async_calls.h"
#include "runtime/task/task.h"
@@ -50,6 +49,7 @@
#include "utils/autoref_ptr.h"
#include "utils/error_code.h"
#include "utils/fmt_logging.h"
+#include "utils/metrics.h"
#include "utils/zlocks.h"
namespace dsn {
@@ -311,10 +311,10 @@ private:
zlock _local_writes_lock;
std::deque<copy_request_ex_ptr> _local_writes;
- perf_counter_wrapper _recent_copy_data_size;
- perf_counter_wrapper _recent_copy_fail_count;
- perf_counter_wrapper _recent_write_data_size;
- perf_counter_wrapper _recent_write_fail_count;
+ METRIC_VAR_DECLARE_counter(nfs_client_copy_bytes);
+ METRIC_VAR_DECLARE_counter(nfs_client_failed_copy_requests);
+ METRIC_VAR_DECLARE_counter(nfs_client_write_bytes);
+ METRIC_VAR_DECLARE_counter(nfs_client_failed_writes);
std::unique_ptr<command_deregister> _nfs_max_copy_rate_megabytes_cmd;
diff --git a/src/nfs/nfs_server_impl.cpp b/src/nfs/nfs_server_impl.cpp
index cadacba1a..25632d4f9 100644
--- a/src/nfs/nfs_server_impl.cpp
+++ b/src/nfs/nfs_server_impl.cpp
@@ -35,17 +35,30 @@
#include <vector>
#include "nfs/nfs_code_definition.h"
-#include "perf_counter/perf_counter.h"
#include "runtime/api_layer1.h"
#include "runtime/task/async_calls.h"
#include "utils/TokenBucket.h"
+#include "utils/autoref_ptr.h"
#include "utils/filesystem.h"
#include "utils/flags.h"
#include "utils/ports.h"
#include "utils/safe_strerror_posix.h"
#include "utils/string_conv.h"
+#include "utils/string_view.h"
#include "utils/utils.h"
+METRIC_DEFINE_counter(
+ server,
+ nfs_server_copy_bytes,
+ dsn::metric_unit::kBytes,
+ "The accumulated data size in bytes that are read from local file in server during nfs copy");
+
+METRIC_DEFINE_counter(
+ server,
+ nfs_server_failed_copy_requests,
+ dsn::metric_unit::kRequests,
+ "The number of nfs copy requests (received by server) that fail to read local file in server");
+
namespace dsn {
class disk_file;
@@ -61,7 +74,10 @@ DSN_TAG_VARIABLE(max_send_rate_megabytes_per_disk, FT_MUTABLE);
DSN_DECLARE_int32(file_close_timer_interval_ms_on_server);
DSN_DECLARE_int32(file_close_expire_time_ms);
-nfs_service_impl::nfs_service_impl() : ::dsn::serverlet<nfs_service_impl>("nfs")
+nfs_service_impl::nfs_service_impl()
+ : ::dsn::serverlet<nfs_service_impl>("nfs"),
+ METRIC_VAR_INIT_server(nfs_server_copy_bytes),
+ METRIC_VAR_INIT_server(nfs_server_failed_copy_requests)
{
_file_close_timer = ::dsn::tasking::enqueue_timer(
LPC_NFS_FILE_CLOSE_TIMER,
@@ -69,16 +85,6 @@ nfs_service_impl::nfs_service_impl() : ::dsn::serverlet<nfs_service_impl>("nfs")
[this] { close_file(); },
std::chrono::milliseconds(FLAGS_file_close_timer_interval_ms_on_server));
- _recent_copy_data_size.init_app_counter("eon.nfs_server",
- "recent_copy_data_size",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "nfs server copy data size in the recent period");
- _recent_copy_fail_count.init_app_counter(
- "eon.nfs_server",
- "recent_copy_fail_count",
- COUNTER_TYPE_VOLATILE_NUMBER,
- "nfs server copy fail count count in the recent period");
-
_send_token_buckets = std::make_unique<dsn::utils::token_buckets>();
register_cli_commands();
}
@@ -161,9 +167,9 @@ void nfs_service_impl::internal_read_callback(error_code err, size_t sz, callbac
if (err != ERR_OK) {
LOG_ERROR("[nfs_service] read file {} failed, err = {}", cp.file_path, err);
- _recent_copy_fail_count->increment();
+ METRIC_VAR_INCREMENT(nfs_server_failed_copy_requests);
} else {
- _recent_copy_data_size->add(sz);
+ METRIC_VAR_INCREMENT_BY(nfs_server_copy_bytes, sz);
}
::dsn::service::copy_response resp;
diff --git a/src/nfs/nfs_server_impl.h b/src/nfs/nfs_server_impl.h
index bdc83bdc9..7a24d8d14 100644
--- a/src/nfs/nfs_server_impl.h
+++ b/src/nfs/nfs_server_impl.h
@@ -35,7 +35,6 @@
#include "aio/file_io.h"
#include "nfs_code_definition.h"
#include "nfs_types.h"
-#include "perf_counter/perf_counter_wrapper.h"
#include "runtime/serverlet.h"
#include "runtime/task/task.h"
#include "runtime/task/task_tracker.h"
@@ -43,6 +42,7 @@
#include "utils/command_manager.h"
#include "utils/error_code.h"
#include "utils/fmt_logging.h"
+#include "utils/metrics.h"
#include "utils/token_buckets.h"
#include "utils/zlocks.h"
@@ -137,8 +137,8 @@ private:
std::unique_ptr<dsn::utils::token_buckets>
_send_token_buckets; // rate limiter of send to remote
- perf_counter_wrapper _recent_copy_data_size;
- perf_counter_wrapper _recent_copy_fail_count;
+ METRIC_VAR_DECLARE_counter(nfs_server_copy_bytes);
+ METRIC_VAR_DECLARE_counter(nfs_server_failed_copy_requests);
std::unique_ptr<command_deregister> _nfs_max_send_rate_megabytes_cmd;
diff --git a/src/nfs/test/CMakeLists.txt b/src/nfs/test/CMakeLists.txt
index 735bb29bd..8dcf2cef7 100644
--- a/src/nfs/test/CMakeLists.txt
+++ b/src/nfs/test/CMakeLists.txt
@@ -33,7 +33,7 @@ set(MY_PROJ_SRC "")
# "GLOB" for non-recursive search
set(MY_SRC_SEARCH_MODE "GLOB")
-set(MY_PROJ_LIBS dsn_nfs dsn_runtime gtest dsn_aio)
+set(MY_PROJ_LIBS dsn_nfs dsn_runtime gtest dsn_aio dsn_http)
set(MY_BOOST_LIBS Boost::system Boost::filesystem Boost::regex)
diff --git a/src/server/pegasus_server_write.cpp b/src/server/pegasus_server_write.cpp
index 9f8db6ca0..e0669fc99 100644
--- a/src/server/pegasus_server_write.cpp
+++ b/src/server/pegasus_server_write.cpp
@@ -18,7 +18,6 @@
*/
#include <rocksdb/status.h>
-#include <stdio.h>
#include <thrift/transport/TTransportException.h>
#include <algorithm>
#include <utility>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 06/12: feat(new_metrics): migrate replica-level metrics for replica class (#1397)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit 0848525a6394771455bcd8031092b12977009d0d
Author: Dan Wang <wa...@apache.org>
AuthorDate: Wed Mar 22 12:32:24 2023 +0800
feat(new_metrics): migrate replica-level metrics for replica class (#1397)
https://github.com/apache/incubator-pegasus/issues/1342
Migrate replica-level metrics in `replica` class to new framework, including the size of
private log, the number of backup requests, the number of read/write/backup requests
that are delayed/rejected due to throttling, the number of read/write requests that are
rejected due to splitting, the number of write requests that are rejected due to bulk load
ingestion, the number of non-idempotent write requests that are rejected due to duplication.
Note that the old metrics for the number of backup requests, the number of delayed/rejected
backup requests due to throttling and the number of rejected non-idempotent write requests
are table-level, which have been changed to replica-level in new metrics, since they could
be aggregated to table-level if necessary.
---
src/perf_counter/test/perf_counter_test.cpp | 1 +
src/replica/replica.cpp | 133 ++++++++++++++++-----------
src/replica/replica.h | 39 ++++----
src/replica/replica_2pc.cpp | 7 +-
src/replica/replica_base.cpp | 2 +
src/replica/replica_base.h | 3 +
src/replica/replica_chkpt.cpp | 8 +-
src/replica/replica_throttle.cpp | 11 +--
src/replica/test/mutation_log_test.cpp | 1 +
src/replica/test/replica_test.cpp | 14 +--
src/server/capacity_unit_calculator.cpp | 7 +-
src/server/capacity_unit_calculator.h | 3 +-
src/server/pegasus_server_impl.cpp | 1 +
src/server/pegasus_server_impl.h | 5 +-
src/server/pegasus_server_impl_init.cpp | 9 +-
src/server/pegasus_server_write.cpp | 2 +
src/server/pegasus_server_write.h | 1 +
src/server/pegasus_write_service.cpp | 11 ++-
src/server/pegasus_write_service.h | 3 +-
src/server/rocksdb_wrapper.cpp | 3 +-
src/server/rocksdb_wrapper.h | 5 +-
src/server/test/pegasus_server_impl_test.cpp | 4 +-
src/utils/clock.cpp | 1 +
23 files changed, 155 insertions(+), 119 deletions(-)
diff --git a/src/perf_counter/test/perf_counter_test.cpp b/src/perf_counter/test/perf_counter_test.cpp
index f83679bd4..0758ef807 100644
--- a/src/perf_counter/test/perf_counter_test.cpp
+++ b/src/perf_counter/test/perf_counter_test.cpp
@@ -38,6 +38,7 @@
// IWYU pragma: no_include <gtest/gtest-test-part.h>
#include <gtest/gtest.h>
#include <stdlib.h>
+#include <algorithm>
#include <chrono>
#include <functional>
#include <memory>
diff --git a/src/replica/replica.cpp b/src/replica/replica.cpp
index be2cbd309..23d4b5ade 100644
--- a/src/replica/replica.cpp
+++ b/src/replica/replica.cpp
@@ -47,6 +47,7 @@
#include "mutation.h"
#include "mutation_log.h"
#include "perf_counter/perf_counter.h"
+#include "perf_counter/perf_counter_wrapper.h"
#include "perf_counter/perf_counters.h"
#include "replica/prepare_list.h"
#include "replica/replica_context.h"
@@ -66,6 +67,66 @@
#include "utils/rand.h"
#include "utils/string_view.h"
+METRIC_DEFINE_gauge_int64(replica,
+ private_log_size_mb,
+ dsn::metric_unit::kMegaBytes,
+ "The size of private log in MB");
+
+METRIC_DEFINE_counter(replica,
+ throttling_delayed_write_requests,
+ dsn::metric_unit::kRequests,
+ "The number of delayed write requests by throttling");
+
+METRIC_DEFINE_counter(replica,
+ throttling_rejected_write_requests,
+ dsn::metric_unit::kRequests,
+ "The number of rejected write requests by throttling");
+
+METRIC_DEFINE_counter(replica,
+ throttling_delayed_read_requests,
+ dsn::metric_unit::kRequests,
+ "The number of delayed read requests by throttling");
+
+METRIC_DEFINE_counter(replica,
+ throttling_rejected_read_requests,
+ dsn::metric_unit::kRequests,
+ "The number of rejected read requests by throttling");
+
+METRIC_DEFINE_counter(replica,
+ backup_requests,
+ dsn::metric_unit::kRequests,
+ "The number of backup requests");
+
+METRIC_DEFINE_counter(replica,
+ throttling_delayed_backup_requests,
+ dsn::metric_unit::kRequests,
+ "The number of delayed backup requests by throttling");
+
+METRIC_DEFINE_counter(replica,
+ throttling_rejected_backup_requests,
+ dsn::metric_unit::kRequests,
+ "The number of rejected backup requests by throttling");
+
+METRIC_DEFINE_counter(replica,
+ splitting_rejected_write_requests,
+ dsn::metric_unit::kRequests,
+ "The number of rejected write requests by splitting");
+
+METRIC_DEFINE_counter(replica,
+ splitting_rejected_read_requests,
+ dsn::metric_unit::kRequests,
+ "The number of rejected read requests by splitting");
+
+METRIC_DEFINE_counter(replica,
+ bulk_load_ingestion_rejected_write_requests,
+ dsn::metric_unit::kRequests,
+ "The number of rejected write requests by bulk load ingestion");
+
+METRIC_DEFINE_counter(replica,
+ dup_rejected_non_idempotent_write_requests,
+ dsn::metric_unit::kRequests,
+ "The number of rejected non-idempotent write requests by duplication");
+
namespace dsn {
namespace replication {
@@ -118,7 +179,19 @@ replica::replica(replica_stub *stub,
// todo(jiashuo1): app.duplicating need rename
_is_duplication_master(app.duplicating),
_is_duplication_follower(is_duplication_follower),
- _backup_mgr(new replica_backup_manager(this))
+ _backup_mgr(new replica_backup_manager(this)),
+ METRIC_VAR_INIT_replica(private_log_size_mb),
+ METRIC_VAR_INIT_replica(throttling_delayed_write_requests),
+ METRIC_VAR_INIT_replica(throttling_rejected_write_requests),
+ METRIC_VAR_INIT_replica(throttling_delayed_read_requests),
+ METRIC_VAR_INIT_replica(throttling_rejected_read_requests),
+ METRIC_VAR_INIT_replica(backup_requests),
+ METRIC_VAR_INIT_replica(throttling_delayed_backup_requests),
+ METRIC_VAR_INIT_replica(throttling_rejected_backup_requests),
+ METRIC_VAR_INIT_replica(splitting_rejected_write_requests),
+ METRIC_VAR_INIT_replica(splitting_rejected_read_requests),
+ METRIC_VAR_INIT_replica(bulk_load_ingestion_rejected_write_requests),
+ METRIC_VAR_INIT_replica(dup_rejected_non_idempotent_write_requests)
{
CHECK(!_app_info.app_type.empty(), "");
CHECK_NOTNULL(stub, "");
@@ -132,59 +205,9 @@ replica::replica(replica_stub *stub,
_disk_migrator = std::make_unique<replica_disk_migrator>(this);
_replica_follower = std::make_unique<replica_follower>(this);
- std::string counter_str = fmt::format("private.log.size(MB)@{}", gpid);
- _counter_private_log_size.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_NUMBER, counter_str.c_str());
-
- counter_str = fmt::format("recent.write.throttling.delay.count@{}", gpid);
- _counter_recent_write_throttling_delay_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
- counter_str = fmt::format("recent.write.throttling.reject.count@{}", gpid);
- _counter_recent_write_throttling_reject_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
- counter_str = fmt::format("recent.read.throttling.delay.count@{}", gpid);
- _counter_recent_read_throttling_delay_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
- counter_str = fmt::format("recent.read.throttling.reject.count@{}", gpid);
- _counter_recent_read_throttling_reject_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
- counter_str =
- fmt::format("recent.backup.request.throttling.delay.count@{}", _app_info.app_name);
- _counter_recent_backup_request_throttling_delay_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
- counter_str =
- fmt::format("recent.backup.request.throttling.reject.count@{}", _app_info.app_name);
- _counter_recent_backup_request_throttling_reject_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
- counter_str = fmt::format("dup.disabled_non_idempotent_write_count@{}", _app_info.app_name);
- _counter_dup_disabled_non_idempotent_write_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
- counter_str = fmt::format("recent.read.splitting.reject.count@{}", gpid);
- _counter_recent_read_splitting_reject_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
- counter_str = fmt::format("recent.write.splitting.reject.count@{}", gpid);
- _counter_recent_write_splitting_reject_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
- counter_str = fmt::format("recent.write.bulk.load.ingestion.reject.count@{}", gpid);
- _counter_recent_write_bulk_load_ingestion_reject_count.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_VOLATILE_NUMBER, counter_str.c_str());
-
// init table level latency perf counters
init_table_level_latency_counters();
- counter_str = fmt::format("backup_request_qps@{}", _app_info.app_name);
- _counter_backup_request_qps.init_app_counter(
- "eon.replica", counter_str.c_str(), COUNTER_TYPE_RATE, counter_str.c_str());
-
if (need_restore) {
// add an extra env for restore
_extra_envs.insert(
@@ -261,7 +284,7 @@ void replica::on_client_read(dsn::message_ex *request, bool ignore_throttling)
return;
}
- CHECK_REQUEST_IF_SPLITTING(read)
+ CHECK_REQUEST_IF_SPLITTING(read);
if (status() == partition_status::PS_INACTIVE ||
status() == partition_status::PS_POTENTIAL_SECONDARY) {
@@ -294,7 +317,7 @@ void replica::on_client_read(dsn::message_ex *request, bool ignore_throttling)
if (!ignore_throttling && throttle_backup_request(request)) {
return;
}
- _counter_backup_request_qps->increment();
+ METRIC_VAR_INCREMENT(backup_requests);
}
uint64_t start_time_ns = dsn_now_ns();
@@ -516,8 +539,6 @@ void replica::close()
_disk_migrator.reset();
}
- _counter_private_log_size.clear();
-
// duplication_impl may have ongoing tasks.
// release it before release replica.
_duplication_mgr.reset();
@@ -599,5 +620,7 @@ error_code replica::store_app_info(app_info &info, const std::string &path)
return err;
}
+int64_t replica::get_backup_request_count() const { return METRIC_VAR_VALUE(backup_requests); }
+
} // namespace replication
} // namespace dsn
diff --git a/src/replica/replica.h b/src/replica/replica.h
index 59218d17a..e645cffc7 100644
--- a/src/replica/replica.h
+++ b/src/replica/replica.h
@@ -57,7 +57,6 @@
#include "metadata_types.h"
#include "mutation.h"
#include "mutation_log.h"
-#include "perf_counter/perf_counter_wrapper.h"
#include "prepare_list.h"
#include "replica/backup/cold_backup_context.h"
#include "replica/replica_base.h"
@@ -70,6 +69,7 @@
#include "utils/autoref_ptr.h"
#include "utils/error_code.h"
#include "utils/flags.h"
+#include "utils/metrics.h"
#include "utils/thread_access_checker.h"
#include "utils/throttling_controller.h"
#include "utils/uniq_timestamp_us.h"
@@ -78,6 +78,7 @@ namespace dsn {
class gpid;
class perf_counter;
class rpc_address;
+
namespace dist {
namespace block_service {
class block_filesystem;
@@ -118,10 +119,13 @@ class test_checker;
}
#define CHECK_REQUEST_IF_SPLITTING(op_type) \
- if (_validate_partition_hash) { \
+ do { \
+ if (!_validate_partition_hash) { \
+ break; \
+ } \
if (_split_mgr->should_reject_request()) { \
response_client_##op_type(request, ERR_SPLITTING); \
- _counter_recent_##op_type##_splitting_reject_count->increment(); \
+ METRIC_VAR_INCREMENT(splitting_rejected_##op_type##_requests); \
return; \
} \
if (!_split_mgr->check_partition_hash( \
@@ -129,7 +133,7 @@ class test_checker;
response_client_##op_type(request, ERR_PARENT_PARTITION_MISUSED); \
return; \
} \
- }
+ } while (0)
DSN_DECLARE_bool(reject_write_when_disk_insufficient);
@@ -520,6 +524,9 @@ private:
void update_app_max_replica_count(int32_t max_replica_count);
void update_app_name(const std::string &app_name);
+ // Currently only used for unit test to get the count of backup requests.
+ int64_t get_backup_request_count() const;
+
private:
friend class ::dsn::replication::test::test_checker;
friend class ::dsn::replication::mutation_queue;
@@ -638,19 +645,19 @@ private:
std::unique_ptr<replica_follower> _replica_follower;
// perf counters
- perf_counter_wrapper _counter_private_log_size;
- perf_counter_wrapper _counter_recent_write_throttling_delay_count;
- perf_counter_wrapper _counter_recent_write_throttling_reject_count;
- perf_counter_wrapper _counter_recent_read_throttling_delay_count;
- perf_counter_wrapper _counter_recent_read_throttling_reject_count;
- perf_counter_wrapper _counter_recent_backup_request_throttling_delay_count;
- perf_counter_wrapper _counter_recent_backup_request_throttling_reject_count;
- perf_counter_wrapper _counter_recent_write_splitting_reject_count;
- perf_counter_wrapper _counter_recent_read_splitting_reject_count;
- perf_counter_wrapper _counter_recent_write_bulk_load_ingestion_reject_count;
+ METRIC_VAR_DECLARE_gauge_int64(private_log_size_mb);
+ METRIC_VAR_DECLARE_counter(throttling_delayed_write_requests);
+ METRIC_VAR_DECLARE_counter(throttling_rejected_write_requests);
+ METRIC_VAR_DECLARE_counter(throttling_delayed_read_requests);
+ METRIC_VAR_DECLARE_counter(throttling_rejected_read_requests);
+ METRIC_VAR_DECLARE_counter(backup_requests);
+ METRIC_VAR_DECLARE_counter(throttling_delayed_backup_requests);
+ METRIC_VAR_DECLARE_counter(throttling_rejected_backup_requests);
+ METRIC_VAR_DECLARE_counter(splitting_rejected_write_requests);
+ METRIC_VAR_DECLARE_counter(splitting_rejected_read_requests);
+ METRIC_VAR_DECLARE_counter(bulk_load_ingestion_rejected_write_requests);
+ METRIC_VAR_DECLARE_counter(dup_rejected_non_idempotent_write_requests);
std::vector<perf_counter *> _counters_table_level_latency;
- perf_counter_wrapper _counter_dup_disabled_non_idempotent_write_count;
- perf_counter_wrapper _counter_backup_request_qps;
dsn::task_tracker _tracker;
// the thread access checker
diff --git a/src/replica/replica_2pc.cpp b/src/replica/replica_2pc.cpp
index 2e85c4bc4..2fa006dd8 100644
--- a/src/replica/replica_2pc.cpp
+++ b/src/replica/replica_2pc.cpp
@@ -72,6 +72,7 @@
#include "utils/flags.h"
#include "utils/fmt_logging.h"
#include "utils/latency_tracer.h"
+#include "utils/metrics.h"
#include "utils/ports.h"
#include "utils/thread_access_checker.h"
#include "utils/uniq_timestamp_us.h"
@@ -164,12 +165,12 @@ void replica::on_client_write(dsn::message_ex *request, bool ignore_throttling)
if (is_duplication_master() && !spec->rpc_request_is_write_idempotent) {
// Ignore non-idempotent write, because duplication provides no guarantee of atomicity to
// make this write produce the same result on multiple clusters.
- _counter_dup_disabled_non_idempotent_write_count->increment();
+ METRIC_VAR_INCREMENT(dup_rejected_non_idempotent_write_requests);
response_client_write(request, ERR_OPERATION_DISABLED);
return;
}
- CHECK_REQUEST_IF_SPLITTING(write)
+ CHECK_REQUEST_IF_SPLITTING(write);
if (partition_status::PS_PRIMARY != status()) {
response_client_write(request, ERR_INVALID_STATE);
@@ -185,7 +186,7 @@ void replica::on_client_write(dsn::message_ex *request, bool ignore_throttling)
if (_is_bulk_load_ingestion) {
if (request->rpc_code() != dsn::apps::RPC_RRDB_RRDB_BULK_LOAD) {
// reject write requests during ingestion
- _counter_recent_write_bulk_load_ingestion_reject_count->increment();
+ METRIC_VAR_INCREMENT(bulk_load_ingestion_rejected_write_requests);
response_client_write(request, ERR_OPERATION_DISABLED);
} else {
response_client_write(request, ERR_NO_NEED_OPERATE);
diff --git a/src/replica/replica_base.cpp b/src/replica/replica_base.cpp
index 11e08ae05..3168ad651 100644
--- a/src/replica/replica_base.cpp
+++ b/src/replica/replica_base.cpp
@@ -18,6 +18,8 @@
#include "replica_base.h"
#include <fmt/core.h>
+#include <fmt/ostream.h>
+#include <iosfwd>
METRIC_DEFINE_entity(replica);
diff --git a/src/replica/replica_base.h b/src/replica/replica_base.h
index 7c5b7747e..ccb39b914 100644
--- a/src/replica/replica_base.h
+++ b/src/replica/replica_base.h
@@ -26,7 +26,10 @@
#pragma once
+#include <string>
+
#include "common/gpid.h"
+#include "utils/fmt_logging.h"
#include "utils/metrics.h"
#include "utils/string_view.h"
diff --git a/src/replica/replica_chkpt.cpp b/src/replica/replica_chkpt.cpp
index b24f11157..5985e5d8d 100644
--- a/src/replica/replica_chkpt.cpp
+++ b/src/replica/replica_chkpt.cpp
@@ -70,6 +70,7 @@
#include "utils/filesystem.h"
#include "utils/flags.h"
#include "utils/fmt_logging.h"
+#include "utils/metrics.h"
#include "utils/thread_access_checker.h"
namespace dsn {
@@ -168,9 +169,10 @@ void replica::on_checkpoint_timer()
valid_start_offset,
(int64_t)FLAGS_log_private_reserve_max_size_mb * 1024 * 1024,
(int64_t)FLAGS_log_private_reserve_max_time_seconds);
- if (status() == partition_status::PS_PRIMARY)
- _counter_private_log_size->set(_private_log->total_size() /
- 1000000);
+ if (status() == partition_status::PS_PRIMARY) {
+ METRIC_VAR_SET(private_log_size_mb,
+ _private_log->total_size() >> 20);
+ }
});
}
}
diff --git a/src/replica/replica_throttle.cpp b/src/replica/replica_throttle.cpp
index 6a1a294b9..ed6bd1cce 100644
--- a/src/replica/replica_throttle.cpp
+++ b/src/replica/replica_throttle.cpp
@@ -25,14 +25,13 @@
#include "common/replica_envs.h"
#include "common/replication.codes.h"
#include "dsn.layer2_types.h"
-#include "perf_counter/perf_counter.h"
-#include "perf_counter/perf_counter_wrapper.h"
#include "replica.h"
#include "runtime/rpc/rpc_message.h"
#include "runtime/task/async_calls.h"
#include "utils/autoref_ptr.h"
#include "utils/error_code.h"
#include "utils/fmt_logging.h"
+#include "utils/metrics.h"
#include "utils/throttling_controller.h"
namespace dsn {
@@ -51,7 +50,7 @@ namespace replication {
[ this, req = message_ptr(request) ]() { on_client_##op_type(req, true); }, \
get_gpid().thread_hash(), \
std::chrono::milliseconds(delay_ms)); \
- _counter_recent_##op_type##_throttling_delay_count->increment(); \
+ METRIC_VAR_INCREMENT(throttling_delayed_##op_type##_requests); \
} else { /** type == throttling_controller::REJECT **/ \
if (delay_ms > 0) { \
tasking::enqueue(LPC_##op_type##_THROTTLING_DELAY, \
@@ -64,7 +63,7 @@ namespace replication {
} else { \
response_client_##op_type(request, ERR_BUSY); \
} \
- _counter_recent_##op_type##_throttling_reject_count->increment(); \
+ METRIC_VAR_INCREMENT(throttling_rejected_##op_type##_requests); \
} \
return true; \
} \
@@ -95,9 +94,9 @@ bool replica::throttle_backup_request(message_ex *request)
[ this, req = message_ptr(request) ]() { on_client_read(req, true); },
get_gpid().thread_hash(),
std::chrono::milliseconds(delay_ms));
- _counter_recent_backup_request_throttling_delay_count->increment();
+ METRIC_VAR_INCREMENT(throttling_delayed_backup_requests);
} else { /** type == throttling_controller::REJECT **/
- _counter_recent_backup_request_throttling_reject_count->increment();
+ METRIC_VAR_INCREMENT(throttling_rejected_backup_requests);
}
return true;
}
diff --git a/src/replica/test/mutation_log_test.cpp b/src/replica/test/mutation_log_test.cpp
index 2b7ce0bab..44e8f7ac7 100644
--- a/src/replica/test/mutation_log_test.cpp
+++ b/src/replica/test/mutation_log_test.cpp
@@ -26,6 +26,7 @@
#include "replica/mutation_log.h"
+#include <fcntl.h>
// IWYU pragma: no_include <gtest/gtest-message.h>
// IWYU pragma: no_include <gtest/gtest-test-part.h>
#include <gtest/gtest.h>
diff --git a/src/replica/test/replica_test.cpp b/src/replica/test/replica_test.cpp
index 42be2a2ae..562fe8a50 100644
--- a/src/replica/test/replica_test.cpp
+++ b/src/replica/test/replica_test.cpp
@@ -19,7 +19,6 @@
// IWYU pragma: no_include <gtest/gtest-test-part.h>
#include <gtest/gtest.h>
#include <stdint.h>
-#include <unistd.h>
#include <iostream>
#include <map>
#include <memory>
@@ -94,10 +93,7 @@ public:
return stub->_counter_recent_write_size_exceed_threshold_count->get_value();
}
- int get_table_level_backup_request_qps()
- {
- return _mock_replica->_counter_backup_request_qps->get_integer_value();
- }
+ int64_t get_backup_request_count() const { return _mock_replica->get_backup_request_count(); }
bool get_validate_partition_hash() const { return _mock_replica->_validate_partition_hash; }
@@ -278,7 +274,7 @@ TEST_F(replica_test, write_size_limited)
ASSERT_EQ(get_write_size_exceed_threshold_count(), count);
}
-TEST_F(replica_test, backup_request_qps)
+TEST_F(replica_test, backup_request_count)
{
// create backup request
struct dsn::message_header header;
@@ -290,11 +286,7 @@ TEST_F(replica_test, backup_request_qps)
backup_request->io_session = sim_net->create_client_session(rpc_address());
_mock_replica->on_client_read(backup_request);
-
- // We have to sleep >= 0.1s, or the value this perf-counter will be 0, according to the
- // implementation of perf-counter which type is COUNTER_TYPE_RATE.
- usleep(1e5);
- ASSERT_GT(get_table_level_backup_request_qps(), 0);
+ ASSERT_EQ(get_backup_request_count(), 1);
}
TEST_F(replica_test, query_data_version_test)
diff --git a/src/server/capacity_unit_calculator.cpp b/src/server/capacity_unit_calculator.cpp
index 7a62b3d56..55b9d104c 100644
--- a/src/server/capacity_unit_calculator.cpp
+++ b/src/server/capacity_unit_calculator.cpp
@@ -20,19 +20,18 @@
#include "capacity_unit_calculator.h"
#include <rocksdb/status.h>
-#include <stdio.h>
#include <sys/param.h>
#include <cmath>
-#include <string>
+#include <cstdint>
-#include "common/gpid.h"
#include "hotkey_collector.h"
-#include "perf_counter/perf_counter.h"
#include "rrdb/rrdb_types.h"
#include "runtime/rpc/rpc_message.h"
+#include "utils/autoref_ptr.h"
#include "utils/blob.h"
#include "utils/flags.h"
#include "utils/fmt_logging.h"
+#include "utils/string_view.h"
#include "utils/token_bucket_throttling_controller.h"
METRIC_DEFINE_counter(replica,
diff --git a/src/server/capacity_unit_calculator.h b/src/server/capacity_unit_calculator.h
index 6d30a07ef..d69e3394b 100644
--- a/src/server/capacity_unit_calculator.h
+++ b/src/server/capacity_unit_calculator.h
@@ -23,12 +23,13 @@
#include <memory>
#include <vector>
-#include "perf_counter/perf_counter_wrapper.h"
#include "replica/replica_base.h"
+#include "utils/metrics.h"
namespace dsn {
class blob;
class message_ex;
+
namespace apps {
class full_data;
class key_value;
diff --git a/src/server/pegasus_server_impl.cpp b/src/server/pegasus_server_impl.cpp
index 38642c5f9..711cda2d3 100644
--- a/src/server/pegasus_server_impl.cpp
+++ b/src/server/pegasus_server_impl.cpp
@@ -57,6 +57,7 @@
#include "pegasus_rpc_types.h"
#include "pegasus_server_write.h"
#include "perf_counter/perf_counter.h"
+#include "perf_counter/perf_counter_wrapper.h"
#include "replica_admin_types.h"
#include "rrdb/rrdb.code.definition.h"
#include "rrdb/rrdb_types.h"
diff --git a/src/server/pegasus_server_impl.h b/src/server/pegasus_server_impl.h
index 7f313cf68..c9c5fe2a1 100644
--- a/src/server/pegasus_server_impl.h
+++ b/src/server/pegasus_server_impl.h
@@ -42,13 +42,13 @@
#include "pegasus_scan_context.h"
#include "pegasus_utils.h"
#include "pegasus_value_schema.h"
-#include "perf_counter/perf_counter_wrapper.h"
#include "range_read_limiter.h"
#include "replica/replication_app_base.h"
#include "runtime/task/task.h"
#include "runtime/task/task_tracker.h"
#include "utils/error_code.h"
#include "utils/flags.h"
+#include "utils/metrics.h"
#include "utils/rand.h"
#include "utils/synchronize.h"
@@ -69,6 +69,9 @@ class WriteBufferManager;
namespace dsn {
class blob;
class message_ex;
+class perf_counter_wrapper;
+class rpc_address;
+
namespace replication {
class detect_hotkey_request;
class detect_hotkey_response;
diff --git a/src/server/pegasus_server_impl_init.cpp b/src/server/pegasus_server_impl_init.cpp
index 0840c094b..cd51f0575 100644
--- a/src/server/pegasus_server_impl_init.cpp
+++ b/src/server/pegasus_server_impl_init.cpp
@@ -17,7 +17,6 @@
* under the License.
*/
-#include <fmt/core.h>
#include <rocksdb/cache.h>
#include <rocksdb/filter_policy.h>
#include <rocksdb/options.h>
@@ -29,6 +28,7 @@
#include <cstdint>
#include <memory>
#include <mutex>
+#include <set>
#include <string>
#include <unordered_map>
#include <utility>
@@ -53,6 +53,8 @@
#include "server/range_read_limiter.h"
#include "utils/flags.h"
#include "utils/fmt_logging.h"
+#include "utils/metrics.h"
+#include "utils/string_view.h"
#include "utils/strings.h"
#include "utils/token_bucket_throttling_controller.h"
@@ -117,10 +119,7 @@ METRIC_DEFINE_counter(replica,
dsn::metric_unit::kRequests,
"The number of abnormal read requests");
-METRIC_DEFINE_counter(replica,
- throttling_rejected_read_requests,
- dsn::metric_unit::kRequests,
- "The number of rejected read requests by throttling");
+METRIC_DECLARE_counter(throttling_rejected_read_requests);
METRIC_DEFINE_gauge_int64(replica,
rdb_total_sst_files,
diff --git a/src/server/pegasus_server_write.cpp b/src/server/pegasus_server_write.cpp
index e6cb5331d..9f8db6ca0 100644
--- a/src/server/pegasus_server_write.cpp
+++ b/src/server/pegasus_server_write.cpp
@@ -36,10 +36,12 @@
#include "runtime/rpc/rpc_holder.h"
#include "runtime/rpc/rpc_message.h"
#include "server/pegasus_write_service.h"
+#include "utils/autoref_ptr.h"
#include "utils/blob.h"
#include "utils/flags.h"
#include "utils/fmt_logging.h"
#include "utils/ports.h"
+#include "utils/string_view.h"
METRIC_DEFINE_counter(replica,
corrupt_writes,
diff --git a/src/server/pegasus_server_write.h b/src/server/pegasus_server_write.h
index d8a358164..2329ec586 100644
--- a/src/server/pegasus_server_write.h
+++ b/src/server/pegasus_server_write.h
@@ -30,6 +30,7 @@
#include "replica/replica_base.h"
#include "rrdb/rrdb_types.h"
#include "runtime/task/task_code.h"
+#include "utils/metrics.h"
namespace dsn {
class blob;
diff --git a/src/server/pegasus_write_service.cpp b/src/server/pegasus_write_service.cpp
index 969e7d141..22c0fc1fe 100644
--- a/src/server/pegasus_write_service.cpp
+++ b/src/server/pegasus_write_service.cpp
@@ -20,21 +20,21 @@
#include <fmt/core.h>
#include <fmt/ostream.h>
#include <rocksdb/status.h>
-#include <algorithm>
+#include <stddef.h>
+#include <functional>
#include <iosfwd>
-#include <string>
+#include <set>
+#include <vector>
#include "base/pegasus_rpc_types.h"
#include "bulk_load_types.h"
#include "capacity_unit_calculator.h"
#include "common/duplication_common.h"
-#include "common/gpid.h"
#include "common/replication.codes.h"
#include "duplication_internal_types.h"
#include "pegasus_value_schema.h"
#include "pegasus_write_service.h"
#include "pegasus_write_service_impl.h"
-#include "perf_counter/perf_counter.h"
#include "rrdb/rrdb.code.definition.h"
#include "rrdb/rrdb_types.h"
#include "runtime/api_layer1.h"
@@ -42,10 +42,11 @@
#include "runtime/task/async_calls.h"
#include "runtime/task/task_code.h"
#include "server/pegasus_server_impl.h"
-#include "utils/defer.h"
+#include "utils/autoref_ptr.h"
#include "utils/error_code.h"
#include "utils/flags.h"
#include "utils/fmt_logging.h"
+#include "utils/string_view.h"
namespace dsn {
class blob;
diff --git a/src/server/pegasus_write_service.h b/src/server/pegasus_write_service.h
index 9e79f9122..6c63c42d1 100644
--- a/src/server/pegasus_write_service.h
+++ b/src/server/pegasus_write_service.h
@@ -21,15 +21,16 @@
#include <cstdint>
#include <memory>
-#include <vector>
#include "common//duplication_common.h"
#include "common/common.h"
#include "replica/replica_base.h"
#include "utils/errors.h"
+#include "utils/metrics.h"
namespace dsn {
class blob;
+
namespace apps {
class check_and_mutate_request;
class check_and_mutate_response;
diff --git a/src/server/rocksdb_wrapper.cpp b/src/server/rocksdb_wrapper.cpp
index 701b02355..23f2d7c42 100644
--- a/src/server/rocksdb_wrapper.cpp
+++ b/src/server/rocksdb_wrapper.cpp
@@ -27,12 +27,11 @@
#include "pegasus_key_schema.h"
#include "pegasus_utils.h"
#include "pegasus_write_service_impl.h"
-#include "perf_counter/perf_counter.h"
-#include "perf_counter/perf_counter_wrapper.h"
#include "server/logging_utils.h"
#include "server/meta_store.h"
#include "server/pegasus_server_impl.h"
#include "server/pegasus_write_service.h"
+#include "utils/autoref_ptr.h"
#include "utils/blob.h"
#include "utils/fail_point.h"
#include "utils/fmt_logging.h"
diff --git a/src/server/rocksdb_wrapper.h b/src/server/rocksdb_wrapper.h
index fef30a629..e3eb600db 100644
--- a/src/server/rocksdb_wrapper.h
+++ b/src/server/rocksdb_wrapper.h
@@ -29,6 +29,7 @@
#include "pegasus_value_schema.h"
#include "replica/replica_base.h"
+#include "utils/metrics.h"
#include "utils/string_view.h"
namespace rocksdb {
@@ -36,10 +37,6 @@ class ColumnFamilyHandle;
class DB;
} // namespace rocksdb
-namespace dsn {
-class perf_counter_wrapper;
-} // namespace dsn
-
namespace pegasus {
namespace server {
diff --git a/src/server/test/pegasus_server_impl_test.cpp b/src/server/test/pegasus_server_impl_test.cpp
index 075193fef..880eeb403 100644
--- a/src/server/test/pegasus_server_impl_test.cpp
+++ b/src/server/test/pegasus_server_impl_test.cpp
@@ -33,15 +33,15 @@
#include "pegasus_const.h"
#include "pegasus_server_test_base.h"
-#include "perf_counter/perf_counter.h"
-#include "perf_counter/perf_counter_wrapper.h"
#include "rrdb/rrdb.code.definition.h"
#include "rrdb/rrdb_types.h"
#include "runtime/serverlet.h"
#include "server/pegasus_read_service.h"
+#include "utils/autoref_ptr.h"
#include "utils/blob.h"
#include "utils/error_code.h"
#include "utils/filesystem.h"
+#include "utils/metrics.h"
namespace pegasus {
namespace server {
diff --git a/src/utils/clock.cpp b/src/utils/clock.cpp
index 84764e908..cb1fd400c 100644
--- a/src/utils/clock.cpp
+++ b/src/utils/clock.cpp
@@ -17,6 +17,7 @@
#include "clock.h"
+#include "runtime/api_layer1.h"
#include "time_utils.h"
uint64_t dsn_now_ns() { return dsn::utils::clock::instance()->now_ns(); }
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 05/12: feat(new_metrics): migrate replica-level metrics for capacity_unit_calculator (#1387)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit 9d1a5665e51853311aa3fa628bd9c056176f73d7
Author: Dan Wang <wa...@apache.org>
AuthorDate: Tue Mar 14 17:18:35 2023 +0800
feat(new_metrics): migrate replica-level metrics for capacity_unit_calculator (#1387)
https://github.com/apache/incubator-pegasus/issues/1334
Migrate the metrics in capacity_unit_calculator to new framework, including read/write capacity units
and the number of bytes consumed by get, multi_get, batch_get, scan, put, multi_put, check_and_set,
check_and_mutate and backup requests.
---
src/server/capacity_unit_calculator.cpp | 142 ++++++++++++++++++--------------
src/server/capacity_unit_calculator.h | 26 +++---
src/server/pegasus_write_service.cpp | 47 +++++------
src/utils/metrics.h | 6 +-
4 files changed, 118 insertions(+), 103 deletions(-)
diff --git a/src/server/capacity_unit_calculator.cpp b/src/server/capacity_unit_calculator.cpp
index 157577b98..7a62b3d56 100644
--- a/src/server/capacity_unit_calculator.cpp
+++ b/src/server/capacity_unit_calculator.cpp
@@ -35,6 +35,61 @@
#include "utils/fmt_logging.h"
#include "utils/token_bucket_throttling_controller.h"
+METRIC_DEFINE_counter(replica,
+ read_capacity_units,
+ dsn::metric_unit::kCapacityUnits,
+ "The number of capacity units for read requests");
+
+METRIC_DEFINE_counter(replica,
+ write_capacity_units,
+ dsn::metric_unit::kCapacityUnits,
+ "The number of capacity units for write requests");
+
+METRIC_DEFINE_counter(replica,
+ get_bytes,
+ dsn::metric_unit::kBytes,
+ "The number of bytes for GET requests");
+
+METRIC_DEFINE_counter(replica,
+ multi_get_bytes,
+ dsn::metric_unit::kBytes,
+ "The number of bytes for MULTI_GET requests");
+
+METRIC_DEFINE_counter(replica,
+ batch_get_bytes,
+ dsn::metric_unit::kBytes,
+ "The number of bytes for BATCH_GET requests");
+
+METRIC_DEFINE_counter(replica,
+ scan_bytes,
+ dsn::metric_unit::kBytes,
+ "The number of bytes for SCAN requests");
+
+METRIC_DEFINE_counter(replica,
+ put_bytes,
+ dsn::metric_unit::kBytes,
+ "The number of bytes for PUT requests");
+
+METRIC_DEFINE_counter(replica,
+ multi_put_bytes,
+ dsn::metric_unit::kBytes,
+ "The number of bytes for MULTI_PUT requests");
+
+METRIC_DEFINE_counter(replica,
+ check_and_set_bytes,
+ dsn::metric_unit::kBytes,
+ "The number of bytes for CHECK_AND_SET requests");
+
+METRIC_DEFINE_counter(replica,
+ check_and_mutate_bytes,
+ dsn::metric_unit::kBytes,
+ "The number of bytes for CHECK_AND_MUTATE requests");
+
+METRIC_DEFINE_counter(replica,
+ backup_request_bytes,
+ dsn::metric_unit::kBytes,
+ "The number of bytes for backup requests");
+
namespace pegasus {
namespace server {
@@ -58,6 +113,17 @@ capacity_unit_calculator::capacity_unit_calculator(
std::shared_ptr<hotkey_collector> write_hotkey_collector,
std::shared_ptr<throttling_controller> read_size_throttling_controller)
: replica_base(r),
+ METRIC_VAR_INIT_replica(read_capacity_units),
+ METRIC_VAR_INIT_replica(write_capacity_units),
+ METRIC_VAR_INIT_replica(get_bytes),
+ METRIC_VAR_INIT_replica(multi_get_bytes),
+ METRIC_VAR_INIT_replica(batch_get_bytes),
+ METRIC_VAR_INIT_replica(scan_bytes),
+ METRIC_VAR_INIT_replica(put_bytes),
+ METRIC_VAR_INIT_replica(multi_put_bytes),
+ METRIC_VAR_INIT_replica(check_and_set_bytes),
+ METRIC_VAR_INIT_replica(check_and_mutate_bytes),
+ METRIC_VAR_INIT_replica(backup_request_bytes),
_read_hotkey_collector(read_hotkey_collector),
_write_hotkey_collector(write_hotkey_collector),
_read_size_throttling_controller(read_size_throttling_controller)
@@ -68,55 +134,6 @@ capacity_unit_calculator::capacity_unit_calculator(
_log_read_cu_size = log(FLAGS_perf_counter_read_capacity_unit_size) / log(2);
_log_write_cu_size = log(FLAGS_perf_counter_write_capacity_unit_size) / log(2);
-
- std::string str_gpid = r->get_gpid().to_string();
- char name[256];
- snprintf(name, 255, "recent.read.cu@%s", str_gpid.c_str());
- _pfc_recent_read_cu.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_VOLATILE_NUMBER,
- "statistic the recent read capacity units");
- snprintf(name, 255, "recent.write.cu@%s", str_gpid.c_str());
- _pfc_recent_write_cu.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_VOLATILE_NUMBER,
- "statistic the recent write capacity units");
-
- snprintf(name, 255, "get_bytes@%s", str_gpid.c_str());
- _pfc_get_bytes.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the get bytes");
-
- snprintf(name, 255, "multi_get_bytes@%s", str_gpid.c_str());
- _pfc_multi_get_bytes.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the multi get bytes");
-
- snprintf(name, 255, "batch_get_bytes@%s", str_gpid.c_str());
- _pfc_batch_get_bytes.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the batch get bytes");
-
- snprintf(name, 255, "scan_bytes@%s", str_gpid.c_str());
- _pfc_scan_bytes.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the scan bytes");
-
- snprintf(name, 255, "put_bytes@%s", str_gpid.c_str());
- _pfc_put_bytes.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the put bytes");
-
- snprintf(name, 255, "multi_put_bytes@%s", str_gpid.c_str());
- _pfc_multi_put_bytes.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the multi put bytes");
-
- snprintf(name, 255, "check_and_set_bytes@%s", str_gpid.c_str());
- _pfc_check_and_set_bytes.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the check and set bytes");
-
- snprintf(name, 255, "check_and_mutate_bytes@%s", str_gpid.c_str());
- _pfc_check_and_mutate_bytes.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the check and mutate bytes");
-
- snprintf(name, 255, "backup_request_bytes@%s", str_gpid.c_str());
- _pfc_backup_request_bytes.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the backup request bytes");
}
int64_t capacity_unit_calculator::add_read_cu(int64_t read_data_size)
@@ -125,7 +142,7 @@ int64_t capacity_unit_calculator::add_read_cu(int64_t read_data_size)
read_data_size > 0
? (read_data_size + FLAGS_perf_counter_read_capacity_unit_size - 1) >> _log_read_cu_size
: 1;
- _pfc_recent_read_cu->add(read_cu);
+ METRIC_VAR_INCREMENT_BY(read_capacity_units, read_cu);
_read_size_throttling_controller->consume_token(read_data_size);
return read_cu;
}
@@ -136,7 +153,7 @@ int64_t capacity_unit_calculator::add_write_cu(int64_t write_data_size)
? (write_data_size + FLAGS_perf_counter_write_capacity_unit_size - 1) >>
_log_write_cu_size
: 1;
- _pfc_recent_write_cu->add(write_cu);
+ METRIC_VAR_INCREMENT_BY(write_capacity_units, write_cu);
return write_cu;
}
@@ -146,7 +163,7 @@ void capacity_unit_calculator::add_get_cu(dsn::message_ex *req,
const dsn::blob &value)
{
auto total_size = key.size() + value.size();
- _pfc_get_bytes->add(total_size);
+ METRIC_VAR_INCREMENT_BY(get_bytes, total_size);
add_backup_request_bytes(req, total_size);
if (status != rocksdb::Status::kOk && status != rocksdb::Status::kNotFound) {
return;
@@ -173,7 +190,7 @@ void capacity_unit_calculator::add_multi_get_cu(dsn::message_ex *req,
data_size += hash_key.size() + kv.key.size() + kv.value.size();
}
auto total_size = hash_key.size() + multi_get_bytes;
- _pfc_multi_get_bytes->add(total_size);
+ METRIC_VAR_INCREMENT_BY(multi_get_bytes, total_size);
add_backup_request_bytes(req, total_size);
if (status != rocksdb::Status::kOk && status != rocksdb::Status::kNotFound &&
@@ -201,7 +218,7 @@ void capacity_unit_calculator::add_batch_get_cu(dsn::message_ex *req,
_read_hotkey_collector->capture_hash_key(data.hash_key, 1);
}
- _pfc_batch_get_bytes->add(data_size);
+ METRIC_VAR_INCREMENT_BY(batch_get_bytes, data_size);
add_backup_request_bytes(req, data_size);
if (status != rocksdb::Status::kOk && status != rocksdb::Status::kNotFound &&
@@ -237,7 +254,7 @@ void capacity_unit_calculator::add_scan_cu(dsn::message_ex *req,
data_size += kv.key.size() + kv.value.size();
}
add_read_cu(data_size);
- _pfc_scan_bytes->add(data_size);
+ METRIC_VAR_INCREMENT_BY(scan_bytes, data_size);
add_backup_request_bytes(req, data_size);
}
@@ -269,7 +286,7 @@ void capacity_unit_calculator::add_put_cu(int32_t status,
const dsn::blob &key,
const dsn::blob &value)
{
- _pfc_put_bytes->add(key.size() + value.size());
+ METRIC_VAR_INCREMENT_BY(put_bytes, key.size() + value.size());
if (status != rocksdb::Status::kOk) {
return;
}
@@ -296,7 +313,7 @@ void capacity_unit_calculator::add_multi_put_cu(int32_t status,
multi_put_bytes += kv.key.size() + kv.value.size();
data_size += hash_key.size() + kv.key.size() + kv.value.size();
}
- _pfc_multi_put_bytes->add(hash_key.size() + multi_put_bytes);
+ METRIC_VAR_INCREMENT_BY(multi_put_bytes, hash_key.size() + multi_put_bytes);
uint64_t key_count = kvs.size();
_write_hotkey_collector->capture_hash_key(hash_key, key_count);
@@ -343,8 +360,9 @@ void capacity_unit_calculator::add_check_and_set_cu(int32_t status,
const dsn::blob &value)
{
- _pfc_check_and_set_bytes->add(hash_key.size() + check_sort_key.size() + set_sort_key.size() +
- value.size());
+ METRIC_VAR_INCREMENT_BY(check_and_set_bytes,
+ hash_key.size() + check_sort_key.size() + set_sort_key.size() +
+ value.size());
if (status != rocksdb::Status::kOk && status != rocksdb::Status::kInvalidArgument &&
status != rocksdb::Status::kTryAgain) {
return;
@@ -370,8 +388,8 @@ void capacity_unit_calculator::add_check_and_mutate_cu(
check_and_mutate_bytes += m.sort_key.size() + m.value.size();
data_size += hash_key.size() + m.sort_key.size() + m.value.size();
}
- _pfc_check_and_mutate_bytes->add(hash_key.size() + check_sort_key.size() +
- check_and_mutate_bytes);
+ METRIC_VAR_INCREMENT_BY(check_and_mutate_bytes,
+ hash_key.size() + check_sort_key.size() + check_and_mutate_bytes);
if (status != rocksdb::Status::kOk && status != rocksdb::Status::kInvalidArgument &&
status != rocksdb::Status::kTryAgain) {
@@ -389,7 +407,7 @@ void capacity_unit_calculator::add_check_and_mutate_cu(
void capacity_unit_calculator::add_backup_request_bytes(dsn::message_ex *req, int64_t bytes)
{
if (req->is_backup_request()) {
- _pfc_backup_request_bytes->add(bytes);
+ METRIC_VAR_INCREMENT_BY(backup_request_bytes, bytes);
}
}
diff --git a/src/server/capacity_unit_calculator.h b/src/server/capacity_unit_calculator.h
index 0b91ea4d2..6d30a07ef 100644
--- a/src/server/capacity_unit_calculator.h
+++ b/src/server/capacity_unit_calculator.h
@@ -108,18 +108,20 @@ private:
uint32_t _log_read_cu_size;
uint32_t _log_write_cu_size;
- ::dsn::perf_counter_wrapper _pfc_recent_read_cu;
- ::dsn::perf_counter_wrapper _pfc_recent_write_cu;
-
- ::dsn::perf_counter_wrapper _pfc_get_bytes;
- ::dsn::perf_counter_wrapper _pfc_multi_get_bytes;
- ::dsn::perf_counter_wrapper _pfc_batch_get_bytes;
- ::dsn::perf_counter_wrapper _pfc_scan_bytes;
- ::dsn::perf_counter_wrapper _pfc_put_bytes;
- ::dsn::perf_counter_wrapper _pfc_multi_put_bytes;
- ::dsn::perf_counter_wrapper _pfc_check_and_set_bytes;
- ::dsn::perf_counter_wrapper _pfc_check_and_mutate_bytes;
- ::dsn::perf_counter_wrapper _pfc_backup_request_bytes;
+ METRIC_VAR_DECLARE_counter(read_capacity_units);
+ METRIC_VAR_DECLARE_counter(write_capacity_units);
+
+ METRIC_VAR_DECLARE_counter(get_bytes);
+ METRIC_VAR_DECLARE_counter(multi_get_bytes);
+ METRIC_VAR_DECLARE_counter(batch_get_bytes);
+ METRIC_VAR_DECLARE_counter(scan_bytes);
+
+ METRIC_VAR_DECLARE_counter(put_bytes);
+ METRIC_VAR_DECLARE_counter(multi_put_bytes);
+ METRIC_VAR_DECLARE_counter(check_and_set_bytes);
+ METRIC_VAR_DECLARE_counter(check_and_mutate_bytes);
+
+ METRIC_VAR_DECLARE_counter(backup_request_bytes);
/*
hotkey capturing weight rules:
diff --git a/src/server/pegasus_write_service.cpp b/src/server/pegasus_write_service.cpp
index 4889329d9..969e7d141 100644
--- a/src/server/pegasus_write_service.cpp
+++ b/src/server/pegasus_write_service.cpp
@@ -55,83 +55,82 @@ class message_ex;
METRIC_DEFINE_counter(replica,
put_requests,
dsn::metric_unit::kRequests,
- "The number of PUT requests for each replica");
+ "The number of PUT requests");
METRIC_DEFINE_counter(replica,
multi_put_requests,
dsn::metric_unit::kRequests,
- "The number of MULTI_PUT requests for each replica");
+ "The number of MULTI_PUT requests");
METRIC_DEFINE_counter(replica,
remove_requests,
dsn::metric_unit::kRequests,
- "The number of REMOVE requests for each replica");
+ "The number of REMOVE requests");
METRIC_DEFINE_counter(replica,
multi_remove_requests,
dsn::metric_unit::kRequests,
- "The number of MULTI_REMOVE requests for each replica");
+ "The number of MULTI_REMOVE requests");
METRIC_DEFINE_counter(replica,
incr_requests,
dsn::metric_unit::kRequests,
- "The number of INCR requests for each replica");
+ "The number of INCR requests");
METRIC_DEFINE_counter(replica,
check_and_set_requests,
dsn::metric_unit::kRequests,
- "The number of CHECK_AND_SET requests for each replica");
+ "The number of CHECK_AND_SET requests");
METRIC_DEFINE_counter(replica,
check_and_mutate_requests,
dsn::metric_unit::kRequests,
- "The number of CHECK_AND_MUTATE requests for each replica");
+ "The number of CHECK_AND_MUTATE requests");
METRIC_DEFINE_percentile_int64(replica,
put_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of PUT requests for each replica");
+ "The latency of PUT requests");
METRIC_DEFINE_percentile_int64(replica,
multi_put_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of MULTI_PUT requests for each replica");
+ "The latency of MULTI_PUT requests");
METRIC_DEFINE_percentile_int64(replica,
remove_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of REMOVE requests for each replica");
+ "The latency of REMOVE requests");
METRIC_DEFINE_percentile_int64(replica,
multi_remove_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of MULTI_REMOVE requests for each replica");
+ "The latency of MULTI_REMOVE requests");
METRIC_DEFINE_percentile_int64(replica,
incr_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of INCR requests for each replica");
+ "The latency of INCR requests");
METRIC_DEFINE_percentile_int64(replica,
check_and_set_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of CHECK_AND_SET requests for each replica");
+ "The latency of CHECK_AND_SET requests");
METRIC_DEFINE_percentile_int64(replica,
check_and_mutate_latency_ns,
dsn::metric_unit::kNanoSeconds,
- "The latency of CHECK_AND_MUTATE requests for each replica");
+ "The latency of CHECK_AND_MUTATE requests");
METRIC_DEFINE_counter(replica,
dup_requests,
dsn::metric_unit::kRequests,
- "The number of DUPLICATE requests for each replica");
+ "The number of DUPLICATE requests");
-METRIC_DEFINE_percentile_int64(
- replica,
- dup_time_lag_ms,
- dsn::metric_unit::kMilliSeconds,
- "the time lag (in ms) between master and slave in the duplication for each replica");
+METRIC_DEFINE_percentile_int64(replica,
+ dup_time_lag_ms,
+ dsn::metric_unit::kMilliSeconds,
+ "the time lag (in ms) between master and slave in the duplication");
METRIC_DEFINE_counter(
replica,
@@ -176,12 +175,6 @@ pegasus_write_service::pegasus_write_service(pegasus_server_impl *server)
_put_batch_size(0),
_remove_batch_size(0)
{
- _dup_lagging_write_threshold_ms = dsn_config_get_value_int64(
- "pegasus.server",
- "dup_lagging_write_threshold_ms",
- 10 * 1000,
- "If the duration that a write flows from master to slave is larger than this threshold, "
- "the write is defined a lagging write.");
}
pegasus_write_service::~pegasus_write_service() {}
@@ -372,7 +365,7 @@ int pegasus_write_service::duplicate(int64_t decree,
METRIC_VAR_INCREMENT(dup_requests);
METRIC_VAR_AUTO_LATENCY(
dup_time_lag_ms, request.timestamp * 1000, [this](uint64_t latency) {
- if (latency > _dup_lagging_write_threshold_ms) {
+ if (latency > FLAGS_dup_lagging_write_threshold_ms) {
METRIC_VAR_INCREMENT(dup_lagging_writes);
}
});
diff --git a/src/utils/metrics.h b/src/utils/metrics.h
index facc83f26..333534703 100644
--- a/src/utils/metrics.h
+++ b/src/utils/metrics.h
@@ -168,8 +168,9 @@ class error_code;
// Perform increment-related operations on metrics including gauge and counter.
#define METRIC_VAR_INCREMENT_BY(name, x) \
do { \
- if (x != 0) { \
- _##name->increment_by(x); \
+ const auto v = (x); \
+ if (v != 0) { \
+ _##name->increment_by(v); \
} \
} while (0)
@@ -611,6 +612,7 @@ enum class metric_unit : size_t
kSeconds,
kBytes,
kMegaBytes,
+ kCapacityUnits,
kRequests,
kSeeks,
kPointLookups,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 03/12: feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1) (#1374)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit fcae8db24495b8af2ce9fb15fc3c99bc4aeb2474
Author: Dan Wang <wa...@apache.org>
AuthorDate: Tue Mar 7 14:32:56 2023 +0800
feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1) (#1374)
This PR is to migrate replica-level metrics of `pegasus_server_impl` to new framework
(https://github.com/apache/incubator-pegasus/issues/1333). Since there are many replica-level
metrics in `pegasus_server_impl`, this PR is part 1 for this migration and other metrics (all
of which are gauges) will be migrated in the later PRs.
---
src/server/pegasus_server_impl.cpp | 250 ++++++++++++++-------------
src/server/pegasus_server_impl.h | 32 ++--
src/server/pegasus_server_impl_init.cpp | 126 +++++++-------
src/server/pegasus_write_service_impl.h | 5 +-
src/server/rocksdb_wrapper.cpp | 6 +-
src/server/rocksdb_wrapper.h | 2 +-
src/server/test/pegasus_server_impl_test.cpp | 4 +-
src/utils/metrics.h | 16 +-
src/utils/test/metrics_test.cpp | 2 -
src/utils/time_utils.h | 2 +-
10 files changed, 242 insertions(+), 203 deletions(-)
diff --git a/src/server/pegasus_server_impl.cpp b/src/server/pegasus_server_impl.cpp
index dba81be2e..c9bd093b3 100644
--- a/src/server/pegasus_server_impl.cpp
+++ b/src/server/pegasus_server_impl.cpp
@@ -316,13 +316,50 @@ int pegasus_server_impl::on_batched_write_requests(int64_t decree,
return _server_write->on_batched_write_requests(requests, count, decree, timestamp);
}
+// Since LOG_ERROR_PREFIX depends on log_prefix(), this method could not be declared as static or
+// with anonymous namespace.
+void pegasus_server_impl::log_expired_data(const char *op,
+ const dsn::rpc_address &addr,
+ const dsn::blob &hash_key,
+ const dsn::blob &sort_key) const
+{
+ LOG_ERROR_PREFIX("rocksdb data expired for {} from {}: hash_key = \"{}\", sort_key = \"{}\"",
+ op,
+ addr,
+ pegasus::utils::c_escape_string(hash_key),
+ pegasus::utils::c_escape_string(sort_key));
+}
+
+void pegasus_server_impl::log_expired_data(const char *op,
+ const dsn::rpc_address &addr,
+ const dsn::blob &key) const
+{
+ dsn::blob hash_key, sort_key;
+ pegasus_restore_key(key, hash_key, sort_key);
+ log_expired_data(op, addr, hash_key, sort_key);
+}
+
+void pegasus_server_impl::log_expired_data(const char *op,
+ const dsn::rpc_address &addr,
+ const rocksdb::Slice &key) const
+{
+ dsn::blob raw_key(key.data(), 0, key.size());
+ log_expired_data(op, addr, raw_key);
+}
+
+#define LOG_EXPIRED_DATA_IF_VERBOSE(...) \
+ do { \
+ if (dsn_unlikely(FLAGS_rocksdb_verbose_log)) { \
+ log_expired_data(__FUNCTION__, rpc.remote_address(), ##__VA_ARGS__); \
+ } \
+ } while (0)
+
void pegasus_server_impl::on_get(get_rpc rpc)
{
- CHECK(_is_open, "");
- _pfc_get_qps->increment();
- uint64_t start_time = dsn_now_ns();
+ CHECK_TRUE(_is_open);
+
+ METRIC_VAR_INCREMENT(get_requests);
- const auto &key = rpc.request();
auto &resp = rpc.response();
resp.app_id = _gpid.get_app_id();
resp.partition_index = _gpid.get_partition_index();
@@ -334,21 +371,20 @@ void pegasus_server_impl::on_get(get_rpc rpc)
return;
}
+ METRIC_VAR_AUTO_LATENCY(get_latency_ns);
+
+ const auto &key = rpc.request();
rocksdb::Slice skey(key.data(), key.length());
std::string value;
rocksdb::Status status = _db->Get(_data_cf_rd_opts, _data_cf, skey, &value);
if (status.ok()) {
if (check_if_record_expired(utils::epoch_now(), value)) {
- _pfc_recent_expire_count->increment();
- if (FLAGS_rocksdb_verbose_log) {
- LOG_ERROR_PREFIX("rocksdb data expired for get from {}", rpc.remote_address());
- }
+ METRIC_VAR_INCREMENT(read_expired_values);
+ LOG_EXPIRED_DATA_IF_VERBOSE(key);
status = rocksdb::Status::NotFound();
}
- }
-
- if (!status.ok()) {
+ } else {
if (FLAGS_rocksdb_verbose_log) {
::dsn::blob hash_key, sort_key;
pegasus_restore_key(key, hash_key, sort_key);
@@ -371,7 +407,7 @@ void pegasus_server_impl::on_get(get_rpc rpc)
usleep(10 * 1000);
#endif
- uint64_t time_used = dsn_now_ns() - start_time;
+ auto time_used = METRIC_VAR_AUTO_LATENCY_DURATION_NS(get_latency_ns);
if (is_get_abnormal(time_used, value.size())) {
::dsn::blob hash_key, sort_key;
pegasus_restore_key(key, hash_key, sort_key);
@@ -384,7 +420,7 @@ void pegasus_server_impl::on_get(get_rpc rpc)
status.ToString(),
value.size(),
time_used);
- _pfc_recent_abnormal_count->increment();
+ METRIC_VAR_INCREMENT(abnormal_read_requests);
}
resp.error = status.code();
@@ -393,17 +429,14 @@ void pegasus_server_impl::on_get(get_rpc rpc)
}
_cu_calculator->add_get_cu(rpc.dsn_request(), resp.error, key, resp.value);
- _pfc_get_latency->set(dsn_now_ns() - start_time);
}
void pegasus_server_impl::on_multi_get(multi_get_rpc rpc)
{
- CHECK(_is_open, "");
- _pfc_multi_get_qps->increment();
- uint64_t start_time = dsn_now_ns();
+ CHECK_TRUE(_is_open);
+
+ METRIC_VAR_INCREMENT(multi_get_requests);
- const auto &request = rpc.request();
- dsn::message_ex *req = rpc.dsn_request();
auto &resp = rpc.response();
resp.app_id = _gpid.get_app_id();
resp.partition_index = _gpid.get_partition_index();
@@ -415,6 +448,10 @@ void pegasus_server_impl::on_multi_get(multi_get_rpc rpc)
return;
}
+ METRIC_VAR_AUTO_LATENCY(multi_get_latency_ns);
+
+ const auto &request = rpc.request();
+ dsn::message_ex *req = rpc.dsn_request();
if (!is_filter_type_supported(request.sort_key_filter_type)) {
LOG_ERROR_PREFIX(
"invalid argument for multi_get from {}: sort key filter type {} not supported",
@@ -422,7 +459,6 @@ void pegasus_server_impl::on_multi_get(multi_get_rpc rpc)
request.sort_key_filter_type);
resp.error = rocksdb::Status::kInvalidArgument;
_cu_calculator->add_multi_get_cu(req, resp.error, request.hash_key, resp.kvs);
- _pfc_multi_get_latency->set(dsn_now_ns() - start_time);
return;
}
@@ -509,8 +545,6 @@ void pegasus_server_impl::on_multi_get(multi_get_rpc rpc)
}
resp.error = rocksdb::Status::kOk;
_cu_calculator->add_multi_get_cu(req, resp.error, request.hash_key, resp.kvs);
- _pfc_multi_get_latency->set(dsn_now_ns() - start_time);
-
return;
}
@@ -713,7 +747,6 @@ void pegasus_server_impl::on_multi_get(multi_get_rpc rpc)
for (int i = 0; i < keys.size(); i++) {
rocksdb::Status &status = statuses[i];
std::string &value = values[i];
- // print log
if (!status.ok()) {
if (FLAGS_rocksdb_verbose_log) {
LOG_ERROR_PREFIX("rocksdb get failed for multi_get from {}: hash_key = \"{}\", "
@@ -727,41 +760,38 @@ void pegasus_server_impl::on_multi_get(multi_get_rpc rpc)
rpc.remote_address(),
status.ToString());
}
- }
- // check ttl
- if (status.ok()) {
- uint32_t expire_ts = pegasus_extract_expire_ts(_pegasus_data_version, value);
- if (expire_ts > 0 && expire_ts <= epoch_now) {
- expire_count++;
- if (FLAGS_rocksdb_verbose_log) {
- LOG_ERROR_PREFIX("rocksdb data expired for multi_get from {}",
- rpc.remote_address());
- }
- status = rocksdb::Status::NotFound();
- }
- }
- // extract value
- if (status.ok()) {
- // check if exceed limit
- if (count >= max_kv_count || size >= max_kv_size) {
- exceed_limit = true;
- break;
- }
- ::dsn::apps::key_value kv;
- kv.key = request.sort_keys[i];
- if (!request.no_value) {
- pegasus_extract_user_data(_pegasus_data_version, std::move(value), kv.value);
+
+ if (status.IsNotFound()) {
+ continue;
}
- count++;
- size += kv.key.length() + kv.value.length();
- resp.kvs.emplace_back(std::move(kv));
- }
- // if error occurred
- if (!status.ok() && !status.IsNotFound()) {
+
error_occurred = true;
final_status = status;
break;
}
+
+ // check ttl
+ if (check_if_record_expired(epoch_now, value)) {
+ expire_count++;
+ LOG_EXPIRED_DATA_IF_VERBOSE(request.hash_key, request.sort_keys[i]);
+ status = rocksdb::Status::NotFound();
+ continue;
+ }
+
+ // check if exceed limit
+ if (dsn_unlikely(count >= max_kv_count || size >= max_kv_size)) {
+ exceed_limit = true;
+ break;
+ }
+
+ ::dsn::apps::key_value kv;
+ kv.key = request.sort_keys[i];
+ if (!request.no_value) {
+ pegasus_extract_user_data(_pegasus_data_version, std::move(value), kv.value);
+ }
+ count++;
+ size += kv.key.length() + kv.value.length();
+ resp.kvs.emplace_back(std::move(kv));
}
if (error_occurred) {
@@ -779,7 +809,7 @@ void pegasus_server_impl::on_multi_get(multi_get_rpc rpc)
usleep(10 * 1000);
#endif
- uint64_t time_used = dsn_now_ns() - start_time;
+ auto time_used = METRIC_VAR_AUTO_LATENCY_DURATION_NS(multi_get_latency_ns);
if (is_multi_get_abnormal(time_used, size, iteration_count)) {
LOG_WARNING_PREFIX(
"rocksdb abnormal multi_get from {}: hash_key = {}, "
@@ -805,25 +835,20 @@ void pegasus_server_impl::on_multi_get(multi_get_rpc rpc)
expire_count,
filter_count,
time_used);
- _pfc_recent_abnormal_count->increment();
+ METRIC_VAR_INCREMENT(abnormal_read_requests);
}
- if (expire_count > 0) {
- _pfc_recent_expire_count->add(expire_count);
- }
- if (filter_count > 0) {
- _pfc_recent_filter_count->add(filter_count);
- }
+ METRIC_VAR_INCREMENT_BY(read_expired_values, expire_count);
+ METRIC_VAR_INCREMENT_BY(read_filtered_values, filter_count);
_cu_calculator->add_multi_get_cu(req, resp.error, request.hash_key, resp.kvs);
- _pfc_multi_get_latency->set(dsn_now_ns() - start_time);
}
void pegasus_server_impl::on_batch_get(batch_get_rpc rpc)
{
- CHECK(_is_open, "");
- _pfc_batch_get_qps->increment();
- int64_t start_time = dsn_now_ns();
+ CHECK_TRUE(_is_open);
+
+ METRIC_VAR_INCREMENT(batch_get_requests);
auto &response = rpc.response();
response.app_id = _gpid.get_app_id();
@@ -836,13 +861,14 @@ void pegasus_server_impl::on_batch_get(batch_get_rpc rpc)
return;
}
+ METRIC_VAR_AUTO_LATENCY(batch_get_latency_ns);
+
const auto &request = rpc.request();
if (request.keys.empty()) {
response.error = rocksdb::Status::kInvalidArgument;
LOG_ERROR_PREFIX("Invalid argument for batch_get from {}: 'keys' field in request is empty",
rpc.remote_address().to_string());
_cu_calculator->add_batch_get_cu(rpc.dsn_request(), response.error, response.data);
- _pfc_batch_get_latency->set(dsn_now_ns() - start_time);
return;
}
@@ -861,6 +887,8 @@ void pegasus_server_impl::on_batch_get(batch_get_rpc rpc)
bool error_occurred = false;
int64_t total_data_size = 0;
uint32_t epoch_now = pegasus::utils::epoch_now();
+ uint64_t expire_count = 0;
+
std::vector<std::string> values;
std::vector<rocksdb::Status> statuses = _db->MultiGet(_data_cf_rd_opts, keys, &values);
response.data.reserve(request.keys.size());
@@ -876,13 +904,8 @@ void pegasus_server_impl::on_batch_get(batch_get_rpc rpc)
if (dsn_likely(status.ok())) {
if (check_if_record_expired(epoch_now, value)) {
- if (FLAGS_rocksdb_verbose_log) {
- LOG_ERROR_PREFIX(
- "rocksdb data expired for batch_get from {}, hash_key = {}, sort_key = {}",
- rpc.remote_address().to_string(),
- pegasus::utils::c_escape_string(hash_key),
- pegasus::utils::c_escape_string(sort_key));
- }
+ ++expire_count;
+ LOG_EXPIRED_DATA_IF_VERBOSE(hash_key, sort_key);
continue;
}
@@ -920,7 +943,7 @@ void pegasus_server_impl::on_batch_get(batch_get_rpc rpc)
response.error = rocksdb::Status::kOk;
}
- int64_t time_used = dsn_now_ns() - start_time;
+ auto time_used = METRIC_VAR_AUTO_LATENCY_DURATION_NS(batch_get_latency_ns);
if (is_batch_get_abnormal(time_used, total_data_size, request.keys.size())) {
LOG_WARNING_PREFIX(
"rocksdb abnormal batch_get from {}: total data size = {}, row count = {}, "
@@ -929,33 +952,36 @@ void pegasus_server_impl::on_batch_get(batch_get_rpc rpc)
total_data_size,
request.keys.size(),
time_used / 1000);
- _pfc_recent_abnormal_count->increment();
+ METRIC_VAR_INCREMENT(abnormal_read_requests);
}
+ METRIC_VAR_INCREMENT_BY(read_expired_values, expire_count);
+
_cu_calculator->add_batch_get_cu(rpc.dsn_request(), response.error, response.data);
- _pfc_batch_get_latency->set(time_used);
}
void pegasus_server_impl::on_sortkey_count(sortkey_count_rpc rpc)
{
- CHECK(_is_open, "");
+ CHECK_TRUE(_is_open);
- _pfc_scan_qps->increment();
- uint64_t start_time = dsn_now_ns();
+ METRIC_VAR_INCREMENT(scan_requests);
- const auto &hash_key = rpc.request();
auto &resp = rpc.response();
resp.app_id = _gpid.get_app_id();
resp.partition_index = _gpid.get_partition_index();
resp.server = _primary_address;
+
if (!_read_size_throttling_controller->available()) {
rpc.error() = dsn::ERR_BUSY;
_counter_recent_read_throttling_reject_count->increment();
return;
}
+ METRIC_VAR_AUTO_LATENCY(scan_latency_ns);
+
// scan
::dsn::blob start_key, stop_key;
+ const auto &hash_key = rpc.request();
pegasus_generate_key(start_key, hash_key, ::dsn::blob());
pegasus_generate_next_blob(stop_key, hash_key);
rocksdb::Slice start(start_key.data(), start_key.length());
@@ -977,18 +1003,14 @@ void pegasus_server_impl::on_sortkey_count(sortkey_count_rpc rpc)
if (check_if_record_expired(epoch_now, it->value())) {
expire_count++;
- if (FLAGS_rocksdb_verbose_log) {
- LOG_ERROR_PREFIX("rocksdb data expired for sortkey_count from {}",
- rpc.remote_address());
- }
+ LOG_EXPIRED_DATA_IF_VERBOSE(it->key());
} else {
resp.count++;
}
it->Next();
}
- if (expire_count > 0) {
- _pfc_recent_expire_count->add(expire_count);
- }
+
+ METRIC_VAR_INCREMENT_BY(read_expired_values, expire_count);
resp.error = it->status().code();
if (!it->status().ok()) {
@@ -1014,7 +1036,6 @@ void pegasus_server_impl::on_sortkey_count(sortkey_count_rpc rpc)
}
_cu_calculator->add_sortkey_count_cu(rpc.dsn_request(), resp.error, hash_key);
- _pfc_scan_latency->set(dsn_now_ns() - start_time);
}
void pegasus_server_impl::on_ttl(ttl_rpc rpc)
@@ -1042,7 +1063,7 @@ void pegasus_server_impl::on_ttl(ttl_rpc rpc)
if (status.ok()) {
expire_ts = pegasus_extract_expire_ts(_pegasus_data_version, value);
if (check_if_ts_expired(now_ts, expire_ts)) {
- _pfc_recent_expire_count->increment();
+ METRIC_VAR_INCREMENT(read_expired_values);
if (FLAGS_rocksdb_verbose_log) {
LOG_ERROR_PREFIX("rocksdb data expired for ttl from {}", rpc.remote_address());
}
@@ -1082,12 +1103,10 @@ void pegasus_server_impl::on_ttl(ttl_rpc rpc)
void pegasus_server_impl::on_get_scanner(get_scanner_rpc rpc)
{
- CHECK(_is_open, "");
- _pfc_scan_qps->increment();
- uint64_t start_time = dsn_now_ns();
+ CHECK_TRUE(_is_open);
+
+ METRIC_VAR_INCREMENT(scan_requests);
- const auto &request = rpc.request();
- dsn::message_ex *req = rpc.dsn_request();
auto &resp = rpc.response();
resp.app_id = _gpid.get_app_id();
resp.partition_index = _gpid.get_partition_index();
@@ -1099,6 +1118,10 @@ void pegasus_server_impl::on_get_scanner(get_scanner_rpc rpc)
return;
}
+ METRIC_VAR_AUTO_LATENCY(scan_latency_ns);
+
+ const auto &request = rpc.request();
+ dsn::message_ex *req = rpc.dsn_request();
if (!is_filter_type_supported(request.hash_key_filter_type)) {
LOG_ERROR_PREFIX(
"invalid argument for get_scanner from {}: hash key filter type {} not supported",
@@ -1106,10 +1129,9 @@ void pegasus_server_impl::on_get_scanner(get_scanner_rpc rpc)
request.hash_key_filter_type);
resp.error = rocksdb::Status::kInvalidArgument;
_cu_calculator->add_scan_cu(req, resp.error, resp.kvs);
- _pfc_scan_latency->set(dsn_now_ns() - start_time);
-
return;
}
+
if (!is_filter_type_supported(request.sort_key_filter_type)) {
LOG_ERROR_PREFIX(
"invalid argument for get_scanner from {}: sort key filter type {} not supported",
@@ -1117,8 +1139,6 @@ void pegasus_server_impl::on_get_scanner(get_scanner_rpc rpc)
request.sort_key_filter_type);
resp.error = rocksdb::Status::kInvalidArgument;
_cu_calculator->add_scan_cu(req, resp.error, resp.kvs);
- _pfc_scan_latency->set(dsn_now_ns() - start_time);
-
return;
}
@@ -1174,8 +1194,6 @@ void pegasus_server_impl::on_get_scanner(get_scanner_rpc rpc)
}
resp.error = rocksdb::Status::kOk;
_cu_calculator->add_scan_cu(req, resp.error, resp.kvs);
- _pfc_scan_latency->set(dsn_now_ns() - start_time);
-
return;
}
@@ -1328,24 +1346,18 @@ void pegasus_server_impl::on_get_scanner(get_scanner_rpc rpc)
resp.context_id = pegasus::SCAN_CONTEXT_ID_COMPLETED;
}
- if (expire_count > 0) {
- _pfc_recent_expire_count->add(expire_count);
- }
- if (filter_count > 0) {
- _pfc_recent_filter_count->add(filter_count);
- }
+ METRIC_VAR_INCREMENT_BY(read_expired_values, expire_count);
+ METRIC_VAR_INCREMENT_BY(read_filtered_values, filter_count);
_cu_calculator->add_scan_cu(req, resp.error, resp.kvs);
- _pfc_scan_latency->set(dsn_now_ns() - start_time);
}
void pegasus_server_impl::on_scan(scan_rpc rpc)
{
- CHECK(_is_open, "");
- _pfc_scan_qps->increment();
- uint64_t start_time = dsn_now_ns();
- const auto &request = rpc.request();
- dsn::message_ex *req = rpc.dsn_request();
+ CHECK_TRUE(_is_open);
+
+ METRIC_VAR_INCREMENT(scan_requests);
+
auto &resp = rpc.response();
resp.app_id = _gpid.get_app_id();
resp.partition_index = _gpid.get_partition_index();
@@ -1357,6 +1369,10 @@ void pegasus_server_impl::on_scan(scan_rpc rpc)
return;
}
+ METRIC_VAR_AUTO_LATENCY(scan_latency_ns);
+
+ const auto &request = rpc.request();
+ dsn::message_ex *req = rpc.dsn_request();
std::unique_ptr<pegasus_scan_context> context = _context_cache.fetch(request.context_id);
if (context) {
rocksdb::Iterator *it = context->iterator.get();
@@ -1479,18 +1495,14 @@ void pegasus_server_impl::on_scan(scan_rpc rpc)
resp.context_id = pegasus::SCAN_CONTEXT_ID_COMPLETED;
}
- if (expire_count > 0) {
- _pfc_recent_expire_count->add(expire_count);
- }
- if (filter_count > 0) {
- _pfc_recent_filter_count->add(filter_count);
- }
+ METRIC_VAR_INCREMENT_BY(read_expired_values, expire_count);
+ METRIC_VAR_INCREMENT_BY(read_filtered_values, filter_count);
+
} else {
resp.error = rocksdb::Status::Code::kNotFound;
}
_cu_calculator->add_scan_cu(req, resp.error, resp.kvs);
- _pfc_scan_latency->set(dsn_now_ns() - start_time);
}
void pegasus_server_impl::on_clear_scanner(const int64_t &args) { _context_cache.fetch(args); }
diff --git a/src/server/pegasus_server_impl.h b/src/server/pegasus_server_impl.h
index bb0cbd8bd..8470b1e53 100644
--- a/src/server/pegasus_server_impl.h
+++ b/src/server/pegasus_server_impl.h
@@ -448,6 +448,15 @@ private:
dsn::replication::manual_compaction_status::type query_compact_status() const override;
+ // Log expired keys for verbose mode.
+ void log_expired_data(const char *op,
+ const dsn::rpc_address &addr,
+ const dsn::blob &hash_key,
+ const dsn::blob &sort_key) const;
+ void log_expired_data(const char *op, const dsn::rpc_address &addr, const dsn::blob &key) const;
+ void
+ log_expired_data(const char *op, const dsn::rpc_address &addr, const rocksdb::Slice &key) const;
+
private:
static const std::chrono::seconds kServerStatUpdateTimeSec;
static const std::string COMPRESSION_HEADER;
@@ -519,20 +528,19 @@ private:
std::shared_ptr<throttling_controller> _read_size_throttling_controller;
- // perf counters
- ::dsn::perf_counter_wrapper _pfc_get_qps;
- ::dsn::perf_counter_wrapper _pfc_multi_get_qps;
- ::dsn::perf_counter_wrapper _pfc_batch_get_qps;
- ::dsn::perf_counter_wrapper _pfc_scan_qps;
+ METRIC_VAR_DECLARE_counter(get_requests);
+ METRIC_VAR_DECLARE_counter(multi_get_requests);
+ METRIC_VAR_DECLARE_counter(batch_get_requests);
+ METRIC_VAR_DECLARE_counter(scan_requests);
- ::dsn::perf_counter_wrapper _pfc_get_latency;
- ::dsn::perf_counter_wrapper _pfc_multi_get_latency;
- ::dsn::perf_counter_wrapper _pfc_batch_get_latency;
- ::dsn::perf_counter_wrapper _pfc_scan_latency;
+ METRIC_VAR_DECLARE_percentile_int64(get_latency_ns);
+ METRIC_VAR_DECLARE_percentile_int64(multi_get_latency_ns);
+ METRIC_VAR_DECLARE_percentile_int64(batch_get_latency_ns);
+ METRIC_VAR_DECLARE_percentile_int64(scan_latency_ns);
- ::dsn::perf_counter_wrapper _pfc_recent_expire_count;
- ::dsn::perf_counter_wrapper _pfc_recent_filter_count;
- ::dsn::perf_counter_wrapper _pfc_recent_abnormal_count;
+ METRIC_VAR_DECLARE_counter(read_expired_values);
+ METRIC_VAR_DECLARE_counter(read_filtered_values);
+ METRIC_VAR_DECLARE_counter(abnormal_read_requests);
// rocksdb internal statistics
// server level
diff --git a/src/server/pegasus_server_impl_init.cpp b/src/server/pegasus_server_impl_init.cpp
index 896f8ab94..27aebdfab 100644
--- a/src/server/pegasus_server_impl_init.cpp
+++ b/src/server/pegasus_server_impl_init.cpp
@@ -62,6 +62,61 @@ class replica;
} // namespace replication
} // namespace dsn
+METRIC_DEFINE_counter(replica,
+ get_requests,
+ dsn::metric_unit::kRequests,
+ "The number of GET requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ multi_get_requests,
+ dsn::metric_unit::kRequests,
+ "The number of MULTI_GET requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ batch_get_requests,
+ dsn::metric_unit::kRequests,
+ "The number of BATCH_GET requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ scan_requests,
+ dsn::metric_unit::kRequests,
+ "The number of SCAN requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ get_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of GET requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ multi_get_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of MULTI_GET requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ batch_get_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of BATCH_GET requests for each replica");
+
+METRIC_DEFINE_percentile_int64(replica,
+ scan_latency_ns,
+ dsn::metric_unit::kNanoSeconds,
+ "The latency of SCAN requests for each replica");
+
+METRIC_DEFINE_counter(replica,
+ read_expired_values,
+ dsn::metric_unit::kValues,
+ "The number of expired values read for each replica");
+
+METRIC_DEFINE_counter(replica,
+ read_filtered_values,
+ dsn::metric_unit::kValues,
+ "The number of filtered values read for each replica");
+
+METRIC_DEFINE_counter(replica,
+ abnormal_read_requests,
+ dsn::metric_unit::kRequests,
+ "The number of abnormal read requests for each replica");
+
namespace pegasus {
namespace server {
@@ -403,7 +458,18 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
_last_durable_decree(0),
_is_checkpointing(false),
_manual_compact_svc(this),
- _partition_version(0)
+ _partition_version(0),
+ METRIC_VAR_INIT_replica(get_requests),
+ METRIC_VAR_INIT_replica(multi_get_requests),
+ METRIC_VAR_INIT_replica(batch_get_requests),
+ METRIC_VAR_INIT_replica(scan_requests),
+ METRIC_VAR_INIT_replica(get_latency_ns),
+ METRIC_VAR_INIT_replica(multi_get_latency_ns),
+ METRIC_VAR_INIT_replica(batch_get_latency_ns),
+ METRIC_VAR_INIT_replica(scan_latency_ns),
+ METRIC_VAR_INIT_replica(read_expired_values),
+ METRIC_VAR_INIT_replica(read_filtered_values),
+ METRIC_VAR_INIT_replica(abnormal_read_requests)
{
_primary_address = dsn::rpc_address(dsn_primary_address()).to_string();
_gpid = get_gpid();
@@ -610,64 +676,6 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
char name[256];
// register the perf counters
- snprintf(name, 255, "get_qps@%s", str_gpid.c_str());
- _pfc_get_qps.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the qps of GET request");
-
- snprintf(name, 255, "multi_get_qps@%s", str_gpid.c_str());
- _pfc_multi_get_qps.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the qps of MULTI_GET request");
-
- snprintf(name, 255, "batch_get_qps@%s", str_gpid.c_str());
- _pfc_batch_get_qps.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the qps of BATCH_GET request");
-
- snprintf(name, 255, "scan_qps@%s", str_gpid.c_str());
- _pfc_scan_qps.init_app_counter(
- "app.pegasus", name, COUNTER_TYPE_RATE, "statistic the qps of SCAN request");
-
- snprintf(name, 255, "get_latency@%s", str_gpid.c_str());
- _pfc_get_latency.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of GET request");
-
- snprintf(name, 255, "multi_get_latency@%s", str_gpid.c_str());
- _pfc_multi_get_latency.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of MULTI_GET request");
-
- snprintf(name, 255, "batch_get_latency@%s", str_gpid.c_str());
- _pfc_batch_get_latency.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of BATCH_GET request");
-
- snprintf(name, 255, "scan_latency@%s", str_gpid.c_str());
- _pfc_scan_latency.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_NUMBER_PERCENTILES,
- "statistic the latency of SCAN request");
-
- snprintf(name, 255, "recent.expire.count@%s", str_gpid.c_str());
- _pfc_recent_expire_count.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_VOLATILE_NUMBER,
- "statistic the recent expired value read count");
-
- snprintf(name, 255, "recent.filter.count@%s", str_gpid.c_str());
- _pfc_recent_filter_count.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_VOLATILE_NUMBER,
- "statistic the recent filtered value read count");
-
- snprintf(name, 255, "recent.abnormal.count@%s", str_gpid.c_str());
- _pfc_recent_abnormal_count.init_app_counter("app.pegasus",
- name,
- COUNTER_TYPE_VOLATILE_NUMBER,
- "statistic the recent abnormal read count");
-
snprintf(name, 255, "disk.storage.sst.count@%s", str_gpid.c_str());
_pfc_rdb_sst_count.init_app_counter(
"app.pegasus", name, COUNTER_TYPE_NUMBER, "statistic the count of sstable files");
diff --git a/src/server/pegasus_write_service_impl.h b/src/server/pegasus_write_service_impl.h
index b75b379a3..344c89560 100644
--- a/src/server/pegasus_write_service_impl.h
+++ b/src/server/pegasus_write_service_impl.h
@@ -89,8 +89,7 @@ public:
explicit impl(pegasus_server_impl *server)
: replica_base(server),
_primary_address(server->_primary_address),
- _pegasus_data_version(server->_pegasus_data_version),
- _pfc_recent_expire_count(server->_pfc_recent_expire_count)
+ _pegasus_data_version(server->_pegasus_data_version)
{
_rocksdb_wrapper = std::make_unique<rocksdb_wrapper>(server);
}
@@ -689,8 +688,6 @@ private:
const std::string _primary_address;
const uint32_t _pegasus_data_version;
- ::dsn::perf_counter_wrapper &_pfc_recent_expire_count;
-
std::unique_ptr<rocksdb_wrapper> _rocksdb_wrapper;
// for setting update_response.error after committed.
diff --git a/src/server/rocksdb_wrapper.cpp b/src/server/rocksdb_wrapper.cpp
index c5872ec13..701b02355 100644
--- a/src/server/rocksdb_wrapper.cpp
+++ b/src/server/rocksdb_wrapper.cpp
@@ -38,6 +38,8 @@
#include "utils/fmt_logging.h"
#include "utils/ports.h"
+METRIC_DECLARE_counter(read_expired_values);
+
namespace pegasus {
namespace server {
@@ -47,7 +49,7 @@ rocksdb_wrapper::rocksdb_wrapper(pegasus_server_impl *server)
_rd_opts(server->_data_cf_rd_opts),
_meta_cf(server->_meta_cf),
_pegasus_data_version(server->_pegasus_data_version),
- _pfc_recent_expire_count(server->_pfc_recent_expire_count),
+ METRIC_VAR_INIT_replica(read_expired_values),
_default_ttl(0)
{
_write_batch = std::make_unique<rocksdb::WriteBatch>();
@@ -69,7 +71,7 @@ int rocksdb_wrapper::get(dsn::string_view raw_key, /*out*/ db_get_context *ctx)
ctx->expire_ts = pegasus_extract_expire_ts(_pegasus_data_version, ctx->raw_value);
if (check_if_ts_expired(utils::epoch_now(), ctx->expire_ts)) {
ctx->expired = true;
- _pfc_recent_expire_count->increment();
+ METRIC_VAR_INCREMENT(read_expired_values);
}
return rocksdb::Status::kOk;
} else if (s.IsNotFound()) {
diff --git a/src/server/rocksdb_wrapper.h b/src/server/rocksdb_wrapper.h
index e3c713512..fef30a629 100644
--- a/src/server/rocksdb_wrapper.h
+++ b/src/server/rocksdb_wrapper.h
@@ -87,7 +87,7 @@ private:
rocksdb::ColumnFamilyHandle *_meta_cf;
const uint32_t _pegasus_data_version;
- dsn::perf_counter_wrapper &_pfc_recent_expire_count;
+ METRIC_VAR_DECLARE_counter(read_expired_values);
volatile uint32_t _default_ttl;
friend class rocksdb_wrapper_test;
diff --git a/src/server/test/pegasus_server_impl_test.cpp b/src/server/test/pegasus_server_impl_test.cpp
index 1363d8054..075193fef 100644
--- a/src/server/test/pegasus_server_impl_test.cpp
+++ b/src/server/test/pegasus_server_impl_test.cpp
@@ -77,7 +77,7 @@ public:
_server->update_app_envs(envs);
// do on_get/on_multi_get operation,
- long before_count = _server->_pfc_recent_abnormal_count->get_integer_value();
+ auto before_count = _server->METRIC_VAR_VALUE(abnormal_read_requests);
if (!test.is_multi_get) {
get_rpc rpc(std::make_unique<dsn::blob>(test_key), dsn::apps::RPC_RRDB_RRDB_GET);
_server->on_get(rpc);
@@ -90,7 +90,7 @@ public:
dsn::apps::RPC_RRDB_RRDB_MULTI_GET);
_server->on_multi_get(rpc);
}
- long after_count = _server->_pfc_recent_abnormal_count->get_integer_value();
+ auto after_count = _server->METRIC_VAR_VALUE(abnormal_read_requests);
ASSERT_EQ(before_count + test.expect_perf_counter_incr, after_count);
}
diff --git a/src/utils/metrics.h b/src/utils/metrics.h
index 2d6da6c0f..da1d056d8 100644
--- a/src/utils/metrics.h
+++ b/src/utils/metrics.h
@@ -165,7 +165,13 @@ class error_code;
#define METRIC_VAR_INIT_replica(name) METRIC_VAR_INIT(name, replica)
// Perform increment-related operations on metrics including gauge and counter.
-#define METRIC_VAR_INCREMENT_BY(name, x) _##name->increment_by(x)
+#define METRIC_VAR_INCREMENT_BY(name, x) \
+ do { \
+ if (x != 0) { \
+ _##name->increment_by(x); \
+ } \
+ } while (0)
+
#define METRIC_VAR_INCREMENT(name) _##name->increment()
// Perform set() operations on metrics including gauge and percentile.
@@ -176,10 +182,15 @@ class error_code;
// such as percentile.
#define METRIC_VAR_SET(name, ...) _##name->set(__VA_ARGS__)
+// Read the current measurement of the metric.
+#define METRIC_VAR_VALUE(name) _##name->value()
+
// Convenient macro that is used to compute latency automatically, which is dedicated to percentile.
#define METRIC_VAR_AUTO_LATENCY(name, ...) \
dsn::auto_latency __##name##_auto_latency(_##name, ##__VA_ARGS__)
+#define METRIC_VAR_AUTO_LATENCY_DURATION_NS(name) __##name##_auto_latency.duration_ns()
+
namespace dsn {
class metric; // IWYU pragma: keep
class metric_entity_prototype; // IWYU pragma: keep
@@ -598,6 +609,7 @@ enum class metric_unit : size_t
kMilliSeconds,
kSeconds,
kRequests,
+ kValues,
kInvalidUnit,
};
@@ -1409,6 +1421,8 @@ public:
}
}
+ inline uint64_t duration_ns() const { return _chrono.duration_ns(); }
+
private:
percentile_ptr<int64_t> _percentile;
utils::chronograph _chrono;
diff --git a/src/utils/test/metrics_test.cpp b/src/utils/test/metrics_test.cpp
index 670af36b7..9388a9a92 100644
--- a/src/utils/test/metrics_test.cpp
+++ b/src/utils/test/metrics_test.cpp
@@ -3134,8 +3134,6 @@ void MetricVarTest::test_set_percentile(size_t n, int64_t val)
EXPECT_EQ(std::vector<int64_t>(n, val), METRIC_VAR_SAMPLES(test_replica_percentile_int64_ns));
}
-#define METRIC_VAR_VALUE(name) _##name->value()
-
#define TEST_METRIC_VAR_INCREMENT(name) \
do { \
ASSERT_EQ(0, METRIC_VAR_VALUE(name)); \
diff --git a/src/utils/time_utils.h b/src/utils/time_utils.h
index 1db1676b0..265cc3c4d 100644
--- a/src/utils/time_utils.h
+++ b/src/utils/time_utils.h
@@ -142,7 +142,7 @@ public:
inline void reset_start_time() { _start_time_ns = dsn_now_ns(); }
- inline uint64_t duration_ns()
+ inline uint64_t duration_ns() const
{
auto now = dsn_now_ns();
CHECK_GE(now, _start_time_ns);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org
[incubator-pegasus] 09/12: feat(new_metrics): add server-level metric entity (#1415)
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch migrate-metrics-dev
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
commit 380ede5d9fb0ca435c2cf664e52cc854a78119a6
Author: Dan Wang <wa...@apache.org>
AuthorDate: Sat Mar 25 11:24:57 2023 +0800
feat(new_metrics): add server-level metric entity (#1415)
https://github.com/apache/incubator-pegasus/issues/1414
Add server-level metric entity, including:
- Define server-level entity prototype in metrics.cpp;
- Implement getter to acquire the (only) instance of server entity;
- Provide convenient macros to simplify operations for server entity.
---
src/utils/metrics.cpp | 8 ++++++++
src/utils/metrics.h | 7 +++++++
2 files changed, 15 insertions(+)
diff --git a/src/utils/metrics.cpp b/src/utils/metrics.cpp
index 2a15fb5aa..b32887c55 100644
--- a/src/utils/metrics.cpp
+++ b/src/utils/metrics.cpp
@@ -30,6 +30,14 @@
#include "utils/string_conv.h"
#include "utils/strings.h"
+METRIC_DEFINE_entity(server);
+
+dsn::metric_entity_ptr server_metric_entity()
+{
+ static auto entity = METRIC_ENTITY_server.instantiate("server");
+ return entity;
+}
+
namespace dsn {
DSN_DEFINE_uint64(metrics,
diff --git a/src/utils/metrics.h b/src/utils/metrics.h
index 9087b78e5..e69268006 100644
--- a/src/utils/metrics.h
+++ b/src/utils/metrics.h
@@ -164,6 +164,7 @@ class error_code;
#define METRIC_VAR_INIT(name, entity, ...) \
_##name(METRIC_##name.instantiate(entity##_metric_entity(), ##__VA_ARGS__))
#define METRIC_VAR_INIT_replica(name, ...) METRIC_VAR_INIT(name, replica, ##__VA_ARGS__)
+#define METRIC_VAR_INIT_server(name, ...) METRIC_VAR_INIT(name, server, ##__VA_ARGS__)
// Perform increment-related operations on metrics including gauge and counter.
#define METRIC_VAR_INCREMENT_BY(name, x) \
@@ -1445,3 +1446,9 @@ private:
};
} // namespace dsn
+
+// Since server_metric_entity() will be called in macros such as METRIC_VAR_INIT_server(), its
+// declaration should be put outside any namespace (for example dsn). server_metric_entity()
+// will not be qualified with any namespace. Once it was qualified with some namespace, its name
+// would not be resolved in any other namespace.
+dsn::metric_entity_ptr server_metric_entity();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pegasus.apache.org
For additional commands, e-mail: commits-help@pegasus.apache.org