SWC-DB  v0.5.12 C++ documentations
SWC-DB© (Super Wide Column Database) - High Performance Scalable Database (https://github.com/kashirin-alex/swc-db)
Shell_Manager.cc
Go to the documentation of this file.
1 /*
2  * SWC-DB© Copyright since 2019 Alex Kashirin <kashirin.alex@gmail.com>
3  * License details at <https://github.com/kashirin-alex/swc-db/#license>
4  */
5 
6 
7 #include "swcdb/utils/cli/Shell_Manager.h"
8 #include "swcdb/db/Protocol/Mngr/req/Report.h"
9 #include "swcdb/db/client/sql/Reader.h"
10 #include "swcdb/core/StateSynchronization.h"
11 
12 
13 namespace SWC { namespace Utils { namespace shell {
14 
15 
16 Mngr::Mngr()
17  : Interface("\033[32mSWC-DB(\033[36mmngr\033[32m)\033[33m> \033[00m",
18  "/tmp/.swc-cli-manager-history",
19  CLI::MANAGER),
20  clients(client::Clients::make(
21  *Env::Config::settings(),
22  Comm::IoContext::make("Mngr", 8),
23  nullptr, // std::make_shared<client::ManagerContext>()
24  nullptr // std::make_shared<client::RangerContext>()
25  )->init()
26  ) {
27 
28  add_option(
29  "cluster",
30  {"report cluster-status state by error",
31  "cluster;"},
32  [ptr=this](std::string& cmd){return ptr->cluster_status(cmd);},
33  "(?i)^(cluster)(\\s|$)"
34  );
35  add_option(
36  "status",
37  {"report managers status by mngr-endpoint, of Schema-Role or by All",
38  "status [Schemas(default) | ALL | endpoint=HOST/IP(|PORT)];"},
39  [ptr=this](std::string& cmd){return ptr->managers_status(cmd);},
40  "(?i)^(report|status)(\\s|$)"
41  );
42  add_option(
43  "column-status",
44  {"report column status with ranges status",
45  "column-status [cid=CID | name=NAME];"},
46  [ptr=this](std::string& cmd){return ptr->column_status(cmd);},
47  "(?i)^(report-column|column-status)"
48  );
49  add_option(
50  "rangers-status",
51  {"report rangers status by Manager of column or Rangers-Role",
52  "rangers-status [without | cid=CID | name=NAME];"},
53  [ptr=this](std::string& cmd){return ptr->rangers_status(cmd);},
54  "(?i)^(report-rangers|rangers-status)"
55  );
56 }
57 
58 Mngr::~Mngr() {
59  clients->stop();
60 }
61 
62 bool Mngr::cluster_status(std::string&) {
63  std::string message;
64  client::Mngr::Hosts hosts;
65  auto groups = clients->managers.groups->get_groups();
66  for(auto& g : groups) {
67  auto tmp = g->get_hosts();
68  hosts.insert(hosts.cend(), tmp.cbegin(), tmp.cend());
69  }
70  if(hosts.empty()) {
71  err = Error::FAILED_EXPECTATION;
72  message.append("Empty swc.mngr.host config\n");
73  return error(message);
74  }
75 
76  Core::StateSynchronization res;
77  for(auto& endpoints : hosts) {
78  Comm::Protocol::Mngr::Req::ClusterStatus::request(
79  clients,
80  endpoints,
81  [this, await=&res]
82  (const Comm::client::ConnQueue::ReqBase::Ptr&,
83  const int& error) noexcept {
84  err = error;
85  await->acknowledge();
86  }
87  );
88  res.wait();
89  if(err) {
90  SWC_PRINT << "code=" << err
91  << " msg=" << Error::get_text(err) << SWC_PRINT_CLOSE;
92  message.append(Error::get_text(err));
93  message.append("\n");
94  return error(message);
95  }
96  res.reset();
97  }
98  SWC_PRINT << "code=" << err
99  << " msg=" << Error::get_text(err) << SWC_PRINT_CLOSE;
100  return true;
101 }
102 
103 bool Mngr::managers_status(std::string& cmd) {
104  std::string message;
105  size_t at = cmd.find_first_of(" ");
106  cmd = cmd.substr(at+1);
107  client::SQL::Reader reader(cmd, message);
108  while(reader.found_space());
109 
110  client::Mngr::Hosts hosts;
111  if(reader.found_token("endpoint", 8)) {
112  while(reader.found_space());
113  reader.expect_eq();
114  if(reader.err)
115  return error(message);
116 
117  std::string host_or_ips;
118  reader.read(host_or_ips, " |");
119  if(reader.err)
120  return error(message);
121  if(host_or_ips.empty())
122  return error("Missing Endpoint");
123 
124  while(reader.found_space());
125  uint16_t port;
126  bool was_set = false;
127  if(*(reader.ptr-1) == '|' || reader.found_token("|", 1)) {
128  reader.read_uint16_t(port, was_set);
129  if(reader.err)
130  return error(message);
131  }
132  if(!was_set)
133  port = Env::Config::settings()->get_i16("swc.mngr.port");
134 
135  Config::Strings ips;
136  std::string host;
137  if(Comm::Resolver::is_ipv4_address(host_or_ips) ||
138  Comm::Resolver::is_ipv6_address(host_or_ips))
139  ips.push_back(host_or_ips);
140  else
141  host = host_or_ips;
142 
143  hosts.emplace_back(
144  Comm::Resolver::get_endpoints(
145  port, ips, host, {}, false
146  )
147  );
148  if(hosts.back().empty())
149  return error("Missing Endpoint");
150 
151  } else if(reader.found_token("all", 3)) {
152  auto groups = clients->managers.groups->get_groups();
153  for(auto& g : groups) {
154  auto tmp = g->get_hosts();
155  hosts.insert(hosts.cend(), tmp.cbegin(), tmp.cend());
156  }
157 
158  } else { // reader.found_token("schemas", 7)
159  hosts.emplace_back();
160  }
161 
162  Core::StateSynchronization res;
163  for(auto& endpoints : hosts) {
164  Comm::Protocol::Mngr::Req::ManagersStatus::request(
165  clients,
166  endpoints,
167  [endpoints, await=&res]
168  (const Comm::client::ConnQueue::ReqBase::Ptr& req, const int& error,
169  const Comm::Protocol::Mngr::Params::Report::RspManagersStatus& rsp) {
170  SWC_PRINT << "# by Manager(";
171  if(error) {
172  for(auto& p : endpoints)
173  SWC_LOG_OSTREAM << p << ',';
174  SWC_LOG_OSTREAM << ") - \033[31mERROR\033[00m: " << error
175  << "(" << Error::get_text(error) << ")";
176  } else {
177  SWC_LOG_OSTREAM << (req->queue? req->queue->get_endpoint_remote()
178  : Comm::EndPoint()) << ") ";
179  rsp.display(SWC_LOG_OSTREAM);
180  }
181  SWC_LOG_OSTREAM << SWC_PRINT_CLOSE;
182  await->acknowledge();
183  }
184  );
185  res.wait();
186  res.reset();
187  }
188  return true;
189 }
190 
191 bool Mngr::column_status(std::string& cmd) {
192  std::string message;
193 
194  size_t at = cmd.find_first_of(" ");
195  cmd = cmd.substr(at+1);
196  cid_t cid = DB::Schema::NO_CID;
197  client::SQL::Reader reader(cmd, message);
198  while(reader.found_space());
199 
200  if(reader.found_token("cid", 3)) {
201  reader.expect_eq();
202  if(!reader.err) {
203  bool was_set = false;
204  reader.read_uint64_t(cid, was_set);
205  }
206  if(reader.err)
207  return error(message);
208 
209  } else if(reader.found_token("name", 4)) {
210  reader.expect_eq();
211  if(reader.err)
212  return error(message);
213 
214  std::string col_name;
215  reader.read(col_name, " ");
216  if(col_name.empty())
217  return error("Missing Column Name");
218 
219  auto schema = reader.get_schema(clients, col_name);
220  if(reader.err)
221  return error(message);
222  cid = schema->cid;
223  } else {
224  bool token = false;
225  reader.expect_token("name", 4, token);
226  return error(message);
227  }
228 
229  Core::StateSynchronization res;
230  Comm::Protocol::Mngr::Req::ColumnStatus::request(
231  clients,
232  Comm::Protocol::Mngr::Params::Report::ReqColumnStatus(cid),
233  [this, await=&res]
234  (const Comm::client::ConnQueue::ReqBase::Ptr&, const int& error,
235  const Comm::Protocol::Mngr::Params::Report::RspColumnStatus& rsp) {
236  if(!(err = error)) {
237  SWC_PRINT << "";
238  rsp.display(SWC_LOG_OSTREAM);
239  SWC_LOG_OSTREAM << SWC_PRINT_CLOSE;
240  }
241  await->acknowledge();
242  }
243  );
244  res.wait();
245 
246  if(err) {
247  message.append(Error::get_text(err));
248  message.append("\n");
249  return error(message);
250  }
251  return true;
252 }
253 
254 bool Mngr::rangers_status(std::string& cmd) {
255  std::string message;
256 
257  size_t at = cmd.find_first_of(" ");
258  cmd = cmd.substr(at+1);
259  cid_t cid = DB::Schema::NO_CID;
260  client::SQL::Reader reader(cmd, message);
261  while(reader.found_space());
262 
263  if(reader.found_token("cid", 3)) {
264  reader.expect_eq();
265  if(!reader.err) {
266  bool was_set = false;
267  reader.read_uint64_t(cid, was_set);
268  }
269  if(reader.err)
270  return error(message);
271 
272  } else if(reader.found_token("name", 4)) {
273  reader.expect_eq();
274  if(reader.err)
275  return error(message);
276 
277  std::string col_name;
278  reader.read(col_name, " ");
279  if(col_name.empty())
280  return error("Missing Column Name");
281 
282  auto schema = reader.get_schema(clients, col_name);
283  if(reader.err)
284  return error(message);
285  cid = schema->cid;
286  }
287 
288  Core::StateSynchronization res;
289  Comm::Protocol::Mngr::Req::RangersStatus::request(
290  clients,
291  cid,
292  [this, await=&res]
293  (const Comm::client::ConnQueue::ReqBase::Ptr&, const int& error,
294  const Comm::Protocol::Mngr::Params::Report::RspRangersStatus& rsp) {
295  if(!(err = error)) {
296  SWC_PRINT << "";
297  rsp.display(SWC_LOG_OSTREAM);
298  SWC_LOG_OSTREAM << SWC_PRINT_CLOSE;
299  }
300  await->acknowledge();
301  }
302  );
303  res.wait();
304 
305  if(err) {
306  message.append(Error::get_text(err));
307  message.append("\n");
308  return error(message);
309  }
310  return true;
311 }
312 
313 
314 }}} // namespace Utils::shell
SWC::Comm::Protocol::Mngr::Req::RangersStatus::request
static SWC_CAN_INLINE void request(const SWC::client::Clients::Ptr &clients, cid_t cid, Cb_t &&cb, const uint32_t timeout=10000)
Definition: Report.h:153
SWC::Utils::shell::Interface::add_option
void add_option(const char *a_name, Core::Vector< std::string > &&a_desc, OptCall_t &&a_call, const char *a_re)
Definition: Shell.cc:249
SWC_LOG_OSTREAM
#define SWC_LOG_OSTREAM
Definition: Logger.h:44
SWC::client::SQL::Reader::expect_eq
void expect_eq()
Definition: Reader.cc:82
SWC::DB::SchemaPrimitives::NO_CID
static constexpr const cid_t NO_CID
Definition: Schema.h:25
SWC::client::SQL::Reader::expect_token
void expect_token(const char *token, uint8_t token_len, bool &found)
Definition: Reader.cc:122
SWC::Utils::shell::Mngr::cluster_status
bool cluster_status(std::string &)
Definition: Shell_Manager.cc:62
StateSynchronization.h
SWC::client::SQL::Reader::read_uint16_t
void read_uint16_t(uint16_t &value, bool &was_set)
Definition: Reader.cc:232
SWC::Comm::Protocol::Mngr::Params::Report::RspManagersStatus::display
void display(std::ostream &out, const std::string &offset="") const
Definition: Report.cc:262
SWC::Comm::Protocol::Mngr::Params::Report::RspRangersStatus
Definition: Report.h:113
SWC::Error::get_text
const char * get_text(const int err) noexcept
Definition: Error.cc:173
SWC::Env::Config::settings
static SWC::Config::Settings::Ptr & settings()
Definition: Settings.h:128
SWC::Comm::EndPoint
asio::ip::tcp::endpoint EndPoint
Definition: Resolver.h:19
SWC::Error::FAILED_EXPECTATION
@ FAILED_EXPECTATION
Definition: Error.h:43
Report.h
SWC::client::SQL::Reader::read_uint64_t
void read_uint64_t(uint64_t &value, bool &was_set, const char *stop=nullptr)
Definition: Reader.cc:274
SWC::Utils::shell::Interface::err
int err
Definition: Shell.h:54
SWC::Utils::shell::CLI
CLI
Definition: Shell.h:28
SWC::Utils::shell::Interface::error
bool error(const std::string &message)
Definition: Shell.cc:345
SWC_PRINT
#define SWC_PRINT
Definition: Logger.h:167
SWC_PRINT_CLOSE
#define SWC_PRINT_CLOSE
Definition: Logger.h:171
SWC::Utils::shell::Mngr::column_status
bool column_status(std::string &cmd)
Definition: Shell_Manager.cc:191
SWC::client::SQL::Reader
Definition: Reader.h:28
SWC::Utils::shell::Interface
Definition: Shell.h:41
SWC::Utils::shell::MANAGER
@ MANAGER
Definition: Shell.h:31
SWC::Comm::Protocol::Mngr::Req::ManagersStatus::request
static SWC_CAN_INLINE void request(const SWC::client::Clients::Ptr &clients, const EndPoints &endpoints, Cb_t &&cb, const uint32_t timeout=10000)
Definition: Report.h:195
SWC::client::SQL::Reader::found_space
bool found_space()
Definition: Reader.cc:37
SWC::Core::Vector::empty
constexpr SWC_CAN_INLINE bool empty() const noexcept
Definition: Vector.h:168
SWC::client::SQL::Reader::get_schema
DB::Schema::Ptr get_schema(const Clients::Ptr &clients, const std::string &col)
Definition: Reader.cc:137
SWC::Core::Vector::back
constexpr SWC_CAN_INLINE reference back() noexcept
Definition: Vector.h:254
SWC::Core::StateSynchronization::wait
SWC_CAN_INLINE void wait()
Definition: StateSynchronization.h:25
SWC::client::SQL::Reader::read
void read(std::string &buf, const char *stop=nullptr, bool keep_escape=false)
Definition: Reader.cc:185
SWC::Comm::Protocol::Mngr::Params::Report::RspManagersStatus
Definition: Report.h:180
SWC
The SWC-DB C++ namespace 'SWC'.
Definition: main.cc:12
SWC::Comm::Protocol::Mngr::Params::Report::RspRangersStatus::display
void display(std::ostream &out, const std::string &offset="") const
Definition: Report.cc:173
SWC::Utils::shell::Mngr::clients
client::Clients::Ptr clients
Definition: Shell_Manager.h:19
SWC::client::SQL::Reader::found_token
bool found_token(const char *token, uint8_t token_len)
Definition: Reader.cc:58
SWC::Utils::shell::Mngr::~Mngr
virtual ~Mngr()
Definition: Shell_Manager.cc:58
SWC::cid_t
uint64_t cid_t
Definition: Identifiers.h:16
SWC::Utils::shell::Mngr::rangers_status
bool rangers_status(std::string &cmd)
Definition: Shell_Manager.cc:254
SWC::Comm::Protocol::Mngr::Req::ClusterStatus::request
static SWC_CAN_INLINE void request(const SWC::client::Clients::Ptr &clients, const EndPoints &endpoints, Cb_t &&cb, const uint32_t timeout=10000)
Definition: Report.h:60
SWC::Comm::Protocol::Mngr::Params::Report::RspColumnStatus::display
void display(std::ostream &out, const std::string &offset="") const
Definition: Report.cc:92
Shell_Manager.h
SWC::Core::Vector< Comm::EndPoints >
SWC::Comm::Protocol::Mngr::Req::ColumnStatus::request
static SWC_CAN_INLINE void request(const SWC::client::Clients::Ptr &clients, cid_t cid, Cb_t &&cb, const uint32_t timeout=10000)
Definition: Report.h:103
SWC::Comm::Resolver::is_ipv4_address
bool is_ipv4_address(const std::string &str) noexcept
Definition: Resolver.cc:141
SWC::client::SQL::Reader::err
int err
Definition: Reader.h:125
SWC::Comm::Protocol::Mngr::Params::Report::ReqColumnStatus
Definition: Report.h:36
SWC::Core::Vector::cend
constexpr SWC_CAN_INLINE const_iterator cend() const noexcept
Definition: Vector.h:232
SWC::client::SQL::Reader::ptr
const char * ptr
Definition: Reader.h:123
SWC::Comm::Protocol::Mngr::Params::Report::RspColumnStatus
Definition: Report.h:59
SWC::Core::StateSynchronization::reset
SWC_CAN_INLINE void reset() noexcept
Definition: StateSynchronization.h:38
SWC::Core::StateSynchronization
Definition: StateSynchronization.h:16
SWC::Utils::shell::Mngr::managers_status
bool managers_status(std::string &cmd)
Definition: Shell_Manager.cc:103
SWC::Comm::client::ConnQueueReqBase::Ptr
std::shared_ptr< ConnQueueReqBase > Ptr
Definition: ClientConnQueue.h:25
SWC::Comm::Resolver::is_ipv6_address
bool is_ipv6_address(const std::string &str) noexcept
Definition: Resolver.cc:147
SWC::Comm::Resolver::get_endpoints
EndPoints get_endpoints(uint16_t defaul_port, const Config::Strings &addrs, const std::string &host, const Networks &nets, bool srv=false)
Definition: Resolver.cc:152
SWC::Core::Vector::push_back
SWC_CAN_INLINE void push_back(ArgsT &&... args)
Definition: Vector.h:331
Reader.h
SWC::Core::Vector::cbegin
constexpr SWC_CAN_INLINE const_iterator cbegin() const noexcept
Definition: Vector.h:216
SWC::Utils::shell::Mngr::Mngr
Mngr()
Definition: Shell_Manager.cc:16
SWC::Core::Vector::emplace_back
SWC_CAN_INLINE reference emplace_back(ArgsT &&... args)
Definition: Vector.h:349
SWC::Core::Vector::insert
SWC_CAN_INLINE iterator insert(size_type offset, ArgsT &&... args)
Definition: Vector.h:367