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: %s (%d)",
99 rc = cpg_fd_get(local_handle, &fd);
101 crm_err(
"Could not obtain the CPG API connection: %s (%d)",
102 cs_strerror(rc), rc);
108 &found_uid, &found_gid))) {
109 crm_err(
"CPG provider is not authentic:" 110 " process %lld (uid: %lld, gid: %lld)",
112 (
long long) found_uid, (
long long) found_gid);
115 crm_err(
"Could not verify authenticity of CPG provider: %s (%d)",
124 cs_repeat(retries, 5, rc = cpg_local_get(local_handle, &local_nodeid));
128 crm_err(
"Could not get local node id from the CPG API: %s (%d)", ais_error2text(rc), rc);
134 cpg_finalize(local_handle);
136 crm_debug(
"Local nodeid is %u", local_nodeid);
144 static ssize_t crm_cs_flush(gpointer
data);
147 crm_cs_flush_cb(gpointer
data)
154 #define CS_SEND_MAX 200 156 crm_cs_flush(gpointer
data)
161 static unsigned int last_sent = 0;
162 cpg_handle_t *handle = (cpg_handle_t *)data;
170 if ((queue_len % 1000) == 0 && queue_len > 1) {
171 crm_err(
"CPG queue has grown to %d", queue_len);
174 crm_warn(
"CPG queue has grown to %d", queue_len);
187 rc = cpg_mcast_joined(*handle, CPG_TYPE_AGREED, iov, 1);
196 (
unsigned long long) iov->iov_len);
205 crm_info(
"Sent %d CPG messages (%d remaining, last=%u): %s (%lld)",
206 sent, queue_len, last_sent, ais_error2text(rc),
209 crm_trace(
"Sent %d CPG messages (%d remaining, last=%u): %s (%lld)",
210 sent, queue_len, last_sent, ais_error2text(rc),
218 delay_ms = QB_MIN(1000,
CS_SEND_MAX + (10 * queue_len));
229 static unsigned int queued = 0;
232 crm_trace(
"Queueing CPG message %u (%llu bytes)",
233 queued, (
unsigned long long) iov->iov_len);
240 pcmk_cpg_dispatch(gpointer user_data)
245 rc = cpg_dispatch(cluster->cpg_handle, CS_DISPATCH_ONE);
247 crm_err(
"Connection to the CPG API failed: %s (%d)", ais_error2text(rc), rc);
248 cluster->cpg_handle = 0;
251 }
else if(cpg_evicted) {
252 crm_err(
"Evicted from CPG membership");
273 crm_err(
"Nodeid mismatch from %d.%d: claimed nodeid=%u", nodeid, pid, msg->
sender.
id);
276 }
else if (msg->
host.
id != 0 && (local_nodeid != msg->
host.
id)) {
291 crm_err(
"Peer with nodeid=%u is unknown", nodeid);
293 }
else if (peer->
uname == NULL) {
294 crm_err(
"No uname for peer with nodeid=%u", nodeid);
297 crm_notice(
"Fixing uname for peer with nodeid=%u", nodeid);
305 crm_trace(
"Got new%s message (size=%d, %d, %d)",
310 *kind = msg->header.
id;
318 char *uncompressed = NULL;
319 unsigned int new_size = msg->
size + 1;
326 uncompressed = calloc(1, new_size);
327 rc = BZ2_bzBuffToBuffDecompress(uncompressed, &new_size, msg->
data, msg->
compressed_size, 1, 0);
330 crm_err(
"Decompression failed: %d", rc);
344 char *pid_s = crm_getpid_s();
351 data = strdup(msg->
data);
362 crm_info(
"Removing peer %s/%u", data,
id);
377 crm_err(
"Invalid message (id=%d, dest=%s:%s, from=%s:%s.%d):" 378 " min=%d, total=%d, size=%d, bz2_size=%d",
388 static int cmp_member_list_nodeid(
const void *first,
391 const struct cpg_address *
const a = *((
const struct cpg_address **) first),
392 *
const b = *((
const struct cpg_address **) second);
393 if (a->nodeid < b->nodeid) {
395 }
else if (a->nodeid > b->nodeid) {
404 const struct cpg_name *groupName,
405 const struct cpg_address *member_list,
size_t member_list_entries,
406 const struct cpg_address *left_list,
size_t left_list_entries,
407 const struct cpg_address *joined_list,
size_t joined_list_entries)
410 gboolean found = FALSE;
413 const struct cpg_address *key, **rival, **sorted;
415 sorted = malloc(member_list_entries *
sizeof(
const struct cpg_address *));
418 for (
size_t iter = 0; iter < member_list_entries; iter++) {
419 sorted[iter] = member_list + iter;
422 qsort(sorted, member_list_entries,
sizeof(
const struct cpg_address *),
423 cmp_member_list_nodeid);
425 for (i = 0; i < left_list_entries; i++) {
428 crm_info(
"Node %u left group %s (peer=%s:%llu, counter=%d.%d)",
429 left_list[i].nodeid, groupName->value,
430 (peer? peer->
uname :
"<none>"),
431 (
unsigned long long) left_list[i].pid, counter, i);
450 rival = bsearch(&key, sorted, member_list_entries,
451 sizeof(
const struct cpg_address *),
452 cmp_member_list_nodeid);
456 }
else if (left_list[i].nodeid == local_nodeid) {
457 crm_info(
"Ignoring the above event %s.%d, comes from a local" 458 " rival process (presumably not us): %llu",
459 groupName->value, counter,
460 (
unsigned long long) left_list[i].pid);
462 crm_info(
"Ignoring the above event %s.%d, comes from" 463 " a rival-rich node: %llu (e.g. %llu process" 465 groupName->value, counter,
466 (
unsigned long long) left_list[i].pid,
467 (
unsigned long long) (*rival)->pid);
474 for (i = 0; i < joined_list_entries; i++) {
475 crm_info(
"Node %u joined group %s (counter=%d.%d, pid=%llu," 476 " unchecked for rivals)",
477 joined_list[i].nodeid, groupName->value, counter, i,
478 (
unsigned long long) left_list[i].pid);
481 for (i = 0; i < member_list_entries; i++) {
484 crm_info(
"Node %u still member of group %s (peer=%s:%llu," 485 " counter=%d.%d, at least once)",
486 member_list[i].nodeid, groupName->value,
487 (peer? peer->
uname :
"<none>"), member_list[i].pid,
490 if (member_list[i].nodeid == local_nodeid
491 && member_list[i].
pid != getpid()) {
493 crm_info(
"Ignoring the above event %s.%d, comes from a local rival" 494 " process: %llu", groupName->value, counter,
495 (
unsigned long long) member_list[i].pid);
505 time_t now = time(NULL);
508 if(peer->
votes == 0) {
511 }
else if(now > (60 + peer->
votes)) {
517 crm_err(
"Node %s[%u] appears to be online even though we think" 518 " it is dead (unchecked for rivals)",
526 if (local_nodeid == member_list[i].nodeid) {
532 crm_err(
"We're not part of CPG group '%s' anymore!", groupName->value);
547 cpg_handle_t handle = 0;
558 cpg_callbacks_t cpg_callbacks = {
559 .cpg_deliver_fn = cluster->cpg.cpg_deliver_fn,
560 .cpg_confchg_fn = cluster->cpg.cpg_confchg_fn,
566 cluster->group.length = 0;
567 cluster->group.value[0] = 0;
571 cluster->group.value[127] = 0;
572 cluster->group.length = 1 + QB_MIN(127, strlen(cluster->group.value));
574 cs_repeat(retries, 30, rc = cpg_initialize(&handle, &cpg_callbacks));
576 crm_err(
"Could not connect to the CPG API: %s (%d)",
577 cs_strerror(rc), rc);
581 rc = cpg_fd_get(handle, &fd);
583 crm_err(
"Could not obtain the CPG API connection: %s (%d)",
584 cs_strerror(rc), rc);
590 &found_uid, &found_gid))) {
591 crm_err(
"CPG provider is not authentic:" 592 " process %lld (uid: %lld, gid: %lld)",
594 (
long long) found_uid, (
long long) found_gid);
598 crm_err(
"Could not verify authenticity of CPG provider: %s (%d)",
606 crm_err(
"Could not get local node id from the CPG API");
613 cs_repeat(retries, 30, rc = cpg_join(handle, &cluster->group));
615 crm_err(
"Could not join the CPG group '%s': %d", crm_system_name, rc);
620 cluster->cpg_handle = handle;
625 cpg_finalize(handle);
650 static int msg_id = 0;
651 static int local_pid = 0;
652 static int local_name_len = 0;
653 static const char *local_name = NULL;
668 if(local_name == NULL) {
671 if(local_name_len == 0 && local_name) {
672 local_name_len = strlen(local_name);
679 if (local_pid == 0) {
680 local_pid = getpid();
691 msg->header.
id =
class;
692 msg->header.error = CS_OK;
699 target = strdup(node->
uname);
708 target = strdup(
"all");
720 msg->
size = 1 + strlen(data);
724 msg = realloc_safe(msg, msg->header.
size);
728 char *compressed = NULL;
729 unsigned int new_size = 0;
730 char *uncompressed = strdup(data);
735 msg = realloc_safe(msg, msg->header.
size);
736 memcpy(msg->
data, compressed, new_size);
742 msg = realloc_safe(msg, msg->header.
size);
750 iov = calloc(1,
sizeof(
struct iovec));
752 iov->iov_len = msg->header.
size;
755 crm_trace(
"Queueing CPG message %u to %s (%llu bytes, %d bytes compressed payload): %.200s",
756 msg->
id, target, (
unsigned long long) iov->iov_len,
759 crm_trace(
"Queueing CPG message %u to %s (%llu bytes, %d bytes payload): %.200s",
760 msg->
id, target, (
unsigned long long) iov->iov_len,
810 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)
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
crm_node_t * crm_update_peer_state(const char *source, crm_node_t *node, const char *state, int membership)
Update a node's state and membership information.
#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.