pacemaker  1.1.14-70404b0
Scalable High-Availability cluster resource manager
 All Data Structures Files Functions Variables Typedefs Enumerations Enumerator Macros Groups Pages
cpg.c
Go to the documentation of this file.
1 /*
2  * Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
3  *
4  * This library is free software; you can redistribute it and/or
5  * modify it under the terms of the GNU Lesser General Public
6  * License as published by the Free Software Foundation; either
7  * version 2.1 of the License, or (at your option) any later version.
8  *
9  * This library is distributed in the hope that it will be useful,
10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12  * Lesser General Public License for more details.
13  *
14  * You should have received a copy of the GNU Lesser General Public
15  * License along with this library; if not, write to the Free Software
16  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
17  */
18 
19 #include <crm_internal.h>
20 #include <bzlib.h>
21 #include <sys/socket.h>
22 #include <netinet/in.h>
23 #include <arpa/inet.h>
24 #include <netdb.h>
25 
26 #include <crm/common/ipc.h>
27 #include <crm/cluster/internal.h>
28 #include <crm/common/mainloop.h>
29 #include <sys/utsname.h>
30 
31 #include <qb/qbipcc.h>
32 #include <qb/qbutil.h>
33 
34 #include <corosync/corodefs.h>
35 #include <corosync/corotypes.h>
36 #include <corosync/hdb.h>
37 #include <corosync/cpg.h>
38 
39 #include <crm/msg_xml.h>
40 
41 cpg_handle_t pcmk_cpg_handle = 0; /* TODO: Remove, use cluster.cpg_handle */
42 
43 static bool cpg_evicted = FALSE;
44 gboolean(*pcmk_cpg_dispatch_fn) (int kind, const char *from, const char *data) = NULL;
45 
46 #define cs_repeat(counter, max, code) do { \
47  code; \
48  if(rc == CS_ERR_TRY_AGAIN || rc == CS_ERR_QUEUE_FULL) { \
49  counter++; \
50  crm_debug("Retrying operation after %ds", counter); \
51  sleep(counter); \
52  } else { \
53  break; \
54  } \
55  } while(counter < max)
56 
57 void
59 {
60  pcmk_cpg_handle = 0;
61  if (cluster->cpg_handle) {
62  crm_trace("Disconnecting CPG");
63  cpg_leave(cluster->cpg_handle, &cluster->group);
64  cpg_finalize(cluster->cpg_handle);
65  cluster->cpg_handle = 0;
66 
67  } else {
68  crm_info("No CPG connection");
69  }
70 }
71 
72 uint32_t get_local_nodeid(cpg_handle_t handle)
73 {
74  int rc = CS_OK;
75  int retries = 0;
76  static uint32_t local_nodeid = 0;
77  cpg_handle_t local_handle = handle;
78  cpg_callbacks_t cb = { };
79 
80  if(local_nodeid != 0) {
81  return local_nodeid;
82  }
83 
84 #if 0
85  /* Should not be necessary */
87  get_ais_details(&local_nodeid, NULL);
88  goto done;
89  }
90 #endif
91 
92  if(handle == 0) {
93  crm_trace("Creating connection");
94  cs_repeat(retries, 5, rc = cpg_initialize(&local_handle, &cb));
95  }
96 
97  if (rc == CS_OK) {
98  retries = 0;
99  crm_trace("Performing lookup");
100  cs_repeat(retries, 5, rc = cpg_local_get(local_handle, &local_nodeid));
101  }
102 
103  if (rc != CS_OK) {
104  crm_err("Could not get local node id from the CPG API: %s (%d)", ais_error2text(rc), rc);
105  }
106  if(handle == 0) {
107  crm_trace("Closing connection");
108  cpg_finalize(local_handle);
109  }
110  crm_debug("Local nodeid is %u", local_nodeid);
111  return local_nodeid;
112 }
113 
114 
117 
118 static ssize_t crm_cs_flush(gpointer data);
119 
120 static gboolean
121 crm_cs_flush_cb(gpointer data)
122 {
123  cs_message_timer = 0;
124  crm_cs_flush(data);
125  return FALSE;
126 }
127 
128 #define CS_SEND_MAX 200
129 static ssize_t
130 crm_cs_flush(gpointer data)
131 {
132  int sent = 0;
133  ssize_t rc = 0;
134  int queue_len = 0;
135  static unsigned int last_sent = 0;
136  cpg_handle_t *handle = (cpg_handle_t *)data;
137 
138  if (*handle == 0) {
139  crm_trace("Connection is dead");
140  return pcmk_ok;
141  }
142 
143  queue_len = g_list_length(cs_message_queue);
144  if ((queue_len % 1000) == 0 && queue_len > 1) {
145  crm_err("CPG queue has grown to %d", queue_len);
146 
147  } else if (queue_len == CS_SEND_MAX) {
148  crm_warn("CPG queue has grown to %d", queue_len);
149  }
150 
151  if (cs_message_timer) {
152  /* There is already a timer, wait until it goes off */
153  crm_trace("Timer active %d", cs_message_timer);
154  return pcmk_ok;
155  }
156 
157  while (cs_message_queue && sent < CS_SEND_MAX) {
158  struct iovec *iov = cs_message_queue->data;
159 
160  errno = 0;
161  rc = cpg_mcast_joined(*handle, CPG_TYPE_AGREED, iov, 1);
162 
163  if (rc != CS_OK) {
164  break;
165  }
166 
167  sent++;
168  last_sent++;
169  crm_trace("CPG message sent, size=%d", iov->iov_len);
170 
171  cs_message_queue = g_list_remove(cs_message_queue, iov);
172  free(iov->iov_base);
173  free(iov);
174  }
175 
176  queue_len -= sent;
177  if (sent > 1 || cs_message_queue) {
178  crm_info("Sent %d CPG messages (%d remaining, last=%u): %s (%d)",
179  sent, queue_len, last_sent, ais_error2text(rc), rc);
180  } else {
181  crm_trace("Sent %d CPG messages (%d remaining, last=%u): %s (%d)",
182  sent, queue_len, last_sent, ais_error2text(rc), rc);
183  }
184 
185  if (cs_message_queue) {
186  uint32_t delay_ms = 100;
187  if(rc != CS_OK) {
188  /* Proportionally more if sending failed but cap at 1s */
189  delay_ms = QB_MIN(1000, CS_SEND_MAX + (10 * queue_len));
190  }
191  cs_message_timer = g_timeout_add(delay_ms, crm_cs_flush_cb, data);
192  }
193 
194  return rc;
195 }
196 
197 gboolean
198 send_cpg_iov(struct iovec * iov)
199 {
200  static unsigned int queued = 0;
201 
202  queued++;
203  crm_trace("Queueing CPG message %u (%d bytes)", queued, iov->iov_len);
204  cs_message_queue = g_list_append(cs_message_queue, iov);
205  crm_cs_flush(&pcmk_cpg_handle);
206  return TRUE;
207 }
208 
209 static int
210 pcmk_cpg_dispatch(gpointer user_data)
211 {
212  int rc = 0;
213  crm_cluster_t *cluster = (crm_cluster_t*) user_data;
214 
215  rc = cpg_dispatch(cluster->cpg_handle, CS_DISPATCH_ONE);
216  if (rc != CS_OK) {
217  crm_err("Connection to the CPG API failed: %s (%d)", ais_error2text(rc), rc);
218  cluster->cpg_handle = 0;
219  return -1;
220 
221  } else if(cpg_evicted) {
222  crm_err("Evicted from CPG membership");
223  return -1;
224  }
225  return 0;
226 }
227 
228 char *
229 pcmk_message_common_cs(cpg_handle_t handle, uint32_t nodeid, uint32_t pid, void *content,
230  uint32_t *kind, const char **from)
231 {
232  char *data = NULL;
233  AIS_Message *msg = (AIS_Message *) content;
234 
235  if(handle) {
236  /* 'msg' came from CPG not the plugin
237  * Do filtering and field massaging
238  */
240  const char *local_name = get_local_node_name();
241 
242  if (msg->sender.id > 0 && msg->sender.id != nodeid) {
243  crm_err("Nodeid mismatch from %d.%d: claimed nodeid=%u", nodeid, pid, msg->sender.id);
244  return NULL;
245 
246  } else if (msg->host.id != 0 && (local_nodeid != msg->host.id)) {
247  /* Not for us */
248  crm_trace("Not for us: %u != %u", msg->host.id, local_nodeid);
249  return NULL;
250  } else if (msg->host.size != 0 && safe_str_neq(msg->host.uname, local_name)) {
251  /* Not for us */
252  crm_trace("Not for us: %s != %s", msg->host.uname, local_name);
253  return NULL;
254  }
255 
256  msg->sender.id = nodeid;
257  if (msg->sender.size == 0) {
258  crm_node_t *peer = crm_get_peer(nodeid, NULL);
259 
260  if (peer == NULL) {
261  crm_err("Peer with nodeid=%u is unknown", nodeid);
262 
263  } else if (peer->uname == NULL) {
264  crm_err("No uname for peer with nodeid=%u", nodeid);
265 
266  } else {
267  crm_notice("Fixing uname for peer with nodeid=%u", nodeid);
268  msg->sender.size = strlen(peer->uname);
269  memset(msg->sender.uname, 0, MAX_NAME);
270  memcpy(msg->sender.uname, peer->uname, msg->sender.size);
271  }
272  }
273  }
274 
275  crm_trace("Got new%s message (size=%d, %d, %d)",
276  msg->is_compressed ? " compressed" : "",
277  ais_data_len(msg), msg->size, msg->compressed_size);
278 
279  if (kind != NULL) {
280  *kind = msg->header.id;
281  }
282  if (from != NULL) {
283  *from = msg->sender.uname;
284  }
285 
286  if (msg->is_compressed && msg->size > 0) {
287  int rc = BZ_OK;
288  char *uncompressed = NULL;
289  unsigned int new_size = msg->size + 1;
290 
291  if (check_message_sanity(msg, NULL) == FALSE) {
292  goto badmsg;
293  }
294 
295  crm_trace("Decompressing message data");
296  uncompressed = calloc(1, new_size);
297  rc = BZ2_bzBuffToBuffDecompress(uncompressed, &new_size, msg->data, msg->compressed_size, 1, 0);
298 
299  if (rc != BZ_OK) {
300  crm_err("Decompression failed: %d", rc);
301  free(uncompressed);
302  goto badmsg;
303  }
304 
305  CRM_ASSERT(rc == BZ_OK);
306  CRM_ASSERT(new_size == msg->size);
307 
308  data = uncompressed;
309 
310  } else if (check_message_sanity(msg, data) == FALSE) {
311  goto badmsg;
312 
313  } else if (safe_str_eq("identify", data)) {
314  int pid = getpid();
315  char *pid_s = crm_itoa(pid);
316 
317  send_cluster_text(crm_class_cluster, pid_s, TRUE, NULL, crm_msg_ais);
318  free(pid_s);
319  return NULL;
320 
321  } else {
322  data = strdup(msg->data);
323  }
324 
325  if (msg->header.id != crm_class_members) {
326  /* Is this even needed anymore? */
327  crm_get_peer(msg->sender.id, msg->sender.uname);
328  }
329 
330  if (msg->header.id == crm_class_rmpeer) {
331  uint32_t id = crm_int_helper(data, NULL);
332 
333  crm_info("Removing peer %s/%u", data, id);
334  reap_crm_member(id, NULL);
335  free(data);
336  return NULL;
337 
338 #if SUPPORT_PLUGIN
339  } else if (is_classic_ais_cluster()) {
341 #endif
342  }
343 
344  crm_trace("Payload: %.200s", data);
345  return data;
346 
347  badmsg:
348  crm_err("Invalid message (id=%d, dest=%s:%s, from=%s:%s.%d):"
349  " min=%d, total=%d, size=%d, bz2_size=%d",
350  msg->id, ais_dest(&(msg->host)), msg_type2text(msg->host.type),
351  ais_dest(&(msg->sender)), msg_type2text(msg->sender.type),
352  msg->sender.pid, (int)sizeof(AIS_Message),
353  msg->header.size, msg->size, msg->compressed_size);
354 
355  free(data);
356  return NULL;
357 }
358 
359 void
360 pcmk_cpg_membership(cpg_handle_t handle,
361  const struct cpg_name *groupName,
362  const struct cpg_address *member_list, size_t member_list_entries,
363  const struct cpg_address *left_list, size_t left_list_entries,
364  const struct cpg_address *joined_list, size_t joined_list_entries)
365 {
366  int i;
367  gboolean found = FALSE;
368  static int counter = 0;
370 
371  for (i = 0; i < left_list_entries; i++) {
372  crm_node_t *peer = crm_find_peer(left_list[i].nodeid, NULL);
373 
374  crm_info("Node %u left group %s (peer=%s, counter=%d.%d)",
375  left_list[i].nodeid, groupName->value,
376  (peer? peer->uname : "<none>"), counter, i);
377  if (peer) {
378  crm_update_peer_proc(__FUNCTION__, peer, crm_proc_cpg, OFFLINESTATUS);
379  }
380  }
381 
382  for (i = 0; i < joined_list_entries; i++) {
383  crm_info("Node %u joined group %s (counter=%d.%d)",
384  joined_list[i].nodeid, groupName->value, counter, i);
385  }
386 
387  for (i = 0; i < member_list_entries; i++) {
388  crm_node_t *peer = crm_get_peer(member_list[i].nodeid, NULL);
389 
390  crm_info("Node %u still member of group %s (peer=%s, counter=%d.%d)",
391  member_list[i].nodeid, groupName->value,
392  (peer? peer->uname : "<none>"), counter, i);
393 
394  /* Anyone that is sending us CPG messages must also be a _CPG_ member.
395  * But its _not_ safe to assume its in the quorum membership.
396  * We may have just found out its dead and are processing the last couple of messages it sent
397  */
398  peer = crm_update_peer_proc(__FUNCTION__, peer, crm_proc_cpg, ONLINESTATUS);
399  if(peer && peer->state && crm_is_peer_active(peer) == FALSE) {
400  time_t now = time(NULL);
401 
402  /* Co-opt the otherwise unused votes field */
403  if(peer->votes == 0) {
404  peer->votes = now;
405 
406  } else if(now > (60 + peer->votes)) {
407  /* On the otherhand, if we're still getting messages, at a certain point
408  * we need to acknowledge our internal cache is probably wrong
409  *
410  * Set the threshold to 1 minute
411  */
412  crm_err("Node %s[%u] appears to be online even though we think it is dead", peer->uname, peer->id);
413  if (crm_update_peer_state(__FUNCTION__, peer, CRM_NODE_MEMBER, 0)) {
414  peer->votes = 0;
415  }
416  }
417  }
418 
419  if (local_nodeid == member_list[i].nodeid) {
420  found = TRUE;
421  }
422  }
423 
424  if (!found) {
425  crm_err("We're not part of CPG group '%s' anymore!", groupName->value);
426  cpg_evicted = TRUE;
427  }
428 
429  counter++;
430 }
431 
432 gboolean
434 {
435  int rc = -1;
436  int fd = 0;
437  int retries = 0;
438  uint32_t id = 0;
439  crm_node_t *peer = NULL;
440  cpg_handle_t handle = 0;
441 
442  struct mainloop_fd_callbacks cpg_fd_callbacks = {
443  .dispatch = pcmk_cpg_dispatch,
444  .destroy = cluster->destroy,
445  };
446 
447  cpg_callbacks_t cpg_callbacks = {
448  .cpg_deliver_fn = cluster->cpg.cpg_deliver_fn,
449  .cpg_confchg_fn = cluster->cpg.cpg_confchg_fn,
450  /* .cpg_deliver_fn = pcmk_cpg_deliver, */
451  /* .cpg_confchg_fn = pcmk_cpg_membership, */
452  };
453 
454  cpg_evicted = FALSE;
455  cluster->group.length = 0;
456  cluster->group.value[0] = 0;
457 
458  /* group.value is char[128] */
459  strncpy(cluster->group.value, crm_system_name, 127);
460  cluster->group.value[127] = 0;
461  cluster->group.length = 1 + QB_MIN(127, strlen(crm_system_name));
462 
463  cs_repeat(retries, 30, rc = cpg_initialize(&handle, &cpg_callbacks));
464  if (rc != CS_OK) {
465  crm_err("Could not connect to the Cluster Process Group API: %d\n", rc);
466  goto bail;
467  }
468 
469  id = get_local_nodeid(handle);
470  if (id == 0) {
471  crm_err("Could not get local node id from the CPG API");
472  goto bail;
473 
474  }
475  cluster->nodeid = id;
476 
477  retries = 0;
478  cs_repeat(retries, 30, rc = cpg_join(handle, &cluster->group));
479  if (rc != CS_OK) {
480  crm_err("Could not join the CPG group '%s': %d", crm_system_name, rc);
481  goto bail;
482  }
483 
484  rc = cpg_fd_get(handle, &fd);
485  if (rc != CS_OK) {
486  crm_err("Could not obtain the CPG API connection: %d\n", rc);
487  goto bail;
488  }
489 
490  pcmk_cpg_handle = handle;
491  cluster->cpg_handle = handle;
492  mainloop_add_fd("corosync-cpg", G_PRIORITY_MEDIUM, fd, cluster, &cpg_fd_callbacks);
493 
494  bail:
495  if (rc != CS_OK) {
496  cpg_finalize(handle);
497  return FALSE;
498  }
499 
500  peer = crm_get_peer(id, NULL);
501  crm_update_peer_proc(__FUNCTION__, peer, crm_proc_cpg, ONLINESTATUS);
502  return TRUE;
503 }
504 
505 gboolean
506 send_cluster_message_cs(xmlNode * msg, gboolean local, crm_node_t * node, enum crm_ais_msg_types dest)
507 {
508  gboolean rc = TRUE;
509  char *data = NULL;
510 
511  data = dump_xml_unformatted(msg);
512  rc = send_cluster_text(crm_class_cluster, data, local, node, dest);
513  free(data);
514  return rc;
515 }
516 
517 gboolean
518 send_cluster_text(int class, const char *data,
519  gboolean local, crm_node_t * node, enum crm_ais_msg_types dest)
520 {
521  static int msg_id = 0;
522  static int local_pid = 0;
523  static int local_name_len = 0;
524  static const char *local_name = NULL;
525 
526  char *target = NULL;
527  struct iovec *iov;
528  AIS_Message *msg = NULL;
530 
531  /* There are only 6 handlers registered to crm_lib_service in plugin.c */
532  CRM_CHECK(class < 6, crm_err("Invalid message class: %d", class);
533  return FALSE);
534 
535 #if !SUPPORT_PLUGIN
536  CRM_CHECK(dest != crm_msg_ais, return FALSE);
537 #endif
538 
539  if(local_name == NULL) {
540  local_name = get_local_node_name();
541  }
542  if(local_name_len == 0 && local_name) {
543  local_name_len = strlen(local_name);
544  }
545 
546  if (data == NULL) {
547  data = "";
548  }
549 
550  if (local_pid == 0) {
551  local_pid = getpid();
552  }
553 
554  if (sender == crm_msg_none) {
555  sender = local_pid;
556  }
557 
558  msg = calloc(1, sizeof(AIS_Message));
559 
560  msg_id++;
561  msg->id = msg_id;
562  msg->header.id = class;
563  msg->header.error = CS_OK;
564 
565  msg->host.type = dest;
566  msg->host.local = local;
567 
568  if (node) {
569  if (node->uname) {
570  target = strdup(node->uname);
571  msg->host.size = strlen(node->uname);
572  memset(msg->host.uname, 0, MAX_NAME);
573  memcpy(msg->host.uname, node->uname, msg->host.size);
574  } else {
575  target = crm_strdup_printf("%u", node->id);
576  }
577  msg->host.id = node->id;
578  } else {
579  target = strdup("all");
580  }
581 
582  msg->sender.id = 0;
583  msg->sender.type = sender;
584  msg->sender.pid = local_pid;
585  msg->sender.size = local_name_len;
586  memset(msg->sender.uname, 0, MAX_NAME);
587  if(local_name && msg->sender.size) {
588  memcpy(msg->sender.uname, local_name, msg->sender.size);
589  }
590 
591  msg->size = 1 + strlen(data);
592  msg->header.size = sizeof(AIS_Message) + msg->size;
593 
594  if (msg->size < CRM_BZ2_THRESHOLD) {
595  msg = realloc_safe(msg, msg->header.size);
596  memcpy(msg->data, data, msg->size);
597 
598  } else {
599  char *compressed = NULL;
600  unsigned int new_size = 0;
601  char *uncompressed = strdup(data);
602 
603  if (crm_compress_string(uncompressed, msg->size, 0, &compressed, &new_size)) {
604 
605  msg->header.size = sizeof(AIS_Message) + new_size;
606  msg = realloc_safe(msg, msg->header.size);
607  memcpy(msg->data, compressed, new_size);
608 
609  msg->is_compressed = TRUE;
610  msg->compressed_size = new_size;
611 
612  } else {
613  msg = realloc_safe(msg, msg->header.size);
614  memcpy(msg->data, data, msg->size);
615  }
616 
617  free(uncompressed);
618  free(compressed);
619  }
620 
621  iov = calloc(1, sizeof(struct iovec));
622  iov->iov_base = msg;
623  iov->iov_len = msg->header.size;
624 
625  if (msg->compressed_size) {
626  crm_trace("Queueing CPG message %u to %s (%d bytes, %d bytes compressed payload): %.200s",
627  msg->id, target, iov->iov_len, msg->compressed_size, data);
628  } else {
629  crm_trace("Queueing CPG message %u to %s (%d bytes, %d bytes payload): %.200s",
630  msg->id, target, iov->iov_len, msg->size, data);
631  }
632  free(target);
633 
634 #if SUPPORT_PLUGIN
635  /* The plugin is the only time we dont use CPG messaging */
637  return send_plugin_text(class, iov);
638  }
639 #endif
640 
641  send_cpg_iov(iov);
642 
643  return TRUE;
644 }
645 
647 text2msg_type(const char *text)
648 {
649  int type = crm_msg_none;
650 
651  CRM_CHECK(text != NULL, return type);
652  if (safe_str_eq(text, "ais")) {
653  type = crm_msg_ais;
654  } else if (safe_str_eq(text, "crm_plugin")) {
655  type = crm_msg_ais;
656  } else if (safe_str_eq(text, CRM_SYSTEM_CIB)) {
657  type = crm_msg_cib;
658  } else if (safe_str_eq(text, CRM_SYSTEM_CRMD)) {
659  type = crm_msg_crmd;
660  } else if (safe_str_eq(text, CRM_SYSTEM_DC)) {
661  type = crm_msg_crmd;
662  } else if (safe_str_eq(text, CRM_SYSTEM_TENGINE)) {
663  type = crm_msg_te;
664  } else if (safe_str_eq(text, CRM_SYSTEM_PENGINE)) {
665  type = crm_msg_pe;
666  } else if (safe_str_eq(text, CRM_SYSTEM_LRMD)) {
667  type = crm_msg_lrmd;
668  } else if (safe_str_eq(text, CRM_SYSTEM_STONITHD)) {
669  type = crm_msg_stonithd;
670  } else if (safe_str_eq(text, "stonith-ng")) {
671  type = crm_msg_stonith_ng;
672  } else if (safe_str_eq(text, "attrd")) {
673  type = crm_msg_attrd;
674 
675  } else {
676  /* This will normally be a transient client rather than
677  * a cluster daemon. Set the type to the pid of the client
678  */
679  int scan_rc = sscanf(text, "%d", &type);
680 
681  if (scan_rc != 1 || type <= crm_msg_stonith_ng) {
682  /* Ensure its sane */
683  type = crm_msg_none;
684  }
685  }
686  return type;
687 }
bool send_plugin_text(int class, struct iovec *iov)
Definition: legacy.c:133
enum crm_ais_msg_types type
Definition: internal.h:39
#define CRM_CHECK(expr, failure_action)
Definition: logging.h:164
char data[0]
Definition: internal.h:56
gboolean send_cpg_iov(struct iovec *iov)
Definition: cpg.c:198
gboolean(* pcmk_cpg_dispatch_fn)(int kind, const char *from, const char *data)
Definition: cpg.c:44
uint32_t local_nodeid
Definition: plugin.c:65
#define crm_notice(fmt, args...)
Definition: logging.h:250
gboolean is_compressed
Definition: internal.h:48
uint32_t size
Definition: internal.h:53
gboolean safe_str_neq(const char *a, const char *b)
Definition: utils.c:668
crm_ais_msg_types
Definition: cluster.h:125
mainloop_io_t * mainloop_add_fd(const char *name, int priority, int fd, void *userdata, struct mainloop_fd_callbacks *callbacks)
Definition: mainloop.c:806
uint32_t nodeid
Definition: cluster.h:94
uint32_t id
Definition: cluster.h:70
gboolean crm_is_peer_active(const crm_node_t *node)
Definition: membership.c:143
const char * get_local_node_name(void)
Definition: cluster.c:289
void(* destroy)(gpointer)
Definition: cluster.h:96
#define pcmk_ok
Definition: error.h:42
uint32_t id
Definition: internal.h:36
crm_node_t * crm_get_peer(unsigned int id, const char *uname)
Definition: membership.c:519
int(* dispatch)(gpointer userdata)
Definition: mainloop.h:90
char * crm_system_name
Definition: utils.c:74
#define CS_SEND_MAX
Definition: cpg.c:128
uint32_t pid
Definition: internal.h:49
AIS_Host sender
Definition: internal.h:53
char * pcmk_message_common_cs(cpg_handle_t handle, uint32_t nodeid, uint32_t pid, void *content, uint32_t *kind, const char **from)
Definition: cpg.c:229
Wrappers for and extensions to glib mainloop.
#define CRM_SYSTEM_DC
Definition: crm.h:80
void plugin_handle_membership(AIS_Message *msg)
Definition: legacy.c:218
void cluster_disconnect_cpg(crm_cluster_t *cluster)
Definition: cpg.c:58
int cs_message_timer
Definition: cpg.c:116
#define crm_warn(fmt, args...)
Definition: logging.h:249
uint32_t id
Definition: internal.h:48
#define crm_debug(fmt, args...)
Definition: logging.h:253
GListPtr cs_message_queue
Definition: cpg.c:115
#define crm_trace(fmt, args...)
Definition: logging.h:254
gboolean local
Definition: internal.h:38
crm_node_t * crm_update_peer_proc(const char *source, crm_node_t *peer, uint32_t flag, const char *status)
Definition: membership.c:723
#define CRM_SYSTEM_PENGINE
Definition: crm.h:86
AIS_Host sender
Definition: internal.h:51
uint32_t id
Definition: internal.h:47
gboolean send_cluster_text(int class, const char *data, gboolean local, crm_node_t *node, enum crm_ais_msg_types dest)
Definition: cpg.c:518
gboolean check_message_sanity(const AIS_Message *msg, const char *data)
Definition: plugin.c:1372
struct crm_ais_msg_s AIS_Message
Definition: internal.h:33
cpg_handle_t pcmk_cpg_handle
Definition: cpg.c:41
#define ais_data_len(msg)
Definition: internal.h:210
uint32_t size
Definition: internal.h:40
#define CRM_NODE_MEMBER
Definition: cluster.h:44
guint reap_crm_member(uint32_t id, const char *name)
Remove all peer cache entries matching a node ID and/or uname.
Definition: membership.c:199
uint32_t compressed_size
Definition: internal.h:54
uint32_t counter
Definition: internal.h:50
#define MAX_NAME
Definition: crm.h:44
#define CRM_SYSTEM_CRMD
Definition: crm.h:84
#define CRM_SYSTEM_STONITHD
Definition: crm.h:88
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.
Definition: membership.c:906
#define CRM_SYSTEM_CIB
Definition: crm.h:83
#define CRM_SYSTEM_TENGINE
Definition: crm.h:87
uint32_t get_local_nodeid(cpg_handle_t handle)
Definition: cpg.c:72
bool crm_compress_string(const char *data, int length, int max, char **result, unsigned int *result_len)
Definition: utils.c:2358
#define crm_err(fmt, args...)
Definition: logging.h:248
#define G_PRIORITY_MEDIUM
Definition: mainloop.h:124
char uname[MAX_NAME]
Definition: internal.h:41
#define OFFLINESTATUS
Definition: util.h:49
enum crm_ais_msg_types text2msg_type(const char *text)
Definition: cpg.c:647
#define CRM_BZ2_THRESHOLD
Definition: xml.h:50
char * dump_xml_unformatted(xmlNode *msg)
Definition: xml.c:3987
#define CRM_SYSTEM_LRMD
Definition: crm.h:85
gboolean send_cluster_message_cs(xmlNode *msg, gboolean local, crm_node_t *node, enum crm_ais_msg_types dest)
Definition: cpg.c:506
#define uint32_t
Definition: stdint.in.h:158
#define CRM_ASSERT(expr)
Definition: error.h:35
char data[0]
Definition: internal.h:58
char * state
Definition: cluster.h:81
Wrappers for and extensions to libqb IPC.
int32_t votes
Definition: cluster.h:75
uint32_t pid
Definition: internal.h:37
char * uname
Definition: cluster.h:79
#define cs_repeat(counter, max, code)
Definition: cpg.c:46
AIS_Host host
Definition: internal.h:50
char * crm_itoa(int an_int)
Definition: utils.c:441
#define safe_str_eq(a, b)
Definition: util.h:74
#define ONLINESTATUS
Definition: util.h:48
char * crm_strdup_printf(char const *format,...) __attribute__((__format__(__printf__
crm_node_t * crm_find_peer(unsigned int id, const char *uname)
Definition: membership.c:383
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)
Definition: cpg.c:360
GList * GListPtr
Definition: crm.h:190
long long crm_int_helper(const char *text, char **end_text)
Definition: utils.c:597
#define crm_info(fmt, args...)
Definition: logging.h:251
gboolean cluster_connect_cpg(crm_cluster_t *cluster)
Definition: cpg.c:433
gboolean is_classic_ais_cluster(void)
Definition: cluster.c:613
enum crm_ais_msg_types type
Definition: internal.h:51
enum cluster_type_e get_cluster_type(void)
Definition: cluster.c:502
gboolean local
Definition: internal.h:50