Code

wire up systemd support in autoconf -- tomek@pipebreaker.pl
[rrdtool.git] / src / rrd_client.c
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) /* {{{ */
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)
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)
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)
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)
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) /* {{{ */
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)
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)
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);
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)
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
331  * removed. */
332 static int chomp (char *str) /* {{{ */
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) /* {{{ */
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) /* {{{ */
372   rrdc_response_t *ret = NULL;
373   int status = 0;
375   char buffer[RRD_CMD_MAX];
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)
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) /* {{{ */
484   if (sd < 0)
485     return 0;
486   else if (daemon_addr == NULL)
487   {
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
494      */
495     if (addr != NULL && strcmp(addr,"") != 0)
496       return 1;
497     else
498       return 0;
499   }
500   else if (strcmp(daemon_addr, sd_path) == 0)
501     return 1;
502   else
503     return 0;
505 } /* }}} int rrdc_is_connected */
507 static int rrdc_connect_unix (const char *path) /* {{{ */
509   struct sockaddr_un sa;
510   int status;
512   assert (path != NULL);
513   assert (sd == -1);
515   sd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
516   if (sd < 0)
517   {
518     status = errno;
519     return (status);
520   }
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));
527   if (status != 0)
528   {
529     status = errno;
530     close_connection ();
531     return (status);
532   }
534   sh = fdopen (sd, "r+");
535   if (sh == NULL)
536   {
537     status = errno;
538     close_connection ();
539     return (status);
540   }
542   return (0);
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];
551   char *addr;
552   char *port;
554   assert (addr_orig != NULL);
555   assert (sd == -1);
557   strncpy(addr_copy, addr_orig, sizeof(addr_copy));
558   addr_copy[sizeof(addr_copy) - 1] = '\0';
559   addr = addr_copy;
561   int status;
562   memset (&ai_hints, 0, sizeof (ai_hints));
563   ai_hints.ai_flags = 0;
564 #ifdef AI_ADDRCONFIG
565   ai_hints.ai_flags |= AI_ADDRCONFIG;
566 #endif
567   ai_hints.ai_family = AF_UNSPEC;
568   ai_hints.ai_socktype = SOCK_STREAM;
570   port = NULL;
571   if (*addr == '[') /* IPv6+port format */
572   {
573     /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
574     addr++;
576     port = strchr (addr, ']');
577     if (port == NULL)
578     {
579       rrd_set_error("malformed address: %s", addr_orig);
580       return (-1);
581     }
582     *port = 0;
583     port++;
585     if (*port == ':')
586       port++;
587     else if (*port == 0)
588       port = NULL;
589     else
590     {
591       rrd_set_error("garbage after address: %s", port);
592       return (-1);
593     }
594   } /* if (*addr == '[') */
595   else
596   {
597     port = rindex(addr, ':');
598     if (port != NULL)
599     {
600       *port = 0;
601       port++;
602     }
603   }
605   ai_res = NULL;
606   status = getaddrinfo (addr,
607                         port == NULL ? RRDCACHED_DEFAULT_PORT : port,
608                         &ai_hints, &ai_res);
609   if (status != 0)
610   {
611     rrd_set_error ("failed to resolve address `%s' (port %s): %s",
612         addr, port == NULL ? RRDCACHED_DEFAULT_PORT : port,
613         gai_strerror (status));
614     return (-1);
615   }
617   for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
618   {
619     sd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
620     if (sd < 0)
621     {
622       status = errno;
623       sd = -1;
624       continue;
625     }
627     status = connect (sd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
628     if (status != 0)
629     {
630       status = errno;
631       close_connection();
632       continue;
633     }
635     sh = fdopen (sd, "r+");
636     if (sh == NULL)
637     {
638       status = errno;
639       close_connection ();
640       continue;
641     }
643     assert (status == 0);
644     break;
645   } /* for (ai_ptr) */
647   freeaddrinfo(ai_res);
649   return (status);
650 } /* }}} int rrdc_connect_network */
652 int rrdc_connect (const char *addr) /* {{{ */
654   int status = 0;
656   if (addr == NULL) {
657     addr = getenv (ENV_RRDCACHED_ADDRESS);
658   }
660   if (addr == NULL || strcmp(addr,"") == 0) {
661     addr = NULL;
662     return 0;
663   }
665   pthread_mutex_lock(&lock);
667   if (sd >= 0 && sd_path != NULL && strcmp(addr, sd_path) == 0)
668   {
669     /* connection to the same daemon; use cached connection */
670     pthread_mutex_unlock (&lock);
671     return (0);
672   }
673   else
674   {
675     close_connection();
676   }
678   rrd_clear_error ();
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);
683   else
684     status = rrdc_connect_network(addr);
686   if (status == 0 && sd >= 0)
687     sd_path = strdup(addr);
688   else
689   {
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 */
693     err = strdup (err);
694     rrd_set_error("Unable to connect to rrdcached: %s",
695                   (status < 0)
696                   ? (err ? err : "Internal error")
697                   : rrd_strerror (status));
698     if (err != NULL)
699       free (err);
700   }
702   pthread_mutex_unlock (&lock);
703   return (status);
704 } /* }}} int rrdc_connect */
706 int rrdc_disconnect (void) /* {{{ */
708   pthread_mutex_lock (&lock);
710   close_connection();
712   pthread_mutex_unlock (&lock);
714   return (0);
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];
721   char *buffer_ptr;
722   size_t buffer_free;
723   size_t buffer_size;
724   rrdc_response_t *res;
725   int status;
726   int i;
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);
734   if (status != 0)
735     return (ENOBUFS);
737   pthread_mutex_lock (&lock);
738   filename = get_path (filename, file_path);
739   if (filename == NULL)
740   {
741     pthread_mutex_unlock (&lock);
742     return (-1);
743   }
745   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
746   if (status != 0)
747   {
748     pthread_mutex_unlock (&lock);
749     return (ENOBUFS);
750   }
752   for (i = 0; i < values_num; i++)
753   {
754     status = buffer_add_value (values[i], &buffer_ptr, &buffer_free);
755     if (status != 0)
756     {
757       pthread_mutex_unlock (&lock);
758       return (ENOBUFS);
759     }
760   }
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';
767   res = NULL;
768   status = request (buffer, buffer_size, &res);
769   pthread_mutex_unlock (&lock);
771   if (status != 0)
772     return (status);
774   status = res->status;
775   response_free (res);
777   return (status);
778 } /* }}} int rrdc_update */
780 int rrdc_flush (const char *filename) /* {{{ */
782   char buffer[RRD_CMD_MAX];
783   char *buffer_ptr;
784   size_t buffer_free;
785   size_t buffer_size;
786   rrdc_response_t *res;
787   int status;
788   char file_path[PATH_MAX];
790   if (filename == NULL)
791     return (-1);
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);
798   if (status != 0)
799     return (ENOBUFS);
801   pthread_mutex_lock (&lock);
802   filename = get_path (filename, file_path);
803   if (filename == NULL)
804   {
805     pthread_mutex_unlock (&lock);
806     return (-1);
807   }
809   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
810   if (status != 0)
811   {
812     pthread_mutex_unlock (&lock);
813     return (ENOBUFS);
814   }
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';
821   res = NULL;
822   status = request (buffer, buffer_size, &res);
823   pthread_mutex_unlock (&lock);
825   if (status != 0)
826     return (status);
828   status = res->status;
829   response_free (res);
831   return (status);
832 } /* }}} int rrdc_flush */
834 rrd_info_t * rrdc_info (const char *filename) /* {{{ */
836   char buffer[RRD_CMD_MAX];
837   char *buffer_ptr;
838   size_t buffer_free;
839   size_t buffer_size;
840   rrdc_response_t *res;
841   int status;
842   char file_path[PATH_MAX];
843   rrd_info_t *data = NULL, *cd;
844   rrd_infoval_t info;
845   unsigned int l;
846   rrd_info_type_t itype;
847   char *k, *s;
849   if (filename == NULL) {
850     rrd_set_error ("rrdc_info: no filename");
851     return (NULL);
852   }
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);
859   if (status != 0) {
860     rrd_set_error ("rrdc_info: out of memory");
861     return (NULL);
862   }
864   pthread_mutex_lock (&lock);
865   filename = get_path (filename, file_path);
866   if (filename == NULL)
867   {
868     pthread_mutex_unlock (&lock);
869     return (NULL);
870   }
872   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
873   if (status != 0)
874   {
875     pthread_mutex_unlock (&lock);
876     rrd_set_error ("rrdc_info: out of memory");
877     return (NULL);
878   }
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';
885   res = NULL;
886   status = request (buffer, buffer_size, &res);
887   pthread_mutex_unlock (&lock);
889   if (status != 0) {
890     rrd_set_error ("rrdcached: %s", res->message);
891     return (NULL);
892   }
893   data = cd = NULL;
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; }
898         }
899     if(!s || !*s) break;
900         itype = atoi(s); /* extract type code */
901         for(;*s;s++) { if(*s == ' ') { *s = 0; s++; break; } }
902     if(!*s) break;
903     /* finally, we're pointing to the value */
904     switch(itype) {
905     case RD_I_VAL:
906         if(*s == 'N') { info.u_val = DNAN; } else { info.u_val = atof(s); }
907         break;
908     case RD_I_CNT:
909         info.u_cnt = atol(s);
910         break;
911     case RD_I_INT:
912         info.u_int = atoi(s);
913         break;
914     case RD_I_STR:
915         chomp(s);
916         info.u_str = (char*)malloc(sizeof(char) * (strlen(s) + 1));
917         strcpy(info.u_str,s);
918         break;
919     case RD_I_BLO:
920         rrd_set_error ("rrdc_info: BLOB objects are not supported");
921         return (NULL);
922     default:
923         rrd_set_error ("rrdc_info: Unsupported info type %d",itype);
924         return (NULL);
925     }
926         
927     cd = rrd_info_push(cd, sprintf_alloc("%s",k), itype, info);
928         if(!data) data = cd;
929   }
930   response_free (res);
932   return (data);
933 } /* }}} int rrdc_info */
935 time_t rrdc_last (const char *filename) /* {{{ */
937   char buffer[RRD_CMD_MAX];
938   char *buffer_ptr;
939   size_t buffer_free;
940   size_t buffer_size;
941   rrdc_response_t *res;
942   int status;
943   char file_path[PATH_MAX];
944   time_t lastup;
946   if (filename == NULL) {
947     rrd_set_error ("rrdc_last: no filename");
948     return (-1);
949   }
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);
956   if (status != 0) {
957     rrd_set_error ("rrdc_last: out of memory");
958     return (-1);
959   }
961   pthread_mutex_lock (&lock);
962   filename = get_path (filename, file_path);
963   if (filename == NULL)
964   {
965     pthread_mutex_unlock (&lock);
966     return (-1);
967   }
969   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
970   if (status != 0)
971   {
972     pthread_mutex_unlock (&lock);
973     rrd_set_error ("rrdc_last: out of memory");
974     return (-1);
975   }
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';
982   res = NULL;
983   status = request (buffer, buffer_size, &res);
984   pthread_mutex_unlock (&lock);
986   if (status != 0) {
987     rrd_set_error ("rrdcached: %s", res->message);
988     return (-1);
989   }
990   lastup = atol(res->message);
991   response_free (res);
993   return (lastup);
994 } /* }}} int rrdc_last */
996 time_t rrdc_first (const char *filename, int rraindex) /* {{{ */
998   char buffer[RRD_CMD_MAX];
999   char *buffer_ptr;
1000   size_t buffer_free;
1001   size_t buffer_size;
1002   rrdc_response_t *res;
1003   int status;
1004   char file_path[PATH_MAX];
1005   time_t firstup;
1007   if (filename == NULL) {
1008     rrd_set_error ("rrdc_first: no filename specified");
1009     return (-1);
1010   }
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);
1017   if (status != 0) {
1018     rrd_set_error ("rrdc_first: out of memory");
1019     return (-1);
1020   }
1022   pthread_mutex_lock (&lock);
1023   filename = get_path (filename, file_path);
1024   if (filename == NULL)
1025   {
1026     pthread_mutex_unlock (&lock);
1027     return (-1);
1028   }
1030   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
1031   if (status != 0)
1032   {
1033     pthread_mutex_unlock (&lock);
1034     rrd_set_error ("rrdc_first: out of memory");
1035     return (-1);
1036   }
1037   status = buffer_add_ulong (rraindex, &buffer_ptr, &buffer_free);
1038   if (status != 0)
1039   {
1040     pthread_mutex_unlock (&lock);
1041     rrd_set_error ("rrdc_first: out of memory");
1042     return (-1);
1043   }
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';
1050   res = NULL;
1051   status = request (buffer, buffer_size, &res);
1052   pthread_mutex_unlock (&lock);
1054   if (status != 0) {
1055     rrd_set_error ("rrdcached: %s", res->message);
1056     return (-1);
1057   }
1058   firstup = atol(res->message);
1059   response_free (res);
1061   return (firstup);
1062 } /* }}} int rrdc_first */
1064 int rrdc_create (const char *filename, /* {{{ */
1065     unsigned long pdp_step,
1066     time_t last_up,
1067     int no_overwrite,
1068     int argc,
1069     const char **argv)
1071   char buffer[RRD_CMD_MAX];
1072   char *buffer_ptr;
1073   size_t buffer_free;
1074   size_t buffer_size;
1075   rrdc_response_t *res;
1076   int status;
1077   char file_path[PATH_MAX];
1078   int i;
1080   if (filename == NULL) {
1081     rrd_set_error ("rrdc_create: no filename specified");
1082     return (-1);
1083   }
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);
1090   if (status != 0) {
1091     rrd_set_error ("rrdc_create: out of memory");
1092     return (-1);
1093   }
1095   pthread_mutex_lock (&lock);
1096   filename = get_path (filename, file_path);
1097   if (filename == NULL)
1098   {
1099     pthread_mutex_unlock (&lock);
1100     return (-1);
1101   }
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);
1108   if(no_overwrite) {
1109     status = buffer_add_string ("-O", &buffer_ptr, &buffer_free);
1110   }
1111   if (status != 0)
1112   {
1113     pthread_mutex_unlock (&lock);
1114     rrd_set_error ("rrdc_create: out of memory");
1115     return (-1);
1116   }
1118   for( i=0; i<argc; i++ ) {
1119     if( argv[i] ) {
1120       status = buffer_add_string (argv[i], &buffer_ptr, &buffer_free);
1121       if (status != 0)
1122       {
1123         pthread_mutex_unlock (&lock);
1124         rrd_set_error ("rrdc_create: out of memory");
1125         return (-1);
1126       }
1127         }
1128   }
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';
1136   res = NULL;
1137   status = request (buffer, buffer_size, &res);
1138   pthread_mutex_unlock (&lock);
1140   if (status != 0) {
1141     rrd_set_error ("rrdcached: %s", res->message);
1142     return (-1);
1143   }
1144   response_free (res);
1145   return(0);
1146 } /* }}} int rrdc_create */
1148 int rrdc_fetch (const char *filename, /* {{{ */
1149     const char *cf,
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];
1157   char *buffer_ptr;
1158   size_t buffer_free;
1159   size_t buffer_size;
1160   rrdc_response_t *res;
1161   char path_buffer[PATH_MAX];
1162   const char *path_ptr;
1164   char *str_tmp;
1165   unsigned long flush_version;
1167   time_t start;
1168   time_t end;
1169   unsigned long step;
1170   unsigned long ds_num;
1171   char **ds_names;
1173   rrd_value_t *data;
1174   size_t data_size;
1175   size_t data_fill;
1177   int status;
1178   size_t current_line;
1179   time_t t;
1181   if ((filename == NULL) || (cf == NULL))
1182     return (-1);
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);
1190   if (status != 0)
1191     return (ENOBUFS);
1193   /* change to path for rrdcached */
1194   path_ptr = get_path (filename, path_buffer);
1195   if (path_ptr == NULL)
1196     return (EINVAL);
1198   status = buffer_add_string (path_ptr, &buffer_ptr, &buffer_free);
1199   if (status != 0)
1200     return (ENOBUFS);
1202   status = buffer_add_string (cf, &buffer_ptr, &buffer_free);
1203   if (status != 0)
1204     return (ENOBUFS);
1206   if ((ret_start != NULL) && (*ret_start > 0))
1207   {
1208     char tmp[64];
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);
1212     if (status != 0)
1213       return (ENOBUFS);
1215     if ((ret_end != NULL) && (*ret_end > 0))
1216     {
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);
1220       if (status != 0)
1221         return (ENOBUFS);
1222     }
1223   }
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';
1230   res = NULL;
1231   status = request (buffer, buffer_size, &res);
1232   if (status != 0)
1233     return (status);
1235   status = res->status;
1236   if (status < 0)
1237   {
1238     rrd_set_error ("rrdcached: %s", res->message);
1239     response_free (res);
1240     return (status);
1241   }
1242   /* }}} Send request */
1244   ds_names = NULL;
1245   ds_num = 0;
1246   data = NULL;
1247   current_line = 0;
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__); \
1254     free (data); \
1255     if (ds_names != 0) { size_t k; for (k = 0; k < ds_num; k++) free (ds_names[k]); } \
1256     free (ds_names); \
1257     response_free (res); \
1258     return (status); \
1259   } while (0)
1261 #define READ_NUMERIC_FIELD(name,type,var) do { \
1262     char *key; \
1263     unsigned long value; \
1264     assert (current_line < res->lines_num); \
1265     status = parse_ulong_header (res->lines[current_line], &key, &value); \
1266     if (status != 0) \
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; \
1271     current_line++; \
1272   } while (0)
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.",
1284         flush_version);
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);
1291   if (start >= 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);
1297   if (step < 1)
1298     BAIL_OUT (-1, "Invalid number for Step: %lu", step);
1300   READ_NUMERIC_FIELD ("DSCount", unsigned long, ds_num);
1301   if (ds_num < 1)
1302     BAIL_OUT (-1, "Invalid number for DSCount: %lu", ds_num);
1303   
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);
1311   if (status != 0)
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);
1315   current_line++;
1317   data_size = ds_num * (end - start) / step;
1318   if (data_size < 1)
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));
1326   if (data == NULL)
1327     BAIL_OUT (-1, "Out of memory");
1328   
1330   data_fill = 0;
1331   for (t = start + step; t <= end; t += step, current_line++)
1332   {
1333     time_t tmp;
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);
1339     if (status != 0)
1340       BAIL_OUT (-1, "Cannot parse value line");
1342     data_fill += (size_t) ds_num;
1343   }
1345   *ret_start = start;
1346   *ret_end = end;
1347   *ret_step = step;
1348   *ret_ds_num = ds_num;
1349   *ret_ds_names = ds_names;
1350   *ret_data = data;
1352   response_free (res);
1353   return (0);
1354 #undef READ_NUMERIC_FIELD
1355 #undef BAIL_OUT
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
1360  */
1361 int rrdc_flush_if_daemon (const char *opt_daemon, const char *filename) /* {{{ */
1363   int status = 0;
1365   rrdc_connect(opt_daemon);
1367   if (rrdc_is_connected(opt_daemon))
1368   {
1369     rrd_clear_error();
1370     status = rrdc_flush (filename);
1372     if (status != 0 && !rrd_test_error())
1373     {
1374       if (status > 0)
1375       {
1376         rrd_set_error("rrdc_flush (%s) failed: %s",
1377                       filename, rrd_strerror(status));
1378       }
1379       else if (status < 0)
1380       {
1381         rrd_set_error("rrdc_flush (%s) failed with status %i.",
1382                       filename, status);
1383       }
1384     }
1385   } /* if (rrdc_is_connected(..)) */
1387   return status;
1388 } /* }}} int rrdc_flush_if_daemon */
1391 int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */
1393   rrdc_stats_t *head;
1394   rrdc_stats_t *tail;
1396   rrdc_response_t *res;
1398   int status;
1399   size_t i;
1401   /* Protocol example: {{{
1402    * ->  STATS
1403    * <-  5 Statistics follow
1404    * <-  QueueLength: 0
1405    * <-  UpdatesWritten: 0
1406    * <-  DataSetsWritten: 0
1407    * <-  TreeNodesNumber: 0
1408    * <-  TreeDepth: 0
1409    * }}} */
1411   res = NULL;
1412   pthread_mutex_lock (&lock);
1413   status = request ("STATS\n", strlen ("STATS\n"), &res);
1414   pthread_mutex_unlock (&lock);
1416   if (status != 0)
1417     return (status);
1419   if (res->status <= 0)
1420   {
1421     response_free (res);
1422     return (EIO);
1423   }
1425   head = NULL;
1426   tail = NULL;
1427   for (i = 0; i < res->lines_num; i++)
1428   {
1429     char *key;
1430     char *value;
1431     char *endptr;
1432     rrdc_stats_t *s;
1434     key = res->lines[i];
1435     value = strchr (key, ':');
1436     if (value == NULL)
1437       continue;
1438     *value = 0;
1439     value++;
1441     while ((value[0] == ' ') || (value[0] == '\t'))
1442       value++;
1444     s = (rrdc_stats_t *) malloc (sizeof (rrdc_stats_t));
1445     if (s == NULL)
1446       continue;
1447     memset (s, 0, sizeof (*s));
1449     s->name = strdup (key);
1451     endptr = NULL;
1452     if ((strcmp ("QueueLength", key) == 0)
1453         || (strcmp ("TreeDepth", key) == 0)
1454         || (strcmp ("TreeNodesNumber", key) == 0))
1455     {
1456       s->type = RRDC_STATS_TYPE_GAUGE;
1457       s->value.gauge = strtod (value, &endptr);
1458     }
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))
1465     {
1466       s->type = RRDC_STATS_TYPE_COUNTER;
1467       s->value.counter = (uint64_t) strtoll (value, &endptr, /* base = */ 0);
1468     }
1469     else
1470     {
1471       free (s);
1472       continue;
1473     }
1475     /* Conversion failed */
1476     if (endptr == value)
1477     {
1478       free (s);
1479       continue;
1480     }
1482     if (head == NULL)
1483     {
1484       head = s;
1485       tail = s;
1486       s->next = NULL;
1487     }
1488     else
1489     {
1490       tail->next = s;
1491       tail = s;
1492     }
1493   } /* for (i = 0; i < res->lines_num; i++) */
1495   response_free (res);
1497   if (head == NULL)
1498     return (EPROTO);
1500   *ret_stats = head;
1501   return (0);
1502 } /* }}} int rrdc_stats_get */
1504 void rrdc_stats_free (rrdc_stats_t *ret_stats) /* {{{ */
1506   rrdc_stats_t *this;
1508   this = ret_stats;
1509   while (this != NULL)
1510   {
1511     rrdc_stats_t *next;
1513     next = this->next;
1515     if (this->name != NULL)
1516     {
1517       free ((char *)this->name);
1518       this->name = NULL;
1519     }
1520     free (this);
1522     this = next;
1523   } /* while (this != NULL) */
1524 } /* }}} void rrdc_stats_free */
1526 /*
1527  * vim: set sw=2 sts=2 ts=8 et fdm=marker :
1528  */