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  **/
  27
  28 #include "rrd.h"
  29 #include "rrd_tool.h"
  30 #include "rrd_client.h"
  31
  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>
  44
  45 #ifndef ENODATA
  46 #define ENODATA ENOENT
  47 #endif
  48
  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;
  57
  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 */
  62
  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;
  75
  76   if ((path == NULL) || (resolved_path == NULL) || (sd_path == NULL))
  77     return (NULL);
  78
  79   if ((*sd_path == '/')
  80       || (strncmp ("unix:", sd_path, strlen ("unix:")) == 0))
  81     is_unix = 1;
  82
  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   }
  99
 100   return path;
 101 } /* }}} char *get_path */
 102
 103 static size_t strsplit (char *string, char **fields, size_t size) /* {{{ */
 104 {
 105   size_t i;
 106   char *ptr;
 107   char *saveptr;
 108
 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++;
 116
 117     if (i >= size)
 118       break;
 119   }
 120
 121   return (i);
 122 } /* }}} size_t strsplit */
 123
 124 static int parse_header (char *line, /* {{{ */
 125     char **ret_key, char **ret_value)
 126 {
 127   char *tmp;
 128
 129   *ret_key = line;
 130
 131   tmp = strchr (line, ':');
 132   if (tmp == NULL)
 133     return (-1);
 134
 135   do
 136   {
 137     *tmp = 0;
 138     tmp++;
 139   }
 140   while ((tmp[0] == ' ') || (tmp[0] == '\t'));
 141
 142   if (*tmp == 0)
 143     return (-1);
 144
 145   *ret_value = tmp;
 146   return (0);
 147 } /* }}} int parse_header */
 148
 149 static int parse_ulong_header (char *line, /* {{{ */
 150     char **ret_key, unsigned long *ret_value)
 151 {
 152   char *str_value;
 153   char *endptr;
 154   int status;
 155
 156   str_value = NULL;
 157   status = parse_header (line, ret_key, &str_value);
 158   if (status != 0)
 159     return (status);
 160
 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);
 166
 167   return (0);
 168 } /* }}} int parse_ulong_header */
 169
 170 static int parse_char_array_header (char *line, /* {{{ */
 171     char **ret_key, char **array, size_t array_len, int alloc)
 172 {
 173   char *tmp_array[array_len];
 174   char *value;
 175   size_t num;
 176   int status;
 177
 178   value = NULL;
 179   status = parse_header (line, ret_key, &value);
 180   if (status != 0)
 181     return (-1);
 182
 183   num = strsplit (value, tmp_array, array_len);
 184   if (num != array_len)
 185     return (-1);
 186
 187   if (alloc == 0)
 188   {
 189     memcpy (array, tmp_array, sizeof (tmp_array));
 190   }
 191   else
 192   {
 193     size_t i;
 194
 195     for (i = 0; i < array_len; i++)
 196       array[i] = strdup (tmp_array[i]);
 197   }
 198
 199   return (0);
 200 } /* }}} int parse_char_array_header */
 201
 202 static int parse_value_array_header (char *line, /* {{{ */
 203     time_t *ret_time, rrd_value_t *array, size_t array_len)
 204 {
 205   char *str_key;
 206   char *str_array[array_len];
 207   char *endptr;
 208   int status;
 209   size_t i;
 210
 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);
 216
 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);
 222
 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   }
 230
 231   return (0);
 232 } /* }}} int parse_value_array_header */
 233
 234 /* One must hold `lock' when calling `close_connection'. */
 235 static void close_connection (void) /* {{{ */
 236 {
 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   }
 248
 249   if (sd_path != NULL)
 250     free (sd_path);
 251   sd_path = NULL;
 252 } /* }}} void close_connection */
 253
 254 static int buffer_add_string (const char *str, /* {{{ */
 255     char **buffer_ret, size_t *buffer_size_ret)
 256 {
 257   char *buffer;
 258   size_t buffer_size;
 259   size_t buffer_pos;
 260   size_t i;
 261   int status;
 262
 263   buffer = *buffer_ret;
 264   buffer_size = *buffer_size_ret;
 265   buffer_pos = 0;
 266
 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) */
 294
 295   if (status != 0)
 296     return (-1);
 297
 298   *buffer_ret = buffer + buffer_pos;
 299   *buffer_size_ret = buffer_size - buffer_pos;
 300
 301   return (0);
 302 } /* }}} int buffer_add_string */
 303
 304 static int buffer_add_value (const char *value, /* {{{ */
 305     char **buffer_ret, size_t *buffer_size_ret)
 306 {
 307   char temp[4096];
 308
 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;
 315
 316   return (buffer_add_string (temp, buffer_ret, buffer_size_ret));
 317 } /* }}} int buffer_add_value */
 318
 319 /* Remove trailing newline (NL) and carriage return (CR) characters. Similar to
 320  * the Perl function `chomp'. Returns the number of characters that have been
 321  * removed. */
 322 static int chomp (char *str) /* {{{ */
 323 {
 324   size_t len;
 325   int removed;
 326
 327   if (str == NULL)
 328     return (-1);
 329
 330   len = strlen (str);
 331   removed = 0;
 332   while ((len > 0) && ((str[len - 1] == '\n') || (str[len - 1] == '\r')))
 333   {
 334     str[len - 1] = 0;
 335     len--;
 336     removed++;
 337   }
 338
 339   return (removed);
 340 } /* }}} int chomp */
 341
 342 static void response_free (rrdc_response_t *res) /* {{{ */
 343 {
 344   if (res == NULL)
 345     return;
 346
 347   if (res->lines != NULL)
 348   {
 349     size_t i;
 350
 351     for (i = 0; i < res->lines_num; i++)
 352       if (res->lines[i] != NULL)
 353         free (res->lines[i]);
 354     free (res->lines);
 355   }
 356
 357   free (res);
 358 } /* }}} void response_free */
 359
 360 static int response_read (rrdc_response_t **ret_response) /* {{{ */
 361 {
 362   rrdc_response_t *ret;
 363
 364   char buffer[4096];
 365   char *buffer_ptr;
 366
 367   size_t i;
 368
 369   if (sh == NULL)
 370     return (-1);
 371
 372   ret = (rrdc_response_t *) malloc (sizeof (rrdc_response_t));
 373   if (ret == NULL)
 374     return (-2);
 375   memset (ret, 0, sizeof (*ret));
 376   ret->lines = NULL;
 377   ret->lines_num = 0;
 378
 379   buffer_ptr = fgets (buffer, sizeof (buffer), sh);
 380   if (buffer_ptr == NULL) {
 381     close_connection();
 382     return (-3);
 383   }
 384   chomp (buffer);
 385
 386   ret->status = strtol (buffer, &ret->message, 0);
 387   if (buffer == ret->message)
 388   {
 389     response_free (ret);
 390     close_connection();
 391     return (-4);
 392   }
 393   /* Skip leading whitespace of the status message */
 394   ret->message += strspn (ret->message, " \t");
 395
 396   if (ret->status <= 0)
 397   {
 398     if (ret->status < 0)
 399       rrd_set_error("rrdcached: %s", ret->message);
 400     *ret_response = ret;
 401     return (0);
 402   }
 403
 404   ret->lines = (char **) malloc (sizeof (char *) * ret->status);
 405   if (ret->lines == NULL)
 406   {
 407     response_free (ret);
 408     close_connection();
 409     return (-5);
 410   }
 411   memset (ret->lines, 0, sizeof (char *) * ret->status);
 412   ret->lines_num = (size_t) ret->status;
 413
 414   for (i = 0; i < ret->lines_num; i++)
 415   {
 416     buffer_ptr = fgets (buffer, sizeof (buffer), sh);
 417     if (buffer_ptr == NULL)
 418     {
 419       response_free (ret);
 420       close_connection();
 421       return (-6);
 422     }
 423     chomp (buffer);
 424
 425     ret->lines[i] = strdup (buffer);
 426     if (ret->lines[i] == NULL)
 427     {
 428       response_free (ret);
 429       close_connection();
 430       return (-7);
 431     }
 432   }
 433
 434   *ret_response = ret;
 435   return (0);
 436 } /* }}} rrdc_response_t *response_read */
 437
 438 static int request (const char *buffer, size_t buffer_size, /* {{{ */
 439     rrdc_response_t **ret_response)
 440 {
 441   int status;
 442   rrdc_response_t *res;
 443
 444   if (sh == NULL)
 445     return (ENOTCONN);
 446
 447   status = (int) fwrite (buffer, buffer_size, /* nmemb = */ 1, sh);
 448   if (status != 1)
 449   {
 450     close_connection ();
 451     rrd_set_error("request: socket error (%d) while talking to rrdcached",
 452                   status);
 453     return (-1);
 454   }
 455   fflush (sh);
 456
 457   res = NULL;
 458   status = response_read (&res);
 459
 460   if (status != 0)
 461   {
 462     if (status < 0)
 463       rrd_set_error("request: internal error while talking to rrdcached");
 464     return (status);
 465   }
 466
 467   *ret_response = res;
 468   return (0);
 469 } /* }}} int request */
 470
 471 /* determine whether we are connected to the specified daemon_addr if
 472  * NULL, return whether we are connected at all
 473  */
 474 int rrdc_is_connected(const char *daemon_addr) /* {{{ */
 475 {
 476   if (sd < 0)
 477     return 0;
 478   else if (daemon_addr == NULL)
 479   {
 480     /* here we have to handle the case i.e.
 481      *   UPDATE --daemon ...; UPDATEV (no --daemon) ...
 482      * In other words: we have a cached connection,
 483      * but it is not specified in the current command.
 484      * Daemon is only implied in this case if set in ENV
 485      */
 486     if (getenv(ENV_RRDCACHED_ADDRESS) != NULL)
 487       return 1;
 488     else
 489       return 0;
 490   }
 491   else if (strcmp(daemon_addr, sd_path) == 0)
 492     return 1;
 493   else
 494     return 0;
 495
 496 } /* }}} int rrdc_is_connected */
 497
 498 static int rrdc_connect_unix (const char *path) /* {{{ */
 499 {
 500   struct sockaddr_un sa;
 501   int status;
 502
 503   assert (path != NULL);
 504   assert (sd == -1);
 505
 506   sd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
 507   if (sd < 0)
 508   {
 509     status = errno;
 510     return (status);
 511   }
 512
 513   memset (&sa, 0, sizeof (sa));
 514   sa.sun_family = AF_UNIX;
 515   strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
 516
 517   status = connect (sd, (struct sockaddr *) &sa, sizeof (sa));
 518   if (status != 0)
 519   {
 520     status = errno;
 521     close_connection ();
 522     return (status);
 523   }
 524
 525   sh = fdopen (sd, "r+");
 526   if (sh == NULL)
 527   {
 528     status = errno;
 529     close_connection ();
 530     return (status);
 531   }
 532
 533   return (0);
 534 } /* }}} int rrdc_connect_unix */
 535
 536 static int rrdc_connect_network (const char *addr_orig) /* {{{ */
 537 {
 538   struct addrinfo ai_hints;
 539   struct addrinfo *ai_res;
 540   struct addrinfo *ai_ptr;
 541   char addr_copy[NI_MAXHOST];
 542   char *addr;
 543   char *port;
 544
 545   assert (addr_orig != NULL);
 546   assert (sd == -1);
 547
 548   strncpy(addr_copy, addr_orig, sizeof(addr_copy));
 549   addr_copy[sizeof(addr_copy) - 1] = '\0';
 550   addr = addr_copy;
 551
 552   int status;
 553   memset (&ai_hints, 0, sizeof (ai_hints));
 554   ai_hints.ai_flags = 0;
 555 #ifdef AI_ADDRCONFIG
 556   ai_hints.ai_flags |= AI_ADDRCONFIG;
 557 #endif
 558   ai_hints.ai_family = AF_UNSPEC;
 559   ai_hints.ai_socktype = SOCK_STREAM;
 560
 561   port = NULL;
 562   if (*addr == '[') /* IPv6+port format */
 563   {
 564     /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
 565     addr++;
 566
 567     port = strchr (addr, ']');
 568     if (port == NULL)
 569     {
 570       rrd_set_error("malformed address: %s", addr_orig);
 571       return (-1);
 572     }
 573     *port = 0;
 574     port++;
 575
 576     if (*port == ':')
 577       port++;
 578     else if (*port == 0)
 579       port = NULL;
 580     else
 581     {
 582       rrd_set_error("garbage after address: %s", port);
 583       return (-1);
 584     }
 585   } /* if (*addr == '[') */
 586   else
 587   {
 588     port = rindex(addr, ':');
 589     if (port != NULL)
 590     {
 591       *port = 0;
 592       port++;
 593     }
 594   }
 595
 596   ai_res = NULL;
 597   status = getaddrinfo (addr,
 598                         port == NULL ? RRDCACHED_DEFAULT_PORT : port,
 599                         &ai_hints, &ai_res);
 600   if (status != 0)
 601   {
 602     rrd_set_error ("failed to resolve address `%s' (port %s): %s",
 603         addr, port == NULL ? RRDCACHED_DEFAULT_PORT : port,
 604         gai_strerror (status));
 605     return (-1);
 606   }
 607
 608   for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 609   {
 610     sd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
 611     if (sd < 0)
 612     {
 613       status = errno;
 614       sd = -1;
 615       continue;
 616     }
 617
 618     status = connect (sd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 619     if (status != 0)
 620     {
 621       status = errno;
 622       close_connection();
 623       continue;
 624     }
 625
 626     sh = fdopen (sd, "r+");
 627     if (sh == NULL)
 628     {
 629       status = errno;
 630       close_connection ();
 631       continue;
 632     }
 633
 634     assert (status == 0);
 635     break;
 636   } /* for (ai_ptr) */
 637
 638   return (status);
 639 } /* }}} int rrdc_connect_network */
 640
 641 int rrdc_connect (const char *addr) /* {{{ */
 642 {
 643   int status = 0;
 644
 645   if (addr == NULL)
 646     addr = getenv (ENV_RRDCACHED_ADDRESS);
 647
 648   if (addr == NULL)
 649     return 0;
 650
 651   pthread_mutex_lock(&lock);
 652
 653   if (sd >= 0 && sd_path != NULL && strcmp(addr, sd_path) == 0)
 654   {
 655     /* connection to the same daemon; use cached connection */
 656     pthread_mutex_unlock (&lock);
 657     return (0);
 658   }
 659   else
 660   {
 661     close_connection();
 662   }
 663
 664   rrd_clear_error ();
 665   if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
 666     status = rrdc_connect_unix (addr + strlen ("unix:"));
 667   else if (addr[0] == '/')
 668     status = rrdc_connect_unix (addr);
 669   else
 670     status = rrdc_connect_network(addr);
 671
 672   if (status == 0 && sd >= 0)
 673     sd_path = strdup(addr);
 674   else
 675   {
 676     char *err = rrd_test_error () ? rrd_get_error () : "Internal error";
 677     /* err points the string that gets written to by rrd_set_error(), thus we
 678      * cannot pass it to that function */
 679     err = strdup (err);
 680     rrd_set_error("Unable to connect to rrdcached: %s",
 681                   (status < 0)
 682                   ? (err ? err : "Internal error")
 683                   : rrd_strerror (status));
 684     if (err != NULL)
 685       free (err);
 686   }
 687
 688   pthread_mutex_unlock (&lock);
 689   return (status);
 690 } /* }}} int rrdc_connect */
 691
 692 int rrdc_disconnect (void) /* {{{ */
 693 {
 694   pthread_mutex_lock (&lock);
 695
 696   close_connection();
 697
 698   pthread_mutex_unlock (&lock);
 699
 700   return (0);
 701 } /* }}} int rrdc_disconnect */
 702
 703 int rrdc_update (const char *filename, int values_num, /* {{{ */
 704                 const char * const *values)
 705 {
 706   char buffer[4096];
 707   char *buffer_ptr;
 708   size_t buffer_free;
 709   size_t buffer_size;
 710   rrdc_response_t *res;
 711   int status;
 712   int i;
 713   char file_path[PATH_MAX];
 714
 715   memset (buffer, 0, sizeof (buffer));
 716   buffer_ptr = &buffer[0];
 717   buffer_free = sizeof (buffer);
 718
 719   status = buffer_add_string ("update", &buffer_ptr, &buffer_free);
 720   if (status != 0)
 721     return (ENOBUFS);
 722
 723   pthread_mutex_lock (&lock);
 724   filename = get_path (filename, file_path);
 725   if (filename == NULL)
 726   {
 727     pthread_mutex_unlock (&lock);
 728     return (-1);
 729   }
 730
 731   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
 732   if (status != 0)
 733   {
 734     pthread_mutex_unlock (&lock);
 735     return (ENOBUFS);
 736   }
 737
 738   for (i = 0; i < values_num; i++)
 739   {
 740     status = buffer_add_value (values[i], &buffer_ptr, &buffer_free);
 741     if (status != 0)
 742     {
 743       pthread_mutex_unlock (&lock);
 744       return (ENOBUFS);
 745     }
 746   }
 747
 748   assert (buffer_free < sizeof (buffer));
 749   buffer_size = sizeof (buffer) - buffer_free;
 750   assert (buffer[buffer_size - 1] == ' ');
 751   buffer[buffer_size - 1] = '\n';
 752
 753   res = NULL;
 754   status = request (buffer, buffer_size, &res);
 755   pthread_mutex_unlock (&lock);
 756
 757   if (status != 0)
 758     return (status);
 759
 760   status = res->status;
 761   response_free (res);
 762
 763   return (status);
 764 } /* }}} int rrdc_update */
 765
 766 int rrdc_flush (const char *filename) /* {{{ */
 767 {
 768   char buffer[4096];
 769   char *buffer_ptr;
 770   size_t buffer_free;
 771   size_t buffer_size;
 772   rrdc_response_t *res;
 773   int status;
 774   char file_path[PATH_MAX];
 775
 776   if (filename == NULL)
 777     return (-1);
 778
 779   memset (buffer, 0, sizeof (buffer));
 780   buffer_ptr = &buffer[0];
 781   buffer_free = sizeof (buffer);
 782
 783   status = buffer_add_string ("flush", &buffer_ptr, &buffer_free);
 784   if (status != 0)
 785     return (ENOBUFS);
 786
 787   pthread_mutex_lock (&lock);
 788   filename = get_path (filename, file_path);
 789   if (filename == NULL)
 790   {
 791     pthread_mutex_unlock (&lock);
 792     return (-1);
 793   }
 794
 795   status = buffer_add_string (filename, &buffer_ptr, &buffer_free);
 796   if (status != 0)
 797   {
 798     pthread_mutex_unlock (&lock);
 799     return (ENOBUFS);
 800   }
 801
 802   assert (buffer_free < sizeof (buffer));
 803   buffer_size = sizeof (buffer) - buffer_free;
 804   assert (buffer[buffer_size - 1] == ' ');
 805   buffer[buffer_size - 1] = '\n';
 806
 807   res = NULL;
 808   status = request (buffer, buffer_size, &res);
 809   pthread_mutex_unlock (&lock);
 810
 811   if (status != 0)
 812     return (status);
 813
 814   status = res->status;
 815   response_free (res);
 816
 817   return (status);
 818 } /* }}} int rrdc_flush */
 819
 820 int rrdc_fetch (const char *filename, /* {{{ */
 821     const char *cf,
 822     time_t *ret_start, time_t *ret_end,
 823     unsigned long *ret_step,
 824     unsigned long *ret_ds_num,
 825     char ***ret_ds_names,
 826     rrd_value_t **ret_data)
 827 {
 828   char buffer[4096];
 829   char *buffer_ptr;
 830   size_t buffer_free;
 831   size_t buffer_size;
 832   rrdc_response_t *res;
 833   char path_buffer[PATH_MAX];
 834   const char *path_ptr;
 835
 836   char *str_tmp;
 837   unsigned long flush_version;
 838
 839   time_t start;
 840   time_t end;
 841   unsigned long step;
 842   unsigned long ds_num;
 843   char **ds_names;
 844
 845   rrd_value_t *data;
 846   size_t data_size;
 847   size_t data_fill;
 848
 849   int status;
 850   size_t current_line;
 851   time_t t;
 852
 853   if ((filename == NULL) || (cf == NULL))
 854     return (-1);
 855
 856   /* Send request {{{ */
 857   memset (buffer, 0, sizeof (buffer));
 858   buffer_ptr = &buffer[0];
 859   buffer_free = sizeof (buffer);
 860
 861   status = buffer_add_string ("FETCH", &buffer_ptr, &buffer_free);
 862   if (status != 0)
 863     return (ENOBUFS);
 864
 865   /* change to path for rrdcached */
 866   path_ptr = get_path (filename, path_buffer);
 867   if (path_ptr == NULL)
 868     return (EINVAL);
 869
 870   status = buffer_add_string (path_ptr, &buffer_ptr, &buffer_free);
 871   if (status != 0)
 872     return (ENOBUFS);
 873
 874   status = buffer_add_string (cf, &buffer_ptr, &buffer_free);
 875   if (status != 0)
 876     return (ENOBUFS);
 877
 878   if ((ret_start != NULL) && (*ret_start > 0))
 879   {
 880     char tmp[64];
 881     snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_start);
 882     tmp[sizeof (tmp) - 1] = 0;
 883     status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
 884     if (status != 0)
 885       return (ENOBUFS);
 886
 887     if ((ret_end != NULL) && (*ret_end > 0))
 888     {
 889       snprintf (tmp, sizeof (tmp), "%lu", (unsigned long) *ret_end);
 890       tmp[sizeof (tmp) - 1] = 0;
 891       status = buffer_add_string (tmp, &buffer_ptr, &buffer_free);
 892       if (status != 0)
 893         return (ENOBUFS);
 894     }
 895   }
 896
 897   assert (buffer_free < sizeof (buffer));
 898   buffer_size = sizeof (buffer) - buffer_free;
 899   assert (buffer[buffer_size - 1] == ' ');
 900   buffer[buffer_size - 1] = '\n';
 901
 902   res = NULL;
 903   status = request (buffer, buffer_size, &res);
 904   if (status != 0)
 905     return (status);
 906
 907   status = res->status;
 908   if (status < 0)
 909   {
 910     rrd_set_error ("rrdcached: %s", res->message);
 911     response_free (res);
 912     return (status);
 913   }
 914   /* }}} Send request */
 915
 916   ds_names = NULL;
 917   ds_num = 0;
 918   data = NULL;
 919   current_line = 0;
 920
 921   /* Macros to make error handling a little easier (i. e. less to type and
 922    * read. `BAIL_OUT' sets the error message, frees all dynamically allocated
 923    * variables and returns the provided status code. */
 924 #define BAIL_OUT(status, ...) do { \
 925     rrd_set_error ("rrdc_fetch: " __VA_ARGS__); \
 926     free (data); \
 927     if (ds_names != 0) { size_t k; for (k = 0; k < ds_num; k++) free (ds_names[k]); } \
 928     free (ds_names); \
 929     response_free (res); \
 930     return (status); \
 931   } while (0)
 932
 933 #define READ_NUMERIC_FIELD(name,type,var) do { \
 934     char *key; \
 935     unsigned long value; \
 936     assert (current_line < res->lines_num); \
 937     status = parse_ulong_header (res->lines[current_line], &key, &value); \
 938     if (status != 0) \
 939       BAIL_OUT (-1, "Unable to parse header `%s'", name); \
 940     if (strcasecmp (key, name) != 0) \
 941       BAIL_OUT (-1, "Unexpected header line: Expected `%s', got `%s'", name, key); \
 942     var = (type) value; \
 943     current_line++; \
 944   } while (0)
 945
 946   if (res->lines_num < 1)
 947     BAIL_OUT (-1, "Premature end of response packet");
 948
 949   /* We're making some very strong assumptions about the fields below. We
 950    * therefore check the version of the `flush' command first, so that later
 951    * versions can change the order of fields and it's easier to implement
 952    * backwards compatibility. */
 953   READ_NUMERIC_FIELD ("FlushVersion", unsigned long, flush_version);
 954   if (flush_version != 1)
 955     BAIL_OUT (-1, "Don't know how to handle flush format version %lu.",
 956         flush_version);
 957
 958   if (res->lines_num < 5)
 959     BAIL_OUT (-1, "Premature end of response packet");
 960
 961   READ_NUMERIC_FIELD ("Start", time_t, start);
 962   READ_NUMERIC_FIELD ("End", time_t, end);
 963   if (start >= end)
 964     BAIL_OUT (-1, "Malformed start and end times: start = %lu; end = %lu;",
 965         (unsigned long) start,
 966         (unsigned long) end);
 967
 968   READ_NUMERIC_FIELD ("Step", unsigned long, step);
 969   if (step < 1)
 970     BAIL_OUT (-1, "Invalid number for Step: %lu", step);
 971
 972   READ_NUMERIC_FIELD ("DSCount", unsigned long, ds_num);
 973   if (ds_num < 1)
 974     BAIL_OUT (-1, "Invalid number for DSCount: %lu", ds_num);
 975
 976   /* It's time to allocate some memory */
 977   ds_names = calloc ((size_t) ds_num, sizeof (*ds_names));
 978   if (ds_names == NULL)
 979     BAIL_OUT (-1, "Out of memory");
 980
 981   status = parse_char_array_header (res->lines[current_line],
 982       &str_tmp, ds_names, (size_t) ds_num, /* alloc = */ 1);
 983   if (status != 0)
 984     BAIL_OUT (-1, "Unable to parse header `DSName'");
 985   if (strcasecmp ("DSName", str_tmp) != 0)
 986     BAIL_OUT (-1, "Unexpected header line: Expected `DSName', got `%s'", str_tmp);
 987   current_line++;
 988
 989   data_size = ds_num * (end - start) / step;
 990   if (data_size < 1)
 991     BAIL_OUT (-1, "No data returned or headers invalid.");
 992
 993   if (res->lines_num != (6 + (data_size / ds_num)))
 994     BAIL_OUT (-1, "Got %zu lines, expected %zu",
 995         res->lines_num, (6 + (data_size / ds_num)));
 996
 997   data = calloc (data_size, sizeof (*data));
 998   if (data == NULL)
 999     BAIL_OUT (-1, "Out of memory");
1000
1001
1002   data_fill = 0;
1003   for (t = start + step; t <= end; t += step, current_line++)
1004   {
1005     time_t tmp;
1006
1007     assert (current_line < res->lines_num);
1008
1009     status = parse_value_array_header (res->lines[current_line],
1010         &tmp, data + data_fill, (size_t) ds_num);
1011     if (status != 0)
1012       BAIL_OUT (-1, "Cannot parse value line");
1013
1014     data_fill += (size_t) ds_num;
1015   }
1016
1017   *ret_start = start;
1018   *ret_end = end;
1019   *ret_step = step;
1020   *ret_ds_num = ds_num;
1021   *ret_ds_names = ds_names;
1022   *ret_data = data;
1023
1024   response_free (res);
1025   return (0);
1026 #undef READ_NUMERIC_FIELD
1027 #undef BAIL_OUT
1028 } /* }}} int rrdc_flush */
1029
1030 /* convenience function; if there is a daemon specified, or if we can
1031  * detect one from the environment, then flush the file.  Otherwise, no-op
1032  */
1033 int rrdc_flush_if_daemon (const char *opt_daemon, const char *filename) /* {{{ */
1034 {
1035   int status = 0;
1036
1037   rrdc_connect(opt_daemon);
1038
1039   if (rrdc_is_connected(opt_daemon))
1040   {
1041     rrd_clear_error();
1042     status = rrdc_flush (filename);
1043
1044     if (status != 0 && !rrd_test_error())
1045     {
1046       if (status > 0)
1047       {
1048         rrd_set_error("rrdc_flush (%s) failed: %s",
1049                       filename, rrd_strerror(status));
1050       }
1051       else if (status < 0)
1052       {
1053         rrd_set_error("rrdc_flush (%s) failed with status %i.",
1054                       filename, status);
1055       }
1056     }
1057   } /* if (rrdc_is_connected(..)) */
1058
1059   return status;
1060 } /* }}} int rrdc_flush_if_daemon */
1061
1062
1063 int rrdc_stats_get (rrdc_stats_t **ret_stats) /* {{{ */
1064 {
1065   rrdc_stats_t *head;
1066   rrdc_stats_t *tail;
1067
1068   rrdc_response_t *res;
1069
1070   int status;
1071   size_t i;
1072
1073   /* Protocol example: {{{
1074    * ->  STATS
1075    * <-  5 Statistics follow
1076    * <-  QueueLength: 0
1077    * <-  UpdatesWritten: 0
1078    * <-  DataSetsWritten: 0
1079    * <-  TreeNodesNumber: 0
1080    * <-  TreeDepth: 0
1081    * }}} */
1082
1083   res = NULL;
1084   pthread_mutex_lock (&lock);
1085   status = request ("STATS\n", strlen ("STATS\n"), &res);
1086   pthread_mutex_unlock (&lock);
1087
1088   if (status != 0)
1089     return (status);
1090
1091   if (res->status <= 0)
1092   {
1093     response_free (res);
1094     return (EIO);
1095   }
1096
1097   head = NULL;
1098   tail = NULL;
1099   for (i = 0; i < res->lines_num; i++)
1100   {
1101     char *key;
1102     char *value;
1103     char *endptr;
1104     rrdc_stats_t *s;
1105
1106     key = res->lines[i];
1107     value = strchr (key, ':');
1108     if (value == NULL)
1109       continue;
1110     *value = 0;
1111     value++;
1112
1113     while ((value[0] == ' ') || (value[0] == '\t'))
1114       value++;
1115
1116     s = (rrdc_stats_t *) malloc (sizeof (rrdc_stats_t));
1117     if (s == NULL)
1118       continue;
1119     memset (s, 0, sizeof (*s));
1120
1121     s->name = strdup (key);
1122
1123     endptr = NULL;
1124     if ((strcmp ("QueueLength", key) == 0)
1125         || (strcmp ("TreeDepth", key) == 0)
1126         || (strcmp ("TreeNodesNumber", key) == 0))
1127     {
1128       s->type = RRDC_STATS_TYPE_GAUGE;
1129       s->value.gauge = strtod (value, &endptr);
1130     }
1131     else if ((strcmp ("DataSetsWritten", key) == 0)
1132         || (strcmp ("FlushesReceived", key) == 0)
1133         || (strcmp ("JournalBytes", key) == 0)
1134         || (strcmp ("JournalRotate", key) == 0)
1135         || (strcmp ("UpdatesReceived", key) == 0)
1136         || (strcmp ("UpdatesWritten", key) == 0))
1137     {
1138       s->type = RRDC_STATS_TYPE_COUNTER;
1139       s->value.counter = (uint64_t) strtoll (value, &endptr, /* base = */ 0);
1140     }
1141     else
1142     {
1143       free (s);
1144       continue;
1145     }
1146
1147     /* Conversion failed */
1148     if (endptr == value)
1149     {
1150       free (s);
1151       continue;
1152     }
1153
1154     if (head == NULL)
1155     {
1156       head = s;
1157       tail = s;
1158       s->next = NULL;
1159     }
1160     else
1161     {
1162       tail->next = s;
1163       tail = s;
1164     }
1165   } /* for (i = 0; i < res->lines_num; i++) */
1166
1167   response_free (res);
1168
1169   if (head == NULL)
1170     return (EPROTO);
1171
1172   *ret_stats = head;
1173   return (0);
1174 } /* }}} int rrdc_stats_get */
1175
1176 void rrdc_stats_free (rrdc_stats_t *ret_stats) /* {{{ */
1177 {
1178   rrdc_stats_t *this;
1179
1180   this = ret_stats;
1181   while (this != NULL)
1182   {
1183     rrdc_stats_t *next;
1184
1185     next = this->next;
1186
1187     if (this->name != NULL)
1188     {
1189       free ((char *)this->name);
1190       this->name = NULL;
1191     }
1192     free (this);
1193
1194     this = next;
1195   } /* while (this != NULL) */
1196 } /* }}} void rrdc_stats_free */
1197
1198 /*
1199  * vim: set sw=2 sts=2 ts=8 et fdm=marker :
1200  */