src/connectivity.c

   1 /**
   2  * collectd - src/connectivity.c
   3  *
   4  * Permission is hereby granted, free of charge, to any person obtaining a
   5  * copy of this software and associated documentation files (the "Software"),
   6  * to deal in the Software without restriction, including without limitation
   7  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
   8  * and/or sell copies of the Software, and to permit persons to whom the
   9  * Software is furnished to do so, subject to the following conditions:
  10  *
  11  * The above copyright notice and this permission notice shall be included in
  12  * all copies or substantial portions of the Software.
  13  *
  14  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  19  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  20  * DEALINGS IN THE SOFTWARE.
  21  *
  22  * Authors:
  23  *   Red Hat NFVPE
  24  *     Andrew Bays <abays at redhat.com>
  25  *     Aneesh Puttur <aputtur at redhat.com>
  26  **/
  27
  28 #include "collectd.h"
  29
  30 #include "common.h"
  31 #include "plugin.h"
  32 #include "utils_complain.h"
  33 #include "utils_ignorelist.h"
  34
  35 #include <asm/types.h>
  36 #include <errno.h>
  37 #include <net/if.h>
  38 #include <netinet/in.h>
  39 #include <pthread.h>
  40 #include <stdio.h>
  41 #include <string.h>
  42 #include <sys/socket.h>
  43 #include <unistd.h>
  44
  45 #include <libmnl/libmnl.h>
  46 #include <linux/netlink.h>
  47 #include <linux/rtnetlink.h>
  48
  49 #include <yajl/yajl_common.h>
  50 #include <yajl/yajl_gen.h>
  51 #if HAVE_YAJL_YAJL_VERSION_H
  52 #include <yajl/yajl_version.h>
  53 #endif
  54 #if defined(YAJL_MAJOR) && (YAJL_MAJOR > 1)
  55 #define HAVE_YAJL_V2 1
  56 #endif
  57
  58 #define MYPROTO NETLINK_ROUTE
  59
  60 #define LINK_STATE_DOWN 0
  61 #define LINK_STATE_UP 1
  62 #define LINK_STATE_UNKNOWN 2
  63
  64 #define CONNECTIVITY_DOMAIN_FIELD "domain"
  65 #define CONNECTIVITY_DOMAIN_VALUE "stateChange"
  66 #define CONNECTIVITY_EVENT_ID_FIELD "eventId"
  67 #define CONNECTIVITY_EVENT_NAME_FIELD "eventName"
  68 #define CONNECTIVITY_EVENT_NAME_DOWN_VALUE "down"
  69 #define CONNECTIVITY_EVENT_NAME_UP_VALUE "up"
  70 #define CONNECTIVITY_LAST_EPOCH_MICROSEC_FIELD "lastEpochMicrosec"
  71 #define CONNECTIVITY_PRIORITY_FIELD "priority"
  72 #define CONNECTIVITY_PRIORITY_VALUE "high"
  73 #define CONNECTIVITY_REPORTING_ENTITY_NAME_FIELD "reportingEntityName"
  74 #define CONNECTIVITY_REPORTING_ENTITY_NAME_VALUE "collectd connectivity plugin"
  75 #define CONNECTIVITY_SEQUENCE_FIELD "sequence"
  76 #define CONNECTIVITY_SEQUENCE_VALUE "0"
  77 #define CONNECTIVITY_SOURCE_NAME_FIELD "sourceName"
  78 #define CONNECTIVITY_START_EPOCH_MICROSEC_FIELD "startEpochMicrosec"
  79 #define CONNECTIVITY_VERSION_FIELD "version"
  80 #define CONNECTIVITY_VERSION_VALUE "1.0"
  81
  82 #define CONNECTIVITY_NEW_STATE_FIELD "newState"
  83 #define CONNECTIVITY_NEW_STATE_FIELD_DOWN_VALUE "outOfService"
  84 #define CONNECTIVITY_NEW_STATE_FIELD_UP_VALUE "inService"
  85 #define CONNECTIVITY_OLD_STATE_FIELD "oldState"
  86 #define CONNECTIVITY_OLD_STATE_FIELD_DOWN_VALUE "outOfService"
  87 #define CONNECTIVITY_OLD_STATE_FIELD_UP_VALUE "inService"
  88 #define CONNECTIVITY_STATE_CHANGE_FIELDS_FIELD "stateChangeFields"
  89 #define CONNECTIVITY_STATE_CHANGE_FIELDS_VERSION_FIELD                         \
  90   "stateChangeFieldsVersion"
  91 #define CONNECTIVITY_STATE_CHANGE_FIELDS_VERSION_VALUE "1.0"
  92 #define CONNECTIVITY_STATE_INTERFACE_FIELD "stateInterface"
  93
  94 /*
  95  * Private data types
  96  */
  97
  98 struct interface_list_s {
  99   char *interface;
 100
 101   uint32_t status;
 102   uint32_t prev_status;
 103   uint32_t sent;
 104   long long unsigned int timestamp;
 105
 106   struct interface_list_s *next;
 107 };
 108 typedef struct interface_list_s interface_list_t;
 109
 110 /*
 111  * Private variables
 112  */
 113
 114 static ignorelist_t *ignorelist = NULL;
 115
 116 static interface_list_t *interface_list_head = NULL;
 117 static int monitor_all_interfaces = 1;
 118
 119 static int connectivity_netlink_thread_loop = 0;
 120 static int connectivity_netlink_thread_error = 0;
 121 static pthread_t connectivity_netlink_thread_id;
 122 static int connectivity_dequeue_thread_loop = 0;
 123 static pthread_t connectivity_dequeue_thread_id;
 124 static pthread_mutex_t connectivity_threads_lock = PTHREAD_MUTEX_INITIALIZER;
 125 static pthread_mutex_t connectivity_data_lock = PTHREAD_MUTEX_INITIALIZER;
 126 static pthread_cond_t connectivity_cond = PTHREAD_COND_INITIALIZER;
 127 static int nl_sock = -1;
 128 static int event_id = 0;
 129 static int unsent_statuses = 0;
 130
 131 static const char *config_keys[] = {"Interface", "IgnoreSelected"};
 132 static int config_keys_num = STATIC_ARRAY_SIZE(config_keys);
 133
 134 /*
 135  * Prototype
 136  */
 137
 138 static void
 139 connectivity_dispatch_notification(const char *interface, const char *type,
 140                                    gauge_t value, gauge_t old_value,
 141                                    long long unsigned int timestamp);
 142
 143 /*
 144  * Private functions
 145  */
 146
 147 static int gen_message_payload(int state, int old_state, const char *interface,
 148                                long long unsigned int timestamp, char **buf) {
 149   const unsigned char *buf2;
 150   yajl_gen g;
 151   char json_str[DATA_MAX_NAME_LEN];
 152
 153 #if !defined(HAVE_YAJL_V2)
 154   yajl_gen_config conf = {};
 155
 156   conf.beautify = 0;
 157 #endif
 158
 159 #if HAVE_YAJL_V2
 160   size_t len;
 161   g = yajl_gen_alloc(NULL);
 162   yajl_gen_config(g, yajl_gen_beautify, 0);
 163 #else
 164   unsigned int len;
 165   g = yajl_gen_alloc(&conf, NULL);
 166 #endif
 167
 168   yajl_gen_clear(g);
 169
 170   // *** BEGIN common event header ***
 171
 172   if (yajl_gen_map_open(g) != yajl_gen_status_ok)
 173     goto err;
 174
 175   // domain
 176   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_DOMAIN_FIELD,
 177                       strlen(CONNECTIVITY_DOMAIN_FIELD)) != yajl_gen_status_ok)
 178     goto err;
 179
 180   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_DOMAIN_VALUE,
 181                       strlen(CONNECTIVITY_DOMAIN_VALUE)) != yajl_gen_status_ok)
 182     goto err;
 183
 184   // eventId
 185   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_EVENT_ID_FIELD,
 186                       strlen(CONNECTIVITY_EVENT_ID_FIELD)) !=
 187       yajl_gen_status_ok)
 188     goto err;
 189
 190   event_id = event_id + 1;
 191   int event_id_len = sizeof(char) * sizeof(int) * 4 + 1;
 192   memset(json_str, '\0', DATA_MAX_NAME_LEN);
 193   snprintf(json_str, event_id_len, "%d", event_id);
 194
 195   if (yajl_gen_number(g, json_str, strlen(json_str)) != yajl_gen_status_ok) {
 196     goto err;
 197   }
 198
 199   // eventName
 200   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_EVENT_NAME_FIELD,
 201                       strlen(CONNECTIVITY_EVENT_NAME_FIELD)) !=
 202       yajl_gen_status_ok)
 203     goto err;
 204
 205   int event_name_len = 0;
 206   event_name_len = event_name_len + strlen(interface);    // interface name
 207   event_name_len = event_name_len + (state == 0 ? 4 : 2); // "down" or "up"
 208   event_name_len =
 209       event_name_len + 12; // "interface", 2 spaces and null-terminator
 210   memset(json_str, '\0', DATA_MAX_NAME_LEN);
 211   snprintf(json_str, event_name_len, "interface %s %s", interface,
 212            (state == 0 ? CONNECTIVITY_EVENT_NAME_DOWN_VALUE
 213                        : CONNECTIVITY_EVENT_NAME_UP_VALUE));
 214
 215   if (yajl_gen_string(g, (u_char *)json_str, strlen(json_str)) !=
 216       yajl_gen_status_ok) {
 217     goto err;
 218   }
 219
 220   // lastEpochMicrosec
 221   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_LAST_EPOCH_MICROSEC_FIELD,
 222                       strlen(CONNECTIVITY_LAST_EPOCH_MICROSEC_FIELD)) !=
 223       yajl_gen_status_ok)
 224     goto err;
 225
 226   int last_epoch_microsec_len =
 227       sizeof(char) * sizeof(long long unsigned int) * 4 + 1;
 228   memset(json_str, '\0', DATA_MAX_NAME_LEN);
 229   snprintf(json_str, last_epoch_microsec_len, "%llu",
 230            (long long unsigned int)CDTIME_T_TO_US(cdtime()));
 231
 232   if (yajl_gen_number(g, json_str, strlen(json_str)) != yajl_gen_status_ok) {
 233     goto err;
 234   }
 235
 236   // priority
 237   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_PRIORITY_FIELD,
 238                       strlen(CONNECTIVITY_PRIORITY_FIELD)) !=
 239       yajl_gen_status_ok)
 240     goto err;
 241
 242   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_PRIORITY_VALUE,
 243                       strlen(CONNECTIVITY_PRIORITY_VALUE)) !=
 244       yajl_gen_status_ok)
 245     goto err;
 246
 247   // reportingEntityName
 248   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_REPORTING_ENTITY_NAME_FIELD,
 249                       strlen(CONNECTIVITY_REPORTING_ENTITY_NAME_FIELD)) !=
 250       yajl_gen_status_ok)
 251     goto err;
 252
 253   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_REPORTING_ENTITY_NAME_VALUE,
 254                       strlen(CONNECTIVITY_REPORTING_ENTITY_NAME_VALUE)) !=
 255       yajl_gen_status_ok)
 256     goto err;
 257
 258   // sequence
 259   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_SEQUENCE_FIELD,
 260                       strlen(CONNECTIVITY_SEQUENCE_FIELD)) !=
 261       yajl_gen_status_ok)
 262     goto err;
 263
 264   if (yajl_gen_number(g, CONNECTIVITY_SEQUENCE_VALUE,
 265                       strlen(CONNECTIVITY_SEQUENCE_VALUE)) !=
 266       yajl_gen_status_ok)
 267     goto err;
 268
 269   // sourceName
 270   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_SOURCE_NAME_FIELD,
 271                       strlen(CONNECTIVITY_SOURCE_NAME_FIELD)) !=
 272       yajl_gen_status_ok)
 273     goto err;
 274
 275   if (yajl_gen_string(g, (u_char *)interface, strlen(interface)) !=
 276       yajl_gen_status_ok)
 277     goto err;
 278
 279   // startEpochMicrosec
 280   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_START_EPOCH_MICROSEC_FIELD,
 281                       strlen(CONNECTIVITY_START_EPOCH_MICROSEC_FIELD)) !=
 282       yajl_gen_status_ok)
 283     goto err;
 284
 285   int start_epoch_microsec_len =
 286       sizeof(char) * sizeof(long long unsigned int) * 4 + 1;
 287   memset(json_str, '\0', DATA_MAX_NAME_LEN);
 288   snprintf(json_str, start_epoch_microsec_len, "%llu",
 289            (long long unsigned int)timestamp);
 290
 291   if (yajl_gen_number(g, json_str, strlen(json_str)) != yajl_gen_status_ok) {
 292     goto err;
 293   }
 294
 295   // version
 296   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_VERSION_FIELD,
 297                       strlen(CONNECTIVITY_VERSION_FIELD)) != yajl_gen_status_ok)
 298     goto err;
 299
 300   if (yajl_gen_number(g, CONNECTIVITY_VERSION_VALUE,
 301                       strlen(CONNECTIVITY_VERSION_VALUE)) != yajl_gen_status_ok)
 302     goto err;
 303
 304   // *** END common event header ***
 305
 306   // *** BEGIN state change fields ***
 307
 308   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_STATE_CHANGE_FIELDS_FIELD,
 309                       strlen(CONNECTIVITY_STATE_CHANGE_FIELDS_FIELD)) !=
 310       yajl_gen_status_ok)
 311     goto err;
 312
 313   if (yajl_gen_map_open(g) != yajl_gen_status_ok)
 314     goto err;
 315
 316   // newState
 317   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_NEW_STATE_FIELD,
 318                       strlen(CONNECTIVITY_NEW_STATE_FIELD)) !=
 319       yajl_gen_status_ok)
 320     goto err;
 321
 322   int new_state_len =
 323       (state == 0 ? strlen(CONNECTIVITY_NEW_STATE_FIELD_DOWN_VALUE)
 324                   : strlen(CONNECTIVITY_NEW_STATE_FIELD_UP_VALUE));
 325
 326   if (yajl_gen_string(
 327           g, (u_char *)(state == 0 ? CONNECTIVITY_NEW_STATE_FIELD_DOWN_VALUE
 328                                    : CONNECTIVITY_NEW_STATE_FIELD_UP_VALUE),
 329           new_state_len) != yajl_gen_status_ok)
 330     goto err;
 331
 332   // oldState
 333   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_OLD_STATE_FIELD,
 334                       strlen(CONNECTIVITY_OLD_STATE_FIELD)) !=
 335       yajl_gen_status_ok)
 336     goto err;
 337
 338   int old_state_len =
 339       (old_state == 0 ? strlen(CONNECTIVITY_OLD_STATE_FIELD_DOWN_VALUE)
 340                       : strlen(CONNECTIVITY_OLD_STATE_FIELD_UP_VALUE));
 341
 342   if (yajl_gen_string(
 343           g, (u_char *)(old_state == 0 ? CONNECTIVITY_OLD_STATE_FIELD_DOWN_VALUE
 344                                        : CONNECTIVITY_OLD_STATE_FIELD_UP_VALUE),
 345           old_state_len) != yajl_gen_status_ok)
 346     goto err;
 347
 348   // stateChangeFieldsVersion
 349   if (yajl_gen_string(g,
 350                       (u_char *)CONNECTIVITY_STATE_CHANGE_FIELDS_VERSION_FIELD,
 351                       strlen(CONNECTIVITY_STATE_CHANGE_FIELDS_VERSION_FIELD)) !=
 352       yajl_gen_status_ok)
 353     goto err;
 354
 355   if (yajl_gen_number(g, CONNECTIVITY_STATE_CHANGE_FIELDS_VERSION_VALUE,
 356                       strlen(CONNECTIVITY_STATE_CHANGE_FIELDS_VERSION_VALUE)) !=
 357       yajl_gen_status_ok)
 358     goto err;
 359
 360   // stateInterface
 361   if (yajl_gen_string(g, (u_char *)CONNECTIVITY_STATE_INTERFACE_FIELD,
 362                       strlen(CONNECTIVITY_STATE_INTERFACE_FIELD)) !=
 363       yajl_gen_status_ok)
 364     goto err;
 365
 366   if (yajl_gen_string(g, (u_char *)interface, strlen(interface)) !=
 367       yajl_gen_status_ok)
 368     goto err;
 369
 370   if (yajl_gen_map_close(g) != yajl_gen_status_ok)
 371     goto err;
 372
 373   // *** END state change fields ***
 374
 375   if (yajl_gen_map_close(g) != yajl_gen_status_ok)
 376     goto err;
 377
 378   if (yajl_gen_get_buf(g, &buf2, &len) != yajl_gen_status_ok)
 379     goto err;
 380
 381   *buf = strdup((char *)buf2);
 382
 383   if (*buf == NULL) {
 384     ERROR("connectivity plugin: strdup failed during gen_message_payload: %s",
 385           STRERRNO);
 386     goto err;
 387   }
 388
 389   yajl_gen_free(g);
 390
 391   return 0;
 392
 393 err:
 394   yajl_gen_free(g);
 395   ERROR("connectivity plugin: gen_message_payload failed to generate JSON");
 396   return -1;
 397 }
 398
 399 static interface_list_t *add_interface(const char *interface, int status,
 400                                        int prev_status) {
 401   interface_list_t *il = calloc(1, sizeof(*il));
 402
 403   if (il == NULL) {
 404     ERROR("connectivity plugin: calloc failed during add_interface: %s",
 405           STRERRNO);
 406     return NULL;
 407   }
 408
 409   char *interface2 = strdup(interface);
 410   if (interface2 == NULL) {
 411     sfree(il);
 412     ERROR("connectivity plugin: strdup failed during add_interface: %s",
 413           STRERRNO);
 414     return NULL;
 415   }
 416
 417   il->interface = interface2;
 418   il->status = status;
 419   il->prev_status = prev_status;
 420   il->timestamp = (long long unsigned int)CDTIME_T_TO_US(cdtime());
 421   il->sent = 0;
 422   il->next = interface_list_head;
 423   interface_list_head = il;
 424
 425   DEBUG("connectivity plugin: added interface %s", interface2);
 426
 427   return il;
 428 }
 429
 430 static int connectivity_link_state(struct nlmsghdr *msg) {
 431   pthread_mutex_lock(&connectivity_data_lock);
 432
 433   struct nlattr *attr;
 434   struct ifinfomsg *ifi = mnl_nlmsg_get_payload(msg);
 435
 436   /* Scan attribute list for device name. */
 437   mnl_attr_for_each(attr, msg, sizeof(*ifi)) {
 438     if (mnl_attr_get_type(attr) != IFLA_IFNAME)
 439       continue;
 440
 441     if (mnl_attr_validate(attr, MNL_TYPE_STRING) < 0) {
 442       ERROR("connectivity plugin: connectivity_link_state: IFLA_IFNAME "
 443             "mnl_attr_validate "
 444             "failed.");
 445       pthread_mutex_unlock(&connectivity_data_lock);
 446       return MNL_CB_ERROR;
 447     }
 448
 449     const char *dev = mnl_attr_get_str(attr);
 450
 451     // Check the list of interfaces we should monitor, if we've chosen
 452     // a subset.  If we don't care about this one, abort.
 453     if (ignorelist_match(ignorelist, dev) != 0) {
 454       DEBUG("connectivity plugin: Ignoring link state change for unmonitored "
 455             "interface: %s",
 456             dev);
 457       break;
 458     }
 459
 460     interface_list_t *il = NULL;
 461
 462     for (il = interface_list_head; il != NULL; il = il->next)
 463       if (strcmp(dev, il->interface) == 0)
 464         break;
 465
 466     if (il == NULL) {
 467       // We haven't encountered this interface yet, so add it to the linked list
 468       il = add_interface(dev, LINK_STATE_UNKNOWN, LINK_STATE_UNKNOWN);
 469
 470       if (il == NULL) {
 471         ERROR("connectivity plugin: unable to add interface %s during "
 472               "connectivity_link_state",
 473               dev);
 474         return MNL_CB_ERROR;
 475       }
 476     }
 477
 478     uint32_t prev_status;
 479
 480     prev_status = il->status;
 481     il->status =
 482         ((ifi->ifi_flags & IFF_RUNNING) ? LINK_STATE_UP : LINK_STATE_DOWN);
 483     il->timestamp = (long long unsigned int)CDTIME_T_TO_US(cdtime());
 484
 485     // If the new status is different than the previous status,
 486     // store the previous status and set sent to zero, and set the
 487     // global flag to indicate there are statuses to dispatch
 488     if (il->status != prev_status) {
 489       il->prev_status = prev_status;
 490       il->sent = 0;
 491       unsent_statuses = 1;
 492     }
 493
 494     DEBUG("connectivity plugin (%llu): Interface %s status is now %s",
 495           il->timestamp, dev, ((ifi->ifi_flags & IFF_RUNNING) ? "UP" : "DOWN"));
 496
 497     // no need to loop again, we found the interface name attr
 498     // (otherwise the first if-statement in the loop would
 499     // have moved us on with 'continue')
 500     break;
 501   }
 502
 503   pthread_mutex_unlock(&connectivity_data_lock);
 504
 505   return 0;
 506 }
 507
 508 static int msg_handler(struct nlmsghdr *msg) {
 509   switch (msg->nlmsg_type) {
 510   case RTM_NEWADDR:
 511   case RTM_DELADDR:
 512   case RTM_NEWROUTE:
 513   case RTM_DELROUTE:
 514   case RTM_DELLINK:
 515     // Not of interest in current version
 516     break;
 517   case RTM_NEWLINK:
 518     connectivity_link_state(msg);
 519     break;
 520   default:
 521     ERROR("connectivity plugin: msg_handler: Unknown netlink nlmsg_type %d",
 522           msg->nlmsg_type);
 523     break;
 524   }
 525   return 0;
 526 }
 527
 528 static int read_event(int nl, int (*msg_handler)(struct nlmsghdr *)) {
 529   int ret = 0;
 530   int recv_flags = MSG_DONTWAIT;
 531
 532   if (nl == -1)
 533     return ret;
 534
 535   while (42) {
 536     pthread_mutex_lock(&connectivity_threads_lock);
 537
 538     if (connectivity_netlink_thread_loop <= 0) {
 539       pthread_mutex_unlock(&connectivity_threads_lock);
 540       return ret;
 541     }
 542
 543     pthread_mutex_unlock(&connectivity_threads_lock);
 544
 545     char buf[4096];
 546     int status = recv(nl, buf, sizeof(buf), recv_flags);
 547
 548     if (status < 0) {
 549
 550       // If there were no more messages to drain from the socket,
 551       // then signal the dequeue thread and allow it to dispatch
 552       // any saved interface status changes.  Then continue, but
 553       // block and wait for new messages
 554       if (errno == EWOULDBLOCK || errno == EAGAIN) {
 555         pthread_cond_signal(&connectivity_cond);
 556
 557         recv_flags = 0;
 558         continue;
 559       }
 560
 561       /* Anything else is an error */
 562       ERROR("connectivity plugin: read_event: Error recv: %d", status);
 563       return status;
 564     }
 565
 566     // Message received successfully, so we'll stop blocking on the
 567     // receive call for now (until we get a "would block" error, which
 568     // will be handled above)
 569     recv_flags = MSG_DONTWAIT;
 570
 571     if (status == 0) {
 572       DEBUG("connectivity plugin: read_event: EOF");
 573     }
 574
 575     /* We need to handle more than one message per 'recvmsg' */
 576     for (struct nlmsghdr *h = (struct nlmsghdr *)buf;
 577          NLMSG_OK(h, (unsigned int)status); h = NLMSG_NEXT(h, status)) {
 578       /* Finish reading */
 579       if (h->nlmsg_type == NLMSG_DONE)
 580         return ret;
 581
 582       /* Message is some kind of error */
 583       if (h->nlmsg_type == NLMSG_ERROR) {
 584         ERROR("connectivity plugin: read_event: Message is an error");
 585         return -1; // Error
 586       }
 587
 588       /* Call message handler */
 589       if (msg_handler) {
 590         ret = (*msg_handler)(h);
 591         if (ret < 0) {
 592           ERROR("connectivity plugin: read_event: Message handler error %d",
 593                 ret);
 594           return ret;
 595         }
 596       } else {
 597         ERROR("connectivity plugin: read_event: Error NULL message handler");
 598         return -1;
 599       }
 600     }
 601   }
 602
 603   return ret;
 604 }
 605
 606 // NOTE: Caller MUST hold connectivity_data_lock when calling this function
 607 static void send_interface_status() {
 608   for (interface_list_t *il = interface_list_head; il != NULL;
 609        il = il->next) /* {{{ */
 610   {
 611     uint32_t status;
 612     uint32_t prev_status;
 613     uint32_t sent;
 614
 615     status = il->status;
 616     prev_status = il->prev_status;
 617     sent = il->sent;
 618
 619     if (status != prev_status && sent == 0) {
 620       connectivity_dispatch_notification(il->interface, "gauge", status,
 621                                          prev_status, il->timestamp);
 622       il->sent = 1;
 623     }
 624   } /* }}} for (il = interface_list_head; il != NULL; il = il->next) */
 625
 626   unsent_statuses = 0;
 627 }
 628
 629 static void read_interface_status() /* {{{ */
 630 {
 631   pthread_mutex_lock(&connectivity_data_lock);
 632
 633   if (!unsent_statuses)
 634     pthread_cond_wait(&connectivity_cond, &connectivity_data_lock);
 635
 636   send_interface_status();
 637
 638   pthread_mutex_unlock(&connectivity_data_lock);
 639 } /* }}} int *read_interface_status */
 640
 641 static void *connectivity_netlink_thread(void *arg) /* {{{ */
 642 {
 643   pthread_mutex_lock(&connectivity_threads_lock);
 644
 645   while (connectivity_netlink_thread_loop > 0) {
 646     pthread_mutex_unlock(&connectivity_threads_lock);
 647
 648     int status = read_event(nl_sock, msg_handler);
 649
 650     pthread_mutex_lock(&connectivity_threads_lock);
 651
 652     if (status < 0) {
 653       connectivity_netlink_thread_error = 1;
 654       break;
 655     }
 656   } /* while (connectivity_netlink_thread_loop > 0) */
 657
 658   pthread_mutex_unlock(&connectivity_threads_lock);
 659
 660   return (void *)0;
 661 } /* }}} void *connectivity_netlink_thread */
 662
 663 static void *connectivity_dequeue_thread(void *arg) /* {{{ */
 664 {
 665   pthread_mutex_lock(&connectivity_threads_lock);
 666
 667   while (connectivity_dequeue_thread_loop > 0) {
 668     pthread_mutex_unlock(&connectivity_threads_lock);
 669
 670     read_interface_status();
 671
 672     pthread_mutex_lock(&connectivity_threads_lock);
 673   } /* while (connectivity_dequeue_thread_loop > 0) */
 674
 675   pthread_mutex_unlock(&connectivity_threads_lock);
 676
 677   return ((void *)0);
 678 } /* }}} void *connectivity_dequeue_thread */
 679
 680 static int nl_connect() {
 681   struct sockaddr_nl sa_nl = {
 682       .nl_family = AF_NETLINK, .nl_groups = RTMGRP_LINK, .nl_pid = getpid(),
 683   };
 684
 685   nl_sock = socket(AF_NETLINK, SOCK_DGRAM, NETLINK_ROUTE);
 686   if (nl_sock == -1) {
 687     ERROR("connectivity plugin: socket open failed: %s", STRERRNO);
 688     return -1;
 689   }
 690
 691   int rc = bind(nl_sock, (struct sockaddr *)&sa_nl, sizeof(sa_nl));
 692   if (rc == -1) {
 693     ERROR("connectivity plugin: socket bind failed: %s", STRERRNO);
 694     close(nl_sock);
 695     return -1;
 696   }
 697
 698   return 0;
 699 }
 700
 701 static int start_netlink_thread(void) /* {{{ */
 702 {
 703   pthread_mutex_lock(&connectivity_threads_lock);
 704
 705   if (connectivity_netlink_thread_loop != 0) {
 706     pthread_mutex_unlock(&connectivity_threads_lock);
 707     return 0;
 708   }
 709
 710   connectivity_netlink_thread_loop = 1;
 711   connectivity_netlink_thread_error = 0;
 712
 713   int status;
 714
 715   if (nl_sock == -1) {
 716     status = nl_connect();
 717
 718     if (status != 0) {
 719       pthread_mutex_unlock(&connectivity_threads_lock);
 720       return status;
 721     }
 722   }
 723
 724   status = plugin_thread_create(&connectivity_netlink_thread_id,
 725                                 /* attr = */ NULL, connectivity_netlink_thread,
 726                                 /* arg = */ (void *)0, "connectivity");
 727   if (status != 0) {
 728     connectivity_netlink_thread_loop = 0;
 729     ERROR("connectivity plugin: Starting thread failed.");
 730     pthread_mutex_unlock(&connectivity_threads_lock);
 731
 732     int status2 = close(nl_sock);
 733
 734     if (status2 != 0) {
 735       ERROR("connectivity plugin: failed to close socket %d: %d (%s)", nl_sock,
 736             status2, STRERRNO);
 737     } else
 738       nl_sock = -1;
 739
 740     return -1;
 741   }
 742
 743   pthread_mutex_unlock(&connectivity_threads_lock);
 744
 745   return status;
 746 }
 747
 748 static int start_dequeue_thread(void) /* {{{ */
 749 {
 750   pthread_mutex_lock(&connectivity_threads_lock);
 751
 752   if (connectivity_dequeue_thread_loop != 0) {
 753     pthread_mutex_unlock(&connectivity_threads_lock);
 754     return 0;
 755   }
 756
 757   connectivity_dequeue_thread_loop = 1;
 758
 759   int status =
 760       plugin_thread_create(&connectivity_dequeue_thread_id,
 761                            /* attr = */ NULL, connectivity_dequeue_thread,
 762                            /* arg = */ (void *)0, "connectivity");
 763   if (status != 0) {
 764     connectivity_dequeue_thread_loop = 0;
 765     ERROR("connectivity plugin: Starting dequeue thread failed.");
 766     pthread_mutex_unlock(&connectivity_threads_lock);
 767     return -1;
 768   }
 769
 770   pthread_mutex_unlock(&connectivity_threads_lock);
 771
 772   return status;
 773 } /* }}} int start_dequeue_thread */
 774
 775 static int start_threads(void) /* {{{ */
 776 {
 777   int status = start_netlink_thread();
 778   int status2 = start_dequeue_thread();
 779
 780   if (status != 0)
 781     return status;
 782   else
 783     return status2;
 784 } /* }}} int start_threads */
 785
 786 static int stop_netlink_thread(int shutdown) /* {{{ */
 787 {
 788   int socket_status;
 789
 790   if (nl_sock != -1) {
 791     socket_status = close(nl_sock);
 792     if (socket_status != 0) {
 793       ERROR("connectivity plugin: failed to close socket %d: %d (%s)", nl_sock,
 794             socket_status, STRERRNO);
 795     } else
 796       nl_sock = -1;
 797   } else
 798     socket_status = 0;
 799
 800   pthread_mutex_lock(&connectivity_threads_lock);
 801
 802   if (connectivity_netlink_thread_loop == 0) {
 803     pthread_mutex_unlock(&connectivity_threads_lock);
 804     // Thread has already been terminated, nothing more to attempt
 805     return socket_status;
 806   }
 807
 808   // Set thread termination status
 809   connectivity_netlink_thread_loop = 0;
 810   pthread_mutex_unlock(&connectivity_threads_lock);
 811
 812   // Let threads waiting on access to the interface list know to move
 813   // on such that they'll see the thread's termination status
 814   pthread_cond_broadcast(&connectivity_cond);
 815
 816   int thread_status;
 817
 818   if (shutdown == 1) {
 819     // Since the thread is blocking, calling pthread_join
 820     // doesn't actually succeed in stopping it.  It will stick around
 821     // until a NETLINK message is received on the socket (at which
 822     // it will realize that "connectivity_netlink_thread_loop" is 0 and will
 823     // break out of the read loop and be allowed to die).  This is
 824     // fine when the process isn't supposed to be exiting, but in
 825     // the case of a process shutdown, we don't want to have an
 826     // idle thread hanging around.  Calling pthread_cancel here in
 827     // the case of a shutdown is just assures that the thread is
 828     // gone and that the process has been fully terminated.
 829
 830     DEBUG("connectivity plugin: Canceling netlink thread for process shutdown");
 831
 832     thread_status = pthread_cancel(connectivity_netlink_thread_id);
 833
 834     if (thread_status != 0 && thread_status != ESRCH) {
 835       ERROR("connectivity plugin: Unable to cancel netlink thread: %d",
 836             thread_status);
 837       thread_status = -1;
 838     } else
 839       thread_status = 0;
 840   } else {
 841     thread_status =
 842         pthread_join(connectivity_netlink_thread_id, /* return = */ NULL);
 843     if (thread_status != 0 && thread_status != ESRCH) {
 844       ERROR("connectivity plugin: Stopping netlink thread failed: %d",
 845             thread_status);
 846       thread_status = -1;
 847     } else
 848       thread_status = 0;
 849   }
 850
 851   pthread_mutex_lock(&connectivity_threads_lock);
 852   memset(&connectivity_netlink_thread_id, 0,
 853          sizeof(connectivity_netlink_thread_id));
 854   connectivity_netlink_thread_error = 0;
 855   pthread_mutex_unlock(&connectivity_threads_lock);
 856
 857   DEBUG("connectivity plugin: Finished requesting stop of netlink thread");
 858
 859   if (socket_status != 0)
 860     return socket_status;
 861   else
 862     return thread_status;
 863 }
 864
 865 static int stop_dequeue_thread(int shutdown) /* {{{ */
 866 {
 867   pthread_mutex_lock(&connectivity_threads_lock);
 868
 869   if (connectivity_dequeue_thread_loop == 0) {
 870     pthread_mutex_unlock(&connectivity_threads_lock);
 871     return -1;
 872   }
 873
 874   // Set thread termination status
 875   connectivity_dequeue_thread_loop = 0;
 876   pthread_mutex_unlock(&connectivity_threads_lock);
 877
 878   // Let threads waiting on access to the interface list know to move
 879   // on such that they'll see the threads termination status
 880   pthread_cond_broadcast(&connectivity_cond);
 881
 882   int status;
 883
 884   if (shutdown == 1) {
 885     // Calling pthread_cancel here in
 886     // the case of a shutdown just assures that the thread is
 887     // gone and that the process has been fully terminated.
 888
 889     DEBUG("connectivity plugin: Canceling dequeue thread for process shutdown");
 890
 891     status = pthread_cancel(connectivity_dequeue_thread_id);
 892
 893     if (status != 0 && status != ESRCH) {
 894       ERROR("connectivity plugin: Unable to cancel dequeue thread: %d", status);
 895       status = -1;
 896     } else
 897       status = 0;
 898   } else {
 899     status = pthread_join(connectivity_dequeue_thread_id, /* return = */ NULL);
 900     if (status != 0 && status != ESRCH) {
 901       ERROR("connectivity plugin: Stopping dequeue thread failed.");
 902       status = -1;
 903     } else
 904       status = 0;
 905   }
 906
 907   pthread_mutex_lock(&connectivity_threads_lock);
 908   memset(&connectivity_dequeue_thread_id, 0,
 909          sizeof(connectivity_dequeue_thread_id));
 910   pthread_mutex_unlock(&connectivity_threads_lock);
 911
 912   DEBUG("connectivity plugin: Finished requesting stop of dequeue thread");
 913
 914   return status;
 915 } /* }}} int stop_dequeue_thread */
 916
 917 static int stop_threads(int shutdown) /* {{{ */
 918 {
 919   int status = stop_netlink_thread(shutdown);
 920   int status2 = stop_dequeue_thread(shutdown);
 921
 922   if (status != 0)
 923     return status;
 924   else
 925     return status2;
 926 } /* }}} int stop_threads */
 927
 928 static int connectivity_init(void) /* {{{ */
 929 {
 930   if (monitor_all_interfaces) {
 931     NOTICE("connectivity plugin: No interfaces have been selected, so all will "
 932            "be monitored");
 933   }
 934
 935   return start_threads();
 936 } /* }}} int connectivity_init */
 937
 938 static int connectivity_config(const char *key, const char *value) /* {{{ */
 939 {
 940   if (ignorelist == NULL) {
 941     ignorelist = ignorelist_create(/* invert = */ 1);
 942   }
 943
 944   if (strcasecmp(key, "Interface") == 0) {
 945     ignorelist_add(ignorelist, value);
 946     monitor_all_interfaces = 0;
 947   } else if (strcasecmp(key, "IgnoreSelected") == 0) {
 948     int invert = 1;
 949     if (IS_TRUE(value))
 950       invert = 0;
 951     ignorelist_set_invert(ignorelist, invert);
 952   } else {
 953     return -1;
 954   }
 955
 956   return 0;
 957 } /* }}} int connectivity_config */
 958
 959 static void
 960 connectivity_dispatch_notification(const char *interface, const char *type,
 961                                    gauge_t value, gauge_t old_value,
 962                                    long long unsigned int timestamp) {
 963
 964   notification_t n = {(value == LINK_STATE_UP ? NOTIF_OKAY : NOTIF_FAILURE),
 965                       cdtime(),
 966                       "",
 967                       "",
 968                       "connectivity",
 969                       "",
 970                       "",
 971                       "",
 972                       NULL};
 973
 974   sstrncpy(n.host, hostname_g, sizeof(n.host));
 975   sstrncpy(n.plugin_instance, interface, sizeof(n.plugin_instance));
 976   sstrncpy(n.type, "gauge", sizeof(n.type));
 977   sstrncpy(n.type_instance, "interface_status", sizeof(n.type_instance));
 978
 979   char *buf = NULL;
 980
 981   gen_message_payload(value, old_value, interface, timestamp, &buf);
 982
 983   notification_meta_t *m = calloc(1, sizeof(*m));
 984
 985   if (m == NULL) {
 986     sfree(buf);
 987     ERROR("connectivity plugin: unable to allocate metadata: %s", STRERRNO);
 988     return;
 989   }
 990
 991   sstrncpy(m->name, "ves", sizeof(m->name));
 992   m->nm_value.nm_string = sstrdup(buf);
 993   m->type = NM_TYPE_STRING;
 994   n.meta = m;
 995
 996   DEBUG("connectivity plugin: notification message: %s",
 997         n.meta->nm_value.nm_string);
 998
 999   DEBUG("connectivity plugin: dispatching state %d for interface %s",
1000         (int)value, interface);
1001
1002   plugin_dispatch_notification(&n);
1003   plugin_notification_meta_free(n.meta);
1004
1005   // strdup'd in gen_message_payload
1006   if (buf != NULL)
1007     sfree(buf);
1008 }
1009
1010 static int connectivity_read(void) /* {{{ */
1011 {
1012   pthread_mutex_lock(&connectivity_threads_lock);
1013
1014   if (connectivity_netlink_thread_error != 0) {
1015
1016     pthread_mutex_unlock(&connectivity_threads_lock);
1017
1018     ERROR("connectivity plugin: The netlink thread had a problem. Restarting "
1019           "it.");
1020
1021     stop_netlink_thread(0);
1022
1023     for (interface_list_t *il = interface_list_head; il != NULL;
1024          il = il->next) {
1025       il->status = LINK_STATE_UNKNOWN;
1026       il->prev_status = LINK_STATE_UNKNOWN;
1027       il->sent = 0;
1028     }
1029
1030     start_netlink_thread();
1031
1032     return -1;
1033   } /* if (connectivity_netlink_thread_error != 0) */
1034
1035   pthread_mutex_unlock(&connectivity_threads_lock);
1036
1037   return 0;
1038 } /* }}} int connectivity_read */
1039
1040 static int connectivity_shutdown(void) /* {{{ */
1041 {
1042   DEBUG("connectivity plugin: Shutting down thread.");
1043
1044   int status = stop_threads(1);
1045
1046   interface_list_t *il = interface_list_head;
1047   while (il != NULL) {
1048     interface_list_t *il_next;
1049
1050     il_next = il->next;
1051
1052     sfree(il->interface);
1053     sfree(il);
1054
1055     il = il_next;
1056   }
1057
1058   ignorelist_free(ignorelist);
1059
1060   return status;
1061 } /* }}} int connectivity_shutdown */
1062
1063 void module_register(void) {
1064   plugin_register_config("connectivity", connectivity_config, config_keys,
1065                          config_keys_num);
1066   plugin_register_init("connectivity", connectivity_init);
1067   plugin_register_read("connectivity", connectivity_read);
1068   plugin_register_shutdown("connectivity", connectivity_shutdown);
1069 } /* void module_register */