2 * RRDTool - src/rrd_client.c
3 * Copyright (C) 2008-2010 Florian octo Forster
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:
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
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
24 * Florian octo Forster <octo at verplant.org>
25 * Sebastian tokkee Harl <sh at tokkee.org>
30 #include "rrd_client.h"
39 #include <sys/types.h>
40 #include <sys/socket.h>
46 #define ENODATA ENOENT
49 struct rrdc_response_s
56 typedef struct rrdc_response_s rrdc_response_t;
58 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
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.
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.
70 * One must hold `lock' when calling this function. */
71 static const char *get_path (const char *path, char *resolved_path) /* {{{ */
73 const char *ret = path;
76 if ((path == NULL) || (resolved_path == NULL) || (sd_path == NULL))
80 || (strncmp ("unix:", sd_path, strlen ("unix:")) == 0))
85 ret = realpath(path, resolved_path);
87 rrd_set_error("realpath(%s): %s", path, rrd_strerror(errno));
92 if (*path == '/') /* not absolute path */
94 rrd_set_error ("absolute path names not allowed when talking "
95 "to a remote daemon");
101 } /* }}} char *get_path */
103 static size_t strsplit (char *string, char **fields, size_t size) /* {{{ */
112 while ((fields[i] = strtok_r (ptr, " \t\r\n", &saveptr)) != NULL)
122 } /* }}} size_t strsplit */
124 static int parse_header (char *line, /* {{{ */
125 char **ret_key, char **ret_value)
131 tmp = strchr (line, ':');
140 while ((tmp[0] == ' ') || (tmp[0] == '\t'));
147 } /* }}} int parse_header */
149 static int parse_ulong_header (char *line, /* {{{ */
150 char **ret_key, unsigned long *ret_value)
157 status = parse_header (line, ret_key, &str_value);
163 *ret_value = (unsigned long) strtol (str_value, &endptr, /* base = */ 0);
164 if ((endptr == str_value) || (errno != 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)
173 char *tmp_array[array_len];
179 status = parse_header (line, ret_key, &value);
183 num = strsplit (value, tmp_array, array_len);
184 if (num != array_len)
189 memcpy (array, tmp_array, sizeof (tmp_array));
195 for (i = 0; i < array_len; i++)
196 array[i] = strdup (tmp_array[i]);
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)
206 char *str_array[array_len];
212 status = parse_char_array_header (line, &str_key,
213 str_array, array_len, /* alloc = */ 0);
219 *ret_time = (time_t) strtol (str_key, &endptr, /* base = */ 10);
220 if ((endptr == str_key) || (errno != 0))
223 for (i = 0; i < array_len; i++)
226 array[i] = (rrd_value_t) strtod (str_array[i], &endptr);
227 if ((endptr == str_array[i]) || (errno != 0))
232 } /* }}} int parse_value_array_header */
234 /* One must hold `lock' when calling `close_connection'. */
235 static void close_connection (void) /* {{{ */
252 } /* }}} void close_connection */
254 static int buffer_add_string (const char *str, /* {{{ */
255 char **buffer_ret, size_t *buffer_size_ret)
263 buffer = *buffer_ret;
264 buffer_size = *buffer_size_ret;
269 while (buffer_pos < buffer_size)
273 buffer[buffer_pos] = ' ';
278 else if ((str[i] == ' ') || (str[i] == '\\'))
280 if (buffer_pos >= (buffer_size - 1))
282 buffer[buffer_pos] = '\\';
284 buffer[buffer_pos] = str[i];
289 buffer[buffer_pos] = str[i];
293 } /* while (buffer_pos < buffer_size) */
298 *buffer_ret = buffer + buffer_pos;
299 *buffer_size_ret = buffer_size - buffer_pos;
302 } /* }}} int buffer_add_string */
304 static int buffer_add_value (const char *value, /* {{{ */
305 char **buffer_ret, size_t *buffer_size_ret)
307 char temp[RRD_CMD_MAX];
309 if (strncmp (value, "N:", 2) == 0)
310 snprintf (temp, sizeof (temp), "%lu:%s",
311 (unsigned long) time (NULL), value + 2);
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)
322 char temp[RRD_CMD_MAX];
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
332 static int chomp (char *str) /* {{{ */
342 while ((len > 0) && ((str[len - 1] == '\n') || (str[len - 1] == '\r')))
350 } /* }}} int chomp */
352 static void response_free (rrdc_response_t *res) /* {{{ */
357 if (res->lines != NULL)
361 for (i = 0; i < res->lines_num; i++)
362 if (res->lines[i] != NULL)
363 free (res->lines[i]);
368 } /* }}} void response_free */
370 static int response_read (rrdc_response_t **ret_response) /* {{{ */
372 rrdc_response_t *ret = NULL;
375 char buffer[RRD_CMD_MAX];
380 #define DIE(code) do { status = code; goto err_out; } while(0)
385 ret = (rrdc_response_t *) malloc (sizeof (rrdc_response_t));
388 memset (ret, 0, sizeof (*ret));
392 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
393 if (buffer_ptr == NULL)
398 ret->status = strtol (buffer, &ret->message, 0);
399 if (buffer == ret->message)
402 /* Skip leading whitespace of the status message */
403 ret->message += strspn (ret->message, " \t");
405 if (ret->status <= 0)
408 rrd_set_error("rrdcached: %s", ret->message);
412 ret->lines = (char **) malloc (sizeof (char *) * ret->status);
413 if (ret->lines == NULL)
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++)
421 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
422 if (buffer_ptr == NULL)
427 ret->lines[i] = strdup (buffer);
428 if (ret->lines[i] == NULL)
444 } /* }}} rrdc_response_t *response_read */
446 static int request (const char *buffer, size_t buffer_size, /* {{{ */
447 rrdc_response_t **ret_response)
450 rrdc_response_t *res;
455 status = (int) fwrite (buffer, buffer_size, /* nmemb = */ 1, sh);
459 rrd_set_error("request: socket error (%d) while talking to rrdcached",
466 status = response_read (&res);
471 rrd_set_error("request: internal error while talking to rrdcached");
477 } /* }}} int request */
479 /* determine whether we are connected to the specified daemon_addr if
480 * NULL, return whether we are connected at all
482 int rrdc_is_connected(const char *daemon_addr) /* {{{ */
486 else if (daemon_addr == NULL)
488 char *addr = getenv(ENV_RRDCACHED_ADDRESS);
489 /* here we have to handle the case i.e.
490 * UPDATE --daemon ...; UPDATEV (no --daemon) ...
491 * In other words: we have a cached connection,
492 * but it is not specified in the current command.
493 * Daemon is only implied in this case if set in ENV
495 if (addr != NULL && ! strcmp(addr,""))
500 else if (strcmp(daemon_addr, sd_path) == 0)
505 } /* }}} int rrdc_is_connected */
507 static int rrdc_connect_unix (const char *path) /* {{{ */
509 struct sockaddr_un sa;
512 assert (path != NULL);
515 sd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
522 memset (&sa, 0, sizeof (sa));
523 sa.sun_family = AF_UNIX;
524 strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
526 status = connect (sd, (struct sockaddr *) &sa, sizeof (sa));
534 sh = fdopen (sd, "r+");
543 } /* }}} int rrdc_connect_unix */
545 static int rrdc_connect_network (const char *addr_orig) /* {{{ */
547 struct addrinfo ai_hints;
548 struct addrinfo *ai_res;
549 struct addrinfo *ai_ptr;
550 char addr_copy[NI_MAXHOST];
554 assert (addr_orig != NULL);
557 strncpy(addr_copy, addr_orig, sizeof(addr_copy));
558 addr_copy[sizeof(addr_copy) - 1] = '\0';
562 memset (&ai_hints, 0, sizeof (ai_hints));
563 ai_hints.ai_flags = 0;
565 ai_hints.ai_flags |= AI_ADDRCONFIG;
567 ai_hints.ai_family = AF_UNSPEC;
568 ai_hints.ai_socktype = SOCK_STREAM;
571 if (*addr == '[') /* IPv6+port format */
573 /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
576 port = strchr (addr, ']');
579 rrd_set_error("malformed address: %s", addr_orig);
591 rrd_set_error("garbage after address: %s", port);
594 } /* if (*addr == '[') */
597 port = rindex(addr, ':');
606 status = getaddrinfo (addr,
607 port == NULL ? RRDCACHED_DEFAULT_PORT : port,
611 rrd_set_error ("failed to resolve address `%s' (port %s): %s",
612 addr, port == NULL ? RRDCACHED_DEFAULT_PORT : port,
613 gai_strerror (status));
617 for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
619 sd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
627 status = connect (sd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
635 sh = fdopen (sd, "r+");
643 assert (status == 0);
647 freeaddrinfo(ai_res);
650 } /* }}} int rrdc_connect_network */
652 int rrdc_connect (const char *addr) /* {{{ */
657 addr = getenv (ENV_RRDCACHED_ADDRESS);
660 if (addr == NULL || ! strcmp(addr,"")) {
665 pthread_mutex_lock(&lock);
667 if (sd >= 0 && sd_path != NULL && strcmp(addr, sd_path) == 0)
669 /* connection to the same daemon; use cached connection */
670 pthread_mutex_unlock (&lock);
679 if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
680 status = rrdc_connect_unix (addr + strlen ("unix:"));
681 else if (addr[0] == '/')
682 status = rrdc_connect_unix (addr);
684 status = rrdc_connect_network(addr);
686 if (status == 0 && sd >= 0)
687 sd_path = strdup(addr);
690 char *err = rrd_test_error () ? rrd_get_error () : "Internal error";
691 /* err points the string that gets written to by rrd_set_error(), thus we
692 * cannot pass it to that function */
694 rrd_set_error("Unable to connect to rrdcached: %s",
696 ? (err ? err : "Internal error")
697 : rrd_strerror (status));
702 pthread_mutex_unlock (&lock);
704 } /* }}} int rrdc_connect */
706 int rrdc_disconnect (void) /* {{{ */
708 pthread_mutex_lock (&lock);
712 pthread_mutex_unlock (&lock);
715 } /* }}} int rrdc_disconnect */
717 int rrdc_update (const char *filename, int values_num, /* {{{ */
718 const char * const *values)
720 char buffer[RRD_CMD_MAX];
724 rrdc_response_t *res;
727 char file_path[PATH_MAX];
729 memset (buffer, 0, sizeof (buffer));
730 buffer_ptr = &buffer[0];
731 buffer_free = sizeof (buffer);
733 status = buffer_add_string ("update", &buffer_ptr, &buffer_free);
737 pthread_mutex_lock (&lock);
738 filename = get_path (filename, file_path);
739 if (filename == NULL)
741 pthread_mutex_unlock (&lock);
745 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
748 pthread_mutex_unlock (&lock);
752 for (i = 0; i < values_num; i++)
754 status = buffer_add_value (values[i], &buffer_ptr, &buffer_free);
757 pthread_mutex_unlock (&lock);
762 assert (buffer_free < sizeof (buffer));
763 buffer_size = sizeof (buffer) - buffer_free;
764 assert (buffer[buffer_size - 1] == ' ');
765 buffer[buffer_size - 1] = '\n';
768 status = request (buffer, buffer_size, &res);
769 pthread_mutex_unlock (&lock);
774 status = res->status;
778 } /* }}} int rrdc_update */
780 int rrdc_flush (const char *filename) /* {{{ */
782 char buffer[RRD_CMD_MAX];
786 rrdc_response_t *res;
788 char file_path[PATH_MAX];
790 if (filename == NULL)
793 memset (buffer, 0, sizeof (buffer));
794 buffer_ptr = &buffer[0];
795 buffer_free = sizeof (buffer);
797 status = buffer_add_string ("flush", &buffer_ptr, &buffer_free);
801 pthread_mutex_lock (&lock);
802 filename = get_path (filename, file_path);
803 if (filename == NULL)
805 pthread_mutex_unlock (&lock);
809 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
812 pthread_mutex_unlock (&lock);
816 assert (buffer_free < sizeof (buffer));
817 buffer_size = sizeof (buffer) - buffer_free;
818 assert (buffer[buffer_size - 1] == ' ');
819 buffer[buffer_size - 1] = '\n';
822 status = request (buffer, buffer_size, &res);
823 pthread_mutex_unlock (&lock);
828 status = res->status;
832 } /* }}} int rrdc_flush */
834 rrd_info_t * rrdc_info (const char *filename) /* {{{ */
836 char buffer[RRD_CMD_MAX];
840 rrdc_response_t *res;
842 char file_path[PATH_MAX];
843 rrd_info_t *data = NULL, *cd;
846 rrd_info_type_t itype;
849 if (filename == NULL) {
850 rrd_set_error ("rrdc_info: no filename");
854 memset (buffer, 0, sizeof (buffer));
855 buffer_ptr = &buffer[0];
856 buffer_free = sizeof (buffer);
858 status = buffer_add_string ("info", &buffer_ptr, &buffer_free);
860 rrd_set_error ("rrdc_info: out of memory");
864 pthread_mutex_lock (&lock);
865 filename = get_path (filename, file_path);
866 if (filename == NULL)
868 pthread_mutex_unlock (&lock);
872 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
875 pthread_mutex_unlock (&lock);
876 rrd_set_error ("rrdc_info: out of memory");
880 assert (buffer_free < sizeof (buffer));
881 buffer_size = sizeof (buffer) - buffer_free;
882 assert (buffer[buffer_size - 1] == ' ');
883 buffer[buffer_size - 1] = '\n';
886 status = request (buffer, buffer_size, &res);
887 pthread_mutex_unlock (&lock);
890 rrd_set_error ("rrdcached: %s", res->message);
894 for( l=0 ; l < res->lines_num ; l++ ) {
895 /* first extract the keyword */
896 for(k = s = res->lines[l];s && *s;s++) {
897 if(*s == ' ') { *s = 0; s++; break; }
900 itype = atoi(s); /* extract type code */
901 for(;*s;s++) { if(*s == ' ') { *s = 0; s++; break; } }
903 /* finally, we're pointing to the value */
906 if(*s == 'N') { info.u_val = DNAN; } else { info.u_val = atof(s); }
909 info.u_cnt = atol(s);
912 info.u_int = atoi(s);
916 info.u_str = (char*)malloc(sizeof(char) * (strlen(s) + 1));
917 strcpy(info.u_str,s);
920 rrd_set_error ("rrdc_info: BLOB objects are not supported");
923 rrd_set_error ("rrdc_info: Unsupported info type %d",itype);
927 cd = rrd_info_push(cd, sprintf_alloc("%s",k), itype, info);
933 } /* }}} int rrdc_info */
935 time_t rrdc_last (const char *filename) /* {{{ */
937 char buffer[RRD_CMD_MAX];
941 rrdc_response_t *res;
943 char file_path[PATH_MAX];
946 if (filename == NULL) {
947 rrd_set_error ("rrdc_last: no filename");
951 memset (buffer, 0, sizeof (buffer));
952 buffer_ptr = &buffer[0];
953 buffer_free = sizeof (buffer);
955 status = buffer_add_string ("last", &buffer_ptr, &buffer_free);
957 rrd_set_error ("rrdc_last: out of memory");
961 pthread_mutex_lock (&lock);
962 filename = get_path (filename, file_path);
963 if (filename == NULL)
965 pthread_mutex_unlock (&lock);
969 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
972 pthread_mutex_unlock (&lock);
973 rrd_set_error ("rrdc_last: out of memory");
977 assert (buffer_free < sizeof (buffer));
978 buffer_size = sizeof (buffer) - buffer_free;
979 assert (buffer[buffer_size - 1] == ' ');
980 buffer[buffer_size - 1] = '\n';
983 status = request (buffer, buffer_size, &res);
984 pthread_mutex_unlock (&lock);
987 rrd_set_error ("rrdcached: %s", res->message);
990 lastup = atol(res->message);
994 } /* }}} int rrdc_last */
996 time_t rrdc_first (const char *filename, int rraindex) /* {{{ */
998 char buffer[RRD_CMD_MAX];
1002 rrdc_response_t *res;
1004 char file_path[PATH_MAX];
1007 if (filename == NULL) {
1008 rrd_set_error ("rrdc_first: no filename specified");
1012 memset (buffer, 0, sizeof (buffer));
1013 buffer_ptr = &buffer[0];
1014 buffer_free = sizeof (buffer);
1016 status = buffer_add_string ("first", &buffer_ptr, &buffer_free);
1018 rrd_set_error ("rrdc_first: out of memory");
1022 pthread_mutex_lock (&lock);
1023 filename = get_path (filename, file_path);
1024 if (filename == NULL)
1026 pthread_mutex_unlock (&lock);
1030 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
1033 pthread_mutex_unlock (&lock);
1034 rrd_set_error ("rrdc_first: out of memory");
1037 status = buffer_add_ulong (rraindex, &buffer_ptr, &buffer_free);
1040 pthread_mutex_unlock (&lock);
1041 rrd_set_error ("rrdc_first: out of memory");
1045 assert (buffer_free < sizeof (buffer));
1046 buffer_size = sizeof (buffer) - buffer_free;
1047 assert (buffer[buffer_size - 1] == ' ');
1048 buffer[buffer_size - 1] = '\n';
1051 status = request (buffer, buffer_size, &res);
1052 pthread_mutex_unlock (&lock);
1055 rrd_set_error ("rrdcached: %s", res->message);
1058 firstup = atol(res->message);
1059 response_free (res);
1062 } /* }}} int rrdc_first */
1064 int rrdc_create (const char *filename, /* {{{ */
1065 unsigned long pdp_step,
1071 char buffer[RRD_CMD_MAX];
1075 rrdc_response_t *res;
1077 char file_path[PATH_MAX];
1080 if (filename == NULL) {
1081 rrd_set_error ("rrdc_create: no filename specified");
1085 memset (buffer, 0, sizeof (buffer));
1086 buffer_ptr = &buffer[0];
1087 buffer_free = sizeof (buffer);
1089 status = buffer_add_string ("create", &buffer_ptr, &buffer_free);
1091 rrd_set_error ("rrdc_create: out of memory");
1095 pthread_mutex_lock (&lock);
1096 filename = get_path (filename, file_path);
1097 if (filename == NULL)
1099 pthread_mutex_unlock (&lock);
1103 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
1104 status = buffer_add_string ("-b", &buffer_ptr, &buffer_free);
1105 status = buffer_add_ulong (last_up, &buffer_ptr, &buffer_free);
1106 status = buffer_add_string ("-s", &buffer_ptr, &buffer_free);
1107 status = buffer_add_ulong (pdp_step, &buffer_ptr, &buffer_free);
1109 status = buffer_add_string ("-O", &buffer_ptr, &buffer_free);
1113 pthread_mutex_unlock (&lock);
1114 rrd_set_error ("rrdc_create: out of memory");
1118 for( i=0; i<argc; i++ ) {
1120 status = buffer_add_string (argv[i], &buffer_ptr, &buffer_free);
1123 pthread_mutex_unlock (&lock);
1124 rrd_set_error ("rrdc_create: out of memory");
1130 /* buffer ready to send? */
1131 assert (buffer_free < sizeof (buffer));
1132 buffer_size = sizeof (buffer) - buffer_free;
1133 assert (buffer[buffer_size - 1] == ' ');
1134 buffer[buffer_size - 1] = '\n';
1137 status = request (buffer, buffer_size, &res);
1138 pthread_mutex_unlock (&lock);
1141 rrd_set_error ("rrdcached: %s", res->message);
1144 response_free (res);
1146 } /* }}} int rrdc_create */
1148 int rrdc_fetch (const char *filename, /* {{{ */
1150 time_t *ret_start, time_t *ret_end,
1151 unsigned long *ret_step,
1152 unsigned long *ret_ds_num,
1153 char ***ret_ds_names,
1154 rrd_value_t **ret_data)
1156 char buffer[RRD_CMD_MAX];
1160 rrdc_response_t *res;
1161 char path_buffer[PATH_MAX];
1162 const char *path_ptr;
1165 unsigned long flush_version;
1170 unsigned long ds_num;
1178 size_t current_line;
1181 if ((filename == NULL) || (cf == NULL))
1184 /* Send request {{{ */
1185 memset (buffer, 0, sizeof (buffer));
1186 buffer_ptr = &buffer[0];
1187 buffer_free = sizeof (buffer);
1189 status = buffer_add_string ("FETCH", &buffer_ptr, &buffer_free);
1193 /* change to path for rrdcached */
1194 path_ptr = get_path (filename, path_buffer);
1195 if (path_ptr == NULL)
1198 status = buffer_add_string (path_ptr, &buffer_ptr, &buffer_free);
1202 status = buffer_add_string (cf, &buffer_ptr, &buffer_free);
1206 if ((ret_start != NULL) && (*ret_start > 0))
1209 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_start);
1210 tmp[sizeof (tmp) - 1] = 0;
1211 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
1215 if ((ret_end != NULL) && (*ret_end > 0))
1217 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_end);
1218 tmp[sizeof (tmp) - 1] = 0;
1219 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
1225 assert (buffer_free < sizeof (buffer));
1226 buffer_size = sizeof (buffer) - buffer_free;
1227 assert (buffer[buffer_size - 1] == ' ');
1228 buffer[buffer_size - 1] = '\n';
1231 status = request (buffer, buffer_size, &res);
1235 status = res->status;
1238 rrd_set_error ("rrdcached: %s", res->message);
1239 response_free (res);
1242 /* }}} Send request */
1249 /* Macros to make error handling a little easier (i. e. less to type and
1250 * read. `BAIL_OUT' sets the error message, frees all dynamically allocated
1251 * variables and returns the provided status code. */
1252 #define BAIL_OUT(status, ...) do { \
1253 rrd_set_error ("rrdc_fetch: " __VA_ARGS__); \
1255 if (ds_names != 0) { size_t k; for (k = 0; k < ds_num; k++) free (ds_names[k]); } \
1257 response_free (res); \
1261 #define READ_NUMERIC_FIELD(name,type,var) do { \
1263 unsigned long value; \
1264 assert (current_line < res->lines_num); \
1265 status = parse_ulong_header (res->lines[current_line], &key, &value); \
1267 BAIL_OUT (-1, "Unable to parse header `%s'", name); \
1268 if (strcasecmp (key, name) != 0) \
1269 BAIL_OUT (-1, "Unexpected header line: Expected `%s', got `%s'", name, key); \
1270 var = (type) value; \
1274 if (res->lines_num < 1)
1275 BAIL_OUT (-1, "Premature end of response packet");
1277 /* We're making some very strong assumptions about the fields below. We
1278 * therefore check the version of the `flush' command first, so that later
1279 * versions can change the order of fields and it's easier to implement
1280 * backwards compatibility. */
1281 READ_NUMERIC_FIELD ("FlushVersion", unsigned long, flush_version);
1282 if (flush_version != 1)
1283 BAIL_OUT (-1, "Don't know how to handle flush format version %lu.",
1286 if (res->lines_num < 5)
1287 BAIL_OUT (-1, "Premature end of response packet");
1289 READ_NUMERIC_FIELD ("Start", time_t, start);
1290 READ_NUMERIC_FIELD ("End", time_t, end);
1292 BAIL_OUT (-1, "Malformed start and end times: start = %lu; end = %lu;",
1293 (unsigned long) start,
1294 (unsigned long) end);
1296 READ_NUMERIC_FIELD ("Step", unsigned long, step);
1298 BAIL_OUT (-1, "Invalid number for Step: %lu", step);
1300 READ_NUMERIC_FIELD ("DSCount", unsigned long, ds_num);
1302 BAIL_OUT (-1, "Invalid number for DSCount: %lu", ds_num);
1304 /* It's time to allocate some memory */
1305 ds_names = calloc ((size_t) ds_num, sizeof (*ds_names));
1306 if (ds_names == NULL)
1307 BAIL_OUT (-1, "Out of memory");
1309 status = parse_char_array_header (res->lines[current_line],
1310 &str_tmp, ds_names, (size_t) ds_num, /* alloc = */ 1);
1312 BAIL_OUT (-1, "Unable to parse header `DSName'");
1313 if (strcasecmp ("DSName", str_tmp) != 0)
1314 BAIL_OUT (-1, "Unexpected header line: Expected `DSName', got `%s'", str_tmp);
1317 data_size = ds_num * (end - start) / step;
1319 BAIL_OUT (-1, "No data returned or headers invalid.");
1321 if (res->lines_num != (6 + (data_size / ds_num)))
1322 BAIL_OUT (-1, "Got %zu lines, expected %zu",
1323 res->lines_num, (6 + (data_size / ds_num)));
1325 data = calloc (data_size, sizeof (*data));
1327 BAIL_OUT (-1, "Out of memory");
1331 for (t = start + step; t <= end; t += step, current_line++)
1335 assert (current_line < res->lines_num);
1337 status = parse_value_array_header (res->lines[current_line],
1338 &tmp, data + data_fill, (size_t) ds_num);
1340 BAIL_OUT (-1, "Cannot parse value line");
1342 data_fill += (size_t) ds_num;
1348 *ret_ds_num = ds_num;
1349 *ret_ds_names = ds_names;
1352 response_free (res);
1354 #undef READ_NUMERIC_FIELD
1356 } /* }}} int rrdc_flush */
1358 /* convenience function; if there is a daemon specified, or if we can
1359 * detect one from the environment, then flush the file. Otherwise, no-op
1361 int rrdc_flush_if_daemon (const char *opt_daemon, const char *filename) /* {{{ */
1365 rrdc_connect(opt_daemon);
1367 if (rrdc_is_connected(opt_daemon))
1370 status = rrdc_flush (filename);
1372 if (status != 0 && !rrd_test_error())
1376 rrd_set_error("rrdc_flush (%s) failed: %s",
1377 filename, rrd_strerror(status));
1379 else if (status < 0)
1381 rrd_set_error("rrdc_flush (%s) failed with status %i.",
1385 } /* if (rrdc_is_connected(..)) */
1388 } /* }}} int rrdc_flush_if_daemon */
1391 int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */
1396 rrdc_response_t *res;
1401 /* Protocol example: {{{
1403 * <- 5 Statistics follow
1405 * <- UpdatesWritten: 0
1406 * <- DataSetsWritten: 0
1407 * <- TreeNodesNumber: 0
1412 pthread_mutex_lock (&lock);
1413 status = request ("STATS\n", strlen ("STATS\n"), &res);
1414 pthread_mutex_unlock (&lock);
1419 if (res->status <= 0)
1421 response_free (res);
1427 for (i = 0; i < res->lines_num; i++)
1434 key = res->lines[i];
1435 value = strchr (key, ':');
1441 while ((value[0] == ' ') || (value[0] == '\t'))
1444 s = (rrdc_stats_t *) malloc (sizeof (rrdc_stats_t));
1447 memset (s, 0, sizeof (*s));
1449 s->name = strdup (key);
1452 if ((strcmp ("QueueLength", key) == 0)
1453 || (strcmp ("TreeDepth", key) == 0)
1454 || (strcmp ("TreeNodesNumber", key) == 0))
1456 s->type = RRDC_STATS_TYPE_GAUGE;
1457 s->value.gauge = strtod (value, &endptr);
1459 else if ((strcmp ("DataSetsWritten", key) == 0)
1460 || (strcmp ("FlushesReceived", key) == 0)
1461 || (strcmp ("JournalBytes", key) == 0)
1462 || (strcmp ("JournalRotate", key) == 0)
1463 || (strcmp ("UpdatesReceived", key) == 0)
1464 || (strcmp ("UpdatesWritten", key) == 0))
1466 s->type = RRDC_STATS_TYPE_COUNTER;
1467 s->value.counter = (uint64_t) strtoll (value, &endptr, /* base = */ 0);
1475 /* Conversion failed */
1476 if (endptr == value)
1493 } /* for (i = 0; i < res->lines_num; i++) */
1495 response_free (res);
1502 } /* }}} int rrdc_stats_get */
1504 void rrdc_stats_free (rrdc_stats_t *ret_stats) /* {{{ */
1509 while (this != NULL)
1515 if (this->name != NULL)
1517 free ((char *)this->name);
1523 } /* while (this != NULL) */
1524 } /* }}} void rrdc_stats_free */
1527 * vim: set sw=2 sts=2 ts=8 et fdm=marker :