12 #include <sys/socket.h> 13 #include <netinet/in.h> 14 #include <arpa/inet.h> 20 #include <sys/utsname.h> 22 #include <qb/qbipcc.h> 23 #include <qb/qbutil.h> 25 #include <corosync/corodefs.h> 26 #include <corosync/corotypes.h> 27 #include <corosync/hdb.h> 28 #include <corosync/cpg.h> 36 static bool cpg_evicted = FALSE;
39 #define cs_repeat(counter, max, code) do { \ 41 if(rc == CS_ERR_TRY_AGAIN || rc == CS_ERR_QUEUE_FULL) { \ 43 crm_debug("Retrying operation after %ds", counter); \ 48 } while(counter < max) 54 if (cluster->cpg_handle) {
56 cpg_leave(cluster->cpg_handle, &cluster->group);
57 cpg_finalize(cluster->cpg_handle);
58 cluster->cpg_handle = 0;
67 cs_error_t rc = CS_OK;
70 cpg_handle_t local_handle = handle;
71 cpg_callbacks_t cb = { };
78 if(local_nodeid != 0) {
85 get_ais_details(&local_nodeid, NULL);
92 cs_repeat(retries, 5, rc = cpg_initialize(&local_handle, &cb));
94 crm_err(
"Could not connect to the CPG API (rc=%d)", rc);
98 rc = cpg_fd_get(local_handle, &fd);
100 crm_err(
"Could not obtain the CPG API connection (rc=%d)", rc);
106 &found_uid, &found_gid))) {
107 crm_err(
"CPG provider is not authentic:" 108 " process %lld (uid: %lld, gid: %lld)",
110 (
long long) found_uid, (
long long) found_gid);
113 crm_err(
"Could not verify authenticity of CPG provider: %s (%d)",
122 cs_repeat(retries, 5, rc = cpg_local_get(local_handle, &local_nodeid));
126 crm_err(
"Could not get local node id from the CPG API: %s (%d)", ais_error2text(rc), rc);
132 cpg_finalize(local_handle);
134 crm_debug(
"Local nodeid is %u", local_nodeid);
142 static ssize_t crm_cs_flush(gpointer
data);
145 crm_cs_flush_cb(gpointer
data)
152 #define CS_SEND_MAX 200 154 crm_cs_flush(gpointer
data)
159 static unsigned int last_sent = 0;
160 cpg_handle_t *handle = (cpg_handle_t *)data;
168 if ((queue_len % 1000) == 0 && queue_len > 1) {
169 crm_err(
"CPG queue has grown to %d", queue_len);
172 crm_warn(
"CPG queue has grown to %d", queue_len);
185 rc = cpg_mcast_joined(*handle, CPG_TYPE_AGREED, iov, 1);
194 (
unsigned long long) iov->iov_len);
203 crm_info(
"Sent %d CPG messages (%d remaining, last=%u): %s (%lld)",
204 sent, queue_len, last_sent, ais_error2text(rc),
207 crm_trace(
"Sent %d CPG messages (%d remaining, last=%u): %s (%lld)",
208 sent, queue_len, last_sent, ais_error2text(rc),
216 delay_ms = QB_MIN(1000,
CS_SEND_MAX + (10 * queue_len));
227 static unsigned int queued = 0;
230 crm_trace(
"Queueing CPG message %u (%llu bytes)",
231 queued, (
unsigned long long) iov->iov_len);
238 pcmk_cpg_dispatch(gpointer user_data)
243 rc = cpg_dispatch(cluster->cpg_handle, CS_DISPATCH_ONE);
245 crm_err(
"Connection to the CPG API failed: %s (%d)", ais_error2text(rc), rc);
246 cluster->cpg_handle = 0;
249 }
else if(cpg_evicted) {
250 crm_err(
"Evicted from CPG membership");
271 crm_err(
"Nodeid mismatch from %d.%d: claimed nodeid=%u", nodeid, pid, msg->
sender.
id);
274 }
else if (msg->
host.
id != 0 && (local_nodeid != msg->
host.
id)) {
289 crm_err(
"Peer with nodeid=%u is unknown", nodeid);
291 }
else if (peer->
uname == NULL) {
292 crm_err(
"No uname for peer with nodeid=%u", nodeid);
295 crm_notice(
"Fixing uname for peer with nodeid=%u", nodeid);
303 crm_trace(
"Got new%s message (size=%d, %d, %d)",
308 *kind = msg->header.
id;
316 char *uncompressed = NULL;
317 unsigned int new_size = msg->
size + 1;
324 uncompressed = calloc(1, new_size);
325 rc = BZ2_bzBuffToBuffDecompress(uncompressed, &new_size, msg->
data, msg->
compressed_size, 1, 0);
328 crm_err(
"Decompression failed: %d", rc);
342 char *pid_s = crm_getpid_s();
349 data = strdup(msg->
data);
360 crm_info(
"Removing peer %s/%u", data,
id);
375 crm_err(
"Invalid message (id=%d, dest=%s:%s, from=%s:%s.%d):" 376 " min=%d, total=%d, size=%d, bz2_size=%d",
386 static int cmp_member_list_nodeid(
const void *first,
389 const struct cpg_address *
const a = *((
const struct cpg_address **) first),
390 *
const b = *((
const struct cpg_address **) second);
391 if (a->nodeid < b->nodeid) {
393 }
else if (a->nodeid > b->nodeid) {
401 cpgreason2str(cpg_reason_t reason)
404 case CPG_REASON_JOIN:
return " via cpg_join";
405 case CPG_REASON_LEAVE:
return " via cpg_leave";
406 case CPG_REASON_NODEDOWN:
return " via cluster exit";
407 case CPG_REASON_NODEUP:
return " via cluster join";
408 case CPG_REASON_PROCDOWN:
return " for unknown reason";
414 static inline const char *
418 return "unknown node";
419 }
else if (peer->
uname == NULL) {
428 const struct cpg_name *groupName,
429 const struct cpg_address *member_list,
size_t member_list_entries,
430 const struct cpg_address *left_list,
size_t left_list_entries,
431 const struct cpg_address *joined_list,
size_t joined_list_entries)
434 gboolean found = FALSE;
437 const struct cpg_address *key, **sorted;
439 sorted = malloc(member_list_entries *
sizeof(
const struct cpg_address *));
442 for (
size_t iter = 0; iter < member_list_entries; iter++) {
443 sorted[iter] = member_list + iter;
446 qsort(sorted, member_list_entries,
sizeof(
const struct cpg_address *),
447 cmp_member_list_nodeid);
449 for (i = 0; i < left_list_entries; i++) {
451 const struct cpg_address **rival = NULL;
470 rival = bsearch(&key, sorted, member_list_entries,
471 sizeof(
const struct cpg_address *),
472 cmp_member_list_nodeid);
476 crm_info(
"Group %s event %d: %s (node %u pid %u) left%s",
477 groupName->value, counter, peer_name(peer),
478 left_list[i].nodeid, left_list[i].
pid,
479 cpgreason2str(left_list[i].reason));
484 }
else if (left_list[i].nodeid == local_nodeid) {
485 crm_warn(
"Group %s event %d: duplicate local pid %u left%s",
486 groupName->value, counter,
487 left_list[i].pid, cpgreason2str(left_list[i].reason));
490 "%s (node %u) duplicate pid %u left%s (%u remains)",
491 groupName->value, counter, peer_name(peer),
492 left_list[i].nodeid, left_list[i].
pid,
493 cpgreason2str(left_list[i].reason), (*rival)->pid);
499 for (i = 0; i < joined_list_entries; i++) {
500 crm_info(
"Group %s event %d: node %u pid %u joined%s",
501 groupName->value, counter, joined_list[i].nodeid,
502 joined_list[i].pid, cpgreason2str(joined_list[i].reason));
505 for (i = 0; i < member_list_entries; i++) {
508 if (member_list[i].nodeid == local_nodeid
509 && member_list[i].
pid != getpid()) {
511 crm_warn(
"Group %s event %d: detected duplicate local pid %u",
512 groupName->value, counter, member_list[i].pid);
515 crm_info(
"Group %s event %d: %s (node %u pid %u) is member",
516 groupName->value, counter, peer_name(peer),
517 member_list[i].nodeid, member_list[i].
pid);
525 time_t now = time(NULL);
528 if(peer->
votes == 0) {
531 }
else if(now > (60 + peer->
votes)) {
537 crm_warn(
"Node %u is member of group %s but was believed offline",
538 member_list[i].nodeid, groupName->value);
545 if (local_nodeid == member_list[i].nodeid) {
551 crm_err(
"Local node was evicted from group %s", groupName->value);
566 cpg_handle_t handle = 0;
577 cpg_callbacks_t cpg_callbacks = {
578 .cpg_deliver_fn = cluster->cpg.cpg_deliver_fn,
579 .cpg_confchg_fn = cluster->cpg.cpg_confchg_fn,
585 cluster->group.length = 0;
586 cluster->group.value[0] = 0;
590 cluster->group.value[127] = 0;
591 cluster->group.length = 1 + QB_MIN(127, strlen(cluster->group.value));
593 cs_repeat(retries, 30, rc = cpg_initialize(&handle, &cpg_callbacks));
595 crm_err(
"Could not connect to the CPG API (rc=%d)", rc);
599 rc = cpg_fd_get(handle, &fd);
601 crm_err(
"Could not obtain the CPG API connection (rc=%d)", rc);
607 &found_uid, &found_gid))) {
608 crm_err(
"CPG provider is not authentic:" 609 " process %lld (uid: %lld, gid: %lld)",
611 (
long long) found_uid, (
long long) found_gid);
615 crm_err(
"Could not verify authenticity of CPG provider: %s (%d)",
623 crm_err(
"Could not get local node id from the CPG API");
630 cs_repeat(retries, 30, rc = cpg_join(handle, &cluster->group));
632 crm_err(
"Could not join the CPG group '%s': %d", crm_system_name, rc);
637 cluster->cpg_handle = handle;
642 cpg_finalize(handle);
667 static int msg_id = 0;
668 static int local_pid = 0;
669 static int local_name_len = 0;
670 static const char *local_name = NULL;
685 if(local_name == NULL) {
688 if(local_name_len == 0 && local_name) {
689 local_name_len = strlen(local_name);
696 if (local_pid == 0) {
697 local_pid = getpid();
708 msg->header.
id =
class;
709 msg->header.error = CS_OK;
716 target = strdup(node->
uname);
725 target = strdup(
"all");
737 msg->
size = 1 + strlen(data);
741 msg = realloc_safe(msg, msg->header.
size);
745 char *compressed = NULL;
746 unsigned int new_size = 0;
747 char *uncompressed = strdup(data);
752 msg = realloc_safe(msg, msg->header.
size);
753 memcpy(msg->
data, compressed, new_size);
759 msg = realloc_safe(msg, msg->header.
size);
767 iov = calloc(1,
sizeof(
struct iovec));
769 iov->iov_len = msg->header.
size;
772 crm_trace(
"Queueing CPG message %u to %s (%llu bytes, %d bytes compressed payload): %.200s",
773 msg->
id, target, (
unsigned long long) iov->iov_len,
776 crm_trace(
"Queueing CPG message %u to %s (%llu bytes, %d bytes payload): %.200s",
777 msg->
id, target, (
unsigned long long) iov->iov_len,
827 int scan_rc = sscanf(text,
"%d", &type);
bool send_plugin_text(int class, struct iovec *iov)
enum crm_ais_msg_types type
#define CRM_CHECK(expr, failure_action)
gboolean send_cpg_iov(struct iovec *iov)
#define crm_notice(fmt, args...)
gboolean safe_str_neq(const char *a, const char *b)
mainloop_io_t * mainloop_add_fd(const char *name, int priority, int fd, void *userdata, struct mainloop_fd_callbacks *callbacks)
gboolean crm_is_peer_active(const crm_node_t *node)
const char * get_local_node_name(void)
void(* destroy)(gpointer)
long long crm_int_helper(const char *text, char **end_text)
#define PCMK__SPECIAL_PID_AS_0(p)
crm_node_t * crm_get_peer(unsigned int id, const char *uname)
crm_node_t * crm_update_peer_state(const char *source, crm_node_t *node, const char *state, uint64_t membership)
Update a node's state and membership information.
char * strerror(int errnum)
char * pcmk_message_common_cs(cpg_handle_t handle, uint32_t nodeid, uint32_t pid, void *content, uint32_t *kind, const char **from)
Wrappers for and extensions to glib mainloop.
void plugin_handle_membership(AIS_Message *msg)
void cluster_disconnect_cpg(crm_cluster_t *cluster)
int(* dispatch)(gpointer userdata)
#define crm_warn(fmt, args...)
#define crm_debug(fmt, args...)
GListPtr cs_message_queue
#define crm_trace(fmt, args...)
crm_node_t * crm_update_peer_proc(const char *source, crm_node_t *peer, uint32_t flag, const char *status)
#define CRM_SYSTEM_PENGINE
gboolean send_cluster_text(int class, const char *data, gboolean local, crm_node_t *node, enum crm_ais_msg_types dest)
gboolean check_message_sanity(const AIS_Message *msg, const char *data)
struct crm_ais_msg_s AIS_Message
cpg_handle_t pcmk_cpg_handle
#define ais_data_len(msg)
guint reap_crm_member(uint32_t id, const char *name)
Remove all peer cache entries matching a node ID and/or uname.
bool crm_compress_string(const char *data, int length, int max, char **result, unsigned int *result_len)
#define CRM_SYSTEM_STONITHD
#define CRM_SYSTEM_TENGINE
uint32_t get_local_nodeid(cpg_handle_t handle)
gboolean(* pcmk_cpg_dispatch_fn)(int kind, const char *from, const char *data)
#define crm_err(fmt, args...)
#define G_PRIORITY_MEDIUM
enum crm_ais_msg_types text2msg_type(const char *text)
#define CRM_BZ2_THRESHOLD
char * dump_xml_unformatted(xmlNode *msg)
gboolean send_cluster_message_cs(xmlNode *msg, gboolean local, crm_node_t *node, enum crm_ais_msg_types dest)
Wrappers for and extensions to libqb IPC.
#define cs_repeat(counter, max, code)
#define safe_str_eq(a, b)
char * crm_strdup_printf(char const *format,...) __attribute__((__format__(__printf__
crm_node_t * crm_find_peer(unsigned int id, const char *uname)
void pcmk_cpg_membership(cpg_handle_t handle, const struct cpg_name *groupName, const struct cpg_address *member_list, size_t member_list_entries, const struct cpg_address *left_list, size_t left_list_entries, const struct cpg_address *joined_list, size_t joined_list_entries)
#define crm_info(fmt, args...)
gboolean cluster_connect_cpg(crm_cluster_t *cluster)
gboolean is_classic_ais_cluster(void)
enum crm_ais_msg_types type
enum cluster_type_e get_cluster_type(void)
int crm_ipc_is_authentic_process(int sock, uid_t refuid, gid_t refgid, pid_t *gotpid, uid_t *gotuid, gid_t *gotgid)
Check the authenticity of the IPC socket peer process.