src/riemann.c

   1 /*
   2  * collectd - src/riemann.c
   3  *
   4  * Copyright (C) 2012  Pierre-Yves Ritschard <pyr@spootnik.org>
   5  *
   6  * Permission to use, copy, modify, and distribute this software for any
   7  * purpose with or without fee is hereby granted, provided that the above
   8  * copyright notice and this permission notice appear in all copies.
   9  *
  10  * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
  11  * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  12  * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
  13  * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  14  * WHATSOEVER RESULTING FROM LOSS OF MIND, USE, DATA OR PROFITS, WHETHER
  15  * IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING
  16  * OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  17  *
  18  */
  19
  20 #include "collectd.h"
  21 #include "plugin.h"
  22 #include "common.h"
  23 #include "configfile.h"
  24 #include "riemann.pb-c.h"
  25
  26 #include <sys/socket.h>
  27 #include <arpa/inet.h>
  28 #include <errno.h>
  29 #include <netdb.h>
  30 #include <inttypes.h>
  31 #include <pthread.h>
  32
  33 #define RIEMANN_DELAY           1
  34 #define RIEMANN_PORT            "5555"
  35 #define RIEMANN_MAX_TAGS        37
  36 #define RIEMANN_EXTRA_TAGS      32
  37
  38 struct riemann_host {
  39 #define F_CONNECT                0x01
  40         uint8_t                  flags;
  41         pthread_mutex_t          lock;
  42         int                      delay;
  43         char                    *node;
  44         char                    *service;
  45         int                      s;
  46
  47         int                      reference_count;
  48 };
  49
  50 static char     *riemann_tags[RIEMANN_EXTRA_TAGS];
  51 static int       riemann_tagcount;
  52
  53 static int      riemann_send(struct riemann_host *, Msg const *);
  54 static int      riemann_notification(const notification_t *, user_data_t *);
  55 static int      riemann_write(const data_set_t *, const value_list_t *, user_data_t *);
  56 static int      riemann_connect(struct riemann_host *);
  57 static int      riemann_disconnect (struct riemann_host *host);
  58 static void     riemann_free(void *);
  59 static int      riemann_config_host(oconfig_item_t *);
  60 static int      riemann_config(oconfig_item_t *);
  61 void    module_register(void);
  62
  63 static void riemann_event_protobuf_free (Event *event) /* {{{ */
  64 {
  65         size_t i;
  66
  67         if (event == NULL)
  68                 return;
  69
  70         sfree (event->state);
  71         sfree (event->service);
  72         sfree (event->host);
  73         sfree (event->description);
  74
  75         for (i = 0; i < event->n_tags; i++)
  76                 sfree (event->tags[i]);
  77         sfree (event->tags);
  78
  79         sfree (event);
  80 } /* }}} void riemann_event_protobuf_free */
  81
  82 static void riemann_msg_protobuf_free (Msg *msg) /* {{{ */
  83 {
  84         size_t i;
  85
  86         if (msg == NULL)
  87                 return;
  88
  89         for (i = 0; i < msg->n_events; i++)
  90         {
  91                 riemann_event_protobuf_free (msg->events[i]);
  92                 msg->events[i] = NULL;
  93         }
  94
  95         sfree (msg->events);
  96         msg->n_events = 0;
  97
  98         sfree (msg);
  99 } /* }}} void riemann_msg_protobuf_free */
 100
 101 static int
 102 riemann_send(struct riemann_host *host, Msg const *msg)
 103 {
 104         u_char *buffer;
 105         size_t  buffer_len;
 106         int status;
 107
 108         pthread_mutex_lock (&host->lock);
 109
 110         status = riemann_connect (host);
 111         if (status != 0)
 112         {
 113                 pthread_mutex_unlock (&host->lock);
 114                 return status;
 115         }
 116
 117         buffer_len = msg__get_packed_size(msg);
 118         buffer = malloc (buffer_len);
 119         if (buffer == NULL) {
 120                 pthread_mutex_unlock (&host->lock);
 121                 ERROR ("riemann plugin: malloc failed.");
 122                 return ENOMEM;
 123         }
 124         memset (buffer, 0, buffer_len);
 125
 126         msg__pack(msg, buffer);
 127
 128         status = (int) swrite (host->s, buffer, buffer_len);
 129         if (status != 0)
 130         {
 131                 char errbuf[1024];
 132
 133                 riemann_disconnect (host);
 134                 pthread_mutex_unlock (&host->lock);
 135
 136                 ERROR ("riemann plugin: Sending to Riemann at %s:%s failed: %s",
 137                                 host->node,
 138                                 (host->service != NULL) ? host->service : RIEMANN_PORT,
 139                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 140                 sfree (buffer);
 141                 return -1;
 142         }
 143
 144         pthread_mutex_unlock (&host->lock);
 145         sfree (buffer);
 146         return 0;
 147 }
 148
 149 static int riemann_event_add_tag (Event *event, /* {{{ */
 150                 char const *format, ...)
 151 {
 152         va_list ap;
 153         char buffer[1024];
 154         size_t ret;
 155
 156         char **tmp;
 157
 158         tmp = realloc (event->tags, (event->n_tags + 1) * sizeof (*event->tags));
 159         if (tmp == NULL)
 160                 return (ENOMEM);
 161         event->tags = tmp;
 162
 163         va_start (ap, format);
 164         ret = vsnprintf (buffer, sizeof (buffer), format, ap);
 165         if (ret >= sizeof (buffer))
 166                 ret = sizeof (buffer) - 1;
 167         buffer[ret] = 0;
 168         va_end (ap);
 169
 170         event->tags[event->n_tags] = strdup (buffer);
 171         if (event->tags[event->n_tags] == NULL)
 172                 return (ENOMEM);
 173         event->n_tags++;
 174         return (0);
 175 } /* }}} int riemann_event_add_tag */
 176
 177 static Msg *riemann_notification_to_protobuf (struct riemann_host *host, /* {{{ */
 178                 notification_t const *n)
 179 {
 180         Msg *msg;
 181         Event *event;
 182         char service_buffer[6 * DATA_MAX_NAME_LEN];
 183         char const *severity;
 184         notification_meta_t *meta;
 185         int i;
 186
 187         msg = malloc (sizeof (*msg));
 188         if (msg == NULL)
 189         {
 190                 ERROR ("riemann plugin: malloc failed.");
 191                 return (NULL);
 192         }
 193         memset (msg, 0, sizeof (*msg));
 194         msg__init (msg);
 195
 196         msg->events = malloc (sizeof (*msg->events));
 197         if (msg->events == NULL)
 198         {
 199                 ERROR ("riemann plugin: malloc failed.");
 200                 sfree (msg);
 201                 return (NULL);
 202         }
 203
 204         event = malloc (sizeof (*event));
 205         if (event == NULL)
 206         {
 207                 ERROR ("riemann plugin: malloc failed.");
 208                 sfree (msg->events);
 209                 sfree (msg);
 210                 return (NULL);
 211         }
 212         memset (event, 0, sizeof (*event));
 213         event__init (event);
 214
 215         msg->events[0] = event;
 216         msg->n_events = 1;
 217
 218         event->host = strdup (n->host);
 219         event->time = CDTIME_T_TO_TIME_T (n->time);
 220         event->has_time = 1;
 221
 222         switch (n->severity)
 223         {
 224                 case NOTIF_OKAY:        severity = "okay"; break;
 225                 case NOTIF_WARNING:     severity = "warning"; break;
 226                 case NOTIF_FAILURE:     severity = "failure"; break;
 227                 default:                severity = "unknown";
 228         }
 229         event->state = strdup (severity);
 230
 231         riemann_event_add_tag (event, "notification");
 232         if (n->plugin[0] != 0)
 233                 riemann_event_add_tag (event, "plugin:%s", n->plugin);
 234         if (n->plugin_instance[0] != 0)
 235                 riemann_event_add_tag (event, "plugin_instance:%s",
 236                                 n->plugin_instance);
 237
 238         if (n->type[0] != 0)
 239                 riemann_event_add_tag (event, "type:%s", n->type);
 240         if (n->type_instance[0] != 0)
 241                 riemann_event_add_tag (event, "type_instance:%s",
 242                                 n->type_instance);
 243
 244         for (i = 0; i < riemann_tagcount; i++)
 245                 riemann_event_add_tag (event, "%s", riemann_tags[i]);
 246
 247         /* TODO: Use FORMAT_VL() here. */
 248         ssnprintf (service_buffer, sizeof(service_buffer),
 249                         "%s-%s-%s-%s", n->plugin, n->plugin_instance,
 250                         n->type, n->type_instance);
 251         event->service = strdup (service_buffer);
 252
 253         /* Pull in values from threshold */
 254         for (meta = n->meta; meta != NULL; meta = meta->next)
 255         {
 256                 if (strcasecmp ("CurrentValue", meta->name) != 0)
 257                         continue;
 258
 259                 event->metric_d = meta->nm_value.nm_double;
 260                 event->has_metric_d = 1;
 261                 break;
 262         }
 263
 264         DEBUG ("riemann plugin: Successfully created protobuf for notification: "
 265                         "host = \"%s\", service = \"%s\", state = \"%s\"",
 266                         event->host, event->service, event->state);
 267         return (msg);
 268 } /* }}} Msg *riemann_notification_to_protobuf */
 269
 270 static Event *riemann_value_to_protobuf (struct riemann_host const *host, /* {{{ */
 271                 data_set_t const *ds,
 272                 value_list_t const *vl, size_t index,
 273                 gauge_t const *rates)
 274 {
 275         Event *event;
 276         char service_buffer[6 * DATA_MAX_NAME_LEN];
 277         int i;
 278
 279         event = malloc (sizeof (*event));
 280         if (event == NULL)
 281         {
 282                 ERROR ("riemann plugin: malloc failed.");
 283                 return (NULL);
 284         }
 285         memset (event, 0, sizeof (*event));
 286         event__init (event);
 287
 288         event->host = strdup (vl->host);
 289         event->time = CDTIME_T_TO_TIME_T (vl->time);
 290         event->has_time = 1;
 291         event->ttl = CDTIME_T_TO_TIME_T (vl->interval) + host->delay;
 292         event->has_ttl = 1;
 293
 294         riemann_event_add_tag (event, "plugin:%s", vl->plugin);
 295         if (vl->plugin_instance[0] != 0)
 296                 riemann_event_add_tag (event, "plugin_instance:%s",
 297                                 vl->plugin_instance);
 298
 299         riemann_event_add_tag (event, "type:%s", vl->type);
 300         if (vl->type_instance[0] != 0)
 301                 riemann_event_add_tag (event, "type_instance:%s",
 302                                 vl->type_instance);
 303
 304         riemann_event_add_tag (event, "ds_type:%s",
 305                         DS_TYPE_TO_STRING(ds->ds[index].type));
 306         riemann_event_add_tag (event, "ds_name:%s", ds->ds[index].name);
 307         riemann_event_add_tag (event, "ds_index:%zu", index);
 308
 309         for (i = 0; i < riemann_tagcount; i++)
 310                 riemann_event_add_tag (event, "%s", riemann_tags[i]);
 311
 312         if (rates != NULL)
 313         {
 314                 event->has_metric_d = 1;
 315                 event->metric_d = (double) rates[index];
 316         }
 317         else if (ds->ds[index].type == DS_TYPE_GAUGE)
 318         {
 319                 event->has_metric_d = 1;
 320                 event->metric_d = (double) vl->values[index].gauge;
 321         }
 322         else
 323         {
 324                 event->has_metric_sint64 = 1;
 325                 if (ds->ds[index].type == DS_TYPE_DERIVE)
 326                         event->metric_sint64 = (int64_t) vl->values[index].derive;
 327                 else if (ds->ds[index].type == DS_TYPE_ABSOLUTE)
 328                         event->metric_sint64 = (int64_t) vl->values[index].absolute;
 329                 else
 330                         event->metric_sint64 = (int64_t) vl->values[index].counter;
 331         }
 332
 333         /* TODO: Use FORMAT_VL() here. */
 334         ssnprintf (service_buffer, sizeof(service_buffer),
 335                         "%s-%s-%s-%s-%s", vl->plugin, vl->plugin_instance,
 336                         vl->type, vl->type_instance, ds->ds[index].name);
 337         event->service = strdup (service_buffer);
 338
 339         DEBUG ("riemann plugin: Successfully created protobuf for metric: "
 340                         "host = \"%s\", service = \"%s\"",
 341                         event->host, event->service);
 342         return (event);
 343 } /* }}} Event *riemann_value_to_protobuf */
 344
 345 static Msg *riemann_value_list_to_protobuf (struct riemann_host const *host, /* {{{ */
 346                 data_set_t const *ds,
 347                 value_list_t const *vl)
 348 {
 349         Msg *msg;
 350         size_t i;
 351
 352         /* Initialize the Msg structure. */
 353         msg = malloc (sizeof (*msg));
 354         if (msg == NULL)
 355         {
 356                 ERROR ("riemann plugin: malloc failed.");
 357                 return (NULL);
 358         }
 359         memset (msg, 0, sizeof (*msg));
 360         msg__init (msg);
 361
 362         /* Set up events. First, the list of pointers. */
 363         msg->n_events = (size_t) vl->values_len;
 364         msg->events = calloc (msg->n_events, sizeof (*msg->events));
 365         if (msg->events == NULL)
 366         {
 367                 ERROR ("riemann plugin: calloc failed.");
 368                 riemann_msg_protobuf_free (msg);
 369                 return (NULL);
 370         }
 371
 372         for (i = 0; i < msg->n_events; i++)
 373         {
 374                 msg->events[i] = riemann_value_to_protobuf (host, ds, vl,
 375                                 (int) i, /* rates = */ NULL);
 376                 if (msg->events[i] == NULL)
 377                 {
 378                         riemann_msg_protobuf_free (msg);
 379                         return (NULL);
 380                 }
 381         }
 382
 383         return (msg);
 384 } /* }}} Msg *riemann_value_list_to_protobuf */
 385
 386 static int
 387 riemann_notification(const notification_t *n, user_data_t *ud)
 388 {
 389         int                      status;
 390         struct riemann_host     *host = ud->data;
 391         Msg                     *msg;
 392
 393         msg = riemann_notification_to_protobuf (host, n);
 394         if (msg == NULL)
 395                 return (-1);
 396
 397         status = riemann_send (host, msg);
 398         if (status != 0)
 399                 ERROR ("riemann plugin: riemann_send failed with status %i",
 400                                 status);
 401
 402         riemann_msg_protobuf_free (msg);
 403         return (status);
 404 } /* }}} int riemann_notification */
 405
 406 static int
 407 riemann_write(const data_set_t *ds,
 408               const value_list_t *vl,
 409               user_data_t *ud)
 410 {
 411         int                      status;
 412         struct riemann_host     *host = ud->data;
 413         Msg                     *msg;
 414
 415         msg = riemann_value_list_to_protobuf (host, ds, vl);
 416         if (msg == NULL)
 417                 return (-1);
 418
 419         status = riemann_send (host, msg);
 420         if (status != 0)
 421                 ERROR ("riemann plugin: riemann_send failed with status %i",
 422                                 status);
 423
 424         riemann_msg_protobuf_free (msg);
 425         return status;
 426 }
 427
 428 /* host->lock must be held when calling this function. */
 429 static int
 430 riemann_connect(struct riemann_host *host)
 431 {
 432         int                      e;
 433         struct addrinfo         *ai, *res, hints;
 434         char const              *service;
 435
 436         if (host->flags & F_CONNECT)
 437                 return 0;
 438
 439         memset(&hints, 0, sizeof(hints));
 440         memset(&service, 0, sizeof(service));
 441         hints.ai_family = PF_UNSPEC;
 442         hints.ai_socktype = SOCK_DGRAM;
 443
 444         assert (host->node != NULL);
 445         service = (host->service != NULL) ? host->service : RIEMANN_PORT;
 446
 447         if ((e = getaddrinfo(host->node, service, &hints, &res)) != 0) {
 448                 ERROR ("riemann plugin: Unable to resolve host \"%s\": %s",
 449                         host->node, gai_strerror(e));
 450                 return -1;
 451         }
 452
 453         for (ai = res; ai != NULL; ai = ai->ai_next) {
 454                 /*
 455                  * check if another thread did not already succesfully connect
 456                  */
 457                 if (host->flags & F_CONNECT) {
 458                         freeaddrinfo(res);
 459                         return 0;
 460                 }
 461
 462                 if ((host->s = socket(ai->ai_family,
 463                                       ai->ai_socktype,
 464                                       ai->ai_protocol)) == -1) {
 465                         WARNING("riemann_connect: could not open socket");
 466                         freeaddrinfo(res);
 467                         return -1;
 468                 }
 469
 470                 if (connect(host->s, ai->ai_addr, ai->ai_addrlen) != 0) {
 471                         close(host->s);
 472                         host->flags |= ~F_CONNECT;
 473                         freeaddrinfo(res);
 474                         return -1;
 475                 }
 476                 host->flags |= F_CONNECT;
 477                 DEBUG("riemann plugin: got a succesful connection for: %s",
 478                                 host->node);
 479                 break;
 480         }
 481
 482         freeaddrinfo(res);
 483         if (ai == NULL) {
 484                 WARNING("riemann_connect: no suitable hosts found");
 485                 return -1;
 486         }
 487
 488         return 0;
 489 }
 490
 491 /* host->lock must be held when calling this function. */
 492 static int
 493 riemann_disconnect (struct riemann_host *host)
 494 {
 495         if ((host->flags & F_CONNECT) == 0)
 496                 return (0);
 497
 498         close (host->s);
 499         host->s = -1;
 500         host->flags &= ~F_CONNECT;
 501
 502         return (0);
 503 }
 504
 505 static void
 506 riemann_free(void *p)
 507 {
 508         struct riemann_host     *host = p;
 509
 510         if (host == NULL)
 511                 return;
 512
 513         pthread_mutex_lock (&host->lock);
 514
 515         host->reference_count--;
 516         if (host->reference_count > 0)
 517         {
 518                 pthread_mutex_unlock (&host->lock);
 519                 return;
 520         }
 521
 522         riemann_disconnect (host);
 523
 524         sfree(host->service);
 525         pthread_mutex_destroy (&host->lock);
 526         sfree(host);
 527 }
 528
 529 static int
 530 riemann_config_host(oconfig_item_t *ci)
 531 {
 532         struct riemann_host     *host = NULL;
 533         int                      status = 0;
 534         int                      i;
 535         oconfig_item_t          *child;
 536         char                     w_cb_name[DATA_MAX_NAME_LEN];
 537         char                     n_cb_name[DATA_MAX_NAME_LEN];
 538         user_data_t              ud;
 539
 540         if (ci->values_num != 1 ||
 541             ci->values[0].type != OCONFIG_TYPE_STRING) {
 542                 WARNING("riemann hosts need one string argument");
 543                 return -1;
 544         }
 545
 546         if ((host = calloc(1, sizeof (*host))) == NULL) {
 547                 WARNING("riemann host allocation failed");
 548                 return ENOMEM;
 549         }
 550         pthread_mutex_init (&host->lock, NULL);
 551         host->reference_count = 1;
 552         host->node = NULL;
 553         host->service = NULL;
 554         host->delay = RIEMANN_DELAY;
 555
 556         status = cf_util_get_string (ci, &host->node);
 557         if (status != 0) {
 558                 WARNING("riemann plugin: Required host name is missing.");
 559                 riemann_free (host);
 560                 return -1;
 561         }
 562
 563         for (i = 0; i < ci->children_num; i++) {
 564                 /*
 565                  * The code here could be simplified but makes room
 566                  * for easy adding of new options later on.
 567                  */
 568                 child = &ci->children[i];
 569                 status = 0;
 570
 571                 if (strcasecmp(child->key, "port") == 0) {
 572                         status = cf_util_get_service (child, &host->service);
 573                         if (status != 0) {
 574                                 ERROR ("riemann plugin: Invalid argument "
 575                                                 "configured for the \"Port\" "
 576                                                 "option.");
 577                                 break;
 578                         }
 579                 } else if (strcasecmp(child->key, "delay") == 0) {
 580                         if ((status = cf_util_get_int(ci, &host->delay)) != 0)
 581                                 break;
 582                 } else {
 583                         WARNING("riemann plugin: ignoring unknown config "
 584                                 "option: \"%s\"", child->key);
 585                 }
 586         }
 587         if (status != 0) {
 588                 riemann_free (host);
 589                 return status;
 590         }
 591
 592         ssnprintf(w_cb_name, sizeof(w_cb_name), "write-riemann/%s:%s",
 593                   host->node,
 594                   (host->service != NULL) ? host->service : RIEMANN_PORT);
 595         ssnprintf(n_cb_name, sizeof(n_cb_name), "notification-riemann/%s:%s",
 596                   host->node,
 597                   (host->service != NULL) ? host->service : RIEMANN_PORT);
 598         DEBUG("riemann w_cb_name: %s", w_cb_name);
 599         DEBUG("riemann n_cb_name: %s", n_cb_name);
 600         ud.data = host;
 601         ud.free_func = riemann_free;
 602
 603         pthread_mutex_lock (&host->lock);
 604
 605         status = plugin_register_write (w_cb_name, riemann_write, &ud);
 606         if (status != 0)
 607                 WARNING ("riemann plugin: plugin_register_write (\"%s\") "
 608                                 "failed with status %i.",
 609                                 w_cb_name, status);
 610         else /* success */
 611                 host->reference_count++;
 612
 613         status = plugin_register_notification (n_cb_name,
 614                         riemann_notification, &ud);
 615         if (status != 0)
 616                 WARNING ("riemann plugin: plugin_register_notification (\"%s\") "
 617                                 "failed with status %i.",
 618                                 n_cb_name, status);
 619         else /* success */
 620                 host->reference_count++;
 621
 622         if (host->reference_count <= 1)
 623         {
 624                 /* Both callbacks failed => free memory.
 625                  * We need to unlock here, because riemann_free() will lock.
 626                  * This is not a race condition, because we're the only one
 627                  * holding a reference. */
 628                 pthread_mutex_unlock (&host->lock);
 629                 riemann_free (host);
 630                 return (-1);
 631         }
 632
 633         host->reference_count--;
 634         pthread_mutex_unlock (&host->lock);
 635
 636         return status;
 637 }
 638
 639 static int
 640 riemann_config(oconfig_item_t *ci)
 641 {
 642         int              i;
 643         char            *newtag;
 644         oconfig_item_t  *child;
 645
 646         for (i = 0; i < ci->children_num; i++)  {
 647                 child = &ci->children[i];
 648
 649                 if (strcasecmp(child->key, "host") == 0) {
 650                         riemann_config_host(child);
 651                 } else if (strcasecmp(child->key, "tag") == 0) {
 652                         if (riemann_tagcount >= RIEMANN_EXTRA_TAGS) {
 653                                 WARNING("riemann plugin: too many tags");
 654                                 return -1;
 655                         }
 656                         newtag = NULL;
 657                         cf_util_get_string(child, &newtag);
 658                         if (newtag == NULL)
 659                                 return -1;
 660                         riemann_tags[riemann_tagcount++] = newtag;
 661                         DEBUG("riemann_config: got tag: %s", newtag);
 662
 663                 } else {
 664                         WARNING ("riemann plugin: Ignoring unknown "
 665                                  "configuration option \"%s\" at top level.",
 666                                  child->key);
 667                 }
 668         }
 669         return (0);
 670 }
 671
 672 void
 673 module_register(void)
 674 {
 675         DEBUG("riemann: module_register");
 676
 677         plugin_register_complex_config ("riemann", riemann_config);
 678 }
 679
 680 /* vim: set sw=8 sts=8 ts=8 noet : */