cpg_test_agent.c 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741
  1. /*
  2. * Copyright (c) 2010 Red Hat, Inc.
  3. *
  4. * All rights reserved.
  5. *
  6. * Author: Angus Salkeld (asalkeld@redhat.com)
  7. *
  8. * This software licensed under BSD license, the text of which follows:
  9. *
  10. * Redistribution and use in source and binary forms, with or without
  11. * modification, are permitted provided that the following conditions are met:
  12. *
  13. * - Redistributions of source code must retain the above copyright notice,
  14. * this list of conditions and the following disclaimer.
  15. * - Redistributions in binary form must reproduce the above copyright notice,
  16. * this list of conditions and the following disclaimer in the documentation
  17. * and/or other materials provided with the distribution.
  18. * - Neither the name of the MontaVista Software, Inc. nor the names of its
  19. * contributors may be used to endorse or promote products derived from this
  20. * software without specific prior written permission.
  21. *
  22. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  23. * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  24. * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  25. * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  26. * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  27. * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  28. * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  29. * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  30. * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  31. * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  32. * THE POSSIBILITY OF SUCH DAMAGE.
  33. */
  34. #include <errno.h>
  35. #include <unistd.h>
  36. #include <stdio.h>
  37. #include <stdlib.h>
  38. #include <assert.h>
  39. #include <string.h>
  40. #include <sys/types.h>
  41. #include <sys/socket.h>
  42. #include <netinet/in.h>
  43. #include <arpa/inet.h>
  44. #include <netdb.h>
  45. #include <syslog.h>
  46. #include <poll.h>
  47. #include <unistd.h>
  48. #include <fcntl.h>
  49. #include <corosync/totem/coropoll.h>
  50. #include <corosync/list.h>
  51. #include <corosync/cpg.h>
  52. #include <corosync/cfg.h>
  53. #include "../../exec/crypto.h"
  54. #include "common_test_agent.h"
  55. typedef enum {
  56. MSG_OK,
  57. MSG_NODEID_ERR,
  58. MSG_PID_ERR,
  59. MSG_SEQ_ERR,
  60. MSG_SIZE_ERR,
  61. MSG_SHA1_ERR,
  62. } msg_status_t;
  63. typedef struct {
  64. uint32_t nodeid;
  65. pid_t pid;
  66. unsigned char sha1[20];
  67. uint32_t seq;
  68. size_t size;
  69. unsigned char payload[0];
  70. } msg_t;
  71. #define LOG_STR_SIZE 256
  72. typedef struct {
  73. char log[LOG_STR_SIZE];
  74. struct list_head list;
  75. } log_entry_t;
  76. static char big_and_buf[HOW_BIG_AND_BUF];
  77. static int32_t record_config_events_g = 0;
  78. static int32_t record_messages_g = 0;
  79. static cpg_handle_t cpg_handle = 0;
  80. static corosync_cfg_handle_t cfg_handle = 0;
  81. static int32_t cpg_fd = -1;
  82. static int32_t cfg_fd = -1;
  83. static struct list_head config_chg_log_head;
  84. static struct list_head msg_log_head;
  85. static pid_t my_pid;
  86. static uint32_t my_nodeid;
  87. static int32_t my_seq;
  88. static int32_t use_zcb = 0;
  89. static int32_t my_msgs_to_send;
  90. static int32_t total_stored_msgs = 0;
  91. static int32_t in_cnchg = 0;
  92. static int32_t pcmk_test = 0;
  93. static void send_some_more_messages (void * unused);
  94. static char* err_status_string (char * buf, size_t buf_len, msg_status_t status)
  95. {
  96. switch (status) {
  97. case MSG_OK:
  98. strncpy (buf, "OK", buf_len);
  99. break;
  100. case MSG_NODEID_ERR:
  101. strncpy (buf, "NODEID_ERR", buf_len);
  102. break;
  103. case MSG_PID_ERR:
  104. strncpy (buf, "PID_ERR", buf_len);
  105. break;
  106. case MSG_SEQ_ERR:
  107. strncpy (buf, "SEQ_ERR", buf_len);
  108. break;
  109. case MSG_SIZE_ERR:
  110. strncpy (buf, "SIZE_ERR", buf_len);
  111. break;
  112. case MSG_SHA1_ERR:
  113. strncpy (buf, "SHA1_ERR", buf_len);
  114. break;
  115. default:
  116. strncpy (buf, "UNKNOWN_ERR", buf_len);
  117. break;
  118. }
  119. if (buf_len > 0) {
  120. buf[buf_len - 1] = '\0';
  121. }
  122. return buf;
  123. }
  124. static void delivery_callback (
  125. cpg_handle_t handle,
  126. const struct cpg_name *groupName,
  127. uint32_t nodeid,
  128. uint32_t pid,
  129. void *msg,
  130. size_t msg_len)
  131. {
  132. log_entry_t *log_pt;
  133. msg_t *msg_pt = (msg_t*)msg;
  134. msg_status_t status = MSG_OK;
  135. char status_buf[20];
  136. unsigned char sha1_compare[20];
  137. hash_state sha1_hash;
  138. if (record_messages_g == 0) {
  139. return;
  140. }
  141. msg_pt->seq = my_seq;
  142. my_seq++;
  143. if (nodeid != msg_pt->nodeid) {
  144. status = MSG_NODEID_ERR;
  145. }
  146. if (pid != msg_pt->pid) {
  147. status = MSG_PID_ERR;
  148. }
  149. if (msg_len != msg_pt->size) {
  150. status = MSG_SIZE_ERR;
  151. }
  152. sha1_init (&sha1_hash);
  153. sha1_process (&sha1_hash, msg_pt->payload, (msg_pt->size - sizeof (msg_t)));
  154. sha1_done (&sha1_hash, sha1_compare);
  155. if (memcmp (sha1_compare, msg_pt->sha1, 20) != 0) {
  156. syslog (LOG_ERR, "%s(); msg seq:%d; incorrect hash",
  157. __func__, msg_pt->seq);
  158. status = MSG_SHA1_ERR;
  159. }
  160. log_pt = malloc (sizeof(log_entry_t));
  161. list_init (&log_pt->list);
  162. snprintf (log_pt->log, LOG_STR_SIZE, "%d:%d:%d:%s;",
  163. msg_pt->nodeid, msg_pt->pid, msg_pt->seq,
  164. err_status_string (status_buf, 20, status));
  165. list_add_tail (&log_pt->list, &msg_log_head);
  166. total_stored_msgs++;
  167. // if ((total_stored_msgs % 100) == 0) {
  168. // syslog (LOG_INFO, "%s(); %d", __func__, total_stored_msgs);
  169. // }
  170. }
  171. static void config_change_callback (
  172. cpg_handle_t handle,
  173. const struct cpg_name *groupName,
  174. const struct cpg_address *member_list, size_t member_list_entries,
  175. const struct cpg_address *left_list, size_t left_list_entries,
  176. const struct cpg_address *joined_list, size_t joined_list_entries)
  177. {
  178. int i;
  179. log_entry_t *log_pt;
  180. /* group_name,ip,pid,join|leave */
  181. for (i = 0; i < left_list_entries; i++) {
  182. syslog (LOG_INFO, "%s(%d) %d:%s left", __func__, record_config_events_g,
  183. left_list[i].nodeid, groupName->value);
  184. if (record_config_events_g > 0) {
  185. log_pt = malloc (sizeof(log_entry_t));
  186. list_init (&log_pt->list);
  187. snprintf (log_pt->log, LOG_STR_SIZE, "%s,%d,%d,left",
  188. groupName->value, left_list[i].nodeid,left_list[i].pid);
  189. list_add_tail(&log_pt->list, &config_chg_log_head);
  190. }
  191. }
  192. for (i = 0; i < joined_list_entries; i++) {
  193. syslog (LOG_INFO, "%s(%d) %d:%s joined", __func__, record_config_events_g,
  194. left_list[i].nodeid, groupName->value);
  195. if (record_config_events_g > 0) {
  196. log_pt = malloc (sizeof(log_entry_t));
  197. list_init (&log_pt->list);
  198. snprintf (log_pt->log, LOG_STR_SIZE, "%s,%d,%d,join",
  199. groupName->value, joined_list[i].nodeid,joined_list[i].pid);
  200. list_add_tail (&log_pt->list, &config_chg_log_head);
  201. }
  202. }
  203. if (pcmk_test == 1) {
  204. in_cnchg = 1;
  205. send_some_more_messages (NULL);
  206. in_cnchg = 0;
  207. }
  208. }
  209. static void my_shutdown_callback (corosync_cfg_handle_t handle,
  210. corosync_cfg_shutdown_flags_t flags)
  211. {
  212. syslog (LOG_CRIT, "%s flags:%d", __func__, flags);
  213. if (flags & COROSYNC_CFG_SHUTDOWN_FLAG_REQUEST) {
  214. corosync_cfg_replyto_shutdown (cfg_handle, COROSYNC_CFG_SHUTDOWN_FLAG_YES);
  215. }
  216. }
  217. static corosync_cfg_callbacks_t cfg_callbacks = {
  218. .corosync_cfg_shutdown_callback = my_shutdown_callback,
  219. .corosync_cfg_state_track_callback = NULL,
  220. };
  221. static cpg_callbacks_t callbacks = {
  222. .cpg_deliver_fn = delivery_callback,
  223. .cpg_confchg_fn = config_change_callback,
  224. };
  225. static void record_messages (void)
  226. {
  227. record_messages_g = 1;
  228. syslog (LOG_DEBUG,"%s() record:%d", __func__, record_messages_g);
  229. }
  230. static void record_config_events (int sock)
  231. {
  232. char response[100];
  233. record_config_events_g = 1;
  234. syslog (LOG_INFO, "%s() record:%d", __func__, record_config_events_g);
  235. snprintf (response, 100, "%s", OK_STR);
  236. send (sock, response, strlen (response), 0);
  237. }
  238. static void read_config_event (int sock)
  239. {
  240. const char *empty = "None";
  241. struct list_head * list = config_chg_log_head.next;
  242. log_entry_t *entry;
  243. if (list != &config_chg_log_head) {
  244. entry = list_entry (list, log_entry_t, list);
  245. send (sock, entry->log, strlen (entry->log), 0);
  246. list_del (&entry->list);
  247. free (entry);
  248. } else {
  249. syslog (LOG_DEBUG,"%s() no events in list", __func__);
  250. send (sock, empty, strlen (empty), 0);
  251. }
  252. }
  253. static void read_messages (int sock, char* atmost_str)
  254. {
  255. struct list_head * list;
  256. log_entry_t *entry;
  257. int atmost = atoi (atmost_str);
  258. int packed = 0;
  259. if (atmost == 0)
  260. atmost = 1;
  261. if (atmost > (HOW_BIG_AND_BUF / LOG_STR_SIZE))
  262. atmost = (HOW_BIG_AND_BUF / LOG_STR_SIZE);
  263. syslog (LOG_DEBUG, "%s() atmost %d; total_stored_msgs:%d",
  264. __func__, atmost, total_stored_msgs);
  265. big_and_buf[0] = '\0';
  266. for (list = msg_log_head.next;
  267. (!list_empty (&msg_log_head) && packed < atmost); ) {
  268. entry = list_entry (list, log_entry_t, list);
  269. strcat (big_and_buf, entry->log);
  270. packed++;
  271. list = list->next;
  272. list_del (&entry->list);
  273. free (entry);
  274. total_stored_msgs--;
  275. }
  276. syslog (LOG_DEBUG, "%s() sending %d; total_stored_msgs:%d; len:%d",
  277. __func__, packed, total_stored_msgs, (int)strlen (big_and_buf));
  278. if (packed == 0) {
  279. strcpy (big_and_buf, "None");
  280. }
  281. send (sock, big_and_buf, strlen (big_and_buf), 0);
  282. }
  283. static poll_timer_handle more_messages_timer_handle;
  284. static void send_some_more_messages_later (void)
  285. {
  286. cpg_dispatch (cpg_handle, CS_DISPATCH_ALL);
  287. poll_timer_add (
  288. ta_poll_handle_get(),
  289. 300, NULL,
  290. send_some_more_messages,
  291. &more_messages_timer_handle);
  292. }
  293. static void send_some_more_messages_zcb (void)
  294. {
  295. msg_t *my_msg;
  296. int i;
  297. int send_now;
  298. size_t payload_size;
  299. size_t total_size;
  300. hash_state sha1_hash;
  301. cs_error_t res;
  302. cpg_flow_control_state_t fc_state;
  303. void *zcb_buffer;
  304. if (cpg_fd < 0)
  305. return;
  306. send_now = my_msgs_to_send;
  307. payload_size = (rand() % 100000);
  308. total_size = payload_size + sizeof (msg_t);
  309. cpg_zcb_alloc (cpg_handle, total_size, &zcb_buffer);
  310. my_msg = (msg_t*)zcb_buffer;
  311. //syslog (LOG_DEBUG,"%s() send_now:%d", __func__, send_now);
  312. my_msg->pid = my_pid;
  313. my_msg->nodeid = my_nodeid;
  314. my_msg->size = sizeof (msg_t) + payload_size;
  315. my_msg->seq = 0;
  316. for (i = 0; i < payload_size; i++) {
  317. my_msg->payload[i] = i;
  318. }
  319. sha1_init (&sha1_hash);
  320. sha1_process (&sha1_hash, my_msg->payload, payload_size);
  321. sha1_done (&sha1_hash, my_msg->sha1);
  322. for (i = 0; i < send_now; i++) {
  323. res = cpg_flow_control_state_get (cpg_handle, &fc_state);
  324. if (res == CS_OK && fc_state == CPG_FLOW_CONTROL_ENABLED) {
  325. /* lets do this later */
  326. send_some_more_messages_later ();
  327. syslog (LOG_INFO, "%s() flow control enabled.", __func__);
  328. goto free_buffer;
  329. }
  330. res = cpg_zcb_mcast_joined (cpg_handle, CPG_TYPE_AGREED, zcb_buffer, total_size);
  331. if (res == CS_ERR_TRY_AGAIN) {
  332. /* lets do this later */
  333. send_some_more_messages_later ();
  334. // if (i > 0) {
  335. // syslog (LOG_INFO, "%s() TRY_AGAIN %d to send.",
  336. // __func__, my_msgs_to_send);
  337. // }
  338. goto free_buffer;
  339. } else if (res != CS_OK) {
  340. syslog (LOG_ERR, "%s() -> cpg_mcast_joined error:%d, exiting.",
  341. __func__, res);
  342. exit (-2);
  343. }
  344. my_msgs_to_send--;
  345. }
  346. free_buffer:
  347. cpg_zcb_free (cpg_handle, zcb_buffer);
  348. }
  349. #define cs_repeat(counter, max, code) do { \
  350. code; \
  351. if(res == CS_ERR_TRY_AGAIN) { \
  352. counter++; \
  353. sleep(counter); \
  354. } \
  355. } while(res == CS_ERR_TRY_AGAIN && counter < max)
  356. static unsigned char buffer[200000];
  357. static void send_some_more_messages_normal (void)
  358. {
  359. msg_t my_msg;
  360. struct iovec iov[2];
  361. int i;
  362. int send_now;
  363. size_t payload_size;
  364. hash_state sha1_hash;
  365. cs_error_t res;
  366. cpg_flow_control_state_t fc_state;
  367. int retries = 0;
  368. time_t before;
  369. if (cpg_fd < 0)
  370. return;
  371. send_now = my_msgs_to_send;
  372. //syslog (LOG_DEBUG,"%s() send_now:%d", __func__, send_now);
  373. my_msg.pid = my_pid;
  374. my_msg.nodeid = my_nodeid;
  375. payload_size = (rand() % 100000);
  376. my_msg.size = sizeof (msg_t) + payload_size;
  377. my_msg.seq = 0;
  378. for (i = 0; i < payload_size; i++) {
  379. buffer[i] = i;
  380. }
  381. sha1_init (&sha1_hash);
  382. sha1_process (&sha1_hash, buffer, payload_size);
  383. sha1_done (&sha1_hash, my_msg.sha1);
  384. iov[0].iov_len = sizeof (msg_t);
  385. iov[0].iov_base = &my_msg;
  386. iov[1].iov_len = payload_size;
  387. iov[1].iov_base = buffer;
  388. for (i = 0; i < send_now; i++) {
  389. if (in_cnchg && pcmk_test) {
  390. retries = 0;
  391. before = time(NULL);
  392. cs_repeat(retries, 30, res = cpg_mcast_joined(cpg_handle, CPG_TYPE_AGREED, iov, 2));
  393. if (retries > 20) {
  394. syslog (LOG_ERR, "%s() -> cs_repeat: blocked for :%lu secs.",
  395. __func__, (unsigned long)(time(NULL) - before));
  396. }
  397. if (res != CS_OK) {
  398. syslog (LOG_ERR, "%s() -> cpg_mcast_joined error:%d.",
  399. __func__, res);
  400. return;
  401. }
  402. } else {
  403. res = cpg_flow_control_state_get (cpg_handle, &fc_state);
  404. if (res == CS_OK && fc_state == CPG_FLOW_CONTROL_ENABLED) {
  405. /* lets do this later */
  406. send_some_more_messages_later ();
  407. syslog (LOG_INFO, "%s() flow control enabled.", __func__);
  408. return;
  409. }
  410. res = cpg_mcast_joined (cpg_handle, CPG_TYPE_AGREED, iov, 2);
  411. if (res == CS_ERR_TRY_AGAIN) {
  412. /* lets do this later */
  413. send_some_more_messages_later ();
  414. if (i > 0) {
  415. syslog (LOG_INFO, "%s() TRY_AGAIN %d to send.",
  416. __func__, my_msgs_to_send);
  417. }
  418. return;
  419. } else if (res != CS_OK) {
  420. syslog (LOG_ERR, "%s() -> cpg_mcast_joined error:%d, exiting.",
  421. __func__, res);
  422. exit (-2);
  423. }
  424. }
  425. my_msgs_to_send--;
  426. }
  427. }
  428. static void send_some_more_messages (void * unused)
  429. {
  430. if (use_zcb) {
  431. send_some_more_messages_zcb ();
  432. } else {
  433. send_some_more_messages_normal ();
  434. }
  435. }
  436. static void msg_blaster (int sock, char* num_to_send_str)
  437. {
  438. my_msgs_to_send = atoi (num_to_send_str);
  439. my_seq = 1;
  440. my_pid = getpid();
  441. use_zcb = 0;
  442. total_stored_msgs = 0;
  443. cpg_local_get (cpg_handle, &my_nodeid);
  444. /* control the limits */
  445. if (my_msgs_to_send <= 0)
  446. my_msgs_to_send = 1;
  447. if (my_msgs_to_send > 10000)
  448. my_msgs_to_send = 10000;
  449. send_some_more_messages_normal ();
  450. }
  451. static void context_test (int sock)
  452. {
  453. char response[100];
  454. char *cmp;
  455. cpg_context_set (cpg_handle, response);
  456. cpg_context_get (cpg_handle, (void**)&cmp);
  457. if (response != cmp) {
  458. snprintf (response, 100, "%s", FAIL_STR);
  459. }
  460. else {
  461. snprintf (response, 100, "%s", OK_STR);
  462. }
  463. send (sock, response, strlen (response), 0);
  464. }
  465. static void msg_blaster_zcb (int sock, char* num_to_send_str)
  466. {
  467. my_msgs_to_send = atoi (num_to_send_str);
  468. my_seq = 1;
  469. my_pid = getpid();
  470. use_zcb = 1;
  471. total_stored_msgs = 0;
  472. cpg_local_get (cpg_handle, &my_nodeid);
  473. /* control the limits */
  474. if (my_msgs_to_send <= 0)
  475. my_msgs_to_send = 1;
  476. if (my_msgs_to_send > 10000)
  477. my_msgs_to_send = 10000;
  478. send_some_more_messages_zcb ();
  479. }
  480. static corosync_cfg_state_notification_t notification_buffer;
  481. static int cfg_dispatch_wrapper_fn (hdb_handle_t handle,
  482. int fd,
  483. int revents,
  484. void *data)
  485. {
  486. cs_error_t error;
  487. if (revents & POLLHUP || revents & POLLERR) {
  488. syslog (LOG_ERR, "%s() got POLLHUP disconnecting from CFG", __func__);
  489. poll_dispatch_delete (ta_poll_handle_get(), cfg_fd);
  490. close (cfg_fd);
  491. cfg_fd = -1;
  492. return -1;
  493. }
  494. error = corosync_cfg_dispatch (cfg_handle, CS_DISPATCH_ALL);
  495. if (error == CS_ERR_LIBRARY) {
  496. syslog (LOG_ERR, "%s() got LIB error disconnecting from CFG.", __func__);
  497. poll_dispatch_delete (ta_poll_handle_get(), cfg_fd);
  498. close (cfg_fd);
  499. cfg_fd = -1;
  500. return -1;
  501. }
  502. return 0;
  503. }
  504. static int cpg_dispatch_wrapper_fn (hdb_handle_t handle,
  505. int fd,
  506. int revents,
  507. void *data)
  508. {
  509. cs_error_t error;
  510. if (revents & POLLHUP || revents & POLLERR) {
  511. syslog (LOG_ERR, "%s() got POLLHUP disconnecting from CPG", __func__);
  512. poll_dispatch_delete (ta_poll_handle_get(), cpg_fd);
  513. close (cpg_fd);
  514. cpg_fd = -1;
  515. return -1;
  516. }
  517. error = cpg_dispatch (cpg_handle, CS_DISPATCH_ALL);
  518. if (error == CS_ERR_LIBRARY) {
  519. syslog (LOG_ERR, "%s() got LIB error disconnecting from CPG", __func__);
  520. poll_dispatch_delete (ta_poll_handle_get(), cpg_fd);
  521. close (cpg_fd);
  522. cpg_fd = -1;
  523. return -1;
  524. }
  525. return 0;
  526. }
  527. static void do_command (int sock, char* func, char*args[], int num_args)
  528. {
  529. int result;
  530. char response[100];
  531. struct cpg_name group_name;
  532. if (parse_debug)
  533. syslog (LOG_DEBUG,"RPC:%s() called.", func);
  534. if (strcmp ("cpg_mcast_joined",func) == 0) {
  535. struct iovec iov[5];
  536. int a;
  537. for (a = 0; a < num_args; a++) {
  538. iov[a].iov_base = args[a];
  539. iov[a].iov_len = strlen(args[a])+1;
  540. }
  541. cpg_mcast_joined (cpg_handle, CPG_TYPE_AGREED, iov, num_args);
  542. } else if (strcmp ("cpg_join",func) == 0) {
  543. strcpy (group_name.value, args[0]);
  544. group_name.length = strlen(args[0]);
  545. result = cpg_join (cpg_handle, &group_name);
  546. if (result != CS_OK) {
  547. syslog (LOG_ERR,
  548. "Could not join process group, error %d\n", result);
  549. exit (1);
  550. }
  551. syslog (LOG_INFO, "called cpg_join()!");
  552. } else if (strcmp ("cpg_leave",func) == 0) {
  553. strcpy (group_name.value, args[0]);
  554. group_name.length = strlen(args[0]);
  555. result = cpg_leave (cpg_handle, &group_name);
  556. if (result != CS_OK) {
  557. syslog (LOG_ERR,
  558. "Could not leave process group, error %d\n", result);
  559. exit (1);
  560. }
  561. syslog (LOG_INFO, "called cpg_leave()!");
  562. } else if (strcmp ("cpg_initialize",func) == 0) {
  563. int retry_count = 0;
  564. result = cpg_initialize (&cpg_handle, &callbacks);
  565. while (result != CS_OK) {
  566. syslog (LOG_ERR,
  567. "cpg_initialize error %d (attempt %d)\n",
  568. result, retry_count);
  569. if (retry_count >= 3) {
  570. exit (1);
  571. }
  572. sleep(1);
  573. retry_count++;
  574. result = cpg_initialize (&cpg_handle, &callbacks);
  575. }
  576. cpg_fd_get (cpg_handle, &cpg_fd);
  577. poll_dispatch_add (ta_poll_handle_get(), cpg_fd, POLLIN|POLLNVAL, NULL, cpg_dispatch_wrapper_fn);
  578. } else if (strcmp ("cpg_local_get", func) == 0) {
  579. unsigned int local_nodeid;
  580. cpg_local_get (cpg_handle, &local_nodeid);
  581. snprintf (response, 100, "%u",local_nodeid);
  582. send (sock, response, strlen (response), 0);
  583. } else if (strcmp ("cpg_finalize", func) == 0) {
  584. cpg_finalize (cpg_handle);
  585. poll_dispatch_delete (ta_poll_handle_get(), cpg_fd);
  586. cpg_fd = -1;
  587. } else if (strcmp ("record_config_events", func) == 0) {
  588. record_config_events (sock);
  589. } else if (strcmp ("record_messages", func) == 0) {
  590. record_messages ();
  591. } else if (strcmp ("read_config_event", func) == 0) {
  592. read_config_event (sock);
  593. } else if (strcmp ("read_messages", func) == 0) {
  594. read_messages (sock, args[0]);
  595. } else if (strcmp ("msg_blaster_zcb", func) == 0) {
  596. msg_blaster_zcb (sock, args[0]);
  597. } else if (strcmp ("pcmk_test", func) == 0) {
  598. pcmk_test = 1;
  599. } else if (strcmp ("msg_blaster", func) == 0) {
  600. msg_blaster (sock, args[0]);
  601. } else if (strcmp ("context_test", func) == 0) {
  602. context_test (sock);
  603. } else if (strcmp ("are_you_ok_dude", func) == 0) {
  604. snprintf (response, 100, "%s", OK_STR);
  605. send (sock, response, strlen (response), 0);
  606. } else if (strcmp ("cfg_shutdown", func) == 0) {
  607. corosync_cfg_try_shutdown (cfg_handle, COROSYNC_CFG_SHUTDOWN_FLAG_REQUEST);
  608. } else if (strcmp ("cfg_initialize",func) == 0) {
  609. int retry_count = 0;
  610. syslog (LOG_INFO,"%s %s() called!", __func__, func);
  611. result = corosync_cfg_initialize (&cfg_handle, &cfg_callbacks);
  612. while (result != CS_OK) {
  613. syslog (LOG_ERR,
  614. "cfg_initialize error %d (attempt %d)\n",
  615. result, retry_count);
  616. if (retry_count >= 3) {
  617. exit (1);
  618. }
  619. sleep(1);
  620. retry_count++;
  621. result = corosync_cfg_initialize (&cfg_handle, &cfg_callbacks);
  622. }
  623. corosync_cfg_fd_get (cfg_handle, &cfg_fd);
  624. corosync_cfg_state_track (cfg_handle, 0, &notification_buffer);
  625. poll_dispatch_add (ta_poll_handle_get(), cfg_fd, POLLIN|POLLNVAL, NULL, cfg_dispatch_wrapper_fn);
  626. } else {
  627. syslog (LOG_ERR,"%s RPC:%s not supported!", __func__, func);
  628. }
  629. }
  630. int main (int argc, char *argv[])
  631. {
  632. openlog (NULL, LOG_CONS|LOG_PID, LOG_DAEMON);
  633. list_init (&msg_log_head);
  634. list_init (&config_chg_log_head);
  635. return test_agent_run (9034, do_command);
  636. }