Add missing kstat definitions
[collectd.git] / src / write_riemann.c
1 /**
2  * collectd - src/write_riemann.c
3  * Copyright (C) 2012,2013  Pierre-Yves Ritschard
4  * Copyright (C) 2013       Florian octo Forster
5  *
6  * Permission is hereby granted, free of charge, to any person obtaining a
7  * copy of this software and associated documentation files (the "Software"),
8  * to deal in the Software without restriction, including without limitation
9  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
10  * and/or sell copies of the Software, and to permit persons to whom the
11  * Software is furnished to do so, subject to the following conditions:
12  *
13  * The above copyright notice and this permission notice shall be included in
14  * all copies or substantial portions of the Software.
15  *
16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
21  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
22  * DEALINGS IN THE SOFTWARE.
23  *
24  * Authors:
25  *   Pierre-Yves Ritschard <pyr at spootnik.org>
26  *   Florian octo Forster <octo at collectd.org>
27  */
28
29 #include "collectd.h"
30 #include "plugin.h"
31 #include "common.h"
32 #include "configfile.h"
33 #include "utils_cache.h"
34 #include "riemann.pb-c.h"
35
36 #include <sys/socket.h>
37 #include <arpa/inet.h>
38 #include <errno.h>
39 #include <netdb.h>
40 #include <inttypes.h>
41 #include <pthread.h>
42
43 #define RIEMANN_HOST            "localhost"
44 #define RIEMANN_PORT            "5555"
45 #define RIEMANN_TTL_FACTOR      2.0
46 #define RIEMANN_BATCH_MAX      8192
47
48 int write_riemann_threshold_check(const data_set_t *, const value_list_t *, int *);
49
50 struct riemann_host {
51         char                    *name;
52         char                    *event_service_prefix;
53 #define F_CONNECT        0x01
54         uint8_t                  flags;
55         pthread_mutex_t  lock;
56     _Bool            batch_mode;
57         _Bool            notifications;
58         _Bool            check_thresholds;
59         _Bool                    store_rates;
60         _Bool                    always_append_ds;
61         char                    *node;
62         char                    *service;
63         _Bool                    use_tcp;
64         int                          s;
65         double                   ttl_factor;
66     Msg             *batch_msg;
67     cdtime_t         batch_init;
68     int              batch_max;
69         int                          reference_count;
70 };
71
72 static char     **riemann_tags;
73 static size_t     riemann_tags_num;
74 static char     **riemann_attrs;
75 static size_t     riemann_attrs_num;
76
77 static void riemann_event_protobuf_free (Event *event) /* {{{ */
78 {
79         size_t i;
80
81         if (event == NULL)
82                 return;
83
84         sfree (event->state);
85         sfree (event->service);
86         sfree (event->host);
87         sfree (event->description);
88
89         strarray_free (event->tags, event->n_tags);
90         event->tags = NULL;
91         event->n_tags = 0;
92
93         for (i = 0; i < event->n_attributes; i++)
94         {
95                 sfree (event->attributes[i]->key);
96                 sfree (event->attributes[i]->value);
97                 sfree (event->attributes[i]);
98         }
99         sfree (event->attributes);
100         event->n_attributes = 0;
101
102         sfree (event);
103 } /* }}} void riemann_event_protobuf_free */
104
105 static void riemann_msg_protobuf_free(Msg *msg) /* {{{ */
106 {
107         size_t i;
108
109         if (msg == NULL)
110                 return;
111
112         for (i = 0; i < msg->n_events; i++)
113         {
114                 riemann_event_protobuf_free (msg->events[i]);
115                 msg->events[i] = NULL;
116         }
117
118         sfree (msg->events);
119         msg->n_events = 0;
120
121         sfree (msg);
122 } /* }}} void riemann_msg_protobuf_free */
123
124 /* host->lock must be held when calling this function. */
125 static int riemann_connect(struct riemann_host *host) /* {{{ */
126 {
127         int                      e;
128         struct addrinfo         *ai, *res, hints;
129         char const              *node;
130         char const              *service;
131
132         if (host->flags & F_CONNECT)
133                 return 0;
134
135         memset(&hints, 0, sizeof(hints));
136         memset(&service, 0, sizeof(service));
137         hints.ai_family = AF_UNSPEC;
138         hints.ai_socktype = host->use_tcp ? SOCK_STREAM : SOCK_DGRAM;
139 #ifdef AI_ADDRCONFIG
140         hints.ai_flags |= AI_ADDRCONFIG;
141 #endif
142
143         node = (host->node != NULL) ? host->node : RIEMANN_HOST;
144         service = (host->service != NULL) ? host->service : RIEMANN_PORT;
145
146         if ((e = getaddrinfo(node, service, &hints, &res)) != 0) {
147                 ERROR ("write_riemann plugin: Unable to resolve host \"%s\": %s",
148                         node, gai_strerror(e));
149                 return -1;
150         }
151
152         host->s = -1;
153         for (ai = res; ai != NULL; ai = ai->ai_next) {
154                 if ((host->s = socket(ai->ai_family,
155                                       ai->ai_socktype,
156                                       ai->ai_protocol)) == -1) {
157                         continue;
158                 }
159
160                 if (connect(host->s, ai->ai_addr, ai->ai_addrlen) != 0) {
161                         close(host->s);
162                         host->s = -1;
163                         continue;
164                 }
165
166                 host->flags |= F_CONNECT;
167                 DEBUG("write_riemann plugin: got a successful connection for: %s:%s",
168                                 node, service);
169                 break;
170         }
171
172         freeaddrinfo(res);
173
174         if (host->s < 0) {
175                 WARNING("write_riemann plugin: Unable to connect to Riemann at %s:%s",
176                                 node, service);
177                 return -1;
178         }
179         return 0;
180 } /* }}} int riemann_connect */
181
182 /* host->lock must be held when calling this function. */
183 static int riemann_disconnect (struct riemann_host *host) /* {{{ */
184 {
185         if ((host->flags & F_CONNECT) == 0)
186                 return (0);
187
188         close (host->s);
189         host->s = -1;
190         host->flags &= ~F_CONNECT;
191
192         return (0);
193 } /* }}} int riemann_disconnect */
194
195 static int riemann_send_msg (struct riemann_host *host, const Msg *msg) /* {{{ */
196 {
197         int status = 0;
198         u_char *buffer = NULL;
199         size_t  buffer_len;
200
201         status = riemann_connect (host);
202         if (status != 0)
203                 return status;
204
205         buffer_len = msg__get_packed_size(msg);
206
207         if (host->use_tcp)
208                 buffer_len += 4;
209
210         buffer = malloc (buffer_len);
211         if (buffer == NULL) {
212                 ERROR ("write_riemann plugin: malloc failed.");
213                 return ENOMEM;
214         }
215         memset (buffer, 0, buffer_len);
216
217         if (host->use_tcp)
218         {
219                 uint32_t length = htonl ((uint32_t) (buffer_len - 4));
220                 memcpy (buffer, &length, 4);
221                 msg__pack(msg, buffer + 4);
222         }
223         else
224         {
225                 msg__pack(msg, buffer);
226         }
227
228         status = (int) swrite (host->s, buffer, buffer_len);
229         if (status != 0)
230         {
231                 char errbuf[1024];
232                 ERROR ("write_riemann plugin: Sending to Riemann at %s:%s failed: %s",
233                                 (host->node != NULL) ? host->node : RIEMANN_HOST,
234                                 (host->service != NULL) ? host->service : RIEMANN_PORT,
235                                 sstrerror (errno, errbuf, sizeof (errbuf)));
236                 sfree (buffer);
237                 return -1;
238         }
239
240         sfree (buffer);
241         return 0;
242 } /* }}} int riemann_send_msg */
243
244 static int riemann_recv_ack(struct riemann_host *host) /* {{{ */
245 {
246         int status = 0;
247         Msg *msg = NULL;
248         uint32_t header;
249
250         status = (int) sread (host->s, &header, 4);
251
252         if (status != 0)
253                 return -1;
254
255         size_t size = ntohl(header);
256
257         // Buffer on the stack since acknowledges are typically small.
258         u_char buffer[size];
259         memset (buffer, 0, size);
260
261         status = (int) sread (host->s, buffer, size);
262
263         if (status != 0)
264                 return status;
265
266         msg = msg__unpack (NULL, size, buffer);
267
268         if (msg == NULL)
269                 return -1;
270
271         if (!msg->ok)
272         {
273                 ERROR ("write_riemann plugin: Sending to Riemann at %s:%s acknowledgement message reported error: %s",
274                                 (host->node != NULL) ? host->node : RIEMANN_HOST,
275                                 (host->service != NULL) ? host->service : RIEMANN_PORT,
276                                 msg->error);
277
278                 msg__free_unpacked(msg, NULL);
279                 return -1;
280         }
281
282         msg__free_unpacked (msg, NULL);
283         return 0;
284 } /* }}} int riemann_recv_ack */
285
286 /**
287  * Function to send messages (Msg) to riemann.
288  *
289  * Acquires the host lock, disconnects on errors.
290  */
291 static int riemann_send(struct riemann_host *host, Msg const *msg) /* {{{ */
292 {
293         int status = 0;
294         pthread_mutex_lock (&host->lock);
295
296         status = riemann_send_msg(host, msg);
297         if (status != 0) {
298                 riemann_disconnect (host);
299                 pthread_mutex_unlock (&host->lock);
300                 return status;
301         }
302
303         /*
304          * For TCP we need to receive message acknowledgemenent.
305          */
306         if (host->use_tcp)
307         {
308                 status = riemann_recv_ack(host);
309
310                 if (status != 0)
311                 {
312                         riemann_disconnect (host);
313                         pthread_mutex_unlock (&host->lock);
314                         return status;
315                 }
316         }
317
318         pthread_mutex_unlock (&host->lock);
319         return 0;
320 } /* }}} int riemann_send */
321
322 static int riemann_event_add_tag (Event *event, char const *tag) /* {{{ */
323 {
324         return (strarray_add (&event->tags, &event->n_tags, tag));
325 } /* }}} int riemann_event_add_tag */
326
327 static int riemann_event_add_attribute(Event *event, /* {{{ */
328                 char const *key, char const *value)
329 {
330         Attribute **new_attributes;
331         Attribute *a;
332
333         new_attributes = realloc (event->attributes,
334                         sizeof (*event->attributes) * (event->n_attributes + 1));
335         if (new_attributes == NULL)
336         {
337                 ERROR ("write_riemann plugin: realloc failed.");
338                 return (ENOMEM);
339         }
340         event->attributes = new_attributes;
341
342         a = malloc (sizeof (*a));
343         if (a == NULL)
344         {
345                 ERROR ("write_riemann plugin: malloc failed.");
346                 return (ENOMEM);
347         }
348         attribute__init (a);
349
350         a->key = strdup (key);
351         if (value != NULL)
352                 a->value = strdup (value);
353
354         event->attributes[event->n_attributes] = a;
355         event->n_attributes++;
356
357         return (0);
358 } /* }}} int riemann_event_add_attribute */
359
360 static Msg *riemann_notification_to_protobuf(struct riemann_host *host, /* {{{ */
361                 notification_t const *n)
362 {
363         Msg *msg;
364         Event *event;
365         char service_buffer[6 * DATA_MAX_NAME_LEN];
366         char const *severity;
367         notification_meta_t *meta;
368         size_t i;
369
370         msg = malloc (sizeof (*msg));
371         if (msg == NULL)
372         {
373                 ERROR ("write_riemann plugin: malloc failed.");
374                 return (NULL);
375         }
376         memset (msg, 0, sizeof (*msg));
377         msg__init (msg);
378
379         msg->events = malloc (sizeof (*msg->events));
380         if (msg->events == NULL)
381         {
382                 ERROR ("write_riemann plugin: malloc failed.");
383                 sfree (msg);
384                 return (NULL);
385         }
386
387         event = malloc (sizeof (*event));
388         if (event == NULL)
389         {
390                 ERROR ("write_riemann plugin: malloc failed.");
391                 sfree (msg->events);
392                 sfree (msg);
393                 return (NULL);
394         }
395         memset (event, 0, sizeof (*event));
396         event__init (event);
397
398         msg->events[0] = event;
399         msg->n_events = 1;
400
401         event->host = strdup (n->host);
402         event->time = CDTIME_T_TO_TIME_T (n->time);
403         event->has_time = 1;
404
405         switch (n->severity)
406         {
407                 case NOTIF_OKAY:        severity = "ok"; break;
408                 case NOTIF_WARNING:     severity = "warning"; break;
409                 case NOTIF_FAILURE:     severity = "critical"; break;
410                 default:                severity = "unknown";
411         }
412         event->state = strdup (severity);
413
414         riemann_event_add_tag (event, "notification");
415         if (n->host[0] != 0)
416                 riemann_event_add_attribute (event, "host", n->host);
417         if (n->plugin[0] != 0)
418                 riemann_event_add_attribute (event, "plugin", n->plugin);
419         if (n->plugin_instance[0] != 0)
420                 riemann_event_add_attribute (event, "plugin_instance",
421                                 n->plugin_instance);
422
423         if (n->type[0] != 0)
424                 riemann_event_add_attribute (event, "type", n->type);
425         if (n->type_instance[0] != 0)
426                 riemann_event_add_attribute (event, "type_instance",
427                                 n->type_instance);
428
429         for (i = 0; i < riemann_attrs_num; i += 2)
430                 riemann_event_add_attribute(event,
431                                             riemann_attrs[i],
432                                             riemann_attrs[i +1]);
433
434         for (i = 0; i < riemann_tags_num; i++)
435                 riemann_event_add_tag (event, riemann_tags[i]);
436
437         format_name (service_buffer, sizeof (service_buffer),
438                         /* host = */ "", n->plugin, n->plugin_instance,
439                         n->type, n->type_instance);
440         event->service = strdup (&service_buffer[1]);
441
442         if (n->message[0] != 0)
443                 riemann_event_add_attribute (event, "description", n->message);
444
445         /* Pull in values from threshold and add extra attributes */
446         for (meta = n->meta; meta != NULL; meta = meta->next)
447         {
448                 if (strcasecmp ("CurrentValue", meta->name) == 0 && meta->type == NM_TYPE_DOUBLE)
449                 {
450                         event->metric_d = meta->nm_value.nm_double;
451                         event->has_metric_d = 1;
452                         continue;
453                 }
454
455                 if (meta->type == NM_TYPE_STRING) {
456                         riemann_event_add_attribute (event, meta->name, meta->nm_value.nm_string);
457                         continue;
458                 }
459         }
460
461         DEBUG ("write_riemann plugin: Successfully created protobuf for notification: "
462                         "host = \"%s\", service = \"%s\", state = \"%s\"",
463                         event->host, event->service, event->state);
464         return (msg);
465 } /* }}} Msg *riemann_notification_to_protobuf */
466
467 static Event *riemann_value_to_protobuf(struct riemann_host const *host, /* {{{ */
468                 data_set_t const *ds,
469                 value_list_t const *vl, size_t index,
470                                          gauge_t const *rates,
471                                          int status)
472 {
473         Event *event;
474         char name_buffer[5 * DATA_MAX_NAME_LEN];
475         char service_buffer[6 * DATA_MAX_NAME_LEN];
476         double ttl;
477         size_t i;
478
479         event = malloc (sizeof (*event));
480         if (event == NULL)
481         {
482                 ERROR ("write_riemann plugin: malloc failed.");
483                 return (NULL);
484         }
485         memset (event, 0, sizeof (*event));
486         event__init (event);
487
488         event->host = strdup (vl->host);
489         event->time = CDTIME_T_TO_TIME_T (vl->time);
490         event->has_time = 1;
491
492         if (host->check_thresholds) {
493                 switch (status) {
494                         case STATE_OKAY:
495                                 event->state = strdup("ok");
496                                 break;
497                         case STATE_ERROR:
498                                 event->state = strdup("critical");
499                                 break;
500                         case STATE_WARNING:
501                                 event->state = strdup("warning");
502                                 break;
503                         case STATE_MISSING:
504                                 event->state = strdup("unknown");
505                                 break;
506                 }
507         }
508
509         ttl = CDTIME_T_TO_DOUBLE (vl->interval) * host->ttl_factor;
510         event->ttl = (float) ttl;
511         event->has_ttl = 1;
512
513         riemann_event_add_attribute (event, "plugin", vl->plugin);
514         if (vl->plugin_instance[0] != 0)
515                 riemann_event_add_attribute (event, "plugin_instance",
516                                 vl->plugin_instance);
517
518         riemann_event_add_attribute (event, "type", vl->type);
519         if (vl->type_instance[0] != 0)
520                 riemann_event_add_attribute (event, "type_instance",
521                                 vl->type_instance);
522
523         if ((ds->ds[index].type != DS_TYPE_GAUGE) && (rates != NULL))
524         {
525                 char ds_type[DATA_MAX_NAME_LEN];
526
527                 ssnprintf (ds_type, sizeof (ds_type), "%s:rate",
528                                 DS_TYPE_TO_STRING(ds->ds[index].type));
529                 riemann_event_add_attribute (event, "ds_type", ds_type);
530         }
531         else
532         {
533                 riemann_event_add_attribute (event, "ds_type",
534                                 DS_TYPE_TO_STRING(ds->ds[index].type));
535         }
536         riemann_event_add_attribute (event, "ds_name", ds->ds[index].name);
537         {
538                 char ds_index[DATA_MAX_NAME_LEN];
539
540                 ssnprintf (ds_index, sizeof (ds_index), "%zu", index);
541                 riemann_event_add_attribute (event, "ds_index", ds_index);
542         }
543
544         for (i = 0; i < riemann_attrs_num; i += 2)
545                 riemann_event_add_attribute(event,
546                                             riemann_attrs[i],
547                                             riemann_attrs[i +1]);
548
549         for (i = 0; i < riemann_tags_num; i++)
550                 riemann_event_add_tag (event, riemann_tags[i]);
551
552         if (ds->ds[index].type == DS_TYPE_GAUGE)
553         {
554                 event->has_metric_d = 1;
555                 event->metric_d = (double) vl->values[index].gauge;
556         }
557         else if (rates != NULL)
558         {
559                 event->has_metric_d = 1;
560                 event->metric_d = (double) rates[index];
561         }
562         else
563         {
564                 event->has_metric_sint64 = 1;
565                 if (ds->ds[index].type == DS_TYPE_DERIVE)
566                         event->metric_sint64 = (int64_t) vl->values[index].derive;
567                 else if (ds->ds[index].type == DS_TYPE_ABSOLUTE)
568                         event->metric_sint64 = (int64_t) vl->values[index].absolute;
569                 else
570                         event->metric_sint64 = (int64_t) vl->values[index].counter;
571         }
572
573         format_name (name_buffer, sizeof (name_buffer),
574                         /* host = */ "", vl->plugin, vl->plugin_instance,
575                         vl->type, vl->type_instance);
576         if (host->always_append_ds || (ds->ds_num > 1))
577         {
578                 if (host->event_service_prefix == NULL)
579                         ssnprintf (service_buffer, sizeof (service_buffer), "%s/%s",
580                                         &name_buffer[1], ds->ds[index].name);
581                 else
582                         ssnprintf (service_buffer, sizeof (service_buffer), "%s%s/%s",
583                                         host->event_service_prefix, &name_buffer[1], ds->ds[index].name);
584         }
585         else
586         {
587                 if (host->event_service_prefix == NULL)
588                         sstrncpy (service_buffer, &name_buffer[1], sizeof (service_buffer));
589                 else
590                         ssnprintf (service_buffer, sizeof (service_buffer), "%s%s",
591                                         host->event_service_prefix, &name_buffer[1]);
592         }
593
594         event->service = strdup (service_buffer);
595
596         DEBUG ("write_riemann plugin: Successfully created protobuf for metric: "
597                         "host = \"%s\", service = \"%s\"",
598                         event->host, event->service);
599         return (event);
600 } /* }}} Event *riemann_value_to_protobuf */
601
602 static Msg *riemann_value_list_to_protobuf (struct riemann_host const *host, /* {{{ */
603                                             data_set_t const *ds,
604                                             value_list_t const *vl,
605                                             int *statuses)
606 {
607         Msg *msg;
608         size_t i;
609         gauge_t *rates = NULL;
610
611         /* Initialize the Msg structure. */
612         msg = malloc (sizeof (*msg));
613         if (msg == NULL)
614         {
615                 ERROR ("write_riemann plugin: malloc failed.");
616                 return (NULL);
617         }
618         memset (msg, 0, sizeof (*msg));
619         msg__init (msg);
620
621         /* Set up events. First, the list of pointers. */
622         msg->n_events = vl->values_len;
623         msg->events = calloc (msg->n_events, sizeof (*msg->events));
624         if (msg->events == NULL)
625         {
626                 ERROR ("write_riemann plugin: calloc failed.");
627                 riemann_msg_protobuf_free (msg);
628                 return (NULL);
629         }
630
631         if (host->store_rates)
632         {
633                 rates = uc_get_rate (ds, vl);
634                 if (rates == NULL)
635                 {
636                         ERROR ("write_riemann plugin: uc_get_rate failed.");
637                         riemann_msg_protobuf_free (msg);
638                         return (NULL);
639                 }
640         }
641
642         for (i = 0; i < msg->n_events; i++)
643         {
644                 msg->events[i] = riemann_value_to_protobuf (host, ds, vl,
645                                                             (int) i, rates, statuses[i]);
646                 if (msg->events[i] == NULL)
647                 {
648                         riemann_msg_protobuf_free (msg);
649                         sfree (rates);
650                         return (NULL);
651                 }
652         }
653
654         sfree (rates);
655         return (msg);
656 } /* }}} Msg *riemann_value_list_to_protobuf */
657
658
659 /*
660  * Always call while holding host->lock !
661  */
662 static int riemann_batch_flush_nolock (cdtime_t timeout,
663                                        struct riemann_host *host)
664 {
665     cdtime_t    now;
666     int         status = 0;
667
668     if (timeout > 0) {
669         now = cdtime ();
670         if ((host->batch_init + timeout) > now)
671             return status;
672     }
673     riemann_send_msg(host, host->batch_msg);
674     riemann_msg_protobuf_free(host->batch_msg);
675
676         if (host->use_tcp && ((status = riemann_recv_ack(host)) != 0))
677         riemann_disconnect (host);
678
679     host->batch_init = cdtime();
680     host->batch_msg = NULL;
681     return status;
682 }
683
684 static int riemann_batch_flush (cdtime_t timeout,
685         const char *identifier __attribute__((unused)),
686         user_data_t *user_data)
687 {
688     struct riemann_host *host;
689     int status;
690
691     if (user_data == NULL)
692         return (-EINVAL);
693
694     host = user_data->data;
695     pthread_mutex_lock (&host->lock);
696     status = riemann_batch_flush_nolock (timeout, host);
697     if (status != 0)
698         ERROR ("write_riemann plugin: riemann_send failed with status %i",
699                status);
700
701     pthread_mutex_unlock(&host->lock);
702     return status;
703 }
704
705 static int riemann_batch_add_value_list (struct riemann_host *host, /* {{{ */
706                                          data_set_t const *ds,
707                                          value_list_t const *vl,
708                                          int *statuses)
709 {
710         size_t i;
711     Event **events;
712     Msg *msg;
713     size_t len;
714     int ret;
715
716     msg = riemann_value_list_to_protobuf (host, ds, vl, statuses);
717     if (msg == NULL)
718         return -1;
719
720     pthread_mutex_lock(&host->lock);
721
722     if (host->batch_msg == NULL) {
723         host->batch_msg = msg;
724     } else {
725         len = msg->n_events + host->batch_msg->n_events;
726         events = realloc(host->batch_msg->events,
727                          (len * sizeof(*host->batch_msg->events)));
728         if (events == NULL) {
729             pthread_mutex_unlock(&host->lock);
730             ERROR ("write_riemann plugin: out of memory");
731             riemann_msg_protobuf_free (msg);
732             return -1;
733         }
734         host->batch_msg->events = events;
735
736         for (i = host->batch_msg->n_events; i < len; i++)
737             host->batch_msg->events[i] = msg->events[i - host->batch_msg->n_events];
738
739         host->batch_msg->n_events = len;
740         sfree (msg->events);
741         msg->n_events = 0;
742         sfree (msg);
743     }
744
745         len = msg__get_packed_size(host->batch_msg);
746     ret = 0;
747     if ((host->batch_max < 0) || (((size_t) host->batch_max) <= len)) {
748             ret = riemann_batch_flush_nolock(0, host);
749     }
750
751     pthread_mutex_unlock(&host->lock);
752     return ret;
753 } /* }}} Msg *riemann_batch_add_value_list */
754
755 static int riemann_notification(const notification_t *n, user_data_t *ud) /* {{{ */
756 {
757         int                      status;
758         struct riemann_host     *host = ud->data;
759         Msg                     *msg;
760
761         if (!host->notifications)
762                 return 0;
763
764     /*
765      * Never batch for notifications, send them ASAP
766      */
767         msg = riemann_notification_to_protobuf (host, n);
768         if (msg == NULL)
769                 return (-1);
770
771         status = riemann_send (host, msg);
772         if (status != 0)
773                 ERROR ("write_riemann plugin: riemann_send failed with status %i",
774                                 status);
775
776         riemann_msg_protobuf_free (msg);
777         return (status);
778 } /* }}} int riemann_notification */
779
780 static int riemann_write(const data_set_t *ds, /* {{{ */
781                 const value_list_t *vl,
782                 user_data_t *ud)
783 {
784         int                      status = 0;
785         int                      statuses[vl->values_len];
786         struct riemann_host     *host = ud->data;
787
788         if (host->check_thresholds) {
789                 status = write_riemann_threshold_check(ds, vl, statuses);
790                 if (status != 0)
791                         return status;
792         } else {
793                 memset (statuses, 0, sizeof (statuses));
794         }
795
796         if (host->use_tcp == 1 && host->batch_mode) {
797                 riemann_batch_add_value_list (host, ds, vl, statuses);
798         } else {
799                 Msg *msg = riemann_value_list_to_protobuf (host, ds, vl, statuses);
800                 if (msg == NULL)
801                         return (-1);
802
803                 status = riemann_send (host, msg);
804                 if (status != 0)
805                         ERROR ("write_riemann plugin: riemann_send failed with status %i", status);
806
807                 riemann_msg_protobuf_free (msg);
808         }
809
810         return status;
811 } /* }}} int riemann_write */
812
813 static void riemann_free(void *p) /* {{{ */
814 {
815         struct riemann_host     *host = p;
816
817         if (host == NULL)
818                 return;
819
820         pthread_mutex_lock (&host->lock);
821
822         host->reference_count--;
823         if (host->reference_count > 0)
824         {
825                 pthread_mutex_unlock (&host->lock);
826                 return;
827         }
828
829         riemann_disconnect (host);
830
831         sfree(host->service);
832         pthread_mutex_destroy (&host->lock);
833         sfree(host);
834 } /* }}} void riemann_free */
835
836 static int riemann_config_node(oconfig_item_t *ci) /* {{{ */
837 {
838         struct riemann_host     *host = NULL;
839         int                      status = 0;
840         int                      i;
841         oconfig_item_t          *child;
842         char                     callback_name[DATA_MAX_NAME_LEN];
843         user_data_t              ud;
844
845         if ((host = calloc(1, sizeof (*host))) == NULL) {
846                 ERROR ("write_riemann plugin: calloc failed.");
847                 return ENOMEM;
848         }
849         pthread_mutex_init (&host->lock, NULL);
850         host->reference_count = 1;
851         host->node = NULL;
852         host->service = NULL;
853         host->notifications = 1;
854         host->check_thresholds = 0;
855         host->store_rates = 1;
856         host->always_append_ds = 0;
857         host->use_tcp = 1;
858         host->batch_mode = 1;
859         host->batch_max = RIEMANN_BATCH_MAX; /* typical MSS */
860         host->batch_init = cdtime();
861         host->ttl_factor = RIEMANN_TTL_FACTOR;
862
863         status = cf_util_get_string (ci, &host->name);
864         if (status != 0) {
865                 WARNING("write_riemann plugin: Required host name is missing.");
866                 riemann_free (host);
867                 return -1;
868         }
869
870         for (i = 0; i < ci->children_num; i++) {
871                 /*
872                  * The code here could be simplified but makes room
873                  * for easy adding of new options later on.
874                  */
875                 child = &ci->children[i];
876                 status = 0;
877
878                 if (strcasecmp ("Host", child->key) == 0) {
879                         status = cf_util_get_string (child, &host->node);
880                         if (status != 0)
881                                 break;
882                 } else if (strcasecmp ("Notifications", child->key) == 0) {
883                         status = cf_util_get_boolean(child, &host->notifications);
884                         if (status != 0)
885                                 break;
886                 } else if (strcasecmp ("EventServicePrefix", child->key) == 0) {
887                         status = cf_util_get_string (child, &host->event_service_prefix);
888                         if (status != 0)
889                                 break;
890                 } else if (strcasecmp ("CheckThresholds", child->key) == 0) {
891                         status = cf_util_get_boolean(child, &host->check_thresholds);
892                         if (status != 0)
893                                 break;
894         } else if (strcasecmp ("Batch", child->key) == 0) {
895             status = cf_util_get_boolean(child, &host->batch_mode);
896             if (status != 0)
897                 break;
898         } else if (strcasecmp("BatchMaxSize", child->key) == 0) {
899             status = cf_util_get_int(child, &host->batch_max);
900             if (status != 0)
901                 break;
902                 } else if (strcasecmp ("Port", child->key) == 0) {
903                         status = cf_util_get_service (child, &host->service);
904                         if (status != 0) {
905                                 ERROR ("write_riemann plugin: Invalid argument "
906                                                 "configured for the \"Port\" "
907                                                 "option.");
908                                 break;
909                         }
910                 } else if (strcasecmp ("Protocol", child->key) == 0) {
911                         char tmp[16];
912                         status = cf_util_get_string_buffer (child,
913                                         tmp, sizeof (tmp));
914                         if (status != 0)
915                         {
916                                 ERROR ("write_riemann plugin: cf_util_get_"
917                                                 "string_buffer failed with "
918                                                 "status %i.", status);
919                                 break;
920                         }
921
922                         if (strcasecmp ("UDP", tmp) == 0)
923                                 host->use_tcp = 0;
924                         else if (strcasecmp ("TCP", tmp) == 0)
925                                 host->use_tcp = 1;
926                         else
927                                 WARNING ("write_riemann plugin: The value "
928                                                 "\"%s\" is not valid for the "
929                                                 "\"Protocol\" option. Use "
930                                                 "either \"UDP\" or \"TCP\".",
931                                                 tmp);
932                 } else if (strcasecmp ("StoreRates", child->key) == 0) {
933                         status = cf_util_get_boolean (child, &host->store_rates);
934                         if (status != 0)
935                                 break;
936                 } else if (strcasecmp ("AlwaysAppendDS", child->key) == 0) {
937                         status = cf_util_get_boolean (child,
938                                         &host->always_append_ds);
939                         if (status != 0)
940                                 break;
941                 } else if (strcasecmp ("TTLFactor", child->key) == 0) {
942                         double tmp = NAN;
943                         status = cf_util_get_double (child, &tmp);
944                         if (status != 0)
945                                 break;
946                         if (tmp >= 2.0) {
947                                 host->ttl_factor = tmp;
948                         } else if (tmp >= 1.0) {
949                                 NOTICE ("write_riemann plugin: The configured "
950                                                 "TTLFactor is very small "
951                                                 "(%.1f). A value of 2.0 or "
952                                                 "greater is recommended.",
953                                                 tmp);
954                                 host->ttl_factor = tmp;
955                         } else if (tmp > 0.0) {
956                                 WARNING ("write_riemann plugin: The configured "
957                                                 "TTLFactor is too small to be "
958                                                 "useful (%.1f). I'll use it "
959                                                 "since the user knows best, "
960                                                 "but under protest.",
961                                                 tmp);
962                                 host->ttl_factor = tmp;
963                         } else { /* zero, negative and NAN */
964                                 ERROR ("write_riemann plugin: The configured "
965                                                 "TTLFactor is invalid (%.1f).",
966                                                 tmp);
967                         }
968                 } else {
969                         WARNING("write_riemann plugin: ignoring unknown config "
970                                 "option: \"%s\"", child->key);
971                 }
972         }
973         if (status != 0) {
974                 riemann_free (host);
975                 return status;
976         }
977
978         ssnprintf (callback_name, sizeof (callback_name), "write_riemann/%s",
979                         host->name);
980         ud.data = host;
981         ud.free_func = riemann_free;
982
983         pthread_mutex_lock (&host->lock);
984
985         status = plugin_register_write (callback_name, riemann_write, &ud);
986
987     if (host->use_tcp == 1 && host->batch_mode) {
988         ud.free_func = NULL;
989         plugin_register_flush(callback_name, riemann_batch_flush, &ud);
990     }
991         if (status != 0)
992                 WARNING ("write_riemann plugin: plugin_register_write (\"%s\") "
993                                 "failed with status %i.",
994                                 callback_name, status);
995         else /* success */
996                 host->reference_count++;
997
998         status = plugin_register_notification (callback_name,
999                         riemann_notification, &ud);
1000         if (status != 0)
1001                 WARNING ("write_riemann plugin: plugin_register_notification (\"%s\") "
1002                                 "failed with status %i.",
1003                                 callback_name, status);
1004         else /* success */
1005                 host->reference_count++;
1006
1007         if (host->reference_count <= 1)
1008         {
1009                 /* Both callbacks failed => free memory.
1010                  * We need to unlock here, because riemann_free() will lock.
1011                  * This is not a race condition, because we're the only one
1012                  * holding a reference. */
1013                 pthread_mutex_unlock (&host->lock);
1014                 riemann_free (host);
1015                 return (-1);
1016         }
1017
1018         host->reference_count--;
1019         pthread_mutex_unlock (&host->lock);
1020
1021         return status;
1022 } /* }}} int riemann_config_node */
1023
1024 static int riemann_config(oconfig_item_t *ci) /* {{{ */
1025 {
1026         int              i;
1027         oconfig_item_t  *child;
1028         int              status;
1029
1030         for (i = 0; i < ci->children_num; i++)  {
1031                 child = &ci->children[i];
1032
1033                 if (strcasecmp("Node", child->key) == 0) {
1034                         riemann_config_node (child);
1035                 } else if (strcasecmp(child->key, "attribute") == 0) {
1036                         char *key = NULL;
1037                         char *val = NULL;
1038
1039                         if (child->values_num != 2) {
1040                                 WARNING("riemann attributes need both a key and a value.");
1041                                 return (-1);
1042                         }
1043                         if (child->values[0].type != OCONFIG_TYPE_STRING ||
1044                             child->values[1].type != OCONFIG_TYPE_STRING) {
1045                                 WARNING("riemann attribute needs string arguments.");
1046                                 return (-1);
1047                         }
1048                         if ((key = strdup(child->values[0].value.string)) == NULL) {
1049                                 WARNING("cannot allocate memory for attribute key.");
1050                                 return (-1);
1051                         }
1052                         if ((val = strdup(child->values[1].value.string)) == NULL) {
1053                                 WARNING("cannot allocate memory for attribute value.");
1054                                 return (-1);
1055                         }
1056                         strarray_add(&riemann_attrs, &riemann_attrs_num, key);
1057                         strarray_add(&riemann_attrs, &riemann_attrs_num, val);
1058                         DEBUG("write_riemann: got attr: %s => %s", key, val);
1059                         sfree(key);
1060                         sfree(val);
1061                 } else if (strcasecmp(child->key, "tag") == 0) {
1062                         char *tmp = NULL;
1063                         status = cf_util_get_string(child, &tmp);
1064                         if (status != 0)
1065                                 continue;
1066
1067                         strarray_add (&riemann_tags, &riemann_tags_num, tmp);
1068                         DEBUG("write_riemann plugin: Got tag: %s", tmp);
1069                         sfree (tmp);
1070                 } else {
1071                         WARNING ("write_riemann plugin: Ignoring unknown "
1072                                  "configuration option \"%s\" at top level.",
1073                                  child->key);
1074                 }
1075         }
1076         return (0);
1077 } /* }}} int riemann_config */
1078
1079 void module_register(void)
1080 {
1081         plugin_register_complex_config ("write_riemann", riemann_config);
1082 }
1083
1084 /* vim: set sw=8 sts=8 ts=8 noet : */