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 ((path == NULL) || (resolved_path == NULL) || (sd_path == NULL))
77 return (NULL);
79 if ((*sd_path == '/')
80 || (strncmp ("unix:", sd_path, strlen ("unix:")) == 0))
81 is_unix = 1;
83 if (is_unix)
84 {
85 ret = realpath(path, resolved_path);
86 if (ret == NULL)
87 rrd_set_error("realpath(%s): %s", path, rrd_strerror(errno));
88 return ret;
89 }
90 else
91 {
92 if (*path == '/') /* not absolute path */
93 {
94 rrd_set_error ("absolute path names not allowed when talking "
95 "to a remote daemon");
96 return NULL;
97 }
98 }
100 return path;
101 } /* }}} char *get_path */
103 static size_t strsplit (char *string, char **fields, size_t size) /* {{{ */
104 {
105 size_t i;
106 char *ptr;
107 char *saveptr;
109 i = 0;
110 ptr = string;
111 saveptr = NULL;
112 while ((fields[i] = strtok_r (ptr, " \t\r\n", &saveptr)) != NULL)
113 {
114 ptr = NULL;
115 i++;
117 if (i >= size)
118 break;
119 }
121 return (i);
122 } /* }}} size_t strsplit */
124 static int parse_header (char *line, /* {{{ */
125 char **ret_key, char **ret_value)
126 {
127 char *tmp;
129 *ret_key = line;
131 tmp = strchr (line, ':');
132 if (tmp == NULL)
133 return (-1);
135 do
136 {
137 *tmp = 0;
138 tmp++;
139 }
140 while ((tmp[0] == ' ') || (tmp[0] == '\t'));
142 if (*tmp == 0)
143 return (-1);
145 *ret_value = tmp;
146 return (0);
147 } /* }}} int parse_header */
149 static int parse_ulong_header (char *line, /* {{{ */
150 char **ret_key, unsigned long *ret_value)
151 {
152 char *str_value;
153 char *endptr;
154 int status;
156 str_value = NULL;
157 status = parse_header (line, ret_key, &str_value);
158 if (status != 0)
159 return (status);
161 endptr = NULL;
162 errno = 0;
163 *ret_value = (unsigned long) strtol (str_value, &endptr, /* base = */ 0);
164 if ((endptr == str_value) || (errno != 0))
165 return (-1);
167 return (0);
168 } /* }}} int parse_ulong_header */
170 static int parse_char_array_header (char *line, /* {{{ */
171 char **ret_key, char **array, size_t array_len, int alloc)
172 {
173 char *tmp_array[array_len];
174 char *value;
175 size_t num;
176 int status;
178 value = NULL;
179 status = parse_header (line, ret_key, &value);
180 if (status != 0)
181 return (-1);
183 num = strsplit (value, tmp_array, array_len);
184 if (num != array_len)
185 return (-1);
187 if (alloc == 0)
188 {
189 memcpy (array, tmp_array, sizeof (tmp_array));
190 }
191 else
192 {
193 size_t i;
195 for (i = 0; i < array_len; i++)
196 array[i] = strdup (tmp_array[i]);
197 }
199 return (0);
200 } /* }}} int parse_char_array_header */
202 static int parse_value_array_header (char *line, /* {{{ */
203 time_t *ret_time, rrd_value_t *array, size_t array_len)
204 {
205 char *str_key;
206 char *str_array[array_len];
207 char *endptr;
208 int status;
209 size_t i;
211 str_key = NULL;
212 status = parse_char_array_header (line, &str_key,
213 str_array, array_len, /* alloc = */ 0);
214 if (status != 0)
215 return (-1);
217 errno = 0;
218 endptr = NULL;
219 *ret_time = (time_t) strtol (str_key, &endptr, /* base = */ 10);
220 if ((endptr == str_key) || (errno != 0))
221 return (-1);
223 for (i = 0; i < array_len; i++)
224 {
225 endptr = NULL;
226 array[i] = (rrd_value_t) strtod (str_array[i], &endptr);
227 if ((endptr == str_array[i]) || (errno != 0))
228 return (-1);
229 }
231 return (0);
232 } /* }}} int parse_value_array_header */
234 /* One must hold `lock' when calling `close_connection'. */
235 static void close_connection (void) /* {{{ */
236 {
237 if (sh != NULL)
238 {
239 fclose (sh);
240 sh = NULL;
241 sd = -1;
242 }
243 else if (sd >= 0)
244 {
245 close (sd);
246 sd = -1;
247 }
249 if (sd_path != NULL)
250 free (sd_path);
251 sd_path = NULL;
252 } /* }}} void close_connection */
254 static int buffer_add_string (const char *str, /* {{{ */
255 char **buffer_ret, size_t *buffer_size_ret)
256 {
257 char *buffer;
258 size_t buffer_size;
259 size_t buffer_pos;
260 size_t i;
261 int status;
263 buffer = *buffer_ret;
264 buffer_size = *buffer_size_ret;
265 buffer_pos = 0;
267 i = 0;
268 status = -1;
269 while (buffer_pos < buffer_size)
270 {
271 if (str[i] == 0)
272 {
273 buffer[buffer_pos] = ' ';
274 buffer_pos++;
275 status = 0;
276 break;
277 }
278 else if ((str[i] == ' ') || (str[i] == '\\'))
279 {
280 if (buffer_pos >= (buffer_size - 1))
281 break;
282 buffer[buffer_pos] = '\\';
283 buffer_pos++;
284 buffer[buffer_pos] = str[i];
285 buffer_pos++;
286 }
287 else
288 {
289 buffer[buffer_pos] = str[i];
290 buffer_pos++;
291 }
292 i++;
293 } /* while (buffer_pos < buffer_size) */
295 if (status != 0)
296 return (-1);
298 *buffer_ret = buffer + buffer_pos;
299 *buffer_size_ret = buffer_size - buffer_pos;
301 return (0);
302 } /* }}} int buffer_add_string */
304 static int buffer_add_value (const char *value, /* {{{ */
305 char **buffer_ret, size_t *buffer_size_ret)
306 {
307 char temp[4096];
309 if (strncmp (value, "N:", 2) == 0)
310 snprintf (temp, sizeof (temp), "%lu:%s",
311 (unsigned long) time (NULL), value + 2);
312 else
313 strncpy (temp, value, sizeof (temp));
314 temp[sizeof (temp) - 1] = 0;
316 return (buffer_add_string (temp, buffer_ret, buffer_size_ret));
317 } /* }}} int buffer_add_value */
319 /* Remove trailing newline (NL) and carriage return (CR) characters. Similar to
320 * the Perl function `chomp'. Returns the number of characters that have been
321 * removed. */
322 static int chomp (char *str) /* {{{ */
323 {
324 size_t len;
325 int removed;
327 if (str == NULL)
328 return (-1);
330 len = strlen (str);
331 removed = 0;
332 while ((len > 0) && ((str[len - 1] == '\n') || (str[len - 1] == '\r')))
333 {
334 str[len - 1] = 0;
335 len--;
336 removed++;
337 }
339 return (removed);
340 } /* }}} int chomp */
342 static void response_free (rrdc_response_t *res) /* {{{ */
343 {
344 if (res == NULL)
345 return;
347 if (res->lines != NULL)
348 {
349 size_t i;
351 for (i = 0; i < res->lines_num; i++)
352 if (res->lines[i] != NULL)
353 free (res->lines[i]);
354 free (res->lines);
355 }
357 free (res);
358 } /* }}} void response_free */
360 static int response_read (rrdc_response_t **ret_response) /* {{{ */
361 {
362 rrdc_response_t *ret;
364 char buffer[4096];
365 char *buffer_ptr;
367 size_t i;
369 if (sh == NULL)
370 return (-1);
372 ret = (rrdc_response_t *) malloc (sizeof (rrdc_response_t));
373 if (ret == NULL)
374 return (-2);
375 memset (ret, 0, sizeof (*ret));
376 ret->lines = NULL;
377 ret->lines_num = 0;
379 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
380 if (buffer_ptr == NULL) {
381 close_connection();
382 return (-3);
383 }
384 chomp (buffer);
386 ret->status = strtol (buffer, &ret->message, 0);
387 if (buffer == ret->message)
388 {
389 response_free (ret);
390 close_connection();
391 return (-4);
392 }
393 /* Skip leading whitespace of the status message */
394 ret->message += strspn (ret->message, " \t");
396 if (ret->status <= 0)
397 {
398 if (ret->status < 0)
399 rrd_set_error("rrdcached: %s", ret->message);
400 *ret_response = ret;
401 return (0);
402 }
404 ret->lines = (char **) malloc (sizeof (char *) * ret->status);
405 if (ret->lines == NULL)
406 {
407 response_free (ret);
408 close_connection();
409 return (-5);
410 }
411 memset (ret->lines, 0, sizeof (char *) * ret->status);
412 ret->lines_num = (size_t) ret->status;
414 for (i = 0; i < ret->lines_num; i++)
415 {
416 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
417 if (buffer_ptr == NULL)
418 {
419 response_free (ret);
420 close_connection();
421 return (-6);
422 }
423 chomp (buffer);
425 ret->lines[i] = strdup (buffer);
426 if (ret->lines[i] == NULL)
427 {
428 response_free (ret);
429 close_connection();
430 return (-7);
431 }
432 }
434 *ret_response = ret;
435 return (0);
436 } /* }}} rrdc_response_t *response_read */
438 static int request (const char *buffer, size_t buffer_size, /* {{{ */
439 rrdc_response_t **ret_response)
440 {
441 int status;
442 rrdc_response_t *res;
444 if (sh == NULL)
445 return (ENOTCONN);
447 status = (int) fwrite (buffer, buffer_size, /* nmemb = */ 1, sh);
448 if (status != 1)
449 {
450 close_connection ();
451 rrd_set_error("request: socket error (%d) while talking to rrdcached",
452 status);
453 return (-1);
454 }
455 fflush (sh);
457 res = NULL;
458 status = response_read (&res);
460 if (status != 0)
461 {
462 if (status < 0)
463 rrd_set_error("request: internal error while talking to rrdcached");
464 return (status);
465 }
467 *ret_response = res;
468 return (0);
469 } /* }}} int request */
471 /* determine whether we are connected to the specified daemon_addr if
472 * NULL, return whether we are connected at all
473 */
474 int rrdc_is_connected(const char *daemon_addr) /* {{{ */
475 {
476 if (sd < 0)
477 return 0;
478 else if (daemon_addr == NULL)
479 {
480 /* here we have to handle the case i.e.
481 * UPDATE --daemon ...; UPDATEV (no --daemon) ...
482 * In other words: we have a cached connection,
483 * but it is not specified in the current command.
484 * Daemon is only implied in this case if set in ENV
485 */
486 if (getenv(ENV_RRDCACHED_ADDRESS) != NULL)
487 return 1;
488 else
489 return 0;
490 }
491 else if (strcmp(daemon_addr, sd_path) == 0)
492 return 1;
493 else
494 return 0;
496 } /* }}} int rrdc_is_connected */
498 static int rrdc_connect_unix (const char *path) /* {{{ */
499 {
500 struct sockaddr_un sa;
501 int status;
503 assert (path != NULL);
504 assert (sd == -1);
506 sd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
507 if (sd < 0)
508 {
509 status = errno;
510 return (status);
511 }
513 memset (&sa, 0, sizeof (sa));
514 sa.sun_family = AF_UNIX;
515 strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
517 status = connect (sd, (struct sockaddr *) &sa, sizeof (sa));
518 if (status != 0)
519 {
520 status = errno;
521 close_connection ();
522 return (status);
523 }
525 sh = fdopen (sd, "r+");
526 if (sh == NULL)
527 {
528 status = errno;
529 close_connection ();
530 return (status);
531 }
533 return (0);
534 } /* }}} int rrdc_connect_unix */
536 static int rrdc_connect_network (const char *addr_orig) /* {{{ */
537 {
538 struct addrinfo ai_hints;
539 struct addrinfo *ai_res;
540 struct addrinfo *ai_ptr;
541 char addr_copy[NI_MAXHOST];
542 char *addr;
543 char *port;
545 assert (addr_orig != NULL);
546 assert (sd == -1);
548 strncpy(addr_copy, addr_orig, sizeof(addr_copy));
549 addr_copy[sizeof(addr_copy) - 1] = '\0';
550 addr = addr_copy;
552 int status;
553 memset (&ai_hints, 0, sizeof (ai_hints));
554 ai_hints.ai_flags = 0;
555 #ifdef AI_ADDRCONFIG
556 ai_hints.ai_flags |= AI_ADDRCONFIG;
557 #endif
558 ai_hints.ai_family = AF_UNSPEC;
559 ai_hints.ai_socktype = SOCK_STREAM;
561 port = NULL;
562 if (*addr == '[') /* IPv6+port format */
563 {
564 /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
565 addr++;
567 port = strchr (addr, ']');
568 if (port == NULL)
569 {
570 rrd_set_error("malformed address: %s", addr_orig);
571 return (-1);
572 }
573 *port = 0;
574 port++;
576 if (*port == ':')
577 port++;
578 else if (*port == 0)
579 port = NULL;
580 else
581 {
582 rrd_set_error("garbage after address: %s", port);
583 return (-1);
584 }
585 } /* if (*addr == '[') */
586 else
587 {
588 port = rindex(addr, ':');
589 if (port != NULL)
590 {
591 *port = 0;
592 port++;
593 }
594 }
596 ai_res = NULL;
597 status = getaddrinfo (addr,
598 port == NULL ? RRDCACHED_DEFAULT_PORT : port,
599 &ai_hints, &ai_res);
600 if (status != 0)
601 {
602 rrd_set_error ("failed to resolve address `%s' (port %s): %s",
603 addr, port == NULL ? RRDCACHED_DEFAULT_PORT : port,
604 gai_strerror (status));
605 return (-1);
606 }
608 for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
609 {
610 sd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
611 if (sd < 0)
612 {
613 status = errno;
614 sd = -1;
615 continue;
616 }
618 status = connect (sd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
619 if (status != 0)
620 {
621 status = errno;
622 close_connection();
623 continue;
624 }
626 sh = fdopen (sd, "r+");
627 if (sh == NULL)
628 {
629 status = errno;
630 close_connection ();
631 continue;
632 }
634 assert (status == 0);
635 break;
636 } /* for (ai_ptr) */
638 return (status);
639 } /* }}} int rrdc_connect_network */
641 int rrdc_connect (const char *addr) /* {{{ */
642 {
643 int status = 0;
645 if (addr == NULL)
646 addr = getenv (ENV_RRDCACHED_ADDRESS);
648 if (addr == NULL)
649 return 0;
651 pthread_mutex_lock(&lock);
653 if (sd >= 0 && sd_path != NULL && strcmp(addr, sd_path) == 0)
654 {
655 /* connection to the same daemon; use cached connection */
656 pthread_mutex_unlock (&lock);
657 return (0);
658 }
659 else
660 {
661 close_connection();
662 }
664 rrd_clear_error ();
665 if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
666 status = rrdc_connect_unix (addr + strlen ("unix:"));
667 else if (addr[0] == '/')
668 status = rrdc_connect_unix (addr);
669 else
670 status = rrdc_connect_network(addr);
672 if (status == 0 && sd >= 0)
673 sd_path = strdup(addr);
674 else
675 {
676 char *err = rrd_test_error () ? rrd_get_error () : "Internal error";
677 /* err points the string that gets written to by rrd_set_error(), thus we
678 * cannot pass it to that function */
679 err = strdup (err);
680 rrd_set_error("Unable to connect to rrdcached: %s",
681 (status < 0)
682 ? (err ? err : "Internal error")
683 : rrd_strerror (status));
684 if (err != NULL)
685 free (err);
686 }
688 pthread_mutex_unlock (&lock);
689 return (status);
690 } /* }}} int rrdc_connect */
692 int rrdc_disconnect (void) /* {{{ */
693 {
694 pthread_mutex_lock (&lock);
696 close_connection();
698 pthread_mutex_unlock (&lock);
700 return (0);
701 } /* }}} int rrdc_disconnect */
703 int rrdc_update (const char *filename, int values_num, /* {{{ */
704 const char * const *values)
705 {
706 char buffer[4096];
707 char *buffer_ptr;
708 size_t buffer_free;
709 size_t buffer_size;
710 rrdc_response_t *res;
711 int status;
712 int i;
713 char file_path[PATH_MAX];
715 memset (buffer, 0, sizeof (buffer));
716 buffer_ptr = &buffer[0];
717 buffer_free = sizeof (buffer);
719 status = buffer_add_string ("update", &buffer_ptr, &buffer_free);
720 if (status != 0)
721 return (ENOBUFS);
723 pthread_mutex_lock (&lock);
724 filename = get_path (filename, file_path);
725 if (filename == NULL)
726 {
727 pthread_mutex_unlock (&lock);
728 return (-1);
729 }
731 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
732 if (status != 0)
733 {
734 pthread_mutex_unlock (&lock);
735 return (ENOBUFS);
736 }
738 for (i = 0; i < values_num; i++)
739 {
740 status = buffer_add_value (values[i], &buffer_ptr, &buffer_free);
741 if (status != 0)
742 {
743 pthread_mutex_unlock (&lock);
744 return (ENOBUFS);
745 }
746 }
748 assert (buffer_free < sizeof (buffer));
749 buffer_size = sizeof (buffer) - buffer_free;
750 assert (buffer[buffer_size - 1] == ' ');
751 buffer[buffer_size - 1] = '\n';
753 res = NULL;
754 status = request (buffer, buffer_size, &res);
755 pthread_mutex_unlock (&lock);
757 if (status != 0)
758 return (status);
760 status = res->status;
761 response_free (res);
763 return (status);
764 } /* }}} int rrdc_update */
766 int rrdc_flush (const char *filename) /* {{{ */
767 {
768 char buffer[4096];
769 char *buffer_ptr;
770 size_t buffer_free;
771 size_t buffer_size;
772 rrdc_response_t *res;
773 int status;
774 char file_path[PATH_MAX];
776 if (filename == NULL)
777 return (-1);
779 memset (buffer, 0, sizeof (buffer));
780 buffer_ptr = &buffer[0];
781 buffer_free = sizeof (buffer);
783 status = buffer_add_string ("flush", &buffer_ptr, &buffer_free);
784 if (status != 0)
785 return (ENOBUFS);
787 pthread_mutex_lock (&lock);
788 filename = get_path (filename, file_path);
789 if (filename == NULL)
790 {
791 pthread_mutex_unlock (&lock);
792 return (-1);
793 }
795 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
796 if (status != 0)
797 {
798 pthread_mutex_unlock (&lock);
799 return (ENOBUFS);
800 }
802 assert (buffer_free < sizeof (buffer));
803 buffer_size = sizeof (buffer) - buffer_free;
804 assert (buffer[buffer_size - 1] == ' ');
805 buffer[buffer_size - 1] = '\n';
807 res = NULL;
808 status = request (buffer, buffer_size, &res);
809 pthread_mutex_unlock (&lock);
811 if (status != 0)
812 return (status);
814 status = res->status;
815 response_free (res);
817 return (status);
818 } /* }}} int rrdc_flush */
820 int rrdc_fetch (const char *filename, /* {{{ */
821 const char *cf,
822 time_t *ret_start, time_t *ret_end,
823 unsigned long *ret_step,
824 unsigned long *ret_ds_num,
825 char ***ret_ds_names,
826 rrd_value_t **ret_data)
827 {
828 char buffer[4096];
829 char *buffer_ptr;
830 size_t buffer_free;
831 size_t buffer_size;
832 rrdc_response_t *res;
833 char path_buffer[PATH_MAX];
834 const char *path_ptr;
836 char *str_tmp;
837 unsigned long flush_version;
839 time_t start;
840 time_t end;
841 unsigned long step;
842 unsigned long ds_num;
843 char **ds_names;
845 rrd_value_t *data;
846 size_t data_size;
847 size_t data_fill;
849 int status;
850 size_t current_line;
851 time_t t;
853 if ((filename == NULL) || (cf == NULL))
854 return (-1);
856 /* Send request {{{ */
857 memset (buffer, 0, sizeof (buffer));
858 buffer_ptr = &buffer[0];
859 buffer_free = sizeof (buffer);
861 status = buffer_add_string ("FETCH", &buffer_ptr, &buffer_free);
862 if (status != 0)
863 return (ENOBUFS);
865 /* change to path for rrdcached */
866 path_ptr = get_path (filename, path_buffer);
867 if (path_ptr == NULL)
868 return (EINVAL);
870 status = buffer_add_string (path_ptr, &buffer_ptr, &buffer_free);
871 if (status != 0)
872 return (ENOBUFS);
874 status = buffer_add_string (cf, &buffer_ptr, &buffer_free);
875 if (status != 0)
876 return (ENOBUFS);
878 if ((ret_start != NULL) && (*ret_start > 0))
879 {
880 char tmp[64];
881 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_start);
882 tmp[sizeof (tmp) - 1] = 0;
883 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
884 if (status != 0)
885 return (ENOBUFS);
887 if ((ret_end != NULL) && (*ret_end > 0))
888 {
889 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_end);
890 tmp[sizeof (tmp) - 1] = 0;
891 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
892 if (status != 0)
893 return (ENOBUFS);
894 }
895 }
897 assert (buffer_free < sizeof (buffer));
898 buffer_size = sizeof (buffer) - buffer_free;
899 assert (buffer[buffer_size - 1] == ' ');
900 buffer[buffer_size - 1] = '\n';
902 res = NULL;
903 status = request (buffer, buffer_size, &res);
904 if (status != 0)
905 return (status);
907 status = res->status;
908 if (status < 0)
909 {
910 rrd_set_error ("rrdcached: %s", res->message);
911 response_free (res);
912 return (status);
913 }
914 /* }}} Send request */
916 ds_names = NULL;
917 ds_num = 0;
918 data = NULL;
919 current_line = 0;
921 /* Macros to make error handling a little easier (i. e. less to type and
922 * read. `BAIL_OUT' sets the error message, frees all dynamically allocated
923 * variables and returns the provided status code. */
924 #define BAIL_OUT(status, ...) do { \
925 rrd_set_error ("rrdc_fetch: " __VA_ARGS__); \
926 free (data); \
927 if (ds_names != 0) { size_t k; for (k = 0; k < ds_num; k++) free (ds_names[k]); } \
928 free (ds_names); \
929 response_free (res); \
930 return (status); \
931 } while (0)
933 #define READ_NUMERIC_FIELD(name,type,var) do { \
934 char *key; \
935 unsigned long value; \
936 assert (current_line < res->lines_num); \
937 status = parse_ulong_header (res->lines[current_line], &key, &value); \
938 if (status != 0) \
939 BAIL_OUT (-1, "Unable to parse header `%s'", name); \
940 if (strcasecmp (key, name) != 0) \
941 BAIL_OUT (-1, "Unexpected header line: Expected `%s', got `%s'", name, key); \
942 var = (type) value; \
943 current_line++; \
944 } while (0)
946 if (res->lines_num < 1)
947 BAIL_OUT (-1, "Premature end of response packet");
949 /* We're making some very strong assumptions about the fields below. We
950 * therefore check the version of the `flush' command first, so that later
951 * versions can change the order of fields and it's easier to implement
952 * backwards compatibility. */
953 READ_NUMERIC_FIELD ("FlushVersion", unsigned long, flush_version);
954 if (flush_version != 1)
955 BAIL_OUT (-1, "Don't know how to handle flush format version %lu.",
956 flush_version);
958 if (res->lines_num < 5)
959 BAIL_OUT (-1, "Premature end of response packet");
961 READ_NUMERIC_FIELD ("Start", time_t, start);
962 READ_NUMERIC_FIELD ("End", time_t, end);
963 if (start >= end)
964 BAIL_OUT (-1, "Malformed start and end times: start = %lu; end = %lu;",
965 (unsigned long) start,
966 (unsigned long) end);
968 READ_NUMERIC_FIELD ("Step", unsigned long, step);
969 if (step < 1)
970 BAIL_OUT (-1, "Invalid number for Step: %lu", step);
972 READ_NUMERIC_FIELD ("DSCount", unsigned long, ds_num);
973 if (ds_num < 1)
974 BAIL_OUT (-1, "Invalid number for DSCount: %lu", ds_num);
976 /* It's time to allocate some memory */
977 ds_names = calloc ((size_t) ds_num, sizeof (*ds_names));
978 if (ds_names == NULL)
979 BAIL_OUT (-1, "Out of memory");
981 status = parse_char_array_header (res->lines[current_line],
982 &str_tmp, ds_names, (size_t) ds_num, /* alloc = */ 1);
983 if (status != 0)
984 BAIL_OUT (-1, "Unable to parse header `DSName'");
985 if (strcasecmp ("DSName", str_tmp) != 0)
986 BAIL_OUT (-1, "Unexpected header line: Expected `DSName', got `%s'", str_tmp);
987 current_line++;
989 data_size = ds_num * (end - start) / step;
990 if (data_size < 1)
991 BAIL_OUT (-1, "No data returned or headers invalid.");
993 if (res->lines_num != (6 + (data_size / ds_num)))
994 BAIL_OUT (-1, "Got %zu lines, expected %zu",
995 res->lines_num, (6 + (data_size / ds_num)));
997 data = calloc (data_size, sizeof (*data));
998 if (data == NULL)
999 BAIL_OUT (-1, "Out of memory");
1002 data_fill = 0;
1003 for (t = start + step; t <= end; t += step, current_line++)
1004 {
1005 time_t tmp;
1007 assert (current_line < res->lines_num);
1009 status = parse_value_array_header (res->lines[current_line],
1010 &tmp, data + data_fill, (size_t) ds_num);
1011 if (status != 0)
1012 BAIL_OUT (-1, "Cannot parse value line");
1014 data_fill += (size_t) ds_num;
1015 }
1017 *ret_start = start;
1018 *ret_end = end;
1019 *ret_step = step;
1020 *ret_ds_num = ds_num;
1021 *ret_ds_names = ds_names;
1022 *ret_data = data;
1024 response_free (res);
1025 return (0);
1026 #undef READ_NUMERIC_FIELD
1027 #undef BAIL_OUT
1028 } /* }}} int rrdc_flush */
1030 /* convenience function; if there is a daemon specified, or if we can
1031 * detect one from the environment, then flush the file. Otherwise, no-op
1032 */
1033 int rrdc_flush_if_daemon (const char *opt_daemon, const char *filename) /* {{{ */
1034 {
1035 int status = 0;
1037 rrdc_connect(opt_daemon);
1039 if (rrdc_is_connected(opt_daemon))
1040 {
1041 rrd_clear_error();
1042 status = rrdc_flush (filename);
1044 if (status != 0 && !rrd_test_error())
1045 {
1046 if (status > 0)
1047 {
1048 rrd_set_error("rrdc_flush (%s) failed: %s",
1049 filename, rrd_strerror(status));
1050 }
1051 else if (status < 0)
1052 {
1053 rrd_set_error("rrdc_flush (%s) failed with status %i.",
1054 filename, status);
1055 }
1056 }
1057 } /* if (rrdc_is_connected(..)) */
1059 return status;
1060 } /* }}} int rrdc_flush_if_daemon */
1063 int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */
1064 {
1065 rrdc_stats_t *head;
1066 rrdc_stats_t *tail;
1068 rrdc_response_t *res;
1070 int status;
1071 size_t i;
1073 /* Protocol example: {{{
1074 * -> STATS
1075 * <- 5 Statistics follow
1076 * <- QueueLength: 0
1077 * <- UpdatesWritten: 0
1078 * <- DataSetsWritten: 0
1079 * <- TreeNodesNumber: 0
1080 * <- TreeDepth: 0
1081 * }}} */
1083 res = NULL;
1084 pthread_mutex_lock (&lock);
1085 status = request ("STATS\n", strlen ("STATS\n"), &res);
1086 pthread_mutex_unlock (&lock);
1088 if (status != 0)
1089 return (status);
1091 if (res->status <= 0)
1092 {
1093 response_free (res);
1094 return (EIO);
1095 }
1097 head = NULL;
1098 tail = NULL;
1099 for (i = 0; i < res->lines_num; i++)
1100 {
1101 char *key;
1102 char *value;
1103 char *endptr;
1104 rrdc_stats_t *s;
1106 key = res->lines[i];
1107 value = strchr (key, ':');
1108 if (value == NULL)
1109 continue;
1110 *value = 0;
1111 value++;
1113 while ((value[0] == ' ') || (value[0] == '\t'))
1114 value++;
1116 s = (rrdc_stats_t *) malloc (sizeof (rrdc_stats_t));
1117 if (s == NULL)
1118 continue;
1119 memset (s, 0, sizeof (*s));
1121 s->name = strdup (key);
1123 endptr = NULL;
1124 if ((strcmp ("QueueLength", key) == 0)
1125 || (strcmp ("TreeDepth", key) == 0)
1126 || (strcmp ("TreeNodesNumber", key) == 0))
1127 {
1128 s->type = RRDC_STATS_TYPE_GAUGE;
1129 s->value.gauge = strtod (value, &endptr);
1130 }
1131 else if ((strcmp ("DataSetsWritten", key) == 0)
1132 || (strcmp ("FlushesReceived", key) == 0)
1133 || (strcmp ("JournalBytes", key) == 0)
1134 || (strcmp ("JournalRotate", key) == 0)
1135 || (strcmp ("UpdatesReceived", key) == 0)
1136 || (strcmp ("UpdatesWritten", key) == 0))
1137 {
1138 s->type = RRDC_STATS_TYPE_COUNTER;
1139 s->value.counter = (uint64_t) strtoll (value, &endptr, /* base = */ 0);
1140 }
1141 else
1142 {
1143 free (s);
1144 continue;
1145 }
1147 /* Conversion failed */
1148 if (endptr == value)
1149 {
1150 free (s);
1151 continue;
1152 }
1154 if (head == NULL)
1155 {
1156 head = s;
1157 tail = s;
1158 s->next = NULL;
1159 }
1160 else
1161 {
1162 tail->next = s;
1163 tail = s;
1164 }
1165 } /* for (i = 0; i < res->lines_num; i++) */
1167 response_free (res);
1169 if (head == NULL)
1170 return (EPROTO);
1172 *ret_stats = head;
1173 return (0);
1174 } /* }}} int rrdc_stats_get */
1176 void rrdc_stats_free (rrdc_stats_t *ret_stats) /* {{{ */
1177 {
1178 rrdc_stats_t *this;
1180 this = ret_stats;
1181 while (this != NULL)
1182 {
1183 rrdc_stats_t *next;
1185 next = this->next;
1187 if (this->name != NULL)
1188 {
1189 free ((char *)this->name);
1190 this->name = NULL;
1191 }
1192 free (this);
1194 this = next;
1195 } /* while (this != NULL) */
1196 } /* }}} void rrdc_stats_free */
1198 /*
1199 * vim: set sw=2 sts=2 ts=8 et fdm=marker :
1200 */