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       if (errno == EINTR)
 562       {
 563         // Interrupt, so just return
 564         return 0;
 565       }
 566
 567       /* Anything else is an error */
 568       ERROR("connectivity plugin: read_event: Error recv: %d", status);
 569       return status;
 570     }
 571
 572     // Message received successfully, so we'll stop blocking on the
 573     // receive call for now (until we get a "would block" error, which
 574     // will be handled above)
 575     recv_flags = MSG_DONTWAIT;
 576
 577     if (status == 0) {
 578       DEBUG("connectivity plugin: read_event: EOF");
 579     }
 580
 581     /* We need to handle more than one message per 'recvmsg' */
 582     for (struct nlmsghdr *h = (struct nlmsghdr *)buf;
 583          NLMSG_OK(h, (unsigned int)status); h = NLMSG_NEXT(h, status)) {
 584       /* Finish reading */
 585       if (h->nlmsg_type == NLMSG_DONE)
 586         return ret;
 587
 588       /* Message is some kind of error */
 589       if (h->nlmsg_type == NLMSG_ERROR) {
 590         ERROR("connectivity plugin: read_event: Message is an error");
 591         return -1; // Error
 592       }
 593
 594       /* Call message handler */
 595       if (msg_handler) {
 596         ret = (*msg_handler)(h);
 597         if (ret < 0) {
 598           ERROR("connectivity plugin: read_event: Message handler error %d",
 599                 ret);
 600           return ret;
 601         }
 602       } else {
 603         ERROR("connectivity plugin: read_event: Error NULL message handler");
 604         return -1;
 605       }
 606     }
 607   }
 608
 609   return ret;
 610 }
 611
 612 // NOTE: Caller MUST hold connectivity_data_lock when calling this function
 613 static void send_interface_status() {
 614   for (interface_list_t *il = interface_list_head; il != NULL;
 615        il = il->next) /* {{{ */
 616   {
 617     uint32_t status;
 618     uint32_t prev_status;
 619     uint32_t sent;
 620
 621     status = il->status;
 622     prev_status = il->prev_status;
 623     sent = il->sent;
 624
 625     if (status != prev_status && sent == 0) {
 626       connectivity_dispatch_notification(il->interface, "gauge", status,
 627                                          prev_status, il->timestamp);
 628       il->sent = 1;
 629     }
 630   } /* }}} for (il = interface_list_head; il != NULL; il = il->next) */
 631
 632   unsent_statuses = 0;
 633 }
 634
 635 static void read_interface_status() /* {{{ */
 636 {
 637   pthread_mutex_lock(&connectivity_data_lock);
 638
 639   if (!unsent_statuses)
 640     pthread_cond_wait(&connectivity_cond, &connectivity_data_lock);
 641
 642   send_interface_status();
 643
 644   pthread_mutex_unlock(&connectivity_data_lock);
 645 } /* }}} int *read_interface_status */
 646
 647 static void *connectivity_netlink_thread(void *arg) /* {{{ */
 648 {
 649   pthread_mutex_lock(&connectivity_threads_lock);
 650
 651   while (connectivity_netlink_thread_loop > 0) {
 652     pthread_mutex_unlock(&connectivity_threads_lock);
 653
 654     int status = read_event(nl_sock, msg_handler);
 655
 656     pthread_mutex_lock(&connectivity_threads_lock);
 657
 658     if (status < 0) {
 659       connectivity_netlink_thread_error = 1;
 660       break;
 661     }
 662   } /* while (connectivity_netlink_thread_loop > 0) */
 663
 664   pthread_mutex_unlock(&connectivity_threads_lock);
 665
 666   return (void *)0;
 667 } /* }}} void *connectivity_netlink_thread */
 668
 669 static void *connectivity_dequeue_thread(void *arg) /* {{{ */
 670 {
 671   pthread_mutex_lock(&connectivity_threads_lock);
 672
 673   while (connectivity_dequeue_thread_loop > 0) {
 674     pthread_mutex_unlock(&connectivity_threads_lock);
 675
 676     read_interface_status();
 677
 678     pthread_mutex_lock(&connectivity_threads_lock);
 679   } /* while (connectivity_dequeue_thread_loop > 0) */
 680
 681   pthread_mutex_unlock(&connectivity_threads_lock);
 682
 683   return ((void *)0);
 684 } /* }}} void *connectivity_dequeue_thread */
 685
 686 static int nl_connect() {
 687   struct sockaddr_nl sa_nl = {
 688       .nl_family = AF_NETLINK, .nl_groups = RTMGRP_LINK, .nl_pid = getpid(),
 689   };
 690
 691   nl_sock = socket(AF_NETLINK, SOCK_DGRAM, NETLINK_ROUTE);
 692   if (nl_sock == -1) {
 693     ERROR("connectivity plugin: socket open failed: %s", STRERRNO);
 694     return -1;
 695   }
 696
 697   int rc = bind(nl_sock, (struct sockaddr *)&sa_nl, sizeof(sa_nl));
 698   if (rc == -1) {
 699     ERROR("connectivity plugin: socket bind failed: %s", STRERRNO);
 700     close(nl_sock);
 701     return -1;
 702   }
 703
 704   return 0;
 705 }
 706
 707 static int start_netlink_thread(void) /* {{{ */
 708 {
 709   pthread_mutex_lock(&connectivity_threads_lock);
 710
 711   if (connectivity_netlink_thread_loop != 0) {
 712     pthread_mutex_unlock(&connectivity_threads_lock);
 713     return 0;
 714   }
 715
 716   connectivity_netlink_thread_loop = 1;
 717   connectivity_netlink_thread_error = 0;
 718
 719   int status;
 720
 721   if (nl_sock == -1) {
 722     status = nl_connect();
 723
 724     if (status != 0) {
 725       pthread_mutex_unlock(&connectivity_threads_lock);
 726       return status;
 727     }
 728   }
 729
 730   status = plugin_thread_create(&connectivity_netlink_thread_id,
 731                                 /* attr = */ NULL, connectivity_netlink_thread,
 732                                 /* arg = */ (void *)0, "connectivity");
 733   if (status != 0) {
 734     connectivity_netlink_thread_loop = 0;
 735     ERROR("connectivity plugin: Starting thread failed.");
 736     pthread_mutex_unlock(&connectivity_threads_lock);
 737
 738     int status2 = close(nl_sock);
 739
 740     if (status2 != 0) {
 741       ERROR("connectivity plugin: failed to close socket %d: %d (%s)", nl_sock,
 742             status2, STRERRNO);
 743     } else
 744       nl_sock = -1;
 745
 746     return -1;
 747   }
 748
 749   pthread_mutex_unlock(&connectivity_threads_lock);
 750
 751   return status;
 752 }
 753
 754 static int start_dequeue_thread(void) /* {{{ */
 755 {
 756   pthread_mutex_lock(&connectivity_threads_lock);
 757
 758   if (connectivity_dequeue_thread_loop != 0) {
 759     pthread_mutex_unlock(&connectivity_threads_lock);
 760     return 0;
 761   }
 762
 763   connectivity_dequeue_thread_loop = 1;
 764
 765   int status =
 766       plugin_thread_create(&connectivity_dequeue_thread_id,
 767                            /* attr = */ NULL, connectivity_dequeue_thread,
 768                            /* arg = */ (void *)0, "connectivity");
 769   if (status != 0) {
 770     connectivity_dequeue_thread_loop = 0;
 771     ERROR("connectivity plugin: Starting dequeue thread failed.");
 772     pthread_mutex_unlock(&connectivity_threads_lock);
 773     return -1;
 774   }
 775
 776   pthread_mutex_unlock(&connectivity_threads_lock);
 777
 778   return status;
 779 } /* }}} int start_dequeue_thread */
 780
 781 static int start_threads(void) /* {{{ */
 782 {
 783   int status = start_netlink_thread();
 784   int status2 = start_dequeue_thread();
 785
 786   if (status != 0)
 787     return status;
 788   else
 789     return status2;
 790 } /* }}} int start_threads */
 791
 792 static int stop_netlink_thread(int shutdown) /* {{{ */
 793 {
 794   int socket_status;
 795
 796   if (nl_sock != -1) {
 797     socket_status = close(nl_sock);
 798     if (socket_status != 0) {
 799       ERROR("connectivity plugin: failed to close socket %d: %d (%s)", nl_sock,
 800             socket_status, STRERRNO);
 801     } else
 802       nl_sock = -1;
 803   } else
 804     socket_status = 0;
 805
 806   pthread_mutex_lock(&connectivity_threads_lock);
 807
 808   if (connectivity_netlink_thread_loop == 0) {
 809     pthread_mutex_unlock(&connectivity_threads_lock);
 810     // Thread has already been terminated, nothing more to attempt
 811     return socket_status;
 812   }
 813
 814   // Set thread termination status
 815   connectivity_netlink_thread_loop = 0;
 816   pthread_mutex_unlock(&connectivity_threads_lock);
 817
 818   // Let threads waiting on access to the interface list know to move
 819   // on such that they'll see the thread's termination status
 820   pthread_cond_broadcast(&connectivity_cond);
 821
 822   int thread_status;
 823
 824   if (shutdown == 1) {
 825     // Since the thread is blocking, calling pthread_join
 826     // doesn't actually succeed in stopping it.  It will stick around
 827     // until a NETLINK message is received on the socket (at which
 828     // it will realize that "connectivity_netlink_thread_loop" is 0 and will
 829     // break out of the read loop and be allowed to die).  This is
 830     // fine when the process isn't supposed to be exiting, but in
 831     // the case of a process shutdown, we don't want to have an
 832     // idle thread hanging around.  Calling pthread_cancel here in
 833     // the case of a shutdown is just assures that the thread is
 834     // gone and that the process has been fully terminated.
 835
 836     DEBUG("connectivity plugin: Canceling netlink thread for process shutdown");
 837
 838     thread_status = pthread_cancel(connectivity_netlink_thread_id);
 839
 840     if (thread_status != 0 && thread_status != ESRCH) {
 841       ERROR("connectivity plugin: Unable to cancel netlink thread: %d",
 842             thread_status);
 843       thread_status = -1;
 844     } else
 845       thread_status = 0;
 846   } else {
 847     thread_status =
 848         pthread_join(connectivity_netlink_thread_id, /* return = */ NULL);
 849     if (thread_status != 0 && thread_status != ESRCH) {
 850       ERROR("connectivity plugin: Stopping netlink thread failed: %d",
 851             thread_status);
 852       thread_status = -1;
 853     } else
 854       thread_status = 0;
 855   }
 856
 857   pthread_mutex_lock(&connectivity_threads_lock);
 858   memset(&connectivity_netlink_thread_id, 0,
 859          sizeof(connectivity_netlink_thread_id));
 860   connectivity_netlink_thread_error = 0;
 861   pthread_mutex_unlock(&connectivity_threads_lock);
 862
 863   DEBUG("connectivity plugin: Finished requesting stop of netlink thread");
 864
 865   if (socket_status != 0)
 866     return socket_status;
 867   else
 868     return thread_status;
 869 }
 870
 871 static int stop_dequeue_thread(int shutdown) /* {{{ */
 872 {
 873   pthread_mutex_lock(&connectivity_threads_lock);
 874
 875   if (connectivity_dequeue_thread_loop == 0) {
 876     pthread_mutex_unlock(&connectivity_threads_lock);
 877     return -1;
 878   }
 879
 880   // Set thread termination status
 881   connectivity_dequeue_thread_loop = 0;
 882   pthread_mutex_unlock(&connectivity_threads_lock);
 883
 884   // Let threads waiting on access to the interface list know to move
 885   // on such that they'll see the threads termination status
 886   pthread_cond_broadcast(&connectivity_cond);
 887
 888   int status;
 889
 890   if (shutdown == 1) {
 891     // Calling pthread_cancel here in
 892     // the case of a shutdown just assures that the thread is
 893     // gone and that the process has been fully terminated.
 894
 895     DEBUG("connectivity plugin: Canceling dequeue thread for process shutdown");
 896
 897     status = pthread_cancel(connectivity_dequeue_thread_id);
 898
 899     if (status != 0 && status != ESRCH) {
 900       ERROR("connectivity plugin: Unable to cancel dequeue thread: %d", status);
 901       status = -1;
 902     } else
 903       status = 0;
 904   } else {
 905     status = pthread_join(connectivity_dequeue_thread_id, /* return = */ NULL);
 906     if (status != 0 && status != ESRCH) {
 907       ERROR("connectivity plugin: Stopping dequeue thread failed.");
 908       status = -1;
 909     } else
 910       status = 0;
 911   }
 912
 913   pthread_mutex_lock(&connectivity_threads_lock);
 914   memset(&connectivity_dequeue_thread_id, 0,
 915          sizeof(connectivity_dequeue_thread_id));
 916   pthread_mutex_unlock(&connectivity_threads_lock);
 917
 918   DEBUG("connectivity plugin: Finished requesting stop of dequeue thread");
 919
 920   return status;
 921 } /* }}} int stop_dequeue_thread */
 922
 923 static int stop_threads(int shutdown) /* {{{ */
 924 {
 925   int status = stop_netlink_thread(shutdown);
 926   int status2 = stop_dequeue_thread(shutdown);
 927
 928   if (status != 0)
 929     return status;
 930   else
 931     return status2;
 932 } /* }}} int stop_threads */
 933
 934 static int connectivity_init(void) /* {{{ */
 935 {
 936   if (monitor_all_interfaces) {
 937     NOTICE("connectivity plugin: No interfaces have been selected, so all will "
 938            "be monitored");
 939   }
 940
 941   return start_threads();
 942 } /* }}} int connectivity_init */
 943
 944 static int connectivity_config(const char *key, const char *value) /* {{{ */
 945 {
 946   if (ignorelist == NULL) {
 947     ignorelist = ignorelist_create(/* invert = */ 1);
 948   }
 949
 950   if (strcasecmp(key, "Interface") == 0) {
 951     ignorelist_add(ignorelist, value);
 952     monitor_all_interfaces = 0;
 953   } else if (strcasecmp(key, "IgnoreSelected") == 0) {
 954     int invert = 1;
 955     if (IS_TRUE(value))
 956       invert = 0;
 957     ignorelist_set_invert(ignorelist, invert);
 958   } else {
 959     return -1;
 960   }
 961
 962   return 0;
 963 } /* }}} int connectivity_config */
 964
 965 static void
 966 connectivity_dispatch_notification(const char *interface, const char *type,
 967                                    gauge_t value, gauge_t old_value,
 968                                    long long unsigned int timestamp) {
 969
 970   notification_t n = {(value == LINK_STATE_UP ? NOTIF_OKAY : NOTIF_FAILURE),
 971                       cdtime(),
 972                       "",
 973                       "",
 974                       "connectivity",
 975                       "",
 976                       "",
 977                       "",
 978                       NULL};
 979
 980   sstrncpy(n.host, hostname_g, sizeof(n.host));
 981   sstrncpy(n.plugin_instance, interface, sizeof(n.plugin_instance));
 982   sstrncpy(n.type, "gauge", sizeof(n.type));
 983   sstrncpy(n.type_instance, "interface_status", sizeof(n.type_instance));
 984
 985   char *buf = NULL;
 986
 987   gen_message_payload(value, old_value, interface, timestamp, &buf);
 988
 989   notification_meta_t *m = calloc(1, sizeof(*m));
 990
 991   if (m == NULL) {
 992     sfree(buf);
 993     ERROR("connectivity plugin: unable to allocate metadata: %s", STRERRNO);
 994     return;
 995   }
 996
 997   sstrncpy(m->name, "ves", sizeof(m->name));
 998   m->nm_value.nm_string = sstrdup(buf);
 999   m->type = NM_TYPE_STRING;
1000   n.meta = m;
1001
1002   DEBUG("connectivity plugin: notification message: %s",
1003         n.meta->nm_value.nm_string);
1004
1005   DEBUG("connectivity plugin: dispatching state %d for interface %s",
1006         (int)value, interface);
1007
1008   plugin_dispatch_notification(&n);
1009   plugin_notification_meta_free(n.meta);
1010
1011   // strdup'd in gen_message_payload
1012   if (buf != NULL)
1013     sfree(buf);
1014 }
1015
1016 static int connectivity_read(void) /* {{{ */
1017 {
1018   pthread_mutex_lock(&connectivity_threads_lock);
1019
1020   if (connectivity_netlink_thread_error != 0) {
1021
1022     pthread_mutex_unlock(&connectivity_threads_lock);
1023
1024     ERROR("connectivity plugin: The netlink thread had a problem. Restarting "
1025           "it.");
1026
1027     stop_netlink_thread(0);
1028
1029     for (interface_list_t *il = interface_list_head; il != NULL;
1030          il = il->next) {
1031       il->status = LINK_STATE_UNKNOWN;
1032       il->prev_status = LINK_STATE_UNKNOWN;
1033       il->sent = 0;
1034     }
1035
1036     start_netlink_thread();
1037
1038     return -1;
1039   } /* if (connectivity_netlink_thread_error != 0) */
1040
1041   pthread_mutex_unlock(&connectivity_threads_lock);
1042
1043   return 0;
1044 } /* }}} int connectivity_read */
1045
1046 static int connectivity_shutdown(void) /* {{{ */
1047 {
1048   DEBUG("connectivity plugin: Shutting down thread.");
1049
1050   int status = stop_threads(1);
1051
1052   interface_list_t *il = interface_list_head;
1053   while (il != NULL) {
1054     interface_list_t *il_next;
1055
1056     il_next = il->next;
1057
1058     sfree(il->interface);
1059     sfree(il);
1060
1061     il = il_next;
1062   }
1063
1064   ignorelist_free(ignorelist);
1065
1066   return status;
1067 } /* }}} int connectivity_shutdown */
1068
1069 void module_register(void) {
1070   plugin_register_config("connectivity", connectivity_config, config_keys,
1071                          config_keys_num);
1072   plugin_register_init("connectivity", connectivity_init);
1073   plugin_register_read("connectivity", connectivity_read);
1074   plugin_register_shutdown("connectivity", connectivity_shutdown);
1075 } /* void module_register */