1 /**
2 * RRDTool - src/rrd_client.c
3 * Copyright (C) 2008-2010 Florian octo Forster
4 *
5 * Permission is hereby granted, free of charge, to any person obtaining a copy
6 * of this software and associated documentation files (the "Software"), to
7 * deal in the Software without restriction, including without limitation the
8 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
9 * sell copies of the Software, and to permit persons to whom the Software is
10 * 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 DEALINGS
21 * IN THE SOFTWARE.
22 *
23 * Authors:
24 * Florian octo Forster <octo at verplant.org>
25 * Sebastian tokkee Harl <sh at tokkee.org>
26 **/
28 #include "rrd.h"
29 #include "rrd_tool.h"
30 #include "rrd_client.h"
32 #include <stdio.h>
33 #include <stdlib.h>
34 #include <string.h>
35 #include <strings.h>
36 #include <errno.h>
37 #include <assert.h>
38 #include <pthread.h>
39 #include <sys/types.h>
40 #include <sys/socket.h>
41 #include <sys/un.h>
42 #include <netdb.h>
43 #include <limits.h>
45 #ifndef ENODATA
46 #define ENODATA ENOENT
47 #endif
49 struct rrdc_response_s
50 {
51 int status;
52 char *message;
53 char **lines;
54 size_t lines_num;
55 };
56 typedef struct rrdc_response_s rrdc_response_t;
58 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
59 static int sd = -1;
60 static FILE *sh = NULL;
61 static char *sd_path = NULL; /* cache the path for sd */
63 /* get_path: Return a path name appropriate to be sent to the daemon.
64 *
65 * When talking to a local daemon (thru a UNIX socket), relative path names
66 * are resolved to absolute path names to allow for transparent integration
67 * into existing solutions (as requested by Tobi). Else, absolute path names
68 * are not allowed, since path name translation is done by the server.
69 *
70 * One must hold `lock' when calling this function. */
71 static const char *get_path (const char *path, char *resolved_path) /* {{{ */
72 {
73 const char *ret = path;
74 int is_unix = 0;
76 if ((*sd_path == '/')
77 || (strncmp ("unix:", sd_path, strlen ("unix:")) == 0))
78 is_unix = 1;
80 if (*path == '/') /* absolute path */
81 {
82 if (! is_unix)
83 {
84 rrd_set_error ("absolute path names not allowed when talking "
85 "to a remote daemon");
86 return (NULL);
87 }
88 /* else: nothing to do */
89 }
90 else /* relative path */
91 {
92 if (is_unix)
93 {
94 realpath (path, resolved_path);
95 ret = resolved_path;
96 }
97 /* else: nothing to do */
98 }
99 return (ret);
100 } /* }}} char *get_path */
102 static size_t strsplit (char *string, char **fields, size_t size) /* {{{ */
103 {
104 size_t i;
105 char *ptr;
106 char *saveptr;
108 i = 0;
109 ptr = string;
110 saveptr = NULL;
111 while ((fields[i] = strtok_r (ptr, " \t\r\n", &saveptr)) != NULL)
112 {
113 ptr = NULL;
114 i++;
116 if (i >= size)
117 break;
118 }
120 return (i);
121 } /* }}} size_t strsplit */
123 static int parse_header (char *line, /* {{{ */
124 char **ret_key, char **ret_value)
125 {
126 char *tmp;
128 *ret_key = line;
130 tmp = strchr (line, ':');
131 if (tmp == NULL)
132 return (-1);
134 do
135 {
136 *tmp = 0;
137 tmp++;
138 }
139 while ((tmp[0] == ' ') || (tmp[0] == '\t'));
141 if (*tmp == 0)
142 return (-1);
144 *ret_value = tmp;
145 return (0);
146 } /* }}} int parse_header */
148 static int parse_ulong_header (char *line, /* {{{ */
149 char **ret_key, unsigned long *ret_value)
150 {
151 char *str_value;
152 char *endptr;
153 int status;
155 str_value = NULL;
156 status = parse_header (line, ret_key, &str_value);
157 if (status != 0)
158 return (status);
160 endptr = NULL;
161 errno = 0;
162 *ret_value = (unsigned long) strtol (str_value, &endptr, /* base = */ 0);
163 if ((endptr == str_value) || (errno != 0))
164 return (-1);
166 return (0);
167 } /* }}} int parse_ulong_header */
169 static int parse_char_array_header (char *line, /* {{{ */
170 char **ret_key, char **array, size_t array_len, int alloc)
171 {
172 char *tmp_array[array_len];
173 char *value;
174 size_t num;
175 int status;
177 value = NULL;
178 status = parse_header (line, ret_key, &value);
179 if (status != 0)
180 return (-1);
182 num = strsplit (value, tmp_array, array_len);
183 if (num != array_len)
184 return (-1);
186 if (alloc == 0)
187 {
188 memcpy (array, tmp_array, sizeof (tmp_array));
189 }
190 else
191 {
192 size_t i;
194 for (i = 0; i < array_len; i++)
195 array[i] = strdup (tmp_array[i]);
196 }
198 return (0);
199 } /* }}} int parse_char_array_header */
201 static int parse_value_array_header (char *line, /* {{{ */
202 time_t *ret_time, rrd_value_t *array, size_t array_len)
203 {
204 char *str_key;
205 char *str_array[array_len];
206 char *endptr;
207 int status;
208 size_t i;
210 str_key = NULL;
211 status = parse_char_array_header (line, &str_key,
212 str_array, array_len, /* alloc = */ 0);
213 if (status != 0)
214 return (-1);
216 errno = 0;
217 endptr = NULL;
218 *ret_time = (time_t) strtol (str_key, &endptr, /* base = */ 10);
219 if ((endptr == str_key) || (errno != 0))
220 return (-1);
222 for (i = 0; i < array_len; i++)
223 {
224 endptr = NULL;
225 array[i] = (rrd_value_t) strtod (str_array[i], &endptr);
226 if ((endptr == str_array[i]) || (errno != 0))
227 return (-1);
228 }
230 return (0);
231 } /* }}} int parse_value_array_header */
233 /* One must hold `lock' when calling `close_connection'. */
234 static void close_connection (void) /* {{{ */
235 {
236 if (sh != NULL)
237 {
238 fclose (sh);
239 sh = NULL;
240 sd = -1;
241 }
242 else if (sd >= 0)
243 {
244 close (sd);
245 sd = -1;
246 }
248 if (sd_path != NULL)
249 free (sd_path);
250 sd_path = NULL;
251 } /* }}} void close_connection */
253 static int buffer_add_string (const char *str, /* {{{ */
254 char **buffer_ret, size_t *buffer_size_ret)
255 {
256 char *buffer;
257 size_t buffer_size;
258 size_t buffer_pos;
259 size_t i;
260 int status;
262 buffer = *buffer_ret;
263 buffer_size = *buffer_size_ret;
264 buffer_pos = 0;
266 i = 0;
267 status = -1;
268 while (buffer_pos < buffer_size)
269 {
270 if (str[i] == 0)
271 {
272 buffer[buffer_pos] = ' ';
273 buffer_pos++;
274 status = 0;
275 break;
276 }
277 else if ((str[i] == ' ') || (str[i] == '\\'))
278 {
279 if (buffer_pos >= (buffer_size - 1))
280 break;
281 buffer[buffer_pos] = '\\';
282 buffer_pos++;
283 buffer[buffer_pos] = str[i];
284 buffer_pos++;
285 }
286 else
287 {
288 buffer[buffer_pos] = str[i];
289 buffer_pos++;
290 }
291 i++;
292 } /* while (buffer_pos < buffer_size) */
294 if (status != 0)
295 return (-1);
297 *buffer_ret = buffer + buffer_pos;
298 *buffer_size_ret = buffer_size - buffer_pos;
300 return (0);
301 } /* }}} int buffer_add_string */
303 static int buffer_add_value (const char *value, /* {{{ */
304 char **buffer_ret, size_t *buffer_size_ret)
305 {
306 char temp[4096];
308 if (strncmp (value, "N:", 2) == 0)
309 snprintf (temp, sizeof (temp), "%lu:%s",
310 (unsigned long) time (NULL), value + 2);
311 else
312 strncpy (temp, value, sizeof (temp));
313 temp[sizeof (temp) - 1] = 0;
315 return (buffer_add_string (temp, buffer_ret, buffer_size_ret));
316 } /* }}} int buffer_add_value */
318 /* Remove trailing newline (NL) and carriage return (CR) characters. Similar to
319 * the Perl function `chomp'. Returns the number of characters that have been
320 * removed. */
321 static int chomp (char *str) /* {{{ */
322 {
323 size_t len;
324 int removed;
326 if (str == NULL)
327 return (-1);
329 len = strlen (str);
330 removed = 0;
331 while ((len > 0) && ((str[len - 1] == '\n') || (str[len - 1] == '\r')))
332 {
333 str[len - 1] = 0;
334 len--;
335 removed++;
336 }
338 return (removed);
339 } /* }}} int chomp */
341 static void response_free (rrdc_response_t *res) /* {{{ */
342 {
343 if (res == NULL)
344 return;
346 if (res->lines != NULL)
347 {
348 size_t i;
350 for (i = 0; i < res->lines_num; i++)
351 if (res->lines[i] != NULL)
352 free (res->lines[i]);
353 free (res->lines);
354 }
356 free (res);
357 } /* }}} void response_free */
359 static int response_read (rrdc_response_t **ret_response) /* {{{ */
360 {
361 rrdc_response_t *ret;
363 char buffer[4096];
364 char *buffer_ptr;
366 size_t i;
368 if (sh == NULL)
369 return (-1);
371 ret = (rrdc_response_t *) malloc (sizeof (rrdc_response_t));
372 if (ret == NULL)
373 return (-2);
374 memset (ret, 0, sizeof (*ret));
375 ret->lines = NULL;
376 ret->lines_num = 0;
378 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
379 if (buffer_ptr == NULL) {
380 close_connection();
381 return (-3);
382 }
383 chomp (buffer);
385 ret->status = strtol (buffer, &ret->message, 0);
386 if (buffer == ret->message)
387 {
388 response_free (ret);
389 close_connection();
390 return (-4);
391 }
392 /* Skip leading whitespace of the status message */
393 ret->message += strspn (ret->message, " \t");
395 if (ret->status <= 0)
396 {
397 if (ret->status < 0)
398 rrd_set_error("rrdcached: %s", ret->message);
399 *ret_response = ret;
400 return (0);
401 }
403 ret->lines = (char **) malloc (sizeof (char *) * ret->status);
404 if (ret->lines == NULL)
405 {
406 response_free (ret);
407 close_connection();
408 return (-5);
409 }
410 memset (ret->lines, 0, sizeof (char *) * ret->status);
411 ret->lines_num = (size_t) ret->status;
413 for (i = 0; i < ret->lines_num; i++)
414 {
415 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
416 if (buffer_ptr == NULL)
417 {
418 response_free (ret);
419 close_connection();
420 return (-6);
421 }
422 chomp (buffer);
424 ret->lines[i] = strdup (buffer);
425 if (ret->lines[i] == NULL)
426 {
427 response_free (ret);
428 close_connection();
429 return (-7);
430 }
431 }
433 *ret_response = ret;
434 return (0);
435 } /* }}} rrdc_response_t *response_read */
437 static int request (const char *buffer, size_t buffer_size, /* {{{ */
438 rrdc_response_t **ret_response)
439 {
440 int status;
441 rrdc_response_t *res;
443 if (sh == NULL)
444 return (ENOTCONN);
446 status = (int) fwrite (buffer, buffer_size, /* nmemb = */ 1, sh);
447 if (status != 1)
448 {
449 close_connection ();
450 rrd_set_error("request: socket error (%d) while talking to rrdcached",
451 status);
452 return (-1);
453 }
454 fflush (sh);
456 res = NULL;
457 status = response_read (&res);
459 if (status != 0)
460 {
461 if (status < 0)
462 rrd_set_error("request: internal error while talking to rrdcached");
463 return (status);
464 }
466 *ret_response = res;
467 return (0);
468 } /* }}} int request */
470 /* determine whether we are connected to the specified daemon_addr if
471 * NULL, return whether we are connected at all
472 */
473 int rrdc_is_connected(const char *daemon_addr) /* {{{ */
474 {
475 if (sd < 0)
476 return 0;
477 else if (daemon_addr == NULL)
478 {
479 /* here we have to handle the case i.e.
480 * UPDATE --daemon ...; UPDATEV (no --daemon) ...
481 * In other words: we have a cached connection,
482 * but it is not specified in the current command.
483 * Daemon is only implied in this case if set in ENV
484 */
485 if (getenv(ENV_RRDCACHED_ADDRESS) != NULL)
486 return 1;
487 else
488 return 0;
489 }
490 else if (strcmp(daemon_addr, sd_path) == 0)
491 return 1;
492 else
493 return 0;
495 } /* }}} int rrdc_is_connected */
497 static int rrdc_connect_unix (const char *path) /* {{{ */
498 {
499 struct sockaddr_un sa;
500 int status;
502 assert (path != NULL);
503 assert (sd == -1);
505 sd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
506 if (sd < 0)
507 {
508 status = errno;
509 return (status);
510 }
512 memset (&sa, 0, sizeof (sa));
513 sa.sun_family = AF_UNIX;
514 strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
516 status = connect (sd, (struct sockaddr *) &sa, sizeof (sa));
517 if (status != 0)
518 {
519 status = errno;
520 close_connection ();
521 return (status);
522 }
524 sh = fdopen (sd, "r+");
525 if (sh == NULL)
526 {
527 status = errno;
528 close_connection ();
529 return (status);
530 }
532 return (0);
533 } /* }}} int rrdc_connect_unix */
535 static int rrdc_connect_network (const char *addr_orig) /* {{{ */
536 {
537 struct addrinfo ai_hints;
538 struct addrinfo *ai_res;
539 struct addrinfo *ai_ptr;
540 char addr_copy[NI_MAXHOST];
541 char *addr;
542 char *port;
544 assert (addr_orig != NULL);
545 assert (sd == -1);
547 strncpy(addr_copy, addr_orig, sizeof(addr_copy));
548 addr_copy[sizeof(addr_copy) - 1] = '\0';
549 addr = addr_copy;
551 int status;
552 memset (&ai_hints, 0, sizeof (ai_hints));
553 ai_hints.ai_flags = 0;
554 #ifdef AI_ADDRCONFIG
555 ai_hints.ai_flags |= AI_ADDRCONFIG;
556 #endif
557 ai_hints.ai_family = AF_UNSPEC;
558 ai_hints.ai_socktype = SOCK_STREAM;
560 port = NULL;
561 if (*addr == '[') /* IPv6+port format */
562 {
563 /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
564 addr++;
566 port = strchr (addr, ']');
567 if (port == NULL)
568 {
569 rrd_set_error("malformed address: %s", addr_orig);
570 return (-1);
571 }
572 *port = 0;
573 port++;
575 if (*port == ':')
576 port++;
577 else if (*port == 0)
578 port = NULL;
579 else
580 {
581 rrd_set_error("garbage after address: %s", port);
582 return (-1);
583 }
584 } /* if (*addr == '[') */
585 else
586 {
587 port = rindex(addr, ':');
588 if (port != NULL)
589 {
590 *port = 0;
591 port++;
592 }
593 }
595 ai_res = NULL;
596 status = getaddrinfo (addr,
597 port == NULL ? RRDCACHED_DEFAULT_PORT : port,
598 &ai_hints, &ai_res);
599 if (status != 0)
600 {
601 rrd_set_error ("failed to resolve address `%s' (port %s): %s",
602 addr, port == NULL ? RRDCACHED_DEFAULT_PORT : port,
603 gai_strerror (status));
604 return (-1);
605 }
607 for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
608 {
609 sd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
610 if (sd < 0)
611 {
612 status = errno;
613 sd = -1;
614 continue;
615 }
617 status = connect (sd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
618 if (status != 0)
619 {
620 status = errno;
621 close_connection();
622 continue;
623 }
625 sh = fdopen (sd, "r+");
626 if (sh == NULL)
627 {
628 status = errno;
629 close_connection ();
630 continue;
631 }
633 assert (status == 0);
634 break;
635 } /* for (ai_ptr) */
637 return (status);
638 } /* }}} int rrdc_connect_network */
640 int rrdc_connect (const char *addr) /* {{{ */
641 {
642 int status = 0;
644 if (addr == NULL)
645 addr = getenv (ENV_RRDCACHED_ADDRESS);
647 if (addr == NULL)
648 return 0;
650 pthread_mutex_lock(&lock);
652 if (sd >= 0 && sd_path != NULL && strcmp(addr, sd_path) == 0)
653 {
654 /* connection to the same daemon; use cached connection */
655 pthread_mutex_unlock (&lock);
656 return (0);
657 }
658 else
659 {
660 close_connection();
661 }
663 rrd_clear_error ();
664 if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
665 status = rrdc_connect_unix (addr + strlen ("unix:"));
666 else if (addr[0] == '/')
667 status = rrdc_connect_unix (addr);
668 else
669 status = rrdc_connect_network(addr);
671 if (status == 0 && sd >= 0)
672 sd_path = strdup(addr);
673 else
674 {
675 char *err = rrd_test_error () ? rrd_get_error () : "Internal error";
676 /* err points the string that gets written to by rrd_set_error(), thus we
677 * cannot pass it to that function */
678 err = strdup (err);
679 rrd_set_error("Unable to connect to rrdcached: %s",
680 (status < 0)
681 ? (err ? err : "Internal error")
682 : rrd_strerror (status));
683 if (err != NULL)
684 free (err);
685 }
687 pthread_mutex_unlock (&lock);
688 return (status);
689 } /* }}} int rrdc_connect */
691 int rrdc_disconnect (void) /* {{{ */
692 {
693 pthread_mutex_lock (&lock);
695 close_connection();
697 pthread_mutex_unlock (&lock);
699 return (0);
700 } /* }}} int rrdc_disconnect */
702 int rrdc_update (const char *filename, int values_num, /* {{{ */
703 const char * const *values)
704 {
705 char buffer[4096];
706 char *buffer_ptr;
707 size_t buffer_free;
708 size_t buffer_size;
709 rrdc_response_t *res;
710 int status;
711 int i;
712 char file_path[PATH_MAX];
714 memset (buffer, 0, sizeof (buffer));
715 buffer_ptr = &buffer[0];
716 buffer_free = sizeof (buffer);
718 status = buffer_add_string ("update", &buffer_ptr, &buffer_free);
719 if (status != 0)
720 return (ENOBUFS);
722 pthread_mutex_lock (&lock);
723 filename = get_path (filename, file_path);
724 if (filename == NULL)
725 {
726 pthread_mutex_unlock (&lock);
727 return (-1);
728 }
730 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
731 if (status != 0)
732 {
733 pthread_mutex_unlock (&lock);
734 return (ENOBUFS);
735 }
737 for (i = 0; i < values_num; i++)
738 {
739 status = buffer_add_value (values[i], &buffer_ptr, &buffer_free);
740 if (status != 0)
741 {
742 pthread_mutex_unlock (&lock);
743 return (ENOBUFS);
744 }
745 }
747 assert (buffer_free < sizeof (buffer));
748 buffer_size = sizeof (buffer) - buffer_free;
749 assert (buffer[buffer_size - 1] == ' ');
750 buffer[buffer_size - 1] = '\n';
752 res = NULL;
753 status = request (buffer, buffer_size, &res);
754 pthread_mutex_unlock (&lock);
756 if (status != 0)
757 return (status);
759 status = res->status;
760 response_free (res);
762 return (status);
763 } /* }}} int rrdc_update */
765 int rrdc_flush (const char *filename) /* {{{ */
766 {
767 char buffer[4096];
768 char *buffer_ptr;
769 size_t buffer_free;
770 size_t buffer_size;
771 rrdc_response_t *res;
772 int status;
773 char file_path[PATH_MAX];
775 if (filename == NULL)
776 return (-1);
778 memset (buffer, 0, sizeof (buffer));
779 buffer_ptr = &buffer[0];
780 buffer_free = sizeof (buffer);
782 status = buffer_add_string ("flush", &buffer_ptr, &buffer_free);
783 if (status != 0)
784 return (ENOBUFS);
786 pthread_mutex_lock (&lock);
787 filename = get_path (filename, file_path);
788 if (filename == NULL)
789 {
790 pthread_mutex_unlock (&lock);
791 return (-1);
792 }
794 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
795 if (status != 0)
796 {
797 pthread_mutex_unlock (&lock);
798 return (ENOBUFS);
799 }
801 assert (buffer_free < sizeof (buffer));
802 buffer_size = sizeof (buffer) - buffer_free;
803 assert (buffer[buffer_size - 1] == ' ');
804 buffer[buffer_size - 1] = '\n';
806 res = NULL;
807 status = request (buffer, buffer_size, &res);
808 pthread_mutex_unlock (&lock);
810 if (status != 0)
811 return (status);
813 status = res->status;
814 response_free (res);
816 return (status);
817 } /* }}} int rrdc_flush */
819 int rrdc_fetch (const char *filename, /* {{{ */
820 const char *cf,
821 time_t *ret_start, time_t *ret_end,
822 unsigned long *ret_step,
823 unsigned long *ret_ds_num,
824 char ***ret_ds_names,
825 rrd_value_t **ret_data)
826 {
827 char buffer[4096];
828 char *buffer_ptr;
829 size_t buffer_free;
830 size_t buffer_size;
831 rrdc_response_t *res;
832 char path_buffer[PATH_MAX];
833 char *path_ptr;
835 char *str_tmp;
836 unsigned long flush_version;
838 time_t start;
839 time_t end;
840 unsigned long step;
841 unsigned long ds_num;
842 char **ds_names;
844 rrd_value_t *data;
845 size_t data_size;
846 size_t data_fill;
848 int status;
849 size_t current_line;
850 time_t t;
852 if ((filename == NULL) || (cf == NULL))
853 return (-1);
855 /* Send request {{{ */
856 memset (buffer, 0, sizeof (buffer));
857 buffer_ptr = &buffer[0];
858 buffer_free = sizeof (buffer);
860 status = buffer_add_string ("FETCH", &buffer_ptr, &buffer_free);
861 if (status != 0)
862 return (ENOBUFS);
864 /* change to path for rrdcached */
865 path_ptr = get_path (filename, path_buffer);
866 if (path_ptr == NULL)
867 return (EINVAL);
869 status = buffer_add_string (path_ptr, &buffer_ptr, &buffer_free);
870 if (status != 0)
871 return (ENOBUFS);
873 status = buffer_add_string (cf, &buffer_ptr, &buffer_free);
874 if (status != 0)
875 return (ENOBUFS);
877 if ((ret_start != NULL) && (*ret_start > 0))
878 {
879 char tmp[64];
880 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_start);
881 tmp[sizeof (tmp) - 1] = 0;
882 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
883 if (status != 0)
884 return (ENOBUFS);
886 if ((ret_end != NULL) && (*ret_end > 0))
887 {
888 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_end);
889 tmp[sizeof (tmp) - 1] = 0;
890 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
891 if (status != 0)
892 return (ENOBUFS);
893 }
894 }
896 assert (buffer_free < sizeof (buffer));
897 buffer_size = sizeof (buffer) - buffer_free;
898 assert (buffer[buffer_size - 1] == ' ');
899 buffer[buffer_size - 1] = '\n';
901 res = NULL;
902 status = request (buffer, buffer_size, &res);
903 if (status != 0)
904 return (status);
906 status = res->status;
907 if (status < 0)
908 {
909 rrd_set_error ("rrdcached: %s", res->message);
910 response_free (res);
911 return (status);
912 }
913 /* }}} Send request */
915 ds_names = NULL;
916 ds_num = 0;
917 data = NULL;
918 current_line = 0;
920 /* Macros to make error handling a little easier (i. e. less to type and
921 * read. `BAIL_OUT' sets the error message, frees all dynamically allocated
922 * variables and returns the provided status code. */
923 #define BAIL_OUT(status, ...) do { \
924 rrd_set_error ("rrdc_fetch: " __VA_ARGS__); \
925 free (data); \
926 if (ds_names != 0) { size_t k; for (k = 0; k < ds_num; k++) free (ds_names[k]); } \
927 free (ds_names); \
928 response_free (res); \
929 return (status); \
930 } while (0)
932 #define READ_NUMERIC_FIELD(name,type,var) do { \
933 char *key; \
934 unsigned long value; \
935 assert (current_line < res->lines_num); \
936 status = parse_ulong_header (res->lines[current_line], &key, &value); \
937 if (status != 0) \
938 BAIL_OUT (-1, "Unable to parse header `%s'", name); \
939 if (strcasecmp (key, name) != 0) \
940 BAIL_OUT (-1, "Unexpected header line: Expected `%s', got `%s'", name, key); \
941 var = (type) value; \
942 current_line++; \
943 } while (0)
945 if (res->lines_num < 1)
946 BAIL_OUT (-1, "Premature end of response packet");
948 /* We're making some very strong assumptions about the fields below. We
949 * therefore check the version of the `flush' command first, so that later
950 * versions can change the order of fields and it's easier to implement
951 * backwards compatibility. */
952 READ_NUMERIC_FIELD ("FlushVersion", unsigned long, flush_version);
953 if (flush_version != 1)
954 BAIL_OUT (-1, "Don't know how to handle flush format version %lu.",
955 flush_version);
957 if (res->lines_num < 5)
958 BAIL_OUT (-1, "Premature end of response packet");
960 READ_NUMERIC_FIELD ("Start", time_t, start);
961 READ_NUMERIC_FIELD ("End", time_t, end);
962 if (start >= end)
963 BAIL_OUT (-1, "Malformed start and end times: start = %lu; end = %lu;",
964 (unsigned long) start,
965 (unsigned long) end);
967 READ_NUMERIC_FIELD ("Step", unsigned long, step);
968 if (step < 1)
969 BAIL_OUT (-1, "Invalid number for Step: %lu", step);
971 READ_NUMERIC_FIELD ("DSCount", unsigned long, ds_num);
972 if (ds_num < 1)
973 BAIL_OUT (-1, "Invalid number for DSCount: %lu", ds_num);
975 /* It's time to allocate some memory */
976 ds_names = calloc ((size_t) ds_num, sizeof (*ds_names));
977 if (ds_names == NULL)
978 BAIL_OUT (-1, "Out of memory");
980 status = parse_char_array_header (res->lines[current_line],
981 &str_tmp, ds_names, (size_t) ds_num, /* alloc = */ 1);
982 if (status != 0)
983 BAIL_OUT (-1, "Unable to parse header `DSName'");
984 if (strcasecmp ("DSName", str_tmp) != 0)
985 BAIL_OUT (-1, "Unexpected header line: Expected `DSName', got `%s'", str_tmp);
986 current_line++;
988 data_size = ds_num * (end - start) / step;
989 if (data_size < 1)
990 BAIL_OUT (-1, "No data returned or headers invalid.");
992 if (res->lines_num != (6 + (data_size / ds_num)))
993 BAIL_OUT (-1, "Got %zu lines, expected %zu",
994 res->lines_num, (6 + (data_size / ds_num)));
996 data = calloc (data_size, sizeof (*data));
997 if (data == NULL)
998 BAIL_OUT (-1, "Out of memory");
1001 data_fill = 0;
1002 for (t = start + step; t <= end; t += step, current_line++)
1003 {
1004 time_t tmp;
1006 assert (current_line < res->lines_num);
1008 status = parse_value_array_header (res->lines[current_line],
1009 &tmp, data + data_fill, (size_t) ds_num);
1010 if (status != 0)
1011 BAIL_OUT (-1, "Cannot parse value line");
1013 data_fill += (size_t) ds_num;
1014 }
1016 *ret_start = start;
1017 *ret_end = end;
1018 *ret_step = step;
1019 *ret_ds_num = ds_num;
1020 *ret_ds_names = ds_names;
1021 *ret_data = data;
1023 response_free (res);
1024 return (0);
1025 #undef READ_NUMERIC_FIELD
1026 #undef BAIL_OUT
1027 } /* }}} int rrdc_flush */
1029 /* convenience function; if there is a daemon specified, or if we can
1030 * detect one from the environment, then flush the file. Otherwise, no-op
1031 */
1032 int rrdc_flush_if_daemon (const char *opt_daemon, const char *filename) /* {{{ */
1033 {
1034 int status = 0;
1036 rrdc_connect(opt_daemon);
1038 if (rrdc_is_connected(opt_daemon))
1039 {
1040 rrd_clear_error();
1041 status = rrdc_flush (filename);
1043 if (status != 0 && !rrd_test_error())
1044 {
1045 if (status > 0)
1046 {
1047 rrd_set_error("rrdc_flush (%s) failed: %s",
1048 filename, rrd_strerror(status));
1049 }
1050 else if (status < 0)
1051 {
1052 rrd_set_error("rrdc_flush (%s) failed with status %i.",
1053 filename, status);
1054 }
1055 }
1056 } /* if (rrdc_is_connected(..)) */
1058 return status;
1059 } /* }}} int rrdc_flush_if_daemon */
1062 int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */
1063 {
1064 rrdc_stats_t *head;
1065 rrdc_stats_t *tail;
1067 rrdc_response_t *res;
1069 int status;
1070 size_t i;
1072 /* Protocol example: {{{
1073 * -> STATS
1074 * <- 5 Statistics follow
1075 * <- QueueLength: 0
1076 * <- UpdatesWritten: 0
1077 * <- DataSetsWritten: 0
1078 * <- TreeNodesNumber: 0
1079 * <- TreeDepth: 0
1080 * }}} */
1082 res = NULL;
1083 pthread_mutex_lock (&lock);
1084 status = request ("STATS\n", strlen ("STATS\n"), &res);
1085 pthread_mutex_unlock (&lock);
1087 if (status != 0)
1088 return (status);
1090 if (res->status <= 0)
1091 {
1092 response_free (res);
1093 return (EIO);
1094 }
1096 head = NULL;
1097 tail = NULL;
1098 for (i = 0; i < res->lines_num; i++)
1099 {
1100 char *key;
1101 char *value;
1102 char *endptr;
1103 rrdc_stats_t *s;
1105 key = res->lines[i];
1106 value = strchr (key, ':');
1107 if (value == NULL)
1108 continue;
1109 *value = 0;
1110 value++;
1112 while ((value[0] == ' ') || (value[0] == '\t'))
1113 value++;
1115 s = (rrdc_stats_t *) malloc (sizeof (rrdc_stats_t));
1116 if (s == NULL)
1117 continue;
1118 memset (s, 0, sizeof (*s));
1120 s->name = strdup (key);
1122 endptr = NULL;
1123 if ((strcmp ("QueueLength", key) == 0)
1124 || (strcmp ("TreeDepth", key) == 0)
1125 || (strcmp ("TreeNodesNumber", key) == 0))
1126 {
1127 s->type = RRDC_STATS_TYPE_GAUGE;
1128 s->value.gauge = strtod (value, &endptr);
1129 }
1130 else if ((strcmp ("DataSetsWritten", key) == 0)
1131 || (strcmp ("FlushesReceived", key) == 0)
1132 || (strcmp ("JournalBytes", key) == 0)
1133 || (strcmp ("JournalRotate", key) == 0)
1134 || (strcmp ("UpdatesReceived", key) == 0)
1135 || (strcmp ("UpdatesWritten", key) == 0))
1136 {
1137 s->type = RRDC_STATS_TYPE_COUNTER;
1138 s->value.counter = (uint64_t) strtoll (value, &endptr, /* base = */ 0);
1139 }
1140 else
1141 {
1142 free (s);
1143 continue;
1144 }
1146 /* Conversion failed */
1147 if (endptr == value)
1148 {
1149 free (s);
1150 continue;
1151 }
1153 if (head == NULL)
1154 {
1155 head = s;
1156 tail = s;
1157 s->next = NULL;
1158 }
1159 else
1160 {
1161 tail->next = s;
1162 tail = s;
1163 }
1164 } /* for (i = 0; i < res->lines_num; i++) */
1166 response_free (res);
1168 if (head == NULL)
1169 return (EPROTO);
1171 *ret_stats = head;
1172 return (0);
1173 } /* }}} int rrdc_stats_get */
1175 void rrdc_stats_free (rrdc_stats_t *ret_stats) /* {{{ */
1176 {
1177 rrdc_stats_t *this;
1179 this = ret_stats;
1180 while (this != NULL)
1181 {
1182 rrdc_stats_t *next;
1184 next = this->next;
1186 if (this->name != NULL)
1187 {
1188 free ((char *)this->name);
1189 this->name = NULL;
1190 }
1191 free (this);
1193 this = next;
1194 } /* while (this != NULL) */
1195 } /* }}} void rrdc_stats_free */
1197 /*
1198 * vim: set sw=2 sts=2 ts=8 et fdm=marker :
1199 */