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