This is an automated email from the ASF dual-hosted git repository.
wangdan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pegasus.git
The following commit(s) were added to refs/heads/master by this push:
new 2432972b2 feat(meta): list tables with specified name pattern (#2172)
2432972b2 is described below
commit 2432972b2ba862097f2a93f34f02a4560f4a6250
Author: Dan Wang <[email protected]>
AuthorDate: Wed Jan 15 12:07:11 2025 +0800
feat(meta): list tables with specified name pattern (#2172)
Once there are a great number of tables on a cluster(say several
thousands), the
whole result of executing `ls` by Pegasus shell cannot even be seen on the
terminal.
If we want to see one or multiple tables with some specified pattern which
could not
be shown on the terminal by `ls`, the only way is to use bash commands or
even script
to list the target tables.
We would allow specifying name pattern for `ls` command, to show just one
or several
tables according to the pattern instead of all. This is useful while we
want to find just
several tables with obvious characteristic(e.g. the same prefix) among
hundreds or even
thousands of tables on the cluster.
---
idl/dsn.layer2.thrift | 4 +-
idl/duplication.thrift | 2 +-
idl/meta_admin.thrift | 14 +-
src/client/replication_ddl_client.cpp | 295 +++++++---------------
src/client/replication_ddl_client.h | 136 ++++++++--
src/client/test/ddl_client_test.cpp | 7 +-
src/meta/duplication/meta_duplication_service.cpp | 20 +-
src/meta/meta_http_service.cpp | 72 +++---
src/meta/server_state.cpp | 46 +++-
src/server/hotspot_partition_calculator.h | 3 +-
src/server/info_collector.cpp | 5 +-
src/server/pegasus_mutation_duplicator.cpp | 2 +
src/shell/command_helper.h | 28 +-
src/shell/command_utils.cpp | 1 +
src/shell/command_utils.h | 11 -
src/shell/commands/node_management.cpp | 53 ++--
src/shell/commands/rebalance.cpp | 22 +-
src/shell/commands/table_management.cpp | 100 ++++----
src/shell/main.cpp | 13 +-
src/utils/errors.h | 45 +++-
src/utils/output_utils.h | 16 +-
src/utils/strings.cpp | 20 +-
src/utils/strings.h | 13 +-
src/utils/test/utils.cpp | 3 +-
24 files changed, 493 insertions(+), 438 deletions(-)
diff --git a/idl/dsn.layer2.thrift b/idl/dsn.layer2.thrift
index 762a564e7..47d736cb8 100644
--- a/idl/dsn.layer2.thrift
+++ b/idl/dsn.layer2.thrift
@@ -76,9 +76,9 @@ enum app_status
AS_INVALID,
AS_AVAILABLE,
AS_CREATING,
- AS_CREATE_FAILED, // depricated
+ AS_CREATE_FAILED, // deprecated
AS_DROPPING,
- AS_DROP_FAILED, // depricated
+ AS_DROP_FAILED, // deprecated
AS_DROPPED,
AS_RECALLING
}
diff --git a/idl/duplication.thrift b/idl/duplication.thrift
index d2460f2fd..7420169b0 100644
--- a/idl/duplication.thrift
+++ b/idl/duplication.thrift
@@ -254,7 +254,7 @@ struct duplication_sync_response
// per-duplication info and progress of each partition for one or multiple
tables.
struct duplication_list_request
{
- // The pattern used to match an app name, whose type is specified by
`match_type`.
+ // The pattern is used to match an app name, whose type is specified by
`match_type`.
1:string app_name_pattern;
2:utils.pattern_match_type match_type;
}
diff --git a/idl/meta_admin.thrift b/idl/meta_admin.thrift
index 6a6af97c7..b5621c30b 100644
--- a/idl/meta_admin.thrift
+++ b/idl/meta_admin.thrift
@@ -31,6 +31,7 @@ include "dsn.layer2.thrift"
include "duplication.thrift"
include "metadata.thrift"
include "partition_split.thrift"
+include "utils.thrift"
namespace cpp dsn.replication
namespace go admin
@@ -196,13 +197,20 @@ struct configuration_recall_app_response
struct configuration_list_apps_request
{
- 1:dsn.layer2.app_status status = app_status.AS_INVALID;
+ 1:dsn.layer2.app_status status = app_status.AS_INVALID;
+
+ // The pattern is used to match an app name, whose type is specified by
`match_type`.
+ 2:optional string app_name_pattern;
+ 3:optional utils.pattern_match_type match_type;
}
struct configuration_list_apps_response
{
- 1:dsn.error_code err;
- 2:list<dsn.layer2.app_info> infos;
+ 1:dsn.error_code err;
+ 2:list<dsn.layer2.app_info> infos;
+
+ // Extra message to describe the error.
+ 3:optional string hint_message;
}
struct query_app_info_request
diff --git a/src/client/replication_ddl_client.cpp
b/src/client/replication_ddl_client.cpp
index 99c8f5293..a5928389a 100644
--- a/src/client/replication_ddl_client.cpp
+++ b/src/client/replication_ddl_client.cpp
@@ -27,10 +27,11 @@
#include "replication_ddl_client.h"
// IWYU pragma: no_include <ext/alloc_traits.h>
-#include <string.h>
#include <algorithm>
#include <cctype>
#include <cstdint>
+#include <cstdio>
+#include <cstring>
#include <fstream>
#include <iomanip>
#include <iostream>
@@ -192,13 +193,12 @@ dsn::error_code replication_ddl_client::create_app(const
std::string &app_name,
bool success_if_exist)
{
if (partition_count < 1) {
- std::cout << "create app " << app_name << " failed: partition_count
should >= 1"
- << std::endl;
+ fmt::println(stderr, "create app {} failed: partition_count should >=
1", app_name);
return ERR_INVALID_PARAMETERS;
}
if (replica_count < 1) {
- std::cout << "create app " << app_name << " failed: replica_count
should >= 1" << std::endl;
+ fmt::println(stderr, "create app {} failed: replica_count should >=
1", app_name);
return ERR_INVALID_PARAMETERS;
}
@@ -215,25 +215,23 @@ dsn::error_code replication_ddl_client::create_app(const
std::string &app_name,
req->options.is_stateful = !is_stateless;
dsn::replication::configuration_create_app_response resp;
- auto resp_task = request_meta_and_wait_response(RPC_CM_CREATE_APP, req,
resp);
-
- if (resp_task->error() != dsn::ERR_OK) {
- std::cout << "create app " << app_name
- << " failed: [create] call server error: " <<
resp_task->error() << std::endl;
- return resp_task->error();
- }
+ RETURN_EC_NOT_OK_MSG(request_meta_and_wait_response(RPC_CM_CREATE_APP,
req, resp),
+ "create app {} failed: [create] call server error",
+ app_name);
if (resp.err != dsn::ERR_OK) {
- std::cout << "create app " << app_name
- << " failed: [create] received server error: " << resp.err
<< std::endl;
+ fmt::println(
+ stderr, "{}: create app {} failed: [create] received server
error", resp.err, app_name);
return resp.err;
}
- std::cout << "create app " << app_name << " succeed, waiting for app
ready" << std::endl;
+ fmt::println("create app {} succeed, waiting for app ready", app_name);
dsn::error_code error = wait_app_ready(app_name, partition_count,
replica_count);
- if (error == dsn::ERR_OK)
- std::cout << app_name << " is ready now!" << std::endl;
+ if (error == dsn::ERR_OK) {
+ fmt::println("{} is ready now!", app_name);
+ }
+
return error;
}
@@ -247,11 +245,7 @@ dsn::error_code replication_ddl_client::drop_app(const
std::string &app_name, in
req->options.__set_reserve_seconds(reserve_seconds);
dsn::replication::configuration_drop_app_response resp;
- auto resp_task = request_meta_and_wait_response(RPC_CM_DROP_APP, req,
resp);
-
- if (resp_task->error() != dsn::ERR_OK) {
- return resp_task->error();
- }
+ RETURN_EC_NOT_OK(request_meta_and_wait_response(RPC_CM_DROP_APP, req,
resp));
if (resp.err != dsn::ERR_OK) {
return resp.err;
@@ -269,11 +263,7 @@ dsn::error_code replication_ddl_client::recall_app(int32_t
app_id, const std::st
req->new_app_name = new_app_name;
dsn::replication::configuration_recall_app_response resp;
- auto resp_task = request_meta_and_wait_response(RPC_CM_RECALL_APP, req,
resp);
-
- if (resp_task->error() != dsn::ERR_OK) {
- return resp_task->error();
- }
+ RETURN_EC_NOT_OK(request_meta_and_wait_response(RPC_CM_RECALL_APP, req,
resp));
if (resp.err != dsn::ERR_OK) {
return resp.err;
@@ -286,63 +276,58 @@ dsn::error_code
replication_ddl_client::recall_app(int32_t app_id, const std::st
resp.info.app_name, resp.info.partition_count,
resp.info.max_replica_count);
}
-dsn::error_code replication_ddl_client::list_apps(const dsn::app_status::type
status,
- std::vector<::dsn::app_info>
&apps)
+error_s replication_ddl_client::list_apps(dsn::app_status::type status,
+ const std::string &app_name_pattern,
+ utils::pattern_match_type::type
match_type,
+ std::vector<::dsn::app_info> &apps)
{
auto req = std::make_shared<configuration_list_apps_request>();
req->status = status;
+ req->__set_app_name_pattern(app_name_pattern);
+ req->__set_match_type(match_type);
- auto resp_task = request_meta(RPC_CM_LIST_APPS, req);
- resp_task->wait();
- if (resp_task->error() != dsn::ERR_OK) {
- return resp_task->error();
+ dsn::replication::configuration_list_apps_response resp;
+ const auto &req_result = request_meta_and_wait_response(RPC_CM_LIST_APPS,
req, resp);
+ if (!req_result) {
+ return req_result;
}
- dsn::replication::configuration_list_apps_response resp;
- ::dsn::unmarshall(resp_task->get_response(), resp);
if (resp.err != dsn::ERR_OK) {
- return resp.err;
+ return error_s::make(resp.err, resp.hint_message);
}
- apps = resp.infos;
+ apps = std::move(resp.infos);
- return dsn::ERR_OK;
+ return error_s::ok();
}
-dsn::error_code replication_ddl_client::list_apps(const dsn::app_status::type
status,
- bool show_all,
- bool detailed,
- bool json,
- const std::string &file_name)
+error_s replication_ddl_client::list_apps(dsn::app_status::type status,
+ std::vector<::dsn::app_info> &apps)
{
- std::vector<::dsn::app_info> apps;
- auto r = list_apps(status, apps);
- if (r != dsn::ERR_OK) {
- return r;
- }
-
- // print configuration_list_apps_response
- std::streambuf *buf;
- std::ofstream of;
+ return list_apps(status, {}, utils::pattern_match_type::PMT_MATCH_ALL,
apps);
+}
- if (!file_name.empty()) {
- of.open(file_name);
- buf = of.rdbuf();
- } else {
- buf = std::cout.rdbuf();
+error_s replication_ddl_client::list_apps(bool detailed,
+ bool json,
+ const std::string &output_file,
+ dsn::app_status::type status,
+ const std::string &app_name_pattern,
+ utils::pattern_match_type::type
match_type)
+{
+ std::vector<::dsn::app_info> apps;
+ {
+ const auto &result = list_apps(status, app_name_pattern, match_type,
apps);
+ if (!result) {
+ return result;
+ }
}
- std::ostream out(buf);
size_t max_app_name_size = 20;
- for (int i = 0; i < apps.size(); i++) {
- dsn::app_info info = apps[i];
- if (!show_all && info.status != app_status::AS_AVAILABLE) {
- continue;
- }
- max_app_name_size = std::max(max_app_name_size, info.app_name.size() +
2);
+ for (const auto &app : apps) {
+ max_app_name_size = std::max(max_app_name_size, app.app_name.size() +
2);
}
- dsn::utils::multi_table_printer mtp;
+ dsn::utils::multi_table_printer multi_printer;
dsn::utils::table_printer tp_general("general_info");
tp_general.add_title("app_id");
tp_general.add_column("status");
@@ -357,11 +342,7 @@ dsn::error_code replication_ddl_client::list_apps(const
dsn::app_status::type st
tp_general.add_column("envs_count");
int available_app_count = 0;
- for (int i = 0; i < apps.size(); i++) {
- dsn::app_info info = apps[i];
- if (!show_all && info.status != app_status::AS_AVAILABLE) {
- continue;
- }
+ for (const auto &info : apps) {
std::string status_str = enum_to_string(info.status);
status_str = status_str.substr(status_str.find("AS_") + 3);
std::string create_time = "-";
@@ -401,7 +382,7 @@ dsn::error_code replication_ddl_client::list_apps(const
dsn::app_status::type st
tp_general.append_data(drop_expire_time);
tp_general.append_data(info.envs.size());
}
- mtp.add(std::move(tp_general));
+ multi_printer.add(std::move(tp_general));
int total_fully_healthy_app_count = 0;
int total_unhealthy_app_count = 0;
@@ -416,17 +397,17 @@ dsn::error_code replication_ddl_client::list_apps(const
dsn::app_status::type st
tp_health.add_column("unhealthy");
tp_health.add_column("write_unhealthy");
tp_health.add_column("read_unhealthy");
- for (auto &info : apps) {
+ for (const auto &info : apps) {
if (info.status != app_status::AS_AVAILABLE) {
continue;
}
- int32_t app_id;
- int32_t partition_count;
+ int32_t app_id = 0;
+ int32_t partition_count = 0;
std::vector<partition_configuration> pcs;
- r = list_app(info.app_name, app_id, partition_count, pcs);
- if (r != dsn::ERR_OK) {
- LOG_ERROR("list app({}) failed, err = {}", info.app_name, r);
- return r;
+ const auto &err = list_app(info.app_name, app_id, partition_count,
pcs);
+ if (err != ERR_OK) {
+ LOG_ERROR("list app({}) failed, err={}", info.app_name, err);
+ return error_s::make(err);
}
CHECK_EQ(info.app_id, app_id);
CHECK_EQ(info.partition_count, partition_count);
@@ -458,16 +439,19 @@ dsn::error_code replication_ddl_client::list_apps(const
dsn::app_status::type st
tp_health.append_data(write_unhealthy);
tp_health.append_data(read_unhealthy);
- if (fully_healthy == info.partition_count)
- total_fully_healthy_app_count++;
- else
- total_unhealthy_app_count++;
- if (write_unhealthy > 0)
- total_write_unhealthy_app_count++;
- if (read_unhealthy > 0)
- total_read_unhealthy_app_count++;
+ if (fully_healthy == info.partition_count) {
+ ++total_fully_healthy_app_count;
+ } else {
+ ++total_unhealthy_app_count;
+ }
+ if (write_unhealthy > 0) {
+ ++total_write_unhealthy_app_count;
+ }
+ if (read_unhealthy > 0) {
+ ++total_read_unhealthy_app_count;
+ }
}
- mtp.add(std::move(tp_health));
+ multi_printer.add(std::move(tp_health));
}
dsn::utils::table_printer tp_count("summary");
@@ -479,16 +463,24 @@ dsn::error_code replication_ddl_client::list_apps(const
dsn::app_status::type st
total_write_unhealthy_app_count);
tp_count.add_row_name_and_data("read_unhealthy_app_count",
total_read_unhealthy_app_count);
}
- mtp.add(std::move(tp_count));
+ multi_printer.add(std::move(tp_count));
- // TODO(wangdan): use dsn::utils::output() in output_utils.h instead.
- mtp.output(out, json ? tp_output_format::kJsonPretty :
tp_output_format::kTabular);
+ dsn::utils::output(output_file, json, multi_printer);
- return dsn::ERR_OK;
+ return error_s::ok();
+}
+
+error_s replication_ddl_client::list_apps(bool detailed,
+ bool json,
+ const std::string &output_file,
+ const dsn::app_status::type status)
+{
+ return list_apps(
+ detailed, json, output_file, status, {},
utils::pattern_match_type::PMT_MATCH_ALL);
}
dsn::error_code replication_ddl_client::list_nodes(
- const dsn::replication::node_status::type status,
+ dsn::replication::node_status::type status,
std::map<dsn::host_port, dsn::replication::node_status::type> &nodes)
{
auto req = std::make_shared<configuration_list_nodes_request>();
@@ -535,104 +527,6 @@ std::string replication_ddl_client::node_name(const
host_port &hp, bool resolve_
return dns_resolver::instance().resolve_address(hp).to_string();
}
-dsn::error_code replication_ddl_client::list_nodes(const
dsn::replication::node_status::type status,
- bool detailed,
- const std::string
&file_name,
- bool resolve_ip)
-{
- std::map<dsn::host_port, dsn::replication::node_status::type> nodes;
- auto r = list_nodes(status, nodes);
- if (r != dsn::ERR_OK) {
- return r;
- }
-
- std::map<dsn::host_port, list_nodes_helper> tmp_map;
- int alive_node_count = 0;
- for (const auto &[hp, type] : nodes) {
- if (type == dsn::replication::node_status::NS_ALIVE) {
- alive_node_count++;
- }
- std::string status_str = enum_to_string(type);
- status_str = status_str.substr(status_str.find("NS_") + 3);
- tmp_map.emplace(hp, list_nodes_helper(node_name(hp, resolve_ip),
status_str));
- }
-
- if (detailed) {
- std::vector<::dsn::app_info> apps;
- r = list_apps(dsn::app_status::AS_AVAILABLE, apps);
- if (r != dsn::ERR_OK) {
- return r;
- }
-
- for (auto &app : apps) {
- int32_t app_id;
- int32_t partition_count;
- std::vector<partition_configuration> pcs;
- r = list_app(app.app_name, app_id, partition_count, pcs);
- if (r != dsn::ERR_OK) {
- return r;
- }
-
- for (const auto &pc : pcs) {
- if (pc.hp_primary) {
- auto find = tmp_map.find(pc.hp_primary);
- if (find != tmp_map.end()) {
- find->second.primary_count++;
- }
- }
- for (const auto &secondary : pc.hp_secondaries) {
- auto find = tmp_map.find(secondary);
- if (find != tmp_map.end()) {
- find->second.secondary_count++;
- }
- }
- }
- }
- }
-
- // print configuration_list_nodes_response
- std::streambuf *buf;
- std::ofstream of;
-
- if (!file_name.empty()) {
- of.open(file_name);
- buf = of.rdbuf();
- } else {
- buf = std::cout.rdbuf();
- }
- std::ostream out(buf);
-
- dsn::utils::table_printer tp;
- tp.add_title("address");
- tp.add_column("status");
- if (detailed) {
- tp.add_column("replica_count");
- tp.add_column("primary_count");
- tp.add_column("secondary_count");
- }
- for (auto &kv : tmp_map) {
- tp.add_row(kv.second.node_name);
- tp.append_data(kv.second.node_status);
- if (detailed) {
- tp.append_data(kv.second.primary_count +
kv.second.secondary_count);
- tp.append_data(kv.second.primary_count);
- tp.append_data(kv.second.secondary_count);
- }
- }
- tp.output(out);
- out << std::endl;
-
- dsn::utils::table_printer tp_count;
- tp_count.add_row_name_and_data("total_node_count", nodes.size());
- tp_count.add_row_name_and_data("alive_node_count", alive_node_count);
- tp_count.add_row_name_and_data("unalive_node_count", nodes.size() -
alive_node_count);
- tp_count.output(out);
- out << std::endl;
-
- return dsn::ERR_OK;
-#undef RESOLVE
-}
-
dsn::error_code replication_ddl_client::cluster_name(int64_t timeout_ms,
std::string &cluster_name)
{
auto req = std::make_shared<configuration_cluster_info_request>();
@@ -1477,17 +1371,22 @@ void replication_ddl_client::end_meta_request(const
rpc_response_task_ptr &callb
dsn::error_code replication_ddl_client::get_app_envs(const std::string
&app_name,
std::map<std::string,
std::string> &envs)
{
+ // Just match the table with the provided name exactly since we want to
get the envs from
+ // a specific table.
std::vector<::dsn::app_info> apps;
- auto r = list_apps(dsn::app_status::AS_AVAILABLE, apps);
- if (r != dsn::ERR_OK) {
- return r;
- }
-
- for (auto &app : apps) {
- if (app.app_name == app_name) {
- envs = app.envs;
- return dsn::ERR_OK;
+ RETURN_EC_NOT_OK(list_apps(
+ dsn::app_status::AS_AVAILABLE, app_name,
utils::pattern_match_type::PMT_MATCH_EXACT, apps));
+
+ for (const auto &app : apps) {
+ // Once the meta server does not support `app_name` and `match_type`
(still the old
+ // version) for `RPC_CM_LIST_APPS`, the response would include all
available tables.
+ // Thus here we should still check if the table name in the response
is the target.
+ if (app.app_name != app_name) {
+ continue;
}
+
+ envs = app.envs;
+ return dsn::ERR_OK;
}
return dsn::ERR_OBJECT_NOT_FOUND;
diff --git a/src/client/replication_ddl_client.h
b/src/client/replication_ddl_client.h
index 4aab2ae7c..0ddd58144 100644
--- a/src/client/replication_ddl_client.h
+++ b/src/client/replication_ddl_client.h
@@ -95,22 +95,49 @@ public:
error_with<configuration_rename_app_response> rename_app(const std::string
&old_app_name,
const std::string
&new_app_name);
- dsn::error_code list_apps(const dsn::app_status::type status,
- bool show_all,
- bool detailed,
- bool json,
- const std::string &file_name);
-
- dsn::error_code list_apps(const dsn::app_status::type status,
- std::vector<::dsn::app_info> &apps);
-
- dsn::error_code list_nodes(const dsn::replication::node_status::type
status,
- bool detailed,
- const std::string &file_name,
- bool resolve_ip = false);
-
+ // Choose tables and list them to a file, with path specified as
`output_file`. Once
+ // `output_file` is empty, tables would be listed to stdout.
+ //
+ // Choose tables according to following parameters:
+ // - `detailed`: whether to show healthy/unhealthy details.
+ // - `json`: whether to output as json format.
+ // - `status`: the status of the tables chosen to be listed.
`app_status::AS_INVALID` means
+ // no restriction.
+ // - `app_name_pattern`: the name pattern of the tables chosen to be
listed.
+ // - `match_type`: the type in which the name pattern would be matched.
+ error_s list_apps(bool detailed,
+ bool json,
+ const std::string &output_file,
+ dsn::app_status::type status,
+ const std::string &app_name_pattern,
+ utils::pattern_match_type::type match_type);
+
+ // The same as the above, except that there's no restriction on table
name; in other
+ // words, the match type is `PMT_MATCH_ALL`.
+ error_s list_apps(bool detailed,
+ bool json,
+ const std::string &output_file,
+ dsn::app_status::type status);
+
+ // Create and send request to meta server to get the tables chosen to be
listed. `status`,
+ // `app_name_pattern` and `match_type` are the same as the above. `apps`
is the output
+ // parameter, which is just the tables chosen to be listed.
+ error_s list_apps(dsn::app_status::type status,
+ const std::string &app_name_pattern,
+ utils::pattern_match_type::type match_type,
+ std::vector<::dsn::app_info> &apps);
+
+ // The same as the above, except that there's no restriction on table
name; in other
+ // words, the match type is `PMT_MATCH_ALL`.
+ error_s list_apps(dsn::app_status::type status,
std::vector<::dsn::app_info> &apps);
+
+ // Create and send request to meta server to get the nodes chosen to be
listed according
+ // to the following parameters:
+ // - `status`: the status of the nodes chosen to be listed.
`node_status::NS_INVALID` means
+ // no restriction.
+ // - `nodes`: the output parameter, which is just the nodes chosen to be
listed.
dsn::error_code
- list_nodes(const dsn::replication::node_status::type status,
+ list_nodes(dsn::replication::node_status::type status,
std::map<dsn::host_port, dsn::replication::node_status::type>
&nodes);
dsn::error_code cluster_name(int64_t timeout_ms, std::string
&cluster_name);
@@ -287,13 +314,18 @@ private:
dsn::message_ex *request,
dsn::message_ex *resp);
+ // Send RPC request `req` with `code` to meta server:
+ // * `timeout_milliseconds`: timeout for the request;
+ // * `reply_thread_hash`: thread hash for the RPC response task.
+ //
+ // Return the RPC response task.
template <typename TRequest>
rpc_response_task_ptr request_meta(const dsn::task_code &code,
std::shared_ptr<TRequest> &req,
- int timeout_milliseconds = 0,
- int reply_thread_hash = 0)
+ int timeout_milliseconds,
+ int reply_thread_hash)
{
- auto msg = dsn::message_ex::create_request(code, timeout_milliseconds);
+ auto *msg = dsn::message_ex::create_request(code,
timeout_milliseconds);
dsn::marshall(msg, *req);
auto task =
@@ -312,28 +344,54 @@ private:
return task;
}
+ // The same as the above, except that the thread hash for the RPC response
task is set to 0.
+ template <typename TRequest>
+ rpc_response_task_ptr request_meta(const dsn::task_code &code,
+ std::shared_ptr<TRequest> &req,
+ int timeout_milliseconds)
+ {
+ return request_meta(code, req, timeout_milliseconds, 0);
+ }
+
+ // The same as the above, except that `timeout_milliseconds` for the RPC
request is set to
+ // 0, which means `rpc_timeout_milliseconds` configured for each task
would be used as the
+ // timeout. See `message_ex::create_request()` for details.
+ template <typename TRequest>
+ rpc_response_task_ptr request_meta(const dsn::task_code &code,
std::shared_ptr<TRequest> &req)
+ {
+ return request_meta(code, req, 0);
+ }
+
static inline bool is_busy(const dsn::error_code &err)
{
return err == dsn::ERR_BUSY_CREATING || err == dsn::ERR_BUSY_DROPPING;
}
+ // The same as `request_meta()`, except that it would retry multiple times
as configured
+ // once failed or busy, and return error status for the response task. If
succeed, `resp`
+ // would be set as the response.
+ //
+ // NOTE: the returned error is just for the RPC request; please also check
the possible
+ // error status in `resp` if the RPC request succeeds.
template <typename TRequest, typename TResponse>
- rpc_response_task_ptr request_meta_and_wait_response(const dsn::task_code
&code,
-
std::shared_ptr<TRequest> &req,
- TResponse &resp,
- int
timeout_milliseconds = 0,
- int reply_thread_hash
= 0)
+ error_s request_meta_and_wait_response(const dsn::task_code &code,
+ std::shared_ptr<TRequest> &req,
+ TResponse &resp,
+ int timeout_milliseconds,
+ int reply_thread_hash)
{
- rpc_response_task_ptr resp_task;
for (uint32_t i = 1; i <= FLAGS_ddl_client_max_attempt_count; ++i) {
- resp_task = request_meta(code, req, timeout_milliseconds,
reply_thread_hash);
+ auto resp_task = request_meta(code, req, timeout_milliseconds,
reply_thread_hash);
resp_task->wait();
// Failed to send request to meta server. The possible reason
might be:
// * cannot connect to meta server (such as ERR_NETWORK_FAILURE);
// * do not receive any response from meta server (such as
ERR_TIMEOUT)
if (resp_task->error() != dsn::ERR_OK) {
- return resp_task;
+ return FMT_ERR(resp_task->error(),
+ "request meta failed: task={}, error={}",
+ code,
+ resp_task->error());
}
// Once response is nullptr, it must be mocked by unit tests since
network is
@@ -358,7 +416,7 @@ private:
// Once `err` field in the received response is ERR_OK or some
non-busy error, do not
// attempt again.
if (resp.err == dsn::ERR_OK || !is_busy(resp.err)) {
- return resp_task;
+ return error_s::ok();
}
// Would not sleep for the last attempt.
@@ -371,7 +429,29 @@ private:
std::chrono::milliseconds(FLAGS_ddl_client_retry_interval_ms));
}
}
- return resp_task;
+
+ return error_s::ok();
+ }
+
+ // The same as the above, except that the thread hash for the RPC response
task is set to 0.
+ template <typename TRequest, typename TResponse>
+ error_s request_meta_and_wait_response(const dsn::task_code &code,
+ std::shared_ptr<TRequest> &req,
+ TResponse &resp,
+ int timeout_milliseconds)
+ {
+ return request_meta_and_wait_response(code, req, resp,
timeout_milliseconds, 0);
+ }
+
+ // The same as the above, except that `timeout_milliseconds` for the RPC
request is set to
+ // 0, which means `rpc_timeout_milliseconds` configured for each task
would be used as the
+ // timeout. See `message_ex::create_request()` for details.
+ template <typename TRequest, typename TResponse>
+ error_s request_meta_and_wait_response(const dsn::task_code &code,
+ std::shared_ptr<TRequest> &req,
+ TResponse &resp)
+ {
+ return request_meta_and_wait_response(code, req, resp, 0);
}
/// Send request to meta server synchronously.
diff --git a/src/client/test/ddl_client_test.cpp
b/src/client/test/ddl_client_test.cpp
index 13b13d69a..99e578104 100644
--- a/src/client/test/ddl_client_test.cpp
+++ b/src/client/test/ddl_client_test.cpp
@@ -28,8 +28,6 @@
#include "meta_admin_types.h"
#include "rpc/rpc_host_port.h"
#include "runtime/api_layer1.h"
-#include "task/task.h"
-#include "utils/autoref_ptr.h"
#include "utils/error_code.h"
#include "utils/errors.h"
#include "utils/fail_point.h"
@@ -146,7 +144,8 @@ TEST(DDLClientTest, RetryMetaRequest)
resp.err = ERR_UNKNOWN;
auto start_ms = dsn_now_ms();
- auto resp_task =
ddl_client->request_meta_and_wait_response(RPC_CM_CREATE_APP, req, resp);
+ const auto &req_result =
+ ddl_client->request_meta_and_wait_response(RPC_CM_CREATE_APP, req,
resp);
uint64_t duration_ms = dsn_now_ms() - start_ms;
// Check if all the errors have been traversed in sequence and
accepted except the last
@@ -157,7 +156,7 @@ TEST(DDLClientTest, RetryMetaRequest)
EXPECT_LE(test.expected_sleep_ms, duration_ms);
// Check if final send error is matched.
- EXPECT_EQ(test.final_send_error, resp_task->error());
+ EXPECT_EQ(test.final_send_error, req_result.code());
// Check if final response error is matched.
EXPECT_EQ(test.final_resp_error, resp.err);
diff --git a/src/meta/duplication/meta_duplication_service.cpp
b/src/meta/duplication/meta_duplication_service.cpp
index 9f778181e..f1d61a9db 100644
--- a/src/meta/duplication/meta_duplication_service.cpp
+++ b/src/meta/duplication/meta_duplication_service.cpp
@@ -107,25 +107,17 @@ void
meta_duplication_service::list_duplication_info(const duplication_list_requ
continue;
}
- const auto &err =
+ const auto &result =
utils::pattern_match(app_name, request.app_name_pattern,
request.match_type);
- if (err == ERR_NOT_MATCHED) {
+ if (result.code() == ERR_NOT_MATCHED) {
continue;
}
- if (err == ERR_NOT_IMPLEMENTED) {
- const auto &msg = fmt::format("match_type {} is not supported now",
-
static_cast<int>(request.match_type));
- response.err = err;
- response.hint_message = msg;
+ if (result.code() != ERR_OK) {
+ response.err = result.code();
+ response.hint_message = result.message();
+ LOG_ERROR("{}, app_name_pattern={}", result,
request.app_name_pattern);
- LOG_ERROR("{}: app_name_pattern={}", msg,
request.app_name_pattern);
-
- return;
- }
-
- if (err != ERR_OK) {
- response.err = err;
return;
}
diff --git a/src/meta/meta_http_service.cpp b/src/meta/meta_http_service.cpp
index efc605d58..22e732737 100644
--- a/src/meta/meta_http_service.cpp
+++ b/src/meta/meta_http_service.cpp
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+#include <cstddef>
#include <fmt/core.h>
#include <fmt/format.h>
#include <rapidjson/ostreamwrapper.h>
@@ -52,6 +53,7 @@
#include "server_load_balancer.h"
#include "server_state.h"
#include "utils/error_code.h"
+#include "utils/errors.h"
#include "utils/flags.h"
#include "utils/fmt_logging.h"
#include "utils/output_utils.h"
@@ -74,6 +76,20 @@ struct list_nodes_helper
}
};
+#define INIT_AND_CALL_LIST_APPS(target_status, list_apps_resp, http_resp)
\
+ configuration_list_apps_response list_apps_resp;
\
+ do {
\
+ configuration_list_apps_request list_apps_req;
\
+ list_apps_req.status = target_status;
\
+ _service->_state->list_apps(list_apps_req, list_apps_resp);
\
+ if (list_apps_resp.err != ERR_OK) {
\
+ http_resp.status_code = http_status_code::kInternalServerError;
\
+ http_resp.body =
\
+ error_s::make(list_apps_resp.err,
list_apps_resp.hint_message).description(); \
+ return;
\
+ }
\
+ } while (false)
+
void meta_http_service::get_app_handler(const http_request &req, http_response
&resp)
{
std::string app_name;
@@ -208,28 +224,21 @@ void meta_http_service::get_app_handler(const
http_request &req, http_response &
void meta_http_service::list_app_handler(const http_request &req,
http_response &resp)
{
bool detailed = false;
- for (const auto &p : req.query_args) {
- if (p.first == "detail") {
+ for (const auto &[name, value] : req.query_args) {
+ if (name == "detail") {
detailed = true;
- } else {
- resp.status_code = http_status_code::kBadRequest;
- return;
+ continue;
}
- }
- if (!redirect_if_not_primary(req, resp))
- return;
- configuration_list_apps_response response;
- configuration_list_apps_request request;
- request.status = dsn::app_status::AS_INVALID;
- _service->_state->list_apps(request, response);
+ resp.status_code = http_status_code::kBadRequest;
+ return;
+ }
- if (response.err != dsn::ERR_OK) {
- resp.body = response.err;
- resp.status_code = http_status_code::kInternalServerError;
+ if (!redirect_if_not_primary(req, resp)) {
return;
}
- std::vector<::dsn::app_info> &apps = response.infos;
+
+ INIT_AND_CALL_LIST_APPS(app_status::AS_INVALID, list_apps_resp, resp);
// output as json format
std::ostringstream out;
@@ -247,7 +256,7 @@ void meta_http_service::list_app_handler(const http_request
&req, http_response
tp_general.add_column("drop_time");
tp_general.add_column("drop_expire");
tp_general.add_column("envs_count");
- for (const auto &app : apps) {
+ for (const auto &app : list_apps_resp.infos) {
if (app.status != dsn::app_status::AS_AVAILABLE) {
continue;
}
@@ -303,7 +312,7 @@ void meta_http_service::list_app_handler(const http_request
&req, http_response
tp_health.add_column("unhealthy");
tp_health.add_column("write_unhealthy");
tp_health.add_column("read_unhealthy");
- for (auto &info : apps) {
+ for (const auto &info : list_apps_resp.infos) {
if (info.status != app_status::AS_AVAILABLE) {
continue;
}
@@ -391,15 +400,14 @@ void meta_http_service::list_node_handler(const
http_request &req, http_response
for (const auto &node : _service->_dead_set) {
tmp_map.emplace(node, list_nodes_helper(node.to_string(), "UNALIVE"));
}
- int alive_node_count = (_service->_alive_set).size();
- int unalive_node_count = (_service->_dead_set).size();
+
+ size_t alive_node_count = _service->_alive_set.size();
+ size_t unalive_node_count = _service->_dead_set.size();
if (detailed) {
- configuration_list_apps_response response;
- configuration_list_apps_request request;
- request.status = dsn::app_status::AS_AVAILABLE;
- _service->_state->list_apps(request, response);
- for (const auto &app : response.infos) {
+ INIT_AND_CALL_LIST_APPS(app_status::AS_AVAILABLE, list_apps_resp,
resp);
+
+ for (const auto &app : list_apps_resp.infos) {
query_cfg_request request_app;
query_cfg_response response_app;
request_app.app_name = app.app_name;
@@ -515,21 +523,13 @@ void meta_http_service::get_app_envs_handler(const
http_request &req, http_respo
}
// get all of the apps
- configuration_list_apps_response response;
- configuration_list_apps_request request;
- request.status = dsn::app_status::AS_AVAILABLE;
- _service->_state->list_apps(request, response);
- if (response.err != dsn::ERR_OK) {
- resp.body = response.err.to_string();
- resp.status_code = http_status_code::kInternalServerError;
- return;
- }
+ INIT_AND_CALL_LIST_APPS(app_status::AS_AVAILABLE, list_apps_resp, resp);
// using app envs to generate a table_printer
dsn::utils::table_printer tp;
- for (auto &app : response.infos) {
+ for (const auto &app : list_apps_resp.infos) {
if (app.app_name == app_name) {
- for (auto env : app.envs) {
+ for (const auto &env : app.envs) {
tp.add_row_name_and_data(env.first, env.second);
}
break;
diff --git a/src/meta/server_state.cpp b/src/meta/server_state.cpp
index 30d5df6b0..ed309338b 100644
--- a/src/meta/server_state.cpp
+++ b/src/meta/server_state.cpp
@@ -78,6 +78,7 @@
#include "utils/blob.h"
#include "utils/command_manager.h"
#include "utils/config_api.h"
+#include "utils/errors.h"
#include "utils/fail_point.h"
#include "utils/flags.h"
#include "utils/fmt_logging.h"
@@ -1612,16 +1613,49 @@ void server_state::list_apps(const
configuration_list_apps_request &request,
configuration_list_apps_response &response,
dsn::message_ex *msg) const
{
- LOG_DEBUG("list app request, status({})", request.status);
+ LOG_DEBUG("list app request: {}{}status={}",
+ request.__isset.app_name_pattern
+ ? fmt::format("app_name_pattern={}, ",
request.app_name_pattern)
+ : "",
+ request.__isset.match_type
+ ? fmt::format("match_type={}, ",
enum_to_string(request.match_type))
+ : "",
+ request.status);
+
zauto_read_lock l(_lock);
- for (const auto &kv : _all_apps) {
- app_state &app = *(kv.second);
- if (request.status == app_status::AS_INVALID || request.status ==
app.status) {
- if (nullptr == msg ||
_meta_svc->get_access_controller()->allowed(msg, app.app_name)) {
- response.infos.push_back(app);
+
+ for (const auto &[_, app] : _all_apps) {
+ // If the pattern is provided in the request, any table chosen to be
listed must match it.
+ if (request.__isset.app_name_pattern && request.__isset.match_type) {
+ const auto &result =
+ utils::pattern_match(app->app_name, request.app_name_pattern,
request.match_type);
+ if (result.code() == ERR_NOT_MATCHED) {
+ continue;
+ }
+
+ if (result.code() != ERR_OK) {
+ response.err = result.code();
+ response.__set_hint_message(result.message());
+ LOG_ERROR("{}, app_name_pattern={}", result,
request.app_name_pattern);
+
+ return;
}
}
+
+ // Only in the following two cases would a table be chosen to be
listed, according to
+ // the requested status:
+ // - `app_status::AS_INVALID` means no filter, in other words, any
table with any status
+ // could be chosen;
+ // - or, current status of a table is the same as the requested status.
+ if (request.status != app_status::AS_INVALID && request.status !=
app->status) {
+ continue;
+ }
+
+ if (msg == nullptr || _meta_svc->get_access_controller()->allowed(msg,
app->app_name)) {
+ response.infos.push_back(*app);
+ }
}
+
response.err = dsn::ERR_OK;
}
diff --git a/src/server/hotspot_partition_calculator.h
b/src/server/hotspot_partition_calculator.h
index f5e3d93ed..0280f09f7 100644
--- a/src/server/hotspot_partition_calculator.h
+++ b/src/server/hotspot_partition_calculator.h
@@ -91,7 +91,8 @@ private:
// hotkey on the replica automatically
std::vector<std::array<int, 2>> _hotpartition_counter;
- typedef dsn::rpc_holder<detect_hotkey_request, detect_hotkey_response>
detect_hotkey_rpc;
+ using detect_hotkey_rpc =
dsn::rpc_holder<dsn::replication::detect_hotkey_request,
+
dsn::replication::detect_hotkey_response>;
friend class hotspot_partition_test;
};
diff --git a/src/server/info_collector.cpp b/src/server/info_collector.cpp
index a59d42ec1..23fe6a523 100644
--- a/src/server/info_collector.cpp
+++ b/src/server/info_collector.cpp
@@ -74,7 +74,7 @@ DEFINE_TASK_CODE(LPC_PEGASUS_STORAGE_SIZE_STAT_TIMER,
info_collector::info_collector()
{
std::vector<::dsn::host_port> meta_servers;
- replica_helper::parse_server_list(FLAGS_server_list, meta_servers);
+ dsn::replication::replica_helper::parse_server_list(FLAGS_server_list,
meta_servers);
_meta_servers.assign_group("meta-servers");
for (auto &ms : meta_servers) {
@@ -86,7 +86,8 @@ info_collector::info_collector()
_shell_context = std::make_shared<shell_context>();
_shell_context->current_cluster_name = _cluster_name;
_shell_context->meta_list = meta_servers;
- _shell_context->ddl_client.reset(new replication_ddl_client(meta_servers));
+ _shell_context->ddl_client =
+
std::make_unique<dsn::replication::replication_ddl_client>(meta_servers);
// initialize the _client.
CHECK(pegasus_client_factory::initialize(nullptr), "Initialize the pegasus
client failed");
diff --git a/src/server/pegasus_mutation_duplicator.cpp
b/src/server/pegasus_mutation_duplicator.cpp
index f31117435..bb1b7f727 100644
--- a/src/server/pegasus_mutation_duplicator.cpp
+++ b/src/server/pegasus_mutation_duplicator.cpp
@@ -26,6 +26,7 @@
#include <cstdint>
#include <functional>
#include <memory>
+#include <set>
#include <string_view>
#include <tuple>
#include <utility>
@@ -34,6 +35,7 @@
#include "client_lib/pegasus_client_impl.h"
#include "common/common.h"
#include "common/duplication_common.h"
+#include "common/replication.codes.h"
#include "duplication_internal_types.h"
#include "gutil/map_util.h"
#include "pegasus/client.h"
diff --git a/src/shell/command_helper.h b/src/shell/command_helper.h
index 6c4cafdd7..8f02b6fe7 100644
--- a/src/shell/command_helper.h
+++ b/src/shell/command_helper.h
@@ -89,17 +89,24 @@
#define SHELL_PRINTLN_OK(msg, ...) SHELL_PRINT_OK_BASE("{}\n",
fmt::format(msg, ##__VA_ARGS__))
-using namespace dsn::replication;
+// Print messages to stderr and return false if `exp` is evaluated to false.
+#define SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(exp, ...)
\
+ do {
\
+ if (dsn_unlikely(!(exp))) {
\
+ SHELL_PRINTLN_ERROR(__VA_ARGS__);
\
+ return false;
\
+ }
\
+ } while (0)
+
+#define RETURN_FALSE_IF_SAMPLE_INTERVAL_MS_INVALID()
\
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(dsn::buf2uint32(optarg,
sample_interval_ms), \
+ "parse sample_interval_ms({}) failed",
\
+ optarg);
\
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(sample_interval_ms > 0,
"sample_interval_ms should be > 0")
DEFINE_TASK_CODE(LPC_SCAN_DATA, TASK_PRIORITY_COMMON,
::dsn::THREAD_POOL_DEFAULT)
DEFINE_TASK_CODE(LPC_GET_METRICS, TASK_PRIORITY_COMMON,
::dsn::THREAD_POOL_DEFAULT)
-#define RETURN_FALSE_IF_SAMPLE_INTERVAL_MS_INVALID()
\
- PRINT_AND_RETURN_FALSE_IF_NOT(dsn::buf2uint32(optarg, sample_interval_ms),
\
- "parse sample_interval_ms({}) failed\n",
\
- optarg);
\
- PRINT_AND_RETURN_FALSE_IF_NOT(sample_interval_ms > 0, "sample_interval_ms
should be > 0\n")
-
enum scan_data_operator
{
SCAN_COPY,
@@ -1852,11 +1859,12 @@ 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) {
- LOG_ERROR("list apps failed, error = {}", err);
+ const auto &result =
sc->ddl_client->list_apps(dsn::app_status::AS_AVAILABLE, apps);
+ if (!result) {
+ LOG_ERROR("list apps failed, error={}", result);
return false;
}
+
if (!fill_nodes(sc, "replica-server", nodes)) {
LOG_ERROR("get replica server node list failed");
return false;
diff --git a/src/shell/command_utils.cpp b/src/shell/command_utils.cpp
index fc2464e88..58a711268 100644
--- a/src/shell/command_utils.cpp
+++ b/src/shell/command_utils.cpp
@@ -18,6 +18,7 @@
#include "command_utils.h"
#include <fmt/core.h>
+#include <cstdio>
#include <memory>
#include "client/replication_ddl_client.h"
diff --git a/src/shell/command_utils.h b/src/shell/command_utils.h
index 3c4ffacac..afde55a2d 100644
--- a/src/shell/command_utils.h
+++ b/src/shell/command_utils.h
@@ -19,7 +19,6 @@
#pragma once
-#include <cstdio>
#include <functional>
#include <map>
#include <set>
@@ -29,7 +28,6 @@
#include "shell/argh.h"
#include "utils/error_code.h"
#include "utils/errors.h"
-#include "utils/ports.h"
#include "utils/strings.h"
namespace dsn {
@@ -96,15 +94,6 @@ bool validate_ip(shell_context *sc,
/*out*/ dsn::host_port &target_hp,
/*out*/ std::string &err_info);
-// Print messages to stderr and return false if `exp` is evaluated to false.
-#define PRINT_AND_RETURN_FALSE_IF_NOT(exp, ...)
\
- do {
\
- if (dsn_unlikely(!(exp))) {
\
- fmt::print(stderr, __VA_ARGS__);
\
- return false;
\
- }
\
- } while (0)
-
template <typename EnumType>
EnumType type_from_string(const std::map<int, const char *> &type_maps,
const std::string &type_string,
diff --git a/src/shell/commands/node_management.cpp
b/src/shell/commands/node_management.cpp
index 109a8b1cc..7f4f8f84c 100644
--- a/src/shell/commands/node_management.cpp
+++ b/src/shell/commands/node_management.cpp
@@ -27,7 +27,6 @@
#include <algorithm>
// IWYU pragma: no_include <bits/getopt_core.h>
#include <chrono>
-#include <fstream>
#include <initializer_list>
#include <iostream>
#include <map>
@@ -454,14 +453,16 @@ bool ls_nodes(command_executor *, shell_context *sc,
arguments args)
}
}
- dsn::utils::multi_table_printer mtp;
+ dsn::utils::multi_table_printer multi_printer;
if (!(status.empty() && output_file.empty())) {
dsn::utils::table_printer tp("parameters");
- if (!status.empty())
+ if (!status.empty()) {
tp.add_row_name_and_data("status", status);
- if (!output_file.empty())
+ }
+ if (!output_file.empty()) {
tp.add_row_name_and_data("out_file", output_file);
- mtp.add(std::move(tp));
+ }
+ multi_printer.add(std::move(tp));
}
::dsn::replication::node_status::type s =
::dsn::replication::node_status::NS_INVALID;
@@ -469,15 +470,15 @@ bool ls_nodes(command_executor *, shell_context *sc,
arguments args)
s = type_from_string(dsn::replication::_node_status_VALUES_TO_NAMES,
std::string("ns_") + status,
::dsn::replication::node_status::NS_INVALID);
- PRINT_AND_RETURN_FALSE_IF_NOT(s !=
::dsn::replication::node_status::NS_INVALID,
- "parse {} as node_status::type failed",
- status);
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(s !=
::dsn::replication::node_status::NS_INVALID,
+ "parse {} as node_status::type
failed",
+ status);
}
std::map<dsn::host_port, dsn::replication::node_status::type> status_by_hp;
auto r = sc->ddl_client->list_nodes(s, status_by_hp);
if (r != dsn::ERR_OK) {
- std::cout << "list nodes failed, error=" << r << std::endl;
+ fmt::println("list nodes failed, error={}", r);
return true;
}
@@ -494,9 +495,9 @@ bool ls_nodes(command_executor *, shell_context *sc,
arguments args)
if (detailed) {
std::vector<::dsn::app_info> apps;
- r = sc->ddl_client->list_apps(dsn::app_status::AS_AVAILABLE, apps);
- if (r != dsn::ERR_OK) {
- std::cout << "list apps failed, error=" << r << std::endl;
+ const auto &result =
sc->ddl_client->list_apps(dsn::app_status::AS_AVAILABLE, apps);
+ if (!result) {
+ fmt::println("list apps failed, error={}", result);
return true;
}
@@ -506,7 +507,7 @@ bool ls_nodes(command_executor *, shell_context *sc,
arguments args)
std::vector<dsn::partition_configuration> pcs;
r = sc->ddl_client->list_app(app.app_name, app_id,
partition_count, pcs);
if (r != dsn::ERR_OK) {
- std::cout << "list app " << app.app_name << " failed, error="
<< r << std::endl;
+ fmt::println("list app {} failed, error={}", app.app_name, r);
return true;
}
@@ -530,7 +531,7 @@ bool ls_nodes(command_executor *, shell_context *sc,
arguments args)
if (resource_usage) {
std::vector<node_desc> nodes;
if (!fill_nodes(sc, "replica-server", nodes)) {
- std::cout << "get replica server node list failed" << std::endl;
+ fmt::println("get replica server node list failed");
return true;
}
@@ -553,7 +554,7 @@ bool ls_nodes(command_executor *, shell_context *sc,
arguments args)
if (show_qps) {
std::vector<node_desc> nodes;
if (!fill_nodes(sc, "replica-server", nodes)) {
- std::cout << "get replica server node list failed" << std::endl;
+ fmt::println("get replica server node list failed");
return true;
}
@@ -595,7 +596,7 @@ bool ls_nodes(command_executor *, shell_context *sc,
arguments args)
if (show_latency) {
std::vector<node_desc> nodes;
if (!fill_nodes(sc, "replica-server", nodes)) {
- std::cout << "get replica server node list failed" << std::endl;
+ fmt::println("get replica server node list failed");
return true;
}
@@ -616,19 +617,6 @@ bool ls_nodes(command_executor *, shell_context *sc,
arguments args)
}
}
- // print configuration_list_nodes_response
- std::streambuf *buf = nullptr;
- std::ofstream of;
-
- // TODO(wangdan): use dsn::utils::output() in output_utils.h instead.
- if (!output_file.empty()) {
- of.open(output_file);
- buf = of.rdbuf();
- } else {
- buf = std::cout.rdbuf();
- }
- std::ostream out(buf);
-
dsn::utils::table_printer tp("details");
tp.add_title("address");
tp.add_column("status");
@@ -694,16 +682,15 @@ bool ls_nodes(command_executor *, shell_context *sc,
arguments args)
tp.append_data(kv.second.multi_put_p99 / 1e6);
}
}
- mtp.add(std::move(tp));
+ multi_printer.add(std::move(tp));
dsn::utils::table_printer tp_count("summary");
tp_count.add_row_name_and_data("total_node_count", status_by_hp.size());
tp_count.add_row_name_and_data("alive_node_count", alive_node_count);
tp_count.add_row_name_and_data("unalive_node_count", status_by_hp.size() -
alive_node_count);
- mtp.add(std::move(tp_count));
+ multi_printer.add(std::move(tp_count));
- // TODO(wangdan): use dsn::utils::output() in output_utils.h instead.
- mtp.output(out, json ? tp_output_format::kJsonPretty :
tp_output_format::kTabular);
+ dsn::utils::output(output_file, json, multi_printer);
return true;
}
diff --git a/src/shell/commands/rebalance.cpp b/src/shell/commands/rebalance.cpp
index 9b6f19d81..8b07b0f5a 100644
--- a/src/shell/commands/rebalance.cpp
+++ b/src/shell/commands/rebalance.cpp
@@ -34,6 +34,7 @@
#include "meta_admin_types.h"
#include "rpc/rpc_host_port.h"
#include "shell/command_executor.h"
+#include "shell/command_helper.h"
#include "shell/command_utils.h"
#include "shell/commands.h"
#include "shell/sds/sds.h"
@@ -49,9 +50,9 @@ bool set_meta_level(command_executor *e, shell_context *sc,
arguments args)
l = type_from_string(_meta_function_level_VALUES_TO_NAMES,
std::string("fl_") + args.argv[1],
meta_function_level::fl_invalid);
- PRINT_AND_RETURN_FALSE_IF_NOT(l != meta_function_level::fl_invalid,
- "parse {} as meta function level failed\n",
- args.argv[1]);
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(l != meta_function_level::fl_invalid,
+ "parse {} as meta function level
failed",
+ args.argv[1]);
configuration_meta_control_response resp =
sc->ddl_client->control_meta_function_level(l);
if (resp.err == dsn::ERR_OK) {
@@ -106,31 +107,32 @@ bool propose(command_executor *e, shell_context *sc,
arguments args)
break;
case 'g':
ans = request.gpid.parse_from(optarg);
- PRINT_AND_RETURN_FALSE_IF_NOT(ans, "parse {} as gpid failed\n",
optarg);
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(ans, "parse {} as gpid
failed", optarg);
break;
case 'p':
proposal_type += optarg;
break;
case 't':
target = dsn::host_port::from_string(optarg);
- PRINT_AND_RETURN_FALSE_IF_NOT(target, "parse {} as
target_host_port failed\n", optarg);
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(
+ target, "parse {} as target_host_port failed", optarg);
break;
case 'n':
node = dsn::host_port::from_string(optarg);
- PRINT_AND_RETURN_FALSE_IF_NOT(target, "parse {} as node
failed\n", optarg);
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(target, "parse {} as node
failed", optarg);
break;
default:
return false;
}
}
- PRINT_AND_RETURN_FALSE_IF_NOT(target, "need set target by -t\n");
- PRINT_AND_RETURN_FALSE_IF_NOT(node, "need set node by -n\n");
- PRINT_AND_RETURN_FALSE_IF_NOT(request.gpid.get_app_id() != -1, "need set
gpid by -g\n");
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(target, "need set target by -t");
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(node, "need set node by -n");
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(request.gpid.get_app_id() != -1, "need
set gpid by -g");
config_type::type tp =
type_from_string(_config_type_VALUES_TO_NAMES, proposal_type,
config_type::CT_INVALID);
- PRINT_AND_RETURN_FALSE_IF_NOT(
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(
tp != config_type::CT_INVALID, "parse {} as config_type failed.\n",
proposal_type);
request.action_list = {new_proposal_action(target, node, tp)};
dsn::error_code err = sc->ddl_client->send_balancer_proposal(request);
diff --git a/src/shell/commands/table_management.cpp
b/src/shell/commands/table_management.cpp
index 0ab0cd1f4..63d205587 100644
--- a/src/shell/commands/table_management.cpp
+++ b/src/shell/commands/table_management.cpp
@@ -26,9 +26,11 @@
#include <algorithm>
#include <cstdint>
#include <fstream>
+#include <initializer_list>
#include <iostream>
#include <map>
#include <memory>
+#include <set>
#include <string>
#include <utility>
#include <vector>
@@ -39,6 +41,7 @@
#include "meta_admin_types.h"
#include "pegasus_utils.h"
#include "rpc/rpc_host_port.h"
+#include "shell/argh.h"
#include "shell/command_executor.h"
#include "shell/command_helper.h"
#include "shell/command_utils.h"
@@ -52,6 +55,7 @@
#include "utils/ports.h"
#include "utils/string_conv.h"
#include "utils/strings.h"
+#include "utils_types.h"
DSN_DEFINE_uint32(shell, tables_sample_interval_ms, 1000, "The interval
between sampling metrics.");
DSN_DEFINE_validator(tables_sample_interval_ms, [](uint32_t value) -> bool {
return value > 0; });
@@ -63,57 +67,61 @@ double convert_to_ratio(double hit, double total)
bool ls_apps(command_executor *e, shell_context *sc, arguments args)
{
- static struct option long_options[] = {{"all", no_argument, 0, 'a'},
- {"detailed", no_argument, 0, 'd'},
- {"json", no_argument, 0, 'j'},
- {"status", required_argument, 0,
's'},
- {"output", required_argument, 0,
'o'},
- {0, 0, 0, 0}};
+ // ls [-a|--all] [-d|--detailed] [-j|--json] [-o|--output file_name]
+ // [-s|--status all|available|creating|dropping|dropped] "
+ // [-p|--app_name_pattern str] [-m|--match_type
all|exact|anywhere|prefix|postfix]"
- bool show_all = false;
- bool detailed = false;
- bool json = false;
- std::string status;
- std::string output_file;
- optind = 0;
- while (true) {
- int option_index = 0;
- int c;
- c = getopt_long(args.argc, args.argv, "adjs:o:", long_options,
&option_index);
- if (c == -1)
- break;
- switch (c) {
- case 'a':
- show_all = true;
- break;
- case 'd':
- detailed = true;
- break;
- case 'j':
- json = true;
- break;
- case 's':
- status = optarg;
- break;
- case 'o':
- output_file = optarg;
- break;
- default:
- return false;
+ // All valid parameters and flags are given as follows.
+ static const std::set<std::string> params = {
+ "o", "output", "s", "status", "p", "app_name_pattern", "m",
"match_type"};
+ static const std::set<std::string> flags = {"a", "all", "d", "detailed",
"j", "json"};
+
+ argh::parser cmd(args.argc, args.argv,
argh::parser::PREFER_PARAM_FOR_UNREG_OPTION);
+
+ // Check if input parameters and flags are valid.
+ const auto &check = validate_cmd(cmd, params, flags);
+ if (!check) {
+ SHELL_PRINTLN_ERROR("{}", check.description());
+ return false;
+ }
+
+ const bool show_all = cmd[{"-a", "--all"}];
+ const bool detailed = cmd[{"-d", "--detailed"}];
+ const bool json = cmd[{"-j", "--json"}];
+
+ const std::string output_file(cmd({"-o", "--output"}, "").str());
+
+ const std::string status_str(cmd({"-s", "--status"}, "").str());
+ auto status = dsn::app_status::AS_INVALID;
+ if (status_str.empty()) {
+ // `show_all` functions only when target `status` is not specified.
+ if (!show_all) {
+ // That `show_all` is not given means just showing available
tables.
+ status = dsn::app_status::AS_AVAILABLE;
}
+ } else if (status_str != "all") {
+ status = type_from_string(dsn::_app_status_VALUES_TO_NAMES,
+ fmt::format("as_{}", status_str),
+ dsn::app_status::AS_INVALID);
+ SHELL_PRINT_AND_RETURN_FALSE_IF_NOT(status !=
dsn::app_status::AS_INVALID,
+ "parse {} as app_status::type
failed",
+ status_str);
}
- ::dsn::app_status::type s = ::dsn::app_status::AS_INVALID;
- if (!status.empty() && status != "all") {
- s = type_from_string(::dsn::_app_status_VALUES_TO_NAMES,
- std::string("as_") + status,
- ::dsn::app_status::AS_INVALID);
- PRINT_AND_RETURN_FALSE_IF_NOT(
- s != ::dsn::app_status::AS_INVALID, "parse {} as app_status::type
failed", status);
+ // Read the parttern of table name with empty string as default.
+ const std::string app_name_pattern(cmd({"-p", "--app_name_pattern"},
"").str());
+
+ // Read the match type of the pattern for table name with "matching all"
as default,
+ // typically requesting all tables owned by this cluster.
+ auto match_type = dsn::utils::pattern_match_type::PMT_MATCH_ALL;
+ PARSE_OPT_ENUM(match_type, dsn::utils::pattern_match_type::PMT_INVALID,
{"-m", "--match_type"});
+
+ const auto &result = sc->ddl_client->list_apps(
+ detailed, json, output_file, status, app_name_pattern, match_type);
+ if (!result) {
+ fmt::println("list apps failed, error={}", result);
}
- ::dsn::error_code err = sc->ddl_client->list_apps(s, show_all, detailed,
json, output_file);
- if (err != ::dsn::ERR_OK)
- std::cout << "list apps failed, error=" << err << std::endl;
+
return true;
}
diff --git a/src/shell/main.cpp b/src/shell/main.cpp
index 1a7c90d38..0c7defb70 100644
--- a/src/shell/main.cpp
+++ b/src/shell/main.cpp
@@ -91,9 +91,10 @@ static command_executor commands[] = {
},
{
"ls",
- "list all apps",
- "[-a|-all] [-d|--detailed] [-j|--json] [-o|--output file_name] "
- "[-s|--status all|available|creating|dropping|dropped]",
+ "list tables with specified status or pattern",
+ "[-a|--all] [-d|--detailed] [-j|--json] [-o|--output file_name] "
+ "[-s|--status all|available|creating|dropping|dropped] "
+ "[-p|--app_name_pattern str] [-m|--match_type
all|exact|anywhere|prefix|postfix]",
ls_apps,
},
{
@@ -542,9 +543,9 @@ static command_executor commands[] = {
{"query_dup", "query duplication info", "<app_name> [-d|--detail]",
query_dup},
{"dups",
"list duplications of one or multiple tables with specified pattern",
- "[-a|--app_name_pattern str] [-m|--match_type str] [-p|--list_partitions]
"
- "[-g|--progress_gap num] [-u|--show_unfinishd] [-o|--output file_name] "
- "[-j|--json]",
+ "[-a|--app_name_pattern str] [-m|--match_type
all|exact|anywhere|prefix|postfix] "
+ "[-p|--list_partitions] [-g|--progress_gap num] [-u|--show_unfinishd] "
+ "[-o|--output file_name] [-j|--json]",
ls_dups},
{"remove_dup", "remove duplication", "<app_name> <dup_id>", remove_dup},
{"start_dup", "start duplication", "<app_name> <dup_id>", start_dup},
diff --git a/src/utils/errors.h b/src/utils/errors.h
index 734ee9375..d262f2241 100644
--- a/src/utils/errors.h
+++ b/src/utils/errors.h
@@ -64,15 +64,19 @@ public:
error_s(const error_s &rhs) noexcept { copy(rhs); }
error_s &operator=(const error_s &rhs) noexcept
{
+ if (this == &rhs) {
+ return *this;
+ }
+
copy(rhs);
- return (*this);
+ return *this;
}
// movable
- error_s(error_s &&rhs) noexcept = default;
+ error_s(error_s &&) noexcept = default;
error_s &operator=(error_s &&) noexcept = default;
- static error_s make(error_code code, std::string_view reason) { return
error_s(code, reason); }
+ static error_s make(error_code code, std::string_view reason) { return
{code, reason}; }
static error_s make(error_code code)
{
@@ -103,11 +107,18 @@ public:
if (!_info) {
return ERR_OK.to_string();
}
- std::string code = _info->code.to_string();
- return _info->msg.empty() ? code : code + ": " + _info->msg;
+
+ std::string msg(_info->code.to_string());
+ if (!_info->msg.empty()) {
+ fmt::format_to(std::back_inserter(msg), ": {}", _info->msg);
+ }
+
+ return msg;
}
- error_code code() const { return _info ? error_code(_info->code) : ERR_OK;
}
+ [[nodiscard]] error_code code() const { return _info ? _info->code :
ERR_OK; }
+
+ [[nodiscard]] std::string message() const { return _info ? _info->msg :
""; }
error_s &operator<<(const char str[])
{
@@ -145,14 +156,17 @@ public:
}
private:
- error_s(error_code code, std::string_view msg) noexcept : _info(new
error_info(code, msg)) {}
+ error_s(error_code code, std::string_view msg) noexcept
+ : _info(std::make_unique<error_info>(code, msg))
+ {
+ }
struct error_info
{
error_code code;
std::string msg; // TODO(wutao1): use raw char* to improve performance?
- error_info(error_code c, std::string_view s) : code(c), msg(s) {}
+ error_info(error_code c, std::string_view s) noexcept : code(c),
msg(s) {}
};
void copy(const error_s &rhs)
@@ -170,7 +184,6 @@ private:
}
}
-private:
std::unique_ptr<error_info> _info;
};
@@ -233,11 +246,19 @@ USER_DEFINED_STRUCTURE_FORMATTER(::dsn::error_s);
}
\
} while (false)
+#define RETURN_EC_NOT_OK(s)
\
+ do {
\
+ const ::dsn::error_s &_s = (s);
\
+ if (dsn_unlikely(!_s)) {
\
+ return _s.code();
\
+ }
\
+ } while (false)
+
#define RETURN_ES_NOT_OK_MSG(s, ...)
\
do {
\
const ::dsn::error_s &_s = (s);
\
if (dsn_unlikely(!_s)) {
\
- fmt::print(stderr, "{}: {}\n", _s.description(),
fmt::format(__VA_ARGS__)); \
+ fmt::println(stderr, "{}: {}", _s, fmt::format(__VA_ARGS__));
\
return _s;
\
}
\
} while (false)
@@ -246,7 +267,7 @@ USER_DEFINED_STRUCTURE_FORMATTER(::dsn::error_s);
do {
\
const ::dsn::error_s &_s = (s);
\
if (dsn_unlikely(!_s)) {
\
- fmt::print(stderr, "{}: {}\n", _s.description(),
fmt::format(__VA_ARGS__)); \
+ fmt::println(stderr, "{}: {}", _s, fmt::format(__VA_ARGS__));
\
return _s.code();
\
}
\
} while (false)
@@ -255,7 +276,7 @@ USER_DEFINED_STRUCTURE_FORMATTER(::dsn::error_s);
do {
\
::dsn::error_s _s = (s);
\
if (dsn_unlikely(!_s)) {
\
- fmt::print(stderr, "{}: {}\n", _s.description(),
fmt::format(__VA_ARGS__)); \
+ fmt::println(stderr, "{}: {}", _s, fmt::format(__VA_ARGS__));
\
return dsn::error_with<T>(std::move(_s));
\
}
\
} while (false)
diff --git a/src/utils/output_utils.h b/src/utils/output_utils.h
index 0950f92dd..7ffe16588 100644
--- a/src/utils/output_utils.h
+++ b/src/utils/output_utils.h
@@ -227,8 +227,8 @@ private:
std::vector<table_printer> _tps;
};
-// Used as a general interface for printer to output to a file, typically
`table_printer`
-// and `multi_table_printer`.
+// Used as a general interface for printer to output to a file in json or
tabular format,
+// typically `table_printer` and `multi_table_printer`.
template <typename Printer>
void output(const std::string &file_path, bool json, const Printer &printer)
{
@@ -249,8 +249,16 @@ void output(const std::string &file_path, bool json, const
Printer &printer)
: table_printer::output_format::kTabular);
}
-// Used as a general interface for printer to output to stdout, typically
`table_printer`
-// and `multi_table_printer`.
+// Used as a general interface for printer to output to a file in tabular
format, typically
+// `table_printer` and `multi_table_printer`.
+template <typename Printer>
+void output(const std::string &file_path, const Printer &printer)
+{
+ output(file_path, false, printer);
+}
+
+// Used as a general interface for printer to output to stdout in json or
tabular format,
+// typically `table_printer` and `multi_table_printer`.
template <typename Printer>
void output(bool json, const Printer &printer)
{
diff --git a/src/utils/strings.cpp b/src/utils/strings.cpp
index a8acd8058..fb4d2b65f 100644
--- a/src/utils/strings.cpp
+++ b/src/utils/strings.cpp
@@ -26,14 +26,15 @@
#include <absl/strings/ascii.h>
#include <boost/algorithm/string/predicate.hpp>
-#include <cstdio>
#include <openssl/md5.h>
#include <strings.h>
#include <algorithm>
+#include <cstdio>
#include <cstring>
#include <sstream> // IWYU pragma: keep
#include <utility>
+#include "utils/error_code.h"
#include "utils/fmt_logging.h"
#include "utils/strings.h"
@@ -114,9 +115,9 @@ bool mequals(const void *lhs, const void *rhs, size_t n)
#undef CHECK_NULL_PTR
-error_code pattern_match(const std::string &str,
- const std::string &pattern,
- pattern_match_type::type match_type)
+error_s pattern_match(const std::string &str,
+ const std::string &pattern,
+ pattern_match_type::type match_type)
{
bool matched = false;
switch (match_type) {
@@ -145,10 +146,17 @@ error_code pattern_match(const std::string &str,
case pattern_match_type::PMT_MATCH_REGEX:
default:
- return ERR_NOT_IMPLEMENTED;
+ return FMT_ERR(ERR_NOT_IMPLEMENTED,
+ "match_type is not supported: val={}, str={}",
+ static_cast<int>(match_type),
+ enum_to_string(match_type));
+ }
+
+ if (!matched) {
+ return error_s::make(ERR_NOT_MATCHED);
}
- return matched ? ERR_OK : ERR_NOT_MATCHED;
+ return error_s::ok();
}
std::string get_last_component(const std::string &input, const char
splitters[])
diff --git a/src/utils/strings.h b/src/utils/strings.h
index eb5dcbeba..8972e78ec 100644
--- a/src/utils/strings.h
+++ b/src/utils/strings.h
@@ -36,7 +36,7 @@
#include "utils_types.h"
#include "utils/enum_helper.h"
-#include "utils/error_code.h"
+#include "utils/errors.h"
namespace dsn::utils {
@@ -75,9 +75,14 @@ bool iequals(const char *lhs, const std::string &rhs, size_t
n);
// Decide whether the first n bytes of two memory areas are equal, even if one
of them is NULL.
bool mequals(const void *lhs, const void *rhs, size_t n);
-error_code pattern_match(const std::string &str,
- const std::string &pattern,
- pattern_match_type::type match_type);
+// Try to match target string `str` with provided `pattern` in `match_type`.
Return the code
+// representing matched, not matched, or some error with additional message(if
any) as follows:
+// - ERR_OK: matched;
+// - ERR_NOT_MATCHED: not matched;
+// - ERR_NOT_IMPLEMENTED: `match_type` is not supported.
+error_s pattern_match(const std::string &str,
+ const std::string &pattern,
+ pattern_match_type::type match_type);
// Split the `input` string by the only character `separator` into tokens.
Leading and trailing
// spaces of each token will be stripped. Once the token is empty, or become
empty after
diff --git a/src/utils/test/utils.cpp b/src/utils/test/utils.cpp
index c018457e9..01c880acd 100644
--- a/src/utils/test/utils.cpp
+++ b/src/utils/test/utils.cpp
@@ -41,6 +41,7 @@
#include "utils/binary_writer.h"
#include "utils/crc.h"
#include "utils/error_code.h"
+#include "utils/errors.h"
#include "utils/link.h"
#include "utils/rand.h"
#include "utils/strings.h"
@@ -313,7 +314,7 @@ TEST_P(PatternMatchTest, PatternMatch)
{
const auto &test_case = GetParam();
const auto actual_err = pattern_match(test_case.str, test_case.pattern,
test_case.match_type);
- EXPECT_EQ(test_case.expected_err, actual_err);
+ EXPECT_EQ(test_case.expected_err, actual_err.code());
}
INSTANTIATE_TEST_SUITE_P(StringTest, PatternMatchTest,
testing::ValuesIn(pattern_match_tests));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
