X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_client.c;h=287f642db4754253eba0134679bf1e650f60ad45;hp=d1ad5e06ef0cf4ee07b63d20e172cb0166562d06;hb=6f5e93c0c08c5ed796d8d079f0eba2d16c6a4bdb;hpb=a12627275ff8487174cbb907a066f62a00b6ae44 diff --git a/src/rrd_client.c b/src/rrd_client.c index d1ad5e0..287f642 100644 --- a/src/rrd_client.c +++ b/src/rrd_client.c @@ -17,6 +17,7 @@ * * Authors: * Florian octo Forster + * Sebastian tokkee Harl **/ #include "rrd.h" @@ -48,84 +49,67 @@ typedef struct rrdc_response_s rrdc_response_t; static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; static int sd = -1; +static FILE *sh = NULL; static char *sd_path = NULL; /* cache the path for sd */ -static void _disconnect(void); -static ssize_t sread (void *buffer_void, size_t buffer_size) /* {{{ */ +/* get_path: Return a path name appropriate to be sent to the daemon. + * + * When talking to a local daemon (thru a UNIX socket), relative path names + * are resolved to absolute path names to allow for transparent integration + * into existing solutions (as requested by Tobi). Else, absolute path names + * are not allowed, since path name translation is done by the server. + * + * One must hold `lock' when calling this function. */ +static const char *get_path (const char *path, char *resolved_path) /* {{{ */ { - char *buffer; - size_t buffer_used; - size_t buffer_free; - ssize_t status; + const char *ret = path; + int is_unix = 0; - buffer = (char *) buffer_void; - buffer_used = 0; - buffer_free = buffer_size; + if ((*sd_path == '/') + || (strncmp ("unix:", sd_path, strlen ("unix:")) == 0)) + is_unix = 1; - while (buffer_free > 0) + if (*path == '/') /* absolute path */ { - status = read (sd, buffer + buffer_used, buffer_free); - if ((status < 0) && ((errno == EAGAIN) || (errno == EINTR))) - continue; - - if (status < 0) - return (-1); - - if (status == 0) + if (! is_unix) { - _disconnect(); - errno = EPROTO; - return (-1); + rrd_set_error ("absolute path names not allowed when talking " + "to a remote daemon"); + return (NULL); } - - assert ((0 > status) || (buffer_free >= (size_t) status)); - - buffer_free -= status; - buffer_used += status; - - if (buffer[buffer_used - 1] == '\n') - break; + /* else: nothing to do */ } - - if (buffer[buffer_used - 1] != '\n') + else /* relative path */ { - errno = ENOBUFS; - return (-1); + if (is_unix) + { + realpath (path, resolved_path); + ret = resolved_path; + } + /* else: nothing to do */ } + return (ret); +} /* }}} char *get_path */ - buffer[buffer_used - 1] = '\0'; - return (buffer_used); -} /* }}} ssize_t sread */ - -static ssize_t swrite (const void *buf, size_t count) /* {{{ */ +/* One must hold `lock' when calling `close_connection'. */ +static void close_connection (void) /* {{{ */ { - const char *ptr; - size_t nleft; - ssize_t status; - - ptr = (const char *) buf; - nleft = count; - - while (nleft > 0) + if (sh != NULL) { - status = write (sd, (const void *) ptr, nleft); - - if ((status < 0) && ((errno == EAGAIN) || (errno == EINTR))) - continue; - - if (status < 0) - { - _disconnect(); - rrd_set_error("lost connection to rrdcached"); - return (status); - } - - nleft -= status; - ptr += status; + fclose (sh); + sh = NULL; + sd = -1; + } + else if (sd >= 0) + { + close (sd); + sd = -1; } - return (0); -} /* }}} ssize_t swrite */ + if (sd_path != NULL) + free (sd_path); + sd_path = NULL; +} /* }}} void close_connection */ static int buffer_add_string (const char *str, /* {{{ */ char **buffer_ret, size_t *buffer_size_ret) @@ -192,103 +176,151 @@ static int buffer_add_value (const char *value, /* {{{ */ return (buffer_add_string (temp, buffer_ret, buffer_size_ret)); } /* }}} int buffer_add_value */ -static int response_parse (char *buffer, size_t buffer_size, /* {{{ */ - rrdc_response_t **ret_response) +/* Remove trailing newline (NL) and carriage return (CR) characters. Similar to + * the Perl function `chomp'. Returns the number of characters that have been + * removed. */ +static int chomp (char *str) /* {{{ */ { - rrdc_response_t *ret; + size_t len; + int removed; + + if (str == NULL) + return (-1); + + len = strlen (str); + removed = 0; + while ((len > 0) && ((str[len - 1] == '\n') || (str[len - 1] == '\r'))) + { + str[len - 1] = 0; + len--; + removed++; + } + + return (removed); +} /* }}} int chomp */ + +static void response_free (rrdc_response_t *res) /* {{{ */ +{ + if (res == NULL) + return; + + if (res->lines != NULL) + { + size_t i; - char *dummy; - char *saveptr; + for (i = 0; i < res->lines_num; i++) + if (res->lines[i] != NULL) + free (res->lines[i]); + free (res->lines); + } - char *line_ptr; - size_t line_counter; + free (res); +} /* }}} void response_free */ - if (buffer == NULL) - return (EINVAL); - if (buffer_size <= 0) - return (EINVAL); +static int response_read (rrdc_response_t **ret_response) /* {{{ */ +{ + rrdc_response_t *ret; - if (buffer[buffer_size - 1] != 0) + char buffer[4096]; + char *buffer_ptr; + + size_t i; + + if (sh == NULL) return (-1); ret = (rrdc_response_t *) malloc (sizeof (rrdc_response_t)); if (ret == NULL) - return (ENOMEM); + return (-2); memset (ret, 0, sizeof (*ret)); + ret->lines = NULL; + ret->lines_num = 0; - line_counter = 0; + buffer_ptr = fgets (buffer, sizeof (buffer), sh); + if (buffer_ptr == NULL) + return (-3); + chomp (buffer); - dummy = buffer; - saveptr = NULL; - while ((line_ptr = strtok_r (dummy, "\r\n", &saveptr)) != NULL) + ret->status = strtol (buffer, &ret->message, 0); + if (buffer == ret->message) { - dummy = NULL; + response_free (ret); + return (-4); + } + /* Skip leading whitespace of the status message */ + ret->message += strspn (ret->message, " \t"); - if (ret->message == NULL) - { - ret->status = strtol (buffer, &ret->message, 0); - if (buffer == ret->message) - { - free (ret); - return (EPROTO); - } + if (ret->status <= 0) + { + if (ret->status < 0) + rrd_set_error("rrdcached: %s", ret->message); + *ret_response = ret; + return (0); + } - /* Skip leading whitespace of the status message */ - ret->message += strspn (ret->message, " \t"); + ret->lines = (char **) malloc (sizeof (char *) * ret->status); + if (ret->lines == NULL) + { + response_free (ret); + return (-5); + } + memset (ret->lines, 0, sizeof (char *) * ret->status); + ret->lines_num = (size_t) ret->status; - if (ret->status > 0) - { - ret->lines = (char **) malloc (sizeof (char *) * ret->status); - if (ret->lines == NULL) - { - free (ret); - return (ENOMEM); - } - memset (ret->lines, 0, sizeof (char *) * ret->status); - ret->lines_num = (size_t) ret->status; - } - else - { - ret->lines = NULL; - ret->lines_num = 0; - } - } - else /* if (ret->message != NULL) */ + for (i = 0; i < ret->lines_num; i++) + { + buffer_ptr = fgets (buffer, sizeof (buffer), sh); + if (buffer_ptr == NULL) { - if (line_counter < ret->lines_num) - ret->lines[line_counter] = line_ptr; - line_counter++; + response_free (ret); + return (-6); } - } /* while (strtok_r) */ + chomp (buffer); - if (ret->lines_num != line_counter) - { - errno = EPROTO; - if (ret->lines != NULL) - free (ret->lines); - free (ret); - return (-1); + ret->lines[i] = strdup (buffer); + if (ret->lines[i] == NULL) + { + response_free (ret); + return (-7); + } } *ret_response = ret; return (0); -} /* }}} int response_parse */ +} /* }}} rrdc_response_t *response_read */ -static void response_free (rrdc_response_t *res) /* {{{ */ +static int request (const char *buffer, size_t buffer_size, /* {{{ */ + rrdc_response_t **ret_response) { - if (res == NULL) - return; + int status; + rrdc_response_t *res; - if (res->lines != NULL) + if (sh == NULL) + return (ENOTCONN); + + status = (int) fwrite (buffer, buffer_size, /* nmemb = */ 1, sh); + if (status != 1) { - res->lines_num = 0; - free (res->lines); - res->lines = NULL; + close_connection (); + rrd_set_error("request: socket error (%d) while talking to rrdcached", + status); + return (-1); } + fflush (sh); - free (res); -} /* }}} void response_free */ + res = NULL; + status = response_read (&res); + + if (status != 0) + { + if (status < 0) + rrd_set_error("request: internal error while talking to rrdcached"); + return (status); + } + *ret_response = res; + return (0); +} /* }}} int request */ /* determine whether we are connected to the specified daemon_addr if * NULL, return whether we are connected at all @@ -340,21 +372,37 @@ static int rrdc_connect_unix (const char *path) /* {{{ */ if (status != 0) { status = errno; + close_connection (); + return (status); + } + + sh = fdopen (sd, "r+"); + if (sh == NULL) + { + status = errno; + close_connection (); return (status); } return (0); } /* }}} int rrdc_connect_unix */ -static int rrdc_connect_network (const char *addr) /* {{{ */ +static int rrdc_connect_network (const char *addr_orig) /* {{{ */ { struct addrinfo ai_hints; struct addrinfo *ai_res; struct addrinfo *ai_ptr; + char addr_copy[NI_MAXHOST]; + char *addr; + char *port; - assert (addr != NULL); + assert (addr_orig != NULL); assert (sd == -1); + strncpy(addr_copy, addr_orig, sizeof(addr_copy)); + addr_copy[sizeof(addr_copy) - 1] = '\0'; + addr = addr_copy; + int status; memset (&ai_hints, 0, sizeof (ai_hints)); ai_hints.ai_flags = 0; @@ -364,10 +412,52 @@ static int rrdc_connect_network (const char *addr) /* {{{ */ ai_hints.ai_family = AF_UNSPEC; ai_hints.ai_socktype = SOCK_STREAM; + port = NULL; + if (*addr == '[') /* IPv6+port format */ + { + /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */ + addr++; + + port = strchr (addr, ']'); + if (port == NULL) + { + rrd_set_error("malformed address: %s", addr_orig); + return (-1); + } + *port = 0; + port++; + + if (*port == ':') + port++; + else if (*port == 0) + port = NULL; + else + { + rrd_set_error("garbage after address: %s", port); + return (-1); + } + } /* if (*addr = ']') */ + else if (strchr (addr, '.') != NULL) /* Hostname or IPv4 */ + { + port = rindex(addr, ':'); + if (port != NULL) + { + *port = 0; + port++; + } + } + ai_res = NULL; - status = getaddrinfo (addr, RRDCACHED_DEFAULT_PORT, &ai_hints, &ai_res); + status = getaddrinfo (addr, + port == NULL ? RRDCACHED_DEFAULT_PORT : port, + &ai_hints, &ai_res); if (status != 0) - return (status); + { + rrd_set_error ("failed to resolve address `%s' (port %s): %s", + addr, port == NULL ? RRDCACHED_DEFAULT_PORT : port, + gai_strerror (status)); + return (-1); + } for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) { @@ -383,7 +473,15 @@ static int rrdc_connect_network (const char *addr) /* {{{ */ if (status != 0) { status = errno; - _disconnect(); + close_connection(); + continue; + } + + sh = fdopen (sd, "r+"); + if (sh == NULL) + { + status = errno; + close_connection (); continue; } @@ -414,9 +512,10 @@ int rrdc_connect (const char *addr) /* {{{ */ } else { - _disconnect(); + close_connection(); } + rrd_clear_error (); if (strncmp ("unix:", addr, strlen ("unix:")) == 0) status = rrdc_connect_unix (addr + strlen ("unix:")); else if (addr[0] == '/') @@ -427,32 +526,28 @@ int rrdc_connect (const char *addr) /* {{{ */ if (status == 0 && sd >= 0) sd_path = strdup(addr); else + { + char *err = rrd_test_error () ? rrd_get_error () : "Internal error"; + /* err points the string that gets written to by rrd_set_error(), thus we + * cannot pass it to that function */ + err = strdup (err); rrd_set_error("Unable to connect to rrdcached: %s", (status < 0) - ? "Internal error" + ? (err ? err : "Internal error") : rrd_strerror (status)); + if (err != NULL) + free (err); + } pthread_mutex_unlock (&lock); return (status); } /* }}} int rrdc_connect */ -static void _disconnect(void) /* {{{ */ -{ - if (sd >= 0) - close(sd); - - if (sd_path != NULL) - free(sd_path); - - sd = -1; - sd_path = NULL; -} /* }}} static void _disconnect(void) */ - int rrdc_disconnect (void) /* {{{ */ { pthread_mutex_lock (&lock); - _disconnect(); + close_connection(); pthread_mutex_unlock (&lock); @@ -466,8 +561,10 @@ int rrdc_update (const char *filename, int values_num, /* {{{ */ char *buffer_ptr; size_t buffer_free; size_t buffer_size; + rrdc_response_t *res; int status; int i; + char file_path[PATH_MAX]; memset (buffer, 0, sizeof (buffer)); buffer_ptr = &buffer[0]; @@ -477,15 +574,29 @@ int rrdc_update (const char *filename, int values_num, /* {{{ */ if (status != 0) return (ENOBUFS); + pthread_mutex_lock (&lock); + filename = get_path (filename, file_path); + if (filename == NULL) + { + pthread_mutex_unlock (&lock); + return (-1); + } + status = buffer_add_string (filename, &buffer_ptr, &buffer_free); if (status != 0) + { + pthread_mutex_unlock (&lock); return (ENOBUFS); + } for (i = 0; i < values_num; i++) { status = buffer_add_value (values[i], &buffer_ptr, &buffer_free); if (status != 0) + { + pthread_mutex_unlock (&lock); return (ENOBUFS); + } } assert (buffer_free < sizeof (buffer)); @@ -493,37 +604,16 @@ int rrdc_update (const char *filename, int values_num, /* {{{ */ assert (buffer[buffer_size - 1] == ' '); buffer[buffer_size - 1] = '\n'; - pthread_mutex_lock (&lock); - - if (sd < 0) - { - pthread_mutex_unlock (&lock); - return (ENOTCONN); - } + res = NULL; + status = request (buffer, buffer_size, &res); + pthread_mutex_unlock (&lock); - status = swrite (buffer, buffer_size); if (status != 0) - { - pthread_mutex_unlock (&lock); return (status); - } - - status = sread (buffer, sizeof (buffer)); - if (status < 0) - { - status = errno; - pthread_mutex_unlock (&lock); - return (status); - } - else if (status == 0) - { - pthread_mutex_unlock (&lock); - return (ENODATA); - } - pthread_mutex_unlock (&lock); + status = res->status; + response_free (res); - status = atoi (buffer); return (status); } /* }}} int rrdc_update */ @@ -533,7 +623,9 @@ int rrdc_flush (const char *filename) /* {{{ */ char *buffer_ptr; size_t buffer_free; size_t buffer_size; + rrdc_response_t *res; int status; + char file_path[PATH_MAX]; if (filename == NULL) return (-1); @@ -546,51 +638,40 @@ int rrdc_flush (const char *filename) /* {{{ */ if (status != 0) return (ENOBUFS); + pthread_mutex_lock (&lock); + filename = get_path (filename, file_path); + if (filename == NULL) + { + pthread_mutex_unlock (&lock); + return (-1); + } + status = buffer_add_string (filename, &buffer_ptr, &buffer_free); if (status != 0) + { + pthread_mutex_unlock (&lock); return (ENOBUFS); + } assert (buffer_free < sizeof (buffer)); buffer_size = sizeof (buffer) - buffer_free; assert (buffer[buffer_size - 1] == ' '); buffer[buffer_size - 1] = '\n'; - pthread_mutex_lock (&lock); - - if (sd < 0) - { - pthread_mutex_unlock (&lock); - return (ENOTCONN); - } + res = NULL; + status = request (buffer, buffer_size, &res); + pthread_mutex_unlock (&lock); - status = swrite (buffer, buffer_size); if (status != 0) - { - pthread_mutex_unlock (&lock); return (status); - } - - status = sread (buffer, sizeof (buffer)); - if (status < 0) - { - status = errno; - pthread_mutex_unlock (&lock); - return (status); - } - else if (status == 0) - { - pthread_mutex_unlock (&lock); - return (ENODATA); - } - pthread_mutex_unlock (&lock); + status = res->status; + response_free (res); - status = atoi (buffer); return (status); } /* }}} int rrdc_flush */ - /* convenience function; if there is a daemon specified, or if we can * detect one from the environment, then flush the file. Otherwise, no-op */ @@ -602,13 +683,23 @@ int rrdc_flush_if_daemon (const char *opt_daemon, const char *filename) /* {{{ * if (rrdc_is_connected(opt_daemon)) { + rrd_clear_error(); status = rrdc_flush (filename); - if (status != 0) + + if (status != 0 && !rrd_test_error()) { - rrd_set_error ("rrdc_flush (%s) failed with status %i.", - filename, status); + if (status > 0) + { + rrd_set_error("rrdc_flush (%s) failed: %s", + filename, rrd_strerror(status)); + } + else if (status < 0) + { + rrd_set_error("rrdc_flush (%s) failed with status %i.", + filename, status); + } } - } /* if (daemon_addr) */ + } /* if (rrdc_is_connected(..)) */ return status; } /* }}} int rrdc_flush_if_daemon */ @@ -619,21 +710,11 @@ int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */ rrdc_stats_t *head; rrdc_stats_t *tail; - rrdc_response_t *response; + rrdc_response_t *res; - char buffer[4096]; - size_t buffer_size; int status; size_t i; - pthread_mutex_lock (&lock); - - if (sd < 0) - { - pthread_mutex_unlock (&lock); - return (ENOTCONN); - } - /* Protocol example: {{{ * -> STATS * <- 5 Statistics follow @@ -643,63 +724,31 @@ int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */ * <- TreeNodesNumber: 0 * <- TreeDepth: 0 * }}} */ - status = swrite ("STATS\n", strlen ("STATS\n")); - if (status != 0) - { - pthread_mutex_unlock (&lock); - return (status); - } - - status = sread (buffer, sizeof (buffer)); - if (status < 0) - { - status = errno; - pthread_mutex_unlock (&lock); - return (status); - } - else if (status == 0) - { - pthread_mutex_unlock (&lock); - return (ENODATA); - } + res = NULL; + pthread_mutex_lock (&lock); + status = request ("STATS\n", strlen ("STATS\n"), &res); pthread_mutex_unlock (&lock); - /* Assert NULL termination */ - buffer_size = (size_t) status; - if (buffer[buffer_size - 1] != 0) - { - if (buffer_size < sizeof (buffer)) - { - buffer[buffer_size] = 0; - buffer_size++; - } - else - { - return (ENOBUFS); - } - } - - status = response_parse (buffer, buffer_size, &response); if (status != 0) return (status); - if (response->status <= 0) + if (res->status <= 0) { - response_free (response); + response_free (res); return (EIO); } head = NULL; tail = NULL; - for (i = 0; i < response->lines_num; i++) + for (i = 0; i < res->lines_num; i++) { char *key; char *value; char *endptr; rrdc_stats_t *s; - key = response->lines[i]; + key = res->lines[i]; value = strchr (key, ':'); if (value == NULL) continue; @@ -718,14 +767,18 @@ int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */ endptr = NULL; if ((strcmp ("QueueLength", key) == 0) - || (strcmp ("TreeNodesNumber", key) == 0) - || (strcmp ("TreeDepth", key) == 0)) + || (strcmp ("TreeDepth", key) == 0) + || (strcmp ("TreeNodesNumber", key) == 0)) { s->type = RRDC_STATS_TYPE_GAUGE; s->value.gauge = strtod (value, &endptr); } - else if ((strcmp ("UpdatesWritten", key) == 0) - || (strcmp ("DataSetsWritten", key) == 0)) + else if ((strcmp ("DataSetsWritten", key) == 0) + || (strcmp ("FlushesReceived", key) == 0) + || (strcmp ("JournalBytes", key) == 0) + || (strcmp ("JournalRotate", key) == 0) + || (strcmp ("UpdatesReceived", key) == 0) + || (strcmp ("UpdatesWritten", key) == 0)) { s->type = RRDC_STATS_TYPE_COUNTER; s->value.counter = (uint64_t) strtoll (value, &endptr, /* base = */ 0); @@ -754,9 +807,9 @@ int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */ tail->next = s; tail = s; } - } /* for (i = 0; i < response->lines_num; i++) */ + } /* for (i = 0; i < res->lines_num; i++) */ - response_free (response); + response_free (res); if (head == NULL) return (EPROTO); @@ -778,7 +831,7 @@ void rrdc_stats_free (rrdc_stats_t *ret_stats) /* {{{ */ if (this->name != NULL) { - free (this->name); + free ((char *)this->name); this->name = NULL; } free (this);