RRDCacheD plugin: Try to reconnect upon failed operations.
[collectd.git] / src / rrdcached.c
1 /**
2  * collectd - src/rrdcached.c
3  * Copyright (C) 2008-2013  Florian octo Forster
4  *
5  * Permission is hereby granted, free of charge, to any person obtaining a
6  * copy of this software and associated documentation files (the "Software"),
7  * to deal in the Software without restriction, including without limitation
8  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
9  * and/or sell copies of the Software, and to permit persons to whom the
10  * Software is furnished to do so, subject to the following conditions:
11  *
12  * The above copyright notice and this permission notice shall be included in
13  * all copies or substantial portions of the Software.
14  *
15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
20  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
21  * DEALINGS IN THE SOFTWARE.
22  *
23  * Authors:
24  *   Florian octo Forster <octo at collectd.org>
25  **/
26
27 #include "collectd.h"
28 #include "plugin.h"
29 #include "common.h"
30 #include "utils_rrdcreate.h"
31
32 #undef HAVE_CONFIG_H
33 #include <rrd.h>
34 #include <rrd_client.h>
35
36 /*
37  * Private variables
38  */
39 static char *datadir = NULL;
40 static char *daemon_address = NULL;
41 static _Bool config_create_files = 1;
42 static _Bool config_collect_stats = 1;
43 static rrdcreate_config_t rrdcreate_config =
44 {
45         /* stepsize = */ 0,
46         /* heartbeat = */ 0,
47         /* rrarows = */ 1200,
48         /* xff = */ 0.1,
49
50         /* timespans = */ NULL,
51         /* timespans_num = */ 0,
52
53         /* consolidation_functions = */ NULL,
54         /* consolidation_functions_num = */ 0,
55
56         /* async = */ 0
57 };
58
59 /*
60  * Prototypes.
61  */
62 static int rc_write (const data_set_t *ds, const value_list_t *vl,
63     user_data_t __attribute__((unused)) *user_data);
64 static int rc_flush (__attribute__((unused)) cdtime_t timeout,
65     const char *identifier, __attribute__((unused)) user_data_t *ud);
66
67 static int value_list_to_string (char *buffer, int buffer_len,
68     const data_set_t *ds, const value_list_t *vl)
69 {
70   int offset;
71   int status;
72   int i;
73   time_t t;
74
75   assert (0 == strcmp (ds->type, vl->type));
76
77   memset (buffer, '\0', buffer_len);
78
79   t = CDTIME_T_TO_TIME_T (vl->time);
80   status = ssnprintf (buffer, buffer_len, "%lu", (unsigned long) t);
81   if ((status < 1) || (status >= buffer_len))
82     return (-1);
83   offset = status;
84
85   for (i = 0; i < ds->ds_num; i++)
86   {
87     if ((ds->ds[i].type != DS_TYPE_COUNTER)
88         && (ds->ds[i].type != DS_TYPE_GAUGE)
89         && (ds->ds[i].type != DS_TYPE_DERIVE)
90         && (ds->ds[i].type != DS_TYPE_ABSOLUTE))
91       return (-1);
92
93     if (ds->ds[i].type == DS_TYPE_COUNTER)
94     {
95       status = ssnprintf (buffer + offset, buffer_len - offset,
96           ":%llu", vl->values[i].counter);
97     }
98     else if (ds->ds[i].type == DS_TYPE_GAUGE) 
99     {
100       status = ssnprintf (buffer + offset, buffer_len - offset,
101           ":%f", vl->values[i].gauge);
102     }
103     else if (ds->ds[i].type == DS_TYPE_DERIVE) {
104       status = ssnprintf (buffer + offset, buffer_len - offset,
105           ":%"PRIi64, vl->values[i].derive);
106     }
107     else /* if (ds->ds[i].type == DS_TYPE_ABSOLUTE) */ {
108       status = ssnprintf (buffer + offset, buffer_len - offset,
109           ":%"PRIu64, vl->values[i].absolute);
110  
111     }
112
113     if ((status < 1) || (status >= (buffer_len - offset)))
114       return (-1);
115
116     offset += status;
117   } /* for ds->ds_num */
118
119   return (0);
120 } /* int value_list_to_string */
121
122 static int value_list_to_filename (char *buffer, size_t buffer_size,
123     value_list_t const *vl)
124 {
125   char const suffix[] = ".rrd";
126   int status;
127   size_t len;
128
129   if (datadir != NULL)
130   {
131     size_t datadir_len = strlen (datadir) + 1;
132
133     if (datadir_len >= buffer_size)
134       return (ENOMEM);
135
136     sstrncpy (buffer, datadir, buffer_size);
137     buffer[datadir_len - 1] = '/';
138     buffer[datadir_len] = 0;
139
140     buffer += datadir_len;
141     buffer_size -= datadir_len;
142   }
143
144   status = FORMAT_VL (buffer, buffer_size, vl);
145   if (status != 0)
146     return (status);
147
148   len = strlen (buffer);
149   assert (len < buffer_size);
150   buffer += len;
151   buffer_size -= len;
152
153   if (buffer_size <= sizeof (suffix))
154     return (ENOMEM);
155
156   memcpy (buffer, suffix, sizeof (suffix));
157   return (0);
158 } /* int value_list_to_filename */
159
160 static int rc_config_get_int_positive (oconfig_item_t const *ci, int *ret)
161 {
162   int status;
163   int tmp = 0;
164
165   status = cf_util_get_int (ci, &tmp);
166   if (status != 0)
167     return (status);
168   if (tmp < 0)
169     return (EINVAL);
170
171   *ret = tmp;
172   return (0);
173 } /* int rc_config_get_int_positive */
174
175 static int rc_config_get_xff (oconfig_item_t const *ci, double *ret)
176 {
177   double value;
178
179   if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_NUMBER))
180   {
181     ERROR ("rrdcached plugin: The \"%s\" needs exactly one numeric argument "
182         "in the range [0.0, 1.0)", ci->key);
183     return (EINVAL);
184   }
185
186   value = ci->values[0].value.number;
187   if ((value >= 0.0) && (value < 1.0))
188   {
189     *ret = value;
190     return (0);
191   }
192
193   ERROR ("rrdcached plugin: The \"%s\" needs exactly one numeric argument "
194       "in the range [0.0, 1.0)", ci->key);
195   return (EINVAL);
196 } /* int rc_config_get_xff */
197
198 static int rc_config_add_timespan (int timespan)
199 {
200   int *tmp;
201
202   if (timespan <= 0)
203     return (EINVAL);
204
205   tmp = realloc (rrdcreate_config.timespans,
206       sizeof (*rrdcreate_config.timespans)
207       * (rrdcreate_config.timespans_num + 1));
208   if (tmp == NULL)
209     return (ENOMEM);
210   rrdcreate_config.timespans = tmp;
211
212   rrdcreate_config.timespans[rrdcreate_config.timespans_num] = timespan;
213   rrdcreate_config.timespans_num++;
214
215   return (0);
216 } /* int rc_config_add_timespan */
217
218 static int rc_config (oconfig_item_t *ci)
219 {
220   int i;
221
222   for (i = 0; i < ci->children_num; i++)
223   {
224     oconfig_item_t const *child = ci->children + i;
225     const char *key = child->key;
226     int status = 0;
227
228     if (strcasecmp ("DataDir", key) == 0)
229     {
230       status = cf_util_get_string (child, &datadir);
231       if (status == 0)
232       {
233         int len = strlen (datadir);
234
235         while ((len > 0) && (datadir[len - 1] == '/'))
236         {
237           len--;
238           datadir[len] = 0;
239         }
240
241         if (len <= 0)
242           sfree (datadir);
243       }
244     }
245     else if (strcasecmp ("DaemonAddress", key) == 0)
246       status = cf_util_get_string (child, &daemon_address);
247     else if (strcasecmp ("CreateFiles", key) == 0)
248       status = cf_util_get_boolean (child, &config_create_files);
249     else if (strcasecmp ("CreateFilesAsync", key) == 0)
250       status = cf_util_get_boolean (child, &rrdcreate_config.async);
251     else if (strcasecmp ("CollectStatistics", key) == 0)
252       status = cf_util_get_boolean (child, &config_collect_stats);
253     else if (strcasecmp ("StepSize", key) == 0)
254     {
255       int tmp = -1;
256
257       status = rc_config_get_int_positive (child, &tmp);
258       if (status == 0)
259         rrdcreate_config.stepsize = (unsigned long) tmp;
260     }
261     else if (strcasecmp ("HeartBeat", key) == 0)
262       status = rc_config_get_int_positive (child, &rrdcreate_config.heartbeat);
263     else if (strcasecmp ("RRARows", key) == 0)
264       status = rc_config_get_int_positive (child, &rrdcreate_config.rrarows);
265     else if (strcasecmp ("RRATimespan", key) == 0)
266     {
267       int tmp = -1;
268       status = rc_config_get_int_positive (child, &tmp);
269       if (status == 0)
270         status = rc_config_add_timespan (tmp);
271     }
272     else if (strcasecmp ("XFF", key) == 0)
273       status = rc_config_get_xff (child, &rrdcreate_config.xff);
274     else
275     {
276       WARNING ("rrdcached plugin: Ignoring invalid option %s.", key);
277       continue;
278     }
279
280     if (status != 0)
281       WARNING ("rrdcached plugin: Handling the \"%s\" option failed.", key);
282   }
283
284   if (daemon_address != NULL)
285   {
286     plugin_register_write ("rrdcached", rc_write, /* user_data = */ NULL);
287     plugin_register_flush ("rrdcached", rc_flush, /* user_data = */ NULL);
288   }
289   return (0);
290 } /* int rc_config */
291
292 static int try_reconnect (void)
293 {
294   int status;
295
296   rrdc_disconnect ();
297
298   rrd_clear_error ();
299   status = rrdc_connect (daemon_address);
300   if (status != 0)
301   {
302     char *err = rrd_get_error ();
303     ERROR ("rrdcached plugin: Failed to reconnect to RRDCacheD "
304         "at %s: %s (status=%d)", daemon_address, err, status);
305     return (-1);
306   }
307
308   INFO ("rrdcached plugin: Successfully reconnected to RRDCacheD "
309       "at %s", daemon_address);
310   return (0);
311 } /* int try_reconnect */
312
313 static int rc_read (void)
314 {
315   int status;
316   rrdc_stats_t *head;
317   rrdc_stats_t *ptr;
318   _Bool retried = 0;
319
320   value_t values[1];
321   value_list_t vl = VALUE_LIST_INIT;
322
323   if (daemon_address == NULL)
324     return (-1);
325
326   if (!config_collect_stats)
327     return (-1);
328
329   vl.values = values;
330   vl.values_len = 1;
331
332   if ((strncmp ("unix:", daemon_address, strlen ("unix:")) == 0)
333       || (daemon_address[0] == '/'))
334     sstrncpy (vl.host, hostname_g, sizeof (vl.host));
335   else
336     sstrncpy (vl.host, daemon_address, sizeof (vl.host));
337   sstrncpy (vl.plugin, "rrdcached", sizeof (vl.plugin));
338
339   status = rrdc_connect (daemon_address);
340   if (status != 0)
341   {
342     char *err = rrd_get_error ();
343     ERROR ("rrdcached plugin: Failed to connect to RRDCacheD "
344         "at %s: %s (status=%d)", daemon_address, err, status);
345     return (-1);
346   }
347
348   while (42)
349   {
350     /* The RRD client lib does not provide any means for checking a
351      * connection, hence we'll have to retry upon failed operations. */
352     head = NULL;
353     status = rrdc_stats_get (&head);
354     if (status == 0)
355       break;
356
357     if (! retried)
358     {
359       retried = 1;
360       if (try_reconnect () == 0)
361         continue;
362       /* else: report the error and fail */
363     }
364
365     ERROR ("rrdcached plugin: rrdc_stats_get failed with status %i.", status);
366     return (-1);
367   }
368
369   for (ptr = head; ptr != NULL; ptr = ptr->next)
370   {
371     if (ptr->type == RRDC_STATS_TYPE_GAUGE)
372       values[0].gauge = (gauge_t) ptr->value.gauge;
373     else if (ptr->type == RRDC_STATS_TYPE_COUNTER)
374       values[0].counter = (counter_t) ptr->value.counter;
375     else
376       continue;
377
378     if (strcasecmp ("QueueLength", ptr->name) == 0)
379     {
380       sstrncpy (vl.type, "queue_length", sizeof (vl.type));
381       sstrncpy (vl.type_instance, "", sizeof (vl.type_instance));
382     }
383     else if (strcasecmp ("UpdatesWritten", ptr->name) == 0)
384     {
385       sstrncpy (vl.type, "operations", sizeof (vl.type));
386       sstrncpy (vl.type_instance, "write-updates", sizeof (vl.type_instance));
387     }
388     else if (strcasecmp ("DataSetsWritten", ptr->name) == 0)
389     {
390       sstrncpy (vl.type, "operations", sizeof (vl.type));
391       sstrncpy (vl.type_instance, "write-data_sets",
392           sizeof (vl.type_instance));
393     }
394     else if (strcasecmp ("TreeNodesNumber", ptr->name) == 0)
395     {
396       sstrncpy (vl.type, "gauge", sizeof (vl.type));
397       sstrncpy (vl.type_instance, "tree_nodes", sizeof (vl.type_instance));
398     }
399     else if (strcasecmp ("TreeDepth", ptr->name) == 0)
400     {
401       sstrncpy (vl.type, "gauge", sizeof (vl.type));
402       sstrncpy (vl.type_instance, "tree_depth", sizeof (vl.type_instance));
403     }
404     else if (strcasecmp ("FlushesReceived", ptr->name) == 0)
405     {
406       sstrncpy (vl.type, "operations", sizeof (vl.type));
407       sstrncpy (vl.type_instance, "receive-flush", sizeof (vl.type_instance));
408     }
409     else if (strcasecmp ("JournalBytes", ptr->name) == 0)
410     {
411       sstrncpy (vl.type, "counter", sizeof (vl.type));
412       sstrncpy (vl.type_instance, "journal-bytes", sizeof (vl.type_instance));
413     }
414     else if (strcasecmp ("JournalRotate", ptr->name) == 0)
415     {
416       sstrncpy (vl.type, "counter", sizeof (vl.type));
417       sstrncpy (vl.type_instance, "journal-rotates", sizeof (vl.type_instance));
418     }
419     else if (strcasecmp ("UpdatesReceived", ptr->name) == 0)
420     {
421       sstrncpy (vl.type, "operations", sizeof (vl.type));
422       sstrncpy (vl.type_instance, "receive-update", sizeof (vl.type_instance));
423     }
424     else
425     {
426       DEBUG ("rrdcached plugin: rc_read: Unknown statistic `%s'.", ptr->name);
427       continue;
428     }
429
430     plugin_dispatch_values (&vl);
431   } /* for (ptr = head; ptr != NULL; ptr = ptr->next) */
432
433   rrdc_stats_free (head);
434
435   return (0);
436 } /* int rc_read */
437
438 static int rc_init (void)
439 {
440   if (config_collect_stats)
441     plugin_register_read ("rrdcached", rc_read);
442
443   return (0);
444 } /* int rc_init */
445
446 static int rc_write (const data_set_t *ds, const value_list_t *vl,
447     user_data_t __attribute__((unused)) *user_data)
448 {
449   char filename[PATH_MAX];
450   char values[512];
451   char *values_array[2];
452   int status;
453   _Bool retried = 0;
454
455   if (daemon_address == NULL)
456   {
457     ERROR ("rrdcached plugin: daemon_address == NULL.");
458     plugin_unregister_write ("rrdcached");
459     return (-1);
460   }
461
462   if (strcmp (ds->type, vl->type) != 0)
463   {
464     ERROR ("rrdcached plugin: DS type does not match value list type");
465     return (-1);
466   }
467
468   if (value_list_to_filename (filename, sizeof (filename), vl) != 0)
469   {
470     ERROR ("rrdcached plugin: value_list_to_filename failed.");
471     return (-1);
472   }
473
474   if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
475   {
476     ERROR ("rrdcached plugin: value_list_to_string failed.");
477     return (-1);
478   }
479
480   values_array[0] = values;
481   values_array[1] = NULL;
482
483   if (config_create_files)
484   {
485     struct stat statbuf;
486
487     status = stat (filename, &statbuf);
488     if (status != 0)
489     {
490       if (errno != ENOENT)
491       {
492         char errbuf[1024];
493         ERROR ("rrdcached plugin: stat (%s) failed: %s",
494             filename, sstrerror (errno, errbuf, sizeof (errbuf)));
495         return (-1);
496       }
497
498       status = cu_rrd_create_file (filename, ds, vl, &rrdcreate_config);
499       if (status != 0)
500       {
501         ERROR ("rrdcached plugin: cu_rrd_create_file (%s) failed.",
502             filename);
503         return (-1);
504       }
505       else if (rrdcreate_config.async)
506         return (0);
507     }
508   }
509
510   status = rrdc_connect (daemon_address);
511   if (status != 0)
512   {
513     char *err = rrd_get_error ();
514     ERROR ("rrdcached plugin: Failed to connect to RRDCacheD "
515         "at %s: %s (status=%d)", daemon_address, err, status);
516     return (-1);
517   }
518
519   while (42)
520   {
521     /* The RRD client lib does not provide any means for checking a
522      * connection, hence we'll have to retry upon failed operations. */
523     status = rrdc_update (filename, /* values_num = */ 1, (void *) values_array);
524     if (status == 0)
525       break;
526
527     if (! retried)
528     {
529       retried = 1;
530       if (try_reconnect () == 0)
531         continue;
532       /* else: report the error and fail */
533     }
534
535     ERROR ("rrdcached plugin: rrdc_update (%s, [%s], 1) failed with "
536         "status %i.",
537         filename, values_array[0], status);
538     return (-1);
539   }
540
541   return (0);
542 } /* int rc_write */
543
544 static int rc_flush (__attribute__((unused)) cdtime_t timeout, /* {{{ */
545     const char *identifier,
546     __attribute__((unused)) user_data_t *ud)
547 {
548   char filename[PATH_MAX + 1];
549   int status;
550   _Bool retried = 0;
551
552   if (identifier == NULL)
553     return (EINVAL);
554
555   if (datadir != NULL)
556     ssnprintf (filename, sizeof (filename), "%s/%s.rrd", datadir, identifier);
557   else
558     ssnprintf (filename, sizeof (filename), "%s.rrd", identifier);
559
560   status = rrdc_connect (daemon_address);
561   if (status != 0)
562   {
563     char *err = rrd_get_error ();
564     ERROR ("rrdcached plugin: Failed to connect to RRDCacheD "
565         "at %s: %s (status=%d)", daemon_address, err, status);
566     return (-1);
567   }
568
569   while (42)
570   {
571     /* The RRD client lib does not provide any means for checking a
572      * connection, hence we'll have to retry upon failed operations. */
573     status = rrdc_flush (filename);
574     if (status == 0)
575       break;
576
577     if (! retried)
578     {
579       retried = 1;
580       if (try_reconnect () == 0)
581         continue;
582       /* else: report the error and fail */
583     }
584
585     ERROR ("rrdcached plugin: rrdc_flush (%s) failed with status %i.",
586         filename, status);
587     return (-1);
588   }
589   DEBUG ("rrdcached plugin: rrdc_flush (%s): Success.", filename);
590
591   return (0);
592 } /* }}} int rc_flush */
593
594 static int rc_shutdown (void)
595 {
596   rrdc_disconnect ();
597   return (0);
598 } /* int rc_shutdown */
599
600 void module_register (void)
601 {
602   plugin_register_complex_config ("rrdcached", rc_config);
603   plugin_register_init ("rrdcached", rc_init);
604   plugin_register_shutdown ("rrdcached", rc_shutdown);
605 } /* void module_register */
606
607 /*
608  * vim: set sw=2 sts=2 et :
609  */