src/rrd_daemon.c

   1 /**
   2  * RRDTool - src/rrd_daemon.c
   3  * Copyright (C) 2008-2010 Florian octo Forster
   4  * Copyright (C) 2008,2009 Kevin Brintnall
   5  *
   6  * This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the GNU General Public License as published by the
   8  * Free Software Foundation; only version 2 of the License is applicable.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  18  *
  19  * Authors:
  20  *   Florian octo Forster <octo at verplant.org>
  21  *   kevin brintnall <kbrint@rufus.net>
  22  **/
  23
  24 #if 0
  25 /*
  26  * First tell the compiler to stick to the C99 and POSIX standards as close as
  27  * possible.
  28  */
  29 #ifndef __STRICT_ANSI__ /* {{{ */
  30 # define __STRICT_ANSI__
  31 #endif
  32
  33 #ifndef _ISOC99_SOURCE
  34 # define _ISOC99_SOURCE
  35 #endif
  36
  37 #ifdef _POSIX_C_SOURCE
  38 # undef _POSIX_C_SOURCE
  39 #endif
  40 #define _POSIX_C_SOURCE 200112L
  41
  42 /* Single UNIX needed for strdup. */
  43 #ifdef _XOPEN_SOURCE
  44 # undef _XOPEN_SOURCE
  45 #endif
  46 #define _XOPEN_SOURCE 500
  47
  48 #ifndef _REENTRANT
  49 # define _REENTRANT
  50 #endif
  51
  52 #ifndef _THREAD_SAFE
  53 # define _THREAD_SAFE
  54 #endif
  55
  56 #ifdef _GNU_SOURCE
  57 # undef _GNU_SOURCE
  58 #endif
  59 /* }}} */
  60 #endif /* 0 */
  61
  62 /*
  63  * Now for some includes..
  64  */
  65 /* {{{ */
  66 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__) && !defined(HAVE_CONFIG_H)
  67 #include "../win32/config.h"
  68 #else
  69 #ifdef HAVE_CONFIG_H
  70 #include "../rrd_config.h"
  71 #endif
  72 #endif
  73
  74 #include "rrd.h"
  75 #include "rrd_client.h"
  76 #include "unused.h"
  77
  78 #include <stdlib.h>
  79
  80 #ifndef WIN32
  81 #ifdef HAVE_STDINT_H
  82 #  include <stdint.h>
  83 #endif
  84 #include <unistd.h>
  85 #include <strings.h>
  86 #include <inttypes.h>
  87 #include <sys/socket.h>
  88
  89 #else
  90
  91 #endif
  92 #include <stdio.h>
  93 #include <string.h>
  94
  95 #include <sys/types.h>
  96 #include <sys/stat.h>
  97 #include <dirent.h>
  98 #include <fcntl.h>
  99 #include <signal.h>
 100 #include <sys/un.h>
 101 #include <netdb.h>
 102 #include <poll.h>
 103 #include <syslog.h>
 104 #include <pthread.h>
 105 #include <errno.h>
 106 #include <assert.h>
 107 #include <sys/time.h>
 108 #include <time.h>
 109 #include <libgen.h>
 110 #include <grp.h>
 111
 112 #include <glib-2.0/glib.h>
 113 /* }}} */
 114
 115 #define RRDD_LOG(severity, ...) \
 116   do { \
 117     if (stay_foreground) \
 118       fprintf(stderr, __VA_ARGS__); \
 119     syslog ((severity), __VA_ARGS__); \
 120   } while (0)
 121
 122 /*
 123  * Types
 124  */
 125 typedef enum { RESP_ERR = -1, RESP_OK = 0 } response_code;
 126
 127 struct listen_socket_s
 128 {
 129   int fd;
 130   char addr[PATH_MAX + 1];
 131   int family;
 132
 133   /* state for BATCH processing */
 134   time_t batch_start;
 135   int batch_cmd;
 136
 137   /* buffered IO */
 138   char *rbuf;
 139   off_t next_cmd;
 140   off_t next_read;
 141
 142   char *wbuf;
 143   ssize_t wbuf_len;
 144
 145   uint32_t permissions;
 146
 147   gid_t  socket_group;
 148   mode_t socket_permissions;
 149 };
 150 typedef struct listen_socket_s listen_socket_t;
 151
 152 struct command_s;
 153 typedef struct command_s command_t;
 154 /* note: guard against "unused" warnings in the handlers */
 155 #define DISPATCH_PROTO  listen_socket_t UNUSED(*sock),\
 156                         time_t UNUSED(now),\
 157                         char  UNUSED(*buffer),\
 158                         size_t UNUSED(buffer_size)
 159
 160 #define HANDLER_PROTO   command_t UNUSED(*cmd),\
 161                         DISPATCH_PROTO
 162
 163 struct command_s {
 164   char   *cmd;
 165   int (*handler)(HANDLER_PROTO);
 166
 167   char  context;                /* where we expect to see it */
 168 #define CMD_CONTEXT_CLIENT      (1<<0)
 169 #define CMD_CONTEXT_BATCH       (1<<1)
 170 #define CMD_CONTEXT_JOURNAL     (1<<2)
 171 #define CMD_CONTEXT_ANY         (0x7f)
 172
 173   char *syntax;
 174   char *help;
 175 };
 176
 177 struct cache_item_s;
 178 typedef struct cache_item_s cache_item_t;
 179 struct cache_item_s
 180 {
 181   char *file;
 182   char **values;
 183   size_t values_num;            /* number of valid pointers */
 184   size_t values_alloc;          /* number of allocated pointers */
 185   time_t last_flush_time;
 186   time_t last_update_stamp;
 187 #define CI_FLAGS_IN_TREE  (1<<0)
 188 #define CI_FLAGS_IN_QUEUE (1<<1)
 189   int flags;
 190   pthread_cond_t  flushed;
 191   cache_item_t *prev;
 192   cache_item_t *next;
 193 };
 194
 195 struct callback_flush_data_s
 196 {
 197   time_t now;
 198   time_t abs_timeout;
 199   char **keys;
 200   size_t keys_num;
 201 };
 202 typedef struct callback_flush_data_s callback_flush_data_t;
 203
 204 enum queue_side_e
 205 {
 206   HEAD,
 207   TAIL
 208 };
 209 typedef enum queue_side_e queue_side_t;
 210
 211 /* describe a set of journal files */
 212 typedef struct {
 213   char **files;
 214   size_t files_num;
 215 } journal_set;
 216
 217 /* max length of socket command or response */
 218 #define CMD_MAX 4096
 219 #define RBUF_SIZE (CMD_MAX*2)
 220
 221 /*
 222  * Variables
 223  */
 224 static int stay_foreground = 0;
 225 static uid_t daemon_uid;
 226
 227 static listen_socket_t *listen_fds = NULL;
 228 static size_t listen_fds_num = 0;
 229
 230 enum {
 231   RUNNING,              /* normal operation */
 232   FLUSHING,             /* flushing remaining values */
 233   SHUTDOWN              /* shutting down */
 234 } state = RUNNING;
 235
 236 static pthread_t *queue_threads;
 237 static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER;
 238 static int config_queue_threads = 4;
 239
 240 static pthread_t flush_thread;
 241 static pthread_cond_t flush_cond = PTHREAD_COND_INITIALIZER;
 242
 243 static pthread_mutex_t connection_threads_lock = PTHREAD_MUTEX_INITIALIZER;
 244 static pthread_cond_t  connection_threads_done = PTHREAD_COND_INITIALIZER;
 245 static int connection_threads_num = 0;
 246
 247 /* Cache stuff */
 248 static GTree          *cache_tree = NULL;
 249 static cache_item_t   *cache_queue_head = NULL;
 250 static cache_item_t   *cache_queue_tail = NULL;
 251 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
 252
 253 static int config_write_interval = 300;
 254 static int config_write_jitter   = 0;
 255 static int config_flush_interval = 3600;
 256 static int config_flush_at_shutdown = 0;
 257 static char *config_pid_file = NULL;
 258 static char *config_base_dir = NULL;
 259 static size_t _config_base_dir_len = 0;
 260 static int config_write_base_only = 0;
 261 static size_t config_alloc_chunk = 1;
 262
 263 static listen_socket_t **config_listen_address_list = NULL;
 264 static size_t config_listen_address_list_len = 0;
 265
 266 static uint64_t stats_queue_length = 0;
 267 static uint64_t stats_updates_received = 0;
 268 static uint64_t stats_flush_received = 0;
 269 static uint64_t stats_updates_written = 0;
 270 static uint64_t stats_data_sets_written = 0;
 271 static uint64_t stats_journal_bytes = 0;
 272 static uint64_t stats_journal_rotate = 0;
 273 static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER;
 274
 275 /* Journaled updates */
 276 #define JOURNAL_REPLAY(s) ((s) == NULL)
 277 #define JOURNAL_BASE "rrd.journal"
 278 static journal_set *journal_cur = NULL;
 279 static journal_set *journal_old = NULL;
 280 static char *journal_dir = NULL;
 281 static FILE *journal_fh = NULL;         /* current journal file handle */
 282 static long  journal_size = 0;          /* current journal size */
 283 #define JOURNAL_MAX (1 * 1024 * 1024 * 1024)
 284 static pthread_mutex_t journal_lock = PTHREAD_MUTEX_INITIALIZER;
 285 static int journal_write(char *cmd, char *args);
 286 static void journal_done(void);
 287 static void journal_rotate(void);
 288
 289 /* prototypes for forward refernces */
 290 static int handle_request_help (HANDLER_PROTO);
 291
 292 /*
 293  * Functions
 294  */
 295 static void sig_common (const char *sig) /* {{{ */
 296 {
 297   RRDD_LOG(LOG_NOTICE, "caught SIG%s", sig);
 298   state = FLUSHING;
 299   pthread_cond_broadcast(&flush_cond);
 300   pthread_cond_broadcast(&queue_cond);
 301 } /* }}} void sig_common */
 302
 303 static void sig_int_handler (int UNUSED(s)) /* {{{ */
 304 {
 305   sig_common("INT");
 306 } /* }}} void sig_int_handler */
 307
 308 static void sig_term_handler (int UNUSED(s)) /* {{{ */
 309 {
 310   sig_common("TERM");
 311 } /* }}} void sig_term_handler */
 312
 313 static void sig_usr1_handler (int UNUSED(s)) /* {{{ */
 314 {
 315   config_flush_at_shutdown = 1;
 316   sig_common("USR1");
 317 } /* }}} void sig_usr1_handler */
 318
 319 static void sig_usr2_handler (int UNUSED(s)) /* {{{ */
 320 {
 321   config_flush_at_shutdown = 0;
 322   sig_common("USR2");
 323 } /* }}} void sig_usr2_handler */
 324
 325 static void install_signal_handlers(void) /* {{{ */
 326 {
 327   /* These structures are static, because `sigaction' behaves weird if the are
 328    * overwritten.. */
 329   static struct sigaction sa_int;
 330   static struct sigaction sa_term;
 331   static struct sigaction sa_pipe;
 332   static struct sigaction sa_usr1;
 333   static struct sigaction sa_usr2;
 334
 335   /* Install signal handlers */
 336   memset (&sa_int, 0, sizeof (sa_int));
 337   sa_int.sa_handler = sig_int_handler;
 338   sigaction (SIGINT, &sa_int, NULL);
 339
 340   memset (&sa_term, 0, sizeof (sa_term));
 341   sa_term.sa_handler = sig_term_handler;
 342   sigaction (SIGTERM, &sa_term, NULL);
 343
 344   memset (&sa_pipe, 0, sizeof (sa_pipe));
 345   sa_pipe.sa_handler = SIG_IGN;
 346   sigaction (SIGPIPE, &sa_pipe, NULL);
 347
 348   memset (&sa_pipe, 0, sizeof (sa_usr1));
 349   sa_usr1.sa_handler = sig_usr1_handler;
 350   sigaction (SIGUSR1, &sa_usr1, NULL);
 351
 352   memset (&sa_usr2, 0, sizeof (sa_usr2));
 353   sa_usr2.sa_handler = sig_usr2_handler;
 354   sigaction (SIGUSR2, &sa_usr2, NULL);
 355
 356 } /* }}} void install_signal_handlers */
 357
 358 static int open_pidfile(char *action, int oflag) /* {{{ */
 359 {
 360   int fd;
 361   const char *file;
 362   char *file_copy, *dir;
 363
 364   file = (config_pid_file != NULL)
 365     ? config_pid_file
 366     : LOCALSTATEDIR "/run/rrdcached.pid";
 367
 368   /* dirname may modify its argument */
 369   file_copy = strdup(file);
 370   if (file_copy == NULL)
 371   {
 372     fprintf(stderr, "rrdcached: strdup(): %s\n",
 373         rrd_strerror(errno));
 374     return -1;
 375   }
 376
 377   dir = dirname(file_copy);
 378   if (rrd_mkdir_p(dir, 0777) != 0)
 379   {
 380     fprintf(stderr, "Failed to create pidfile directory '%s': %s\n",
 381         dir, rrd_strerror(errno));
 382     return -1;
 383   }
 384
 385   free(file_copy);
 386
 387   fd = open(file, oflag, S_IWUSR|S_IRUSR|S_IRGRP|S_IROTH);
 388   if (fd < 0)
 389     fprintf(stderr, "rrdcached: can't %s pid file '%s' (%s)\n",
 390             action, file, rrd_strerror(errno));
 391
 392   return(fd);
 393 } /* }}} static int open_pidfile */
 394
 395 /* check existing pid file to see whether a daemon is running */
 396 static int check_pidfile(void)
 397 {
 398   int pid_fd;
 399   pid_t pid;
 400   char pid_str[16];
 401
 402   pid_fd = open_pidfile("open", O_RDWR);
 403   if (pid_fd < 0)
 404     return pid_fd;
 405
 406   if (read(pid_fd, pid_str, sizeof(pid_str)) <= 0)
 407     return -1;
 408
 409   pid = atoi(pid_str);
 410   if (pid <= 0)
 411     return -1;
 412
 413   /* another running process that we can signal COULD be
 414    * a competing rrdcached */
 415   if (pid != getpid() && kill(pid, 0) == 0)
 416   {
 417     fprintf(stderr,
 418             "FATAL: Another rrdcached daemon is running?? (pid %d)\n", pid);
 419     close(pid_fd);
 420     return -1;
 421   }
 422
 423   lseek(pid_fd, 0, SEEK_SET);
 424   if (ftruncate(pid_fd, 0) == -1)
 425   {
 426     fprintf(stderr,
 427             "FATAL: Faild to truncate stale PID file. (pid %d)\n", pid);
 428     close(pid_fd);
 429     return -1;
 430   }
 431
 432   fprintf(stderr,
 433           "rrdcached: removed stale PID file (no rrdcached on pid %d)\n"
 434           "rrdcached: starting normally.\n", pid);
 435
 436   return pid_fd;
 437 } /* }}} static int check_pidfile */
 438
 439 static int write_pidfile (int fd) /* {{{ */
 440 {
 441   pid_t pid;
 442   FILE *fh;
 443
 444   pid = getpid ();
 445
 446   fh = fdopen (fd, "w");
 447   if (fh == NULL)
 448   {
 449     RRDD_LOG (LOG_ERR, "write_pidfile: fdopen() failed.");
 450     close(fd);
 451     return (-1);
 452   }
 453
 454   fprintf (fh, "%i\n", (int) pid);
 455   fclose (fh);
 456
 457   return (0);
 458 } /* }}} int write_pidfile */
 459
 460 static int remove_pidfile (void) /* {{{ */
 461 {
 462   char *file;
 463   int status;
 464
 465   file = (config_pid_file != NULL)
 466     ? config_pid_file
 467     : LOCALSTATEDIR "/run/rrdcached.pid";
 468
 469   status = unlink (file);
 470   if (status == 0)
 471     return (0);
 472   return (errno);
 473 } /* }}} int remove_pidfile */
 474
 475 static char *next_cmd (listen_socket_t *sock, ssize_t *len) /* {{{ */
 476 {
 477   char *eol;
 478
 479   eol = memchr(sock->rbuf + sock->next_cmd, '\n',
 480                sock->next_read - sock->next_cmd);
 481
 482   if (eol == NULL)
 483   {
 484     /* no commands left, move remainder back to front of rbuf */
 485     memmove(sock->rbuf, sock->rbuf + sock->next_cmd,
 486             sock->next_read - sock->next_cmd);
 487     sock->next_read -= sock->next_cmd;
 488     sock->next_cmd = 0;
 489     *len = 0;
 490     return NULL;
 491   }
 492   else
 493   {
 494     char *cmd = sock->rbuf + sock->next_cmd;
 495     *eol = '\0';
 496
 497     sock->next_cmd = eol - sock->rbuf + 1;
 498
 499     if (eol > sock->rbuf && *(eol-1) == '\r')
 500       *(--eol) = '\0'; /* handle "\r\n" EOL */
 501
 502     *len = eol - cmd;
 503
 504     return cmd;
 505   }
 506
 507   /* NOTREACHED */
 508   assert(1==0);
 509 } /* }}} char *next_cmd */
 510
 511 /* add the characters directly to the write buffer */
 512 static int add_to_wbuf(listen_socket_t *sock, char *str, size_t len) /* {{{ */
 513 {
 514   char *new_buf;
 515
 516   assert(sock != NULL);
 517
 518   new_buf = rrd_realloc(sock->wbuf, sock->wbuf_len + len + 1);
 519   if (new_buf == NULL)
 520   {
 521     RRDD_LOG(LOG_ERR, "add_to_wbuf: realloc failed");
 522     return -1;
 523   }
 524
 525   strncpy(new_buf + sock->wbuf_len, str, len + 1);
 526
 527   sock->wbuf = new_buf;
 528   sock->wbuf_len += len;
 529
 530   return 0;
 531 } /* }}} static int add_to_wbuf */
 532
 533 /* add the text to the "extra" info that's sent after the status line */
 534 static int add_response_info(listen_socket_t *sock, char *fmt, ...) /* {{{ */
 535 {
 536   va_list argp;
 537   char buffer[CMD_MAX];
 538   int len;
 539
 540   if (JOURNAL_REPLAY(sock)) return 0;
 541   if (sock->batch_start) return 0; /* no extra info returned when in BATCH */
 542
 543   va_start(argp, fmt);
 544 #ifdef HAVE_VSNPRINTF
 545   len = vsnprintf(buffer, sizeof(buffer), fmt, argp);
 546 #else
 547   len = vsprintf(buffer, fmt, argp);
 548 #endif
 549   va_end(argp);
 550   if (len < 0)
 551   {
 552     RRDD_LOG(LOG_ERR, "add_response_info: vnsprintf failed");
 553     return -1;
 554   }
 555
 556   return add_to_wbuf(sock, buffer, len);
 557 } /* }}} static int add_response_info */
 558
 559 static int count_lines(char *str) /* {{{ */
 560 {
 561   int lines = 0;
 562
 563   if (str != NULL)
 564   {
 565     while ((str = strchr(str, '\n')) != NULL)
 566     {
 567       ++lines;
 568       ++str;
 569     }
 570   }
 571
 572   return lines;
 573 } /* }}} static int count_lines */
 574
 575 /* send the response back to the user.
 576  * returns 0 on success, -1 on error
 577  * write buffer is always zeroed after this call */
 578 static int send_response (listen_socket_t *sock, response_code rc,
 579                           char *fmt, ...) /* {{{ */
 580 {
 581   va_list argp;
 582   char buffer[CMD_MAX];
 583   int lines;
 584   ssize_t wrote;
 585   int rclen, len;
 586
 587   if (JOURNAL_REPLAY(sock)) return rc;
 588
 589   if (sock->batch_start)
 590   {
 591     if (rc == RESP_OK)
 592       return rc; /* no response on success during BATCH */
 593     lines = sock->batch_cmd;
 594   }
 595   else if (rc == RESP_OK)
 596     lines = count_lines(sock->wbuf);
 597   else
 598     lines = -1;
 599
 600   rclen = sprintf(buffer, "%d ", lines);
 601   va_start(argp, fmt);
 602 #ifdef HAVE_VSNPRINTF
 603   len = vsnprintf(buffer+rclen, sizeof(buffer)-rclen, fmt, argp);
 604 #else
 605   len = vsprintf(buffer+rclen, fmt, argp);
 606 #endif
 607   va_end(argp);
 608   if (len < 0)
 609     return -1;
 610
 611   len += rclen;
 612
 613   /* append the result to the wbuf, don't write to the user */
 614   if (sock->batch_start)
 615     return add_to_wbuf(sock, buffer, len);
 616
 617   /* first write must be complete */
 618   if (len != write(sock->fd, buffer, len))
 619   {
 620     RRDD_LOG(LOG_INFO, "send_response: could not write status message");
 621     return -1;
 622   }
 623
 624   if (sock->wbuf != NULL && rc == RESP_OK)
 625   {
 626     wrote = 0;
 627     while (wrote < sock->wbuf_len)
 628     {
 629       ssize_t wb = write(sock->fd, sock->wbuf + wrote, sock->wbuf_len - wrote);
 630       if (wb <= 0)
 631       {
 632         RRDD_LOG(LOG_INFO, "send_response: could not write results");
 633         return -1;
 634       }
 635       wrote += wb;
 636     }
 637   }
 638
 639   free(sock->wbuf); sock->wbuf = NULL;
 640   sock->wbuf_len = 0;
 641
 642   return 0;
 643 } /* }}} */
 644
 645 static void wipe_ci_values(cache_item_t *ci, time_t when)
 646 {
 647   ci->values = NULL;
 648   ci->values_num = 0;
 649   ci->values_alloc = 0;
 650
 651   ci->last_flush_time = when;
 652   if (config_write_jitter > 0)
 653     ci->last_flush_time += (rrd_random() % config_write_jitter);
 654 }
 655
 656 /* remove_from_queue
 657  * remove a "cache_item_t" item from the queue.
 658  * must hold 'cache_lock' when calling this
 659  */
 660 static void remove_from_queue(cache_item_t *ci) /* {{{ */
 661 {
 662   if (ci == NULL) return;
 663   if ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) return; /* not queued */
 664
 665   if (ci->prev == NULL)
 666     cache_queue_head = ci->next; /* reset head */
 667   else
 668     ci->prev->next = ci->next;
 669
 670   if (ci->next == NULL)
 671     cache_queue_tail = ci->prev; /* reset the tail */
 672   else
 673     ci->next->prev = ci->prev;
 674
 675   ci->next = ci->prev = NULL;
 676   ci->flags &= ~CI_FLAGS_IN_QUEUE;
 677
 678   pthread_mutex_lock (&stats_lock);
 679   assert (stats_queue_length > 0);
 680   stats_queue_length--;
 681   pthread_mutex_unlock (&stats_lock);
 682
 683 } /* }}} static void remove_from_queue */
 684
 685 /* free the resources associated with the cache_item_t
 686  * must hold cache_lock when calling this function
 687  */
 688 static void *free_cache_item(cache_item_t *ci) /* {{{ */
 689 {
 690   if (ci == NULL) return NULL;
 691
 692   remove_from_queue(ci);
 693
 694   for (size_t i=0; i < ci->values_num; i++)
 695     free(ci->values[i]);
 696
 697   free (ci->values);
 698   free (ci->file);
 699
 700   /* in case anyone is waiting */
 701   pthread_cond_broadcast(&ci->flushed);
 702   pthread_cond_destroy(&ci->flushed);
 703
 704   free (ci);
 705
 706   return NULL;
 707 } /* }}} static void *free_cache_item */
 708
 709 /*
 710  * enqueue_cache_item:
 711  * `cache_lock' must be acquired before calling this function!
 712  */
 713 static int enqueue_cache_item (cache_item_t *ci, /* {{{ */
 714     queue_side_t side)
 715 {
 716   if (ci == NULL)
 717     return (-1);
 718
 719   if (ci->values_num == 0)
 720     return (0);
 721
 722   if (side == HEAD)
 723   {
 724     if (cache_queue_head == ci)
 725       return 0;
 726
 727     /* remove if further down in queue */
 728     remove_from_queue(ci);
 729
 730     ci->prev = NULL;
 731     ci->next = cache_queue_head;
 732     if (ci->next != NULL)
 733       ci->next->prev = ci;
 734     cache_queue_head = ci;
 735
 736     if (cache_queue_tail == NULL)
 737       cache_queue_tail = cache_queue_head;
 738   }
 739   else /* (side == TAIL) */
 740   {
 741     /* We don't move values back in the list.. */
 742     if (ci->flags & CI_FLAGS_IN_QUEUE)
 743       return (0);
 744
 745     assert (ci->next == NULL);
 746     assert (ci->prev == NULL);
 747
 748     ci->prev = cache_queue_tail;
 749
 750     if (cache_queue_tail == NULL)
 751       cache_queue_head = ci;
 752     else
 753       cache_queue_tail->next = ci;
 754
 755     cache_queue_tail = ci;
 756   }
 757
 758   ci->flags |= CI_FLAGS_IN_QUEUE;
 759
 760   pthread_cond_signal(&queue_cond);
 761   pthread_mutex_lock (&stats_lock);
 762   stats_queue_length++;
 763   pthread_mutex_unlock (&stats_lock);
 764
 765   return (0);
 766 } /* }}} int enqueue_cache_item */
 767
 768 /*
 769  * tree_callback_flush:
 770  * Called via `g_tree_foreach' in `flush_thread_main'. `cache_lock' is held
 771  * while this is in progress.
 772  */
 773 static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */
 774     gpointer data)
 775 {
 776   cache_item_t *ci;
 777   callback_flush_data_t *cfd;
 778
 779   ci = (cache_item_t *) value;
 780   cfd = (callback_flush_data_t *) data;
 781
 782   if (ci->flags & CI_FLAGS_IN_QUEUE)
 783     return FALSE;
 784
 785   if (ci->values_num > 0
 786       && (ci->last_flush_time <= cfd->abs_timeout || state != RUNNING))
 787   {
 788     enqueue_cache_item (ci, TAIL);
 789   }
 790   else if (((cfd->now - ci->last_flush_time) >= config_flush_interval)
 791       && (ci->values_num <= 0))
 792   {
 793     assert ((char *) key == ci->file);
 794     if (!rrd_add_ptr((void ***)&cfd->keys, &cfd->keys_num, (void *)key))
 795     {
 796       RRDD_LOG (LOG_ERR, "tree_callback_flush: rrd_add_ptrs failed.");
 797       return (FALSE);
 798     }
 799   }
 800
 801   return (FALSE);
 802 } /* }}} gboolean tree_callback_flush */
 803
 804 static int flush_old_values (int max_age)
 805 {
 806   callback_flush_data_t cfd;
 807   size_t k;
 808
 809   memset (&cfd, 0, sizeof (cfd));
 810   /* Pass the current time as user data so that we don't need to call
 811    * `time' for each node. */
 812   cfd.now = time (NULL);
 813   cfd.keys = NULL;
 814   cfd.keys_num = 0;
 815
 816   if (max_age > 0)
 817     cfd.abs_timeout = cfd.now - max_age;
 818   else
 819     cfd.abs_timeout = cfd.now + 2*config_write_jitter + 1;
 820
 821   /* `tree_callback_flush' will return the keys of all values that haven't
 822    * been touched in the last `config_flush_interval' seconds in `cfd'.
 823    * The char*'s in this array point to the same memory as ci->file, so we
 824    * don't need to free them separately. */
 825   g_tree_foreach (cache_tree, tree_callback_flush, (gpointer) &cfd);
 826
 827   for (k = 0; k < cfd.keys_num; k++)
 828   {
 829     gboolean status = g_tree_remove(cache_tree, cfd.keys[k]);
 830     /* should never fail, since we have held the cache_lock
 831      * the entire time */
 832     assert(status == TRUE);
 833   }
 834
 835   if (cfd.keys != NULL)
 836   {
 837     free (cfd.keys);
 838     cfd.keys = NULL;
 839   }
 840
 841   return (0);
 842 } /* int flush_old_values */
 843
 844 static void *flush_thread_main (void UNUSED(*args)) /* {{{ */
 845 {
 846   struct timeval now;
 847   struct timespec next_flush;
 848   int status;
 849
 850   gettimeofday (&now, NULL);
 851   next_flush.tv_sec = now.tv_sec + config_flush_interval;
 852   next_flush.tv_nsec = 1000 * now.tv_usec;
 853
 854   pthread_mutex_lock(&cache_lock);
 855
 856   while (state == RUNNING)
 857   {
 858     gettimeofday (&now, NULL);
 859     if ((now.tv_sec > next_flush.tv_sec)
 860         || ((now.tv_sec == next_flush.tv_sec)
 861           && ((1000 * now.tv_usec) > next_flush.tv_nsec)))
 862     {
 863       RRDD_LOG(LOG_DEBUG, "flushing old values");
 864
 865       /* Determine the time of the next cache flush. */
 866       next_flush.tv_sec = now.tv_sec + config_flush_interval;
 867
 868       /* Flush all values that haven't been written in the last
 869        * `config_write_interval' seconds. */
 870       flush_old_values (config_write_interval);
 871
 872       /* unlock the cache while we rotate so we don't block incoming
 873        * updates if the fsync() blocks on disk I/O */
 874       pthread_mutex_unlock(&cache_lock);
 875       journal_rotate();
 876       pthread_mutex_lock(&cache_lock);
 877     }
 878
 879     status = pthread_cond_timedwait(&flush_cond, &cache_lock, &next_flush);
 880     if (status != 0 && status != ETIMEDOUT)
 881     {
 882       RRDD_LOG (LOG_ERR, "flush_thread_main: "
 883                 "pthread_cond_timedwait returned %i.", status);
 884     }
 885   }
 886
 887   if (config_flush_at_shutdown)
 888     flush_old_values (-1); /* flush everything */
 889
 890   state = SHUTDOWN;
 891
 892   pthread_mutex_unlock(&cache_lock);
 893
 894   return NULL;
 895 } /* void *flush_thread_main */
 896
 897 static void *queue_thread_main (void UNUSED(*args)) /* {{{ */
 898 {
 899   pthread_mutex_lock (&cache_lock);
 900
 901   while (state != SHUTDOWN
 902          || (cache_queue_head != NULL && config_flush_at_shutdown))
 903   {
 904     cache_item_t *ci;
 905     char *file;
 906     char **values;
 907     size_t values_num;
 908     int status;
 909
 910     /* Now, check if there's something to store away. If not, wait until
 911      * something comes in. */
 912     if (cache_queue_head == NULL)
 913     {
 914       status = pthread_cond_wait (&queue_cond, &cache_lock);
 915       if ((status != 0) && (status != ETIMEDOUT))
 916       {
 917         RRDD_LOG (LOG_ERR, "queue_thread_main: "
 918             "pthread_cond_wait returned %i.", status);
 919       }
 920     }
 921
 922     /* Check if a value has arrived. This may be NULL if we timed out or there
 923      * was an interrupt such as a signal. */
 924     if (cache_queue_head == NULL)
 925       continue;
 926
 927     ci = cache_queue_head;
 928
 929     /* copy the relevant parts */
 930     file = strdup (ci->file);
 931     if (file == NULL)
 932     {
 933       RRDD_LOG (LOG_ERR, "queue_thread_main: strdup failed.");
 934       continue;
 935     }
 936
 937     assert(ci->values != NULL);
 938     assert(ci->values_num > 0);
 939
 940     values = ci->values;
 941     values_num = ci->values_num;
 942
 943     wipe_ci_values(ci, time(NULL));
 944     remove_from_queue(ci);
 945
 946     pthread_mutex_unlock (&cache_lock);
 947
 948     rrd_clear_error ();
 949     status = rrd_update_r (file, NULL, (int) values_num, (void *) values);
 950     if (status != 0)
 951     {
 952       RRDD_LOG (LOG_NOTICE, "queue_thread_main: "
 953           "rrd_update_r (%s) failed with status %i. (%s)",
 954           file, status, rrd_get_error());
 955     }
 956
 957     journal_write("wrote", file);
 958
 959     /* Search again in the tree.  It's possible someone issued a "FORGET"
 960      * while we were writing the update values. */
 961     pthread_mutex_lock(&cache_lock);
 962     ci = (cache_item_t *) g_tree_lookup(cache_tree, file);
 963     if (ci)
 964       pthread_cond_broadcast(&ci->flushed);
 965     pthread_mutex_unlock(&cache_lock);
 966
 967     if (status == 0)
 968     {
 969       pthread_mutex_lock (&stats_lock);
 970       stats_updates_written++;
 971       stats_data_sets_written += values_num;
 972       pthread_mutex_unlock (&stats_lock);
 973     }
 974
 975     rrd_free_ptrs((void ***) &values, &values_num);
 976     free(file);
 977
 978     pthread_mutex_lock (&cache_lock);
 979   }
 980   pthread_mutex_unlock (&cache_lock);
 981
 982   return (NULL);
 983 } /* }}} void *queue_thread_main */
 984
 985 static int buffer_get_field (char **buffer_ret, /* {{{ */
 986     size_t *buffer_size_ret, char **field_ret)
 987 {
 988   char *buffer;
 989   size_t buffer_pos;
 990   size_t buffer_size;
 991   char *field;
 992   size_t field_size;
 993   int status;
 994
 995   buffer = *buffer_ret;
 996   buffer_pos = 0;
 997   buffer_size = *buffer_size_ret;
 998   field = *buffer_ret;
 999   field_size = 0;
1000
1001   if (buffer_size <= 0)
1002     return (-1);
1003
1004   /* This is ensured by `handle_request'. */
1005   assert (buffer[buffer_size - 1] == '\0');
1006
1007   status = -1;
1008   while (buffer_pos < buffer_size)
1009   {
1010     /* Check for end-of-field or end-of-buffer */
1011     if (buffer[buffer_pos] == ' ' || buffer[buffer_pos] == '\0')
1012     {
1013       field[field_size] = 0;
1014       field_size++;
1015       buffer_pos++;
1016       status = 0;
1017       break;
1018     }
1019     /* Handle escaped characters. */
1020     else if (buffer[buffer_pos] == '\\')
1021     {
1022       if (buffer_pos >= (buffer_size - 1))
1023         break;
1024       buffer_pos++;
1025       field[field_size] = buffer[buffer_pos];
1026       field_size++;
1027       buffer_pos++;
1028     }
1029     /* Normal operation */
1030     else
1031     {
1032       field[field_size] = buffer[buffer_pos];
1033       field_size++;
1034       buffer_pos++;
1035     }
1036   } /* while (buffer_pos < buffer_size) */
1037
1038   if (status != 0)
1039     return (status);
1040
1041   *buffer_ret = buffer + buffer_pos;
1042   *buffer_size_ret = buffer_size - buffer_pos;
1043   *field_ret = field;
1044
1045   return (0);
1046 } /* }}} int buffer_get_field */
1047
1048 /* if we're restricting writes to the base directory,
1049  * check whether the file falls within the dir
1050  * returns 1 if OK, otherwise 0
1051  */
1052 static int check_file_access (const char *file, listen_socket_t *sock) /* {{{ */
1053 {
1054   assert(file != NULL);
1055
1056   if (!config_write_base_only
1057       || JOURNAL_REPLAY(sock)
1058       || config_base_dir == NULL)
1059     return 1;
1060
1061   if (strstr(file, "../") != NULL) goto err;
1062
1063   /* relative paths without "../" are ok */
1064   if (*file != '/') return 1;
1065
1066   /* file must be of the format base + "/" + <1+ char filename> */
1067   if (strlen(file) < _config_base_dir_len + 2) goto err;
1068   if (strncmp(file, config_base_dir, _config_base_dir_len) != 0) goto err;
1069   if (*(file + _config_base_dir_len) != '/') goto err;
1070
1071   return 1;
1072
1073 err:
1074   if (sock != NULL && sock->fd >= 0)
1075     send_response(sock, RESP_ERR, "%s\n", rrd_strerror(EACCES));
1076
1077   return 0;
1078 } /* }}} static int check_file_access */
1079
1080 /* when using a base dir, convert relative paths to absolute paths.
1081  * if necessary, modifies the "filename" pointer to point
1082  * to the new path created in "tmp".  "tmp" is provided
1083  * by the caller and sizeof(tmp) must be >= PATH_MAX.
1084  *
1085  * this allows us to optimize for the expected case (absolute path)
1086  * with a no-op.
1087  */
1088 static void get_abs_path(char **filename, char *tmp)
1089 {
1090   assert(tmp != NULL);
1091   assert(filename != NULL && *filename != NULL);
1092
1093   if (config_base_dir == NULL || **filename == '/')
1094     return;
1095
1096   snprintf(tmp, PATH_MAX, "%s/%s", config_base_dir, *filename);
1097   *filename = tmp;
1098 } /* }}} static int get_abs_path */
1099
1100 static int flush_file (const char *filename) /* {{{ */
1101 {
1102   cache_item_t *ci;
1103
1104   pthread_mutex_lock (&cache_lock);
1105
1106   ci = (cache_item_t *) g_tree_lookup (cache_tree, filename);
1107   if (ci == NULL)
1108   {
1109     pthread_mutex_unlock (&cache_lock);
1110     return (ENOENT);
1111   }
1112
1113   if (ci->values_num > 0)
1114   {
1115     /* Enqueue at head */
1116     enqueue_cache_item (ci, HEAD);
1117     pthread_cond_wait(&ci->flushed, &cache_lock);
1118   }
1119
1120   /* DO NOT DO ANYTHING WITH ci HERE!!  The entry
1121    * may have been purged during our cond_wait() */
1122
1123   pthread_mutex_unlock(&cache_lock);
1124
1125   return (0);
1126 } /* }}} int flush_file */
1127
1128 static int syntax_error(listen_socket_t *sock, command_t *cmd) /* {{{ */
1129 {
1130   char *err = "Syntax error.\n";
1131
1132   if (cmd && cmd->syntax)
1133     err = cmd->syntax;
1134
1135   return send_response(sock, RESP_ERR, "Usage: %s", err);
1136 } /* }}} static int syntax_error() */
1137
1138 static int handle_request_stats (HANDLER_PROTO) /* {{{ */
1139 {
1140   uint64_t copy_queue_length;
1141   uint64_t copy_updates_received;
1142   uint64_t copy_flush_received;
1143   uint64_t copy_updates_written;
1144   uint64_t copy_data_sets_written;
1145   uint64_t copy_journal_bytes;
1146   uint64_t copy_journal_rotate;
1147
1148   uint64_t tree_nodes_number;
1149   uint64_t tree_depth;
1150
1151   pthread_mutex_lock (&stats_lock);
1152   copy_queue_length       = stats_queue_length;
1153   copy_updates_received   = stats_updates_received;
1154   copy_flush_received     = stats_flush_received;
1155   copy_updates_written    = stats_updates_written;
1156   copy_data_sets_written  = stats_data_sets_written;
1157   copy_journal_bytes      = stats_journal_bytes;
1158   copy_journal_rotate     = stats_journal_rotate;
1159   pthread_mutex_unlock (&stats_lock);
1160
1161   pthread_mutex_lock (&cache_lock);
1162   tree_nodes_number = (uint64_t) g_tree_nnodes (cache_tree);
1163   tree_depth        = (uint64_t) g_tree_height (cache_tree);
1164   pthread_mutex_unlock (&cache_lock);
1165
1166   add_response_info(sock,
1167                     "QueueLength: %"PRIu64"\n", copy_queue_length);
1168   add_response_info(sock,
1169                     "UpdatesReceived: %"PRIu64"\n", copy_updates_received);
1170   add_response_info(sock,
1171                     "FlushesReceived: %"PRIu64"\n", copy_flush_received);
1172   add_response_info(sock,
1173                     "UpdatesWritten: %"PRIu64"\n", copy_updates_written);
1174   add_response_info(sock,
1175                     "DataSetsWritten: %"PRIu64"\n", copy_data_sets_written);
1176   add_response_info(sock, "TreeNodesNumber: %"PRIu64"\n", tree_nodes_number);
1177   add_response_info(sock, "TreeDepth: %"PRIu64"\n", tree_depth);
1178   add_response_info(sock, "JournalBytes: %"PRIu64"\n", copy_journal_bytes);
1179   add_response_info(sock, "JournalRotate: %"PRIu64"\n", copy_journal_rotate);
1180
1181   send_response(sock, RESP_OK, "Statistics follow\n");
1182
1183   return (0);
1184 } /* }}} int handle_request_stats */
1185
1186 static int handle_request_flush (HANDLER_PROTO) /* {{{ */
1187 {
1188   char *file, file_tmp[PATH_MAX];
1189   int status;
1190
1191   status = buffer_get_field (&buffer, &buffer_size, &file);
1192   if (status != 0)
1193   {
1194     return syntax_error(sock,cmd);
1195   }
1196   else
1197   {
1198     pthread_mutex_lock(&stats_lock);
1199     stats_flush_received++;
1200     pthread_mutex_unlock(&stats_lock);
1201
1202     get_abs_path(&file, file_tmp);
1203     if (!check_file_access(file, sock)) return 0;
1204
1205     status = flush_file (file);
1206     if (status == 0)
1207       return send_response(sock, RESP_OK, "Successfully flushed %s.\n", file);
1208     else if (status == ENOENT)
1209     {
1210       /* no file in our tree; see whether it exists at all */
1211       struct stat statbuf;
1212
1213       memset(&statbuf, 0, sizeof(statbuf));
1214       if (stat(file, &statbuf) == 0 && S_ISREG(statbuf.st_mode))
1215         return send_response(sock, RESP_OK, "Nothing to flush: %s.\n", file);
1216       else
1217         return send_response(sock, RESP_ERR, "No such file: %s.\n", file);
1218     }
1219     else if (status < 0)
1220       return send_response(sock, RESP_ERR, "Internal error.\n");
1221     else
1222       return send_response(sock, RESP_ERR, "Failed with status %i.\n", status);
1223   }
1224
1225   /* NOTREACHED */
1226   assert(1==0);
1227 } /* }}} int handle_request_flush */
1228
1229 static int handle_request_flushall(HANDLER_PROTO) /* {{{ */
1230 {
1231   RRDD_LOG(LOG_DEBUG, "Received FLUSHALL");
1232
1233   pthread_mutex_lock(&cache_lock);
1234   flush_old_values(-1);
1235   pthread_mutex_unlock(&cache_lock);
1236
1237   return send_response(sock, RESP_OK, "Started flush.\n");
1238 } /* }}} static int handle_request_flushall */
1239
1240 static int handle_request_pending(HANDLER_PROTO) /* {{{ */
1241 {
1242   int status;
1243   char *file, file_tmp[PATH_MAX];
1244   cache_item_t *ci;
1245
1246   status = buffer_get_field(&buffer, &buffer_size, &file);
1247   if (status != 0)
1248     return syntax_error(sock,cmd);
1249
1250   get_abs_path(&file, file_tmp);
1251
1252   pthread_mutex_lock(&cache_lock);
1253   ci = g_tree_lookup(cache_tree, file);
1254   if (ci == NULL)
1255   {
1256     pthread_mutex_unlock(&cache_lock);
1257     return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(ENOENT));
1258   }
1259
1260   for (size_t i=0; i < ci->values_num; i++)
1261     add_response_info(sock, "%s\n", ci->values[i]);
1262
1263   pthread_mutex_unlock(&cache_lock);
1264   return send_response(sock, RESP_OK, "updates pending\n");
1265 } /* }}} static int handle_request_pending */
1266
1267 static int handle_request_forget(HANDLER_PROTO) /* {{{ */
1268 {
1269   int status;
1270   gboolean found;
1271   char *file, file_tmp[PATH_MAX];
1272
1273   status = buffer_get_field(&buffer, &buffer_size, &file);
1274   if (status != 0)
1275     return syntax_error(sock,cmd);
1276
1277   get_abs_path(&file, file_tmp);
1278   if (!check_file_access(file, sock)) return 0;
1279
1280   pthread_mutex_lock(&cache_lock);
1281   found = g_tree_remove(cache_tree, file);
1282   pthread_mutex_unlock(&cache_lock);
1283
1284   if (found == TRUE)
1285   {
1286     if (!JOURNAL_REPLAY(sock))
1287       journal_write("forget", file);
1288
1289     return send_response(sock, RESP_OK, "Gone!\n");
1290   }
1291   else
1292     return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(ENOENT));
1293
1294   /* NOTREACHED */
1295   assert(1==0);
1296 } /* }}} static int handle_request_forget */
1297
1298 static int handle_request_queue (HANDLER_PROTO) /* {{{ */
1299 {
1300   cache_item_t *ci;
1301
1302   pthread_mutex_lock(&cache_lock);
1303
1304   ci = cache_queue_head;
1305   while (ci != NULL)
1306   {
1307     add_response_info(sock, "%d %s\n", ci->values_num, ci->file);
1308     ci = ci->next;
1309   }
1310
1311   pthread_mutex_unlock(&cache_lock);
1312
1313   return send_response(sock, RESP_OK, "in queue.\n");
1314 } /* }}} int handle_request_queue */
1315
1316 static int handle_request_update (HANDLER_PROTO) /* {{{ */
1317 {
1318   char *file, file_tmp[PATH_MAX];
1319   int values_num = 0;
1320   int status;
1321   char orig_buf[CMD_MAX];
1322
1323   cache_item_t *ci;
1324
1325   /* save it for the journal later */
1326   if (!JOURNAL_REPLAY(sock))
1327     strncpy(orig_buf, buffer, buffer_size);
1328
1329   status = buffer_get_field (&buffer, &buffer_size, &file);
1330   if (status != 0)
1331     return syntax_error(sock,cmd);
1332
1333   pthread_mutex_lock(&stats_lock);
1334   stats_updates_received++;
1335   pthread_mutex_unlock(&stats_lock);
1336
1337   get_abs_path(&file, file_tmp);
1338   if (!check_file_access(file, sock)) return 0;
1339
1340   pthread_mutex_lock (&cache_lock);
1341   ci = g_tree_lookup (cache_tree, file);
1342
1343   if (ci == NULL) /* {{{ */
1344   {
1345     struct stat statbuf;
1346     cache_item_t *tmp;
1347
1348     /* don't hold the lock while we setup; stat(2) might block */
1349     pthread_mutex_unlock(&cache_lock);
1350
1351     memset (&statbuf, 0, sizeof (statbuf));
1352     status = stat (file, &statbuf);
1353     if (status != 0)
1354     {
1355       RRDD_LOG (LOG_NOTICE, "handle_request_update: stat (%s) failed.", file);
1356
1357       status = errno;
1358       if (status == ENOENT)
1359         return send_response(sock, RESP_ERR, "No such file: %s\n", file);
1360       else
1361         return send_response(sock, RESP_ERR,
1362                              "stat failed with error %i.\n", status);
1363     }
1364     if (!S_ISREG (statbuf.st_mode))
1365       return send_response(sock, RESP_ERR, "Not a regular file: %s\n", file);
1366
1367     if (access(file, R_OK|W_OK) != 0)
1368       return send_response(sock, RESP_ERR, "Cannot read/write %s: %s\n",
1369                            file, rrd_strerror(errno));
1370
1371     ci = (cache_item_t *) malloc (sizeof (cache_item_t));
1372     if (ci == NULL)
1373     {
1374       RRDD_LOG (LOG_ERR, "handle_request_update: malloc failed.");
1375
1376       return send_response(sock, RESP_ERR, "malloc failed.\n");
1377     }
1378     memset (ci, 0, sizeof (cache_item_t));
1379
1380     ci->file = strdup (file);
1381     if (ci->file == NULL)
1382     {
1383       free (ci);
1384       RRDD_LOG (LOG_ERR, "handle_request_update: strdup failed.");
1385
1386       return send_response(sock, RESP_ERR, "strdup failed.\n");
1387     }
1388
1389     wipe_ci_values(ci, now);
1390     ci->flags = CI_FLAGS_IN_TREE;
1391     pthread_cond_init(&ci->flushed, NULL);
1392
1393     pthread_mutex_lock(&cache_lock);
1394
1395     /* another UPDATE might have added this entry in the meantime */
1396     tmp = g_tree_lookup (cache_tree, file);
1397     if (tmp == NULL)
1398       g_tree_replace (cache_tree, (void *) ci->file, (void *) ci);
1399     else
1400     {
1401       free_cache_item (ci);
1402       ci = tmp;
1403     }
1404
1405     /* state may have changed while we were unlocked */
1406     if (state == SHUTDOWN)
1407       return -1;
1408   } /* }}} */
1409   assert (ci != NULL);
1410
1411   /* don't re-write updates in replay mode */
1412   if (!JOURNAL_REPLAY(sock))
1413     journal_write("update", orig_buf);
1414
1415   while (buffer_size > 0)
1416   {
1417     char *value;
1418     time_t stamp;
1419     char *eostamp;
1420
1421     status = buffer_get_field (&buffer, &buffer_size, &value);
1422     if (status != 0)
1423     {
1424       RRDD_LOG (LOG_INFO, "handle_request_update: Error reading field.");
1425       break;
1426     }
1427
1428     /* make sure update time is always moving forward */
1429     stamp = strtol(value, &eostamp, 10);
1430     if (eostamp == value || eostamp == NULL || *eostamp != ':')
1431     {
1432       pthread_mutex_unlock(&cache_lock);
1433       return send_response(sock, RESP_ERR,
1434                            "Cannot find timestamp in '%s'!\n", value);
1435     }
1436     else if (stamp <= ci->last_update_stamp)
1437     {
1438       pthread_mutex_unlock(&cache_lock);
1439       return send_response(sock, RESP_ERR,
1440                            "illegal attempt to update using time %ld when last"
1441                            " update time is %ld (minimum one second step)\n",
1442                            stamp, ci->last_update_stamp);
1443     }
1444     else
1445       ci->last_update_stamp = stamp;
1446
1447     if (!rrd_add_strdup_chunk(&ci->values, &ci->values_num, value,
1448                               &ci->values_alloc, config_alloc_chunk))
1449     {
1450       RRDD_LOG (LOG_ERR, "handle_request_update: rrd_add_strdup failed.");
1451       continue;
1452     }
1453
1454     values_num++;
1455   }
1456
1457   if (((now - ci->last_flush_time) >= config_write_interval)
1458       && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
1459       && (ci->values_num > 0))
1460   {
1461     enqueue_cache_item (ci, TAIL);
1462   }
1463
1464   pthread_mutex_unlock (&cache_lock);
1465
1466   if (values_num < 1)
1467     return send_response(sock, RESP_ERR, "No values updated.\n");
1468   else
1469     return send_response(sock, RESP_OK,
1470                          "errors, enqueued %i value(s).\n", values_num);
1471
1472   /* NOTREACHED */
1473   assert(1==0);
1474
1475 } /* }}} int handle_request_update */
1476
1477 static int handle_request_fetch (HANDLER_PROTO) /* {{{ */
1478 {
1479   char *file;
1480   char *cf;
1481
1482   char *start_str;
1483   char *end_str;
1484   time_t start_tm;
1485   time_t end_tm;
1486
1487   unsigned long step;
1488   unsigned long ds_cnt;
1489   char **ds_namv;
1490   rrd_value_t *data;
1491
1492   int status;
1493   unsigned long i;
1494   time_t t;
1495   rrd_value_t *data_ptr;
1496
1497   file = NULL;
1498   cf = NULL;
1499   start_str = NULL;
1500   end_str = NULL;
1501
1502   /* Read the arguments */
1503   do /* while (0) */
1504   {
1505     status = buffer_get_field (&buffer, &buffer_size, &file);
1506     if (status != 0)
1507       break;
1508
1509     status = buffer_get_field (&buffer, &buffer_size, &cf);
1510     if (status != 0)
1511       break;
1512
1513     status = buffer_get_field (&buffer, &buffer_size, &start_str);
1514     if (status != 0)
1515     {
1516       start_str = NULL;
1517       status = 0;
1518       break;
1519     }
1520
1521     status = buffer_get_field (&buffer, &buffer_size, &end_str);
1522     if (status != 0)
1523     {
1524       end_str = NULL;
1525       status = 0;
1526       break;
1527     }
1528   } while (0);
1529
1530   if (status != 0)
1531     return (syntax_error(sock,cmd));
1532
1533   status = flush_file (file);
1534   if ((status != 0) && (status != ENOENT))
1535     return (send_response (sock, RESP_ERR,
1536           "flush_file (%s) failed with status %i.\n", file, status));
1537
1538   t = time (NULL); /* "now" */
1539
1540   /* Parse start time */
1541   if (start_str != NULL)
1542   {
1543     char *endptr;
1544     long value;
1545
1546     endptr = NULL;
1547     errno = 0;
1548     value = strtol (start_str, &endptr, /* base = */ 0);
1549     if ((endptr == start_str) || (errno != 0))
1550       return (send_response(sock, RESP_ERR,
1551             "Cannot parse start time `%s': Only simple integers are allowed.\n",
1552             start_str));
1553
1554     if (value > 0)
1555       start_tm = (time_t) value;
1556     else
1557       start_tm = (time_t) (t + value);
1558   }
1559   else
1560   {
1561     start_tm = t - 86400;
1562   }
1563
1564   /* Parse end time */
1565   if (end_str != NULL)
1566   {
1567     char *endptr;
1568     long value;
1569
1570     endptr = NULL;
1571     errno = 0;
1572     value = strtol (end_str, &endptr, /* base = */ 0);
1573     if ((endptr == end_str) || (errno != 0))
1574       return (send_response(sock, RESP_ERR,
1575             "Cannot parse end time `%s': Only simple integers are allowed.\n",
1576             end_str));
1577
1578     if (value > 0)
1579       end_tm = (time_t) value;
1580     else
1581       end_tm = (time_t) (t + value);
1582   }
1583   else
1584   {
1585     end_tm = t;
1586   }
1587
1588   step = -1;
1589   ds_cnt = 0;
1590   ds_namv = NULL;
1591   data = NULL;
1592
1593   status = rrd_fetch_r (file, cf, &start_tm, &end_tm, &step,
1594       &ds_cnt, &ds_namv, &data);
1595   if (status != 0)
1596     return (send_response(sock, RESP_ERR,
1597           "rrd_fetch_r failed: %s\n", rrd_get_error ()));
1598
1599   add_response_info (sock, "FlushVersion: %lu\n", 1);
1600   add_response_info (sock, "Start: %lu\n", (unsigned long) start_tm);
1601   add_response_info (sock, "End: %lu\n", (unsigned long) end_tm);
1602   add_response_info (sock, "Step: %lu\n", step);
1603   add_response_info (sock, "DSCount: %lu\n", ds_cnt);
1604
1605 #define SSTRCAT(buffer,str,buffer_fill) do { \
1606     size_t str_len = strlen (str); \
1607     if ((buffer_fill + str_len) > sizeof (buffer)) \
1608       str_len = sizeof (buffer) - buffer_fill; \
1609     if (str_len > 0) { \
1610       strncpy (buffer + buffer_fill, str, str_len); \
1611       buffer_fill += str_len; \
1612       assert (buffer_fill <= sizeof (buffer)); \
1613       if (buffer_fill == sizeof (buffer)) \
1614         buffer[buffer_fill - 1] = 0; \
1615       else \
1616         buffer[buffer_fill] = 0; \
1617     } \
1618   } while (0)
1619
1620   { /* Add list of DS names */
1621     char linebuf[1024];
1622     size_t linebuf_fill;
1623
1624     memset (linebuf, 0, sizeof (linebuf));
1625     linebuf_fill = 0;
1626     for (i = 0; i < ds_cnt; i++)
1627     {
1628       if (i > 0)
1629         SSTRCAT (linebuf, " ", linebuf_fill);
1630       SSTRCAT (linebuf, ds_namv[i], linebuf_fill);
1631     }
1632     add_response_info (sock, "DSName: %s\n", linebuf);
1633   }
1634
1635   /* Add the actual data */
1636   assert (step > 0);
1637   data_ptr = data;
1638   for (t = start_tm + step; t <= end_tm; t += step)
1639   {
1640     char linebuf[1024];
1641     size_t linebuf_fill;
1642     char tmp[128];
1643
1644     memset (linebuf, 0, sizeof (linebuf));
1645     linebuf_fill = 0;
1646     for (i = 0; i < ds_cnt; i++)
1647     {
1648       snprintf (tmp, sizeof (tmp), " %0.10e", *data_ptr);
1649       tmp[sizeof (tmp) - 1] = 0;
1650       SSTRCAT (linebuf, tmp, linebuf_fill);
1651
1652       data_ptr++;
1653     }
1654
1655     add_response_info (sock, "%10lu:%s\n", (unsigned long) t, linebuf);
1656   } /* for (t) */
1657
1658   return (send_response (sock, RESP_OK, "Success\n"));
1659 #undef SSTRCAT
1660 } /* }}} int handle_request_fetch */
1661
1662 /* we came across a "WROTE" entry during journal replay.
1663  * throw away any values that we have accumulated for this file
1664  */
1665 static int handle_request_wrote (HANDLER_PROTO) /* {{{ */
1666 {
1667   cache_item_t *ci;
1668   const char *file = buffer;
1669
1670   pthread_mutex_lock(&cache_lock);
1671
1672   ci = g_tree_lookup(cache_tree, file);
1673   if (ci == NULL)
1674   {
1675     pthread_mutex_unlock(&cache_lock);
1676     return (0);
1677   }
1678
1679   if (ci->values)
1680     rrd_free_ptrs((void ***) &ci->values, &ci->values_num);
1681
1682   wipe_ci_values(ci, now);
1683   remove_from_queue(ci);
1684
1685   pthread_mutex_unlock(&cache_lock);
1686   return (0);
1687 } /* }}} int handle_request_wrote */
1688
1689 /* start "BATCH" processing */
1690 static int batch_start (HANDLER_PROTO) /* {{{ */
1691 {
1692   int status;
1693   if (sock->batch_start)
1694     return send_response(sock, RESP_ERR, "Already in BATCH\n");
1695
1696   status = send_response(sock, RESP_OK,
1697                          "Go ahead.  End with dot '.' on its own line.\n");
1698   sock->batch_start = time(NULL);
1699   sock->batch_cmd = 0;
1700
1701   return status;
1702 } /* }}} static int batch_start */
1703
1704 /* finish "BATCH" processing and return results to the client */
1705 static int batch_done (HANDLER_PROTO) /* {{{ */
1706 {
1707   assert(sock->batch_start);
1708   sock->batch_start = 0;
1709   sock->batch_cmd  = 0;
1710   return send_response(sock, RESP_OK, "errors\n");
1711 } /* }}} static int batch_done */
1712
1713 static int handle_request_quit (HANDLER_PROTO) /* {{{ */
1714 {
1715   return -1;
1716 } /* }}} static int handle_request_quit */
1717
1718 static command_t list_of_commands[] = { /* {{{ */
1719   {
1720     "UPDATE",
1721     handle_request_update,
1722     CMD_CONTEXT_ANY,
1723     "UPDATE <filename> <values> [<values> ...]\n"
1724     ,
1725     "Adds the given file to the internal cache if it is not yet known and\n"
1726     "appends the given value(s) to the entry. See the rrdcached(1) manpage\n"
1727     "for details.\n"
1728     "\n"
1729     "Each <values> has the following form:\n"
1730     "  <values> = <time>:<value>[:<value>[...]]\n"
1731     "See the rrdupdate(1) manpage for details.\n"
1732   },
1733   {
1734     "WROTE",
1735     handle_request_wrote,
1736     CMD_CONTEXT_JOURNAL,
1737     NULL,
1738     NULL
1739   },
1740   {
1741     "FLUSH",
1742     handle_request_flush,
1743     CMD_CONTEXT_CLIENT | CMD_CONTEXT_BATCH,
1744     "FLUSH <filename>\n"
1745     ,
1746     "Adds the given filename to the head of the update queue and returns\n"
1747     "after it has been dequeued.\n"
1748   },
1749   {
1750     "FLUSHALL",
1751     handle_request_flushall,
1752     CMD_CONTEXT_CLIENT,
1753     "FLUSHALL\n"
1754     ,
1755     "Triggers writing of all pending updates.  Returns immediately.\n"
1756   },
1757   {
1758     "PENDING",
1759     handle_request_pending,
1760     CMD_CONTEXT_CLIENT,
1761     "PENDING <filename>\n"
1762     ,
1763     "Shows any 'pending' updates for a file, in order.\n"
1764     "The updates shown have not yet been written to the underlying RRD file.\n"
1765   },
1766   {
1767     "FORGET",
1768     handle_request_forget,
1769     CMD_CONTEXT_ANY,
1770     "FORGET <filename>\n"
1771     ,
1772     "Removes the file completely from the cache.\n"
1773     "Any pending updates for the file will be lost.\n"
1774   },
1775   {
1776     "QUEUE",
1777     handle_request_queue,
1778     CMD_CONTEXT_CLIENT,
1779     "QUEUE\n"
1780     ,
1781         "Shows all files in the output queue.\n"
1782     "The output is zero or more lines in the following format:\n"
1783     "(where <num_vals> is the number of values to be written)\n"
1784     "\n"
1785     "<num_vals> <filename>\n"
1786   },
1787   {
1788     "STATS",
1789     handle_request_stats,
1790     CMD_CONTEXT_CLIENT,
1791     "STATS\n"
1792     ,
1793     "Returns some performance counters, see the rrdcached(1) manpage for\n"
1794     "a description of the values.\n"
1795   },
1796   {
1797     "HELP",
1798     handle_request_help,
1799     CMD_CONTEXT_CLIENT,
1800     "HELP [<command>]\n",
1801     NULL, /* special! */
1802   },
1803   {
1804     "BATCH",
1805     batch_start,
1806     CMD_CONTEXT_CLIENT,
1807     "BATCH\n"
1808     ,
1809     "The 'BATCH' command permits the client to initiate a bulk load\n"
1810     "   of commands to rrdcached.\n"
1811     "\n"
1812     "Usage:\n"
1813     "\n"
1814     "    client: BATCH\n"
1815     "    server: 0 Go ahead.  End with dot '.' on its own line.\n"
1816     "    client: command #1\n"
1817     "    client: command #2\n"
1818     "    client: ... and so on\n"
1819     "    client: .\n"
1820     "    server: 2 errors\n"
1821     "    server: 7 message for command #7\n"
1822     "    server: 9 message for command #9\n"
1823     "\n"
1824     "For more information, consult the rrdcached(1) documentation.\n"
1825   },
1826   {
1827     ".",   /* BATCH terminator */
1828     batch_done,
1829     CMD_CONTEXT_BATCH,
1830     NULL,
1831     NULL
1832   },
1833   {
1834     "FETCH",
1835     handle_request_fetch,
1836     CMD_CONTEXT_CLIENT,
1837     "FETCH <file> <CF> [<start> [<end>]]\n"
1838     ,
1839     "The 'FETCH' can be used by the client to retrieve values from an RRD file.\n"
1840   },
1841   {
1842     "QUIT",
1843     handle_request_quit,
1844     CMD_CONTEXT_CLIENT | CMD_CONTEXT_BATCH,
1845     "QUIT\n"
1846     ,
1847     "Disconnect from rrdcached.\n"
1848   }
1849 }; /* }}} command_t list_of_commands[] */
1850 static size_t list_of_commands_len = sizeof (list_of_commands)
1851   / sizeof (list_of_commands[0]);
1852
1853 static command_t *find_command(char *cmd)
1854 {
1855   size_t i;
1856
1857   for (i = 0; i < list_of_commands_len; i++)
1858     if (strcasecmp(cmd, list_of_commands[i].cmd) == 0)
1859       return (&list_of_commands[i]);
1860   return NULL;
1861 }
1862
1863 /* We currently use the index in the `list_of_commands' array as a bit position
1864  * in `listen_socket_t.permissions'. This member schould NEVER be accessed from
1865  * outside these functions so that switching to a more elegant storage method
1866  * is easily possible. */
1867 static ssize_t find_command_index (const char *cmd) /* {{{ */
1868 {
1869   size_t i;
1870
1871   for (i = 0; i < list_of_commands_len; i++)
1872     if (strcasecmp(cmd, list_of_commands[i].cmd) == 0)
1873       return ((ssize_t) i);
1874   return (-1);
1875 } /* }}} ssize_t find_command_index */
1876
1877 static int socket_permission_check (listen_socket_t *sock, /* {{{ */
1878     const char *cmd)
1879 {
1880   ssize_t i;
1881
1882   if (JOURNAL_REPLAY(sock))
1883     return (1);
1884
1885   if (cmd == NULL)
1886     return (-1);
1887
1888   if ((strcasecmp ("QUIT", cmd) == 0)
1889       || (strcasecmp ("HELP", cmd) == 0))
1890     return (1);
1891   else if (strcmp (".", cmd) == 0)
1892     cmd = "BATCH";
1893
1894   i = find_command_index (cmd);
1895   if (i < 0)
1896     return (-1);
1897   assert (i < 32);
1898
1899   if ((sock->permissions & (1 << i)) != 0)
1900     return (1);
1901   return (0);
1902 } /* }}} int socket_permission_check */
1903
1904 static int socket_permission_add (listen_socket_t *sock, /* {{{ */
1905     const char *cmd)
1906 {
1907   ssize_t i;
1908
1909   i = find_command_index (cmd);
1910   if (i < 0)
1911     return (-1);
1912   assert (i < 32);
1913
1914   sock->permissions |= (1 << i);
1915   return (0);
1916 } /* }}} int socket_permission_add */
1917
1918 /* check whether commands are received in the expected context */
1919 static int command_check_context(listen_socket_t *sock, command_t *cmd)
1920 {
1921   if (JOURNAL_REPLAY(sock))
1922     return (cmd->context & CMD_CONTEXT_JOURNAL);
1923   else if (sock->batch_start)
1924     return (cmd->context & CMD_CONTEXT_BATCH);
1925   else
1926     return (cmd->context & CMD_CONTEXT_CLIENT);
1927
1928   /* NOTREACHED */
1929   assert(1==0);
1930 }
1931
1932 static int handle_request_help (HANDLER_PROTO) /* {{{ */
1933 {
1934   int status;
1935   char *cmd_str;
1936   char *resp_txt;
1937   command_t *help = NULL;
1938
1939   status = buffer_get_field (&buffer, &buffer_size, &cmd_str);
1940   if (status == 0)
1941     help = find_command(cmd_str);
1942
1943   if (help && (help->syntax || help->help))
1944   {
1945     char tmp[CMD_MAX];
1946
1947     snprintf(tmp, sizeof(tmp)-1, "Help for %s\n", help->cmd);
1948     resp_txt = tmp;
1949
1950     if (help->syntax)
1951       add_response_info(sock, "Usage: %s\n", help->syntax);
1952
1953     if (help->help)
1954       add_response_info(sock, "%s\n", help->help);
1955   }
1956   else
1957   {
1958     size_t i;
1959
1960     resp_txt = "Command overview\n";
1961
1962     for (i = 0; i < list_of_commands_len; i++)
1963     {
1964       if (list_of_commands[i].syntax == NULL)
1965         continue;
1966       add_response_info (sock, "%s", list_of_commands[i].syntax);
1967     }
1968   }
1969
1970   return send_response(sock, RESP_OK, resp_txt);
1971 } /* }}} int handle_request_help */
1972
1973 static int handle_request (DISPATCH_PROTO) /* {{{ */
1974 {
1975   char *buffer_ptr = buffer;
1976   char *cmd_str = NULL;
1977   command_t *cmd = NULL;
1978   int status;
1979
1980   assert (buffer[buffer_size - 1] == '\0');
1981
1982   status = buffer_get_field (&buffer_ptr, &buffer_size, &cmd_str);
1983   if (status != 0)
1984   {
1985     RRDD_LOG (LOG_INFO, "handle_request: Unable parse command.");
1986     return (-1);
1987   }
1988
1989   if (sock != NULL && sock->batch_start)
1990     sock->batch_cmd++;
1991
1992   cmd = find_command(cmd_str);
1993   if (!cmd)
1994     return send_response(sock, RESP_ERR, "Unknown command: %s\n", cmd_str);
1995
1996   if (!socket_permission_check (sock, cmd->cmd))
1997     return send_response(sock, RESP_ERR, "Permission denied.\n");
1998
1999   if (!command_check_context(sock, cmd))
2000     return send_response(sock, RESP_ERR, "Can't use '%s' here.\n", cmd_str);
2001
2002   return cmd->handler(cmd, sock, now, buffer_ptr, buffer_size);
2003 } /* }}} int handle_request */
2004
2005 static void journal_set_free (journal_set *js) /* {{{ */
2006 {
2007   if (js == NULL)
2008     return;
2009
2010   rrd_free_ptrs((void ***) &js->files, &js->files_num);
2011
2012   free(js);
2013 } /* }}} journal_set_free */
2014
2015 static void journal_set_remove (journal_set *js) /* {{{ */
2016 {
2017   if (js == NULL)
2018     return;
2019
2020   for (uint i=0; i < js->files_num; i++)
2021   {
2022     RRDD_LOG(LOG_DEBUG, "removing old journal %s", js->files[i]);
2023     unlink(js->files[i]);
2024   }
2025 } /* }}} journal_set_remove */
2026
2027 /* close current journal file handle.
2028  * MUST hold journal_lock before calling */
2029 static void journal_close(void) /* {{{ */
2030 {
2031   if (journal_fh != NULL)
2032   {
2033     if (fclose(journal_fh) != 0)
2034       RRDD_LOG(LOG_ERR, "cannot close journal: %s", rrd_strerror(errno));
2035   }
2036
2037   journal_fh = NULL;
2038   journal_size = 0;
2039 } /* }}} journal_close */
2040
2041 /* MUST hold journal_lock before calling */
2042 static void journal_new_file(void) /* {{{ */
2043 {
2044   struct timeval now;
2045   int  new_fd;
2046   char new_file[PATH_MAX + 1];
2047
2048   assert(journal_dir != NULL);
2049   assert(journal_cur != NULL);
2050
2051   journal_close();
2052
2053   gettimeofday(&now, NULL);
2054   /* this format assures that the files sort in strcmp() order */
2055   snprintf(new_file, PATH_MAX, "%s/%s.%010d.%06d",
2056            journal_dir, JOURNAL_BASE, (int)now.tv_sec, (int)now.tv_usec);
2057
2058   new_fd = open(new_file, O_WRONLY|O_CREAT|O_APPEND,
2059                 S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
2060   if (new_fd < 0)
2061     goto error;
2062
2063   journal_fh = fdopen(new_fd, "a");
2064   if (journal_fh == NULL)
2065     goto error;
2066
2067   journal_size = ftell(journal_fh);
2068   RRDD_LOG(LOG_DEBUG, "started new journal %s", new_file);
2069
2070   /* record the file in the journal set */
2071   rrd_add_strdup(&journal_cur->files, &journal_cur->files_num, new_file);
2072
2073   return;
2074
2075 error:
2076   RRDD_LOG(LOG_CRIT,
2077            "JOURNALING DISABLED: Error while trying to create %s : %s",
2078            new_file, rrd_strerror(errno));
2079   RRDD_LOG(LOG_CRIT,
2080            "JOURNALING DISABLED: All values will be flushed at shutdown");
2081
2082   close(new_fd);
2083   config_flush_at_shutdown = 1;
2084
2085 } /* }}} journal_new_file */
2086
2087 /* MUST NOT hold journal_lock before calling this */
2088 static void journal_rotate(void) /* {{{ */
2089 {
2090   journal_set *old_js = NULL;
2091
2092   if (journal_dir == NULL)
2093     return;
2094
2095   RRDD_LOG(LOG_DEBUG, "rotating journals");
2096
2097   pthread_mutex_lock(&stats_lock);
2098   ++stats_journal_rotate;
2099   pthread_mutex_unlock(&stats_lock);
2100
2101   pthread_mutex_lock(&journal_lock);
2102
2103   journal_close();
2104
2105   /* rotate the journal sets */
2106   old_js = journal_old;
2107   journal_old = journal_cur;
2108   journal_cur = calloc(1, sizeof(journal_set));
2109
2110   if (journal_cur != NULL)
2111     journal_new_file();
2112   else
2113     RRDD_LOG(LOG_CRIT, "journal_rotate: malloc(journal_set) failed\n");
2114
2115   pthread_mutex_unlock(&journal_lock);
2116
2117   journal_set_remove(old_js);
2118   journal_set_free  (old_js);
2119
2120 } /* }}} static void journal_rotate */
2121
2122 /* MUST hold journal_lock when calling */
2123 static void journal_done(void) /* {{{ */
2124 {
2125   if (journal_cur == NULL)
2126     return;
2127
2128   journal_close();
2129
2130   if (config_flush_at_shutdown)
2131   {
2132     RRDD_LOG(LOG_INFO, "removing journals");
2133     journal_set_remove(journal_old);
2134     journal_set_remove(journal_cur);
2135   }
2136   else
2137   {
2138     RRDD_LOG(LOG_INFO, "expedited shutdown; "
2139              "journals will be used at next startup");
2140   }
2141
2142   journal_set_free(journal_cur);
2143   journal_set_free(journal_old);
2144   free(journal_dir);
2145
2146 } /* }}} static void journal_done */
2147
2148 static int journal_write(char *cmd, char *args) /* {{{ */
2149 {
2150   int chars;
2151
2152   if (journal_fh == NULL)
2153     return 0;
2154
2155   pthread_mutex_lock(&journal_lock);
2156   chars = fprintf(journal_fh, "%s %s\n", cmd, args);
2157   journal_size += chars;
2158
2159   if (journal_size > JOURNAL_MAX)
2160     journal_new_file();
2161
2162   pthread_mutex_unlock(&journal_lock);
2163
2164   if (chars > 0)
2165   {
2166     pthread_mutex_lock(&stats_lock);
2167     stats_journal_bytes += chars;
2168     pthread_mutex_unlock(&stats_lock);
2169   }
2170
2171   return chars;
2172 } /* }}} static int journal_write */
2173
2174 static int journal_replay (const char *file) /* {{{ */
2175 {
2176   FILE *fh;
2177   int entry_cnt = 0;
2178   int fail_cnt = 0;
2179   uint64_t line = 0;
2180   char entry[CMD_MAX];
2181   time_t now;
2182
2183   if (file == NULL) return 0;
2184
2185   {
2186     char *reason = "unknown error";
2187     int status = 0;
2188     struct stat statbuf;
2189
2190     memset(&statbuf, 0, sizeof(statbuf));
2191     if (stat(file, &statbuf) != 0)
2192     {
2193       reason = "stat error";
2194       status = errno;
2195     }
2196     else if (!S_ISREG(statbuf.st_mode))
2197     {
2198       reason = "not a regular file";
2199       status = EPERM;
2200     }
2201     if (statbuf.st_uid != daemon_uid)
2202     {
2203       reason = "not owned by daemon user";
2204       status = EACCES;
2205     }
2206     if (statbuf.st_mode & (S_IWGRP|S_IWOTH))
2207     {
2208       reason = "must not be user/group writable";
2209       status = EACCES;
2210     }
2211
2212     if (status != 0)
2213     {
2214       RRDD_LOG(LOG_ERR, "journal_replay: %s : %s (%s)",
2215                file, rrd_strerror(status), reason);
2216       return 0;
2217     }
2218   }
2219
2220   fh = fopen(file, "r");
2221   if (fh == NULL)
2222   {
2223     if (errno != ENOENT)
2224       RRDD_LOG(LOG_ERR, "journal_replay: cannot open journal file: '%s' (%s)",
2225                file, rrd_strerror(errno));
2226     return 0;
2227   }
2228   else
2229     RRDD_LOG(LOG_NOTICE, "replaying from journal: %s", file);
2230
2231   now = time(NULL);
2232
2233   while(!feof(fh))
2234   {
2235     size_t entry_len;
2236
2237     ++line;
2238     if (fgets(entry, sizeof(entry), fh) == NULL)
2239       break;
2240     entry_len = strlen(entry);
2241
2242     /* check \n termination in case journal writing crashed mid-line */
2243     if (entry_len == 0)
2244       continue;
2245     else if (entry[entry_len - 1] != '\n')
2246     {
2247       RRDD_LOG(LOG_NOTICE, "Malformed journal entry at line %"PRIu64, line);
2248       ++fail_cnt;
2249       continue;
2250     }
2251
2252     entry[entry_len - 1] = '\0';
2253
2254     if (handle_request(NULL, now, entry, entry_len) == 0)
2255       ++entry_cnt;
2256     else
2257       ++fail_cnt;
2258   }
2259
2260   fclose(fh);
2261
2262   RRDD_LOG(LOG_INFO, "Replayed %d entries (%d failures)",
2263            entry_cnt, fail_cnt);
2264
2265   return entry_cnt > 0 ? 1 : 0;
2266 } /* }}} static int journal_replay */
2267
2268 static int journal_sort(const void *v1, const void *v2)
2269 {
2270   char **jn1 = (char **) v1;
2271   char **jn2 = (char **) v2;
2272
2273   return strcmp(*jn1,*jn2);
2274 }
2275
2276 static void journal_init(void) /* {{{ */
2277 {
2278   int had_journal = 0;
2279   DIR *dir;
2280   struct dirent *dent;
2281   char path[PATH_MAX+1];
2282
2283   if (journal_dir == NULL) return;
2284
2285   pthread_mutex_lock(&journal_lock);
2286
2287   journal_cur = calloc(1, sizeof(journal_set));
2288   if (journal_cur == NULL)
2289   {
2290     RRDD_LOG(LOG_CRIT, "journal_rotate: malloc(journal_set) failed\n");
2291     return;
2292   }
2293
2294   RRDD_LOG(LOG_INFO, "checking for journal files");
2295
2296   /* Handle old journal files during transition.  This gives them the
2297    * correct sort order.  TODO: remove after first release
2298    */
2299   {
2300     char old_path[PATH_MAX+1];
2301     snprintf(old_path, PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE ".old" );
2302     snprintf(path,     PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE ".0000");
2303     rename(old_path, path);
2304
2305     snprintf(old_path, PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE        );
2306     snprintf(path,     PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE ".0001");
2307     rename(old_path, path);
2308   }
2309
2310   dir = opendir(journal_dir);
2311   if (!dir) {
2312     RRDD_LOG(LOG_CRIT, "journal_init: opendir(%s) failed\n", journal_dir);
2313     return;
2314   }
2315   while ((dent = readdir(dir)) != NULL)
2316   {
2317     /* looks like a journal file? */
2318     if (strncmp(dent->d_name, JOURNAL_BASE, strlen(JOURNAL_BASE)))
2319       continue;
2320
2321     snprintf(path, PATH_MAX, "%s/%s", journal_dir, dent->d_name);
2322
2323     if (!rrd_add_strdup(&journal_cur->files, &journal_cur->files_num, path))
2324     {
2325       RRDD_LOG(LOG_CRIT, "journal_init: cannot add journal file %s!",
2326                dent->d_name);
2327       break;
2328     }
2329   }
2330   closedir(dir);
2331
2332   qsort(journal_cur->files, journal_cur->files_num,
2333         sizeof(journal_cur->files[0]), journal_sort);
2334
2335   for (uint i=0; i < journal_cur->files_num; i++)
2336     had_journal += journal_replay(journal_cur->files[i]);
2337
2338   journal_new_file();
2339
2340   /* it must have been a crash.  start a flush */
2341   if (had_journal && config_flush_at_shutdown)
2342     flush_old_values(-1);
2343
2344   pthread_mutex_unlock(&journal_lock);
2345
2346   RRDD_LOG(LOG_INFO, "journal processing complete");
2347
2348 } /* }}} static void journal_init */
2349
2350 static void free_listen_socket(listen_socket_t *sock) /* {{{ */
2351 {
2352   assert(sock != NULL);
2353
2354   free(sock->rbuf);  sock->rbuf = NULL;
2355   free(sock->wbuf);  sock->wbuf = NULL;
2356   free(sock);
2357 } /* }}} void free_listen_socket */
2358
2359 static void close_connection(listen_socket_t *sock) /* {{{ */
2360 {
2361   if (sock->fd >= 0)
2362   {
2363     close(sock->fd);
2364     sock->fd = -1;
2365   }
2366
2367   free_listen_socket(sock);
2368
2369 } /* }}} void close_connection */
2370
2371 static void *connection_thread_main (void *args) /* {{{ */
2372 {
2373   listen_socket_t *sock;
2374   int fd;
2375
2376   sock = (listen_socket_t *) args;
2377   fd = sock->fd;
2378
2379   /* init read buffers */
2380   sock->next_read = sock->next_cmd = 0;
2381   sock->rbuf = malloc(RBUF_SIZE);
2382   if (sock->rbuf == NULL)
2383   {
2384     RRDD_LOG(LOG_ERR, "connection_thread_main: cannot malloc read buffer");
2385     close_connection(sock);
2386     return NULL;
2387   }
2388
2389   pthread_mutex_lock (&connection_threads_lock);
2390   connection_threads_num++;
2391   pthread_mutex_unlock (&connection_threads_lock);
2392
2393   while (state == RUNNING)
2394   {
2395     char *cmd;
2396     ssize_t cmd_len;
2397     ssize_t rbytes;
2398     time_t now;
2399
2400     struct pollfd pollfd;
2401     int status;
2402
2403     pollfd.fd = fd;
2404     pollfd.events = POLLIN | POLLPRI;
2405     pollfd.revents = 0;
2406
2407     status = poll (&pollfd, 1, /* timeout = */ 500);
2408     if (state != RUNNING)
2409       break;
2410     else if (status == 0) /* timeout */
2411       continue;
2412     else if (status < 0) /* error */
2413     {
2414       status = errno;
2415       if (status != EINTR)
2416         RRDD_LOG (LOG_ERR, "connection_thread_main: poll(2) failed.");
2417       continue;
2418     }
2419
2420     if ((pollfd.revents & POLLHUP) != 0) /* normal shutdown */
2421       break;
2422     else if ((pollfd.revents & (POLLIN | POLLPRI)) == 0)
2423     {
2424       RRDD_LOG (LOG_WARNING, "connection_thread_main: "
2425           "poll(2) returned something unexpected: %#04hx",
2426           pollfd.revents);
2427       break;
2428     }
2429
2430     rbytes = read(fd, sock->rbuf + sock->next_read,
2431                   RBUF_SIZE - sock->next_read);
2432     if (rbytes < 0)
2433     {
2434       RRDD_LOG(LOG_ERR, "connection_thread_main: read() failed.");
2435       break;
2436     }
2437     else if (rbytes == 0)
2438       break; /* eof */
2439
2440     sock->next_read += rbytes;
2441
2442     if (sock->batch_start)
2443       now = sock->batch_start;
2444     else
2445       now = time(NULL);
2446
2447     while ((cmd = next_cmd(sock, &cmd_len)) != NULL)
2448     {
2449       status = handle_request (sock, now, cmd, cmd_len+1);
2450       if (status != 0)
2451         goto out_close;
2452     }
2453   }
2454
2455 out_close:
2456   close_connection(sock);
2457
2458   /* Remove this thread from the connection threads list */
2459   pthread_mutex_lock (&connection_threads_lock);
2460   connection_threads_num--;
2461   if (connection_threads_num <= 0)
2462     pthread_cond_broadcast(&connection_threads_done);
2463   pthread_mutex_unlock (&connection_threads_lock);
2464
2465   return (NULL);
2466 } /* }}} void *connection_thread_main */
2467
2468 static int open_listen_socket_unix (const listen_socket_t *sock) /* {{{ */
2469 {
2470   int fd;
2471   struct sockaddr_un sa;
2472   listen_socket_t *temp;
2473   int status;
2474   const char *path;
2475   char *path_copy, *dir;
2476
2477   path = sock->addr;
2478   if (strncmp(path, "unix:", strlen("unix:")) == 0)
2479     path += strlen("unix:");
2480
2481   /* dirname may modify its argument */
2482   path_copy = strdup(path);
2483   if (path_copy == NULL)
2484   {
2485     fprintf(stderr, "rrdcached: strdup(): %s\n",
2486         rrd_strerror(errno));
2487     return (-1);
2488   }
2489
2490   dir = dirname(path_copy);
2491   if (rrd_mkdir_p(dir, 0777) != 0)
2492   {
2493     fprintf(stderr, "Failed to create socket directory '%s': %s\n",
2494         dir, rrd_strerror(errno));
2495     return (-1);
2496   }
2497
2498   free(path_copy);
2499
2500   temp = (listen_socket_t *) rrd_realloc (listen_fds,
2501       sizeof (listen_fds[0]) * (listen_fds_num + 1));
2502   if (temp == NULL)
2503   {
2504     fprintf (stderr, "rrdcached: open_listen_socket_unix: realloc failed.\n");
2505     return (-1);
2506   }
2507   listen_fds = temp;
2508   memcpy (listen_fds + listen_fds_num, sock, sizeof (listen_fds[0]));
2509
2510   fd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
2511   if (fd < 0)
2512   {
2513     fprintf (stderr, "rrdcached: unix socket(2) failed: %s\n",
2514              rrd_strerror(errno));
2515     return (-1);
2516   }
2517
2518   memset (&sa, 0, sizeof (sa));
2519   sa.sun_family = AF_UNIX;
2520   strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
2521
2522   /* if we've gotten this far, we own the pid file.  any daemon started
2523    * with the same args must not be alive.  therefore, ensure that we can
2524    * create the socket...
2525    */
2526   unlink(path);
2527
2528   status = bind (fd, (struct sockaddr *) &sa, sizeof (sa));
2529   if (status != 0)
2530   {
2531     fprintf (stderr, "rrdcached: bind(%s) failed: %s.\n",
2532              path, rrd_strerror(errno));
2533     close (fd);
2534     return (-1);
2535   }
2536
2537   /* tweak the sockets group ownership */
2538   if (sock->socket_group != (gid_t)-1)
2539   {
2540     if ( (chown(path, getuid(), sock->socket_group) != 0) ||
2541          (chmod(path, (S_IRUSR|S_IWUSR|S_IXUSR | S_IRGRP|S_IWGRP)) != 0) )
2542     {
2543       fprintf(stderr, "rrdcached: failed to set socket group permissions (%s)\n", strerror(errno));
2544     }
2545   }
2546
2547   if (sock->socket_permissions != (mode_t)-1)
2548   {
2549     if (chmod(path, sock->socket_permissions) != 0)
2550       fprintf(stderr, "rrdcached: failed to set socket file permissions (%o): %s\n",
2551           (unsigned int)sock->socket_permissions, strerror(errno));
2552   }
2553
2554   status = listen (fd, /* backlog = */ 10);
2555   if (status != 0)
2556   {
2557     fprintf (stderr, "rrdcached: listen(%s) failed: %s.\n",
2558              path, rrd_strerror(errno));
2559     close (fd);
2560     unlink (path);
2561     return (-1);
2562   }
2563
2564   listen_fds[listen_fds_num].fd = fd;
2565   listen_fds[listen_fds_num].family = PF_UNIX;
2566   strncpy(listen_fds[listen_fds_num].addr, path,
2567           sizeof (listen_fds[listen_fds_num].addr) - 1);
2568   listen_fds_num++;
2569
2570   return (0);
2571 } /* }}} int open_listen_socket_unix */
2572
2573 static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */
2574 {
2575   struct addrinfo ai_hints;
2576   struct addrinfo *ai_res;
2577   struct addrinfo *ai_ptr;
2578   char addr_copy[NI_MAXHOST];
2579   char *addr;
2580   char *port;
2581   int status;
2582
2583   strncpy (addr_copy, sock->addr, sizeof(addr_copy)-1);
2584   addr_copy[sizeof (addr_copy) - 1] = 0;
2585   addr = addr_copy;
2586
2587   memset (&ai_hints, 0, sizeof (ai_hints));
2588   ai_hints.ai_flags = 0;
2589 #ifdef AI_ADDRCONFIG
2590   ai_hints.ai_flags |= AI_ADDRCONFIG;
2591 #endif
2592   ai_hints.ai_family = AF_UNSPEC;
2593   ai_hints.ai_socktype = SOCK_STREAM;
2594
2595   port = NULL;
2596   if (*addr == '[') /* IPv6+port format */
2597   {
2598     /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
2599     addr++;
2600
2601     port = strchr (addr, ']');
2602     if (port == NULL)
2603     {
2604       fprintf (stderr, "rrdcached: Malformed address: %s\n", sock->addr);
2605       return (-1);
2606     }
2607     *port = 0;
2608     port++;
2609
2610     if (*port == ':')
2611       port++;
2612     else if (*port == 0)
2613       port = NULL;
2614     else
2615     {
2616       fprintf (stderr, "rrdcached: Garbage after address: %s\n", port);
2617       return (-1);
2618     }
2619   } /* if (*addr == '[') */
2620   else
2621   {
2622     port = rindex(addr, ':');
2623     if (port != NULL)
2624     {
2625       *port = 0;
2626       port++;
2627     }
2628   }
2629   ai_res = NULL;
2630   status = getaddrinfo (addr,
2631                         port == NULL ? RRDCACHED_DEFAULT_PORT : port,
2632                         &ai_hints, &ai_res);
2633   if (status != 0)
2634   {
2635     fprintf (stderr, "rrdcached: getaddrinfo(%s) failed: %s\n",
2636              addr, gai_strerror (status));
2637     return (-1);
2638   }
2639
2640   for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
2641   {
2642     int fd;
2643     listen_socket_t *temp;
2644     int one = 1;
2645
2646     temp = (listen_socket_t *) rrd_realloc (listen_fds,
2647         sizeof (listen_fds[0]) * (listen_fds_num + 1));
2648     if (temp == NULL)
2649     {
2650       fprintf (stderr,
2651                "rrdcached: open_listen_socket_network: realloc failed.\n");
2652       continue;
2653     }
2654     listen_fds = temp;
2655     memcpy (listen_fds + listen_fds_num, sock, sizeof (listen_fds[0]));
2656
2657     fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
2658     if (fd < 0)
2659     {
2660       fprintf (stderr, "rrdcached: network socket(2) failed: %s.\n",
2661                rrd_strerror(errno));
2662       continue;
2663     }
2664
2665     setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
2666
2667     status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
2668     if (status != 0)
2669     {
2670       fprintf (stderr, "rrdcached: bind(%s) failed: %s.\n",
2671                sock->addr, rrd_strerror(errno));
2672       close (fd);
2673       continue;
2674     }
2675
2676     status = listen (fd, /* backlog = */ 10);
2677     if (status != 0)
2678     {
2679       fprintf (stderr, "rrdcached: listen(%s) failed: %s\n.",
2680                sock->addr, rrd_strerror(errno));
2681       close (fd);
2682       freeaddrinfo(ai_res);
2683       return (-1);
2684     }
2685
2686     listen_fds[listen_fds_num].fd = fd;
2687     listen_fds[listen_fds_num].family = ai_ptr->ai_family;
2688     listen_fds_num++;
2689   } /* for (ai_ptr) */
2690
2691   freeaddrinfo(ai_res);
2692   return (0);
2693 } /* }}} static int open_listen_socket_network */
2694
2695 static int open_listen_socket (const listen_socket_t *sock) /* {{{ */
2696 {
2697   assert(sock != NULL);
2698   assert(sock->addr != NULL);
2699
2700   if (strncmp ("unix:", sock->addr, strlen ("unix:")) == 0
2701       || sock->addr[0] == '/')
2702     return (open_listen_socket_unix(sock));
2703   else
2704     return (open_listen_socket_network(sock));
2705 } /* }}} int open_listen_socket */
2706
2707 static int close_listen_sockets (void) /* {{{ */
2708 {
2709   size_t i;
2710
2711   for (i = 0; i < listen_fds_num; i++)
2712   {
2713     close (listen_fds[i].fd);
2714
2715     if (listen_fds[i].family == PF_UNIX)
2716       unlink(listen_fds[i].addr);
2717   }
2718
2719   free (listen_fds);
2720   listen_fds = NULL;
2721   listen_fds_num = 0;
2722
2723   return (0);
2724 } /* }}} int close_listen_sockets */
2725
2726 static void *listen_thread_main (void UNUSED(*args)) /* {{{ */
2727 {
2728   struct pollfd *pollfds;
2729   int pollfds_num;
2730   int status;
2731   int i;
2732
2733   if (listen_fds_num < 1)
2734   {
2735     RRDD_LOG(LOG_ERR, "listen_thread_main: no listen_fds !");
2736     return (NULL);
2737   }
2738
2739   pollfds_num = listen_fds_num;
2740   pollfds = (struct pollfd *) malloc (sizeof (*pollfds) * pollfds_num);
2741   if (pollfds == NULL)
2742   {
2743     RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
2744     return (NULL);
2745   }
2746   memset (pollfds, 0, sizeof (*pollfds) * pollfds_num);
2747
2748   RRDD_LOG(LOG_INFO, "listening for connections");
2749
2750   while (state == RUNNING)
2751   {
2752     for (i = 0; i < pollfds_num; i++)
2753     {
2754       pollfds[i].fd = listen_fds[i].fd;
2755       pollfds[i].events = POLLIN | POLLPRI;
2756       pollfds[i].revents = 0;
2757     }
2758
2759     status = poll (pollfds, pollfds_num, /* timeout = */ 1000);
2760     if (state != RUNNING)
2761       break;
2762     else if (status == 0) /* timeout */
2763       continue;
2764     else if (status < 0) /* error */
2765     {
2766       status = errno;
2767       if (status != EINTR)
2768       {
2769         RRDD_LOG (LOG_ERR, "listen_thread_main: poll(2) failed.");
2770       }
2771       continue;
2772     }
2773
2774     for (i = 0; i < pollfds_num; i++)
2775     {
2776       listen_socket_t *client_sock;
2777       struct sockaddr_storage client_sa;
2778       socklen_t client_sa_size;
2779       pthread_t tid;
2780       pthread_attr_t attr;
2781
2782       if (pollfds[i].revents == 0)
2783         continue;
2784
2785       if ((pollfds[i].revents & (POLLIN | POLLPRI)) == 0)
2786       {
2787         RRDD_LOG (LOG_ERR, "listen_thread_main: "
2788             "poll(2) returned something unexpected for listen FD #%i.",
2789             pollfds[i].fd);
2790         continue;
2791       }
2792
2793       client_sock = (listen_socket_t *) malloc (sizeof (listen_socket_t));
2794       if (client_sock == NULL)
2795       {
2796         RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
2797         continue;
2798       }
2799       memcpy(client_sock, &listen_fds[i], sizeof(listen_fds[0]));
2800
2801       client_sa_size = sizeof (client_sa);
2802       client_sock->fd = accept (pollfds[i].fd,
2803           (struct sockaddr *) &client_sa, &client_sa_size);
2804       if (client_sock->fd < 0)
2805       {
2806         RRDD_LOG (LOG_ERR, "listen_thread_main: accept(2) failed.");
2807         free(client_sock);
2808         continue;
2809       }
2810
2811       pthread_attr_init (&attr);
2812       pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
2813
2814       status = pthread_create (&tid, &attr, connection_thread_main,
2815                                client_sock);
2816       if (status != 0)
2817       {
2818         RRDD_LOG (LOG_ERR, "listen_thread_main: pthread_create failed.");
2819         close_connection(client_sock);
2820         continue;
2821       }
2822     } /* for (pollfds_num) */
2823   } /* while (state == RUNNING) */
2824
2825   RRDD_LOG(LOG_INFO, "starting shutdown");
2826
2827   close_listen_sockets ();
2828
2829   pthread_mutex_lock (&connection_threads_lock);
2830   while (connection_threads_num > 0)
2831     pthread_cond_wait(&connection_threads_done, &connection_threads_lock);
2832   pthread_mutex_unlock (&connection_threads_lock);
2833
2834   free(pollfds);
2835
2836   return (NULL);
2837 } /* }}} void *listen_thread_main */
2838
2839 static int daemonize (void) /* {{{ */
2840 {
2841   int pid_fd;
2842   char *base_dir;
2843
2844   daemon_uid = geteuid();
2845
2846   pid_fd = open_pidfile("create", O_CREAT|O_EXCL|O_WRONLY);
2847   if (pid_fd < 0)
2848     pid_fd = check_pidfile();
2849   if (pid_fd < 0)
2850     return pid_fd;
2851
2852   /* open all the listen sockets */
2853   if (config_listen_address_list_len > 0)
2854   {
2855     for (size_t i = 0; i < config_listen_address_list_len; i++)
2856       open_listen_socket (config_listen_address_list[i]);
2857
2858     rrd_free_ptrs((void ***) &config_listen_address_list,
2859                   &config_listen_address_list_len);
2860   }
2861   else
2862   {
2863     listen_socket_t sock;
2864     memset(&sock, 0, sizeof(sock));
2865     strncpy(sock.addr, RRDCACHED_DEFAULT_ADDRESS, sizeof(sock.addr)-1);
2866     open_listen_socket (&sock);
2867   }
2868
2869   if (listen_fds_num < 1)
2870   {
2871     fprintf (stderr, "rrdcached: FATAL: cannot open any listen sockets\n");
2872     goto error;
2873   }
2874
2875   if (!stay_foreground)
2876   {
2877     pid_t child;
2878
2879     child = fork ();
2880     if (child < 0)
2881     {
2882       fprintf (stderr, "daemonize: fork(2) failed.\n");
2883       goto error;
2884     }
2885     else if (child > 0)
2886       exit(0);
2887
2888     /* Become session leader */
2889     setsid ();
2890
2891     /* Open the first three file descriptors to /dev/null */
2892     close (2);
2893     close (1);
2894     close (0);
2895
2896     open ("/dev/null", O_RDWR);
2897     if (dup(0) == -1 || dup(0) == -1){
2898         RRDD_LOG (LOG_ERR, "faild to run dup.\n");
2899     }
2900   } /* if (!stay_foreground) */
2901
2902   /* Change into the /tmp directory. */
2903   base_dir = (config_base_dir != NULL)
2904     ? config_base_dir
2905     : "/tmp";
2906
2907   if (chdir (base_dir) != 0)
2908   {
2909     fprintf (stderr, "daemonize: chdir (%s) failed.\n", base_dir);
2910     goto error;
2911   }
2912
2913   install_signal_handlers();
2914
2915   openlog ("rrdcached", LOG_PID, LOG_DAEMON);
2916   RRDD_LOG(LOG_INFO, "starting up");
2917
2918   cache_tree = g_tree_new_full ((GCompareDataFunc) strcmp, NULL, NULL,
2919                                 (GDestroyNotify) free_cache_item);
2920   if (cache_tree == NULL)
2921   {
2922     RRDD_LOG (LOG_ERR, "daemonize: g_tree_new failed.");
2923     goto error;
2924   }
2925
2926   return write_pidfile (pid_fd);
2927
2928 error:
2929   remove_pidfile();
2930   return -1;
2931 } /* }}} int daemonize */
2932
2933 static int cleanup (void) /* {{{ */
2934 {
2935   pthread_cond_broadcast (&flush_cond);
2936   pthread_join (flush_thread, NULL);
2937
2938   pthread_cond_broadcast (&queue_cond);
2939   for (int i = 0; i < config_queue_threads; i++)
2940     pthread_join (queue_threads[i], NULL);
2941
2942   if (config_flush_at_shutdown)
2943   {
2944     assert(cache_queue_head == NULL);
2945     RRDD_LOG(LOG_INFO, "clean shutdown; all RRDs flushed");
2946   }
2947
2948   free(queue_threads);
2949   free(config_base_dir);
2950
2951   pthread_mutex_lock(&cache_lock);
2952   g_tree_destroy(cache_tree);
2953
2954   pthread_mutex_lock(&journal_lock);
2955   journal_done();
2956
2957   RRDD_LOG(LOG_INFO, "goodbye");
2958   closelog ();
2959
2960   remove_pidfile ();
2961   free(config_pid_file);
2962
2963   return (0);
2964 } /* }}} int cleanup */
2965
2966 static int read_options (int argc, char **argv) /* {{{ */
2967 {
2968   int option;
2969   int status = 0;
2970
2971   char **permissions = NULL;
2972   size_t permissions_len = 0;
2973
2974   gid_t  socket_group = (gid_t)-1;
2975   mode_t socket_permissions = (mode_t)-1;
2976
2977   while ((option = getopt(argc, argv, "gl:s:m:P:f:w:z:t:Bb:p:Fj:a:h?")) != -1)
2978   {
2979     switch (option)
2980     {
2981       case 'g':
2982         stay_foreground=1;
2983         break;
2984
2985       case 'l':
2986       {
2987         listen_socket_t *new;
2988
2989         new = malloc(sizeof(listen_socket_t));
2990         if (new == NULL)
2991         {
2992           fprintf(stderr, "read_options: malloc failed.\n");
2993           return(2);
2994         }
2995         memset(new, 0, sizeof(listen_socket_t));
2996
2997         strncpy(new->addr, optarg, sizeof(new->addr)-1);
2998
2999         /* Add permissions to the socket {{{ */
3000         if (permissions_len != 0)
3001         {
3002           size_t i;
3003           for (i = 0; i < permissions_len; i++)
3004           {
3005             status = socket_permission_add (new, permissions[i]);
3006             if (status != 0)
3007             {
3008               fprintf (stderr, "read_options: Adding permission \"%s\" to "
3009                   "socket failed. Most likely, this permission doesn't "
3010                   "exist. Check your command line.\n", permissions[i]);
3011               status = 4;
3012             }
3013           }
3014         }
3015         else /* if (permissions_len == 0) */
3016         {
3017           /* Add permission for ALL commands to the socket. */
3018           size_t i;
3019           for (i = 0; i < list_of_commands_len; i++)
3020           {
3021             status = socket_permission_add (new, list_of_commands[i].cmd);
3022             if (status != 0)
3023             {
3024               fprintf (stderr, "read_options: Adding permission \"%s\" to "
3025                   "socket failed. This should never happen, ever! Sorry.\n",
3026                   permissions[i]);
3027               status = 4;
3028             }
3029           }
3030         }
3031         /* }}} Done adding permissions. */
3032
3033         new->socket_group = socket_group;
3034         new->socket_permissions = socket_permissions;
3035
3036         if (!rrd_add_ptr((void ***)&config_listen_address_list,
3037                          &config_listen_address_list_len, new))
3038         {
3039           fprintf(stderr, "read_options: rrd_add_ptr failed.\n");
3040           return (2);
3041         }
3042       }
3043       break;
3044
3045       /* set socket group permissions */
3046       case 's':
3047       {
3048         gid_t group_gid;
3049         struct group *grp;
3050
3051         group_gid = strtoul(optarg, NULL, 10);
3052         if (errno != EINVAL && group_gid>0)
3053         {
3054           /* we were passed a number */
3055           grp = getgrgid(group_gid);
3056         }
3057         else
3058         {
3059           grp = getgrnam(optarg);
3060         }
3061
3062         if (grp)
3063         {
3064           socket_group = grp->gr_gid;
3065         }
3066         else
3067         {
3068           /* no idea what the user wanted... */
3069           fprintf (stderr, "read_options: couldn't map \"%s\" to a group, Sorry\n", optarg);
3070           return (5);
3071         }
3072       }
3073       break;
3074
3075       /* set socket file permissions */
3076       case 'm':
3077       {
3078         long  tmp;
3079         char *endptr = NULL;
3080
3081         tmp = strtol (optarg, &endptr, 8);
3082         if ((endptr == optarg) || (! endptr) || (*endptr != '\0')
3083             || (tmp > 07777) || (tmp < 0)) {
3084           fprintf (stderr, "read_options: Invalid file mode \"%s\".\n",
3085               optarg);
3086           return (5);
3087         }
3088
3089         socket_permissions = (mode_t)tmp;
3090       }
3091       break;
3092
3093       case 'P':
3094       {
3095         char *optcopy;
3096         char *saveptr;
3097         char *dummy;
3098         char *ptr;
3099
3100         rrd_free_ptrs ((void *) &permissions, &permissions_len);
3101
3102         optcopy = strdup (optarg);
3103         dummy = optcopy;
3104         saveptr = NULL;
3105         while ((ptr = strtok_r (dummy, ", ", &saveptr)) != NULL)
3106         {
3107           dummy = NULL;
3108           rrd_add_strdup ((void *) &permissions, &permissions_len, ptr);
3109         }
3110
3111         free (optcopy);
3112       }
3113       break;
3114
3115       case 'f':
3116       {
3117         int temp;
3118
3119         temp = atoi (optarg);
3120         if (temp > 0)
3121           config_flush_interval = temp;
3122         else
3123         {
3124           fprintf (stderr, "Invalid flush interval: %s\n", optarg);
3125           status = 3;
3126         }
3127       }
3128       break;
3129
3130       case 'w':
3131       {
3132         int temp;
3133
3134         temp = atoi (optarg);
3135         if (temp > 0)
3136           config_write_interval = temp;
3137         else
3138         {
3139           fprintf (stderr, "Invalid write interval: %s\n", optarg);
3140           status = 2;
3141         }
3142       }
3143       break;
3144
3145       case 'z':
3146       {
3147         int temp;
3148
3149         temp = atoi(optarg);
3150         if (temp > 0)
3151           config_write_jitter = temp;
3152         else
3153         {
3154           fprintf (stderr, "Invalid write jitter: -z %s\n", optarg);
3155           status = 2;
3156         }
3157
3158         break;
3159       }
3160
3161       case 't':
3162       {
3163         int threads;
3164         threads = atoi(optarg);
3165         if (threads >= 1)
3166           config_queue_threads = threads;
3167         else
3168         {
3169           fprintf (stderr, "Invalid thread count: -t %s\n", optarg);
3170           return 1;
3171         }
3172       }
3173       break;
3174
3175       case 'B':
3176         config_write_base_only = 1;
3177         break;
3178
3179       case 'b':
3180       {
3181         size_t len;
3182         char base_realpath[PATH_MAX];
3183
3184         if (config_base_dir != NULL)
3185           free (config_base_dir);
3186         config_base_dir = strdup (optarg);
3187         if (config_base_dir == NULL)
3188         {
3189           fprintf (stderr, "read_options: strdup failed.\n");
3190           return (3);
3191         }
3192
3193         if (rrd_mkdir_p (config_base_dir, 0777) != 0)
3194         {
3195           fprintf (stderr, "Failed to create base directory '%s': %s\n",
3196               config_base_dir, rrd_strerror (errno));
3197           return (3);
3198         }
3199
3200         /* make sure that the base directory is not resolved via
3201          * symbolic links.  this makes some performance-enhancing
3202          * assumptions possible (we don't have to resolve paths
3203          * that start with a "/")
3204          */
3205         if (realpath(config_base_dir, base_realpath) == NULL)
3206         {
3207           fprintf (stderr, "Failed to canonicalize the base directory '%s': "
3208               "%s\n", config_base_dir, rrd_strerror(errno));
3209           return 5;
3210         }
3211
3212         len = strlen (config_base_dir);
3213         while ((len > 0) && (config_base_dir[len - 1] == '/'))
3214         {
3215           config_base_dir[len - 1] = 0;
3216           len--;
3217         }
3218
3219         if (len < 1)
3220         {
3221           fprintf (stderr, "Invalid base directory: %s\n", optarg);
3222           return (4);
3223         }
3224
3225         _config_base_dir_len = len;
3226
3227         len = strlen (base_realpath);
3228         while ((len > 0) && (base_realpath[len - 1] == '/'))
3229         {
3230           base_realpath[len - 1] = '\0';
3231           len--;
3232         }
3233
3234         if (strncmp(config_base_dir,
3235                          base_realpath, sizeof(base_realpath)) != 0)
3236         {
3237           fprintf(stderr,
3238                   "Base directory (-b) resolved via file system links!\n"
3239                   "Please consult rrdcached '-b' documentation!\n"
3240                   "Consider specifying the real directory (%s)\n",
3241                   base_realpath);
3242           return 5;
3243         }
3244       }
3245       break;
3246
3247       case 'p':
3248       {
3249         if (config_pid_file != NULL)
3250           free (config_pid_file);
3251         config_pid_file = strdup (optarg);
3252         if (config_pid_file == NULL)
3253         {
3254           fprintf (stderr, "read_options: strdup failed.\n");
3255           return (3);
3256         }
3257       }
3258       break;
3259
3260       case 'F':
3261         config_flush_at_shutdown = 1;
3262         break;
3263
3264       case 'j':
3265       {
3266         char journal_dir_actual[PATH_MAX];
3267         const char *dir;
3268         dir = journal_dir = strdup(realpath((const char *)optarg, journal_dir_actual));
3269
3270         status = rrd_mkdir_p(dir, 0777);
3271         if (status != 0)
3272         {
3273           fprintf(stderr, "Failed to create journal directory '%s': %s\n",
3274               dir, rrd_strerror(errno));
3275           return 6;
3276         }
3277
3278         if (access(dir, R_OK|W_OK|X_OK) != 0)
3279         {
3280           fprintf(stderr, "Must specify a writable directory with -j! (%s)\n",
3281                   errno ? rrd_strerror(errno) : "");
3282           return 6;
3283         }
3284       }
3285       break;
3286
3287       case 'a':
3288       {
3289         int temp = atoi(optarg);
3290         if (temp > 0)
3291           config_alloc_chunk = temp;
3292         else
3293         {
3294           fprintf(stderr, "Invalid allocation size: %s\n", optarg);
3295           return 10;
3296         }
3297       }
3298       break;
3299
3300       case 'h':
3301       case '?':
3302         printf ("RRDCacheD %s\n"
3303             "Copyright (C) 2008,2009 Florian octo Forster and Kevin Brintnall\n"
3304             "\n"
3305             "Usage: rrdcached [options]\n"
3306             "\n"
3307             "Valid options are:\n"
3308             "  -l <address>  Socket address to listen to.\n"
3309             "  -P <perms>    Sets the permissions to assign to all following "
3310                             "sockets\n"
3311             "  -w <seconds>  Interval in which to write data.\n"
3312             "  -z <delay>    Delay writes up to <delay> seconds to spread load\n"
3313             "  -t <threads>  Number of write threads.\n"
3314             "  -f <seconds>  Interval in which to flush dead data.\n"
3315             "  -p <file>     Location of the PID-file.\n"
3316             "  -b <dir>      Base directory to change to.\n"
3317             "  -B            Restrict file access to paths within -b <dir>\n"
3318             "  -g            Do not fork and run in the foreground.\n"
3319             "  -j <dir>      Directory in which to create the journal files.\n"
3320             "  -F            Always flush all updates at shutdown\n"
3321             "  -s <id|name>  Group owner of all following UNIX sockets\n"
3322             "                (the socket will also have read/write permissions "
3323                             "for that group)\n"
3324             "  -m <mode>     File permissions (octal) of all following UNIX "
3325                             "sockets\n"
3326             "  -a <size>     Memory allocation chunk size. Default is 1."
3327             "\n"
3328             "For more information and a detailed description of all options "
3329             "please refer\n"
3330             "to the rrdcached(1) manual page.\n",
3331             VERSION);
3332         if (option == 'h')
3333           status = -1;
3334         else
3335           status = 1;
3336         break;
3337     } /* switch (option) */
3338   } /* while (getopt) */
3339
3340   /* advise the user when values are not sane */
3341   if (config_flush_interval < 2 * config_write_interval)
3342     fprintf(stderr, "WARNING: flush interval (-f) should be at least"
3343             " 2x write interval (-w) !\n");
3344   if (config_write_jitter > config_write_interval)
3345     fprintf(stderr, "WARNING: write delay (-z) should NOT be larger than"
3346             " write interval (-w) !\n");
3347
3348   if (config_write_base_only && config_base_dir == NULL)
3349     fprintf(stderr, "WARNING: -B does not make sense without -b!\n"
3350             "  Consult the rrdcached documentation\n");
3351
3352   if (journal_dir == NULL)
3353     config_flush_at_shutdown = 1;
3354
3355   rrd_free_ptrs ((void *) &permissions, &permissions_len);
3356
3357   return (status);
3358 } /* }}} int read_options */
3359
3360 int main (int argc, char **argv)
3361 {
3362   int status;
3363
3364   status = read_options (argc, argv);
3365   if (status != 0)
3366   {
3367     if (status < 0)
3368       status = 0;
3369     return (status);
3370   }
3371
3372   status = daemonize ();
3373   if (status != 0)
3374   {
3375     fprintf (stderr, "rrdcached: daemonize failed, exiting.\n");
3376     return (1);
3377   }
3378
3379   journal_init();
3380
3381   /* start the queue threads */
3382   queue_threads = calloc(config_queue_threads, sizeof(*queue_threads));
3383   if (queue_threads == NULL)
3384   {
3385     RRDD_LOG (LOG_ERR, "FATAL: cannot calloc queue threads");
3386     cleanup();
3387     return (1);
3388   }
3389   for (int i = 0; i < config_queue_threads; i++)
3390   {
3391     memset (&queue_threads[i], 0, sizeof (*queue_threads));
3392     status = pthread_create (&queue_threads[i], NULL, queue_thread_main, NULL);
3393     if (status != 0)
3394     {
3395       RRDD_LOG (LOG_ERR, "FATAL: cannot create queue thread");
3396       cleanup();
3397       return (1);
3398     }
3399   }
3400
3401   /* start the flush thread */
3402   memset(&flush_thread, 0, sizeof(flush_thread));
3403   status = pthread_create (&flush_thread, NULL, flush_thread_main, NULL);
3404   if (status != 0)
3405   {
3406     RRDD_LOG (LOG_ERR, "FATAL: cannot create flush thread");
3407     cleanup();
3408     return (1);
3409   }
3410
3411   listen_thread_main (NULL);
3412   cleanup ();
3413
3414   return (0);
3415 } /* int main */
3416
3417 /*
3418  * vim: set sw=2 sts=2 ts=8 et fdm=marker :
3419  */