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);
195 crm_trace(
"CPG message sent, size=%d", iov->iov_len);
204 crm_info(
"Sent %d CPG messages (%d remaining, last=%u): %s (%d)",
205 sent, queue_len, last_sent, ais_error2text(rc), rc);
207 crm_trace(
"Sent %d CPG messages (%d remaining, last=%u): %s (%d)",
208 sent, queue_len, last_sent, ais_error2text(rc), rc);
215 delay_ms = QB_MIN(1000,
CS_SEND_MAX + (10 * queue_len));
226 static unsigned int queued = 0;
229 crm_trace(
"Queueing CPG message %u (%d bytes)", queued, iov->iov_len);
236 pcmk_cpg_dispatch(gpointer user_data)
241 rc = cpg_dispatch(cluster->cpg_handle, CS_DISPATCH_ONE);
243 crm_err(
"Connection to the CPG API failed: %s (%d)", ais_error2text(rc), rc);
244 cluster->cpg_handle = 0;
247 }
else if(cpg_evicted) {
248 crm_err(
"Evicted from CPG membership");
269 crm_err(
"Nodeid mismatch from %d.%d: claimed nodeid=%u", nodeid, pid, msg->
sender.
id);
272 }
else if (msg->
host.
id != 0 && (local_nodeid != msg->
host.
id)) {
287 crm_err(
"Peer with nodeid=%u is unknown", nodeid);
289 }
else if (peer->
uname == NULL) {
290 crm_err(
"No uname for peer with nodeid=%u", nodeid);
293 crm_notice(
"Fixing uname for peer with nodeid=%u", nodeid);
301 crm_trace(
"Got new%s message (size=%d, %d, %d)",
306 *kind = msg->header.
id;
314 char *uncompressed = NULL;
315 unsigned int new_size = msg->
size + 1;
322 uncompressed = calloc(1, new_size);
323 rc = BZ2_bzBuffToBuffDecompress(uncompressed, &new_size, msg->
data, msg->
compressed_size, 1, 0);
326 crm_err(
"Decompression failed: %d", rc);
348 data = strdup(msg->
data);
359 crm_info(
"Removing peer %s/%u", data,
id);
374 crm_err(
"Invalid message (id=%d, dest=%s:%s, from=%s:%s.%d):" 375 " min=%d, total=%d, size=%d, bz2_size=%d",
385 static int cmp_member_list_nodeid(
const void *first,
388 const struct cpg_address *
const a = *((
const struct cpg_address **) first),
389 *
const b = *((
const struct cpg_address **) second);
390 if (a->nodeid < b->nodeid) {
392 }
else if (a->nodeid > b->nodeid) {
401 const struct cpg_name *groupName,
402 const struct cpg_address *member_list,
size_t member_list_entries,
403 const struct cpg_address *left_list,
size_t left_list_entries,
404 const struct cpg_address *joined_list,
size_t joined_list_entries)
407 gboolean found = FALSE;
410 const struct cpg_address *key, **rival, **sorted;
412 sorted = malloc(member_list_entries *
sizeof(
const struct cpg_address *));
415 for (
size_t iter = 0; iter < member_list_entries; iter++) {
416 sorted[iter] = member_list + iter;
419 qsort(sorted, member_list_entries,
sizeof(
const struct cpg_address *),
420 cmp_member_list_nodeid);
422 for (i = 0; i < left_list_entries; i++) {
425 crm_info(
"Node %u left group %s (peer=%s:%llu, counter=%d.%d)",
426 left_list[i].nodeid, groupName->value,
427 (peer? peer->
uname :
"<none>"),
428 (
unsigned long long) left_list[i].pid, counter, i);
447 rival = bsearch(&key, sorted, member_list_entries,
448 sizeof(
const struct cpg_address *),
449 cmp_member_list_nodeid);
453 }
else if (left_list[i].nodeid == local_nodeid) {
454 crm_info(
"Ignoring the above event %s.%d, comes from a local" 455 " rival process (presumably not us): %llu",
456 groupName->value, counter,
457 (
unsigned long long) left_list[i].pid);
459 crm_info(
"Ignoring the above event %s.%d, comes from" 460 " a rival-rich node: %llu (e.g. %llu process" 462 groupName->value, counter,
463 (
unsigned long long) left_list[i].pid,
464 (
unsigned long long) (*rival)->pid);
471 for (i = 0; i < joined_list_entries; i++) {
472 crm_info(
"Node %u joined group %s (counter=%d.%d, pid=%llu," 473 " unchecked for rivals)",
474 joined_list[i].nodeid, groupName->value, counter, i,
475 (
unsigned long long) left_list[i].pid);
478 for (i = 0; i < member_list_entries; i++) {
481 crm_info(
"Node %u still member of group %s (peer=%s:%llu," 482 " counter=%d.%d, at least once)",
483 member_list[i].nodeid, groupName->value,
484 (peer? peer->
uname :
"<none>"), member_list[i].pid,
487 if (member_list[i].nodeid == local_nodeid
488 && member_list[i].
pid != getpid()) {
490 crm_info(
"Ignoring the above event %s.%d, comes from a local rival" 491 " process: %llu", groupName->value, counter,
492 (
unsigned long long) member_list[i].pid);
502 time_t now = time(NULL);
505 if(peer->
votes == 0) {
508 }
else if(now > (60 + peer->
votes)) {
514 crm_err(
"Node %s[%u] appears to be online even though we think" 515 " it is dead (unchecked for rivals)",
523 if (local_nodeid == member_list[i].nodeid) {
529 crm_err(
"We're not part of CPG group '%s' anymore!", groupName->value);
544 cpg_handle_t handle = 0;
555 cpg_callbacks_t cpg_callbacks = {
556 .cpg_deliver_fn = cluster->cpg.cpg_deliver_fn,
557 .cpg_confchg_fn = cluster->cpg.cpg_confchg_fn,
563 cluster->group.length = 0;
564 cluster->group.value[0] = 0;
568 cluster->group.value[127] = 0;
571 cs_repeat(retries, 30, rc = cpg_initialize(&handle, &cpg_callbacks));
573 crm_err(
"Could not connect to the CPG API: %s (%d)",
574 cs_strerror(rc), rc);
578 rc = cpg_fd_get(handle, &fd);
580 crm_err(
"Could not obtain the CPG API connection: %s (%d)",
581 cs_strerror(rc), rc);
587 &found_uid, &found_gid))) {
588 crm_err(
"CPG provider is not authentic:" 589 " process %lld (uid: %lld, gid: %lld)",
591 (
long long) found_uid, (
long long) found_gid);
595 crm_err(
"Could not verify authenticity of CPG provider: %s (%d)",
603 crm_err(
"Could not get local node id from the CPG API");
610 cs_repeat(retries, 30, rc = cpg_join(handle, &cluster->group));
617 cluster->cpg_handle = handle;
622 cpg_finalize(handle);
647 static int msg_id = 0;
648 static int local_pid = 0;
649 static int local_name_len = 0;
650 static const char *local_name = NULL;
665 if(local_name == NULL) {
668 if(local_name_len == 0 && local_name) {
669 local_name_len = strlen(local_name);
676 if (local_pid == 0) {
677 local_pid = getpid();
688 msg->header.
id =
class;
689 msg->header.error = CS_OK;
696 target = strdup(node->
uname);
705 target = strdup(
"all");
717 msg->
size = 1 + strlen(data);
721 msg = realloc_safe(msg, msg->header.
size);
725 char *compressed = NULL;
726 unsigned int new_size = 0;
727 char *uncompressed = strdup(data);
732 msg = realloc_safe(msg, msg->header.
size);
733 memcpy(msg->
data, compressed, new_size);
739 msg = realloc_safe(msg, msg->header.
size);
747 iov = calloc(1,
sizeof(
struct iovec));
749 iov->iov_len = msg->header.
size;
752 crm_trace(
"Queueing CPG message %u to %s (%d bytes, %d bytes compressed payload): %.200s",
755 crm_trace(
"Queueing CPG message %u to %s (%d bytes, %d bytes payload): %.200s",
756 msg->
id, target, iov->iov_len, msg->
size, data);
805 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)
#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.
#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)
bool crm_compress_string(const char *data, int length, int max, char **result, unsigned int *result_len)
#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)
char * crm_itoa(int an_int)
#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)
long long crm_int_helper(const char *text, char **end_text)
#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.