commands.c 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753
  1. // SPDX-License-Identifier: GPL-3.0-or-later
  2. #include "common.h"
  3. static uv_thread_t thread;
  4. static uv_loop_t* loop;
  5. static uv_async_t async;
  6. static struct completion completion;
  7. static uv_pipe_t server_pipe;
  8. char cmd_prefix_by_status[] = {
  9. CMD_PREFIX_INFO,
  10. CMD_PREFIX_ERROR,
  11. CMD_PREFIX_ERROR
  12. };
  13. static int command_server_initialized = 0;
  14. static int command_thread_error;
  15. static int command_thread_shutdown;
  16. static unsigned clients = 0;
  17. struct command_context {
  18. /* embedded client pipe structure at address 0 */
  19. uv_pipe_t client;
  20. uv_work_t work;
  21. uv_write_t write_req;
  22. cmd_t idx;
  23. char *args;
  24. char *message;
  25. cmd_status_t status;
  26. char command_string[MAX_COMMAND_LENGTH];
  27. unsigned command_string_size;
  28. };
  29. /* Forward declarations */
  30. static cmd_status_t cmd_help_execute(char *args, char **message);
  31. static cmd_status_t cmd_reload_health_execute(char *args, char **message);
  32. static cmd_status_t cmd_reopen_logs_execute(char *args, char **message);
  33. static cmd_status_t cmd_exit_execute(char *args, char **message);
  34. static cmd_status_t cmd_fatal_execute(char *args, char **message);
  35. static cmd_status_t cmd_reload_claiming_state_execute(char *args, char **message);
  36. static cmd_status_t cmd_reload_labels_execute(char *args, char **message);
  37. static cmd_status_t cmd_read_config_execute(char *args, char **message);
  38. static cmd_status_t cmd_write_config_execute(char *args, char **message);
  39. static cmd_status_t cmd_ping_execute(char *args, char **message);
  40. static cmd_status_t cmd_aclk_state(char *args, char **message);
  41. static cmd_status_t cmd_version(char *args, char **message);
  42. static cmd_status_t cmd_dumpconfig(char *args, char **message);
  43. static command_info_t command_info_array[] = {
  44. {"help", cmd_help_execute, CMD_TYPE_HIGH_PRIORITY}, // show help menu
  45. {"reload-health", cmd_reload_health_execute, CMD_TYPE_ORTHOGONAL}, // reload health configuration
  46. {"reopen-logs", cmd_reopen_logs_execute, CMD_TYPE_ORTHOGONAL}, // Close and reopen log files
  47. {"shutdown-agent", cmd_exit_execute, CMD_TYPE_EXCLUSIVE}, // exit cleanly
  48. {"fatal-agent", cmd_fatal_execute, CMD_TYPE_HIGH_PRIORITY}, // exit with fatal error
  49. {"reload-claiming-state", cmd_reload_claiming_state_execute, CMD_TYPE_ORTHOGONAL}, // reload claiming state
  50. {"reload-labels", cmd_reload_labels_execute, CMD_TYPE_ORTHOGONAL}, // reload the labels
  51. {"read-config", cmd_read_config_execute, CMD_TYPE_CONCURRENT},
  52. {"write-config", cmd_write_config_execute, CMD_TYPE_ORTHOGONAL},
  53. {"ping", cmd_ping_execute, CMD_TYPE_ORTHOGONAL},
  54. {"aclk-state", cmd_aclk_state, CMD_TYPE_ORTHOGONAL},
  55. {"version", cmd_version, CMD_TYPE_ORTHOGONAL},
  56. {"dumpconfig", cmd_dumpconfig, CMD_TYPE_ORTHOGONAL}
  57. };
  58. /* Mutexes for commands of type CMD_TYPE_ORTHOGONAL */
  59. static uv_mutex_t command_lock_array[CMD_TOTAL_COMMANDS];
  60. /* Commands of type CMD_TYPE_EXCLUSIVE are writers */
  61. static uv_rwlock_t exclusive_rwlock;
  62. /*
  63. * Locking order:
  64. * 1. exclusive_rwlock
  65. * 2. command_lock_array[]
  66. */
  67. /* Forward declarations */
  68. static void cmd_lock_exclusive(unsigned index);
  69. static void cmd_lock_orthogonal(unsigned index);
  70. static void cmd_lock_idempotent(unsigned index);
  71. static void cmd_lock_high_priority(unsigned index);
  72. static command_lock_t *cmd_lock_by_type[] = {
  73. cmd_lock_exclusive,
  74. cmd_lock_orthogonal,
  75. cmd_lock_idempotent,
  76. cmd_lock_high_priority
  77. };
  78. /* Forward declarations */
  79. static void cmd_unlock_exclusive(unsigned index);
  80. static void cmd_unlock_orthogonal(unsigned index);
  81. static void cmd_unlock_idempotent(unsigned index);
  82. static void cmd_unlock_high_priority(unsigned index);
  83. static command_lock_t *cmd_unlock_by_type[] = {
  84. cmd_unlock_exclusive,
  85. cmd_unlock_orthogonal,
  86. cmd_unlock_idempotent,
  87. cmd_unlock_high_priority
  88. };
  89. static cmd_status_t cmd_help_execute(char *args, char **message)
  90. {
  91. (void)args;
  92. *message = mallocz(MAX_COMMAND_LENGTH);
  93. strncpyz(*message,
  94. "\nThe commands are (arguments are in brackets):\n"
  95. "help\n"
  96. " Show this help menu.\n"
  97. "reload-health\n"
  98. " Reload health configuration.\n"
  99. "reload-labels\n"
  100. " Reload all labels.\n"
  101. "save-database\n"
  102. " Save internal DB to disk for memory mode save.\n"
  103. "reopen-logs\n"
  104. " Close and reopen log files.\n"
  105. "shutdown-agent\n"
  106. " Cleanup and exit the netdata agent.\n"
  107. "fatal-agent\n"
  108. " Log the state and halt the netdata agent.\n"
  109. "reload-claiming-state\n"
  110. " Reload agent claiming state from disk.\n"
  111. "ping\n"
  112. " Return with 'pong' if agent is alive.\n"
  113. "aclk-state [json]\n"
  114. " Returns current state of ACLK and Cloud connection. (optionally in json).\n"
  115. "dumpconfig\n"
  116. " Returns the current netdata.conf on stdout.\n"
  117. "version\n"
  118. " Returns the netdata version.\n",
  119. MAX_COMMAND_LENGTH - 1);
  120. return CMD_STATUS_SUCCESS;
  121. }
  122. static cmd_status_t cmd_reload_health_execute(char *args, char **message)
  123. {
  124. (void)args;
  125. (void)message;
  126. nd_log_limits_unlimited();
  127. netdata_log_info("COMMAND: Reloading HEALTH configuration.");
  128. health_reload();
  129. nd_log_limits_reset();
  130. return CMD_STATUS_SUCCESS;
  131. }
  132. static cmd_status_t cmd_reopen_logs_execute(char *args, char **message)
  133. {
  134. (void)args;
  135. (void)message;
  136. nd_log_limits_unlimited();
  137. nd_log_reopen_log_files();
  138. nd_log_limits_reset();
  139. return CMD_STATUS_SUCCESS;
  140. }
  141. static cmd_status_t cmd_exit_execute(char *args, char **message)
  142. {
  143. (void)args;
  144. (void)message;
  145. nd_log_limits_unlimited();
  146. netdata_log_info("COMMAND: Cleaning up to exit.");
  147. netdata_cleanup_and_exit(0, NULL, NULL, NULL);
  148. exit(0);
  149. return CMD_STATUS_SUCCESS;
  150. }
  151. static cmd_status_t cmd_fatal_execute(char *args, char **message)
  152. {
  153. (void)args;
  154. (void)message;
  155. fatal("COMMAND: netdata now exits.");
  156. return CMD_STATUS_SUCCESS;
  157. }
  158. static cmd_status_t cmd_reload_claiming_state_execute(char *args, char **message)
  159. {
  160. (void)args;
  161. (void)message;
  162. #if defined(DISABLE_CLOUD) || !defined(ENABLE_ACLK)
  163. netdata_log_info("The claiming feature has been explicitly disabled");
  164. *message = strdupz("This agent cannot be claimed, it was built without support for Cloud");
  165. return CMD_STATUS_FAILURE;
  166. #endif
  167. netdata_log_info("COMMAND: Reloading Agent Claiming configuration.");
  168. claim_reload_all();
  169. return CMD_STATUS_SUCCESS;
  170. }
  171. static cmd_status_t cmd_reload_labels_execute(char *args, char **message)
  172. {
  173. (void)args;
  174. netdata_log_info("COMMAND: reloading host labels.");
  175. reload_host_labels();
  176. BUFFER *wb = buffer_create(10, NULL);
  177. rrdlabels_log_to_buffer(localhost->rrdlabels, wb);
  178. (*message)=strdupz(buffer_tostring(wb));
  179. buffer_free(wb);
  180. return CMD_STATUS_SUCCESS;
  181. }
  182. static cmd_status_t cmd_read_config_execute(char *args, char **message)
  183. {
  184. size_t n = strlen(args);
  185. char *separator = strchr(args,'|');
  186. if (separator == NULL)
  187. return CMD_STATUS_FAILURE;
  188. char *separator2 = strchr(separator + 1,'|');
  189. if (separator2 == NULL)
  190. return CMD_STATUS_FAILURE;
  191. char *temp = callocz(n + 1, 1);
  192. strcpy(temp, args);
  193. size_t offset = separator - args;
  194. temp[offset] = 0;
  195. size_t offset2 = separator2 - args;
  196. temp[offset2] = 0;
  197. const char *conf_file = temp; /* "cloud" is cloud.conf, otherwise netdata.conf */
  198. struct config *tmp_config = strcmp(conf_file, "cloud") ? &netdata_config : &cloud_config;
  199. char *value = appconfig_get(tmp_config, temp + offset + 1, temp + offset2 + 1, NULL);
  200. if (value == NULL)
  201. {
  202. netdata_log_error("Cannot execute read-config conf_file=%s section=%s / key=%s because no value set",
  203. conf_file,
  204. temp + offset + 1,
  205. temp + offset2 + 1);
  206. freez(temp);
  207. return CMD_STATUS_FAILURE;
  208. }
  209. else
  210. {
  211. (*message) = strdupz(value);
  212. freez(temp);
  213. return CMD_STATUS_SUCCESS;
  214. }
  215. }
  216. static cmd_status_t cmd_write_config_execute(char *args, char **message)
  217. {
  218. UNUSED(message);
  219. netdata_log_info("write-config %s", args);
  220. size_t n = strlen(args);
  221. char *separator = strchr(args,'|');
  222. if (separator == NULL)
  223. return CMD_STATUS_FAILURE;
  224. char *separator2 = strchr(separator + 1,'|');
  225. if (separator2 == NULL)
  226. return CMD_STATUS_FAILURE;
  227. char *separator3 = strchr(separator2 + 1,'|');
  228. if (separator3 == NULL)
  229. return CMD_STATUS_FAILURE;
  230. char *temp = callocz(n + 1, 1);
  231. strcpy(temp, args);
  232. size_t offset = separator - args;
  233. temp[offset] = 0;
  234. size_t offset2 = separator2 - args;
  235. temp[offset2] = 0;
  236. size_t offset3 = separator3 - args;
  237. temp[offset3] = 0;
  238. const char *conf_file = temp; /* "cloud" is cloud.conf, otherwise netdata.conf */
  239. struct config *tmp_config = strcmp(conf_file, "cloud") ? &netdata_config : &cloud_config;
  240. appconfig_set(tmp_config, temp + offset + 1, temp + offset2 + 1, temp + offset3 + 1);
  241. netdata_log_info("write-config conf_file=%s section=%s key=%s value=%s",conf_file, temp + offset + 1, temp + offset2 + 1,
  242. temp + offset3 + 1);
  243. freez(temp);
  244. return CMD_STATUS_SUCCESS;
  245. }
  246. static cmd_status_t cmd_ping_execute(char *args, char **message)
  247. {
  248. (void)args;
  249. *message = strdupz("pong");
  250. return CMD_STATUS_SUCCESS;
  251. }
  252. static cmd_status_t cmd_aclk_state(char *args, char **message)
  253. {
  254. netdata_log_info("COMMAND: Reopening aclk/cloud state.");
  255. if (strstr(args, "json"))
  256. *message = aclk_state_json();
  257. else
  258. *message = aclk_state();
  259. return CMD_STATUS_SUCCESS;
  260. }
  261. static cmd_status_t cmd_version(char *args, char **message)
  262. {
  263. (void)args;
  264. char version[MAX_COMMAND_LENGTH];
  265. snprintfz(version, MAX_COMMAND_LENGTH -1, "%s %s", program_name, program_version);
  266. *message = strdupz(version);
  267. return CMD_STATUS_SUCCESS;
  268. }
  269. static cmd_status_t cmd_dumpconfig(char *args, char **message)
  270. {
  271. (void)args;
  272. BUFFER *wb = buffer_create(1024, NULL);
  273. config_generate(wb, 0);
  274. *message = strdupz(buffer_tostring(wb));
  275. buffer_free(wb);
  276. return CMD_STATUS_SUCCESS;
  277. }
  278. static void cmd_lock_exclusive(unsigned index)
  279. {
  280. (void)index;
  281. uv_rwlock_wrlock(&exclusive_rwlock);
  282. }
  283. static void cmd_lock_orthogonal(unsigned index)
  284. {
  285. uv_rwlock_rdlock(&exclusive_rwlock);
  286. uv_mutex_lock(&command_lock_array[index]);
  287. }
  288. static void cmd_lock_idempotent(unsigned index)
  289. {
  290. (void)index;
  291. uv_rwlock_rdlock(&exclusive_rwlock);
  292. }
  293. static void cmd_lock_high_priority(unsigned index)
  294. {
  295. (void)index;
  296. }
  297. static void cmd_unlock_exclusive(unsigned index)
  298. {
  299. (void)index;
  300. uv_rwlock_wrunlock(&exclusive_rwlock);
  301. }
  302. static void cmd_unlock_orthogonal(unsigned index)
  303. {
  304. uv_rwlock_rdunlock(&exclusive_rwlock);
  305. uv_mutex_unlock(&command_lock_array[index]);
  306. }
  307. static void cmd_unlock_idempotent(unsigned index)
  308. {
  309. (void)index;
  310. uv_rwlock_rdunlock(&exclusive_rwlock);
  311. }
  312. static void cmd_unlock_high_priority(unsigned index)
  313. {
  314. (void)index;
  315. }
  316. static void pipe_close_cb(uv_handle_t* handle)
  317. {
  318. /* Also frees command context */
  319. freez(handle);
  320. }
  321. static void pipe_write_cb(uv_write_t* req, int status)
  322. {
  323. (void)status;
  324. uv_pipe_t *client = req->data;
  325. uv_close((uv_handle_t *)client, pipe_close_cb);
  326. --clients;
  327. buffer_free(client->data);
  328. // netdata_log_info("Command Clients = %u", clients);
  329. }
  330. static inline void add_char_to_command_reply(BUFFER *reply_string, unsigned *reply_string_size, char character)
  331. {
  332. buffer_fast_charcat(reply_string, character);
  333. *reply_string_size +=1;
  334. }
  335. static inline void add_string_to_command_reply(BUFFER *reply_string, unsigned *reply_string_size, char *str)
  336. {
  337. unsigned len;
  338. len = strlen(str);
  339. buffer_fast_strcat(reply_string, str, len);
  340. *reply_string_size += len;
  341. }
  342. static void send_command_reply(struct command_context *cmd_ctx, cmd_status_t status, char *message)
  343. {
  344. int ret;
  345. BUFFER *reply_string = buffer_create(128, NULL);
  346. char exit_status_string[MAX_EXIT_STATUS_LENGTH + 1] = {'\0', };
  347. unsigned reply_string_size = 0;
  348. uv_buf_t write_buf;
  349. uv_stream_t *client = (uv_stream_t *)(uv_pipe_t *)cmd_ctx;
  350. snprintfz(exit_status_string, MAX_EXIT_STATUS_LENGTH, "%u", status);
  351. add_char_to_command_reply(reply_string, &reply_string_size, CMD_PREFIX_EXIT_CODE);
  352. add_string_to_command_reply(reply_string, &reply_string_size, exit_status_string);
  353. add_char_to_command_reply(reply_string, &reply_string_size, '\0');
  354. if (message) {
  355. add_char_to_command_reply(reply_string, &reply_string_size, cmd_prefix_by_status[status]);
  356. add_string_to_command_reply(reply_string, &reply_string_size, message);
  357. }
  358. cmd_ctx->write_req.data = client;
  359. client->data = reply_string;
  360. write_buf.base = reply_string->buffer;
  361. write_buf.len = reply_string_size;
  362. ret = uv_write(&cmd_ctx->write_req, (uv_stream_t *)client, &write_buf, 1, pipe_write_cb);
  363. if (ret) {
  364. netdata_log_error("uv_write(): %s", uv_strerror(ret));
  365. }
  366. }
  367. cmd_status_t execute_command(cmd_t idx, char *args, char **message)
  368. {
  369. cmd_status_t status;
  370. cmd_type_t type = command_info_array[idx].type;
  371. cmd_lock_by_type[type](idx);
  372. status = command_info_array[idx].func(args, message);
  373. cmd_unlock_by_type[type](idx);
  374. return status;
  375. }
  376. static void after_schedule_command(uv_work_t *req, int status)
  377. {
  378. struct command_context *cmd_ctx = req->data;
  379. (void)status;
  380. send_command_reply(cmd_ctx, cmd_ctx->status, cmd_ctx->message);
  381. if (cmd_ctx->message)
  382. freez(cmd_ctx->message);
  383. }
  384. static void schedule_command(uv_work_t *req)
  385. {
  386. register_libuv_worker_jobs();
  387. worker_is_busy(UV_EVENT_SCHEDULE_CMD);
  388. struct command_context *cmd_ctx = req->data;
  389. cmd_ctx->status = execute_command(cmd_ctx->idx, cmd_ctx->args, &cmd_ctx->message);
  390. worker_is_idle();
  391. }
  392. /* This will alter the state of the command_info_array.cmd_str
  393. */
  394. static void parse_commands(struct command_context *cmd_ctx)
  395. {
  396. char *message = NULL, *pos, *lstrip, *rstrip;
  397. cmd_t i;
  398. cmd_status_t status;
  399. status = CMD_STATUS_FAILURE;
  400. /* Skip white-space characters */
  401. for (pos = cmd_ctx->command_string ; isspace(*pos) && ('\0' != *pos) ; ++pos) ;
  402. for (i = 0 ; i < CMD_TOTAL_COMMANDS ; ++i) {
  403. if (!strncmp(pos, command_info_array[i].cmd_str, strlen(command_info_array[i].cmd_str))) {
  404. if (CMD_EXIT == i) {
  405. /* musl C does not like libuv workqueues calling exit() */
  406. execute_command(CMD_EXIT, NULL, NULL);
  407. }
  408. for (lstrip=pos + strlen(command_info_array[i].cmd_str); isspace(*lstrip) && ('\0' != *lstrip); ++lstrip) ;
  409. for (rstrip=lstrip+strlen(lstrip)-1; rstrip>lstrip && isspace(*rstrip); *(rstrip--) = 0 ) ;
  410. cmd_ctx->work.data = cmd_ctx;
  411. cmd_ctx->idx = i;
  412. cmd_ctx->args = lstrip;
  413. cmd_ctx->message = NULL;
  414. fatal_assert(0 == uv_queue_work(loop, &cmd_ctx->work, schedule_command, after_schedule_command));
  415. break;
  416. }
  417. }
  418. if (CMD_TOTAL_COMMANDS == i) {
  419. /* no command found */
  420. message = strdupz("Illegal command. Please type \"help\" for instructions.");
  421. send_command_reply(cmd_ctx, status, message);
  422. freez(message);
  423. }
  424. }
  425. static void pipe_read_cb(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf)
  426. {
  427. struct command_context *cmd_ctx = (struct command_context *)client;
  428. if (0 == nread) {
  429. netdata_log_info("%s: Zero bytes read by command pipe.", __func__);
  430. } else if (UV_EOF == nread) {
  431. netdata_log_info("EOF found in command pipe.");
  432. parse_commands(cmd_ctx);
  433. } else if (nread < 0) {
  434. netdata_log_error("%s: %s", __func__, uv_strerror(nread));
  435. }
  436. if (nread < 0) { /* stop stream due to EOF or error */
  437. (void)uv_read_stop((uv_stream_t *)client);
  438. } else if (nread) {
  439. size_t to_copy;
  440. to_copy = MIN((size_t) nread, MAX_COMMAND_LENGTH - 1 - cmd_ctx->command_string_size);
  441. memcpy(cmd_ctx->command_string + cmd_ctx->command_string_size, buf->base, to_copy);
  442. cmd_ctx->command_string_size += to_copy;
  443. cmd_ctx->command_string[cmd_ctx->command_string_size] = '\0';
  444. }
  445. if (buf && buf->len) {
  446. freez(buf->base);
  447. }
  448. if (nread < 0 && UV_EOF != nread) {
  449. uv_close((uv_handle_t *)client, pipe_close_cb);
  450. --clients;
  451. // netdata_log_info("Command Clients = %u", clients);
  452. }
  453. }
  454. static void alloc_cb(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
  455. {
  456. (void)handle;
  457. buf->base = mallocz(suggested_size);
  458. buf->len = suggested_size;
  459. }
  460. static void connection_cb(uv_stream_t *server, int status)
  461. {
  462. int ret;
  463. uv_pipe_t *client;
  464. struct command_context *cmd_ctx;
  465. fatal_assert(status == 0);
  466. /* combined allocation of client pipe and command context */
  467. cmd_ctx = mallocz(sizeof(*cmd_ctx));
  468. client = (uv_pipe_t *)cmd_ctx;
  469. ret = uv_pipe_init(server->loop, client, 1);
  470. if (ret) {
  471. netdata_log_error("uv_pipe_init(): %s", uv_strerror(ret));
  472. freez(cmd_ctx);
  473. return;
  474. }
  475. ret = uv_accept(server, (uv_stream_t *)client);
  476. if (ret) {
  477. netdata_log_error("uv_accept(): %s", uv_strerror(ret));
  478. uv_close((uv_handle_t *)client, pipe_close_cb);
  479. return;
  480. }
  481. ++clients;
  482. // netdata_log_info("Command Clients = %u", clients);
  483. /* Start parsing a new command */
  484. cmd_ctx->command_string_size = 0;
  485. cmd_ctx->command_string[0] = '\0';
  486. ret = uv_read_start((uv_stream_t*)client, alloc_cb, pipe_read_cb);
  487. if (ret) {
  488. netdata_log_error("uv_read_start(): %s", uv_strerror(ret));
  489. uv_close((uv_handle_t *)client, pipe_close_cb);
  490. --clients;
  491. // netdata_log_info("Command Clients = %u", clients);
  492. return;
  493. }
  494. }
  495. static void async_cb(uv_async_t *handle)
  496. {
  497. uv_stop(handle->loop);
  498. }
  499. static void command_thread(void *arg)
  500. {
  501. int ret;
  502. uv_fs_t req;
  503. (void) arg;
  504. loop = mallocz(sizeof(uv_loop_t));
  505. ret = uv_loop_init(loop);
  506. if (ret) {
  507. netdata_log_error("uv_loop_init(): %s", uv_strerror(ret));
  508. command_thread_error = ret;
  509. goto error_after_loop_init;
  510. }
  511. loop->data = NULL;
  512. ret = uv_async_init(loop, &async, async_cb);
  513. if (ret) {
  514. netdata_log_error("uv_async_init(): %s", uv_strerror(ret));
  515. command_thread_error = ret;
  516. goto error_after_async_init;
  517. }
  518. async.data = NULL;
  519. ret = uv_pipe_init(loop, &server_pipe, 0);
  520. if (ret) {
  521. netdata_log_error("uv_pipe_init(): %s", uv_strerror(ret));
  522. command_thread_error = ret;
  523. goto error_after_pipe_init;
  524. }
  525. const char *pipename = daemon_pipename();
  526. (void)uv_fs_unlink(loop, &req, pipename, NULL);
  527. uv_fs_req_cleanup(&req);
  528. ret = uv_pipe_bind(&server_pipe, pipename);
  529. if (ret) {
  530. netdata_log_error("uv_pipe_bind(): %s", uv_strerror(ret));
  531. command_thread_error = ret;
  532. goto error_after_pipe_bind;
  533. }
  534. ret = uv_listen((uv_stream_t *)&server_pipe, SOMAXCONN, connection_cb);
  535. if (ret) {
  536. /* Fallback to backlog of 1 */
  537. netdata_log_info("uv_listen() failed with backlog = %d, falling back to backlog = 1.", SOMAXCONN);
  538. ret = uv_listen((uv_stream_t *)&server_pipe, 1, connection_cb);
  539. }
  540. if (ret) {
  541. netdata_log_error("uv_listen(): %s", uv_strerror(ret));
  542. command_thread_error = ret;
  543. goto error_after_uv_listen;
  544. }
  545. command_thread_error = 0;
  546. command_thread_shutdown = 0;
  547. /* wake up initialization thread */
  548. completion_mark_complete(&completion);
  549. while (command_thread_shutdown == 0) {
  550. uv_run(loop, UV_RUN_DEFAULT);
  551. }
  552. /* cleanup operations of the event loop */
  553. netdata_log_info("Shutting down command event loop.");
  554. uv_close((uv_handle_t *)&async, NULL);
  555. uv_close((uv_handle_t*)&server_pipe, NULL);
  556. uv_run(loop, UV_RUN_DEFAULT); /* flush all libuv handles */
  557. netdata_log_info("Shutting down command loop complete.");
  558. fatal_assert(0 == uv_loop_close(loop));
  559. freez(loop);
  560. return;
  561. error_after_uv_listen:
  562. error_after_pipe_bind:
  563. uv_close((uv_handle_t*)&server_pipe, NULL);
  564. error_after_pipe_init:
  565. uv_close((uv_handle_t *)&async, NULL);
  566. error_after_async_init:
  567. uv_run(loop, UV_RUN_DEFAULT); /* flush all libuv handles */
  568. fatal_assert(0 == uv_loop_close(loop));
  569. error_after_loop_init:
  570. freez(loop);
  571. /* wake up initialization thread */
  572. completion_mark_complete(&completion);
  573. }
  574. static void sanity_check(void)
  575. {
  576. /* The size of command_info_array must be CMD_TOTAL_COMMANDS elements */
  577. BUILD_BUG_ON(CMD_TOTAL_COMMANDS != sizeof(command_info_array) / sizeof(command_info_array[0]));
  578. }
  579. void commands_init(void)
  580. {
  581. cmd_t i;
  582. int error;
  583. sanity_check();
  584. if (command_server_initialized)
  585. return;
  586. netdata_log_info("Initializing command server.");
  587. for (i = 0 ; i < CMD_TOTAL_COMMANDS ; ++i) {
  588. fatal_assert(0 == uv_mutex_init(&command_lock_array[i]));
  589. }
  590. fatal_assert(0 == uv_rwlock_init(&exclusive_rwlock));
  591. completion_init(&completion);
  592. error = uv_thread_create(&thread, command_thread, NULL);
  593. if (error) {
  594. netdata_log_error("uv_thread_create(): %s", uv_strerror(error));
  595. goto after_error;
  596. }
  597. /* wait for worker thread to initialize */
  598. completion_wait_for(&completion);
  599. completion_destroy(&completion);
  600. uv_thread_set_name_np(thread, "DAEMON_COMMAND");
  601. if (command_thread_error) {
  602. error = uv_thread_join(&thread);
  603. if (error) {
  604. netdata_log_error("uv_thread_create(): %s", uv_strerror(error));
  605. }
  606. goto after_error;
  607. }
  608. command_server_initialized = 1;
  609. return;
  610. after_error:
  611. netdata_log_error("Failed to initialize command server. The netdata cli tool will be unable to send commands.");
  612. }
  613. void commands_exit(void)
  614. {
  615. cmd_t i;
  616. if (!command_server_initialized)
  617. return;
  618. command_thread_shutdown = 1;
  619. netdata_log_info("Shutting down command server.");
  620. /* wake up event loop */
  621. fatal_assert(0 == uv_async_send(&async));
  622. fatal_assert(0 == uv_thread_join(&thread));
  623. for (i = 0 ; i < CMD_TOTAL_COMMANDS ; ++i) {
  624. uv_mutex_destroy(&command_lock_array[i]);
  625. }
  626. uv_rwlock_destroy(&exclusive_rwlock);
  627. netdata_log_info("Command server has stopped.");
  628. command_server_initialized = 0;
  629. }