10 #include <sys/socket.h> 11 #include <netinet/in.h> 12 #include <arpa/inet.h> 18 #include <sys/utsname.h> 20 #include <qb/qbipcc.h> 21 #include <qb/qbutil.h> 23 #include <corosync/corodefs.h> 24 #include <corosync/corotypes.h> 25 #include <corosync/hdb.h> 26 #include <corosync/cpg.h> 32 static bool cpg_evicted = FALSE;
35 #define cs_repeat(counter, max, code) do { \ 37 if(rc == CS_ERR_TRY_AGAIN || rc == CS_ERR_QUEUE_FULL) { \ 39 crm_debug("Retrying operation after %ds", counter); \ 44 } while(counter < max) 50 if (cluster->cpg_handle) {
52 cpg_leave(cluster->cpg_handle, &cluster->group);
53 cpg_finalize(cluster->cpg_handle);
54 cluster->cpg_handle = 0;
66 cpg_handle_t local_handle = handle;
67 cpg_callbacks_t cb = { };
69 if(local_nodeid != 0) {
75 cs_repeat(retries, 5, rc = cpg_initialize(&local_handle, &cb));
81 cs_repeat(retries, 5, rc = cpg_local_get(local_handle, &local_nodeid));
85 crm_err(
"Could not get local node id from the CPG API: %s (%d)", ais_error2text(rc), rc);
89 cpg_finalize(local_handle);
91 crm_debug(
"Local nodeid is %u", local_nodeid);
99 static ssize_t crm_cs_flush(gpointer
data);
102 crm_cs_flush_cb(gpointer
data)
109 #define CS_SEND_MAX 200 111 crm_cs_flush(gpointer
data)
116 static unsigned int last_sent = 0;
117 cpg_handle_t *handle = (cpg_handle_t *)data;
125 if ((queue_len % 1000) == 0 && queue_len > 1) {
126 crm_err(
"CPG queue has grown to %d", queue_len);
129 crm_warn(
"CPG queue has grown to %d", queue_len);
142 rc = cpg_mcast_joined(*handle, CPG_TYPE_AGREED, iov, 1);
151 (
unsigned long long) iov->iov_len);
160 crm_info(
"Sent %d CPG messages (%d remaining, last=%u): %s (%lld)",
161 sent, queue_len, last_sent, ais_error2text(rc),
164 crm_trace(
"Sent %d CPG messages (%d remaining, last=%u): %s (%lld)",
165 sent, queue_len, last_sent, ais_error2text(rc),
173 delay_ms = QB_MIN(1000,
CS_SEND_MAX + (10 * queue_len));
184 static unsigned int queued = 0;
187 crm_trace(
"Queueing CPG message %u (%llu bytes)",
188 queued, (
unsigned long long) iov->iov_len);
195 pcmk_cpg_dispatch(gpointer user_data)
200 rc = cpg_dispatch(cluster->cpg_handle, CS_DISPATCH_ONE);
202 crm_err(
"Connection to the CPG API failed: %s (%d)", ais_error2text(rc), rc);
203 cluster->cpg_handle = 0;
206 }
else if(cpg_evicted) {
207 crm_err(
"Evicted from CPG membership");
226 crm_err(
"Nodeid mismatch from %d.%d: claimed nodeid=%u", nodeid, pid, msg->
sender.
id);
229 }
else if (msg->
host.
id != 0 && (local_nodeid != msg->
host.
id)) {
244 crm_err(
"Peer with nodeid=%u is unknown", nodeid);
246 }
else if (peer->
uname == NULL) {
247 crm_err(
"No uname for peer with nodeid=%u", nodeid);
250 crm_notice(
"Fixing uname for peer with nodeid=%u", nodeid);
258 crm_trace(
"Got new%s message (size=%d, %d, %d)",
263 *kind = msg->header.
id;
271 char *uncompressed = NULL;
272 unsigned int new_size = msg->
size + 1;
279 uncompressed = calloc(1, new_size);
280 rc = BZ2_bzBuffToBuffDecompress(uncompressed, &new_size, msg->
data, msg->
compressed_size, 1, 0);
298 char *pid_s = crm_getpid_s();
305 data = strdup(msg->
data);
315 crm_err(
"Invalid message (id=%d, dest=%s:%s, from=%s:%s.%d):" 316 " min=%d, total=%d, size=%d, bz2_size=%d",
326 #define PEER_NAME(peer) ((peer)? ((peer)->uname? (peer)->uname : "<unknown>") : "<none>") 330 const struct cpg_name *groupName,
331 const struct cpg_address *member_list,
size_t member_list_entries,
332 const struct cpg_address *left_list,
size_t left_list_entries,
333 const struct cpg_address *joined_list,
size_t joined_list_entries)
336 gboolean found = FALSE;
337 static int counter = 0;
340 for (i = 0; i < left_list_entries; i++) {
343 crm_info(
"Group event %s.%d: node %u (%s) left",
344 groupName->value, counter, left_list[i].nodeid,
351 for (i = 0; i < joined_list_entries; i++) {
352 crm_info(
"Group event %s.%d: node %u joined",
353 groupName->value, counter, joined_list[i].nodeid);
356 for (i = 0; i < member_list_entries; i++) {
359 crm_info(
"Group event %s.%d: node %u (%s) is member",
360 groupName->value, counter, member_list[i].nodeid,
376 time_t now = time(NULL);
382 }
else if (now > (peer->
when_lost + 60)) {
384 crm_warn(
"Node %u member of group %s but believed offline",
385 member_list[i].nodeid, groupName->value);
390 if (local_nodeid == member_list[i].nodeid) {
396 crm_err(
"We're not part of CPG group '%s' anymore!", groupName->value);
411 cpg_handle_t handle = 0;
419 cpg_callbacks_t cpg_callbacks = {
420 .cpg_deliver_fn = cluster->cpg.cpg_deliver_fn,
421 .cpg_confchg_fn = cluster->cpg.cpg_confchg_fn,
427 cluster->group.length = 0;
428 cluster->group.value[0] = 0;
431 strncpy(cluster->group.value, message_name, 127);
432 cluster->group.value[127] = 0;
433 cluster->group.length = 1 + QB_MIN(127, strlen(cluster->group.value));
435 cs_repeat(retries, 30, rc = cpg_initialize(&handle, &cpg_callbacks));
437 crm_err(
"Could not connect to the Cluster Process Group API: %d", rc);
443 crm_err(
"Could not get local node id from the CPG API");
450 cs_repeat(retries, 30, rc = cpg_join(handle, &cluster->group));
452 crm_err(
"Could not join the CPG group '%s': %d", message_name, rc);
456 rc = cpg_fd_get(handle, &fd);
458 crm_err(
"Could not obtain the CPG API connection: %d", rc);
463 cluster->cpg_handle = handle;
468 cpg_finalize(handle);
493 static int msg_id = 0;
494 static int local_pid = 0;
495 static int local_name_len = 0;
496 static const char *local_name = NULL;
507 crm_err(
"Invalid message class: %d", msg_class);
513 if(local_name == NULL) {
516 if(local_name_len == 0 && local_name) {
517 local_name_len = strlen(local_name);
524 if (local_pid == 0) {
525 local_pid = getpid();
536 msg->header.
id = msg_class;
537 msg->header.error = CS_OK;
544 target = strdup(node->
uname);
553 target = strdup(
"all");
565 msg->
size = 1 + strlen(data);
569 msg = realloc_safe(msg, msg->header.
size);
573 char *compressed = NULL;
574 unsigned int new_size = 0;
575 char *uncompressed = strdup(data);
580 msg = realloc_safe(msg, msg->header.
size);
581 memcpy(msg->
data, compressed, new_size);
587 msg = realloc_safe(msg, msg->header.
size);
595 iov = calloc(1,
sizeof(
struct iovec));
597 iov->iov_len = msg->header.
size;
600 crm_trace(
"Queueing CPG message %u to %s (%llu bytes, %d bytes compressed payload): %.200s",
601 msg->
id, target, (
unsigned long long) iov->iov_len,
604 crm_trace(
"Queueing CPG message %u to %s (%llu bytes, %d bytes payload): %.200s",
605 msg->
id, target, (
unsigned long long) iov->iov_len,
646 int scan_rc = sscanf(text,
"%d", &type);
enum crm_ais_msg_types type
#define CRM_CHECK(expr, failure_action)
gboolean send_cpg_iov(struct iovec *iov)
#define crm_notice(fmt, args...)
const char * bz2_strerror(int rc)
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)
const char * get_local_node_name(void)
void(* destroy)(gpointer)
crm_node_t * crm_get_peer(unsigned int id, const char *uname)
gboolean send_cluster_text(enum crm_ais_msg_class msg_class, const char *data, gboolean local, crm_node_t *node, enum crm_ais_msg_types dest)
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 cluster_disconnect_cpg(crm_cluster_t *cluster)
int(* dispatch)(gpointer userdata)
#define crm_warn(fmt, args...)
#define crm_debug(fmt, args...)
const char * pcmk_message_name(const char *name)
Get name to be used as identifier for cluster messages.
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 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)
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)
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)
char * crm_strdup_printf(char const *format,...) __attribute__((__format__(__printf__
#define crm_info(fmt, args...)
gboolean cluster_connect_cpg(crm_cluster_t *cluster)
enum crm_ais_msg_types type