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;
77 || (strncmp ("unix:", sd_path, strlen ("unix:")) == 0))
80 if (*path == '/') /* absolute path */
84 rrd_set_error ("absolute path names not allowed when talking "
85 "to a remote daemon");
88 /* else: nothing to do */
90 else /* relative path */
94 realpath (path, resolved_path);
97 /* else: nothing to do */
100 } /* }}} char *get_path */
102 static size_t strsplit (char *string, char **fields, size_t size) /* {{{ */
111 while ((fields[i] = strtok_r (ptr, " \t\r\n", &saveptr)) != NULL)
121 } /* }}} size_t strsplit */
123 static int parse_header (char *line, /* {{{ */
124 char **ret_key, char **ret_value)
130 tmp = strchr (line, ':');
139 while ((tmp[0] == ' ') || (tmp[0] == '\t'));
146 } /* }}} int parse_header */
148 static int parse_ulong_header (char *line, /* {{{ */
149 char **ret_key, unsigned long *ret_value)
156 status = parse_header (line, ret_key, &str_value);
162 *ret_value = (unsigned long) strtol (str_value, &endptr, /* base = */ 0);
163 if ((endptr == str_value) || (errno != 0))
167 } /* }}} int parse_ulong_header */
169 static int parse_char_array_header (char *line, /* {{{ */
170 char **ret_key, char **array, size_t array_len, int alloc)
172 char *tmp_array[array_len];
178 status = parse_header (line, ret_key, &value);
182 num = strsplit (value, tmp_array, array_len);
183 if (num != array_len)
188 memcpy (array, tmp_array, sizeof (tmp_array));
194 for (i = 0; i < array_len; i++)
195 array[i] = strdup (tmp_array[i]);
199 } /* }}} int parse_char_array_header */
201 static int parse_value_array_header (char *line, /* {{{ */
202 time_t *ret_time, rrd_value_t *array, size_t array_len)
205 char *str_array[array_len];
211 status = parse_char_array_header (line, &str_key,
212 str_array, array_len, /* alloc = */ 0);
218 *ret_time = (time_t) strtol (str_key, &endptr, /* base = */ 10);
219 if ((endptr == str_key) || (errno != 0))
222 for (i = 0; i < array_len; i++)
225 array[i] = (rrd_value_t) strtod (str_array[i], &endptr);
226 if ((endptr == str_array[i]) || (errno != 0))
231 } /* }}} int parse_value_array_header */
233 /* One must hold `lock' when calling `close_connection'. */
234 static void close_connection (void) /* {{{ */
251 } /* }}} void close_connection */
253 static int buffer_add_string (const char *str, /* {{{ */
254 char **buffer_ret, size_t *buffer_size_ret)
262 buffer = *buffer_ret;
263 buffer_size = *buffer_size_ret;
268 while (buffer_pos < buffer_size)
272 buffer[buffer_pos] = ' ';
277 else if ((str[i] == ' ') || (str[i] == '\\'))
279 if (buffer_pos >= (buffer_size - 1))
281 buffer[buffer_pos] = '\\';
283 buffer[buffer_pos] = str[i];
288 buffer[buffer_pos] = str[i];
292 } /* while (buffer_pos < buffer_size) */
297 *buffer_ret = buffer + buffer_pos;
298 *buffer_size_ret = buffer_size - buffer_pos;
301 } /* }}} int buffer_add_string */
303 static int buffer_add_value (const char *value, /* {{{ */
304 char **buffer_ret, size_t *buffer_size_ret)
308 if (strncmp (value, "N:", 2) == 0)
309 snprintf (temp, sizeof (temp), "%lu:%s",
310 (unsigned long) time (NULL), value + 2);
312 strncpy (temp, value, sizeof (temp));
313 temp[sizeof (temp) - 1] = 0;
315 return (buffer_add_string (temp, buffer_ret, buffer_size_ret));
316 } /* }}} int buffer_add_value */
318 /* Remove trailing newline (NL) and carriage return (CR) characters. Similar to
319 * the Perl function `chomp'. Returns the number of characters that have been
321 static int chomp (char *str) /* {{{ */
331 while ((len > 0) && ((str[len - 1] == '\n') || (str[len - 1] == '\r')))
339 } /* }}} int chomp */
341 static void response_free (rrdc_response_t *res) /* {{{ */
346 if (res->lines != NULL)
350 for (i = 0; i < res->lines_num; i++)
351 if (res->lines[i] != NULL)
352 free (res->lines[i]);
357 } /* }}} void response_free */
359 static int response_read (rrdc_response_t **ret_response) /* {{{ */
361 rrdc_response_t *ret;
371 ret = (rrdc_response_t *) malloc (sizeof (rrdc_response_t));
374 memset (ret, 0, sizeof (*ret));
378 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
379 if (buffer_ptr == NULL)
383 ret->status = strtol (buffer, &ret->message, 0);
384 if (buffer == ret->message)
389 /* Skip leading whitespace of the status message */
390 ret->message += strspn (ret->message, " \t");
392 if (ret->status <= 0)
395 rrd_set_error("rrdcached: %s", ret->message);
400 ret->lines = (char **) malloc (sizeof (char *) * ret->status);
401 if (ret->lines == NULL)
406 memset (ret->lines, 0, sizeof (char *) * ret->status);
407 ret->lines_num = (size_t) ret->status;
409 for (i = 0; i < ret->lines_num; i++)
411 buffer_ptr = fgets (buffer, sizeof (buffer), sh);
412 if (buffer_ptr == NULL)
419 ret->lines[i] = strdup (buffer);
420 if (ret->lines[i] == NULL)
429 } /* }}} rrdc_response_t *response_read */
431 static int request (const char *buffer, size_t buffer_size, /* {{{ */
432 rrdc_response_t **ret_response)
435 rrdc_response_t *res;
440 status = (int) fwrite (buffer, buffer_size, /* nmemb = */ 1, sh);
444 rrd_set_error("request: socket error (%d) while talking to rrdcached",
451 status = response_read (&res);
456 rrd_set_error("request: internal error while talking to rrdcached");
462 } /* }}} int request */
464 /* determine whether we are connected to the specified daemon_addr if
465 * NULL, return whether we are connected at all
467 int rrdc_is_connected(const char *daemon_addr) /* {{{ */
471 else if (daemon_addr == NULL)
473 /* here we have to handle the case i.e.
474 * UPDATE --daemon ...; UPDATEV (no --daemon) ...
475 * In other words: we have a cached connection,
476 * but it is not specified in the current command.
477 * Daemon is only implied in this case if set in ENV
479 if (getenv(ENV_RRDCACHED_ADDRESS) != NULL)
484 else if (strcmp(daemon_addr, sd_path) == 0)
489 } /* }}} int rrdc_is_connected */
491 static int rrdc_connect_unix (const char *path) /* {{{ */
493 struct sockaddr_un sa;
496 assert (path != NULL);
499 sd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
506 memset (&sa, 0, sizeof (sa));
507 sa.sun_family = AF_UNIX;
508 strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
510 status = connect (sd, (struct sockaddr *) &sa, sizeof (sa));
518 sh = fdopen (sd, "r+");
527 } /* }}} int rrdc_connect_unix */
529 static int rrdc_connect_network (const char *addr_orig) /* {{{ */
531 struct addrinfo ai_hints;
532 struct addrinfo *ai_res;
533 struct addrinfo *ai_ptr;
534 char addr_copy[NI_MAXHOST];
538 assert (addr_orig != NULL);
541 strncpy(addr_copy, addr_orig, sizeof(addr_copy));
542 addr_copy[sizeof(addr_copy) - 1] = '\0';
546 memset (&ai_hints, 0, sizeof (ai_hints));
547 ai_hints.ai_flags = 0;
549 ai_hints.ai_flags |= AI_ADDRCONFIG;
551 ai_hints.ai_family = AF_UNSPEC;
552 ai_hints.ai_socktype = SOCK_STREAM;
555 if (*addr == '[') /* IPv6+port format */
557 /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
560 port = strchr (addr, ']');
563 rrd_set_error("malformed address: %s", addr_orig);
575 rrd_set_error("garbage after address: %s", port);
578 } /* if (*addr == '[') */
581 port = rindex(addr, ':');
590 status = getaddrinfo (addr,
591 port == NULL ? RRDCACHED_DEFAULT_PORT : port,
595 rrd_set_error ("failed to resolve address `%s' (port %s): %s",
596 addr, port == NULL ? RRDCACHED_DEFAULT_PORT : port,
597 gai_strerror (status));
601 for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
603 sd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
611 status = connect (sd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
619 sh = fdopen (sd, "r+");
627 assert (status == 0);
632 } /* }}} int rrdc_connect_network */
634 int rrdc_connect (const char *addr) /* {{{ */
639 addr = getenv (ENV_RRDCACHED_ADDRESS);
644 pthread_mutex_lock(&lock);
646 if (sd >= 0 && sd_path != NULL && strcmp(addr, sd_path) == 0)
648 /* connection to the same daemon; use cached connection */
649 pthread_mutex_unlock (&lock);
658 if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
659 status = rrdc_connect_unix (addr + strlen ("unix:"));
660 else if (addr[0] == '/')
661 status = rrdc_connect_unix (addr);
663 status = rrdc_connect_network(addr);
665 if (status == 0 && sd >= 0)
666 sd_path = strdup(addr);
669 char *err = rrd_test_error () ? rrd_get_error () : "Internal error";
670 /* err points the string that gets written to by rrd_set_error(), thus we
671 * cannot pass it to that function */
673 rrd_set_error("Unable to connect to rrdcached: %s",
675 ? (err ? err : "Internal error")
676 : rrd_strerror (status));
681 pthread_mutex_unlock (&lock);
683 } /* }}} int rrdc_connect */
685 int rrdc_disconnect (void) /* {{{ */
687 pthread_mutex_lock (&lock);
691 pthread_mutex_unlock (&lock);
694 } /* }}} int rrdc_disconnect */
696 int rrdc_update (const char *filename, int values_num, /* {{{ */
697 const char * const *values)
703 rrdc_response_t *res;
706 char file_path[PATH_MAX];
708 memset (buffer, 0, sizeof (buffer));
709 buffer_ptr = &buffer[0];
710 buffer_free = sizeof (buffer);
712 status = buffer_add_string ("update", &buffer_ptr, &buffer_free);
716 pthread_mutex_lock (&lock);
717 filename = get_path (filename, file_path);
718 if (filename == NULL)
720 pthread_mutex_unlock (&lock);
724 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
727 pthread_mutex_unlock (&lock);
731 for (i = 0; i < values_num; i++)
733 status = buffer_add_value (values[i], &buffer_ptr, &buffer_free);
736 pthread_mutex_unlock (&lock);
741 assert (buffer_free < sizeof (buffer));
742 buffer_size = sizeof (buffer) - buffer_free;
743 assert (buffer[buffer_size - 1] == ' ');
744 buffer[buffer_size - 1] = '\n';
747 status = request (buffer, buffer_size, &res);
748 pthread_mutex_unlock (&lock);
753 status = res->status;
757 } /* }}} int rrdc_update */
759 int rrdc_flush (const char *filename) /* {{{ */
765 rrdc_response_t *res;
767 char file_path[PATH_MAX];
769 if (filename == NULL)
772 memset (buffer, 0, sizeof (buffer));
773 buffer_ptr = &buffer[0];
774 buffer_free = sizeof (buffer);
776 status = buffer_add_string ("flush", &buffer_ptr, &buffer_free);
780 pthread_mutex_lock (&lock);
781 filename = get_path (filename, file_path);
782 if (filename == NULL)
784 pthread_mutex_unlock (&lock);
788 status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
791 pthread_mutex_unlock (&lock);
795 assert (buffer_free < sizeof (buffer));
796 buffer_size = sizeof (buffer) - buffer_free;
797 assert (buffer[buffer_size - 1] == ' ');
798 buffer[buffer_size - 1] = '\n';
801 status = request (buffer, buffer_size, &res);
802 pthread_mutex_unlock (&lock);
807 status = res->status;
811 } /* }}} int rrdc_flush */
813 int rrdc_fetch (const char *filename, /* {{{ */
815 time_t *ret_start, time_t *ret_end,
816 unsigned long *ret_step,
817 unsigned long *ret_ds_num,
818 char ***ret_ds_names,
819 rrd_value_t **ret_data)
825 rrdc_response_t *res;
826 char path_buffer[PATH_MAX];
830 unsigned long flush_version;
835 unsigned long ds_num;
846 if ((filename == NULL) || (cf == NULL))
849 /* Send request {{{ */
850 memset (buffer, 0, sizeof (buffer));
851 buffer_ptr = &buffer[0];
852 buffer_free = sizeof (buffer);
854 status = buffer_add_string ("FETCH", &buffer_ptr, &buffer_free);
858 /* change to path for rrdcached */
859 path_ptr = get_path (filename, path_buffer);
860 if (path_ptr == NULL)
863 status = buffer_add_string (path_ptr, &buffer_ptr, &buffer_free);
867 status = buffer_add_string (cf, &buffer_ptr, &buffer_free);
871 if ((ret_start != NULL) && (*ret_start > 0))
874 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_start);
875 tmp[sizeof (tmp) - 1] = 0;
876 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
880 if ((ret_end != NULL) && (*ret_end > 0))
882 snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_end);
883 tmp[sizeof (tmp) - 1] = 0;
884 status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
890 assert (buffer_free < sizeof (buffer));
891 buffer_size = sizeof (buffer) - buffer_free;
892 assert (buffer[buffer_size - 1] == ' ');
893 buffer[buffer_size - 1] = '\n';
896 status = request (buffer, buffer_size, &res);
900 status = res->status;
903 rrd_set_error ("rrdcached: %s", res->message);
907 /* }}} Send request */
914 /* Macros to make error handling a little easier (i. e. less to type and
915 * read. `BAIL_OUT' sets the error message, frees all dynamically allocated
916 * variables and returns the provided status code. */
917 #define BAIL_OUT(status, ...) do { \
918 rrd_set_error ("rrdc_fetch: " __VA_ARGS__); \
920 if (ds_names != 0) { size_t k; for (k = 0; k < ds_num; k++) free (ds_names[k]); } \
922 response_free (res); \
926 #define READ_NUMERIC_FIELD(name,type,var) do { \
928 unsigned long value; \
929 assert (current_line < res->lines_num); \
930 status = parse_ulong_header (res->lines[current_line], &key, &value); \
932 BAIL_OUT (-1, "Unable to parse header `%s'", name); \
933 if (strcasecmp (key, name) != 0) \
934 BAIL_OUT (-1, "Unexpected header line: Expected `%s', got `%s'", name, key); \
935 var = (type) value; \
939 if (res->lines_num < 1)
940 BAIL_OUT (-1, "Premature end of response packet");
942 /* We're making some very strong assumptions about the fields below. We
943 * therefore check the version of the `flush' command first, so that later
944 * versions can change the order of fields and it's easier to implement
945 * backwards compatibility. */
946 READ_NUMERIC_FIELD ("FlushVersion", unsigned long, flush_version);
947 if (flush_version != 1)
948 BAIL_OUT (-1, "Don't know how to handle flush format version %lu.",
951 if (res->lines_num < 5)
952 BAIL_OUT (-1, "Premature end of response packet");
954 READ_NUMERIC_FIELD ("Start", time_t, start);
955 READ_NUMERIC_FIELD ("End", time_t, end);
957 BAIL_OUT (-1, "Malformed start and end times: start = %lu; end = %lu;",
958 (unsigned long) start,
959 (unsigned long) end);
961 READ_NUMERIC_FIELD ("Step", unsigned long, step);
963 BAIL_OUT (-1, "Invalid number for Step: %lu", step);
965 READ_NUMERIC_FIELD ("DSCount", unsigned long, ds_num);
967 BAIL_OUT (-1, "Invalid number for DSCount: %lu", ds_num);
969 /* It's time to allocate some memory */
970 ds_names = calloc ((size_t) ds_num, sizeof (*ds_names));
971 if (ds_names == NULL)
972 BAIL_OUT (-1, "Out of memory");
974 status = parse_char_array_header (res->lines[current_line],
975 &str_tmp, ds_names, (size_t) ds_num, /* alloc = */ 1);
977 BAIL_OUT (-1, "Unable to parse header `DSName'");
978 if (strcasecmp ("DSName", str_tmp) != 0)
979 BAIL_OUT (-1, "Unexpected header line: Expected `DSName', got `%s'", str_tmp);
982 data_size = ds_num * (end - start) / step;
984 BAIL_OUT (-1, "No data returned or headers invalid.");
986 if (res->lines_num != (6 + (data_size / ds_num)))
987 BAIL_OUT (-1, "Got %zu lines, expected %zu",
988 res->lines_num, (6 + (data_size / ds_num)));
990 data = calloc (data_size, sizeof (*data));
992 BAIL_OUT (-1, "Out of memory");
996 for (t = start + step; t <= end; t += step, current_line++)
1000 assert (current_line < res->lines_num);
1002 status = parse_value_array_header (res->lines[current_line],
1003 &tmp, data + data_fill, (size_t) ds_num);
1005 BAIL_OUT (-1, "Cannot parse value line");
1007 data_fill += (size_t) ds_num;
1013 *ret_ds_num = ds_num;
1014 *ret_ds_names = ds_names;
1017 response_free (res);
1019 #undef READ_NUMERIC_FIELD
1021 } /* }}} int rrdc_flush */
1023 /* convenience function; if there is a daemon specified, or if we can
1024 * detect one from the environment, then flush the file. Otherwise, no-op
1026 int rrdc_flush_if_daemon (const char *opt_daemon, const char *filename) /* {{{ */
1030 rrdc_connect(opt_daemon);
1032 if (rrdc_is_connected(opt_daemon))
1035 status = rrdc_flush (filename);
1037 if (status != 0 && !rrd_test_error())
1041 rrd_set_error("rrdc_flush (%s) failed: %s",
1042 filename, rrd_strerror(status));
1044 else if (status < 0)
1046 rrd_set_error("rrdc_flush (%s) failed with status %i.",
1050 } /* if (rrdc_is_connected(..)) */
1053 } /* }}} int rrdc_flush_if_daemon */
1056 int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */
1061 rrdc_response_t *res;
1066 /* Protocol example: {{{
1068 * <- 5 Statistics follow
1070 * <- UpdatesWritten: 0
1071 * <- DataSetsWritten: 0
1072 * <- TreeNodesNumber: 0
1077 pthread_mutex_lock (&lock);
1078 status = request ("STATS\n", strlen ("STATS\n"), &res);
1079 pthread_mutex_unlock (&lock);
1084 if (res->status <= 0)
1086 response_free (res);
1092 for (i = 0; i < res->lines_num; i++)
1099 key = res->lines[i];
1100 value = strchr (key, ':');
1106 while ((value[0] == ' ') || (value[0] == '\t'))
1109 s = (rrdc_stats_t *) malloc (sizeof (rrdc_stats_t));
1112 memset (s, 0, sizeof (*s));
1114 s->name = strdup (key);
1117 if ((strcmp ("QueueLength", key) == 0)
1118 || (strcmp ("TreeDepth", key) == 0)
1119 || (strcmp ("TreeNodesNumber", key) == 0))
1121 s->type = RRDC_STATS_TYPE_GAUGE;
1122 s->value.gauge = strtod (value, &endptr);
1124 else if ((strcmp ("DataSetsWritten", key) == 0)
1125 || (strcmp ("FlushesReceived", key) == 0)
1126 || (strcmp ("JournalBytes", key) == 0)
1127 || (strcmp ("JournalRotate", key) == 0)
1128 || (strcmp ("UpdatesReceived", key) == 0)
1129 || (strcmp ("UpdatesWritten", key) == 0))
1131 s->type = RRDC_STATS_TYPE_COUNTER;
1132 s->value.counter = (uint64_t) strtoll (value, &endptr, /* base = */ 0);
1140 /* Conversion failed */
1141 if (endptr == value)
1158 } /* for (i = 0; i < res->lines_num; i++) */
1160 response_free (res);
1167 } /* }}} int rrdc_stats_get */
1169 void rrdc_stats_free (rrdc_stats_t *ret_stats) /* {{{ */
1174 while (this != NULL)
1180 if (this->name != NULL)
1182 free ((char *)this->name);
1188 } /* while (this != NULL) */
1189 } /* }}} void rrdc_stats_free */
1192 * vim: set sw=2 sts=2 ts=8 et fdm=marker :