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 static int buffer_add_ulong (const unsigned long value, /* {{{ */
320 char **buffer_ret, size_t *buffer_size_ret)
321 {
322 char temp[4096];
324 snprintf (temp, sizeof (temp), "%lu", value);
325 temp[sizeof (temp) - 1] = 0;
326 return (buffer_add_string (temp, buffer_ret, buffer_size_ret));
327 } /* }}} int buffer_add_ulong */
329 /* Remove trailing newline (NL) and carriage return (CR) characters. Similar to
330 * the Perl function `chomp'. Returns the number of characters that have been
331 * removed. */
332 static int chomp (char *str) /* {{{ */
333 {
334 size_t len;
335 int removed;
337 if (str == NULL)
338 return (-1);
340 len = strlen (str);
341 removed = 0;
342 while ((len > 0) && ((str[len - 1] == '\n') || (str[len - 1] == '\r')))
343 {
344 str[len - 1] = 0;
345 len--;
346 removed++;
347 }
349 return (removed);
350 } /* }}} int chomp */
352 static void response_free (rrdc_response_t *res) /* {{{ */
353 {
354 if (res == NULL)
355 return;
357 if (res->lines != NULL)
358 {
359 size_t i;
361 for (i = 0; i < res->lines_num; i++)
362 if (res->lines[i] != NULL)
363 free (res->lines[i]);
364 free (res->lines);
365 }
367 free (res);
368 } /* }}} void response_free */
370 static int response_read (rrdc_response_t **ret_response) /* {{{ */
371 {
372 rrdc_response_t *ret = NULL;
373 int status = 0;
375 char buffer[4096];
376 char *buffer_ptr;
378 size_t i;
380 #define DIE(code) do { status = code; goto err_out; } while(0)
382 if (sh == NULL)
383 DIE(-1);
385 ret = (rrdc_response_t *) malloc (sizeof (rrdc_response_t));
386 if (ret == NULL)
387 DIE(-2);
388 memset (ret, 0, sizeof (*ret));
389 ret->lines = NULL;
390 ret->lines_num = 0;
392 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
393 if (buffer_ptr == NULL)
394 DIE(-3);
396 chomp (buffer);
398 ret->status = strtol (buffer, &ret->message, 0);
399 if (buffer == ret->message)
400 DIE(-4);
402 /* Skip leading whitespace of the status message */
403 ret->message += strspn (ret->message, " \t");
405 if (ret->status <= 0)
406 {
407 if (ret->status < 0)
408 rrd_set_error("rrdcached: %s", ret->message);
409 goto out;
410 }
412 ret->lines = (char **) malloc (sizeof (char *) * ret->status);
413 if (ret->lines == NULL)
414 DIE(-5);
416 memset (ret->lines, 0, sizeof (char *) * ret->status);
417 ret->lines_num = (size_t) ret->status;
419 for (i = 0; i < ret->lines_num; i++)
420 {
421 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
422 if (buffer_ptr == NULL)
423 DIE(-6);
425 chomp (buffer);
427 ret->lines[i] = strdup (buffer);
428 if (ret->lines[i] == NULL)
429 DIE(-7);
430 }
432 out:
433 *ret_response = ret;
434 fflush(sh);
435 return (status);
437 err_out:
438 response_free(ret);
439 close_connection();
440 return (status);
442 #undef DIE
444 } /* }}} rrdc_response_t *response_read */
446 static int request (const char *buffer, size_t buffer_size, /* {{{ */
447 rrdc_response_t **ret_response)
448 {
449 int status;
450 rrdc_response_t *res;
452 if (sh == NULL)
453 return (ENOTCONN);
455 status = (int) fwrite (buffer, buffer_size, /* nmemb = */ 1, sh);
456 if (status != 1)
457 {
458 close_connection ();
459 rrd_set_error("request: socket error (%d) while talking to rrdcached",
460 status);
461 return (-1);
462 }
463 fflush (sh);
465 res = NULL;
466 status = response_read (&res);
468 if (status != 0)
469 {
470 if (status < 0)
471 rrd_set_error("request: internal error while talking to rrdcached");
472 return (status);
473 }
475 *ret_response = res;
476 return (0);
477 } /* }}} int request */
479 /* determine whether we are connected to the specified daemon_addr if
480 * NULL, return whether we are connected at all
481 */
482 int rrdc_is_connected(const char *daemon_addr) /* {{{ */
483 {
484 if (sd < 0)
485 return 0;
486 else if (daemon_addr == NULL)
487 {
488 /* here we have to handle the case i.e.
489 * UPDATE --daemon ...; UPDATEV (no --daemon) ...
490 * In other words: we have a cached connection,
491 * but it is not specified in the current command.
492 * Daemon is only implied in this case if set in ENV
493 */
494 if (getenv(ENV_RRDCACHED_ADDRESS) != NULL)
495 return 1;
496 else
497 return 0;
498 }
499 else if (strcmp(daemon_addr, sd_path) == 0)
500 return 1;
501 else
502 return 0;
504 } /* }}} int rrdc_is_connected */
506 static int rrdc_connect_unix (const char *path) /* {{{ */
507 {
508 struct sockaddr_un sa;
509 int status;
511 assert (path != NULL);
512 assert (sd == -1);
514 sd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
515 if (sd < 0)
516 {
517 status = errno;
518 return (status);
519 }
521 memset (&sa, 0, sizeof (sa));
522 sa.sun_family = AF_UNIX;
523 strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
525 status = connect (sd, (struct sockaddr *) &sa, sizeof (sa));
526 if (status != 0)
527 {
528 status = errno;
529 close_connection ();
530 return (status);
531 }
533 sh = fdopen (sd, "r+");
534 if (sh == NULL)
535 {
536 status = errno;
537 close_connection ();
538 return (status);
539 }
541 return (0);
542 } /* }}} int rrdc_connect_unix */
544 static int rrdc_connect_network (const char *addr_orig) /* {{{ */
545 {
546 struct addrinfo ai_hints;
547 struct addrinfo *ai_res;
548 struct addrinfo *ai_ptr;
549 char addr_copy[NI_MAXHOST];
550 char *addr;
551 char *port;
553 assert (addr_orig != NULL);
554 assert (sd == -1);
556 strncpy(addr_copy, addr_orig, sizeof(addr_copy));
557 addr_copy[sizeof(addr_copy) - 1] = '\0';
558 addr = addr_copy;
560 int status;
561 memset (&ai_hints, 0, sizeof (ai_hints));
562 ai_hints.ai_flags = 0;
563 #ifdef AI_ADDRCONFIG
564 ai_hints.ai_flags |= AI_ADDRCONFIG;
565 #endif
566 ai_hints.ai_family = AF_UNSPEC;
567 ai_hints.ai_socktype = SOCK_STREAM;
569 port = NULL;
570 if (*addr == '[') /* IPv6+port format */
571 {
572 /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
573 addr++;
575 port = strchr (addr, ']');
576 if (port == NULL)
577 {
578 rrd_set_error("malformed address: %s", addr_orig);
579 return (-1);
580 }
581 *port = 0;
582 port++;
584 if (*port == ':')
585 port++;
586 else if (*port == 0)
587 port = NULL;
588 else
589 {
590 rrd_set_error("garbage after address: %s", port);
591 return (-1);
592 }
593 } /* if (*addr == '[') */
594 else
595 {
596 port = rindex(addr, ':');
597 if (port != NULL)
598 {
599 *port = 0;
600 port++;
601 }
602 }
604 ai_res = NULL;
605 status = getaddrinfo (addr,
606 port == NULL ? RRDCACHED_DEFAULT_PORT : port,
607 &ai_hints, &ai_res);
608 if (status != 0)
609 {
610 rrd_set_error ("failed to resolve address `%s' (port %s): %s",
611 addr, port == NULL ? RRDCACHED_DEFAULT_PORT : port,
612 gai_strerror (status));
613 return (-1);
614 }
616 for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
617 {
618 sd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
619 if (sd < 0)
620 {
621 status = errno;
622 sd = -1;
623 continue;
624 }
626 status = connect (sd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
627 if (status != 0)
628 {
629 status = errno;
630 close_connection();
631 continue;
632 }
634 sh = fdopen (sd, "r+");
635 if (sh == NULL)
636 {
637 status = errno;
638 close_connection ();
639 continue;
640 }
642 assert (status == 0);
643 break;
644 } /* for (ai_ptr) */
646 return (status);
647 } /* }}} int rrdc_connect_network */
649 int rrdc_connect (const char *addr) /* {{{ */
650 {
651 int status = 0;
653 if (addr == NULL)
654 addr = getenv (ENV_RRDCACHED_ADDRESS);
656 if (addr == NULL)
657 return 0;
659 pthread_mutex_lock(&lock);
661 if (sd >= 0 && sd_path != NULL && strcmp(addr, sd_path) == 0)
662 {
663 /* connection to the same daemon; use cached connection */
664 pthread_mutex_unlock (&lock);
665 return (0);
666 }
667 else
668 {
669 close_connection();
670 }
672 rrd_clear_error ();
673 if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
674 status = rrdc_connect_unix (addr + strlen ("unix:"));
675 else if (addr[0] == '/')
676 status = rrdc_connect_unix (addr);
677 else
678 status = rrdc_connect_network(addr);
680 if (status == 0 && sd >= 0)
681 sd_path = strdup(addr);
682 else
683 {
684 char *err = rrd_test_error () ? rrd_get_error () : "Internal error";
685 /* err points the string that gets written to by rrd_set_error(), thus we
686 * cannot pass it to that function */
687 err = strdup (err);
688 rrd_set_error("Unable to connect to rrdcached: %s",
689 (status < 0)
690 ? (err ? err : "Internal error")
691 : rrd_strerror (status));
692 if (err != NULL)
693 free (err);
694 }
696 pthread_mutex_unlock (&lock);
697 return (status);
698 } /* }}} int rrdc_connect */
700 int rrdc_disconnect (void) /* {{{ */
701 {
702 pthread_mutex_lock (&lock);
704 close_connection();
706 pthread_mutex_unlock (&lock);
708 return (0);
709 } /* }}} int rrdc_disconnect */
711 int rrdc_update (const char *filename, int values_num, /* {{{ */
712 const char * const *values)
713 {
714 char buffer[4096];
715 char *buffer_ptr;
716 size_t buffer_free;
717 size_t buffer_size;
718 rrdc_response_t *res;
719 int status;
720 int i;
721 char file_path[PATH_MAX];
723 memset (buffer, 0, sizeof (buffer));
724 buffer_ptr = &buffer[0];
725 buffer_free = sizeof (buffer);
727 status = buffer_add_string ("update", &buffer_ptr, &buffer_free);
728 if (status != 0)
729 return (ENOBUFS);
731 pthread_mutex_lock (&lock);
732 filename = get_path (filename, file_path);
733 if (filename == NULL)
734 {
735 pthread_mutex_unlock (&lock);
736 return (-1);
737 }
739 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
740 if (status != 0)
741 {
742 pthread_mutex_unlock (&lock);
743 return (ENOBUFS);
744 }
746 for (i = 0; i < values_num; i++)
747 {
748 status = buffer_add_value (values[i], &buffer_ptr, &buffer_free);
749 if (status != 0)
750 {
751 pthread_mutex_unlock (&lock);
752 return (ENOBUFS);
753 }
754 }
756 assert (buffer_free < sizeof (buffer));
757 buffer_size = sizeof (buffer) - buffer_free;
758 assert (buffer[buffer_size - 1] == ' ');
759 buffer[buffer_size - 1] = '\n';
761 res = NULL;
762 status = request (buffer, buffer_size, &res);
763 pthread_mutex_unlock (&lock);
765 if (status != 0)
766 return (status);
768 status = res->status;
769 response_free (res);
771 return (status);
772 } /* }}} int rrdc_update */
774 int rrdc_flush (const char *filename) /* {{{ */
775 {
776 char buffer[4096];
777 char *buffer_ptr;
778 size_t buffer_free;
779 size_t buffer_size;
780 rrdc_response_t *res;
781 int status;
782 char file_path[PATH_MAX];
784 if (filename == NULL)
785 return (-1);
787 memset (buffer, 0, sizeof (buffer));
788 buffer_ptr = &buffer[0];
789 buffer_free = sizeof (buffer);
791 status = buffer_add_string ("flush", &buffer_ptr, &buffer_free);
792 if (status != 0)
793 return (ENOBUFS);
795 pthread_mutex_lock (&lock);
796 filename = get_path (filename, file_path);
797 if (filename == NULL)
798 {
799 pthread_mutex_unlock (&lock);
800 return (-1);
801 }
803 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
804 if (status != 0)
805 {
806 pthread_mutex_unlock (&lock);
807 return (ENOBUFS);
808 }
810 assert (buffer_free < sizeof (buffer));
811 buffer_size = sizeof (buffer) - buffer_free;
812 assert (buffer[buffer_size - 1] == ' ');
813 buffer[buffer_size - 1] = '\n';
815 res = NULL;
816 status = request (buffer, buffer_size, &res);
817 pthread_mutex_unlock (&lock);
819 if (status != 0)
820 return (status);
822 status = res->status;
823 response_free (res);
825 return (status);
826 } /* }}} int rrdc_flush */
828 rrd_info_t * rrdc_info (const char *filename) /* {{{ */
829 {
830 char buffer[4096];
831 char *buffer_ptr;
832 size_t buffer_free;
833 size_t buffer_size;
834 rrdc_response_t *res;
835 int status;
836 char file_path[PATH_MAX];
837 rrd_info_t *data = NULL, *cd;
838 rrd_infoval_t info;
839 unsigned int l;
840 rrd_info_type_t itype;
841 char *k, *s;
843 if (filename == NULL) {
844 rrd_set_error ("rrdc_info: no filename");
845 return (NULL);
846 }
848 memset (buffer, 0, sizeof (buffer));
849 buffer_ptr = &buffer[0];
850 buffer_free = sizeof (buffer);
852 status = buffer_add_string ("info", &buffer_ptr, &buffer_free);
853 if (status != 0) {
854 rrd_set_error ("rrdc_info: out of memory");
855 return (NULL);
856 }
858 pthread_mutex_lock (&lock);
859 filename = get_path (filename, file_path);
860 if (filename == NULL)
861 {
862 pthread_mutex_unlock (&lock);
863 return (NULL);
864 }
866 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
867 if (status != 0)
868 {
869 pthread_mutex_unlock (&lock);
870 rrd_set_error ("rrdc_info: out of memory");
871 return (NULL);
872 }
874 assert (buffer_free < sizeof (buffer));
875 buffer_size = sizeof (buffer) - buffer_free;
876 assert (buffer[buffer_size - 1] == ' ');
877 buffer[buffer_size - 1] = '\n';
879 res = NULL;
880 status = request (buffer, buffer_size, &res);
881 pthread_mutex_unlock (&lock);
883 if (status != 0) {
884 rrd_set_error ("rrdcached: %s", res->message);
885 return (NULL);
886 }
887 data = cd = NULL;
888 for( l=0 ; l < res->lines_num ; l++ ) {
889 /* first extract the keyword */
890 for(k = s = res->lines[l];s && *s;s++) {
891 if(*s == ' ') { *s = 0; s++; break; }
892 }
893 if(!s || !*s) break;
894 itype = atoi(s); /* extract type code */
895 for(;*s;s++) { if(*s == ' ') { *s = 0; s++; break; } }
896 if(!*s) break;
897 /* finally, we're pointing to the value */
898 switch(itype) {
899 case RD_I_VAL:
900 if(*s == 'N') { info.u_val = DNAN; } else { info.u_val = atof(s); }
901 break;
902 case RD_I_CNT:
903 info.u_cnt = atol(s);
904 break;
905 case RD_I_INT:
906 info.u_int = atoi(s);
907 break;
908 case RD_I_STR:
909 chomp(s);
910 info.u_str = (char*)malloc(sizeof(char) * (strlen(s) + 1));
911 strcpy(info.u_str,s);
912 break;
913 case RD_I_BLO:
914 rrd_set_error ("rrdc_info: BLOB objects are not supported");
915 return (NULL);
916 default:
917 rrd_set_error ("rrdc_info: Unsupported info type %d",itype);
918 return (NULL);
919 }
921 cd = rrd_info_push(cd, sprintf_alloc("%s",k), itype, info);
922 if(!data) data = cd;
923 }
924 response_free (res);
926 return (data);
927 } /* }}} int rrdc_info */
929 time_t rrdc_last (const char *filename) /* {{{ */
930 {
931 char buffer[4096];
932 char *buffer_ptr;
933 size_t buffer_free;
934 size_t buffer_size;
935 rrdc_response_t *res;
936 int status;
937 char file_path[PATH_MAX];
938 time_t lastup;
940 if (filename == NULL) {
941 rrd_set_error ("rrdc_last: no filename");
942 return (-1);
943 }
945 memset (buffer, 0, sizeof (buffer));
946 buffer_ptr = &buffer[0];
947 buffer_free = sizeof (buffer);
949 status = buffer_add_string ("last", &buffer_ptr, &buffer_free);
950 if (status != 0) {
951 rrd_set_error ("rrdc_last: out of memory");
952 return (-1);
953 }
955 pthread_mutex_lock (&lock);
956 filename = get_path (filename, file_path);
957 if (filename == NULL)
958 {
959 pthread_mutex_unlock (&lock);
960 return (-1);
961 }
963 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
964 if (status != 0)
965 {
966 pthread_mutex_unlock (&lock);
967 rrd_set_error ("rrdc_last: out of memory");
968 return (-1);
969 }
971 assert (buffer_free < sizeof (buffer));
972 buffer_size = sizeof (buffer) - buffer_free;
973 assert (buffer[buffer_size - 1] == ' ');
974 buffer[buffer_size - 1] = '\n';
976 res = NULL;
977 status = request (buffer, buffer_size, &res);
978 pthread_mutex_unlock (&lock);
980 if (status != 0) {
981 rrd_set_error ("rrdcached: %s", res->message);
982 return (-1);
983 }
984 lastup = atol(res->message);
985 response_free (res);
987 return (lastup);
988 } /* }}} int rrdc_last */
990 time_t rrdc_first (const char *filename, int rraindex) /* {{{ */
991 {
992 char buffer[4096];
993 char *buffer_ptr;
994 size_t buffer_free;
995 size_t buffer_size;
996 rrdc_response_t *res;
997 int status;
998 char file_path[PATH_MAX];
999 time_t firstup;
1001 if (filename == NULL) {
1002 rrd_set_error ("rrdc_first: no filename specified");
1003 return (-1);
1004 }
1006 memset (buffer, 0, sizeof (buffer));
1007 buffer_ptr = &buffer[0];
1008 buffer_free = sizeof (buffer);
1010 status = buffer_add_string ("first", &buffer_ptr, &buffer_free);
1011 if (status != 0) {
1012 rrd_set_error ("rrdc_first: out of memory");
1013 return (-1);
1014 }
1016 pthread_mutex_lock (&lock);
1017 filename = get_path (filename, file_path);
1018 if (filename == NULL)
1019 {
1020 pthread_mutex_unlock (&lock);
1021 return (-1);
1022 }
1024 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
1025 if (status != 0)
1026 {
1027 pthread_mutex_unlock (&lock);
1028 rrd_set_error ("rrdc_first: out of memory");
1029 return (-1);
1030 }
1031 status = buffer_add_ulong (rraindex, &buffer_ptr, &buffer_free);
1032 if (status != 0)
1033 {
1034 pthread_mutex_unlock (&lock);
1035 rrd_set_error ("rrdc_first: out of memory");
1036 return (-1);
1037 }
1039 assert (buffer_free < sizeof (buffer));
1040 buffer_size = sizeof (buffer) - buffer_free;
1041 assert (buffer[buffer_size - 1] == ' ');
1042 buffer[buffer_size - 1] = '\n';
1044 res = NULL;
1045 status = request (buffer, buffer_size, &res);
1046 pthread_mutex_unlock (&lock);
1048 if (status != 0) {
1049 rrd_set_error ("rrdcached: %s", res->message);
1050 return (-1);
1051 }
1052 firstup = atol(res->message);
1053 response_free (res);
1055 return (firstup);
1056 } /* }}} int rrdc_first */
1058 int rrdc_create (const char *filename, /* {{{ */
1059 unsigned long pdp_step,
1060 time_t last_up,
1061 int no_overwrite,
1062 int argc,
1063 const char **argv)
1064 {
1065 char buffer[4096];
1066 char *buffer_ptr;
1067 size_t buffer_free;
1068 size_t buffer_size;
1069 rrdc_response_t *res;
1070 int status;
1071 char file_path[PATH_MAX];
1072 int i;
1074 if (filename == NULL) {
1075 rrd_set_error ("rrdc_create: no filename specified");
1076 return (-1);
1077 }
1079 memset (buffer, 0, sizeof (buffer));
1080 buffer_ptr = &buffer[0];
1081 buffer_free = sizeof (buffer);
1083 status = buffer_add_string ("create", &buffer_ptr, &buffer_free);
1084 if (status != 0) {
1085 rrd_set_error ("rrdc_create: out of memory");
1086 return (-1);
1087 }
1089 pthread_mutex_lock (&lock);
1090 filename = get_path (filename, file_path);
1091 if (filename == NULL)
1092 {
1093 pthread_mutex_unlock (&lock);
1094 return (-1);
1095 }
1097 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
1098 status = buffer_add_string ("-b", &buffer_ptr, &buffer_free);
1099 status = buffer_add_ulong (last_up, &buffer_ptr, &buffer_free);
1100 status = buffer_add_string ("-s", &buffer_ptr, &buffer_free);
1101 status = buffer_add_ulong (pdp_step, &buffer_ptr, &buffer_free);
1102 if(no_overwrite) {
1103 status = buffer_add_string ("-O", &buffer_ptr, &buffer_free);
1104 }
1105 if (status != 0)
1106 {
1107 pthread_mutex_unlock (&lock);
1108 rrd_set_error ("rrdc_create: out of memory");
1109 return (-1);
1110 }
1112 for( i=0; i<argc; i++ ) {
1113 if( argv[i] ) {
1114 status = buffer_add_string (argv[i], &buffer_ptr, &buffer_free);
1115 if (status != 0)
1116 {
1117 pthread_mutex_unlock (&lock);
1118 rrd_set_error ("rrdc_create: out of memory");
1119 return (-1);
1120 }
1121 }
1122 }
1124 /* buffer ready to send? */
1125 assert (buffer_free < sizeof (buffer));
1126 buffer_size = sizeof (buffer) - buffer_free;
1127 assert (buffer[buffer_size - 1] == ' ');
1128 buffer[buffer_size - 1] = '\n';
1130 res = NULL;
1131 status = request (buffer, buffer_size, &res);
1132 pthread_mutex_unlock (&lock);
1134 if (status != 0) {
1135 rrd_set_error ("rrdcached: %s", res->message);
1136 return (-1);
1137 }
1138 response_free (res);
1139 return(0);
1140 } /* }}} int rrdc_create */
1142 int rrdc_fetch (const char *filename, /* {{{ */
1143 const char *cf,
1144 time_t *ret_start, time_t *ret_end,
1145 unsigned long *ret_step,
1146 unsigned long *ret_ds_num,
1147 char ***ret_ds_names,
1148 rrd_value_t **ret_data)
1149 {
1150 char buffer[4096];
1151 char *buffer_ptr;
1152 size_t buffer_free;
1153 size_t buffer_size;
1154 rrdc_response_t *res;
1155 char path_buffer[PATH_MAX];
1156 const char *path_ptr;
1158 char *str_tmp;
1159 unsigned long flush_version;
1161 time_t start;
1162 time_t end;
1163 unsigned long step;
1164 unsigned long ds_num;
1165 char **ds_names;
1167 rrd_value_t *data;
1168 size_t data_size;
1169 size_t data_fill;
1171 int status;
1172 size_t current_line;
1173 time_t t;
1175 if ((filename == NULL) || (cf == NULL))
1176 return (-1);
1178 /* Send request {{{ */
1179 memset (buffer, 0, sizeof (buffer));
1180 buffer_ptr = &buffer[0];
1181 buffer_free = sizeof (buffer);
1183 status = buffer_add_string ("FETCH", &buffer_ptr, &buffer_free);
1184 if (status != 0)
1185 return (ENOBUFS);
1187 /* change to path for rrdcached */
1188 path_ptr = get_path (filename, path_buffer);
1189 if (path_ptr == NULL)
1190 return (EINVAL);
1192 status = buffer_add_string (path_ptr, &buffer_ptr, &buffer_free);
1193 if (status != 0)
1194 return (ENOBUFS);
1196 status = buffer_add_string (cf, &buffer_ptr, &buffer_free);
1197 if (status != 0)
1198 return (ENOBUFS);
1200 if ((ret_start != NULL) && (*ret_start > 0))
1201 {
1202 char tmp[64];
1203 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_start);
1204 tmp[sizeof (tmp) - 1] = 0;
1205 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
1206 if (status != 0)
1207 return (ENOBUFS);
1209 if ((ret_end != NULL) && (*ret_end > 0))
1210 {
1211 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_end);
1212 tmp[sizeof (tmp) - 1] = 0;
1213 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
1214 if (status != 0)
1215 return (ENOBUFS);
1216 }
1217 }
1219 assert (buffer_free < sizeof (buffer));
1220 buffer_size = sizeof (buffer) - buffer_free;
1221 assert (buffer[buffer_size - 1] == ' ');
1222 buffer[buffer_size - 1] = '\n';
1224 res = NULL;
1225 status = request (buffer, buffer_size, &res);
1226 if (status != 0)
1227 return (status);
1229 status = res->status;
1230 if (status < 0)
1231 {
1232 rrd_set_error ("rrdcached: %s", res->message);
1233 response_free (res);
1234 return (status);
1235 }
1236 /* }}} Send request */
1238 ds_names = NULL;
1239 ds_num = 0;
1240 data = NULL;
1241 current_line = 0;
1243 /* Macros to make error handling a little easier (i. e. less to type and
1244 * read. `BAIL_OUT' sets the error message, frees all dynamically allocated
1245 * variables and returns the provided status code. */
1246 #define BAIL_OUT(status, ...) do { \
1247 rrd_set_error ("rrdc_fetch: " __VA_ARGS__); \
1248 free (data); \
1249 if (ds_names != 0) { size_t k; for (k = 0; k < ds_num; k++) free (ds_names[k]); } \
1250 free (ds_names); \
1251 response_free (res); \
1252 return (status); \
1253 } while (0)
1255 #define READ_NUMERIC_FIELD(name,type,var) do { \
1256 char *key; \
1257 unsigned long value; \
1258 assert (current_line < res->lines_num); \
1259 status = parse_ulong_header (res->lines[current_line], &key, &value); \
1260 if (status != 0) \
1261 BAIL_OUT (-1, "Unable to parse header `%s'", name); \
1262 if (strcasecmp (key, name) != 0) \
1263 BAIL_OUT (-1, "Unexpected header line: Expected `%s', got `%s'", name, key); \
1264 var = (type) value; \
1265 current_line++; \
1266 } while (0)
1268 if (res->lines_num < 1)
1269 BAIL_OUT (-1, "Premature end of response packet");
1271 /* We're making some very strong assumptions about the fields below. We
1272 * therefore check the version of the `flush' command first, so that later
1273 * versions can change the order of fields and it's easier to implement
1274 * backwards compatibility. */
1275 READ_NUMERIC_FIELD ("FlushVersion", unsigned long, flush_version);
1276 if (flush_version != 1)
1277 BAIL_OUT (-1, "Don't know how to handle flush format version %lu.",
1278 flush_version);
1280 if (res->lines_num < 5)
1281 BAIL_OUT (-1, "Premature end of response packet");
1283 READ_NUMERIC_FIELD ("Start", time_t, start);
1284 READ_NUMERIC_FIELD ("End", time_t, end);
1285 if (start >= end)
1286 BAIL_OUT (-1, "Malformed start and end times: start = %lu; end = %lu;",
1287 (unsigned long) start,
1288 (unsigned long) end);
1290 READ_NUMERIC_FIELD ("Step", unsigned long, step);
1291 if (step < 1)
1292 BAIL_OUT (-1, "Invalid number for Step: %lu", step);
1294 READ_NUMERIC_FIELD ("DSCount", unsigned long, ds_num);
1295 if (ds_num < 1)
1296 BAIL_OUT (-1, "Invalid number for DSCount: %lu", ds_num);
1298 /* It's time to allocate some memory */
1299 ds_names = calloc ((size_t) ds_num, sizeof (*ds_names));
1300 if (ds_names == NULL)
1301 BAIL_OUT (-1, "Out of memory");
1303 status = parse_char_array_header (res->lines[current_line],
1304 &str_tmp, ds_names, (size_t) ds_num, /* alloc = */ 1);
1305 if (status != 0)
1306 BAIL_OUT (-1, "Unable to parse header `DSName'");
1307 if (strcasecmp ("DSName", str_tmp) != 0)
1308 BAIL_OUT (-1, "Unexpected header line: Expected `DSName', got `%s'", str_tmp);
1309 current_line++;
1311 data_size = ds_num * (end - start) / step;
1312 if (data_size < 1)
1313 BAIL_OUT (-1, "No data returned or headers invalid.");
1315 if (res->lines_num != (6 + (data_size / ds_num)))
1316 BAIL_OUT (-1, "Got %zu lines, expected %zu",
1317 res->lines_num, (6 + (data_size / ds_num)));
1319 data = calloc (data_size, sizeof (*data));
1320 if (data == NULL)
1321 BAIL_OUT (-1, "Out of memory");
1324 data_fill = 0;
1325 for (t = start + step; t <= end; t += step, current_line++)
1326 {
1327 time_t tmp;
1329 assert (current_line < res->lines_num);
1331 status = parse_value_array_header (res->lines[current_line],
1332 &tmp, data + data_fill, (size_t) ds_num);
1333 if (status != 0)
1334 BAIL_OUT (-1, "Cannot parse value line");
1336 data_fill += (size_t) ds_num;
1337 }
1339 *ret_start = start;
1340 *ret_end = end;
1341 *ret_step = step;
1342 *ret_ds_num = ds_num;
1343 *ret_ds_names = ds_names;
1344 *ret_data = data;
1346 response_free (res);
1347 return (0);
1348 #undef READ_NUMERIC_FIELD
1349 #undef BAIL_OUT
1350 } /* }}} int rrdc_flush */
1352 /* convenience function; if there is a daemon specified, or if we can
1353 * detect one from the environment, then flush the file. Otherwise, no-op
1354 */
1355 int rrdc_flush_if_daemon (const char *opt_daemon, const char *filename) /* {{{ */
1356 {
1357 int status = 0;
1359 rrdc_connect(opt_daemon);
1361 if (rrdc_is_connected(opt_daemon))
1362 {
1363 rrd_clear_error();
1364 status = rrdc_flush (filename);
1366 if (status != 0 && !rrd_test_error())
1367 {
1368 if (status > 0)
1369 {
1370 rrd_set_error("rrdc_flush (%s) failed: %s",
1371 filename, rrd_strerror(status));
1372 }
1373 else if (status < 0)
1374 {
1375 rrd_set_error("rrdc_flush (%s) failed with status %i.",
1376 filename, status);
1377 }
1378 }
1379 } /* if (rrdc_is_connected(..)) */
1381 return status;
1382 } /* }}} int rrdc_flush_if_daemon */
1385 int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */
1386 {
1387 rrdc_stats_t *head;
1388 rrdc_stats_t *tail;
1390 rrdc_response_t *res;
1392 int status;
1393 size_t i;
1395 /* Protocol example: {{{
1396 * -> STATS
1397 * <- 5 Statistics follow
1398 * <- QueueLength: 0
1399 * <- UpdatesWritten: 0
1400 * <- DataSetsWritten: 0
1401 * <- TreeNodesNumber: 0
1402 * <- TreeDepth: 0
1403 * }}} */
1405 res = NULL;
1406 pthread_mutex_lock (&lock);
1407 status = request ("STATS\n", strlen ("STATS\n"), &res);
1408 pthread_mutex_unlock (&lock);
1410 if (status != 0)
1411 return (status);
1413 if (res->status <= 0)
1414 {
1415 response_free (res);
1416 return (EIO);
1417 }
1419 head = NULL;
1420 tail = NULL;
1421 for (i = 0; i < res->lines_num; i++)
1422 {
1423 char *key;
1424 char *value;
1425 char *endptr;
1426 rrdc_stats_t *s;
1428 key = res->lines[i];
1429 value = strchr (key, ':');
1430 if (value == NULL)
1431 continue;
1432 *value = 0;
1433 value++;
1435 while ((value[0] == ' ') || (value[0] == '\t'))
1436 value++;
1438 s = (rrdc_stats_t *) malloc (sizeof (rrdc_stats_t));
1439 if (s == NULL)
1440 continue;
1441 memset (s, 0, sizeof (*s));
1443 s->name = strdup (key);
1445 endptr = NULL;
1446 if ((strcmp ("QueueLength", key) == 0)
1447 || (strcmp ("TreeDepth", key) == 0)
1448 || (strcmp ("TreeNodesNumber", key) == 0))
1449 {
1450 s->type = RRDC_STATS_TYPE_GAUGE;
1451 s->value.gauge = strtod (value, &endptr);
1452 }
1453 else if ((strcmp ("DataSetsWritten", key) == 0)
1454 || (strcmp ("FlushesReceived", key) == 0)
1455 || (strcmp ("JournalBytes", key) == 0)
1456 || (strcmp ("JournalRotate", key) == 0)
1457 || (strcmp ("UpdatesReceived", key) == 0)
1458 || (strcmp ("UpdatesWritten", key) == 0))
1459 {
1460 s->type = RRDC_STATS_TYPE_COUNTER;
1461 s->value.counter = (uint64_t) strtoll (value, &endptr, /* base = */ 0);
1462 }
1463 else
1464 {
1465 free (s);
1466 continue;
1467 }
1469 /* Conversion failed */
1470 if (endptr == value)
1471 {
1472 free (s);
1473 continue;
1474 }
1476 if (head == NULL)
1477 {
1478 head = s;
1479 tail = s;
1480 s->next = NULL;
1481 }
1482 else
1483 {
1484 tail->next = s;
1485 tail = s;
1486 }
1487 } /* for (i = 0; i < res->lines_num; i++) */
1489 response_free (res);
1491 if (head == NULL)
1492 return (EPROTO);
1494 *ret_stats = head;
1495 return (0);
1496 } /* }}} int rrdc_stats_get */
1498 void rrdc_stats_free (rrdc_stats_t *ret_stats) /* {{{ */
1499 {
1500 rrdc_stats_t *this;
1502 this = ret_stats;
1503 while (this != NULL)
1504 {
1505 rrdc_stats_t *next;
1507 next = this->next;
1509 if (this->name != NULL)
1510 {
1511 free ((char *)this->name);
1512 this->name = NULL;
1513 }
1514 free (this);
1516 this = next;
1517 } /* while (this != NULL) */
1518 } /* }}} void rrdc_stats_free */
1520 /*
1521 * vim: set sw=2 sts=2 ts=8 et fdm=marker :
1522 */