| /* |
| * 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. |
| */ |
| |
| #pragma once |
| |
| #include <getopt.h> |
| #include <thread> |
| #include <iomanip> |
| #include <fstream> |
| #include <queue> |
| #include <boost/algorithm/string.hpp> |
| #include <rocksdb/db.h> |
| #include <rocksdb/sst_dump_tool.h> |
| #include <rocksdb/env.h> |
| #include <rocksdb/statistics.h> |
| #include <dsn/cpp/json_helper.h> |
| #include <dsn/dist/remote_command.h> |
| #include <dsn/dist/replication/replication_ddl_client.h> |
| #include <dsn/dist/replication/mutation_log_tool.h> |
| #include <dsn/perf_counter/perf_counter_utils.h> |
| #include <dsn/utility/string_view.h> |
| #include <dsn/utils/time_utils.h> |
| |
| #include <rrdb/rrdb.code.definition.h> |
| #include <rrdb/rrdb_types.h> |
| #include <pegasus/version.h> |
| #include <pegasus/git_commit.h> |
| #include <pegasus/error.h> |
| #include <geo/lib/geo_client.h> |
| |
| #include "base/pegasus_key_schema.h" |
| #include "base/pegasus_value_schema.h" |
| #include "base/pegasus_utils.h" |
| |
| #include "command_executor.h" |
| #include "command_utils.h" |
| |
| using namespace dsn::replication; |
| |
| #define STR_I(var) #var |
| #define STR(var) STR_I(var) |
| #ifndef DSN_BUILD_TYPE |
| #define PEGASUS_BUILD_TYPE "" |
| #else |
| #define PEGASUS_BUILD_TYPE STR(DSN_BUILD_TYPE) |
| #endif |
| |
| DEFINE_TASK_CODE(LPC_SCAN_DATA, TASK_PRIORITY_COMMON, ::dsn::THREAD_POOL_DEFAULT) |
| enum scan_data_operator |
| { |
| SCAN_COPY, |
| SCAN_CLEAR, |
| SCAN_COUNT, |
| SCAN_GEN_GEO |
| }; |
| class top_container |
| { |
| public: |
| struct top_heap_item |
| { |
| std::string hash_key; |
| std::string sort_key; |
| long row_size; |
| top_heap_item(std::string &&hash_key_, std::string &&sort_key_, long row_size_) |
| : hash_key(std::move(hash_key_)), sort_key(std::move(sort_key_)), row_size(row_size_) |
| { |
| } |
| }; |
| struct top_heap_compare |
| { |
| bool operator()(top_heap_item i1, top_heap_item i2) { return i1.row_size < i2.row_size; } |
| }; |
| typedef std::priority_queue<top_heap_item, std::vector<top_heap_item>, top_heap_compare> |
| top_heap; |
| |
| top_container(int count) : _count(count) {} |
| |
| void push(std::string &&hash_key, std::string &&sort_key, long row_size) |
| { |
| dsn::utils::auto_lock<dsn::utils::ex_lock_nr> l(_lock); |
| if (_heap.size() < _count) { |
| _heap.emplace(std::move(hash_key), std::move(sort_key), row_size); |
| } else { |
| const top_heap_item &top = _heap.top(); |
| if (top.row_size < row_size) { |
| _heap.pop(); |
| _heap.emplace(std::move(hash_key), std::move(sort_key), row_size); |
| } |
| } |
| } |
| |
| top_heap &all() { return _heap; } |
| |
| private: |
| int _count; |
| top_heap _heap; |
| dsn::utils::ex_lock_nr _lock; |
| }; |
| |
| enum class histogram_type |
| { |
| HASH_KEY_SIZE, |
| SORT_KEY_SIZE, |
| VALUE_SIZE, |
| ROW_SIZE |
| }; |
| |
| struct scan_data_context |
| { |
| scan_data_operator op; |
| int split_id; |
| int max_batch_count; |
| int timeout_ms; |
| bool no_overwrite; // if set true, then use check_and_set() instead of set() |
| // when inserting data to destination table for copy_data, |
| // to not overwrite old data if it aleady exist. |
| pegasus::pegasus_client::filter_type sort_key_filter_type; |
| std::string sort_key_filter_pattern; |
| pegasus::pegasus_client::filter_type value_filter_type; |
| std::string value_filter_pattern; |
| pegasus::pegasus_client::pegasus_scanner_wrapper scanner; |
| pegasus::pegasus_client *client; |
| pegasus::geo::geo_client *geoclient; |
| std::atomic_bool *error_occurred; |
| std::atomic_long split_rows; |
| std::atomic_long split_request_count; |
| std::atomic_bool split_completed; |
| bool stat_size; |
| std::shared_ptr<rocksdb::Statistics> statistics; |
| int top_count; |
| top_container top_rows; |
| bool count_hash_key; |
| std::string last_hash_key; |
| std::atomic_long split_hash_key_count; |
| scan_data_context(scan_data_operator op_, |
| int split_id_, |
| int max_batch_count_, |
| int timeout_ms_, |
| pegasus::pegasus_client::pegasus_scanner_wrapper scanner_, |
| pegasus::pegasus_client *client_, |
| pegasus::geo::geo_client *geoclient_, |
| std::atomic_bool *error_occurred_, |
| bool stat_size_ = false, |
| std::shared_ptr<rocksdb::Statistics> statistics_ = nullptr, |
| int top_count_ = 0, |
| bool count_hash_key_ = false) |
| : op(op_), |
| split_id(split_id_), |
| max_batch_count(max_batch_count_), |
| timeout_ms(timeout_ms_), |
| no_overwrite(false), |
| sort_key_filter_type(pegasus::pegasus_client::FT_NO_FILTER), |
| value_filter_type(pegasus::pegasus_client::FT_NO_FILTER), |
| scanner(scanner_), |
| client(client_), |
| geoclient(geoclient_), |
| error_occurred(error_occurred_), |
| split_rows(0), |
| split_request_count(0), |
| split_completed(false), |
| stat_size(stat_size_), |
| statistics(statistics_), |
| top_count(top_count_), |
| top_rows(top_count_), |
| count_hash_key(count_hash_key_), |
| split_hash_key_count(0) |
| { |
| // max_batch_count should > 1 because scan may be terminated |
| // when split_request_count = 1 |
| dassert(max_batch_count > 1, ""); |
| } |
| void set_sort_key_filter(pegasus::pegasus_client::filter_type type, const std::string &pattern) |
| { |
| sort_key_filter_type = type; |
| sort_key_filter_pattern = pattern; |
| } |
| void set_value_filter(pegasus::pegasus_client::filter_type type, const std::string &pattern) |
| { |
| value_filter_type = type; |
| value_filter_pattern = pattern; |
| } |
| void set_no_overwrite() { no_overwrite = true; } |
| }; |
| inline void update_atomic_max(std::atomic_long &max, long value) |
| { |
| while (true) { |
| long old = max.load(); |
| if (value <= old || max.compare_exchange_weak(old, value)) { |
| break; |
| } |
| } |
| } |
| inline pegasus::pegasus_client::filter_type parse_filter_type(const std::string &name, |
| bool include_exact) |
| { |
| if (include_exact && name == "exact") |
| return pegasus::pegasus_client::FT_MATCH_EXACT; |
| else |
| return (pegasus::pegasus_client::filter_type)type_from_string( |
| dsn::apps::_filter_type_VALUES_TO_NAMES, |
| std::string("ft_match_") + name, |
| ::dsn::apps::filter_type::FT_NO_FILTER); |
| } |
| // return true if the data is valid for the filter |
| inline bool validate_filter(pegasus::pegasus_client::filter_type filter_type, |
| const std::string &filter_pattern, |
| const std::string &value) |
| { |
| switch (filter_type) { |
| case pegasus::pegasus_client::FT_NO_FILTER: |
| return true; |
| case pegasus::pegasus_client::FT_MATCH_EXACT: |
| return filter_pattern == value; |
| case pegasus::pegasus_client::FT_MATCH_ANYWHERE: |
| case pegasus::pegasus_client::FT_MATCH_PREFIX: |
| case pegasus::pegasus_client::FT_MATCH_POSTFIX: { |
| if (filter_pattern.length() == 0) |
| return true; |
| if (value.length() < filter_pattern.length()) |
| return false; |
| if (filter_type == pegasus::pegasus_client::FT_MATCH_ANYWHERE) { |
| return dsn::string_view(value).find(filter_pattern) != dsn::string_view::npos; |
| } else if (filter_type == pegasus::pegasus_client::FT_MATCH_PREFIX) { |
| return ::memcmp(value.data(), filter_pattern.data(), filter_pattern.length()) == 0; |
| } else { // filter_type == pegasus::pegasus_client::FT_MATCH_POSTFIX |
| return ::memcmp(value.data() + value.length() - filter_pattern.length(), |
| filter_pattern.data(), |
| filter_pattern.length()) == 0; |
| } |
| } |
| default: |
| dassert(false, "unsupported filter type: %d", filter_type); |
| } |
| return false; |
| } |
| // return true if the data is valid for the filter |
| inline bool |
| validate_filter(scan_data_context *context, const std::string &sort_key, const std::string &value) |
| { |
| // for sort key, we only need to check MATCH_EXACT, because it is not supported |
| // on the server side, but MATCH_PREFIX is already satisified. |
| if (context->sort_key_filter_type == pegasus::pegasus_client::FT_MATCH_EXACT && |
| sort_key.length() > context->sort_key_filter_pattern.length()) |
| return false; |
| return validate_filter(context->value_filter_type, context->value_filter_pattern, value); |
| } |
| inline void scan_data_next(scan_data_context *context) |
| { |
| while (!context->split_completed.load() && !context->error_occurred->load() && |
| context->split_request_count.load() < context->max_batch_count) { |
| context->split_request_count++; |
| context->scanner->async_next([context](int ret, |
| std::string &&hash_key, |
| std::string &&sort_key, |
| std::string &&value, |
| pegasus::pegasus_client::internal_info &&info) { |
| if (ret == pegasus::PERR_OK) { |
| if (validate_filter(context, sort_key, value)) { |
| switch (context->op) { |
| case SCAN_COPY: |
| context->split_request_count++; |
| if (context->no_overwrite) { |
| auto callback = [context]( |
| int err, |
| pegasus::pegasus_client::check_and_set_results &&results, |
| pegasus::pegasus_client::internal_info &&info) { |
| if (err != pegasus::PERR_OK) { |
| if (!context->split_completed.exchange(true)) { |
| fprintf(stderr, |
| "ERROR: split[%d] async check and set failed: %s\n", |
| context->split_id, |
| context->client->get_error_string(err)); |
| context->error_occurred->store(true); |
| } |
| } else { |
| if (results.set_succeed) { |
| context->split_rows++; |
| } |
| scan_data_next(context); |
| } |
| // should put "split_request_count--" at end of the scope, |
| // to prevent that split_request_count becomes 0 in the middle. |
| context->split_request_count--; |
| }; |
| pegasus::pegasus_client::check_and_set_options options; |
| context->client->async_check_and_set( |
| hash_key, |
| sort_key, |
| pegasus::pegasus_client::cas_check_type::CT_VALUE_NOT_EXIST, |
| "", |
| sort_key, |
| value, |
| options, |
| std::move(callback), |
| context->timeout_ms); |
| } else { |
| auto callback = |
| [context](int err, pegasus::pegasus_client::internal_info &&info) { |
| if (err != pegasus::PERR_OK) { |
| if (!context->split_completed.exchange(true)) { |
| fprintf(stderr, |
| "ERROR: split[%d] async set failed: %s\n", |
| context->split_id, |
| context->client->get_error_string(err)); |
| context->error_occurred->store(true); |
| } |
| } else { |
| context->split_rows++; |
| scan_data_next(context); |
| } |
| // should put "split_request_count--" at end of the scope, |
| // to prevent that split_request_count becomes 0 in the middle. |
| context->split_request_count--; |
| }; |
| context->client->async_set(hash_key, |
| sort_key, |
| value, |
| std::move(callback), |
| context->timeout_ms); |
| } |
| break; |
| case SCAN_CLEAR: |
| context->split_request_count++; |
| context->client->async_del( |
| hash_key, |
| sort_key, |
| [context](int err, pegasus::pegasus_client::internal_info &&info) { |
| if (err != pegasus::PERR_OK) { |
| if (!context->split_completed.exchange(true)) { |
| fprintf(stderr, |
| "ERROR: split[%d] async del failed: %s\n", |
| context->split_id, |
| context->client->get_error_string(err)); |
| context->error_occurred->store(true); |
| } |
| } else { |
| context->split_rows++; |
| scan_data_next(context); |
| } |
| // should put "split_request_count--" at end of the scope, |
| // to prevent that split_request_count becomes 0 in the middle. |
| context->split_request_count--; |
| }, |
| context->timeout_ms); |
| break; |
| case SCAN_COUNT: |
| context->split_rows++; |
| if (context->stat_size && context->statistics) { |
| long hash_key_size = hash_key.size(); |
| context->statistics->measureTime( |
| static_cast<uint32_t>(histogram_type::HASH_KEY_SIZE), |
| hash_key_size); |
| |
| long sort_key_size = sort_key.size(); |
| context->statistics->measureTime( |
| static_cast<uint32_t>(histogram_type::SORT_KEY_SIZE), |
| sort_key_size); |
| |
| long value_size = value.size(); |
| context->statistics->measureTime( |
| static_cast<uint32_t>(histogram_type::VALUE_SIZE), value_size); |
| |
| long row_size = hash_key_size + sort_key_size + value_size; |
| context->statistics->measureTime( |
| static_cast<uint32_t>(histogram_type::ROW_SIZE), row_size); |
| |
| if (context->top_count > 0) { |
| context->top_rows.push( |
| std::move(hash_key), std::move(sort_key), row_size); |
| } |
| } |
| if (context->count_hash_key) { |
| if (hash_key != context->last_hash_key) { |
| context->split_hash_key_count++; |
| context->last_hash_key = std::move(hash_key); |
| } |
| } |
| scan_data_next(context); |
| break; |
| case SCAN_GEN_GEO: |
| context->split_request_count++; |
| context->geoclient->async_set( |
| hash_key, |
| sort_key, |
| value, |
| [context](int err, pegasus::pegasus_client::internal_info &&info) { |
| if (err != pegasus::PERR_OK) { |
| if (!context->split_completed.exchange(true)) { |
| fprintf(stderr, |
| "ERROR: split[%d] async set failed: %s\n", |
| context->split_id, |
| context->client->get_error_string(err)); |
| context->error_occurred->store(true); |
| } |
| } else { |
| context->split_rows++; |
| scan_data_next(context); |
| } |
| // should put "split_request_count--" at end of the scope, |
| // to prevent that split_request_count becomes 0 in the middle. |
| context->split_request_count--; |
| }, |
| context->timeout_ms); |
| break; |
| default: |
| dassert(false, "op = %d", context->op); |
| break; |
| } |
| } else { |
| scan_data_next(context); |
| } |
| } else if (ret == pegasus::PERR_SCAN_COMPLETE) { |
| context->split_completed.store(true); |
| } else { |
| if (!context->split_completed.exchange(true)) { |
| fprintf(stderr, |
| "ERROR: split[%d] scan next failed: %s\n", |
| context->split_id, |
| context->client->get_error_string(ret)); |
| context->error_occurred->store(true); |
| } |
| } |
| // should put "split_request_count--" at end of the scope, |
| // to prevent that split_request_count becomes 0 in the middle. |
| context->split_request_count--; |
| }); |
| |
| if (context->count_hash_key) { |
| // disable parallel scan if count_hash_key == true |
| break; |
| } |
| } |
| } |
| |
| struct node_desc |
| { |
| std::string desc; |
| dsn::rpc_address address; |
| node_desc(const std::string &s, const dsn::rpc_address &n) : desc(s), address(n) {} |
| }; |
| // type: all | replica-server | meta-server |
| inline bool fill_nodes(shell_context *sc, const std::string &type, std::vector<node_desc> &nodes) |
| { |
| if (type == "all" || type == "meta-server") { |
| for (auto &addr : sc->meta_list) { |
| nodes.emplace_back("meta-server", addr); |
| } |
| } |
| |
| if (type == "all" || type == "replica-server") { |
| std::map<dsn::rpc_address, dsn::replication::node_status::type> rs_nodes; |
| ::dsn::error_code err = |
| sc->ddl_client->list_nodes(dsn::replication::node_status::NS_ALIVE, rs_nodes); |
| if (err != ::dsn::ERR_OK) { |
| fprintf(stderr, "ERROR: list node failed: %s\n", err.to_string()); |
| return false; |
| } |
| for (auto &kv : rs_nodes) { |
| nodes.emplace_back("replica-server", kv.first); |
| } |
| } |
| |
| return true; |
| } |
| |
| inline std::vector<std::pair<bool, std::string>> |
| call_remote_command(shell_context *sc, |
| const std::vector<node_desc> &nodes, |
| const std::string &cmd, |
| const std::vector<std::string> &arguments) |
| { |
| std::vector<std::pair<bool, std::string>> results; |
| std::vector<dsn::task_ptr> tasks; |
| tasks.resize(nodes.size()); |
| results.resize(nodes.size()); |
| for (int i = 0; i < nodes.size(); ++i) { |
| auto callback = [&results, i](::dsn::error_code err, const std::string &resp) { |
| if (err == ::dsn::ERR_OK) { |
| results[i].first = true; |
| results[i].second = resp; |
| } else { |
| results[i].first = false; |
| results[i].second = err.to_string(); |
| } |
| }; |
| tasks[i] = dsn::dist::cmd::async_call_remote( |
| nodes[i].address, cmd, arguments, callback, std::chrono::milliseconds(5000)); |
| } |
| for (int i = 0; i < nodes.size(); ++i) { |
| tasks[i]->wait(); |
| } |
| return results; |
| } |
| |
| inline bool parse_app_pegasus_perf_counter_name(const std::string &name, |
| int32_t &app_id, |
| int32_t &partition_index, |
| std::string &counter_name) |
| { |
| std::string::size_type find = name.find_last_of('@'); |
| if (find == std::string::npos) |
| return false; |
| int n = sscanf(name.c_str() + find + 1, "%d.%d", &app_id, &partition_index); |
| if (n != 2) |
| return false; |
| std::string::size_type find2 = name.find_last_of('*'); |
| if (find2 == std::string::npos) |
| return false; |
| counter_name = name.substr(find2 + 1, find - find2 - 1); |
| return true; |
| } |
| |
| inline bool parse_app_perf_counter_name(const std::string &name, |
| std::string &app_name, |
| std::string &counter_name) |
| { |
| /** |
| * name format: |
| * 1.{node}*{section}*{counter_name}@{app_name}.{percent_line} |
| * 2.{node}*{section}*{counter_name}@{app_name} |
| */ |
| std::string::size_type find = name.find_last_of('@'); |
| if (find == std::string::npos) |
| return false; |
| |
| std::string::size_type find2 = name.find_last_of('.'); |
| if (find2 == std::string::npos) { |
| app_name = name.substr(find + 1); |
| } else { |
| app_name = name.substr(find + 1, find2 - find - 1); |
| } |
| |
| std::string::size_type find3 = name.find_last_of('*'); |
| if (find3 == std::string::npos) |
| return false; |
| counter_name = name.substr(find3 + 1, find - find3 - 1); |
| return true; |
| } |
| |
| struct row_data |
| { |
| double get_total_qps() const |
| { |
| return get_qps + multi_get_qps + scan_qps + put_qps + multi_put_qps + remove_qps + |
| multi_remove_qps + incr_qps + check_and_set_qps + check_and_mutate_qps + |
| duplicate_qps; |
| } |
| |
| double get_total_cu() const { return recent_read_cu + recent_write_cu; } |
| |
| double get_total_read_qps() const { return get_qps + multi_get_qps + scan_qps; } |
| |
| double get_total_write_qps() const |
| { |
| return put_qps + remove_qps + multi_put_qps + multi_remove_qps + check_and_set_qps + |
| check_and_mutate_qps; |
| } |
| |
| std::string row_name; |
| int32_t app_id = 0; |
| int32_t partition_count = 0; |
| double get_qps = 0; |
| double multi_get_qps = 0; |
| double put_qps = 0; |
| double multi_put_qps = 0; |
| double remove_qps = 0; |
| double multi_remove_qps = 0; |
| double incr_qps = 0; |
| double check_and_set_qps = 0; |
| double check_and_mutate_qps = 0; |
| double scan_qps = 0; |
| double duplicate_qps = 0; |
| double dup_shipped_ops = 0; |
| double dup_failed_shipping_ops = 0; |
| double recent_read_cu = 0; |
| double recent_write_cu = 0; |
| double recent_expire_count = 0; |
| double recent_filter_count = 0; |
| double recent_abnormal_count = 0; |
| double recent_write_throttling_delay_count = 0; |
| double recent_write_throttling_reject_count = 0; |
| double storage_mb = 0; |
| double storage_count = 0; |
| double rdb_block_cache_hit_count = 0; |
| double rdb_block_cache_total_count = 0; |
| double rdb_index_and_filter_blocks_mem_usage = 0; |
| double rdb_memtable_mem_usage = 0; |
| double rdb_estimate_num_keys = 0; |
| double rdb_bf_seek_negatives = 0; |
| double rdb_bf_seek_total = 0; |
| double rdb_bf_point_positive_true = 0; |
| double rdb_bf_point_positive_total = 0; |
| double rdb_bf_point_negatives = 0; |
| double backup_request_qps = 0; |
| double get_bytes = 0; |
| double multi_get_bytes = 0; |
| double scan_bytes = 0; |
| double put_bytes = 0; |
| double multi_put_bytes = 0; |
| double check_and_set_bytes = 0; |
| double check_and_mutate_bytes = 0; |
| }; |
| |
| inline bool |
| update_app_pegasus_perf_counter(row_data &row, const std::string &counter_name, double value) |
| { |
| if (counter_name == "get_qps") |
| row.get_qps += value; |
| else if (counter_name == "multi_get_qps") |
| row.multi_get_qps += value; |
| else if (counter_name == "put_qps") |
| row.put_qps += value; |
| else if (counter_name == "multi_put_qps") |
| row.multi_put_qps += value; |
| else if (counter_name == "remove_qps") |
| row.remove_qps += value; |
| else if (counter_name == "multi_remove_qps") |
| row.multi_remove_qps += value; |
| else if (counter_name == "incr_qps") |
| row.incr_qps += value; |
| else if (counter_name == "check_and_set_qps") |
| row.check_and_set_qps += value; |
| else if (counter_name == "check_and_mutate_qps") |
| row.check_and_mutate_qps += value; |
| else if (counter_name == "scan_qps") |
| row.scan_qps += value; |
| else if (counter_name == "duplicate_qps") |
| row.duplicate_qps += value; |
| else if (counter_name == "dup_shipped_ops") |
| row.dup_shipped_ops += value; |
| else if (counter_name == "dup_failed_shipping_ops") |
| row.dup_failed_shipping_ops += value; |
| else if (counter_name == "recent.read.cu") |
| row.recent_read_cu += value; |
| else if (counter_name == "recent.write.cu") |
| row.recent_write_cu += value; |
| else if (counter_name == "recent.expire.count") |
| row.recent_expire_count += value; |
| else if (counter_name == "recent.filter.count") |
| row.recent_filter_count += value; |
| else if (counter_name == "recent.abnormal.count") |
| row.recent_abnormal_count += value; |
| else if (counter_name == "recent.write.throttling.delay.count") |
| row.recent_write_throttling_delay_count += value; |
| else if (counter_name == "recent.write.throttling.reject.count") |
| row.recent_write_throttling_reject_count += value; |
| else if (counter_name == "disk.storage.sst(MB)") |
| row.storage_mb += value; |
| else if (counter_name == "disk.storage.sst.count") |
| row.storage_count += value; |
| else if (counter_name == "rdb.block_cache.hit_count") |
| row.rdb_block_cache_hit_count += value; |
| else if (counter_name == "rdb.block_cache.total_count") |
| row.rdb_block_cache_total_count += value; |
| else if (counter_name == "rdb.index_and_filter_blocks.memory_usage") |
| row.rdb_index_and_filter_blocks_mem_usage += value; |
| else if (counter_name == "rdb.memtable.memory_usage") |
| row.rdb_memtable_mem_usage += value; |
| else if (counter_name == "rdb.estimate_num_keys") |
| row.rdb_estimate_num_keys += value; |
| else if (counter_name == "rdb.bf_seek_negatives") |
| row.rdb_bf_seek_negatives += value; |
| else if (counter_name == "rdb.bf_seek_total") |
| row.rdb_bf_seek_total += value; |
| else if (counter_name == "rdb.bf_point_positive_true") |
| row.rdb_bf_point_positive_true += value; |
| else if (counter_name == "rdb.bf_point_positive_total") |
| row.rdb_bf_point_positive_total += value; |
| else if (counter_name == "rdb.bf_point_negatives") |
| row.rdb_bf_point_negatives += value; |
| else if (counter_name == "backup_request_qps") |
| row.backup_request_qps += value; |
| else if (counter_name == "get_bytes") |
| row.get_bytes += value; |
| else if (counter_name == "multi_get_bytes") |
| row.multi_get_bytes += value; |
| else if (counter_name == "scan_bytes") |
| row.scan_bytes += value; |
| else if (counter_name == "put_bytes") |
| row.put_bytes += value; |
| else if (counter_name == "multi_put_bytes") |
| row.multi_put_bytes += value; |
| else if (counter_name == "check_and_set_bytes") |
| row.check_and_set_bytes += value; |
| else if (counter_name == "check_and_mutate_bytes") |
| row.check_and_mutate_bytes += value; |
| else |
| return false; |
| return true; |
| } |
| |
| inline bool get_apps_and_nodes(shell_context *sc, |
| std::vector<::dsn::app_info> &apps, |
| std::vector<node_desc> &nodes) |
| { |
| dsn::error_code err = sc->ddl_client->list_apps(dsn::app_status::AS_AVAILABLE, apps); |
| if (err != dsn::ERR_OK) { |
| derror("list apps failed, error = %s", err.to_string()); |
| return false; |
| } |
| if (!fill_nodes(sc, "replica-server", nodes)) { |
| derror("get replica server node list failed"); |
| return false; |
| } |
| return true; |
| } |
| |
| inline bool |
| get_app_partitions(shell_context *sc, |
| const std::vector<::dsn::app_info> &apps, |
| std::map<int32_t, std::vector<dsn::partition_configuration>> &app_partitions) |
| { |
| for (const ::dsn::app_info &app : apps) { |
| int32_t app_id = 0; |
| int32_t partition_count = 0; |
| dsn::error_code err = sc->ddl_client->list_app( |
| app.app_name, app_id, partition_count, app_partitions[app.app_id]); |
| if (err != ::dsn::ERR_OK) { |
| derror("list app %s failed, error = %s", app.app_name.c_str(), err.to_string()); |
| return false; |
| } |
| dassert(app_id == app.app_id, "%d VS %d", app_id, app.app_id); |
| dassert(partition_count == app.partition_count, |
| "%d VS %d", |
| partition_count, |
| app.partition_count); |
| } |
| return true; |
| } |
| |
| inline bool decode_node_perf_counter_info(const dsn::rpc_address &node_addr, |
| const std::pair<bool, std::string> &result, |
| dsn::perf_counter_info &info) |
| { |
| if (!result.first) { |
| derror("query perf counter info from node %s failed", node_addr.to_string()); |
| return false; |
| } |
| dsn::blob bb(result.second.data(), 0, result.second.size()); |
| if (!dsn::json::json_forwarder<dsn::perf_counter_info>::decode(bb, info)) { |
| derror("decode perf counter info from node %s failed, result = %s", |
| node_addr.to_string(), |
| result.second.c_str()); |
| return false; |
| } |
| if (info.result != "OK") { |
| derror("query perf counter info from node %s returns error, error = %s", |
| node_addr.to_string(), |
| info.result.c_str()); |
| return false; |
| } |
| return true; |
| } |
| |
| // rows: key-app name, value-perf counters for each partition |
| inline bool get_app_partition_stat(shell_context *sc, |
| std::map<std::string, std::vector<row_data>> &rows) |
| { |
| // get apps and nodes |
| std::vector<::dsn::app_info> apps; |
| std::vector<node_desc> nodes; |
| if (!get_apps_and_nodes(sc, apps, nodes)) { |
| return false; |
| } |
| |
| // get the relationship between app_id and app_name |
| std::map<int32_t, std::string> app_id_name; |
| std::map<std::string, int32_t> app_name_id; |
| for (::dsn::app_info &app : apps) { |
| app_id_name[app.app_id] = app.app_name; |
| app_name_id[app.app_name] = app.app_id; |
| rows[app.app_name].resize(app.partition_count); |
| } |
| |
| // get app_id --> partitions |
| std::map<int32_t, std::vector<dsn::partition_configuration>> app_partitions; |
| if (!get_app_partitions(sc, apps, app_partitions)) { |
| return false; |
| } |
| |
| // get all of the perf counters with format ".*@.*" |
| std::vector<std::pair<bool, std::string>> results = |
| call_remote_command(sc, nodes, "perf-counters", {".*@.*"}); |
| |
| for (int i = 0; i < nodes.size(); ++i) { |
| // decode info of perf-counters on node i |
| dsn::perf_counter_info info; |
| if (!decode_node_perf_counter_info(nodes[i].address, results[i], info)) { |
| return false; |
| } |
| |
| for (dsn::perf_counter_metric &m : info.counters) { |
| // get app_id/partition_id/counter_name/app_name from the name of perf-counter |
| int32_t app_id_x, partition_index_x; |
| std::string counter_name; |
| std::string app_name; |
| |
| if (parse_app_pegasus_perf_counter_name( |
| m.name, app_id_x, partition_index_x, counter_name)) { |
| // only primary partition will be counted |
| auto find = app_partitions.find(app_id_x); |
| if (find != app_partitions.end() && |
| find->second[partition_index_x].primary == nodes[i].address) { |
| row_data &row = rows[app_id_name[app_id_x]][partition_index_x]; |
| row.row_name = std::to_string(partition_index_x); |
| row.app_id = app_id_x; |
| update_app_pegasus_perf_counter(row, counter_name, m.value); |
| } |
| } else if (parse_app_perf_counter_name(m.name, app_name, counter_name)) { |
| // if the app_name from perf-counter isn't existed(maybe the app was dropped), it |
| // will be ignored. |
| if (app_name_id.find(app_name) == app_name_id.end()) { |
| continue; |
| } |
| // perf-counter value will be set into partition index 0. |
| row_data &row = rows[app_name][0]; |
| row.app_id = app_name_id[app_name]; |
| update_app_pegasus_perf_counter(row, counter_name, m.value); |
| } |
| } |
| } |
| return true; |
| } |
| |
| inline bool |
| get_app_stat(shell_context *sc, const std::string &app_name, std::vector<row_data> &rows) |
| { |
| std::vector<::dsn::app_info> apps; |
| std::vector<node_desc> nodes; |
| if (!get_apps_and_nodes(sc, apps, nodes)) |
| return false; |
| |
| ::dsn::app_info *app_info = nullptr; |
| if (!app_name.empty()) { |
| for (auto &app : apps) { |
| if (app.app_name == app_name) { |
| app_info = &app; |
| break; |
| } |
| } |
| if (app_info == nullptr) { |
| derror("app %s not found", app_name.c_str()); |
| return false; |
| } |
| } |
| |
| std::vector<std::string> arguments; |
| char tmp[256]; |
| if (app_name.empty()) { |
| sprintf(tmp, ".*@.*"); |
| } else { |
| sprintf(tmp, ".*@%d\\..*", app_info->app_id); |
| } |
| arguments.emplace_back(tmp); |
| std::vector<std::pair<bool, std::string>> results = |
| call_remote_command(sc, nodes, "perf-counters", arguments); |
| |
| if (app_name.empty()) { |
| std::map<int32_t, std::vector<dsn::partition_configuration>> app_partitions; |
| if (!get_app_partitions(sc, apps, app_partitions)) |
| return false; |
| |
| rows.resize(app_partitions.size()); |
| int idx = 0; |
| std::map<int32_t, int> app_row_idx; // app_id --> row_idx |
| for (::dsn::app_info &app : apps) { |
| rows[idx].row_name = app.app_name; |
| rows[idx].app_id = app.app_id; |
| rows[idx].partition_count = app.partition_count; |
| app_row_idx[app.app_id] = idx; |
| idx++; |
| } |
| |
| for (int i = 0; i < nodes.size(); ++i) { |
| dsn::rpc_address node_addr = nodes[i].address; |
| dsn::perf_counter_info info; |
| if (!decode_node_perf_counter_info(node_addr, results[i], info)) |
| return false; |
| for (dsn::perf_counter_metric &m : info.counters) { |
| int32_t app_id_x, partition_index_x; |
| std::string counter_name; |
| if (!parse_app_pegasus_perf_counter_name( |
| m.name, app_id_x, partition_index_x, counter_name)) { |
| continue; |
| } |
| auto find = app_partitions.find(app_id_x); |
| if (find == app_partitions.end()) |
| continue; |
| dsn::partition_configuration &pc = find->second[partition_index_x]; |
| if (pc.primary != node_addr) |
| continue; |
| update_app_pegasus_perf_counter(rows[app_row_idx[app_id_x]], counter_name, m.value); |
| } |
| } |
| } else { |
| rows.resize(app_info->partition_count); |
| for (int i = 0; i < app_info->partition_count; i++) |
| rows[i].row_name = std::to_string(i); |
| int32_t app_id = 0; |
| int32_t partition_count = 0; |
| std::vector<dsn::partition_configuration> partitions; |
| dsn::error_code err = |
| sc->ddl_client->list_app(app_name, app_id, partition_count, partitions); |
| if (err != ::dsn::ERR_OK) { |
| derror("list app %s failed, error = %s", app_name.c_str(), err.to_string()); |
| return false; |
| } |
| dassert(app_id == app_info->app_id, "%d VS %d", app_id, app_info->app_id); |
| dassert(partition_count == app_info->partition_count, |
| "%d VS %d", |
| partition_count, |
| app_info->partition_count); |
| |
| for (int i = 0; i < nodes.size(); ++i) { |
| dsn::rpc_address node_addr = nodes[i].address; |
| dsn::perf_counter_info info; |
| if (!decode_node_perf_counter_info(node_addr, results[i], info)) |
| return false; |
| for (dsn::perf_counter_metric &m : info.counters) { |
| int32_t app_id_x, partition_index_x; |
| std::string counter_name; |
| bool parse_ret = parse_app_pegasus_perf_counter_name( |
| m.name, app_id_x, partition_index_x, counter_name); |
| dassert(parse_ret, "name = %s", m.name.c_str()); |
| dassert(app_id_x == app_id, "name = %s", m.name.c_str()); |
| dassert(partition_index_x < partition_count, "name = %s", m.name.c_str()); |
| if (partitions[partition_index_x].primary != node_addr) |
| continue; |
| update_app_pegasus_perf_counter(rows[partition_index_x], counter_name, m.value); |
| } |
| } |
| } |
| return true; |
| } |
| |
| struct node_capacity_unit_stat |
| { |
| // timestamp when node perf_counter_info has updated. |
| std::string timestamp; |
| std::string node_address; |
| // mapping: app_id --> (read_cu, write_cu) |
| std::map<int32_t, std::pair<int64_t, int64_t>> cu_value_by_app; |
| |
| std::string dump_to_json() const |
| { |
| std::map<int32_t, std::vector<int64_t>> values; |
| for (auto &kv : cu_value_by_app) { |
| auto &pair = kv.second; |
| if (pair.first != 0 || pair.second != 0) |
| values.emplace(kv.first, std::vector<int64_t>{pair.first, pair.second}); |
| } |
| std::stringstream out; |
| rapidjson::OStreamWrapper wrapper(out); |
| dsn::json::JsonWriter writer(wrapper); |
| dsn::json::json_encode(writer, values); |
| return out.str(); |
| } |
| }; |
| |
| inline bool get_capacity_unit_stat(shell_context *sc, |
| std::vector<node_capacity_unit_stat> &nodes_stat) |
| { |
| std::vector<node_desc> nodes; |
| if (!fill_nodes(sc, "replica-server", nodes)) { |
| derror("get replica server node list failed"); |
| return false; |
| } |
| |
| std::vector<std::pair<bool, std::string>> results = |
| call_remote_command(sc, nodes, "perf-counters-by-substr", {".cu@"}); |
| |
| nodes_stat.resize(nodes.size()); |
| for (int i = 0; i < nodes.size(); ++i) { |
| dsn::rpc_address node_addr = nodes[i].address; |
| dsn::perf_counter_info info; |
| if (!decode_node_perf_counter_info(node_addr, results[i], info)) { |
| dwarn("decode perf counter from node(%s) failed, just ignore it", |
| node_addr.to_string()); |
| continue; |
| } |
| nodes_stat[i].timestamp = info.timestamp_str; |
| nodes_stat[i].node_address = node_addr.to_string(); |
| for (dsn::perf_counter_metric &m : info.counters) { |
| int32_t app_id, pidx; |
| std::string counter_name; |
| bool r = parse_app_pegasus_perf_counter_name(m.name, app_id, pidx, counter_name); |
| dassert(r, "name = %s", m.name.c_str()); |
| if (counter_name == "recent.read.cu") { |
| nodes_stat[i].cu_value_by_app[app_id].first += (int64_t)m.value; |
| } else if (counter_name == "recent.write.cu") { |
| nodes_stat[i].cu_value_by_app[app_id].second += (int64_t)m.value; |
| } |
| } |
| } |
| return true; |
| } |
| |
| struct app_storage_size_stat |
| { |
| // timestamp when this stat is generated. |
| std::string timestamp; |
| // mapping: app_id --> [app_partition_count, stat_partition_count, storage_size_in_mb] |
| std::map<int32_t, std::vector<int64_t>> st_value_by_app; |
| |
| std::string dump_to_json() const |
| { |
| std::stringstream out; |
| rapidjson::OStreamWrapper wrapper(out); |
| dsn::json::JsonWriter writer(wrapper); |
| dsn::json::json_encode(writer, st_value_by_app); |
| return out.str(); |
| } |
| }; |
| |
| inline bool get_storage_size_stat(shell_context *sc, app_storage_size_stat &st_stat) |
| { |
| std::vector<::dsn::app_info> apps; |
| std::vector<node_desc> nodes; |
| if (!get_apps_and_nodes(sc, apps, nodes)) { |
| derror("get apps and nodes failed"); |
| return false; |
| } |
| |
| std::map<int32_t, std::vector<dsn::partition_configuration>> app_partitions; |
| if (!get_app_partitions(sc, apps, app_partitions)) { |
| derror("get app partitions failed"); |
| return false; |
| } |
| for (auto &kv : app_partitions) { |
| auto &v = kv.second; |
| for (auto &c : v) { |
| // use partition_flags to record if this partition's storage size is calculated, |
| // because `app_partitions' is a temporary variable, so we can re-use partition_flags. |
| c.partition_flags = 0; |
| } |
| } |
| |
| std::vector<std::pair<bool, std::string>> results = call_remote_command( |
| sc, nodes, "perf-counters-by-prefix", {"replica*app.pegasus*disk.storage.sst(MB)"}); |
| |
| for (int i = 0; i < nodes.size(); ++i) { |
| dsn::rpc_address node_addr = nodes[i].address; |
| dsn::perf_counter_info info; |
| if (!decode_node_perf_counter_info(node_addr, results[i], info)) { |
| dwarn("decode perf counter from node(%s) failed, just ignore it", |
| node_addr.to_string()); |
| continue; |
| } |
| for (dsn::perf_counter_metric &m : info.counters) { |
| int32_t app_id_x, partition_index_x; |
| std::string counter_name; |
| bool parse_ret = parse_app_pegasus_perf_counter_name( |
| m.name, app_id_x, partition_index_x, counter_name); |
| dassert(parse_ret, "name = %s", m.name.c_str()); |
| if (counter_name != "disk.storage.sst(MB)") |
| continue; |
| auto find = app_partitions.find(app_id_x); |
| if (find == app_partitions.end()) // app id not found |
| continue; |
| dsn::partition_configuration &pc = find->second[partition_index_x]; |
| if (pc.primary != node_addr) // not primary replica |
| continue; |
| if (pc.partition_flags != 0) // already calculated |
| continue; |
| pc.partition_flags = 1; |
| int64_t app_partition_count = find->second.size(); |
| auto st_it = st_stat.st_value_by_app |
| .emplace(app_id_x, std::vector<int64_t>{app_partition_count, 0, 0}) |
| .first; |
| st_it->second[1]++; // stat_partition_count |
| st_it->second[2] += m.value; // storage_size_in_mb |
| } |
| } |
| |
| char buf[20]; |
| dsn::utils::time_ms_to_date_time(dsn_now_ms(), buf, sizeof(buf)); |
| st_stat.timestamp = buf; |
| return true; |
| } |
| |
| inline configuration_proposal_action new_proposal_action(const dsn::rpc_address &target, |
| const dsn::rpc_address &node, |
| config_type::type type) |
| { |
| configuration_proposal_action act; |
| act.__set_target(target); |
| act.__set_node(node); |
| act.__set_type(type); |
| return act; |
| } |
