res/res_corosync.c

   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         <support_level>extended</support_level>
  30  ***/
  31
  32 #include "asterisk.h"
  33
  34 ASTERISK_FILE_VERSION(__FILE__, "$Revision$");
  35
  36 #include <corosync/cpg.h>
  37 #include <corosync/cfg.h>
  38
  39 #include "asterisk/module.h"
  40 #include "asterisk/logger.h"
  41 #include "asterisk/poll-compat.h"
  42 #include "asterisk/config.h"
  43 #include "asterisk/event.h"
  44 #include "asterisk/cli.h"
  45 #include "asterisk/devicestate.h"
  46
  47 AST_RWLOCK_DEFINE_STATIC(event_types_lock);
  48
  49 static struct {
  50         const char *name;
  51         struct ast_event_sub *sub;
  52         unsigned char publish;
  53         unsigned char subscribe;
  54 } event_types[] = {
  55         [AST_EVENT_MWI] = { .name = "mwi", },
  56         [AST_EVENT_DEVICE_STATE_CHANGE] = { .name = "device_state", },
  57 };
  58
  59 static struct {
  60         pthread_t id;
  61         int alert_pipe[2];
  62         unsigned int stop:1;
  63 } dispatch_thread = {
  64         .id = AST_PTHREADT_NULL,
  65         .alert_pipe = { -1, -1 },
  66 };
  67
  68 static cpg_handle_t cpg_handle;
  69 static corosync_cfg_handle_t cfg_handle;
  70
  71 static void cfg_state_track_cb(
  72                 corosync_cfg_state_notification_buffer_t *notification_buffer,
  73                 cs_error_t error);
  74
  75 static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle,
  76                 corosync_cfg_shutdown_flags_t flags);
  77
  78 static corosync_cfg_callbacks_t cfg_callbacks = {
  79         .corosync_cfg_state_track_callback = cfg_state_track_cb,
  80         .corosync_cfg_shutdown_callback = cfg_shutdown_cb,
  81 };
  82
  83 static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name,
  84                 uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len);
  85
  86 static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name,
  87                 const struct cpg_address *member_list, size_t member_list_entries,
  88                 const struct cpg_address *left_list, size_t left_list_entries,
  89                 const struct cpg_address *joined_list, size_t joined_list_entries);
  90
  91 static cpg_callbacks_t cpg_callbacks = {
  92         .cpg_deliver_fn = cpg_deliver_cb,
  93         .cpg_confchg_fn = cpg_confchg_cb,
  94 };
  95
  96 static void ast_event_cb(const struct ast_event *event, void *data);
  97
  98 static void cfg_state_track_cb(
  99                 corosync_cfg_state_notification_buffer_t *notification_buffer,
 100                 cs_error_t error)
 101 {
 102 }
 103
 104 static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle,
 105                 corosync_cfg_shutdown_flags_t flags)
 106 {
 107 }
 108
 109 static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name,
 110                 uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len)
 111 {
 112         struct ast_event *event;
 113
 114         if (msg_len < ast_event_minimum_length()) {
 115                 ast_debug(1, "Ignoring event that's too small. %u < %u\n",
 116                         (unsigned int) msg_len,
 117                         (unsigned int) ast_event_minimum_length());
 118                 return;
 119         }
 120
 121         if (!ast_eid_cmp(&ast_eid_default, ast_event_get_ie_raw(msg, AST_EVENT_IE_EID))) {
 122                 /* Don't feed events back in that originated locally. */
 123                 return;
 124         }
 125
 126         ast_rwlock_rdlock(&event_types_lock);
 127         if (!event_types[ast_event_get_type(msg)].subscribe) {
 128                 /* We are not configured to subscribe to these events. */
 129                 ast_rwlock_unlock(&event_types_lock);
 130                 return;
 131         }
 132         ast_rwlock_unlock(&event_types_lock);
 133
 134         if (!(event = ast_malloc(msg_len))) {
 135                 return;
 136         }
 137
 138         memcpy(event, msg, msg_len);
 139
 140         ast_event_queue_and_cache(event);
 141 }
 142
 143 static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name,
 144                 const struct cpg_address *member_list, size_t member_list_entries,
 145                 const struct cpg_address *left_list, size_t left_list_entries,
 146                 const struct cpg_address *joined_list, size_t joined_list_entries)
 147 {
 148         unsigned int i;
 149
 150         /* If any new nodes have joined, dump our cache of events we are publishing
 151          * that originated from this server. */
 152
 153         if (!joined_list_entries) {
 154                 return;
 155         }
 156
 157         for (i = 0; i < ARRAY_LEN(event_types); i++) {
 158                 struct ast_event_sub *event_sub;
 159
 160                 ast_rwlock_rdlock(&event_types_lock);
 161                 if (!event_types[i].publish) {
 162                         ast_rwlock_unlock(&event_types_lock);
 163                         continue;
 164                 }
 165                 ast_rwlock_unlock(&event_types_lock);
 166
 167                 event_sub = ast_event_subscribe_new(i, ast_event_cb, NULL);
 168                 ast_event_sub_append_ie_raw(event_sub, AST_EVENT_IE_EID,
 169                                         &ast_eid_default, sizeof(ast_eid_default));
 170                 ast_event_dump_cache(event_sub);
 171                 ast_event_sub_destroy(event_sub);
 172         }
 173 }
 174
 175 static void *dispatch_thread_handler(void *data)
 176 {
 177         cs_error_t cs_err;
 178         struct pollfd pfd[3] = {
 179                 { .events = POLLIN, },
 180                 { .events = POLLIN, },
 181                 { .events = POLLIN, },
 182         };
 183
 184         if ((cs_err = cpg_fd_get(cpg_handle, &pfd[0].fd)) != CS_OK) {
 185                 ast_log(LOG_ERROR, "Failed to get CPG fd.  This module is now broken.\n");
 186                 return NULL;
 187         }
 188
 189         if ((cs_err = corosync_cfg_fd_get(cfg_handle, &pfd[1].fd)) != CS_OK) {
 190                 ast_log(LOG_ERROR, "Failed to get CFG fd.  This module is now broken.\n");
 191                 return NULL;
 192         }
 193
 194         pfd[2].fd = dispatch_thread.alert_pipe[0];
 195
 196         while (!dispatch_thread.stop) {
 197                 int res;
 198
 199                 pfd[0].revents = 0;
 200                 pfd[1].revents = 0;
 201                 pfd[2].revents = 0;
 202
 203                 res = ast_poll(pfd, ARRAY_LEN(pfd), -1);
 204                 if (res == -1 && errno != EINTR && errno != EAGAIN) {
 205                         ast_log(LOG_ERROR, "poll() error: %s (%d)\n", strerror(errno), errno);
 206                         continue;
 207                 }
 208
 209                 if (pfd[0].revents & POLLIN) {
 210                         if ((cs_err = cpg_dispatch(cpg_handle, CS_DISPATCH_ALL)) != CS_OK) {
 211                                 ast_log(LOG_WARNING, "Failed CPG dispatch: %d\n", cs_err);
 212                         }
 213                 }
 214
 215                 if (pfd[1].revents & POLLIN) {
 216                         if ((cs_err = corosync_cfg_dispatch(cfg_handle, CS_DISPATCH_ALL)) != CS_OK) {
 217                                 ast_log(LOG_WARNING, "Failed CFG dispatch: %d\n", cs_err);
 218                         }
 219                 }
 220         }
 221
 222         return NULL;
 223 }
 224
 225 static void ast_event_cb(const struct ast_event *event, void *data)
 226 {
 227         cs_error_t cs_err;
 228         struct iovec iov = {
 229                 .iov_base = (void *) event,
 230                 .iov_len = ast_event_get_size(event),
 231         };
 232
 233         if (ast_eid_cmp(&ast_eid_default,
 234                         ast_event_get_ie_raw(event, AST_EVENT_IE_EID))) {
 235                 /* If the event didn't originate from this server, don't send it back out. */
 236                 return;
 237         }
 238
 239         /* The ast_event subscription will only exist if we are configured to publish
 240          * these events, so just send away. */
 241
 242         if ((cs_err = cpg_mcast_joined(cpg_handle, CPG_TYPE_FIFO, &iov, 1)) != CS_OK) {
 243                 ast_log(LOG_WARNING, "CPG mcast failed (%d)\n", cs_err);
 244         }
 245 }
 246
 247 static char *corosync_show_members(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
 248 {
 249         cs_error_t cs_err;
 250         cpg_iteration_handle_t cpg_iter;
 251         struct cpg_iteration_description_t cpg_desc;
 252         unsigned int i;
 253
 254         switch (cmd) {
 255         case CLI_INIT:
 256                 e->command = "corosync show members";
 257                 e->usage =
 258                         "Usage: corosync show members\n"
 259                         "       Show corosync cluster members\n";
 260                 return NULL;
 261
 262         case CLI_GENERATE:
 263                 return NULL;    /* no completion */
 264         }
 265
 266         if (a->argc != e->args) {
 267                 return CLI_SHOWUSAGE;
 268         }
 269
 270         cs_err = cpg_iteration_initialize(cpg_handle, CPG_ITERATION_ALL, NULL, &cpg_iter);
 271
 272         if (cs_err != CS_OK) {
 273                 ast_cli(a->fd, "Failed to initialize CPG iterator.\n");
 274                 return CLI_FAILURE;
 275         }
 276
 277         ast_cli(a->fd, "\n"
 278                     "=============================================================\n"
 279                     "=== Cluster members =========================================\n"
 280                     "=============================================================\n"
 281                     "===\n");
 282
 283         for (i = 1, cs_err = cpg_iteration_next(cpg_iter, &cpg_desc);
 284                         cs_err == CS_OK;
 285                         cs_err = cpg_iteration_next(cpg_iter, &cpg_desc), i++) {
 286                 corosync_cfg_node_address_t addrs[8];
 287                 int num_addrs = 0;
 288                 unsigned int j;
 289
 290                 cs_err = corosync_cfg_get_node_addrs(cfg_handle, cpg_desc.nodeid,
 291                                 ARRAY_LEN(addrs), &num_addrs, addrs);
 292                 if (cs_err != CS_OK) {
 293                         ast_log(LOG_WARNING, "Failed to get node addresses\n");
 294                         continue;
 295                 }
 296
 297                 ast_cli(a->fd, "=== Node %d\n", i);
 298                 ast_cli(a->fd, "=== --> Group: %s\n", cpg_desc.group.value);
 299
 300                 for (j = 0; j < num_addrs; j++) {
 301                         struct sockaddr *sa = (struct sockaddr *) addrs[j].address;
 302                         size_t sa_len = (size_t) addrs[j].address_length;
 303                         char buf[128];
 304
 305                         getnameinfo(sa, sa_len, buf, sizeof(buf), NULL, 0, NI_NUMERICHOST);
 306
 307                         ast_cli(a->fd, "=== --> Address %d: %s\n", j + 1, buf);
 308                 }
 309
 310         }
 311
 312         ast_cli(a->fd, "===\n"
 313                        "=============================================================\n"
 314                        "\n");
 315
 316         cpg_iteration_finalize(cpg_iter);
 317
 318         return CLI_SUCCESS;
 319 }
 320
 321 static char *corosync_show_config(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
 322 {
 323         unsigned int i;
 324
 325         switch (cmd) {
 326         case CLI_INIT:
 327                 e->command = "corosync show config";
 328                 e->usage =
 329                         "Usage: corosync show config\n"
 330                         "       Show configuration loaded from res_corosync.conf\n";
 331                 return NULL;
 332
 333         case CLI_GENERATE:
 334                 return NULL;    /* no completion */
 335         }
 336
 337         if (a->argc != e->args) {
 338                 return CLI_SHOWUSAGE;
 339         }
 340
 341         ast_cli(a->fd, "\n"
 342                     "=============================================================\n"
 343                     "=== res_corosync config =====================================\n"
 344                     "=============================================================\n"
 345                     "===\n");
 346
 347         ast_rwlock_rdlock(&event_types_lock);
 348         for (i = 0; i < ARRAY_LEN(event_types); i++) {
 349                 if (event_types[i].publish) {
 350                         ast_cli(a->fd, "=== ==> Publishing Event Type: %s\n",
 351                                         event_types[i].name);
 352                 }
 353                 if (event_types[i].subscribe) {
 354                         ast_cli(a->fd, "=== ==> Subscribing to Event Type: %s\n",
 355                                         event_types[i].name);
 356                 }
 357         }
 358         ast_rwlock_unlock(&event_types_lock);
 359
 360         ast_cli(a->fd, "===\n"
 361                        "=============================================================\n"
 362                        "\n");
 363
 364         return CLI_SUCCESS;
 365 }
 366
 367 static struct ast_cli_entry corosync_cli[] = {
 368         AST_CLI_DEFINE(corosync_show_config, "Show configuration"),
 369         AST_CLI_DEFINE(corosync_show_members, "Show cluster members"),
 370 };
 371
 372 enum {
 373         PUBLISH,
 374         SUBSCRIBE,
 375 };
 376
 377 static int set_event(const char *event_type, int pubsub)
 378 {
 379         unsigned int i;
 380
 381         for (i = 0; i < ARRAY_LEN(event_types); i++) {
 382                 if (!event_types[i].name || strcasecmp(event_type, event_types[i].name)) {
 383                         continue;
 384                 }
 385
 386                 switch (pubsub) {
 387                 case PUBLISH:
 388                         event_types[i].publish = 1;
 389                         break;
 390                 case SUBSCRIBE:
 391                         event_types[i].subscribe = 1;
 392                         break;
 393                 }
 394
 395                 break;
 396         }
 397
 398         return (i == ARRAY_LEN(event_types)) ? -1 : 0;
 399 }
 400
 401 static int load_general_config(struct ast_config *cfg)
 402 {
 403         struct ast_variable *v;
 404         int res = 0;
 405         unsigned int i;
 406
 407         ast_rwlock_wrlock(&event_types_lock);
 408
 409         for (i = 0; i < ARRAY_LEN(event_types); i++) {
 410                 event_types[i].publish = 0;
 411                 event_types[i].subscribe = 0;
 412         }
 413
 414         for (v = ast_variable_browse(cfg, "general"); v && !res; v = v->next) {
 415                 if (!strcasecmp(v->name, "publish_event")) {
 416                         res = set_event(v->value, PUBLISH);
 417                 } else if (!strcasecmp(v->name, "subscribe_event")) {
 418                         res = set_event(v->value, SUBSCRIBE);
 419                 } else {
 420                         ast_log(LOG_WARNING, "Unknown option '%s'\n", v->name);
 421                 }
 422         }
 423
 424         for (i = 0; i < ARRAY_LEN(event_types); i++) {
 425                 if (event_types[i].publish && !event_types[i].sub) {
 426                         event_types[i].sub = ast_event_subscribe(i,
 427                                                 ast_event_cb, "Corosync", NULL,
 428                                                 AST_EVENT_IE_END);
 429                 } else if (!event_types[i].publish && event_types[i].sub) {
 430                         event_types[i].sub = ast_event_unsubscribe(event_types[i].sub);
 431                 }
 432         }
 433
 434         ast_rwlock_unlock(&event_types_lock);
 435
 436         return res;
 437 }
 438
 439 static int load_config(unsigned int reload)
 440 {
 441         static const char filename[] = "res_corosync.conf";
 442         struct ast_config *cfg;
 443         const char *cat = NULL;
 444         struct ast_flags config_flags = { 0 };
 445         int res = 0;
 446
 447         cfg = ast_config_load(filename, config_flags);
 448
 449         if (cfg == CONFIG_STATUS_FILEMISSING || cfg == CONFIG_STATUS_FILEINVALID) {
 450                 return -1;
 451         }
 452
 453         while ((cat = ast_category_browse(cfg, cat))) {
 454                 if (!strcasecmp(cat, "general")) {
 455                         res = load_general_config(cfg);
 456                 } else {
 457                         ast_log(LOG_WARNING, "Unknown configuration section '%s'\n", cat);
 458                 }
 459         }
 460
 461         ast_config_destroy(cfg);
 462
 463         return res;
 464 }
 465
 466 static void cleanup_module(void)
 467 {
 468         cs_error_t cs_err;
 469         unsigned int i;
 470
 471         for (i = 0; i < ARRAY_LEN(event_types); i++) {
 472                 if (event_types[i].sub) {
 473                         event_types[i].sub = ast_event_unsubscribe(event_types[i].sub);
 474                 }
 475                 event_types[i].publish = 0;
 476                 event_types[i].subscribe = 0;
 477         }
 478
 479         if (dispatch_thread.id != AST_PTHREADT_NULL) {
 480                 char meepmeep = 'x';
 481                 dispatch_thread.stop = 1;
 482                 if (ast_carefulwrite(dispatch_thread.alert_pipe[1], &meepmeep, 1,
 483                                         5000) == -1) {
 484                         ast_log(LOG_ERROR, "Failed to write to pipe: %s (%d)\n",
 485                                         strerror(errno), errno);
 486                 }
 487                 pthread_join(dispatch_thread.id, NULL);
 488         }
 489
 490         if (dispatch_thread.alert_pipe[0] != -1) {
 491                 close(dispatch_thread.alert_pipe[0]);
 492                 dispatch_thread.alert_pipe[0] = -1;
 493         }
 494
 495         if (dispatch_thread.alert_pipe[1] != -1) {
 496                 close(dispatch_thread.alert_pipe[1]);
 497                 dispatch_thread.alert_pipe[1] = -1;
 498         }
 499
 500         if (cpg_handle && (cs_err = cpg_finalize(cpg_handle)) != CS_OK) {
 501                 ast_log(LOG_ERROR, "Failed to finalize cpg (%d)\n", (int) cs_err);
 502         }
 503         cpg_handle = 0;
 504
 505         if (cfg_handle && (cs_err = corosync_cfg_finalize(cfg_handle)) != CS_OK) {
 506                 ast_log(LOG_ERROR, "Failed to finalize cfg (%d)\n", (int) cs_err);
 507         }
 508         cfg_handle = 0;
 509 }
 510
 511 static int load_module(void)
 512 {
 513         cs_error_t cs_err;
 514         enum ast_module_load_result res = AST_MODULE_LOAD_FAILURE;
 515         struct cpg_name name;
 516
 517         if ((cs_err = corosync_cfg_initialize(&cfg_handle, &cfg_callbacks)) != CS_OK) {
 518                 ast_log(LOG_ERROR, "Failed to initialize cfg (%d)\n", (int) cs_err);
 519                 return AST_MODULE_LOAD_DECLINE;
 520         }
 521
 522         if ((cs_err = cpg_initialize(&cpg_handle, &cpg_callbacks)) != CS_OK) {
 523                 ast_log(LOG_ERROR, "Failed to initialize cpg (%d)\n", (int) cs_err);
 524                 goto failed;
 525         }
 526
 527         ast_copy_string(name.value, "asterisk", sizeof(name.value));
 528         name.length = strlen(name.value);
 529
 530         if ((cs_err = cpg_join(cpg_handle, &name)) != CS_OK) {
 531                 ast_log(LOG_ERROR, "Failed to join (%d)\n", (int) cs_err);
 532                 goto failed;
 533         }
 534
 535         if (pipe(dispatch_thread.alert_pipe) == -1) {
 536                 ast_log(LOG_ERROR, "Failed to create alert pipe: %s (%d)\n",
 537                                 strerror(errno), errno);
 538                 goto failed;
 539         }
 540
 541         if (ast_pthread_create_background(&dispatch_thread.id, NULL,
 542                         dispatch_thread_handler, NULL)) {
 543                 ast_log(LOG_ERROR, "Error starting CPG dispatch thread.\n");
 544                 goto failed;
 545         }
 546
 547         if (load_config(0)) {
 548                 /* simply not configured is not a fatal error */
 549                 res = AST_MODULE_LOAD_DECLINE;
 550                 goto failed;
 551         }
 552
 553         ast_cli_register_multiple(corosync_cli, ARRAY_LEN(corosync_cli));
 554
 555         ast_enable_distributed_devstate();
 556
 557         return AST_MODULE_LOAD_SUCCESS;
 558
 559 failed:
 560         cleanup_module();
 561
 562         return res;
 563 }
 564
 565 static int unload_module(void)
 566 {
 567         ast_cli_unregister_multiple(corosync_cli, ARRAY_LEN(corosync_cli));
 568
 569         cleanup_module();
 570
 571         return 0;
 572 }
 573
 574 AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "Corosync");