You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@pegasus.apache.org by "empiredan (via GitHub)" <gi...@apache.org> on 2023/02/28 07:00:03 UTC
[GitHub] [incubator-pegasus] empiredan opened a new pull request, #1374: feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1)
empiredan opened a new pull request, #1374:
URL: https://github.com/apache/incubator-pegasus/pull/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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
[GitHub] [incubator-pegasus] acelyc111 merged pull request #1374: feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1)
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 merged PR #1374:
URL: https://github.com/apache/incubator-pegasus/pull/1374
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
[GitHub] [incubator-pegasus] acelyc111 commented on a diff in pull request #1374: feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1)
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1374:
URL: https://github.com/apache/incubator-pegasus/pull/1374#discussion_r1121172181
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -670,41 +673,44 @@ 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 (_verbose_log) {
- LOG_ERROR_PREFIX("rocksdb data expired for multi_get from {}",
- rpc.remote_address());
- }
- status = rocksdb::Status::NotFound();
+
+ if (status.IsNotFound()) {
+ continue;
}
+
+ error_occurred = true;
+ final_status = status;
+ break;
}
- // 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);
+
+ // check ttl
+ if (check_if_record_expired(epoch_now, value)) {
+ expire_count++;
+ if (_verbose_log) {
+ LOG_ERROR_PREFIX(
+ "rocksdb data expired for multi_get from {}: hash_key = \"{}\", "
+ "sort_key = \"{}\"",
+ rpc.remote_address(),
+ ::pegasus::utils::c_escape_string(request.hash_key),
+ ::pegasus::utils::c_escape_string(request.sort_keys[i]));
}
Review Comment:
Seems duplicated several times, how about use another macro to wrap these code?
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -670,41 +673,44 @@ 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 (_verbose_log) {
- LOG_ERROR_PREFIX("rocksdb data expired for multi_get from {}",
- rpc.remote_address());
- }
- status = rocksdb::Status::NotFound();
+
+ if (status.IsNotFound()) {
+ continue;
}
+
+ error_occurred = true;
+ final_status = status;
+ break;
}
- // 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);
+
+ // check ttl
+ if (check_if_record_expired(epoch_now, value)) {
+ expire_count++;
+ if (_verbose_log) {
+ LOG_ERROR_PREFIX(
+ "rocksdb data expired for multi_get from {}: hash_key = \"{}\", "
+ "sort_key = \"{}\"",
+ rpc.remote_address(),
+ ::pegasus::utils::c_escape_string(request.hash_key),
+ ::pegasus::utils::c_escape_string(request.sort_keys[i]));
}
- count++;
- size += kv.key.length() + kv.value.length();
- resp.kvs.emplace_back(std::move(kv));
+ status = rocksdb::Status::NotFound();
+ continue;
}
- // if error occurred
- if (!status.ok() && !status.IsNotFound()) {
- error_occurred = true;
- final_status = status;
+
+ // check if exceed limit
+ if (count >= max_kv_count || size >= max_kv_size) {
Review Comment:
How about add `dsn_likely/unlikey` in hot paths?
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -748,25 +754,24 @@ 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);
+ METRIC_VAR_INCREMENT_BY(read_expired_values, expire_count);
}
if (filter_count > 0) {
- _pfc_recent_filter_count->add(filter_count);
+ METRIC_VAR_INCREMENT_BY(read_filtered_values, filter_count);
}
Review Comment:
How about move the judgement into METRIC_VAR_INCREMENT_BY or counter metric?
##########
src/server/pegasus_write_service_impl.h:
##########
@@ -689,7 +691,7 @@ class pegasus_write_service::impl : public dsn::replication::replica_base
const std::string _primary_address;
const uint32_t _pegasus_data_version;
- ::dsn::perf_counter_wrapper &_pfc_recent_expire_count;
+ METRIC_VAR_DECLARE_counter(read_expired_values);
Review Comment:
Anywhere we use it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
[GitHub] [incubator-pegasus] empiredan commented on a diff in pull request #1374: feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1)
Posted by "empiredan (via GitHub)" <gi...@apache.org>.
empiredan commented on code in PR #1374:
URL: https://github.com/apache/incubator-pegasus/pull/1374#discussion_r1127302260
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -277,21 +334,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 (_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(__FUNCTION__, rpc.remote_address(), key);
Review Comment:
OK, I'll make it more graceful with macro.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
[GitHub] [incubator-pegasus] empiredan commented on a diff in pull request #1374: feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1)
Posted by "empiredan (via GitHub)" <gi...@apache.org>.
empiredan commented on code in PR #1374:
URL: https://github.com/apache/incubator-pegasus/pull/1374#discussion_r1126042317
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -670,41 +673,44 @@ 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 (_verbose_log) {
- LOG_ERROR_PREFIX("rocksdb data expired for multi_get from {}",
- rpc.remote_address());
- }
- status = rocksdb::Status::NotFound();
+
+ if (status.IsNotFound()) {
+ continue;
}
+
+ error_occurred = true;
+ final_status = status;
+ break;
}
- // 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);
+
+ // check ttl
+ if (check_if_record_expired(epoch_now, value)) {
+ expire_count++;
+ if (_verbose_log) {
+ LOG_ERROR_PREFIX(
+ "rocksdb data expired for multi_get from {}: hash_key = \"{}\", "
+ "sort_key = \"{}\"",
+ rpc.remote_address(),
+ ::pegasus::utils::c_escape_string(request.hash_key),
+ ::pegasus::utils::c_escape_string(request.sort_keys[i]));
}
- count++;
- size += kv.key.length() + kv.value.length();
- resp.kvs.emplace_back(std::move(kv));
+ status = rocksdb::Status::NotFound();
+ continue;
}
- // if error occurred
- if (!status.ok() && !status.IsNotFound()) {
- error_occurred = true;
- final_status = status;
+
+ // check if exceed limit
+ if (count >= max_kv_count || size >= max_kv_size) {
Review Comment:
OK, add `dsn_unlikely` here.
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -670,41 +673,44 @@ 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 (_verbose_log) {
- LOG_ERROR_PREFIX("rocksdb data expired for multi_get from {}",
- rpc.remote_address());
- }
- status = rocksdb::Status::NotFound();
+
+ if (status.IsNotFound()) {
+ continue;
}
+
+ error_occurred = true;
+ final_status = status;
+ break;
}
- // 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);
+
+ // check ttl
+ if (check_if_record_expired(epoch_now, value)) {
+ expire_count++;
+ if (_verbose_log) {
+ LOG_ERROR_PREFIX(
+ "rocksdb data expired for multi_get from {}: hash_key = \"{}\", "
+ "sort_key = \"{}\"",
+ rpc.remote_address(),
+ ::pegasus::utils::c_escape_string(request.hash_key),
+ ::pegasus::utils::c_escape_string(request.sort_keys[i]));
}
Review Comment:
OK, I'll encapsulate this into functions naming `log_expired_data*`.
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -748,25 +754,24 @@ 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);
+ METRIC_VAR_INCREMENT_BY(read_expired_values, expire_count);
}
if (filter_count > 0) {
- _pfc_recent_filter_count->add(filter_count);
+ METRIC_VAR_INCREMENT_BY(read_filtered_values, filter_count);
}
Review Comment:
Check if it is zero in `METRIC_VAR_INCREMENT_BY`.
##########
src/server/pegasus_write_service_impl.h:
##########
@@ -689,7 +691,7 @@ class pegasus_write_service::impl : public dsn::replication::replica_base
const std::string _primary_address;
const uint32_t _pegasus_data_version;
- ::dsn::perf_counter_wrapper &_pfc_recent_expire_count;
+ METRIC_VAR_DECLARE_counter(read_expired_values);
Review Comment:
This variable can be dropped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org
[GitHub] [incubator-pegasus] acelyc111 commented on a diff in pull request #1374: feat(new_metrics): migrate replica-level metrics for pegasus_server_impl (part 1)
Posted by "acelyc111 (via GitHub)" <gi...@apache.org>.
acelyc111 commented on code in PR #1374:
URL: https://github.com/apache/incubator-pegasus/pull/1374#discussion_r1126634892
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -277,21 +334,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 (_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(__FUNCTION__, rpc.remote_address(), key);
Review Comment:
If using macro, `__FUNCTION__` can be omitted?
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -259,13 +259,70 @@ 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
Review Comment:
Is there any reason of why not using macros?
##########
src/server/pegasus_server_impl.cpp:
##########
@@ -259,13 +259,70 @@ 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_if_verbose(const char *op,
+ const dsn::rpc_address &addr,
+ const dsn::blob &hash_key,
+ const dsn::blob &sort_key) const
+{
+ if (!_verbose_log) {
+ return;
+ }
+
+ log_expired_data(op, addr, hash_key, sort_key);
+}
+
+void pegasus_server_impl::log_expired_data_if_verbose(const char *op,
+ const dsn::rpc_address &addr,
+ const dsn::blob &key) const
+{
+ if (!_verbose_log) {
+ return;
+ }
+
+ log_expired_data(op, addr, key);
+}
+
+void pegasus_server_impl::log_expired_data_if_verbose(const char *op,
+ const dsn::rpc_address &addr,
+ const rocksdb::Slice &key) const
+{
+ if (!_verbose_log) {
Review Comment:
```suggestion
if (dsn_likely(!_verbose_log)) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@pegasus.apache.org
For additional commands, e-mail: dev-help@pegasus.apache.org