Asterisk - The Open Source Telephony Project GIT-master-85241bd
res_corosync.c
Go to the documentation of this file.
1/*
2 * Asterisk -- An open source telephony toolkit.
3 *
4 * Copyright (C) 2007, Digium, Inc.
5 * Copyright (C) 2012, Russell Bryant
6 *
7 * Russell Bryant <russell@russellbryant.net>
8 *
9 * See http://www.asterisk.org for more information about
10 * the Asterisk project. Please do not directly contact
11 * any of the maintainers of this project for assistance;
12 * the project provides a web site, mailing lists and IRC
13 * channels for your use.
14 *
15 * This program is free software, distributed under the terms of
16 * the GNU General Public License Version 2. See the LICENSE file
17 * at the top of the source tree.
18 */
19
20/*!
21 * \file
22 * \author Russell Bryant <russell@russellbryant.net>
23 *
24 * This module is based on and replaces the previous res_ais module.
25 */
26
27/*** MODULEINFO
28 <depend>corosync</depend>
29 <defaultenabled>no</defaultenabled>
30 <support_level>extended</support_level>
31 ***/
32
33#include "asterisk.h"
34
35#include <corosync/cpg.h>
36#include <corosync/cfg.h>
37
38#include "asterisk/module.h"
39#include "asterisk/logger.h"
41#include "asterisk/config.h"
42#include "asterisk/event.h"
43#include "asterisk/cli.h"
45#include "asterisk/mwi.h"
46#include "asterisk/stasis.h"
50
53
54/*! \brief Timeout for Corosync's poll process */
55#define COROSYNC_POLL_TIMEOUT (10 * 1000)
56
57static void publish_mwi_to_stasis(struct ast_event *event);
60
61/*! \brief Join to corosync */
62static int corosync_node_joined = 0;
63
64/*! \brief All the nodes that we're aware of */
65static struct ao2_container *nodes;
66
67/*! \brief The internal topic used for message forwarding and pings */
69
70/*! \brief Our \ref stasis message router */
72
73/*! \brief Internal accessor for our topic */
74static struct stasis_topic *corosync_topic(void)
75{
77}
78
80 /*! The corosync ID */
81 int id;
82 /*! The Asterisk EID */
83 struct ast_eid eid;
84 /*! The IP address of the node */
86};
87
88/*! \brief Corosync ipc dispatch/request and reply size */
89#define COROSYNC_IPC_BUFFER_SIZE (8192 * 128)
90
91/*! \brief Version of pthread_create to ensure stack is large enough */
92#define corosync_pthread_create_background(a, b, c, d) \
93 ast_pthread_create_stack(a, b, c, d, \
94 (AST_BACKGROUND_STACKSIZE + (3 * COROSYNC_IPC_BUFFER_SIZE)), \
95 __FILE__, __FUNCTION__, __LINE__, #c)
96
98{
99 struct corosync_node *node;
100
102 if (!node) {
103 return NULL;
104 }
105
106 memcpy(&node->eid, (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID), sizeof(node->eid));
109
110 return node;
111}
112
113static int corosync_node_hash_fn(const void *obj, const int flags)
114{
115 const struct corosync_node *node;
116 const int *id;
117
118 switch (flags & OBJ_SEARCH_MASK) {
119 case OBJ_SEARCH_KEY:
120 id = obj;
121 break;
123 node = obj;
124 id = &node->id;
125 break;
126 default:
127 ast_assert(0);
128 return 0;
129 }
130 return *id;
131}
132
133static int corosync_node_cmp_fn(void *obj, void *arg, int flags)
134{
135 struct corosync_node *left = obj;
136 struct corosync_node *right = arg;
137 const int *id = arg;
138 int cmp;
139
140 switch (flags & OBJ_SEARCH_MASK) {
142 id = &right->id;
143 /* Fall through */
144 case OBJ_SEARCH_KEY:
145 cmp = (left->id == *id);
146 break;
148 cmp = (left->id == right->id);
149 break;
150 default:
151 /* Sort can only work on something with a full or partial key. */
152 ast_assert(0);
153 cmp = 1;
154 break;
155 }
156 return cmp ? CMP_MATCH : 0;
157}
158
159
160/*! \brief A payload wrapper around a corosync ping event */
162 /*! The corosync ping event being passed over \ref stasis */
164};
165
166/*! \brief Destructor for the \ref corosync_ping_payload wrapper object */
167static void corosync_ping_payload_dtor(void *obj)
168{
169 struct corosync_ping_payload *payload = obj;
170
171 ast_free(payload->event);
172}
173
174/*! \brief Convert a Corosync PING to a \ref ast_event */
176{
177 struct corosync_ping_payload *payload;
178 struct ast_event *event;
179 struct ast_eid *event_eid;
180
181 if (!message) {
182 return NULL;
183 }
184
185 payload = stasis_message_data(message);
186
187 if (!payload->event) {
188 return NULL;
189 }
190
191 event_eid = (struct ast_eid *)ast_event_get_ie_raw(payload->event, AST_EVENT_IE_EID);
192
194 AST_EVENT_IE_EID, AST_EVENT_IE_PLTYPE_RAW, event_eid, sizeof(*event_eid),
196
197 return event;
198}
199
200STASIS_MESSAGE_TYPE_DEFN_LOCAL(corosync_ping_message_type,
201 .to_event = corosync_ping_to_event, );
202
203/*! \brief Publish a Corosync ping to \ref stasis */
205{
206 struct corosync_ping_payload *payload;
207 struct stasis_message *message;
208 struct ast_eid *event_eid;
209
212
213 if (!corosync_ping_message_type()) {
214 return;
215 }
216
217 payload = ao2_t_alloc(sizeof(*payload), corosync_ping_payload_dtor, "Create ping payload");
218 if (!payload) {
219 return;
220 }
221 event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
223 AST_EVENT_IE_EID, AST_EVENT_IE_PLTYPE_RAW, event_eid, sizeof(*event_eid),
225
226 message = stasis_message_create(corosync_ping_message_type(), payload);
227 if (!message) {
228 ao2_t_ref(payload, -1, "Destroy payload on off nominal");
229 return;
230 }
231
233
234 ao2_t_ref(payload, -1, "Hand ref to stasis");
235 ao2_t_ref(message, -1, "Hand ref to stasis");
236}
237
238static struct {
239 const char *name;
241 unsigned char publish;
242 unsigned char publish_default;
243 unsigned char subscribe;
244 unsigned char subscribe_default;
245 struct stasis_topic *(* topic_fn)(void);
246 struct stasis_cache *(* cache_fn)(void);
247 struct stasis_message_type *(* message_type_fn)(void);
248 void (* publish_to_stasis)(struct ast_event *);
249} event_types[] = {
250 [AST_EVENT_MWI] = { .name = "mwi",
251 .topic_fn = ast_mwi_topic_all,
252 .cache_fn = ast_mwi_state_cache,
253 .message_type_fn = ast_mwi_state_type,
254 .publish_to_stasis = publish_mwi_to_stasis, },
255 [AST_EVENT_DEVICE_STATE_CHANGE] = { .name = "device_state",
256 .topic_fn = ast_device_state_topic_all,
257 .cache_fn = ast_device_state_cache,
258 .message_type_fn = ast_device_state_message_type,
259 .publish_to_stasis = publish_device_state_to_stasis, },
260 [AST_EVENT_PING] = { .name = "ping",
261 .publish_default = 1,
262 .subscribe_default = 1,
263 .topic_fn = corosync_topic,
264 .message_type_fn = corosync_ping_message_type,
265 .publish_to_stasis = publish_corosync_ping_to_stasis, },
266 [AST_EVENT_CLUSTER_DISCOVERY] = { .name = "cluster_discovery",
267 .publish_default = 1,
268 .subscribe_default = 1,
269 .topic_fn = ast_system_topic,
270 .message_type_fn = ast_cluster_discovery_type,
271 .publish_to_stasis = publish_cluster_discovery_to_stasis, },
273
274static struct {
275 pthread_t id;
277 unsigned int stop:1;
278} dispatch_thread = {
279 .id = AST_PTHREADT_NULL,
280 .alert_pipe = { -1, -1 },
282
283static cpg_handle_t cpg_handle;
284static corosync_cfg_handle_t cfg_handle;
285
286#ifdef HAVE_COROSYNC_CFG_STATE_TRACK
287static void cfg_state_track_cb(
288 corosync_cfg_state_notification_buffer_t *notification_buffer,
289 cs_error_t error);
290#endif /* HAVE_COROSYNC_CFG_STATE_TRACK */
291
292static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle,
293 corosync_cfg_shutdown_flags_t flags);
294
295static corosync_cfg_callbacks_t cfg_callbacks = {
296#ifdef HAVE_COROSYNC_CFG_STATE_TRACK
297 .corosync_cfg_state_track_callback = cfg_state_track_cb,
298#endif /* HAVE_COROSYNC_CFG_STATE_TRACK */
299 .corosync_cfg_shutdown_callback = cfg_shutdown_cb,
300};
301
302/*! \brief Publish cluster discovery to \ref stasis */
304{
305 struct ast_json *json;
306 struct ast_json_payload *payload;
307 struct stasis_message *message;
308 char eid[18];
309 const char *addr;
310
311 ast_eid_to_str(eid, sizeof(eid), &node->eid);
312 addr = ast_sockaddr_stringify_addr(&node->addr);
313
314 ast_log(AST_LOG_NOTICE, "Node %u (%s) at %s %s the cluster\n",
315 node->id,
316 eid,
317 addr,
318 joined ? "joined" : "left");
319
320 json = ast_json_pack("{s: s, s: i, s: s, s: i}",
321 "address", addr,
322 "node_id", node->id,
323 "eid", eid,
324 "joined", joined);
325 if (!json) {
326 return;
327 }
328
329 payload = ast_json_payload_create(json);
330 if (!payload) {
331 ast_json_unref(json);
332 return;
333 }
334
336 if (!message) {
337 ast_json_unref(json);
338 ao2_ref(payload, -1);
339 return;
340 }
341
343 ast_json_unref(json);
344 ao2_ref(payload, -1);
345 ao2_ref(message, -1);
346}
347
348static void send_cluster_notify(void);
349
350/*! \brief Publish a received cluster discovery \ref ast_event to \ref stasis */
352{
353 struct corosync_node *node;
355 struct ast_eid *event_eid;
356
358
359 event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
360 if (!event_eid || !ast_eid_cmp(&ast_eid_default, event_eid)) {
361 /* Don't feed events back in that originated locally. */
362 return;
363 }
364
367 if (node) {
368 /* We already know about this node */
370 ao2_ref(node, -1);
371 return;
372 }
373
375 if (!node) {
377 return;
378 }
381
383
384 ao2_ref(node, -1);
385
386 /*
387 * When we get news that someone else has joined, we need to let them
388 * know we exist as well.
389 */
391}
392
393/*! \brief Publish a received MWI \ref ast_event to \ref stasis */
395{
396 const char *mailbox;
397 const char *context;
398 unsigned int new_msgs;
399 unsigned int old_msgs;
400 struct ast_eid *event_eid;
401
403
408 event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
409
411 return;
412 }
413
414 if (new_msgs > INT_MAX) {
415 new_msgs = INT_MAX;
416 }
417
418 if (old_msgs > INT_MAX) {
419 old_msgs = INT_MAX;
420 }
421
422 if (ast_publish_mwi_state_full(mailbox, context, (int)new_msgs,
423 (int)old_msgs, NULL, event_eid)) {
424 char eid[18];
425 ast_eid_to_str(eid, sizeof(eid), event_eid);
426 ast_log(LOG_WARNING, "Failed to publish MWI message for %s@%s from %s\n",
428 }
429}
430
431/*! \brief Publish a received device state \ref ast_event to \ref stasis */
433{
434 const char *device;
436 unsigned int cachable;
437 struct ast_eid *event_eid;
438
440
444 event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
445
446 if (ast_strlen_zero(device)) {
447 return;
448 }
449
450 if (ast_publish_device_state_full(device, state, cachable, event_eid)) {
451 char eid[18];
452 ast_eid_to_str(eid, sizeof(eid), event_eid);
453 ast_log(LOG_WARNING, "Failed to publish device state message for %s from %s\n",
454 device, eid);
455 }
456}
457
458static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name,
459 uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len);
460
461static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name,
462 const struct cpg_address *member_list, size_t member_list_entries,
463 const struct cpg_address *left_list, size_t left_list_entries,
464 const struct cpg_address *joined_list, size_t joined_list_entries);
465
466static cpg_callbacks_t cpg_callbacks = {
467 .cpg_deliver_fn = cpg_deliver_cb,
468 .cpg_confchg_fn = cpg_confchg_cb,
469};
470
471#ifdef HAVE_COROSYNC_CFG_STATE_TRACK
472static void cfg_state_track_cb(
473 corosync_cfg_state_notification_buffer_t *notification_buffer,
474 cs_error_t error)
475{
476}
477#endif /* HAVE_COROSYNC_CFG_STATE_TRACK */
478
479static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle,
480 corosync_cfg_shutdown_flags_t flags)
481{
482}
483
484static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name,
485 uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len)
486{
487 struct ast_event *event;
488 void (*publish_handler)(struct ast_event *) = NULL;
489 enum ast_event_type event_type;
490 struct ast_eid *event_eid;
491
492 if (msg_len < ast_event_minimum_length()) {
493 ast_debug(1, "Ignoring event that's too small. %u < %u\n",
494 (unsigned int) msg_len,
495 (unsigned int) ast_event_minimum_length());
496 return;
497 }
498
499 event_eid = (struct ast_eid *)ast_event_get_ie_raw(msg, AST_EVENT_IE_EID);
500 if (!event_eid || !ast_eid_cmp(&ast_eid_default, event_eid)) {
501 /* Don't feed events back in that originated locally. */
502 return;
503 }
504
505 event_type = ast_event_get_type(msg);
506 if (event_type > AST_EVENT_TOTAL) {
507 /* Egads, we don't support this */
508 return;
509 }
510
512 ast_debug(5, "cpg_deliver_cb rdlock\n");
513 publish_handler = event_types[event_type].publish_to_stasis;
514 if (!event_types[event_type].subscribe || !publish_handler) {
515 /* We are not configured to subscribe to these events or
516 we have no way to publish it internally. */
518 ast_debug(5, "cpg_deliver_cb unlock\n");
519 return;
520 }
522 ast_debug(5, "cpg_deliver_cb unlock\n");
523
524 if (!(event = ast_malloc(msg_len))) {
525 return;
526 }
527
528 memcpy(event, msg, msg_len);
529
530 if (event_type == AST_EVENT_PING) {
531 const struct ast_eid *eid;
532 char buf[128] = "";
533
535 ast_eid_to_str(buf, sizeof(buf), (struct ast_eid *) eid);
536 ast_log(LOG_NOTICE, "Got event PING from server with EID: '%s'\n", buf);
537 }
538 ast_debug(5, "Publishing event %s (%u) to stasis\n",
539 ast_event_get_type_name(event), event_type);
540 publish_handler(event);
542}
543
545{
546 cs_error_t cs_err;
547 struct iovec iov;
548
549 iov.iov_base = (void *)event;
550 iov.iov_len = ast_event_get_size(event);
551
552 ast_debug(5, "Publishing event %s (%u) to corosync\n",
554
555 /* The stasis subscription will only exist if we are configured to publish
556 * these events, so just send away. */
558 ast_debug(5, "publish_event_to_corosync rdlock\n");
559 if ((cs_err = cpg_mcast_joined(cpg_handle, CPG_TYPE_FIFO, &iov, 1)) != CS_OK) {
560 ast_log(LOG_WARNING, "CPG mcast failed (%u) for event %s (%u)\n",
562 }
564 ast_debug(5, "publish_event_to_corosync unlock\n");
565 } else {
566 ast_log(LOG_WARNING, "CPG mcast not executed for event %s (%u): initializing CPG.\n",
568 }
569}
570
572{
573 struct ast_event *event;
574 struct ast_eid *event_eid;
575
577 if (!event) {
578 return;
579 }
580
581 event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
582 if (!event_eid || ast_eid_cmp(&ast_eid_default, event_eid)) {
583 /* If the event didn't originate from this server, don't send it back out. */
585 return;
586 }
587
589 const struct ast_eid *eid;
590 char buf[128] = "";
591
593 ast_eid_to_str(buf, sizeof(buf), (struct ast_eid *) eid);
594 ast_log(LOG_NOTICE, "Sending event PING from this server with EID: '%s'\n", buf);
595 }
596
599}
600
601static void stasis_message_cb(void *data, struct stasis_subscription *sub, struct stasis_message *message)
602{
603 if (!message) {
604 return;
605 }
606
608}
609
610static int dump_cache_cb(void *obj, void *arg, int flags)
611{
612 struct stasis_message *message = obj;
613
614 if (!message) {
615 return 0;
616 }
617
619
620 return 0;
621}
622
623static int clear_node_cache(void *obj, void *arg, int flags)
624{
625 struct stasis_message *cached_msg = obj;
626 struct stasis_topic *topic = arg;
627 struct stasis_message *msg;
628 struct ast_eid *msg_eid;
629
630 if (!cached_msg) {
631 return 0;
632 }
633
634 msg_eid = (struct ast_eid *)stasis_message_eid(cached_msg);
635 if(msg_eid && ast_eid_cmp(&ast_eid_default, msg_eid))
636 {
637 msg = stasis_cache_clear_create(cached_msg);
638 if (msg) {
639 stasis_publish(topic, msg);
640 ao2_cleanup(msg);
641 }
642 }
643
644 return 0;
645}
646
647static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name,
648 const struct cpg_address *member_list, size_t member_list_entries,
649 const struct cpg_address *left_list, size_t left_list_entries,
650 const struct cpg_address *joined_list, size_t joined_list_entries)
651{
652 unsigned int i;
653
654
655 for (i = 0; i < left_list_entries; i++) {
656 const struct cpg_address *cpg_node = &left_list[i];
657 struct corosync_node* node;
658 unsigned int j;
659
660 node = ao2_find(nodes, &cpg_node->nodeid, OBJ_UNLINK | OBJ_SEARCH_KEY);
661 if (!node) {
662 continue;
663 }
664
665 for (j = 0; j < ARRAY_LEN(event_types); j++) {
666 struct ao2_container *messages;
667 int messages_count;
668
670 ast_debug(5, "cpg_confchg_cb rdlock\n");
671 if (!event_types[j].subscribe) {
673 ast_debug(5, "cpg_confchg_cb unlock\n");
674 continue;
675 }
676
679 ast_debug(5, "cpg_confchg_cb unlock\n");
680 continue;
681 }
683 ast_debug(5, "cpg_confchg_cb unlock\n");
684
686
687 messages_count = ao2_container_count(messages);
688 ast_log(LOG_NOTICE, "Clearing %i events of type %s of node %i from stasis cache.\n", messages_count, event_types[j].name, node->id);
690 ast_log(LOG_NOTICE, "Cleared events of type %s from stasis cache.\n", event_types[j].name);
691
692 ao2_t_ref(messages, -1, "Dispose of flushed cache");
693 }
694
696 ao2_ref(node, -1);
697 }
698
699 /* If any new nodes have joined, dump our cache of events we are publishing
700 * that originated from this server. */
701 if (!joined_list_entries) {
702 return;
703 }
704
705 for (i = 0; i < ARRAY_LEN(event_types); i++) {
706 struct ao2_container *messages;
707 int messages_count;
708
710 ast_debug(5, "cpg_confchg_cb rdlock\n");
711 if (!event_types[i].publish) {
713 ast_debug(5, "cpg_confchg_cb unlock\n");
714 continue;
715 }
716
719 ast_debug(5, "cpg_confchg_cb unlock\n");
720 continue;
721 }
723 ast_debug(5, "cpg_confchg_cb unlock\n");
724
726
727 messages_count = ao2_container_count(messages);
728 ast_log(LOG_NOTICE, "Sending %i events of type %s to corosync.\n", messages_count, event_types[i].name);
730 ast_log(LOG_NOTICE, "Sent events of type %s to corosync.\n", event_types[i].name);
731
732 ao2_t_ref(messages, -1, "Dispose of dumped cache");
733 }
734}
735
736/*! \brief Informs the cluster of our EID and our IP addresses */
737static void send_cluster_notify(void)
738{
739 struct ast_event *event;
740 unsigned int node_id;
741 cs_error_t cs_err;
742 corosync_cfg_node_address_t corosync_addr;
743 int num_addrs = 0;
744 struct sockaddr *sa;
745 size_t sa_len;
746 char buf[128];
747 int res;
748
750 ast_debug(5, "send_cluster_notify rdlock\n");
751
752 if ((cs_err = corosync_cfg_local_get(cfg_handle, &node_id)) != CS_OK) {
753 ast_log(LOG_WARNING, "Failed to extract Corosync node ID for this node. Not informing cluster of existance.\n");
754 return;
755 }
756
757 if (((cs_err = corosync_cfg_get_node_addrs(cfg_handle, node_id, 1, &num_addrs, &corosync_addr)) != CS_OK) || (num_addrs < 1)) {
758 ast_log(LOG_WARNING, "Failed to get local Corosync address. Not informing cluster of existance.\n");
759 return;
760 }
761
763 ast_debug(5, "send_cluster_notify unlock\n");
764 }
765
766 sa = (struct sockaddr *)corosync_addr.address;
767 sa_len = (size_t)corosync_addr.address_length;
768 if ((res = getnameinfo(sa, sa_len, buf, sizeof(buf), NULL, 0, NI_NUMERICHOST))) {
769 ast_log(LOG_WARNING, "Failed to determine name of local Corosync address: %s (%d). Not informing cluster of existance.\n",
770 gai_strerror(res), res);
771 return;
772 }
773
780}
781
782static void *dispatch_thread_handler(void *data)
783{
784 cs_error_t cs_err;
785 struct pollfd pfd[3] = {
786 { .events = POLLIN, },
787 { .events = POLLIN, },
788 { .events = POLLIN, },
789 };
790
792 ast_debug(5, "dispatch_thread_handler rdlock\n");
793 if ((cs_err = cpg_fd_get(cpg_handle, &pfd[0].fd)) != CS_OK) {
794 ast_log(LOG_ERROR, "Failed to get CPG fd. This module is now broken.\n");
796 ast_debug(5, "dispatch_thread_handler unlock\n");
797 return NULL;
798 }
799
800 if ((cs_err = corosync_cfg_fd_get(cfg_handle, &pfd[1].fd)) != CS_OK) {
801 ast_log(LOG_ERROR, "Failed to get CFG fd. This module is now broken.\n");
803 ast_debug(5, "dispatch_thread_handler unlock\n");
804 return NULL;
805 }
806
807 pfd[2].fd = dispatch_thread.alert_pipe[0];
809 ast_debug(5, "dispatch_thread_handler unlock\n");
810 } else {
811 ast_log(LOG_ERROR, "Failed to get fd: initializing CPG. This module is now broken.\n");
812 return NULL;
813 }
815 while (!dispatch_thread.stop) {
816 int res;
817
818 cs_err = CS_OK;
819
820 pfd[0].revents = 0;
821 pfd[1].revents = 0;
822 pfd[2].revents = 0;
823
824 res = ast_poll(pfd, ARRAY_LEN(pfd), COROSYNC_POLL_TIMEOUT);
825 if (res == -1 && errno != EINTR && errno != EAGAIN) {
826 ast_log(LOG_ERROR, "poll() error: %s (%d)\n", strerror(errno), errno);
827 cs_err = CS_ERR_BAD_HANDLE;
828 } else if (res == 0) {
829 unsigned int local_nodeid;
830
832 ast_debug(5, "dispatch_thread_handler rdlock\n");
833 if ((cs_err = cpg_local_get(cpg_handle, &local_nodeid)) == CS_OK) {
834 struct cpg_name name;
835 struct cpg_address address[CPG_MEMBERS_MAX];
836 int entries = CPG_MEMBERS_MAX;
837
838 ast_copy_string(name.value, "asterisk", sizeof(name.value));
839 name.length = strlen(name.value);
840 if ((cs_err = cpg_membership_get(cpg_handle, &name, address, &entries)) == CS_OK) {
841 int i;
842 int found = 0;
843
844 ast_debug(1, "CPG group has %i node membership\n", entries);
845 for (i = 0; (i < entries) && !found; i++) {
846 if (address[i].nodeid == local_nodeid)
847 found = 1;
848 }
849 if (!found) {
850 ast_log(LOG_WARNING, "Failed to check CPG node membership\n");
852 cs_err = CS_ERR_BAD_HANDLE;
853 }
854 } else {
855 ast_log(LOG_WARNING, "Failed to get CPG node membership: %u\n", cs_err);
857 cs_err = CS_ERR_BAD_HANDLE;
858 }
859 } else {
860 ast_log(LOG_WARNING, "Failed to get CPG local node id: %u\n", cs_err);
862 cs_err = CS_ERR_BAD_HANDLE;
863 }
865 ast_debug(5, "dispatch_thread_handler unlock\n");
866 } else {
867 ast_log(LOG_WARNING, "Failed to check CPG node membership: initializing CPG.\n");
869 cs_err = CS_ERR_BAD_HANDLE;
870 }
871 } else {
873 ast_debug(5, "dispatch_thread_handler rdlock\n");
874 if (pfd[0].revents & POLLIN) {
875 if ((cs_err = cpg_dispatch(cpg_handle, CS_DISPATCH_ALL)) != CS_OK) {
876 ast_log(LOG_WARNING, "Failed CPG dispatch: %u\n", cs_err);
877 }
878 }
879
880 if (pfd[1].revents & POLLIN) {
881 if ((cs_err = corosync_cfg_dispatch(cfg_handle, CS_DISPATCH_ALL)) != CS_OK) {
882 ast_log(LOG_WARNING, "Failed CFG dispatch: %u\n", cs_err);
883 }
884 }
886 ast_debug(5, "dispatch_thread_handler unlock\n");
887 } else {
888 ast_log(LOG_WARNING, "Failed to dispatch: initializing CPG.\n");
889 }
890 }
891 if (cs_err == CS_ERR_LIBRARY || cs_err == CS_ERR_BAD_HANDLE) {
892
893 /* If corosync gets restarted out from under Asterisk, try to recover. */
894
895 ast_log(LOG_NOTICE, "Attempting to recover from corosync failure.\n");
896
898 struct cpg_name name;
899 ast_debug(5, "dispatch_thread_handler wrlock\n");
900
902 if (cpg_handle && (cs_err = cpg_finalize(cpg_handle)) != CS_OK) {
903 ast_log(LOG_ERROR, "Failed to finalize cpg (%d)\n", (int) cs_err);
904 }
905
906 if (cfg_handle && (cs_err = corosync_cfg_finalize(cfg_handle)) != CS_OK) {
907 ast_log(LOG_ERROR, "Failed to finalize cfg (%d)\n", (int) cs_err);
908 }
909
910 if ((cs_err = corosync_cfg_initialize(&cfg_handle, &cfg_callbacks)) != CS_OK) {
911 ast_log(LOG_ERROR, "Failed to initialize cfg (%d)\n", (int) cs_err);
913 ast_debug(5, "dispatch_thread_handler unlock\n");
914 sleep(5);
915 continue;
916 }
917
918 if ((cs_err = cpg_initialize(&cpg_handle, &cpg_callbacks) != CS_OK)) {
919 ast_log(LOG_ERROR, "Failed to initialize cpg (%d)\n", (int) cs_err);
921 ast_debug(5, "dispatch_thread_handler unlock\n");
922 sleep(5);
923 continue;
924 }
925
926 if ((cs_err = cpg_fd_get(cpg_handle, &pfd[0].fd)) != CS_OK) {
927 ast_log(LOG_ERROR, "Failed to get CPG fd.\n");
929 ast_debug(5, "dispatch_thread_handler unlock\n");
930 sleep(5);
931 continue;
932 }
933
934 if ((cs_err = corosync_cfg_fd_get(cfg_handle, &pfd[1].fd)) != CS_OK) {
935 ast_log(LOG_ERROR, "Failed to get CFG fd.\n");
937 ast_debug(5, "dispatch_thread_handler unlock\n");
938 sleep(5);
939 continue;
940 }
941
942 ast_copy_string(name.value, "asterisk", sizeof(name.value));
943 name.length = strlen(name.value);
944 if ((cs_err = cpg_join(cpg_handle, &name)) != CS_OK) {
945 ast_log(LOG_ERROR, "Failed to join cpg (%d)\n", (int) cs_err);
947 ast_debug(5, "dispatch_thread_handler unlock\n");
948 sleep(5);
949 continue;
950 }
953 ast_debug(5, "dispatch_thread_handler unlock\n");
954 ast_log(LOG_NOTICE, "Corosync recovery complete.\n");
956 } else {
957 ast_log(LOG_NOTICE, "Failed to recover from corosync failure: initializing CPG.\n");
958 }
959 }
960 }
961
962 return NULL;
963}
964
965static char *corosync_show_members(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
966{
967 cs_error_t cs_err;
968 cpg_iteration_handle_t cpg_iter;
969 struct cpg_iteration_description_t cpg_desc;
970 unsigned int i;
971
972 switch (cmd) {
973 case CLI_INIT:
974 e->command = "corosync show members";
975 e->usage =
976 "Usage: corosync show members\n"
977 " Show corosync cluster members\n";
978 return NULL;
979
980 case CLI_GENERATE:
981 return NULL; /* no completion */
982 }
983
984 if (a->argc != e->args) {
985 return CLI_SHOWUSAGE;
986 }
987
989 ast_debug(5, "corosync_show_members rdlock\n");
990 cs_err = cpg_iteration_initialize(cpg_handle, CPG_ITERATION_ALL, NULL, &cpg_iter);
991
992 if (cs_err != CS_OK) {
993 ast_cli(a->fd, "Failed to initialize CPG iterator: %u.\n", cs_err);
994 cpg_iteration_finalize(cpg_iter);
996 ast_debug(5, "corosync_show_members unlock\n");
997 return CLI_FAILURE;
998 }
999
1000 ast_cli(a->fd, "\n"
1001 "=============================================================\n"
1002 "=== Cluster members =========================================\n"
1003 "=============================================================\n"
1004 "===\n");
1005
1006 for (i = 1, cs_err = cpg_iteration_next(cpg_iter, &cpg_desc);
1007 cs_err == CS_OK;
1008 cs_err = cpg_iteration_next(cpg_iter, &cpg_desc), i++) {
1009 #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
1010 corosync_cfg_node_address_t addrs[8];
1011 int num_addrs = 0;
1012 unsigned int j;
1013 #endif
1014
1015 ast_cli(a->fd, "=== Node %u\n", i);
1016 ast_cli(a->fd, "=== --> Group: %s\n", cpg_desc.group.value);
1017
1018 #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
1019 /*
1020 * Corosync 2.x cfg lib needs to allocate 1M on stack after calling
1021 * corosync_cfg_get_node_addrs. netconsole thread has allocated only 0.5M
1022 * resulting in crash.
1023 */
1024 cs_err = corosync_cfg_get_node_addrs(cfg_handle, cpg_desc.nodeid,
1025 ARRAY_LEN(addrs), &num_addrs, addrs);
1026 if (cs_err != CS_OK) {
1027 ast_log(LOG_WARNING, "Failed to get node addresses\n");
1028 continue;
1029 }
1030
1031 for (j = 0; j < num_addrs; j++) {
1032 struct sockaddr *sa = (struct sockaddr *) addrs[j].address;
1033 size_t sa_len = (size_t) addrs[j].address_length;
1034 char buf[128];
1035
1036 getnameinfo(sa, sa_len, buf, sizeof(buf), NULL, 0, NI_NUMERICHOST);
1037
1038 ast_cli(a->fd, "=== --> Address %u: %s\n", j + 1, buf);
1039 }
1040 #else
1041 ast_cli(a->fd, "=== --> Nodeid: %"PRIu32"\n", cpg_desc.nodeid);
1042 #endif
1043 }
1044
1045 ast_cli(a->fd, "===\n"
1046 "=============================================================\n"
1047 "\n");
1048
1049 cpg_iteration_finalize(cpg_iter);
1051 ast_debug(5, "corosync_show_members unlock\n");
1052 } else {
1053 ast_cli(a->fd, "Failed to initialize CPG iterator: initializing CPG.\n");
1054 }
1055
1056 return CLI_SUCCESS;
1057}
1058
1059static char *corosync_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
1060{
1061 struct ast_event *event;
1062
1063 switch (cmd) {
1064 case CLI_INIT:
1065 e->command = "corosync ping";
1066 e->usage =
1067 "Usage: corosync ping\n"
1068 " Send a test ping to the cluster.\n"
1069 "A NOTICE will be in the log for every ping received\n"
1070 "on a server.\n If you send a ping, you should see a NOTICE\n"
1071 "in the log for every server in the cluster.\n";
1072 return NULL;
1073
1074 case CLI_GENERATE:
1075 return NULL; /* no completion */
1076 }
1077
1078 if (a->argc != e->args) {
1079 return CLI_SHOWUSAGE;
1080 }
1081
1083
1084 if (!event) {
1085 return CLI_FAILURE;
1086 }
1087
1088 event_types[AST_EVENT_PING].publish_to_stasis(event);
1089
1091 return CLI_SUCCESS;
1092}
1093
1094static char *corosync_show_config(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
1095{
1096 unsigned int i;
1097
1098 switch (cmd) {
1099 case CLI_INIT:
1100 e->command = "corosync show config";
1101 e->usage =
1102 "Usage: corosync show config\n"
1103 " Show configuration loaded from res_corosync.conf\n";
1104 return NULL;
1105
1106 case CLI_GENERATE:
1107 return NULL; /* no completion */
1108 }
1109
1110 if (a->argc != e->args) {
1111 return CLI_SHOWUSAGE;
1112 }
1113
1114 ast_cli(a->fd, "\n"
1115 "=============================================================\n"
1116 "=== res_corosync config =====================================\n"
1117 "=============================================================\n"
1118 "===\n");
1119
1121 ast_debug(5, "corosync_show_config rdlock\n");
1122 for (i = 0; i < ARRAY_LEN(event_types); i++) {
1123 if (event_types[i].publish) {
1124 ast_cli(a->fd, "=== ==> Publishing Event Type: %s\n",
1125 event_types[i].name);
1126 }
1127 if (event_types[i].subscribe) {
1128 ast_cli(a->fd, "=== ==> Subscribing to Event Type: %s\n",
1129 event_types[i].name);
1130 }
1131 }
1133 ast_debug(5, "corosync_show_config unlock\n");
1134
1135 ast_cli(a->fd, "===\n"
1136 "=============================================================\n"
1137 "\n");
1138
1139 return CLI_SUCCESS;
1140}
1141
1142static struct ast_cli_entry corosync_cli[] = {
1143 AST_CLI_DEFINE(corosync_show_config, "Show configuration"),
1144 AST_CLI_DEFINE(corosync_show_members, "Show cluster members"),
1145 AST_CLI_DEFINE(corosync_ping, "Send a test ping to the cluster"),
1146};
1147
1148enum {
1151};
1152
1153static int set_event(const char *event_type, int pubsub)
1154{
1155 unsigned int i;
1156
1157 for (i = 0; i < ARRAY_LEN(event_types); i++) {
1158 if (!event_types[i].name || strcasecmp(event_type, event_types[i].name)) {
1159 continue;
1160 }
1161
1162 switch (pubsub) {
1163 case PUBLISH:
1164 event_types[i].publish = 1;
1165 break;
1166 case SUBSCRIBE:
1167 event_types[i].subscribe = 1;
1168 break;
1169 }
1170
1171 break;
1172 }
1173
1174 return (i == ARRAY_LEN(event_types)) ? -1 : 0;
1175}
1176
1177static int load_general_config(struct ast_config *cfg)
1178{
1179 struct ast_variable *v;
1180 int res = 0;
1181 unsigned int i;
1182
1184 ast_debug(5, "load_general_config wrlock\n");
1185
1186 for (i = 0; i < ARRAY_LEN(event_types); i++) {
1187 event_types[i].publish = event_types[i].publish_default;
1188 event_types[i].subscribe = event_types[i].subscribe_default;
1189 }
1190
1191 for (v = ast_variable_browse(cfg, "general"); v && !res; v = v->next) {
1192 if (!strcasecmp(v->name, "publish_event")) {
1193 res = set_event(v->value, PUBLISH);
1194 } else if (!strcasecmp(v->name, "subscribe_event")) {
1195 res = set_event(v->value, SUBSCRIBE);
1196 } else {
1197 ast_log(LOG_WARNING, "Unknown option '%s'\n", v->name);
1198 }
1199 }
1200
1201 for (i = 0; i < ARRAY_LEN(event_types); i++) {
1202 if (event_types[i].publish && !event_types[i].sub) {
1204 corosync_topic());
1208 NULL);
1209 } else if (!event_types[i].publish && event_types[i].sub) {
1213 }
1214 }
1215
1217 ast_debug(5, "load_general_config unlock\n");
1218
1219 return res;
1220}
1221
1222static int load_config(unsigned int reload)
1223{
1224 static const char filename[] = "res_corosync.conf";
1225 struct ast_config *cfg;
1226 const char *cat = NULL;
1227 struct ast_flags config_flags = { 0 };
1228 int res = 0;
1229
1230 cfg = ast_config_load(filename, config_flags);
1231
1233 return -1;
1234 }
1235
1236 while ((cat = ast_category_browse(cfg, cat))) {
1237 if (!strcasecmp(cat, "general")) {
1238 res = load_general_config(cfg);
1239 } else {
1240 ast_log(LOG_WARNING, "Unknown configuration section '%s'\n", cat);
1241 }
1242 }
1243
1244 ast_config_destroy(cfg);
1245
1246 return res;
1247}
1248
1249static void cleanup_module(void)
1250{
1251 cs_error_t cs_err;
1252 unsigned int i;
1253
1254 if (stasis_router) {
1255
1256 /* Unsubscribe all topic forwards and cancel all message routes */
1257 for (i = 0; i < ARRAY_LEN(event_types); i++) {
1258 struct ao2_container *messages = NULL;
1259 int messages_count;
1260 unsigned char subscribe = 0;
1261
1263 ast_debug(5, "cleanup_module wrlock\n");
1264 subscribe = event_types[i].subscribe;
1265
1266 if (event_types[i].sub) {
1269 }
1270 event_types[i].publish = 0;
1271 event_types[i].subscribe = 0;
1273 ast_debug(5, "cleanup_module unlock\n");
1274
1277 messages_count = ao2_container_count(messages);
1278 ast_log(LOG_NOTICE, "Clearing %i events of type %s of other nodes from stasis cache.\n", messages_count, event_types[i].name);
1280 ast_log(LOG_NOTICE, "Cleared events of type %s from stasis cache.\n", event_types[i].name);
1281 ao2_t_ref(messages, -1, "Dispose of flushed cache");
1282 }
1283 }
1284
1287 }
1288
1290 ao2_t_ref(corosync_aggregate_topic, -1, "Dispose of topic on cleanup");
1292 }
1293
1294 STASIS_MESSAGE_TYPE_CLEANUP(corosync_ping_message_type);
1295
1297 char meepmeep = 'x';
1298 dispatch_thread.stop = 1;
1299 if (ast_carefulwrite(dispatch_thread.alert_pipe[1], &meepmeep, 1,
1300 5000) == -1) {
1301 ast_log(LOG_ERROR, "Failed to write to pipe: %s (%d)\n",
1302 strerror(errno), errno);
1303 }
1304 pthread_join(dispatch_thread.id, NULL);
1305 }
1306
1307 if (dispatch_thread.alert_pipe[0] != -1) {
1308 close(dispatch_thread.alert_pipe[0]);
1309 dispatch_thread.alert_pipe[0] = -1;
1310 }
1311
1312 if (dispatch_thread.alert_pipe[1] != -1) {
1313 close(dispatch_thread.alert_pipe[1]);
1314 dispatch_thread.alert_pipe[1] = -1;
1315 }
1316
1318 ast_debug(5, "cleanup_module wrlock\n");
1319 if (cpg_handle && (cs_err = cpg_finalize(cpg_handle)) != CS_OK) {
1320 ast_log(LOG_ERROR, "Failed to finalize cpg (%d)\n", (int) cs_err);
1321 }
1322 cpg_handle = 0;
1323
1324 if (cfg_handle && (cs_err = corosync_cfg_finalize(cfg_handle)) != CS_OK) {
1325 ast_log(LOG_ERROR, "Failed to finalize cfg (%d)\n", (int) cs_err);
1326 }
1327 cfg_handle = 0;
1330 ast_debug(5, "cleanup_module unlock\n");
1331 }
1333 nodes = NULL;
1334}
1335
1336static int load_module(void)
1337{
1338 cs_error_t cs_err;
1339 struct cpg_name name;
1340
1342 ast_log(LOG_ERROR, "Entity ID is not set.\n");
1344 }
1345
1348 if (!nodes) {
1349 goto failed;
1350 }
1351
1352 corosync_aggregate_topic = stasis_topic_create("corosync:aggregator");
1354 ast_log(AST_LOG_ERROR, "Failed to create stasis topic for corosync\n");
1355 goto failed;
1356 }
1357
1359 if (!stasis_router) {
1360 ast_log(AST_LOG_ERROR, "Failed to create message router for corosync topic\n");
1361 goto failed;
1362 }
1365
1366 if (STASIS_MESSAGE_TYPE_INIT(corosync_ping_message_type) != 0) {
1367 ast_log(AST_LOG_ERROR, "Failed to initialize corosync ping message type\n");
1368 goto failed;
1369 }
1370
1371 if (load_config(0)) {
1372 /* simply not configured is not a fatal error */
1373 goto failed;
1374 }
1375
1378 ast_debug(5, "load_module wrlock\n");
1379 if ((cs_err = corosync_cfg_initialize(&cfg_handle, &cfg_callbacks)) != CS_OK) {
1380 ast_log(LOG_ERROR, "Failed to initialize cfg: (%d)\n", (int) cs_err);
1382 ast_debug(5, "load_module unlock\n");
1383 goto failed;
1384 }
1385
1386 if ((cs_err = cpg_initialize(&cpg_handle, &cpg_callbacks)) != CS_OK) {
1387 ast_log(LOG_ERROR, "Failed to initialize cpg: (%d)\n", (int) cs_err);
1389 ast_debug(5, "load_module unlock\n");
1390 goto failed;
1391 }
1392
1393 ast_copy_string(name.value, "asterisk", sizeof(name.value));
1394 name.length = strlen(name.value);
1395
1396 if ((cs_err = cpg_join(cpg_handle, &name)) != CS_OK) {
1397 ast_log(LOG_ERROR, "Failed to join: (%d)\n", (int) cs_err);
1399 ast_debug(5, "load_module unlock\n");
1400 goto failed;
1401 }
1402
1403 if (pipe(dispatch_thread.alert_pipe) == -1) {
1404 ast_log(LOG_ERROR, "Failed to create alert pipe: %s (%d)\n",
1405 strerror(errno), errno);
1407 ast_debug(5, "load_module unlock\n");
1408 goto failed;
1409 }
1411
1413 ast_debug(5, "load_module unlock\n");
1416 ast_log(LOG_ERROR, "Error starting CPG dispatch thread.\n");
1417 goto failed;
1418 }
1419
1421 } else {
1422 goto failed;
1423 }
1424
1426
1427failed:
1429
1431}
1432
1433static int unload_module(void)
1434{
1436
1438
1439 return 0;
1440}
1441
Asterisk main include file. File version handling, generic pbx functions.
#define ast_free(a)
Definition: astmm.h:180
#define ast_malloc(len)
A wrapper for malloc()
Definition: astmm.h:191
#define ast_log
Definition: astobj2.c:42
#define ao2_t_ref(o, delta, tag)
Definition: astobj2.h:460
@ CMP_MATCH
Definition: astobj2.h:1027
@ AO2_ALLOC_OPT_LOCK_NOLOCK
Definition: astobj2.h:367
@ AO2_ALLOC_OPT_LOCK_MUTEX
Definition: astobj2.h:363
#define ao2_callback(c, flags, cb_fn, arg)
ao2_callback() is a generic function that applies cb_fn() to all objects in a container,...
Definition: astobj2.h:1693
int ao2_container_count(struct ao2_container *c)
Returns the number of elements in a container.
#define ao2_cleanup(obj)
Definition: astobj2.h:1934
#define ao2_link_flags(container, obj, flags)
Add an object to a container.
Definition: astobj2.h:1554
#define ao2_find(container, arg, flags)
Definition: astobj2.h:1736
#define ao2_unlock(a)
Definition: astobj2.h:729
#define ao2_lock(a)
Definition: astobj2.h:717
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
#define ao2_alloc_options(data_size, destructor_fn, options)
Definition: astobj2.h:404
#define ao2_t_alloc(data_size, destructor_fn, debug_msg)
Definition: astobj2.h:407
@ OBJ_SEARCH_PARTIAL_KEY
The arg parameter is a partial search key similar to OBJ_SEARCH_KEY.
Definition: astobj2.h:1116
@ OBJ_SEARCH_OBJECT
The arg parameter is an object of the same type.
Definition: astobj2.h:1087
@ OBJ_NOLOCK
Assume that the ao2_container is already locked.
Definition: astobj2.h:1063
@ OBJ_NODATA
Definition: astobj2.h:1044
@ OBJ_SEARCH_MASK
Search option field mask.
Definition: astobj2.h:1072
@ OBJ_UNLINK
Definition: astobj2.h:1039
@ OBJ_SEARCH_KEY
The arg parameter is a search key, but is not an object.
Definition: astobj2.h:1101
#define ao2_container_alloc_hash(ao2_options, container_options, n_buckets, hash_fn, sort_fn, cmp_fn)
Allocate and initialize a hash container with the desired number of buckets.
Definition: astobj2.h:1303
enum cc_state state
Definition: ccss.c:393
Standard Command Line Interface.
#define CLI_SHOWUSAGE
Definition: cli.h:45
#define CLI_SUCCESS
Definition: cli.h:44
int ast_cli_unregister_multiple(struct ast_cli_entry *e, int len)
Unregister multiple commands.
Definition: clicompat.c:30
#define AST_CLI_DEFINE(fn, txt,...)
Definition: cli.h:197
void ast_cli(int fd, const char *fmt,...)
Definition: clicompat.c:6
@ CLI_INIT
Definition: cli.h:152
@ CLI_GENERATE
Definition: cli.h:153
#define CLI_FAILURE
Definition: cli.h:46
#define ast_cli_register_multiple(e, len)
Register multiple commands.
Definition: cli.h:265
Device state management.
struct stasis_message_type * ast_device_state_message_type(void)
Get the Stasis message type for device state messages.
struct stasis_cache * ast_device_state_cache(void)
Backend cache for ast_device_state_topic_cached()
Definition: devicestate.c:673
struct stasis_topic * ast_device_state_topic_all(void)
Get the Stasis topic for device state messages.
Definition: devicestate.c:668
ast_device_state
Device States.
Definition: devicestate.h:52
int ast_publish_device_state_full(const char *device, enum ast_device_state state, enum ast_devstate_cache cachable, struct ast_eid *eid)
Publish a device state update with EID.
Definition: devicestate.c:709
char buf[BUFSIZE]
Definition: eagi_proxy.c:66
size_t ast_event_minimum_length(void)
Get the minimum length of an ast_event.
Definition: event.c:529
struct ast_event * ast_event_new(enum ast_event_type event_type,...)
Create a new event.
Definition: event.c:402
size_t ast_event_get_size(const struct ast_event *event)
Get the size of an event.
Definition: event.c:228
uint32_t ast_event_get_ie_uint(const struct ast_event *event, enum ast_event_ie_type ie_type)
Get the value of an information element that has an integer payload.
Definition: event.c:293
void ast_event_destroy(struct ast_event *event)
Destroy an event.
Definition: event.c:524
const char * ast_event_get_ie_str(const struct ast_event *event, enum ast_event_ie_type ie_type)
Get the value of an information element that has a string payload.
Definition: event.c:302
const void * ast_event_get_ie_raw(const struct ast_event *event, enum ast_event_ie_type ie_type)
Get the value of an information element that has a raw payload.
Definition: event.c:311
const char * ast_event_get_type_name(const struct ast_event *event)
Get the string representation of the type of the given event.
Definition: event.c:194
enum ast_event_type ast_event_get_type(const struct ast_event *event)
Get the type for an event.
Definition: event.c:288
@ AST_EVENT_IE_END
Definition: event_defs.h:70
@ AST_EVENT_IE_STATE
Generic State IE Used by AST_EVENT_DEVICE_STATE_CHANGE Payload type: UINT The actual state values dep...
Definition: event_defs.h:121
@ AST_EVENT_IE_CONTEXT
Context IE Used by AST_EVENT_MWI Payload type: str.
Definition: event_defs.h:127
@ AST_EVENT_IE_LOCAL_ADDR
Definition: event_defs.h:281
@ AST_EVENT_IE_DEVICE
Device Name Used by AST_EVENT_DEVICE_STATE_CHANGE Payload type: STR.
Definition: event_defs.h:113
@ AST_EVENT_IE_EID
Entity ID Used by All events Payload type: RAW This IE indicates which server the event originated fr...
Definition: event_defs.h:272
@ AST_EVENT_IE_MAILBOX
Mailbox name.
Definition: event_defs.h:89
@ AST_EVENT_IE_CACHABLE
Event non-cacheability flag Used by: All events Payload type: UINT.
Definition: event_defs.h:306
@ AST_EVENT_IE_NODE_ID
Cluster node ID Used by: Corosync Payload type: UINT.
Definition: event_defs.h:313
@ AST_EVENT_IE_OLDMSGS
Number of Used by: AST_EVENT_MWI Payload type: UINT.
Definition: event_defs.h:83
@ AST_EVENT_IE_NEWMSGS
Number of new messages Used by: AST_EVENT_MWI Payload type: UINT.
Definition: event_defs.h:77
ast_event_type
Definition: event_defs.h:28
@ AST_EVENT_TOTAL
Definition: event_defs.h:64
@ AST_EVENT_PING
Definition: event_defs.h:60
@ AST_EVENT_DEVICE_STATE_CHANGE
Definition: event_defs.h:48
@ AST_EVENT_MWI
Definition: event_defs.h:38
@ AST_EVENT_CLUSTER_DISCOVERY
Definition: event_defs.h:62
@ AST_EVENT_IE_PLTYPE_RAW
Definition: event_defs.h:330
@ AST_EVENT_IE_PLTYPE_UINT
Definition: event_defs.h:326
@ AST_EVENT_IE_PLTYPE_STR
Definition: event_defs.h:328
char * address
Definition: f2c.h:59
Configuration File Parser.
#define ast_config_load(filename, flags)
Load a config file.
char * ast_category_browse(struct ast_config *config, const char *prev_name)
Browse categories.
Definition: extconf.c:3326
#define CONFIG_STATUS_FILEMISSING
#define CONFIG_STATUS_FILEINVALID
void ast_config_destroy(struct ast_config *cfg)
Destroys a config.
Definition: extconf.c:1289
struct ast_variable * ast_variable_browse(const struct ast_config *config, const char *category_name)
Definition: extconf.c:1215
Support for logging to various files, console and syslog Configuration in file logger....
#define AST_LOG_ERROR
#define ast_debug(level,...)
Log a DEBUG message.
#define LOG_ERROR
#define AST_LOG_NOTICE
#define LOG_NOTICE
#define LOG_WARNING
void ast_json_unref(struct ast_json *value)
Decrease refcount on value. If refcount reaches zero, value is freed.
Definition: json.c:73
struct ast_json_payload * ast_json_payload_create(struct ast_json *json)
Create an ao2 object to pass json blobs as data payloads for stasis.
Definition: json.c:756
struct ast_json * ast_json_pack(char const *format,...)
Helper for creating complex JSON values.
Definition: json.c:612
#define ast_rwlock_wrlock(a)
Definition: lock.h:236
#define AST_PTHREADT_NULL
Definition: lock.h:66
#define ast_rwlock_rdlock(a)
Definition: lock.h:235
#define AST_RWLOCK_DEFINE_STATIC(rwlock)
Definition: lock.h:543
#define ast_rwlock_trywrlock(a)
Definition: lock.h:238
#define ast_rwlock_unlock(a)
Definition: lock.h:234
#define ast_rwlock_tryrdlock(a)
Definition: lock.h:237
int errno
Asterisk module definitions.
#define ASTERISK_GPL_KEY
The text the key() function should return.
Definition: module.h:46
@ AST_MODULE_LOAD_SUCCESS
Definition: module.h:70
@ AST_MODULE_LOAD_DECLINE
Module has failed to load, may be in an inconsistent state.
Definition: module.h:78
Asterisk MWI API.
struct stasis_message_type * ast_mwi_state_type(void)
Get the Stasis Message Bus API message type for MWI messages.
struct stasis_cache * ast_mwi_state_cache(void)
Backend cache for ast_mwi_topic_cached().
Definition: mwi.c:94
int ast_publish_mwi_state_full(const char *mailbox, const char *context, int new_msgs, int old_msgs, const char *channel_id, struct ast_eid *eid)
Publish a MWI state update via stasis with all parameters.
Definition: mwi.c:393
struct stasis_topic * ast_mwi_topic_all(void)
Get the Stasis Message Bus API topic for MWI messages.
Definition: mwi.c:89
int ast_sockaddr_parse(struct ast_sockaddr *addr, const char *str, int flags)
Parse an IPv4 or IPv6 address string.
Definition: netsock2.c:230
static char * ast_sockaddr_stringify_addr(const struct ast_sockaddr *addr)
Wrapper around ast_sockaddr_stringify_fmt() to return an address only.
Definition: netsock2.h:286
#define ast_poll(a, b, c)
Definition: poll-compat.h:88
static int reload(void)
static char * corosync_show_config(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name, uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len)
Definition: res_corosync.c:484
static void publish_event_to_corosync(struct ast_event *event)
Definition: res_corosync.c:544
static struct ast_event * corosync_ping_to_event(struct stasis_message *message)
Convert a Corosync PING to a ast_event.
Definition: res_corosync.c:175
void(* publish_to_stasis)(struct ast_event *)
Definition: res_corosync.c:248
unsigned int stop
Definition: res_corosync.c:277
static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle, corosync_cfg_shutdown_flags_t flags)
Definition: res_corosync.c:479
pthread_t id
Definition: res_corosync.c:275
static int clear_node_cache(void *obj, void *arg, int flags)
Definition: res_corosync.c:623
static cpg_callbacks_t cpg_callbacks
Definition: res_corosync.c:466
static void publish_to_corosync(struct stasis_message *message)
Definition: res_corosync.c:571
static void publish_cluster_discovery_to_stasis_full(struct corosync_node *node, int joined)
Publish cluster discovery to Stasis Message Bus API.
Definition: res_corosync.c:303
static void * dispatch_thread_handler(void *data)
Definition: res_corosync.c:782
#define COROSYNC_POLL_TIMEOUT
Timeout for Corosync's poll process.
Definition: res_corosync.c:55
static struct ao2_container * nodes
All the nodes that we're aware of.
Definition: res_corosync.c:65
static void publish_corosync_ping_to_stasis(struct ast_event *event)
Publish a Corosync ping to Stasis Message Bus API.
Definition: res_corosync.c:204
static struct corosync_node * corosync_node_alloc(struct ast_event *event)
Definition: res_corosync.c:97
struct stasis_message_type *(* message_type_fn)(void)
Definition: res_corosync.c:247
static struct ast_cli_entry corosync_cli[]
unsigned char publish_default
Definition: res_corosync.c:242
#define corosync_pthread_create_background(a, b, c, d)
Version of pthread_create to ensure stack is large enough.
Definition: res_corosync.c:92
int alert_pipe[2]
Definition: res_corosync.c:276
static void cleanup_module(void)
static int corosync_node_joined
Join to corosync.
Definition: res_corosync.c:62
static void corosync_ping_payload_dtor(void *obj)
Destructor for the corosync_ping_payload wrapper object.
Definition: res_corosync.c:167
static struct stasis_message_router * stasis_router
Our Stasis Message Bus API message router.
Definition: res_corosync.c:71
static corosync_cfg_callbacks_t cfg_callbacks
Definition: res_corosync.c:295
static struct @433 dispatch_thread
static void send_cluster_notify(void)
Informs the cluster of our EID and our IP addresses.
Definition: res_corosync.c:737
static int load_general_config(struct ast_config *cfg)
static void publish_device_state_to_stasis(struct ast_event *event)
Publish a received device state ast_event to Stasis Message Bus API.
Definition: res_corosync.c:432
unsigned char subscribe
Definition: res_corosync.c:243
static struct stasis_topic * corosync_topic(void)
Internal accessor for our topic.
Definition: res_corosync.c:74
static int corosync_node_hash_fn(const void *obj, const int flags)
Definition: res_corosync.c:113
struct stasis_cache *(* cache_fn)(void)
Definition: res_corosync.c:246
struct stasis_forward * sub
Definition: res_corosync.c:240
const char * name
Definition: res_corosync.c:239
unsigned char subscribe_default
Definition: res_corosync.c:244
struct stasis_topic *(* topic_fn)(void)
Definition: res_corosync.c:245
static char * corosync_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static struct stasis_topic * corosync_aggregate_topic
The internal topic used for message forwarding and pings.
Definition: res_corosync.c:68
unsigned char publish
Definition: res_corosync.c:241
static ast_rwlock_t event_types_lock
Definition: res_corosync.c:51
@ SUBSCRIBE
@ PUBLISH
static void publish_cluster_discovery_to_stasis(struct ast_event *event)
Publish a received cluster discovery ast_event to Stasis Message Bus API.
Definition: res_corosync.c:351
static char * corosync_show_members(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
Definition: res_corosync.c:965
static int load_module(void)
static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name, 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: res_corosync.c:647
STASIS_MESSAGE_TYPE_DEFN_LOCAL(corosync_ping_message_type,.to_event=corosync_ping_to_event,)
static corosync_cfg_handle_t cfg_handle
Definition: res_corosync.c:284
static int unload_module(void)
static void stasis_message_cb(void *data, struct stasis_subscription *sub, struct stasis_message *message)
Definition: res_corosync.c:601
static int load_config(unsigned int reload)
static int set_event(const char *event_type, int pubsub)
static void publish_mwi_to_stasis(struct ast_event *event)
Publish a received MWI ast_event to Stasis Message Bus API.
Definition: res_corosync.c:394
static ast_rwlock_t init_cpg_lock
Definition: res_corosync.c:52
AST_MODULE_INFO_STANDARD_EXTENDED(ASTERISK_GPL_KEY, "Corosync")
static cpg_handle_t cpg_handle
Definition: res_corosync.c:283
static int dump_cache_cb(void *obj, void *arg, int flags)
Definition: res_corosync.c:610
static int corosync_node_cmp_fn(void *obj, void *arg, int flags)
Definition: res_corosync.c:133
static struct @432 event_types[]
#define NULL
Definition: resample.c:96
Stasis Message Bus API. See Stasis Message Bus API for detailed documentation.
struct stasis_message * stasis_cache_clear_create(struct stasis_message *message)
A message which instructs the caching topic to remove an entry from its cache.
Definition: stasis_cache.c:778
#define STASIS_MESSAGE_TYPE_CLEANUP(name)
Boiler-plate messaging macro for cleaning up message types.
Definition: stasis.h:1515
struct stasis_forward * stasis_forward_cancel(struct stasis_forward *forward)
Definition: stasis.c:1548
struct ao2_container * stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type)
Dump all entity items from the cache to a subscription.
Definition: stasis_cache.c:757
struct stasis_topic * stasis_topic_create(const char *name)
Create a new topic.
Definition: stasis.c:617
struct ao2_container * stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid)
Dump cached items to a subscription for a specific entity.
Definition: stasis_cache.c:718
const struct ast_eid * stasis_message_eid(const struct stasis_message *msg)
Get the entity id for a stasis_message.
struct ast_event * stasis_message_to_event(struct stasis_message *msg)
Build the Generic event system representation of the message.
#define STASIS_MESSAGE_TYPE_INIT(name)
Boiler-plate messaging macro for initializing message types.
Definition: stasis.h:1493
void * stasis_message_data(const struct stasis_message *msg)
Get the data contained in a message.
struct stasis_message * stasis_message_create(struct stasis_message_type *type, void *data)
Create a new message.
struct stasis_forward * stasis_forward_all(struct stasis_topic *from_topic, struct stasis_topic *to_topic)
Create a subscription which forwards all messages from one topic to another.
Definition: stasis.c:1578
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
Publish a message to a topic's subscribers.
Definition: stasis.c:1511
int stasis_message_router_set_congestion_limits(struct stasis_message_router *router, long low_water, long high_water)
Set the high and low alert water marks of the stasis message router.
#define stasis_message_router_create(topic)
Create a new message router object.
void stasis_message_router_remove(struct stasis_message_router *router, struct stasis_message_type *message_type)
Remove a route from a message router.
int stasis_message_router_add(struct stasis_message_router *router, struct stasis_message_type *message_type, stasis_subscription_cb callback, void *data)
Add a route to a message router.
void stasis_message_router_unsubscribe_and_join(struct stasis_message_router *router)
Unsubscribe the router from the upstream topic, blocking until the final message has been processed.
struct stasis_message_type * ast_cluster_discovery_type(void)
A stasis_message_type for Cluster discovery.
struct stasis_topic * ast_system_topic(void)
A Stasis Message Bus API topic which publishes messages regarding system changes.
static force_inline int attribute_pure ast_strlen_zero(const char *s)
Definition: strings.h:65
void ast_copy_string(char *dst, const char *src, size_t size)
Size-limited null-terminating string copy.
Definition: strings.h:425
Generic container type.
descriptor for a cli entry.
Definition: cli.h:171
int args
This gets set in ast_cli_register()
Definition: cli.h:185
char * command
Definition: cli.h:186
const char * usage
Definition: cli.h:177
An Entity ID is essentially a MAC address, brief and unique.
Definition: utils.h:813
unsigned char eid[6]
Definition: utils.h:814
An event.
Definition: event.c:81
Structure used to handle boolean flags.
Definition: utils.h:199
Abstract JSON element (object, array, string, int, ...).
Socket address structure.
Definition: netsock2.h:97
Structure for variables, used for configurations and for channel variables.
struct ast_variable * next
struct ast_sockaddr addr
Definition: res_corosync.c:85
struct ast_eid eid
Definition: res_corosync.c:83
A payload wrapper around a corosync ping event.
Definition: res_corosync.c:161
struct ast_event * event
Definition: res_corosync.c:163
Definition: astman.c:222
Definition: test_heap.c:38
Forwarding information.
Definition: stasis.c:1531
struct ast_eid eid
An API for managing task processing threads that can be shared across modules.
#define AST_TASKPROCESSOR_HIGH_WATER_LEVEL
Definition: taskprocessor.h:64
static struct test_val a
int error(const char *format,...)
Definition: utils/frame.c:999
#define ast_assert(a)
Definition: utils.h:739
int ast_carefulwrite(int fd, char *s, int len, int timeoutms)
Try to write string, but wait no more than ms milliseconds before timing out.
Definition: utils.c:1771
int ast_eid_cmp(const struct ast_eid *eid1, const struct ast_eid *eid2)
Compare two EIDs.
Definition: utils.c:3094
char * ast_eid_to_str(char *s, int maxlen, struct ast_eid *eid)
Convert an EID to a string.
Definition: utils.c:2839
int ast_eid_is_empty(const struct ast_eid *eid)
Check if EID is empty.
Definition: utils.c:3099
#define ARRAY_LEN(a)
Definition: utils.h:666
struct ast_eid ast_eid_default
Global EID.
Definition: options.c:93