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