Code

Imported upstream version 1.4.2.
[pkg-rrdtool.git] / src / rrd_client.c
1 /**
2  * RRDTool - src/rrd_client.c
3  * Copyright (C) 2008 Florian octo Forster
4  *
5  * This program is free software; you can redistribute it and/or modify it
6  * under the terms of the GNU General Public License as published by the
7  * Free Software Foundation; only version 2 of the License is applicable.
8  *
9  * This program is distributed in the hope that it will be useful, but
10  * WITHOUT ANY WARRANTY; without even the implied warranty of
11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
12  * General Public License for more details.
13  *
14  * You should have received a copy of the GNU General Public License along
15  * with this program; if not, write to the Free Software Foundation, Inc.,
16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
17  *
18  * Authors:
19  *   Florian octo Forster <octo at verplant.org>
20  *   Sebastian tokkee Harl <sh at tokkee.org>
21  **/
23 #include "rrd.h"
24 #include "rrd_tool.h"
25 #include "rrd_client.h"
27 #include <stdio.h>
28 #include <stdlib.h>
29 #include <string.h>
30 #include <strings.h>
31 #include <errno.h>
32 #include <assert.h>
33 #include <pthread.h>
34 #include <sys/types.h>
35 #include <sys/socket.h>
36 #include <sys/un.h>
37 #include <netdb.h>
38 #include <limits.h>
40 #ifndef ENODATA
41 #define ENODATA ENOENT
42 #endif
44 struct rrdc_response_s
45 {
46   int status;
47   char *message;
48   char **lines;
49   size_t lines_num;
50 };
51 typedef struct rrdc_response_s rrdc_response_t;
53 static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
54 static int sd = -1;
55 static FILE *sh = NULL;
56 static char *sd_path = NULL; /* cache the path for sd */
58 /* get_path: Return a path name appropriate to be sent to the daemon.
59  *
60  * When talking to a local daemon (thru a UNIX socket), relative path names
61  * are resolved to absolute path names to allow for transparent integration
62  * into existing solutions (as requested by Tobi). Else, absolute path names
63  * are not allowed, since path name translation is done by the server.
64  *
65  * One must hold `lock' when calling this function. */
66 static const char *get_path (const char *path, char *resolved_path) /* {{{ */
67 {
68   const char *ret = path;
69   int is_unix = 0;
71   if ((*sd_path == '/')
72       || (strncmp ("unix:", sd_path, strlen ("unix:")) == 0))
73     is_unix = 1;
75   if (*path == '/') /* absolute path */
76   {
77     if (! is_unix)
78     {
79       rrd_set_error ("absolute path names not allowed when talking "
80           "to a remote daemon");
81       return (NULL);
82     }
83     /* else: nothing to do */
84   }
85   else /* relative path */
86   {
87     if (is_unix)
88     {
89       realpath (path, resolved_path);
90       ret = resolved_path;
91     }
92     /* else: nothing to do */
93   }
94   return (ret);
95 } /* }}} char *get_path */
97 /* One must hold `lock' when calling `close_connection'. */
98 static void close_connection (void) /* {{{ */
99 {
100   if (sh != NULL)
101   {
102     fclose (sh);
103     sh = NULL;
104     sd = -1;
105   }
106   else if (sd >= 0)
107   {
108     close (sd);
109     sd = -1;
110   }
112   if (sd_path != NULL)
113     free (sd_path);
114   sd_path = NULL;
115 } /* }}} void close_connection */
117 static int buffer_add_string (const char *str, /* {{{ */
118     char **buffer_ret, size_t *buffer_size_ret)
120   char *buffer;
121   size_t buffer_size;
122   size_t buffer_pos;
123   size_t i;
124   int status;
126   buffer = *buffer_ret;
127   buffer_size = *buffer_size_ret;
128   buffer_pos = 0;
130   i = 0;
131   status = -1;
132   while (buffer_pos < buffer_size)
133   {
134     if (str[i] == 0)
135     {
136       buffer[buffer_pos] = ' ';
137       buffer_pos++;
138       status = 0;
139       break;
140     }
141     else if ((str[i] == ' ') || (str[i] == '\\'))
142     {
143       if (buffer_pos >= (buffer_size - 1))
144         break;
145       buffer[buffer_pos] = '\\';
146       buffer_pos++;
147       buffer[buffer_pos] = str[i];
148       buffer_pos++;
149     }
150     else
151     {
152       buffer[buffer_pos] = str[i];
153       buffer_pos++;
154     }
155     i++;
156   } /* while (buffer_pos < buffer_size) */
158   if (status != 0)
159     return (-1);
161   *buffer_ret = buffer + buffer_pos;
162   *buffer_size_ret = buffer_size - buffer_pos;
164   return (0);
165 } /* }}} int buffer_add_string */
167 static int buffer_add_value (const char *value, /* {{{ */
168     char **buffer_ret, size_t *buffer_size_ret)
170   char temp[4096];
172   if (strncmp (value, "N:", 2) == 0)
173     snprintf (temp, sizeof (temp), "%lu:%s",
174         (unsigned long) time (NULL), value + 2);
175   else
176     strncpy (temp, value, sizeof (temp));
177   temp[sizeof (temp) - 1] = 0;
179   return (buffer_add_string (temp, buffer_ret, buffer_size_ret));
180 } /* }}} int buffer_add_value */
182 /* Remove trailing newline (NL) and carriage return (CR) characters. Similar to
183  * the Perl function `chomp'. Returns the number of characters that have been
184  * removed. */
185 static int chomp (char *str) /* {{{ */
187   size_t len;
188   int removed;
190   if (str == NULL)
191     return (-1);
193   len = strlen (str);
194   removed = 0;
195   while ((len > 0) && ((str[len - 1] == '\n') || (str[len - 1] == '\r')))
196   {
197     str[len - 1] = 0;
198     len--;
199     removed++;
200   }
202   return (removed);
203 } /* }}} int chomp */
205 static void response_free (rrdc_response_t *res) /* {{{ */
207   if (res == NULL)
208     return;
210   if (res->lines != NULL)
211   {
212     size_t i;
214     for (i = 0; i < res->lines_num; i++)
215       if (res->lines[i] != NULL)
216         free (res->lines[i]);
217     free (res->lines);
218   }
220   free (res);
221 } /* }}} void response_free */
223 static int response_read (rrdc_response_t **ret_response) /* {{{ */
225   rrdc_response_t *ret;
227   char buffer[4096];
228   char *buffer_ptr;
230   size_t i;
232   if (sh == NULL)
233     return (-1);
235   ret = (rrdc_response_t *) malloc (sizeof (rrdc_response_t));
236   if (ret == NULL)
237     return (-2);
238   memset (ret, 0, sizeof (*ret));
239   ret->lines = NULL;
240   ret->lines_num = 0;
242   buffer_ptr = fgets (buffer, sizeof (buffer), sh);
243   if (buffer_ptr == NULL)
244     return (-3);
245   chomp (buffer);
247   ret->status = strtol (buffer, &ret->message, 0);
248   if (buffer == ret->message)
249   {
250     response_free (ret);
251     return (-4);
252   }
253   /* Skip leading whitespace of the status message */
254   ret->message += strspn (ret->message, " \t");
256   if (ret->status <= 0)
257   {
258     if (ret->status < 0)
259       rrd_set_error("rrdcached: %s", ret->message);
260     *ret_response = ret;
261     return (0);
262   }
264   ret->lines = (char **) malloc (sizeof (char *) * ret->status);
265   if (ret->lines == NULL)
266   {
267     response_free (ret);
268     return (-5);
269   }
270   memset (ret->lines, 0, sizeof (char *) * ret->status);
271   ret->lines_num = (size_t) ret->status;
273   for (i = 0; i < ret->lines_num; i++)
274   {
275     buffer_ptr = fgets (buffer, sizeof (buffer), sh);
276     if (buffer_ptr == NULL)
277     {
278       response_free (ret);
279       return (-6);
280     }
281     chomp (buffer);
283     ret->lines[i] = strdup (buffer);
284     if (ret->lines[i] == NULL)
285     {
286       response_free (ret);
287       return (-7);
288     }
289   }
291   *ret_response = ret;
292   return (0);
293 } /* }}} rrdc_response_t *response_read */
295 static int request (const char *buffer, size_t buffer_size, /* {{{ */
296     rrdc_response_t **ret_response)
298   int status;
299   rrdc_response_t *res;
301   if (sh == NULL)
302     return (ENOTCONN);
304   status = (int) fwrite (buffer, buffer_size, /* nmemb = */ 1, sh);
305   if (status != 1)
306   {
307     close_connection ();
308     rrd_set_error("request: socket error (%d) while talking to rrdcached",
309                   status);
310     return (-1);
311   }
312   fflush (sh);
314   res = NULL;
315   status = response_read (&res);
317   if (status != 0)
318   {
319     if (status < 0)
320       rrd_set_error("request: internal error while talking to rrdcached");
321     return (status);
322   }
324   *ret_response = res;
325   return (0);
326 } /* }}} int request */
328 /* determine whether we are connected to the specified daemon_addr if
329  * NULL, return whether we are connected at all
330  */
331 int rrdc_is_connected(const char *daemon_addr) /* {{{ */
333   if (sd < 0)
334     return 0;
335   else if (daemon_addr == NULL)
336   {
337     /* here we have to handle the case i.e.
338      *   UPDATE --daemon ...; UPDATEV (no --daemon) ...
339      * In other words: we have a cached connection,
340      * but it is not specified in the current command.
341      * Daemon is only implied in this case if set in ENV
342      */
343     if (getenv(ENV_RRDCACHED_ADDRESS) != NULL)
344       return 1;
345     else
346       return 0;
347   }
348   else if (strcmp(daemon_addr, sd_path) == 0)
349     return 1;
350   else
351     return 0;
353 } /* }}} int rrdc_is_connected */
355 static int rrdc_connect_unix (const char *path) /* {{{ */
357   struct sockaddr_un sa;
358   int status;
360   assert (path != NULL);
361   assert (sd == -1);
363   sd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
364   if (sd < 0)
365   {
366     status = errno;
367     return (status);
368   }
370   memset (&sa, 0, sizeof (sa));
371   sa.sun_family = AF_UNIX;
372   strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
374   status = connect (sd, (struct sockaddr *) &sa, sizeof (sa));
375   if (status != 0)
376   {
377     status = errno;
378     close_connection ();
379     return (status);
380   }
382   sh = fdopen (sd, "r+");
383   if (sh == NULL)
384   {
385     status = errno;
386     close_connection ();
387     return (status);
388   }
390   return (0);
391 } /* }}} int rrdc_connect_unix */
393 static int rrdc_connect_network (const char *addr_orig) /* {{{ */
395   struct addrinfo ai_hints;
396   struct addrinfo *ai_res;
397   struct addrinfo *ai_ptr;
398   char addr_copy[NI_MAXHOST];
399   char *addr;
400   char *port;
402   assert (addr_orig != NULL);
403   assert (sd == -1);
405   strncpy(addr_copy, addr_orig, sizeof(addr_copy));
406   addr_copy[sizeof(addr_copy) - 1] = '\0';
407   addr = addr_copy;
409   int status;
410   memset (&ai_hints, 0, sizeof (ai_hints));
411   ai_hints.ai_flags = 0;
412 #ifdef AI_ADDRCONFIG
413   ai_hints.ai_flags |= AI_ADDRCONFIG;
414 #endif
415   ai_hints.ai_family = AF_UNSPEC;
416   ai_hints.ai_socktype = SOCK_STREAM;
418   port = NULL;
419   if (*addr == '[') /* IPv6+port format */
420   {
421     /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
422     addr++;
424     port = strchr (addr, ']');
425     if (port == NULL)
426     {
427       rrd_set_error("malformed address: %s", addr_orig);
428       return (-1);
429     }
430     *port = 0;
431     port++;
433     if (*port == ':')
434       port++;
435     else if (*port == 0)
436       port = NULL;
437     else
438     {
439       rrd_set_error("garbage after address: %s", port);
440       return (-1);
441     }
442   } /* if (*addr == '[') */
443   else
444   {
445     port = rindex(addr, ':');
446     if (port != NULL)
447     {
448       *port = 0;
449       port++;
450     }
451   }
453   ai_res = NULL;
454   status = getaddrinfo (addr,
455                         port == NULL ? RRDCACHED_DEFAULT_PORT : port,
456                         &ai_hints, &ai_res);
457   if (status != 0)
458   {
459     rrd_set_error ("failed to resolve address `%s' (port %s): %s",
460         addr, port == NULL ? RRDCACHED_DEFAULT_PORT : port,
461         gai_strerror (status));
462     return (-1);
463   }
465   for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
466   {
467     sd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
468     if (sd < 0)
469     {
470       status = errno;
471       sd = -1;
472       continue;
473     }
475     status = connect (sd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
476     if (status != 0)
477     {
478       status = errno;
479       close_connection();
480       continue;
481     }
483     sh = fdopen (sd, "r+");
484     if (sh == NULL)
485     {
486       status = errno;
487       close_connection ();
488       continue;
489     }
491     assert (status == 0);
492     break;
493   } /* for (ai_ptr) */
495   return (status);
496 } /* }}} int rrdc_connect_network */
498 int rrdc_connect (const char *addr) /* {{{ */
500   int status = 0;
502   if (addr == NULL)
503     addr = getenv (ENV_RRDCACHED_ADDRESS);
505   if (addr == NULL)
506     return 0;
508   pthread_mutex_lock(&lock);
510   if (sd >= 0 && sd_path != NULL && strcmp(addr, sd_path) == 0)
511   {
512     /* connection to the same daemon; use cached connection */
513     pthread_mutex_unlock (&lock);
514     return (0);
515   }
516   else
517   {
518     close_connection();
519   }
521   rrd_clear_error ();
522   if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
523     status = rrdc_connect_unix (addr + strlen ("unix:"));
524   else if (addr[0] == '/')
525     status = rrdc_connect_unix (addr);
526   else
527     status = rrdc_connect_network(addr);
529   if (status == 0 && sd >= 0)
530     sd_path = strdup(addr);
531   else
532   {
533     char *err = rrd_test_error () ? rrd_get_error () : "Internal error";
534     /* err points the string that gets written to by rrd_set_error(), thus we
535      * cannot pass it to that function */
536     err = strdup (err);
537     rrd_set_error("Unable to connect to rrdcached: %s",
538                   (status < 0)
539                   ? (err ? err : "Internal error")
540                   : rrd_strerror (status));
541     if (err != NULL)
542       free (err);
543   }
545   pthread_mutex_unlock (&lock);
546   return (status);
547 } /* }}} int rrdc_connect */
549 int rrdc_disconnect (void) /* {{{ */
551   pthread_mutex_lock (&lock);
553   close_connection();
555   pthread_mutex_unlock (&lock);
557   return (0);
558 } /* }}} int rrdc_disconnect */
560 int rrdc_update (const char *filename, int values_num, /* {{{ */
561                 const char * const *values)
563   char buffer[4096];
564   char *buffer_ptr;
565   size_t buffer_free;
566   size_t buffer_size;
567   rrdc_response_t *res;
568   int status;
569   int i;
570   char file_path[PATH_MAX];
572   memset (buffer, 0, sizeof (buffer));
573   buffer_ptr = &buffer[0];
574   buffer_free = sizeof (buffer);
576   status = buffer_add_string ("update", &buffer_ptr, &buffer_free);
577   if (status != 0)
578     return (ENOBUFS);
580   pthread_mutex_lock (&lock);
581   filename = get_path (filename, file_path);
582   if (filename == NULL)
583   {
584     pthread_mutex_unlock (&lock);
585     return (-1);
586   }
588   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
589   if (status != 0)
590   {
591     pthread_mutex_unlock (&lock);
592     return (ENOBUFS);
593   }
595   for (i = 0; i < values_num; i++)
596   {
597     status = buffer_add_value (values[i], &buffer_ptr, &buffer_free);
598     if (status != 0)
599     {
600       pthread_mutex_unlock (&lock);
601       return (ENOBUFS);
602     }
603   }
605   assert (buffer_free < sizeof (buffer));
606   buffer_size = sizeof (buffer) - buffer_free;
607   assert (buffer[buffer_size - 1] == ' ');
608   buffer[buffer_size - 1] = '\n';
610   res = NULL;
611   status = request (buffer, buffer_size, &res);
612   pthread_mutex_unlock (&lock);
614   if (status != 0)
615     return (status);
617   status = res->status;
618   response_free (res);
620   return (status);
621 } /* }}} int rrdc_update */
623 int rrdc_flush (const char *filename) /* {{{ */
625   char buffer[4096];
626   char *buffer_ptr;
627   size_t buffer_free;
628   size_t buffer_size;
629   rrdc_response_t *res;
630   int status;
631   char file_path[PATH_MAX];
633   if (filename == NULL)
634     return (-1);
636   memset (buffer, 0, sizeof (buffer));
637   buffer_ptr = &buffer[0];
638   buffer_free = sizeof (buffer);
640   status = buffer_add_string ("flush", &buffer_ptr, &buffer_free);
641   if (status != 0)
642     return (ENOBUFS);
644   pthread_mutex_lock (&lock);
645   filename = get_path (filename, file_path);
646   if (filename == NULL)
647   {
648     pthread_mutex_unlock (&lock);
649     return (-1);
650   }
652   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
653   if (status != 0)
654   {
655     pthread_mutex_unlock (&lock);
656     return (ENOBUFS);
657   }
659   assert (buffer_free < sizeof (buffer));
660   buffer_size = sizeof (buffer) - buffer_free;
661   assert (buffer[buffer_size - 1] == ' ');
662   buffer[buffer_size - 1] = '\n';
664   res = NULL;
665   status = request (buffer, buffer_size, &res);
666   pthread_mutex_unlock (&lock);
668   if (status != 0)
669     return (status);
671   status = res->status;
672   response_free (res);
674   return (status);
675 } /* }}} int rrdc_flush */
678 /* convenience function; if there is a daemon specified, or if we can
679  * detect one from the environment, then flush the file.  Otherwise, no-op
680  */
681 int rrdc_flush_if_daemon (const char *opt_daemon, const char *filename) /* {{{ */
683   int status = 0;
685   rrdc_connect(opt_daemon);
687   if (rrdc_is_connected(opt_daemon))
688   {
689     rrd_clear_error();
690     status = rrdc_flush (filename);
692     if (status != 0 && !rrd_test_error())
693     {
694       if (status > 0)
695       {
696         rrd_set_error("rrdc_flush (%s) failed: %s",
697                       filename, rrd_strerror(status));
698       }
699       else if (status < 0)
700       {
701         rrd_set_error("rrdc_flush (%s) failed with status %i.",
702                       filename, status);
703       }
704     }
705   } /* if (rrdc_is_connected(..)) */
707   return status;
708 } /* }}} int rrdc_flush_if_daemon */
711 int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */
713   rrdc_stats_t *head;
714   rrdc_stats_t *tail;
716   rrdc_response_t *res;
718   int status;
719   size_t i;
721   /* Protocol example: {{{
722    * ->  STATS
723    * <-  5 Statistics follow
724    * <-  QueueLength: 0
725    * <-  UpdatesWritten: 0
726    * <-  DataSetsWritten: 0
727    * <-  TreeNodesNumber: 0
728    * <-  TreeDepth: 0
729    * }}} */
731   res = NULL;
732   pthread_mutex_lock (&lock);
733   status = request ("STATS\n", strlen ("STATS\n"), &res);
734   pthread_mutex_unlock (&lock);
736   if (status != 0)
737     return (status);
739   if (res->status <= 0)
740   {
741     response_free (res);
742     return (EIO);
743   }
745   head = NULL;
746   tail = NULL;
747   for (i = 0; i < res->lines_num; i++)
748   {
749     char *key;
750     char *value;
751     char *endptr;
752     rrdc_stats_t *s;
754     key = res->lines[i];
755     value = strchr (key, ':');
756     if (value == NULL)
757       continue;
758     *value = 0;
759     value++;
761     while ((value[0] == ' ') || (value[0] == '\t'))
762       value++;
764     s = (rrdc_stats_t *) malloc (sizeof (rrdc_stats_t));
765     if (s == NULL)
766       continue;
767     memset (s, 0, sizeof (*s));
769     s->name = strdup (key);
771     endptr = NULL;
772     if ((strcmp ("QueueLength", key) == 0)
773         || (strcmp ("TreeDepth", key) == 0)
774         || (strcmp ("TreeNodesNumber", key) == 0))
775     {
776       s->type = RRDC_STATS_TYPE_GAUGE;
777       s->value.gauge = strtod (value, &endptr);
778     }
779     else if ((strcmp ("DataSetsWritten", key) == 0)
780         || (strcmp ("FlushesReceived", key) == 0)
781         || (strcmp ("JournalBytes", key) == 0)
782         || (strcmp ("JournalRotate", key) == 0)
783         || (strcmp ("UpdatesReceived", key) == 0)
784         || (strcmp ("UpdatesWritten", key) == 0))
785     {
786       s->type = RRDC_STATS_TYPE_COUNTER;
787       s->value.counter = (uint64_t) strtoll (value, &endptr, /* base = */ 0);
788     }
789     else
790     {
791       free (s);
792       continue;
793     }
795     /* Conversion failed */
796     if (endptr == value)
797     {
798       free (s);
799       continue;
800     }
802     if (head == NULL)
803     {
804       head = s;
805       tail = s;
806       s->next = NULL;
807     }
808     else
809     {
810       tail->next = s;
811       tail = s;
812     }
813   } /* for (i = 0; i < res->lines_num; i++) */
815   response_free (res);
817   if (head == NULL)
818     return (EPROTO);
820   *ret_stats = head;
821   return (0);
822 } /* }}} int rrdc_stats_get */
824 void rrdc_stats_free (rrdc_stats_t *ret_stats) /* {{{ */
826   rrdc_stats_t *this;
828   this = ret_stats;
829   while (this != NULL)
830   {
831     rrdc_stats_t *next;
833     next = this->next;
835     if (this->name != NULL)
836     {
837       free ((char *)this->name);
838       this->name = NULL;
839     }
840     free (this);
842     this = next;
843   } /* while (this != NULL) */
844 } /* }}} void rrdc_stats_free */
846 /*
847  * vim: set sw=2 sts=2 ts=8 et fdm=marker :
848  */