2 * RRDTool - src/rrd_daemon.c
3 * Copyright (C) 2008-2010 Florian octo Forster
4 * Copyright (C) 2008,2009 Kevin Brintnall
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.
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.
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
20 * Florian octo Forster <octo at verplant.org>
21 * kevin brintnall <kbrint@rufus.net>
26 * First tell the compiler to stick to the C99 and POSIX standards as close as
29 #ifndef __STRICT_ANSI__ /* {{{ */
30 # define __STRICT_ANSI__
33 #ifndef _ISOC99_SOURCE
34 # define _ISOC99_SOURCE
37 #ifdef _POSIX_C_SOURCE
38 # undef _POSIX_C_SOURCE
40 #define _POSIX_C_SOURCE 200112L
42 /* Single UNIX needed for strdup. */
46 #define _XOPEN_SOURCE 500
63 * Now for some includes..
66 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__) && !defined(HAVE_CONFIG_H)
67 #include "../win32/config.h"
70 #include "../rrd_config.h"
75 #include "rrd_client.h"
87 #include <sys/socket.h>
95 #include <sys/types.h>
107 #include <sys/time.h>
114 #endif /* HAVE_LIBWRAP */
116 #include <glib-2.0/glib.h>
119 #define RRDD_LOG(severity, ...) \
121 if (stay_foreground) { \
122 fprintf(stderr, __VA_ARGS__); \
123 fprintf(stderr, "\n"); } \
124 syslog ((severity), __VA_ARGS__); \
130 typedef enum { RESP_ERR = -1, RESP_OK = 0 } response_code;
132 struct listen_socket_s
135 char addr[PATH_MAX + 1];
138 /* state for BATCH processing */
150 uint32_t permissions;
153 mode_t socket_permissions;
155 typedef struct listen_socket_s listen_socket_t;
158 typedef struct command_s command_t;
159 /* note: guard against "unused" warnings in the handlers */
160 #define DISPATCH_PROTO listen_socket_t UNUSED(*sock),\
162 char UNUSED(*buffer),\
163 size_t UNUSED(buffer_size)
165 #define HANDLER_PROTO command_t UNUSED(*cmd),\
170 int (*handler)(HANDLER_PROTO);
172 char context; /* where we expect to see it */
173 #define CMD_CONTEXT_CLIENT (1<<0)
174 #define CMD_CONTEXT_BATCH (1<<1)
175 #define CMD_CONTEXT_JOURNAL (1<<2)
176 #define CMD_CONTEXT_ANY (0x7f)
183 typedef struct cache_item_s cache_item_t;
188 size_t values_num; /* number of valid pointers */
189 size_t values_alloc; /* number of allocated pointers */
190 time_t last_flush_time;
191 time_t last_update_stamp;
192 #define CI_FLAGS_IN_TREE (1<<0)
193 #define CI_FLAGS_IN_QUEUE (1<<1)
195 pthread_cond_t flushed;
200 struct callback_flush_data_s
207 typedef struct callback_flush_data_s callback_flush_data_t;
214 typedef enum queue_side_e queue_side_t;
216 /* describe a set of journal files */
222 #define RBUF_SIZE (RRD_CMD_MAX*2)
227 static int stay_foreground = 0;
228 static uid_t daemon_uid;
230 static listen_socket_t *listen_fds = NULL;
231 static size_t listen_fds_num = 0;
233 static listen_socket_t default_socket;
236 RUNNING, /* normal operation */
237 FLUSHING, /* flushing remaining values */
238 SHUTDOWN /* shutting down */
241 static pthread_t *queue_threads;
242 static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER;
243 static int config_queue_threads = 4;
245 static pthread_t flush_thread;
246 static pthread_cond_t flush_cond = PTHREAD_COND_INITIALIZER;
248 static pthread_mutex_t connection_threads_lock = PTHREAD_MUTEX_INITIALIZER;
249 static pthread_cond_t connection_threads_done = PTHREAD_COND_INITIALIZER;
250 static int connection_threads_num = 0;
253 static GTree *cache_tree = NULL;
254 static cache_item_t *cache_queue_head = NULL;
255 static cache_item_t *cache_queue_tail = NULL;
256 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
258 static int config_write_interval = 300;
259 static int config_write_jitter = 0;
260 static int config_flush_interval = 3600;
261 static int config_flush_at_shutdown = 0;
262 static char *config_pid_file = NULL;
263 static char *config_base_dir = NULL;
264 static size_t _config_base_dir_len = 0;
265 static int config_write_base_only = 0;
266 static size_t config_alloc_chunk = 1;
268 static listen_socket_t **config_listen_address_list = NULL;
269 static size_t config_listen_address_list_len = 0;
271 static uint64_t stats_queue_length = 0;
272 static uint64_t stats_updates_received = 0;
273 static uint64_t stats_flush_received = 0;
274 static uint64_t stats_updates_written = 0;
275 static uint64_t stats_data_sets_written = 0;
276 static uint64_t stats_journal_bytes = 0;
277 static uint64_t stats_journal_rotate = 0;
278 static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER;
280 static int opt_no_overwrite = 0; /* default for the daemon */
282 /* Journaled updates */
283 #define JOURNAL_REPLAY(s) ((s) == NULL)
284 #define JOURNAL_BASE "rrd.journal"
285 static journal_set *journal_cur = NULL;
286 static journal_set *journal_old = NULL;
287 static char *journal_dir = NULL;
288 static FILE *journal_fh = NULL; /* current journal file handle */
289 static long journal_size = 0; /* current journal size */
290 #define JOURNAL_MAX (1 * 1024 * 1024 * 1024)
291 static pthread_mutex_t journal_lock = PTHREAD_MUTEX_INITIALIZER;
292 static int journal_write(char *cmd, char *args);
293 static void journal_done(void);
294 static void journal_rotate(void);
296 /* prototypes for forward refernces */
297 static int handle_request_help (HANDLER_PROTO);
302 static void sig_common (const char *sig) /* {{{ */
304 RRDD_LOG(LOG_NOTICE, "caught SIG%s", sig);
306 pthread_cond_broadcast(&flush_cond);
307 pthread_cond_broadcast(&queue_cond);
308 } /* }}} void sig_common */
310 static void sig_int_handler (int UNUSED(s)) /* {{{ */
313 } /* }}} void sig_int_handler */
315 static void sig_term_handler (int UNUSED(s)) /* {{{ */
318 } /* }}} void sig_term_handler */
320 static void sig_usr1_handler (int UNUSED(s)) /* {{{ */
322 config_flush_at_shutdown = 1;
324 } /* }}} void sig_usr1_handler */
326 static void sig_usr2_handler (int UNUSED(s)) /* {{{ */
328 config_flush_at_shutdown = 0;
330 } /* }}} void sig_usr2_handler */
332 static void install_signal_handlers(void) /* {{{ */
334 /* These structures are static, because `sigaction' behaves weird if the are
336 static struct sigaction sa_int;
337 static struct sigaction sa_term;
338 static struct sigaction sa_pipe;
339 static struct sigaction sa_usr1;
340 static struct sigaction sa_usr2;
342 /* Install signal handlers */
343 memset (&sa_int, 0, sizeof (sa_int));
344 sa_int.sa_handler = sig_int_handler;
345 sigaction (SIGINT, &sa_int, NULL);
347 memset (&sa_term, 0, sizeof (sa_term));
348 sa_term.sa_handler = sig_term_handler;
349 sigaction (SIGTERM, &sa_term, NULL);
351 memset (&sa_pipe, 0, sizeof (sa_pipe));
352 sa_pipe.sa_handler = SIG_IGN;
353 sigaction (SIGPIPE, &sa_pipe, NULL);
355 memset (&sa_pipe, 0, sizeof (sa_usr1));
356 sa_usr1.sa_handler = sig_usr1_handler;
357 sigaction (SIGUSR1, &sa_usr1, NULL);
359 memset (&sa_usr2, 0, sizeof (sa_usr2));
360 sa_usr2.sa_handler = sig_usr2_handler;
361 sigaction (SIGUSR2, &sa_usr2, NULL);
363 } /* }}} void install_signal_handlers */
365 static int open_pidfile(char *action, int oflag) /* {{{ */
369 char *file_copy, *dir;
371 file = (config_pid_file != NULL)
373 : LOCALSTATEDIR "/run/rrdcached.pid";
375 /* dirname may modify its argument */
376 file_copy = strdup(file);
377 if (file_copy == NULL)
379 fprintf(stderr, "rrdcached: strdup(): %s\n",
380 rrd_strerror(errno));
384 dir = dirname(file_copy);
385 if (rrd_mkdir_p(dir, 0777) != 0)
387 fprintf(stderr, "Failed to create pidfile directory '%s': %s\n",
388 dir, rrd_strerror(errno));
394 fd = open(file, oflag, S_IWUSR|S_IRUSR|S_IRGRP|S_IROTH);
396 fprintf(stderr, "rrdcached: can't %s pid file '%s' (%s)\n",
397 action, file, rrd_strerror(errno));
400 } /* }}} static int open_pidfile */
402 /* check existing pid file to see whether a daemon is running */
403 static int check_pidfile(void)
409 pid_fd = open_pidfile("open", O_RDWR);
413 if (read(pid_fd, pid_str, sizeof(pid_str)) <= 0)
420 /* another running process that we can signal COULD be
421 * a competing rrdcached */
422 if (pid != getpid() && kill(pid, 0) == 0)
425 "FATAL: Another rrdcached daemon is running?? (pid %d)\n", pid);
430 lseek(pid_fd, 0, SEEK_SET);
431 if (ftruncate(pid_fd, 0) == -1)
434 "FATAL: Faild to truncate stale PID file. (pid %d)\n", pid);
440 "rrdcached: removed stale PID file (no rrdcached on pid %d)\n"
441 "rrdcached: starting normally.\n", pid);
444 } /* }}} static int check_pidfile */
446 static int write_pidfile (int fd) /* {{{ */
453 fh = fdopen (fd, "w");
456 RRDD_LOG (LOG_ERR, "write_pidfile: fdopen() failed.");
461 fprintf (fh, "%i\n", (int) pid);
465 } /* }}} int write_pidfile */
467 static int remove_pidfile (void) /* {{{ */
472 file = (config_pid_file != NULL)
474 : LOCALSTATEDIR "/run/rrdcached.pid";
476 status = unlink (file);
480 } /* }}} int remove_pidfile */
482 static char *next_cmd (listen_socket_t *sock, ssize_t *len) /* {{{ */
486 eol = memchr(sock->rbuf + sock->next_cmd, '\n',
487 sock->next_read - sock->next_cmd);
491 /* no commands left, move remainder back to front of rbuf */
492 memmove(sock->rbuf, sock->rbuf + sock->next_cmd,
493 sock->next_read - sock->next_cmd);
494 sock->next_read -= sock->next_cmd;
501 char *cmd = sock->rbuf + sock->next_cmd;
504 sock->next_cmd = eol - sock->rbuf + 1;
506 if (eol > sock->rbuf && *(eol-1) == '\r')
507 *(--eol) = '\0'; /* handle "\r\n" EOL */
516 } /* }}} char *next_cmd */
518 /* add the characters directly to the write buffer */
519 static int add_to_wbuf(listen_socket_t *sock, char *str, size_t len) /* {{{ */
523 assert(sock != NULL);
525 new_buf = rrd_realloc(sock->wbuf, sock->wbuf_len + len + 1);
528 RRDD_LOG(LOG_ERR, "add_to_wbuf: realloc failed");
532 strncpy(new_buf + sock->wbuf_len, str, len + 1);
534 sock->wbuf = new_buf;
535 sock->wbuf_len += len;
538 } /* }}} static int add_to_wbuf */
540 /* add the text to the "extra" info that's sent after the status line */
541 static int add_response_info(listen_socket_t *sock, char *fmt, ...) /* {{{ */
544 char buffer[RRD_CMD_MAX];
547 if (JOURNAL_REPLAY(sock)) return 0;
548 if (sock->batch_start) return 0; /* no extra info returned when in BATCH */
551 #ifdef HAVE_VSNPRINTF
552 len = vsnprintf(buffer, sizeof(buffer), fmt, argp);
554 len = vsprintf(buffer, fmt, argp);
559 RRDD_LOG(LOG_ERR, "add_response_info: vnsprintf failed");
563 return add_to_wbuf(sock, buffer, len);
564 } /* }}} static int add_response_info */
566 static int count_lines(char *str) /* {{{ */
572 while ((str = strchr(str, '\n')) != NULL)
580 } /* }}} static int count_lines */
582 /* send the response back to the user.
583 * returns 0 on success, -1 on error
584 * write buffer is always zeroed after this call */
585 static int send_response (listen_socket_t *sock, response_code rc,
586 char *fmt, ...) /* {{{ */
589 char buffer[RRD_CMD_MAX];
594 if (JOURNAL_REPLAY(sock)) return rc;
596 if (sock->batch_start)
599 return rc; /* no response on success during BATCH */
600 lines = sock->batch_cmd;
602 else if (rc == RESP_OK)
603 lines = count_lines(sock->wbuf);
607 rclen = sprintf(buffer, "%d ", lines);
609 #ifdef HAVE_VSNPRINTF
610 len = vsnprintf(buffer+rclen, sizeof(buffer)-rclen, fmt, argp);
612 len = vsprintf(buffer+rclen, fmt, argp);
620 /* append the result to the wbuf, don't write to the user */
621 if (sock->batch_start)
622 return add_to_wbuf(sock, buffer, len);
624 /* first write must be complete */
625 if (len != write(sock->fd, buffer, len))
627 RRDD_LOG(LOG_INFO, "send_response: could not write status message");
631 if (sock->wbuf != NULL && rc == RESP_OK)
634 while (wrote < sock->wbuf_len)
636 ssize_t wb = write(sock->fd, sock->wbuf + wrote, sock->wbuf_len - wrote);
639 RRDD_LOG(LOG_INFO, "send_response: could not write results");
646 free(sock->wbuf); sock->wbuf = NULL;
652 static void wipe_ci_values(cache_item_t *ci, time_t when)
656 ci->values_alloc = 0;
658 ci->last_flush_time = when;
659 if (config_write_jitter > 0)
660 ci->last_flush_time += (rrd_random() % config_write_jitter);
664 * remove a "cache_item_t" item from the queue.
665 * must hold 'cache_lock' when calling this
667 static void remove_from_queue(cache_item_t *ci) /* {{{ */
669 if (ci == NULL) return;
670 if ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) return; /* not queued */
672 if (ci->prev == NULL)
673 cache_queue_head = ci->next; /* reset head */
675 ci->prev->next = ci->next;
677 if (ci->next == NULL)
678 cache_queue_tail = ci->prev; /* reset the tail */
680 ci->next->prev = ci->prev;
682 ci->next = ci->prev = NULL;
683 ci->flags &= ~CI_FLAGS_IN_QUEUE;
685 pthread_mutex_lock (&stats_lock);
686 assert (stats_queue_length > 0);
687 stats_queue_length--;
688 pthread_mutex_unlock (&stats_lock);
690 } /* }}} static void remove_from_queue */
692 /* free the resources associated with the cache_item_t
693 * must hold cache_lock when calling this function
695 static void *free_cache_item(cache_item_t *ci) /* {{{ */
697 if (ci == NULL) return NULL;
699 remove_from_queue(ci);
701 for (size_t i=0; i < ci->values_num; i++)
707 /* in case anyone is waiting */
708 pthread_cond_broadcast(&ci->flushed);
709 pthread_cond_destroy(&ci->flushed);
714 } /* }}} static void *free_cache_item */
717 * enqueue_cache_item:
718 * `cache_lock' must be acquired before calling this function!
720 static int enqueue_cache_item (cache_item_t *ci, /* {{{ */
726 if (ci->values_num == 0)
731 if (cache_queue_head == ci)
734 /* remove if further down in queue */
735 remove_from_queue(ci);
738 ci->next = cache_queue_head;
739 if (ci->next != NULL)
741 cache_queue_head = ci;
743 if (cache_queue_tail == NULL)
744 cache_queue_tail = cache_queue_head;
746 else /* (side == TAIL) */
748 /* We don't move values back in the list.. */
749 if (ci->flags & CI_FLAGS_IN_QUEUE)
752 assert (ci->next == NULL);
753 assert (ci->prev == NULL);
755 ci->prev = cache_queue_tail;
757 if (cache_queue_tail == NULL)
758 cache_queue_head = ci;
760 cache_queue_tail->next = ci;
762 cache_queue_tail = ci;
765 ci->flags |= CI_FLAGS_IN_QUEUE;
767 pthread_cond_signal(&queue_cond);
768 pthread_mutex_lock (&stats_lock);
769 stats_queue_length++;
770 pthread_mutex_unlock (&stats_lock);
773 } /* }}} int enqueue_cache_item */
776 * tree_callback_flush:
777 * Called via `g_tree_foreach' in `flush_thread_main'. `cache_lock' is held
778 * while this is in progress.
780 static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */
784 callback_flush_data_t *cfd;
786 ci = (cache_item_t *) value;
787 cfd = (callback_flush_data_t *) data;
789 if (ci->flags & CI_FLAGS_IN_QUEUE)
792 if (ci->values_num > 0
793 && (ci->last_flush_time <= cfd->abs_timeout || state != RUNNING))
795 enqueue_cache_item (ci, TAIL);
797 else if (((cfd->now - ci->last_flush_time) >= config_flush_interval)
798 && (ci->values_num <= 0))
800 assert ((char *) key == ci->file);
801 if (!rrd_add_ptr((void ***)&cfd->keys, &cfd->keys_num, (void *)key))
803 RRDD_LOG (LOG_ERR, "tree_callback_flush: rrd_add_ptrs failed.");
809 } /* }}} gboolean tree_callback_flush */
811 static int flush_old_values (int max_age)
813 callback_flush_data_t cfd;
816 memset (&cfd, 0, sizeof (cfd));
817 /* Pass the current time as user data so that we don't need to call
818 * `time' for each node. */
819 cfd.now = time (NULL);
824 cfd.abs_timeout = cfd.now - max_age;
826 cfd.abs_timeout = cfd.now + 2*config_write_jitter + 1;
828 /* `tree_callback_flush' will return the keys of all values that haven't
829 * been touched in the last `config_flush_interval' seconds in `cfd'.
830 * The char*'s in this array point to the same memory as ci->file, so we
831 * don't need to free them separately. */
832 g_tree_foreach (cache_tree, tree_callback_flush, (gpointer) &cfd);
834 for (k = 0; k < cfd.keys_num; k++)
836 gboolean status = g_tree_remove(cache_tree, cfd.keys[k]);
837 /* should never fail, since we have held the cache_lock
839 assert(status == TRUE);
842 if (cfd.keys != NULL)
849 } /* int flush_old_values */
851 static void *flush_thread_main (void UNUSED(*args)) /* {{{ */
854 struct timespec next_flush;
857 gettimeofday (&now, NULL);
858 next_flush.tv_sec = now.tv_sec + config_flush_interval;
859 next_flush.tv_nsec = 1000 * now.tv_usec;
861 pthread_mutex_lock(&cache_lock);
863 while (state == RUNNING)
865 gettimeofday (&now, NULL);
866 if ((now.tv_sec > next_flush.tv_sec)
867 || ((now.tv_sec == next_flush.tv_sec)
868 && ((1000 * now.tv_usec) > next_flush.tv_nsec)))
870 RRDD_LOG(LOG_DEBUG, "flushing old values");
872 /* Determine the time of the next cache flush. */
873 next_flush.tv_sec = now.tv_sec + config_flush_interval;
875 /* Flush all values that haven't been written in the last
876 * `config_write_interval' seconds. */
877 flush_old_values (config_write_interval);
879 /* unlock the cache while we rotate so we don't block incoming
880 * updates if the fsync() blocks on disk I/O */
881 pthread_mutex_unlock(&cache_lock);
883 pthread_mutex_lock(&cache_lock);
886 status = pthread_cond_timedwait(&flush_cond, &cache_lock, &next_flush);
887 if (status != 0 && status != ETIMEDOUT)
889 RRDD_LOG (LOG_ERR, "flush_thread_main: "
890 "pthread_cond_timedwait returned %i.", status);
894 if (config_flush_at_shutdown)
895 flush_old_values (-1); /* flush everything */
899 pthread_mutex_unlock(&cache_lock);
902 } /* void *flush_thread_main */
904 static void *queue_thread_main (void UNUSED(*args)) /* {{{ */
906 pthread_mutex_lock (&cache_lock);
908 while (state != SHUTDOWN
909 || (cache_queue_head != NULL && config_flush_at_shutdown))
917 /* Now, check if there's something to store away. If not, wait until
918 * something comes in. */
919 if (cache_queue_head == NULL)
921 status = pthread_cond_wait (&queue_cond, &cache_lock);
922 if ((status != 0) && (status != ETIMEDOUT))
924 RRDD_LOG (LOG_ERR, "queue_thread_main: "
925 "pthread_cond_wait returned %i.", status);
929 /* Check if a value has arrived. This may be NULL if we timed out or there
930 * was an interrupt such as a signal. */
931 if (cache_queue_head == NULL)
934 ci = cache_queue_head;
936 /* copy the relevant parts */
937 file = strdup (ci->file);
940 RRDD_LOG (LOG_ERR, "queue_thread_main: strdup failed.");
944 assert(ci->values != NULL);
945 assert(ci->values_num > 0);
948 values_num = ci->values_num;
950 wipe_ci_values(ci, time(NULL));
951 remove_from_queue(ci);
953 pthread_mutex_unlock (&cache_lock);
956 status = rrd_update_r (file, NULL, (int) values_num, (void *) values);
959 RRDD_LOG (LOG_NOTICE, "queue_thread_main: "
960 "rrd_update_r (%s) failed with status %i. (%s)",
961 file, status, rrd_get_error());
964 journal_write("wrote", file);
966 /* Search again in the tree. It's possible someone issued a "FORGET"
967 * while we were writing the update values. */
968 pthread_mutex_lock(&cache_lock);
969 ci = (cache_item_t *) g_tree_lookup(cache_tree, file);
971 pthread_cond_broadcast(&ci->flushed);
972 pthread_mutex_unlock(&cache_lock);
976 pthread_mutex_lock (&stats_lock);
977 stats_updates_written++;
978 stats_data_sets_written += values_num;
979 pthread_mutex_unlock (&stats_lock);
982 rrd_free_ptrs((void ***) &values, &values_num);
985 pthread_mutex_lock (&cache_lock);
987 pthread_mutex_unlock (&cache_lock);
990 } /* }}} void *queue_thread_main */
992 static int buffer_get_field (char **buffer_ret, /* {{{ */
993 size_t *buffer_size_ret, char **field_ret)
1002 buffer = *buffer_ret;
1004 buffer_size = *buffer_size_ret;
1005 field = *buffer_ret;
1008 if (buffer_size <= 0)
1011 /* This is ensured by `handle_request'. */
1012 assert (buffer[buffer_size - 1] == '\0');
1015 while (buffer_pos < buffer_size)
1017 /* Check for end-of-field or end-of-buffer */
1018 if (buffer[buffer_pos] == ' ' || buffer[buffer_pos] == '\0')
1020 field[field_size] = 0;
1026 /* Handle escaped characters. */
1027 else if (buffer[buffer_pos] == '\\')
1029 if (buffer_pos >= (buffer_size - 1))
1032 field[field_size] = buffer[buffer_pos];
1036 /* Normal operation */
1039 field[field_size] = buffer[buffer_pos];
1043 } /* while (buffer_pos < buffer_size) */
1048 *buffer_ret = buffer + buffer_pos;
1049 *buffer_size_ret = buffer_size - buffer_pos;
1053 } /* }}} int buffer_get_field */
1055 /* if we're restricting writes to the base directory,
1056 * check whether the file falls within the dir
1057 * returns 1 if OK, otherwise 0
1059 static int check_file_access (const char *file, listen_socket_t *sock) /* {{{ */
1061 assert(file != NULL);
1063 if (!config_write_base_only
1064 || JOURNAL_REPLAY(sock)
1065 || config_base_dir == NULL)
1068 if (strstr(file, "../") != NULL) goto err;
1070 /* relative paths without "../" are ok */
1071 if (*file != '/') return 1;
1073 /* file must be of the format base + "/" + <1+ char filename> */
1074 if (strlen(file) < _config_base_dir_len + 2) goto err;
1075 if (strncmp(file, config_base_dir, _config_base_dir_len) != 0) goto err;
1076 if (*(file + _config_base_dir_len) != '/') goto err;
1081 if (sock != NULL && sock->fd >= 0)
1082 send_response(sock, RESP_ERR, "%s\n", rrd_strerror(EACCES));
1085 } /* }}} static int check_file_access */
1087 /* when using a base dir, convert relative paths to absolute paths.
1088 * if necessary, modifies the "filename" pointer to point
1089 * to the new path created in "tmp". "tmp" is provided
1090 * by the caller and sizeof(tmp) must be >= PATH_MAX.
1092 * this allows us to optimize for the expected case (absolute path)
1095 static void get_abs_path(char **filename, char *tmp)
1097 assert(tmp != NULL);
1098 assert(filename != NULL && *filename != NULL);
1100 if (config_base_dir == NULL || **filename == '/')
1103 snprintf(tmp, PATH_MAX, "%s/%s", config_base_dir, *filename);
1105 } /* }}} static int get_abs_path */
1107 static int flush_file (const char *filename) /* {{{ */
1111 pthread_mutex_lock (&cache_lock);
1113 ci = (cache_item_t *) g_tree_lookup (cache_tree, filename);
1116 pthread_mutex_unlock (&cache_lock);
1120 if (ci->values_num > 0)
1122 /* Enqueue at head */
1123 enqueue_cache_item (ci, HEAD);
1124 pthread_cond_wait(&ci->flushed, &cache_lock);
1127 /* DO NOT DO ANYTHING WITH ci HERE!! The entry
1128 * may have been purged during our cond_wait() */
1130 pthread_mutex_unlock(&cache_lock);
1133 } /* }}} int flush_file */
1135 static int syntax_error(listen_socket_t *sock, command_t *cmd) /* {{{ */
1137 char *err = "Syntax error.\n";
1139 if (cmd && cmd->syntax)
1142 return send_response(sock, RESP_ERR, "Usage: %s", err);
1143 } /* }}} static int syntax_error() */
1145 static int handle_request_stats (HANDLER_PROTO) /* {{{ */
1147 uint64_t copy_queue_length;
1148 uint64_t copy_updates_received;
1149 uint64_t copy_flush_received;
1150 uint64_t copy_updates_written;
1151 uint64_t copy_data_sets_written;
1152 uint64_t copy_journal_bytes;
1153 uint64_t copy_journal_rotate;
1155 uint64_t tree_nodes_number;
1156 uint64_t tree_depth;
1158 pthread_mutex_lock (&stats_lock);
1159 copy_queue_length = stats_queue_length;
1160 copy_updates_received = stats_updates_received;
1161 copy_flush_received = stats_flush_received;
1162 copy_updates_written = stats_updates_written;
1163 copy_data_sets_written = stats_data_sets_written;
1164 copy_journal_bytes = stats_journal_bytes;
1165 copy_journal_rotate = stats_journal_rotate;
1166 pthread_mutex_unlock (&stats_lock);
1168 pthread_mutex_lock (&cache_lock);
1169 tree_nodes_number = (uint64_t) g_tree_nnodes (cache_tree);
1170 tree_depth = (uint64_t) g_tree_height (cache_tree);
1171 pthread_mutex_unlock (&cache_lock);
1173 add_response_info(sock,
1174 "QueueLength: %"PRIu64"\n", copy_queue_length);
1175 add_response_info(sock,
1176 "UpdatesReceived: %"PRIu64"\n", copy_updates_received);
1177 add_response_info(sock,
1178 "FlushesReceived: %"PRIu64"\n", copy_flush_received);
1179 add_response_info(sock,
1180 "UpdatesWritten: %"PRIu64"\n", copy_updates_written);
1181 add_response_info(sock,
1182 "DataSetsWritten: %"PRIu64"\n", copy_data_sets_written);
1183 add_response_info(sock, "TreeNodesNumber: %"PRIu64"\n", tree_nodes_number);
1184 add_response_info(sock, "TreeDepth: %"PRIu64"\n", tree_depth);
1185 add_response_info(sock, "JournalBytes: %"PRIu64"\n", copy_journal_bytes);
1186 add_response_info(sock, "JournalRotate: %"PRIu64"\n", copy_journal_rotate);
1188 send_response(sock, RESP_OK, "Statistics follow\n");
1191 } /* }}} int handle_request_stats */
1193 static int handle_request_flush (HANDLER_PROTO) /* {{{ */
1195 char *file, file_tmp[PATH_MAX];
1198 status = buffer_get_field (&buffer, &buffer_size, &file);
1201 return syntax_error(sock,cmd);
1205 pthread_mutex_lock(&stats_lock);
1206 stats_flush_received++;
1207 pthread_mutex_unlock(&stats_lock);
1209 get_abs_path(&file, file_tmp);
1210 if (!check_file_access(file, sock)) return 0;
1212 status = flush_file (file);
1214 return send_response(sock, RESP_OK, "Successfully flushed %s.\n", file);
1215 else if (status == ENOENT)
1217 /* no file in our tree; see whether it exists at all */
1218 struct stat statbuf;
1220 memset(&statbuf, 0, sizeof(statbuf));
1221 if (stat(file, &statbuf) == 0 && S_ISREG(statbuf.st_mode))
1222 return send_response(sock, RESP_OK, "Nothing to flush: %s.\n", file);
1224 return send_response(sock, RESP_ERR, "No such file: %s.\n", file);
1226 else if (status < 0)
1227 return send_response(sock, RESP_ERR, "Internal error.\n");
1229 return send_response(sock, RESP_ERR, "Failed with status %i.\n", status);
1234 } /* }}} int handle_request_flush */
1236 static int handle_request_flushall(HANDLER_PROTO) /* {{{ */
1238 RRDD_LOG(LOG_DEBUG, "Received FLUSHALL");
1240 pthread_mutex_lock(&cache_lock);
1241 flush_old_values(-1);
1242 pthread_mutex_unlock(&cache_lock);
1244 return send_response(sock, RESP_OK, "Started flush.\n");
1245 } /* }}} static int handle_request_flushall */
1247 static int handle_request_pending(HANDLER_PROTO) /* {{{ */
1250 char *file, file_tmp[PATH_MAX];
1253 status = buffer_get_field(&buffer, &buffer_size, &file);
1255 return syntax_error(sock,cmd);
1257 get_abs_path(&file, file_tmp);
1259 pthread_mutex_lock(&cache_lock);
1260 ci = g_tree_lookup(cache_tree, file);
1263 pthread_mutex_unlock(&cache_lock);
1264 return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(ENOENT));
1267 for (size_t i=0; i < ci->values_num; i++)
1268 add_response_info(sock, "%s\n", ci->values[i]);
1270 pthread_mutex_unlock(&cache_lock);
1271 return send_response(sock, RESP_OK, "updates pending\n");
1272 } /* }}} static int handle_request_pending */
1274 static int handle_request_forget(HANDLER_PROTO) /* {{{ */
1278 char *file, file_tmp[PATH_MAX];
1280 status = buffer_get_field(&buffer, &buffer_size, &file);
1282 return syntax_error(sock,cmd);
1284 get_abs_path(&file, file_tmp);
1285 if (!check_file_access(file, sock)) return 0;
1287 pthread_mutex_lock(&cache_lock);
1288 found = g_tree_remove(cache_tree, file);
1289 pthread_mutex_unlock(&cache_lock);
1293 if (!JOURNAL_REPLAY(sock))
1294 journal_write("forget", file);
1296 return send_response(sock, RESP_OK, "Gone!\n");
1299 return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(ENOENT));
1303 } /* }}} static int handle_request_forget */
1305 static int handle_request_queue (HANDLER_PROTO) /* {{{ */
1309 pthread_mutex_lock(&cache_lock);
1311 ci = cache_queue_head;
1314 add_response_info(sock, "%d %s\n", ci->values_num, ci->file);
1318 pthread_mutex_unlock(&cache_lock);
1320 return send_response(sock, RESP_OK, "in queue.\n");
1321 } /* }}} int handle_request_queue */
1323 static int handle_request_update (HANDLER_PROTO) /* {{{ */
1325 char *file, file_tmp[PATH_MAX];
1328 char orig_buf[RRD_CMD_MAX];
1332 /* save it for the journal later */
1333 if (!JOURNAL_REPLAY(sock))
1334 strncpy(orig_buf, buffer, buffer_size);
1336 status = buffer_get_field (&buffer, &buffer_size, &file);
1338 return syntax_error(sock,cmd);
1340 pthread_mutex_lock(&stats_lock);
1341 stats_updates_received++;
1342 pthread_mutex_unlock(&stats_lock);
1344 get_abs_path(&file, file_tmp);
1345 if (!check_file_access(file, sock)) return 0;
1347 pthread_mutex_lock (&cache_lock);
1348 ci = g_tree_lookup (cache_tree, file);
1350 if (ci == NULL) /* {{{ */
1352 struct stat statbuf;
1355 /* don't hold the lock while we setup; stat(2) might block */
1356 pthread_mutex_unlock(&cache_lock);
1358 memset (&statbuf, 0, sizeof (statbuf));
1359 status = stat (file, &statbuf);
1362 RRDD_LOG (LOG_NOTICE, "handle_request_update: stat (%s) failed.", file);
1365 if (status == ENOENT)
1366 return send_response(sock, RESP_ERR, "No such file: %s\n", file);
1368 return send_response(sock, RESP_ERR,
1369 "stat failed with error %i.\n", status);
1371 if (!S_ISREG (statbuf.st_mode))
1372 return send_response(sock, RESP_ERR, "Not a regular file: %s\n", file);
1374 if (access(file, R_OK|W_OK) != 0)
1375 return send_response(sock, RESP_ERR, "Cannot read/write %s: %s\n",
1376 file, rrd_strerror(errno));
1378 ci = (cache_item_t *) malloc (sizeof (cache_item_t));
1381 RRDD_LOG (LOG_ERR, "handle_request_update: malloc failed.");
1383 return send_response(sock, RESP_ERR, "malloc failed.\n");
1385 memset (ci, 0, sizeof (cache_item_t));
1387 ci->file = strdup (file);
1388 if (ci->file == NULL)
1391 RRDD_LOG (LOG_ERR, "handle_request_update: strdup failed.");
1393 return send_response(sock, RESP_ERR, "strdup failed.\n");
1396 wipe_ci_values(ci, now);
1397 ci->flags = CI_FLAGS_IN_TREE;
1398 pthread_cond_init(&ci->flushed, NULL);
1400 pthread_mutex_lock(&cache_lock);
1402 /* another UPDATE might have added this entry in the meantime */
1403 tmp = g_tree_lookup (cache_tree, file);
1405 g_tree_replace (cache_tree, (void *) ci->file, (void *) ci);
1408 free_cache_item (ci);
1412 /* state may have changed while we were unlocked */
1413 if (state == SHUTDOWN)
1416 assert (ci != NULL);
1418 /* don't re-write updates in replay mode */
1419 if (!JOURNAL_REPLAY(sock))
1420 journal_write("update", orig_buf);
1422 while (buffer_size > 0)
1428 status = buffer_get_field (&buffer, &buffer_size, &value);
1431 RRDD_LOG (LOG_INFO, "handle_request_update: Error reading field.");
1435 /* make sure update time is always moving forward */
1436 stamp = strtol(value, &eostamp, 10);
1437 if (eostamp == value || eostamp == NULL || *eostamp != ':')
1439 pthread_mutex_unlock(&cache_lock);
1440 return send_response(sock, RESP_ERR,
1441 "Cannot find timestamp in '%s'!\n", value);
1443 else if (stamp <= ci->last_update_stamp)
1445 pthread_mutex_unlock(&cache_lock);
1446 return send_response(sock, RESP_ERR,
1447 "illegal attempt to update using time %ld when last"
1448 " update time is %ld (minimum one second step)\n",
1449 stamp, ci->last_update_stamp);
1452 ci->last_update_stamp = stamp;
1454 if (!rrd_add_strdup_chunk(&ci->values, &ci->values_num, value,
1455 &ci->values_alloc, config_alloc_chunk))
1457 RRDD_LOG (LOG_ERR, "handle_request_update: rrd_add_strdup failed.");
1464 if (((now - ci->last_flush_time) >= config_write_interval)
1465 && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
1466 && (ci->values_num > 0))
1468 enqueue_cache_item (ci, TAIL);
1471 pthread_mutex_unlock (&cache_lock);
1474 return send_response(sock, RESP_ERR, "No values updated.\n");
1476 return send_response(sock, RESP_OK,
1477 "errors, enqueued %i value(s).\n", values_num);
1482 } /* }}} int handle_request_update */
1484 static int handle_request_fetch (HANDLER_PROTO) /* {{{ */
1486 char *file, file_tmp[PATH_MAX];
1495 unsigned long ds_cnt;
1502 rrd_value_t *data_ptr;
1509 /* Read the arguments */
1512 status = buffer_get_field (&buffer, &buffer_size, &file);
1516 status = buffer_get_field (&buffer, &buffer_size, &cf);
1520 status = buffer_get_field (&buffer, &buffer_size, &start_str);
1528 status = buffer_get_field (&buffer, &buffer_size, &end_str);
1538 return (syntax_error(sock,cmd));
1540 get_abs_path(&file, file_tmp);
1541 if (!check_file_access(file, sock)) return 0;
1543 status = flush_file (file);
1544 if ((status != 0) && (status != ENOENT))
1545 return (send_response (sock, RESP_ERR,
1546 "flush_file (%s) failed with status %i.\n", file, status));
1548 t = time (NULL); /* "now" */
1550 /* Parse start time */
1551 if (start_str != NULL)
1558 value = strtol (start_str, &endptr, /* base = */ 0);
1559 if ((endptr == start_str) || (errno != 0))
1560 return (send_response(sock, RESP_ERR,
1561 "Cannot parse start time `%s': Only simple integers are allowed.\n",
1565 start_tm = (time_t) value;
1567 start_tm = (time_t) (t + value);
1571 start_tm = t - 86400;
1574 /* Parse end time */
1575 if (end_str != NULL)
1582 value = strtol (end_str, &endptr, /* base = */ 0);
1583 if ((endptr == end_str) || (errno != 0))
1584 return (send_response(sock, RESP_ERR,
1585 "Cannot parse end time `%s': Only simple integers are allowed.\n",
1589 end_tm = (time_t) value;
1591 end_tm = (time_t) (t + value);
1603 status = rrd_fetch_r (file, cf, &start_tm, &end_tm, &step,
1604 &ds_cnt, &ds_namv, &data);
1606 return (send_response(sock, RESP_ERR,
1607 "rrd_fetch_r failed: %s\n", rrd_get_error ()));
1609 add_response_info (sock, "FlushVersion: %lu\n", 1);
1610 add_response_info (sock, "Start: %lu\n", (unsigned long) start_tm);
1611 add_response_info (sock, "End: %lu\n", (unsigned long) end_tm);
1612 add_response_info (sock, "Step: %lu\n", step);
1613 add_response_info (sock, "DSCount: %lu\n", ds_cnt);
1615 #define SSTRCAT(buffer,str,buffer_fill) do { \
1616 size_t str_len = strlen (str); \
1617 if ((buffer_fill + str_len) > sizeof (buffer)) \
1618 str_len = sizeof (buffer) - buffer_fill; \
1619 if (str_len > 0) { \
1620 strncpy (buffer + buffer_fill, str, str_len); \
1621 buffer_fill += str_len; \
1622 assert (buffer_fill <= sizeof (buffer)); \
1623 if (buffer_fill == sizeof (buffer)) \
1624 buffer[buffer_fill - 1] = 0; \
1626 buffer[buffer_fill] = 0; \
1630 { /* Add list of DS names */
1632 size_t linebuf_fill;
1634 memset (linebuf, 0, sizeof (linebuf));
1636 for (i = 0; i < ds_cnt; i++)
1639 SSTRCAT (linebuf, " ", linebuf_fill);
1640 SSTRCAT (linebuf, ds_namv[i], linebuf_fill);
1641 rrd_freemem(ds_namv[i]);
1643 rrd_freemem(ds_namv);
1644 add_response_info (sock, "DSName: %s\n", linebuf);
1647 /* Add the actual data */
1650 for (t = start_tm + step; t <= end_tm; t += step)
1653 size_t linebuf_fill;
1656 memset (linebuf, 0, sizeof (linebuf));
1658 for (i = 0; i < ds_cnt; i++)
1660 snprintf (tmp, sizeof (tmp), " %0.10e", *data_ptr);
1661 tmp[sizeof (tmp) - 1] = 0;
1662 SSTRCAT (linebuf, tmp, linebuf_fill);
1667 add_response_info (sock, "%10lu:%s\n", (unsigned long) t, linebuf);
1671 return (send_response (sock, RESP_OK, "Success\n"));
1673 } /* }}} int handle_request_fetch */
1675 /* we came across a "WROTE" entry during journal replay.
1676 * throw away any values that we have accumulated for this file
1678 static int handle_request_wrote (HANDLER_PROTO) /* {{{ */
1681 const char *file = buffer;
1683 pthread_mutex_lock(&cache_lock);
1685 ci = g_tree_lookup(cache_tree, file);
1688 pthread_mutex_unlock(&cache_lock);
1693 rrd_free_ptrs((void ***) &ci->values, &ci->values_num);
1695 wipe_ci_values(ci, now);
1696 remove_from_queue(ci);
1698 pthread_mutex_unlock(&cache_lock);
1700 } /* }}} int handle_request_wrote */
1702 static int handle_request_info (HANDLER_PROTO) /* {{{ */
1704 char *file, file_tmp[PATH_MAX];
1708 /* obtain filename */
1709 status = buffer_get_field(&buffer, &buffer_size, &file);
1711 return syntax_error(sock,cmd);
1712 /* get full pathname */
1713 get_abs_path(&file, file_tmp);
1714 if (!check_file_access(file, sock)) {
1715 return send_response(sock, RESP_ERR, "Cannot read: %s\n", file);
1719 info = rrd_info_r(file);
1721 return send_response(sock, RESP_ERR, "RRD Error: %s\n", rrd_get_error());
1723 for (rrd_info_t *data = info; data != NULL; data = data->next) {
1724 switch (data->type) {
1726 if (isnan(data->value.u_val))
1727 add_response_info(sock,"%s %d NaN\n",data->key, data->type);
1729 add_response_info(sock,"%s %d %0.10e\n", data->key, data->type, data->value.u_val);
1732 add_response_info(sock,"%s %d %lu\n", data->key, data->type, data->value.u_cnt);
1735 add_response_info(sock,"%s %d %d\n", data->key, data->type, data->value.u_int);
1738 add_response_info(sock,"%s %d %s\n", data->key, data->type, data->value.u_str);
1741 add_response_info(sock,"%s %d %lu\n", data->key, data->type, data->value.u_blo.size);
1746 rrd_info_free(info);
1748 return send_response(sock, RESP_OK, "Info for %s follows\n",file);
1749 } /* }}} static int handle_request_info */
1751 static int handle_request_first (HANDLER_PROTO) /* {{{ */
1753 char *i, *file, file_tmp[PATH_MAX];
1758 /* obtain filename */
1759 status = buffer_get_field(&buffer, &buffer_size, &file);
1761 return syntax_error(sock,cmd);
1762 /* get full pathname */
1763 get_abs_path(&file, file_tmp);
1764 if (!check_file_access(file, sock)) {
1765 return send_response(sock, RESP_ERR, "Cannot read: %s\n", file);
1768 status = buffer_get_field(&buffer, &buffer_size, &i);
1770 return syntax_error(sock,cmd);
1773 return send_response(sock, RESP_ERR, "Invalid index specified (%d)\n", idx);
1778 t = rrd_first_r(file,idx);
1780 return send_response(sock, RESP_ERR, "RRD Error: %s\n", rrd_get_error());
1782 return send_response(sock, RESP_OK, "%lu\n",(unsigned)t);
1783 } /* }}} static int handle_request_first */
1786 static int handle_request_last (HANDLER_PROTO) /* {{{ */
1788 char *file, file_tmp[PATH_MAX];
1790 time_t t, from_file, step;
1791 rrd_file_t * rrd_file;
1795 /* obtain filename */
1796 status = buffer_get_field(&buffer, &buffer_size, &file);
1798 return syntax_error(sock,cmd);
1799 /* get full pathname */
1800 get_abs_path(&file, file_tmp);
1801 if (!check_file_access(file, sock)) {
1802 return send_response(sock, RESP_ERR, "Cannot read: %s\n", file);
1806 rrd_file = rrd_open(file,&rrd,RRD_READONLY);
1808 return send_response(sock, RESP_ERR, "RRD Error: %s\n", rrd_get_error());
1810 from_file = rrd.live_head->last_up;
1811 step = rrd.stat_head->pdp_step;
1812 rrd_close(rrd_file);
1813 pthread_mutex_lock(&cache_lock);
1814 ci = g_tree_lookup(cache_tree, file);
1816 t = ci->last_update_stamp;
1819 pthread_mutex_unlock(&cache_lock);
1823 return send_response(sock, RESP_ERR, "Error: rrdcached: Invalid timestamp returned\n");
1825 return send_response(sock, RESP_OK, "%lu\n",(unsigned)t);
1826 } /* }}} static int handle_request_last */
1828 static int handle_request_create (HANDLER_PROTO) /* {{{ */
1830 char *file, file_tmp[PATH_MAX];
1835 unsigned long step = 300;
1836 time_t last_up = time(NULL)-10;
1837 int no_overwrite = opt_no_overwrite;
1840 /* obtain filename */
1841 status = buffer_get_field(&buffer, &buffer_size, &file);
1843 return syntax_error(sock,cmd);
1844 /* get full pathname */
1845 get_abs_path(&file, file_tmp);
1846 if (!check_file_access(file, sock)) {
1847 return send_response(sock, RESP_ERR, "Cannot read: %s\n", file);
1849 RRDD_LOG(LOG_INFO, "rrdcreate request for %s",file);
1851 while ((status = buffer_get_field(&buffer, &buffer_size, &tok)) == 0 && tok) {
1852 if( ! strncmp(tok,"-b",2) ) {
1853 status = buffer_get_field(&buffer, &buffer_size, &tok );
1854 if (status != 0) return syntax_error(sock,cmd);
1855 last_up = (time_t) atol(tok);
1858 if( ! strncmp(tok,"-s",2) ) {
1859 status = buffer_get_field(&buffer, &buffer_size, &tok );
1860 if (status != 0) return syntax_error(sock,cmd);
1864 if( ! strncmp(tok,"-O",2) ) {
1868 if( ! strncmp(tok,"DS:",3) ) { av[ac++]=tok; continue; }
1869 if( ! strncmp(tok,"RRA:",4) ) { av[ac++]=tok; continue; }
1870 return syntax_error(sock,cmd);
1873 return send_response(sock, RESP_ERR, "The step size cannot be less than 1 second.\n");
1875 if (last_up < 3600 * 24 * 365 * 10) {
1876 return send_response(sock, RESP_ERR, "The first entry must be after 1980.\n");
1880 status = rrd_create_r2(file,step,last_up,no_overwrite,ac,(const char **)av);
1883 return send_response(sock, RESP_OK, "RRD created OK\n");
1885 return send_response(sock, RESP_ERR, "RRD Error: %s\n", rrd_get_error());
1886 } /* }}} static int handle_request_create */
1888 /* start "BATCH" processing */
1889 static int batch_start (HANDLER_PROTO) /* {{{ */
1892 if (sock->batch_start)
1893 return send_response(sock, RESP_ERR, "Already in BATCH\n");
1895 status = send_response(sock, RESP_OK,
1896 "Go ahead. End with dot '.' on its own line.\n");
1897 sock->batch_start = time(NULL);
1898 sock->batch_cmd = 0;
1901 } /* }}} static int batch_start */
1903 /* finish "BATCH" processing and return results to the client */
1904 static int batch_done (HANDLER_PROTO) /* {{{ */
1906 assert(sock->batch_start);
1907 sock->batch_start = 0;
1908 sock->batch_cmd = 0;
1909 return send_response(sock, RESP_OK, "errors\n");
1910 } /* }}} static int batch_done */
1912 static int handle_request_quit (HANDLER_PROTO) /* {{{ */
1915 } /* }}} static int handle_request_quit */
1917 static command_t list_of_commands[] = { /* {{{ */
1920 handle_request_update,
1922 "UPDATE <filename> <values> [<values> ...]\n"
1924 "Adds the given file to the internal cache if it is not yet known and\n"
1925 "appends the given value(s) to the entry. See the rrdcached(1) manpage\n"
1928 "Each <values> has the following form:\n"
1929 " <values> = <time>:<value>[:<value>[...]]\n"
1930 "See the rrdupdate(1) manpage for details.\n"
1934 handle_request_wrote,
1935 CMD_CONTEXT_JOURNAL,
1941 handle_request_flush,
1942 CMD_CONTEXT_CLIENT | CMD_CONTEXT_BATCH,
1943 "FLUSH <filename>\n"
1945 "Adds the given filename to the head of the update queue and returns\n"
1946 "after it has been dequeued.\n"
1950 handle_request_flushall,
1954 "Triggers writing of all pending updates. Returns immediately.\n"
1958 handle_request_pending,
1960 "PENDING <filename>\n"
1962 "Shows any 'pending' updates for a file, in order.\n"
1963 "The updates shown have not yet been written to the underlying RRD file.\n"
1967 handle_request_forget,
1969 "FORGET <filename>\n"
1971 "Removes the file completely from the cache.\n"
1972 "Any pending updates for the file will be lost.\n"
1976 handle_request_queue,
1980 "Shows all files in the output queue.\n"
1981 "The output is zero or more lines in the following format:\n"
1982 "(where <num_vals> is the number of values to be written)\n"
1984 "<num_vals> <filename>\n"
1988 handle_request_stats,
1992 "Returns some performance counters, see the rrdcached(1) manpage for\n"
1993 "a description of the values.\n"
1997 handle_request_help,
1999 "HELP [<command>]\n",
2000 NULL, /* special! */
2008 "The 'BATCH' command permits the client to initiate a bulk load\n"
2009 " of commands to rrdcached.\n"
2014 " server: 0 Go ahead. End with dot '.' on its own line.\n"
2015 " client: command #1\n"
2016 " client: command #2\n"
2017 " client: ... and so on\n"
2019 " server: 2 errors\n"
2020 " server: 7 message for command #7\n"
2021 " server: 9 message for command #9\n"
2023 "For more information, consult the rrdcached(1) documentation.\n"
2026 ".", /* BATCH terminator */
2034 handle_request_fetch,
2036 "FETCH <file> <CF> [<start> [<end>]]\n"
2038 "The 'FETCH' can be used by the client to retrieve values from an RRD file.\n"
2042 handle_request_info,
2044 "INFO <filename>\n",
2045 "The INFO command retrieves information about a specified RRD file.\n"
2046 "This is returned in standard rrdinfo format, a sequence of lines\n"
2047 "with the format <keyname> = <value>\n"
2048 "Note that this is the data as of the last update of the RRD file itself,\n"
2049 "not the last time data was received via rrdcached, so there may be pending\n"
2050 "updates in the queue. If this bothers you, then first run a FLUSH.\n"
2054 handle_request_first,
2056 "FIRST <filename> <rra index>\n",
2057 "The FIRST command retrieves the first data time for a specified RRA in\n"
2062 handle_request_last,
2064 "LAST <filename>\n",
2065 "The LAST command retrieves the last update time for a specified RRD file.\n"
2066 "Note that this is the time of the last update of the RRD file itself, not\n"
2067 "the last time data was received via rrdcached, so there may be pending\n"
2068 "updates in the queue. If this bothers you, then first run a FLUSH.\n"
2072 handle_request_create,
2073 CMD_CONTEXT_CLIENT | CMD_CONTEXT_BATCH,
2074 "CREATE <filename> [-b start] [-s step] [-O] <DS definitions> <RRA definitions>\n",
2075 "The CREATE command will create an RRD file, overwriting any existing file\n"
2076 "unless the -O option is given or rrdcached was started with the -O option.\n"
2077 "The start parameter needs to be in seconds since 1/1/70 (AT-style syntax is\n"
2078 "not acceptable) and the step is in seconds (default is 300).\n"
2079 "The DS and RRA definitions are as for the 'rrdtool create' command.\n"
2083 handle_request_quit,
2084 CMD_CONTEXT_CLIENT | CMD_CONTEXT_BATCH,
2087 "Disconnect from rrdcached.\n"
2089 }; /* }}} command_t list_of_commands[] */
2090 static size_t list_of_commands_len = sizeof (list_of_commands)
2091 / sizeof (list_of_commands[0]);
2093 static command_t *find_command(char *cmd)
2097 for (i = 0; i < list_of_commands_len; i++)
2098 if (strcasecmp(cmd, list_of_commands[i].cmd) == 0)
2099 return (&list_of_commands[i]);
2103 /* We currently use the index in the `list_of_commands' array as a bit position
2104 * in `listen_socket_t.permissions'. This member schould NEVER be accessed from
2105 * outside these functions so that switching to a more elegant storage method
2106 * is easily possible. */
2107 static ssize_t find_command_index (const char *cmd) /* {{{ */
2111 for (i = 0; i < list_of_commands_len; i++)
2112 if (strcasecmp(cmd, list_of_commands[i].cmd) == 0)
2113 return ((ssize_t) i);
2115 } /* }}} ssize_t find_command_index */
2117 static int socket_permission_check (listen_socket_t *sock, /* {{{ */
2122 if (JOURNAL_REPLAY(sock))
2128 if ((strcasecmp ("QUIT", cmd) == 0)
2129 || (strcasecmp ("HELP", cmd) == 0))
2131 else if (strcmp (".", cmd) == 0)
2134 i = find_command_index (cmd);
2139 if ((sock->permissions & (1 << i)) != 0)
2142 } /* }}} int socket_permission_check */
2144 static int socket_permission_add (listen_socket_t *sock, /* {{{ */
2149 i = find_command_index (cmd);
2154 sock->permissions |= (1 << i);
2156 } /* }}} int socket_permission_add */
2158 static void socket_permission_clear (listen_socket_t *sock) /* {{{ */
2160 sock->permissions = 0;
2161 } /* }}} socket_permission_clear */
2163 static void socket_permission_copy (listen_socket_t *dest, /* {{{ */
2164 listen_socket_t *src)
2166 dest->permissions = src->permissions;
2167 } /* }}} socket_permission_copy */
2169 static void socket_permission_set_all (listen_socket_t *sock) /* {{{ */
2173 sock->permissions = 0;
2174 for (i = 0; i < list_of_commands_len; i++)
2175 sock->permissions |= (1 << i);
2176 } /* }}} void socket_permission_set_all */
2178 /* check whether commands are received in the expected context */
2179 static int command_check_context(listen_socket_t *sock, command_t *cmd)
2181 if (JOURNAL_REPLAY(sock))
2182 return (cmd->context & CMD_CONTEXT_JOURNAL);
2183 else if (sock->batch_start)
2184 return (cmd->context & CMD_CONTEXT_BATCH);
2186 return (cmd->context & CMD_CONTEXT_CLIENT);
2192 static int handle_request_help (HANDLER_PROTO) /* {{{ */
2197 command_t *help = NULL;
2199 status = buffer_get_field (&buffer, &buffer_size, &cmd_str);
2201 help = find_command(cmd_str);
2203 if (help && (help->syntax || help->help))
2205 char tmp[RRD_CMD_MAX];
2207 snprintf(tmp, sizeof(tmp)-1, "Help for %s\n", help->cmd);
2211 add_response_info(sock, "Usage: %s\n", help->syntax);
2214 add_response_info(sock, "%s\n", help->help);
2220 resp_txt = "Command overview\n";
2222 for (i = 0; i < list_of_commands_len; i++)
2224 if (list_of_commands[i].syntax == NULL)
2226 add_response_info (sock, "%s", list_of_commands[i].syntax);
2230 return send_response(sock, RESP_OK, resp_txt);
2231 } /* }}} int handle_request_help */
2233 static int handle_request (DISPATCH_PROTO) /* {{{ */
2235 char *buffer_ptr = buffer;
2236 char *cmd_str = NULL;
2237 command_t *cmd = NULL;
2240 assert (buffer[buffer_size - 1] == '\0');
2242 status = buffer_get_field (&buffer_ptr, &buffer_size, &cmd_str);
2245 RRDD_LOG (LOG_INFO, "handle_request: Unable parse command.");
2249 if (sock != NULL && sock->batch_start)
2252 cmd = find_command(cmd_str);
2254 return send_response(sock, RESP_ERR, "Unknown command: %s\n", cmd_str);
2256 if (!socket_permission_check (sock, cmd->cmd))
2257 return send_response(sock, RESP_ERR, "Permission denied.\n");
2259 if (!command_check_context(sock, cmd))
2260 return send_response(sock, RESP_ERR, "Can't use '%s' here.\n", cmd_str);
2262 return cmd->handler(cmd, sock, now, buffer_ptr, buffer_size);
2263 } /* }}} int handle_request */
2265 static void journal_set_free (journal_set *js) /* {{{ */
2270 rrd_free_ptrs((void ***) &js->files, &js->files_num);
2273 } /* }}} journal_set_free */
2275 static void journal_set_remove (journal_set *js) /* {{{ */
2280 for (uint i=0; i < js->files_num; i++)
2282 RRDD_LOG(LOG_DEBUG, "removing old journal %s", js->files[i]);
2283 unlink(js->files[i]);
2285 } /* }}} journal_set_remove */
2287 /* close current journal file handle.
2288 * MUST hold journal_lock before calling */
2289 static void journal_close(void) /* {{{ */
2291 if (journal_fh != NULL)
2293 if (fclose(journal_fh) != 0)
2294 RRDD_LOG(LOG_ERR, "cannot close journal: %s", rrd_strerror(errno));
2299 } /* }}} journal_close */
2301 /* MUST hold journal_lock before calling */
2302 static void journal_new_file(void) /* {{{ */
2306 char new_file[PATH_MAX + 1];
2308 assert(journal_dir != NULL);
2309 assert(journal_cur != NULL);
2313 gettimeofday(&now, NULL);
2314 /* this format assures that the files sort in strcmp() order */
2315 snprintf(new_file, PATH_MAX, "%s/%s.%010d.%06d",
2316 journal_dir, JOURNAL_BASE, (int)now.tv_sec, (int)now.tv_usec);
2318 new_fd = open(new_file, O_WRONLY|O_CREAT|O_APPEND,
2319 S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
2323 journal_fh = fdopen(new_fd, "a");
2324 if (journal_fh == NULL)
2327 journal_size = ftell(journal_fh);
2328 RRDD_LOG(LOG_DEBUG, "started new journal %s", new_file);
2330 /* record the file in the journal set */
2331 rrd_add_strdup(&journal_cur->files, &journal_cur->files_num, new_file);
2337 "JOURNALING DISABLED: Error while trying to create %s : %s",
2338 new_file, rrd_strerror(errno));
2340 "JOURNALING DISABLED: All values will be flushed at shutdown");
2343 config_flush_at_shutdown = 1;
2345 } /* }}} journal_new_file */
2347 /* MUST NOT hold journal_lock before calling this */
2348 static void journal_rotate(void) /* {{{ */
2350 journal_set *old_js = NULL;
2352 if (journal_dir == NULL)
2355 RRDD_LOG(LOG_DEBUG, "rotating journals");
2357 pthread_mutex_lock(&stats_lock);
2358 ++stats_journal_rotate;
2359 pthread_mutex_unlock(&stats_lock);
2361 pthread_mutex_lock(&journal_lock);
2365 /* rotate the journal sets */
2366 old_js = journal_old;
2367 journal_old = journal_cur;
2368 journal_cur = calloc(1, sizeof(journal_set));
2370 if (journal_cur != NULL)
2373 RRDD_LOG(LOG_CRIT, "journal_rotate: malloc(journal_set) failed\n");
2375 pthread_mutex_unlock(&journal_lock);
2377 journal_set_remove(old_js);
2378 journal_set_free (old_js);
2380 } /* }}} static void journal_rotate */
2382 /* MUST hold journal_lock when calling */
2383 static void journal_done(void) /* {{{ */
2385 if (journal_cur == NULL)
2390 if (config_flush_at_shutdown)
2392 RRDD_LOG(LOG_INFO, "removing journals");
2393 journal_set_remove(journal_old);
2394 journal_set_remove(journal_cur);
2398 RRDD_LOG(LOG_INFO, "expedited shutdown; "
2399 "journals will be used at next startup");
2402 journal_set_free(journal_cur);
2403 journal_set_free(journal_old);
2406 } /* }}} static void journal_done */
2408 static int journal_write(char *cmd, char *args) /* {{{ */
2412 if (journal_fh == NULL)
2415 pthread_mutex_lock(&journal_lock);
2416 chars = fprintf(journal_fh, "%s %s\n", cmd, args);
2417 journal_size += chars;
2419 if (journal_size > JOURNAL_MAX)
2422 pthread_mutex_unlock(&journal_lock);
2426 pthread_mutex_lock(&stats_lock);
2427 stats_journal_bytes += chars;
2428 pthread_mutex_unlock(&stats_lock);
2432 } /* }}} static int journal_write */
2434 static int journal_replay (const char *file) /* {{{ */
2440 char entry[RRD_CMD_MAX];
2443 if (file == NULL) return 0;
2446 char *reason = "unknown error";
2448 struct stat statbuf;
2450 memset(&statbuf, 0, sizeof(statbuf));
2451 if (stat(file, &statbuf) != 0)
2453 reason = "stat error";
2456 else if (!S_ISREG(statbuf.st_mode))
2458 reason = "not a regular file";
2461 if (statbuf.st_uid != daemon_uid)
2463 reason = "not owned by daemon user";
2466 if (statbuf.st_mode & (S_IWGRP|S_IWOTH))
2468 reason = "must not be user/group writable";
2474 RRDD_LOG(LOG_ERR, "journal_replay: %s : %s (%s)",
2475 file, rrd_strerror(status), reason);
2480 fh = fopen(file, "r");
2483 if (errno != ENOENT)
2484 RRDD_LOG(LOG_ERR, "journal_replay: cannot open journal file: '%s' (%s)",
2485 file, rrd_strerror(errno));
2489 RRDD_LOG(LOG_NOTICE, "replaying from journal: %s", file);
2498 if (fgets(entry, sizeof(entry), fh) == NULL)
2500 entry_len = strlen(entry);
2502 /* check \n termination in case journal writing crashed mid-line */
2505 else if (entry[entry_len - 1] != '\n')
2507 RRDD_LOG(LOG_NOTICE, "Malformed journal entry at line %"PRIu64, line);
2512 entry[entry_len - 1] = '\0';
2514 if (handle_request(NULL, now, entry, entry_len) == 0)
2522 RRDD_LOG(LOG_INFO, "Replayed %d entries (%d failures)",
2523 entry_cnt, fail_cnt);
2525 return entry_cnt > 0 ? 1 : 0;
2526 } /* }}} static int journal_replay */
2528 static int journal_sort(const void *v1, const void *v2)
2530 char **jn1 = (char **) v1;
2531 char **jn2 = (char **) v2;
2533 return strcmp(*jn1,*jn2);
2536 static void journal_init(void) /* {{{ */
2538 int had_journal = 0;
2540 struct dirent *dent;
2541 char path[PATH_MAX+1];
2543 if (journal_dir == NULL) return;
2545 pthread_mutex_lock(&journal_lock);
2547 journal_cur = calloc(1, sizeof(journal_set));
2548 if (journal_cur == NULL)
2550 RRDD_LOG(LOG_CRIT, "journal_rotate: malloc(journal_set) failed\n");
2554 RRDD_LOG(LOG_INFO, "checking for journal files");
2556 /* Handle old journal files during transition. This gives them the
2557 * correct sort order. TODO: remove after first release
2560 char old_path[PATH_MAX+1];
2561 snprintf(old_path, PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE ".old" );
2562 snprintf(path, PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE ".0000");
2563 rename(old_path, path);
2565 snprintf(old_path, PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE );
2566 snprintf(path, PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE ".0001");
2567 rename(old_path, path);
2570 dir = opendir(journal_dir);
2572 RRDD_LOG(LOG_CRIT, "journal_init: opendir(%s) failed\n", journal_dir);
2575 while ((dent = readdir(dir)) != NULL)
2577 /* looks like a journal file? */
2578 if (strncmp(dent->d_name, JOURNAL_BASE, strlen(JOURNAL_BASE)))
2581 snprintf(path, PATH_MAX, "%s/%s", journal_dir, dent->d_name);
2583 if (!rrd_add_strdup(&journal_cur->files, &journal_cur->files_num, path))
2585 RRDD_LOG(LOG_CRIT, "journal_init: cannot add journal file %s!",
2592 qsort(journal_cur->files, journal_cur->files_num,
2593 sizeof(journal_cur->files[0]), journal_sort);
2595 for (uint i=0; i < journal_cur->files_num; i++)
2596 had_journal += journal_replay(journal_cur->files[i]);
2600 /* it must have been a crash. start a flush */
2601 if (had_journal && config_flush_at_shutdown)
2602 flush_old_values(-1);
2604 pthread_mutex_unlock(&journal_lock);
2606 RRDD_LOG(LOG_INFO, "journal processing complete");
2608 } /* }}} static void journal_init */
2610 static void free_listen_socket(listen_socket_t *sock) /* {{{ */
2612 assert(sock != NULL);
2614 free(sock->rbuf); sock->rbuf = NULL;
2615 free(sock->wbuf); sock->wbuf = NULL;
2617 } /* }}} void free_listen_socket */
2619 static void close_connection(listen_socket_t *sock) /* {{{ */
2627 free_listen_socket(sock);
2629 } /* }}} void close_connection */
2631 static void *connection_thread_main (void *args) /* {{{ */
2633 listen_socket_t *sock;
2636 sock = (listen_socket_t *) args;
2639 /* init read buffers */
2640 sock->next_read = sock->next_cmd = 0;
2641 sock->rbuf = malloc(RBUF_SIZE);
2642 if (sock->rbuf == NULL)
2644 RRDD_LOG(LOG_ERR, "connection_thread_main: cannot malloc read buffer");
2645 close_connection(sock);
2649 pthread_mutex_lock (&connection_threads_lock);
2651 /* LIBWRAP does not support multiple threads! By putting this code
2652 inside pthread_mutex_lock we do not have to worry about request_info
2653 getting overwritten by another thread.
2655 struct request_info req;
2656 request_init(&req, RQ_DAEMON, "rrdcache\0", RQ_FILE, fd, NULL );
2658 if(!hosts_access(&req)) {
2659 RRDD_LOG(LOG_INFO, "refused connection from %s", eval_client(&req));
2660 pthread_mutex_unlock (&connection_threads_lock);
2661 close_connection(sock);
2664 #endif /* HAVE_LIBWRAP */
2665 connection_threads_num++;
2666 pthread_mutex_unlock (&connection_threads_lock);
2668 while (state == RUNNING)
2675 struct pollfd pollfd;
2679 pollfd.events = POLLIN | POLLPRI;
2682 status = poll (&pollfd, 1, /* timeout = */ 500);
2683 if (state != RUNNING)
2685 else if (status == 0) /* timeout */
2687 else if (status < 0) /* error */
2690 if (status != EINTR)
2691 RRDD_LOG (LOG_ERR, "connection_thread_main: poll(2) failed.");
2695 if ((pollfd.revents & POLLHUP) != 0) /* normal shutdown */
2697 else if ((pollfd.revents & (POLLIN | POLLPRI)) == 0)
2699 RRDD_LOG (LOG_WARNING, "connection_thread_main: "
2700 "poll(2) returned something unexpected: %#04hx",
2705 rbytes = read(fd, sock->rbuf + sock->next_read,
2706 RBUF_SIZE - sock->next_read);
2709 RRDD_LOG(LOG_ERR, "connection_thread_main: read() failed.");
2712 else if (rbytes == 0)
2715 sock->next_read += rbytes;
2717 if (sock->batch_start)
2718 now = sock->batch_start;
2722 while ((cmd = next_cmd(sock, &cmd_len)) != NULL)
2724 status = handle_request (sock, now, cmd, cmd_len+1);
2731 close_connection(sock);
2733 /* Remove this thread from the connection threads list */
2734 pthread_mutex_lock (&connection_threads_lock);
2735 connection_threads_num--;
2736 if (connection_threads_num <= 0)
2737 pthread_cond_broadcast(&connection_threads_done);
2738 pthread_mutex_unlock (&connection_threads_lock);
2741 } /* }}} void *connection_thread_main */
2743 static int open_listen_socket_unix (const listen_socket_t *sock) /* {{{ */
2746 struct sockaddr_un sa;
2747 listen_socket_t *temp;
2750 char *path_copy, *dir;
2753 if (strncmp(path, "unix:", strlen("unix:")) == 0)
2754 path += strlen("unix:");
2756 /* dirname may modify its argument */
2757 path_copy = strdup(path);
2758 if (path_copy == NULL)
2760 fprintf(stderr, "rrdcached: strdup(): %s\n",
2761 rrd_strerror(errno));
2765 dir = dirname(path_copy);
2766 if (rrd_mkdir_p(dir, 0777) != 0)
2768 fprintf(stderr, "Failed to create socket directory '%s': %s\n",
2769 dir, rrd_strerror(errno));
2775 temp = (listen_socket_t *) rrd_realloc (listen_fds,
2776 sizeof (listen_fds[0]) * (listen_fds_num + 1));
2779 fprintf (stderr, "rrdcached: open_listen_socket_unix: realloc failed.\n");
2783 memcpy (listen_fds + listen_fds_num, sock, sizeof (listen_fds[0]));
2785 fd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
2788 fprintf (stderr, "rrdcached: unix socket(2) failed: %s\n",
2789 rrd_strerror(errno));
2793 memset (&sa, 0, sizeof (sa));
2794 sa.sun_family = AF_UNIX;
2795 strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
2797 /* if we've gotten this far, we own the pid file. any daemon started
2798 * with the same args must not be alive. therefore, ensure that we can
2799 * create the socket...
2803 status = bind (fd, (struct sockaddr *) &sa, sizeof (sa));
2806 fprintf (stderr, "rrdcached: bind(%s) failed: %s.\n",
2807 path, rrd_strerror(errno));
2812 /* tweak the sockets group ownership */
2813 if (sock->socket_group != (gid_t)-1)
2815 if ( (chown(path, getuid(), sock->socket_group) != 0) ||
2816 (chmod(path, (S_IRUSR|S_IWUSR|S_IXUSR | S_IRGRP|S_IWGRP)) != 0) )
2818 fprintf(stderr, "rrdcached: failed to set socket group permissions (%s)\n", strerror(errno));
2822 if (sock->socket_permissions != (mode_t)-1)
2824 if (chmod(path, sock->socket_permissions) != 0)
2825 fprintf(stderr, "rrdcached: failed to set socket file permissions (%o): %s\n",
2826 (unsigned int)sock->socket_permissions, strerror(errno));
2829 status = listen (fd, /* backlog = */ 10);
2832 fprintf (stderr, "rrdcached: listen(%s) failed: %s.\n",
2833 path, rrd_strerror(errno));
2839 listen_fds[listen_fds_num].fd = fd;
2840 listen_fds[listen_fds_num].family = PF_UNIX;
2841 strncpy(listen_fds[listen_fds_num].addr, path,
2842 sizeof (listen_fds[listen_fds_num].addr) - 1);
2846 } /* }}} int open_listen_socket_unix */
2848 static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */
2850 struct addrinfo ai_hints;
2851 struct addrinfo *ai_res;
2852 struct addrinfo *ai_ptr;
2853 char addr_copy[NI_MAXHOST];
2858 strncpy (addr_copy, sock->addr, sizeof(addr_copy)-1);
2859 addr_copy[sizeof (addr_copy) - 1] = 0;
2862 memset (&ai_hints, 0, sizeof (ai_hints));
2863 ai_hints.ai_flags = 0;
2864 #ifdef AI_ADDRCONFIG
2865 ai_hints.ai_flags |= AI_ADDRCONFIG;
2867 ai_hints.ai_family = AF_UNSPEC;
2868 ai_hints.ai_socktype = SOCK_STREAM;
2871 if (*addr == '[') /* IPv6+port format */
2873 /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
2876 port = strchr (addr, ']');
2879 fprintf (stderr, "rrdcached: Malformed address: %s\n", sock->addr);
2887 else if (*port == 0)
2891 fprintf (stderr, "rrdcached: Garbage after address: %s\n", port);
2894 } /* if (*addr == '[') */
2897 port = rindex(addr, ':');
2905 status = getaddrinfo (addr,
2906 port == NULL ? RRDCACHED_DEFAULT_PORT : port,
2907 &ai_hints, &ai_res);
2910 fprintf (stderr, "rrdcached: getaddrinfo(%s) failed: %s\n",
2911 addr, gai_strerror (status));
2915 for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
2918 listen_socket_t *temp;
2921 temp = (listen_socket_t *) rrd_realloc (listen_fds,
2922 sizeof (listen_fds[0]) * (listen_fds_num + 1));
2926 "rrdcached: open_listen_socket_network: realloc failed.\n");
2930 memcpy (listen_fds + listen_fds_num, sock, sizeof (listen_fds[0]));
2932 fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
2935 fprintf (stderr, "rrdcached: network socket(2) failed: %s.\n",
2936 rrd_strerror(errno));
2940 setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
2942 status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
2945 fprintf (stderr, "rrdcached: bind(%s) failed: %s.\n",
2946 sock->addr, rrd_strerror(errno));
2951 status = listen (fd, /* backlog = */ 10);
2954 fprintf (stderr, "rrdcached: listen(%s) failed: %s\n.",
2955 sock->addr, rrd_strerror(errno));
2957 freeaddrinfo(ai_res);
2961 listen_fds[listen_fds_num].fd = fd;
2962 listen_fds[listen_fds_num].family = ai_ptr->ai_family;
2964 } /* for (ai_ptr) */
2966 freeaddrinfo(ai_res);
2968 } /* }}} static int open_listen_socket_network */
2970 static int open_listen_socket (const listen_socket_t *sock) /* {{{ */
2972 assert(sock != NULL);
2973 assert(sock->addr != NULL);
2975 if (strncmp ("unix:", sock->addr, strlen ("unix:")) == 0
2976 || sock->addr[0] == '/')
2977 return (open_listen_socket_unix(sock));
2979 return (open_listen_socket_network(sock));
2980 } /* }}} int open_listen_socket */
2982 static int close_listen_sockets (void) /* {{{ */
2986 for (i = 0; i < listen_fds_num; i++)
2988 close (listen_fds[i].fd);
2990 if (listen_fds[i].family == PF_UNIX)
2991 unlink(listen_fds[i].addr);
2999 } /* }}} int close_listen_sockets */
3001 static void *listen_thread_main (void UNUSED(*args)) /* {{{ */
3003 struct pollfd *pollfds;
3008 if (listen_fds_num < 1)
3010 RRDD_LOG(LOG_ERR, "listen_thread_main: no listen_fds !");
3014 pollfds_num = listen_fds_num;
3015 pollfds = (struct pollfd *) malloc (sizeof (*pollfds) * pollfds_num);
3016 if (pollfds == NULL)
3018 RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
3021 memset (pollfds, 0, sizeof (*pollfds) * pollfds_num);
3023 RRDD_LOG(LOG_INFO, "listening for connections");
3025 while (state == RUNNING)
3027 for (i = 0; i < pollfds_num; i++)
3029 pollfds[i].fd = listen_fds[i].fd;
3030 pollfds[i].events = POLLIN | POLLPRI;
3031 pollfds[i].revents = 0;
3034 status = poll (pollfds, pollfds_num, /* timeout = */ 1000);
3035 if (state != RUNNING)
3037 else if (status == 0) /* timeout */
3039 else if (status < 0) /* error */
3042 if (status != EINTR)
3044 RRDD_LOG (LOG_ERR, "listen_thread_main: poll(2) failed.");
3049 for (i = 0; i < pollfds_num; i++)
3051 listen_socket_t *client_sock;
3052 struct sockaddr_storage client_sa;
3053 socklen_t client_sa_size;
3055 pthread_attr_t attr;
3057 if (pollfds[i].revents == 0)
3060 if ((pollfds[i].revents & (POLLIN | POLLPRI)) == 0)
3062 RRDD_LOG (LOG_ERR, "listen_thread_main: "
3063 "poll(2) returned something unexpected for listen FD #%i.",
3068 client_sock = (listen_socket_t *) malloc (sizeof (listen_socket_t));
3069 if (client_sock == NULL)
3071 RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
3074 memcpy(client_sock, &listen_fds[i], sizeof(listen_fds[0]));
3076 client_sa_size = sizeof (client_sa);
3077 client_sock->fd = accept (pollfds[i].fd,
3078 (struct sockaddr *) &client_sa, &client_sa_size);
3079 if (client_sock->fd < 0)
3081 RRDD_LOG (LOG_ERR, "listen_thread_main: accept(2) failed.");
3086 pthread_attr_init (&attr);
3087 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
3089 status = pthread_create (&tid, &attr, connection_thread_main,
3093 RRDD_LOG (LOG_ERR, "listen_thread_main: pthread_create failed.");
3094 close_connection(client_sock);
3097 } /* for (pollfds_num) */
3098 } /* while (state == RUNNING) */
3100 RRDD_LOG(LOG_INFO, "starting shutdown");
3102 close_listen_sockets ();
3104 pthread_mutex_lock (&connection_threads_lock);
3105 while (connection_threads_num > 0)
3106 pthread_cond_wait(&connection_threads_done, &connection_threads_lock);
3107 pthread_mutex_unlock (&connection_threads_lock);
3112 } /* }}} void *listen_thread_main */
3114 static int daemonize (void) /* {{{ */
3119 daemon_uid = geteuid();
3121 pid_fd = open_pidfile("create", O_CREAT|O_EXCL|O_WRONLY);
3123 pid_fd = check_pidfile();
3127 /* open all the listen sockets */
3128 if (config_listen_address_list_len > 0)
3130 for (size_t i = 0; i < config_listen_address_list_len; i++)
3131 open_listen_socket (config_listen_address_list[i]);
3133 rrd_free_ptrs((void ***) &config_listen_address_list,
3134 &config_listen_address_list_len);
3138 strncpy(default_socket.addr, RRDCACHED_DEFAULT_ADDRESS,
3139 sizeof(default_socket.addr) - 1);
3140 default_socket.addr[sizeof(default_socket.addr) - 1] = '\0';
3142 if (default_socket.permissions == 0)
3143 socket_permission_set_all (&default_socket);
3145 open_listen_socket (&default_socket);
3148 if (listen_fds_num < 1)
3150 fprintf (stderr, "rrdcached: FATAL: cannot open any listen sockets\n");
3154 if (!stay_foreground)
3161 fprintf (stderr, "daemonize: fork(2) failed.\n");
3167 /* Become session leader */
3170 /* Open the first three file descriptors to /dev/null */
3175 open ("/dev/null", O_RDWR);
3176 if (dup(0) == -1 || dup(0) == -1){
3177 RRDD_LOG (LOG_ERR, "faild to run dup.\n");
3179 } /* if (!stay_foreground) */
3181 /* Change into the /tmp directory. */
3182 base_dir = (config_base_dir != NULL)
3186 if (chdir (base_dir) != 0)
3188 fprintf (stderr, "daemonize: chdir (%s) failed.\n", base_dir);
3192 install_signal_handlers();
3194 openlog ("rrdcached", LOG_PID, LOG_DAEMON);
3195 RRDD_LOG(LOG_INFO, "starting up");
3197 cache_tree = g_tree_new_full ((GCompareDataFunc) strcmp, NULL, NULL,
3198 (GDestroyNotify) free_cache_item);
3199 if (cache_tree == NULL)
3201 RRDD_LOG (LOG_ERR, "daemonize: g_tree_new failed.");
3205 return write_pidfile (pid_fd);
3210 } /* }}} int daemonize */
3212 static int cleanup (void) /* {{{ */
3214 pthread_cond_broadcast (&flush_cond);
3215 pthread_join (flush_thread, NULL);
3217 pthread_cond_broadcast (&queue_cond);
3218 for (int i = 0; i < config_queue_threads; i++)
3219 pthread_join (queue_threads[i], NULL);
3221 if (config_flush_at_shutdown)
3223 assert(cache_queue_head == NULL);
3224 RRDD_LOG(LOG_INFO, "clean shutdown; all RRDs flushed");
3227 free(queue_threads);
3228 free(config_base_dir);
3230 pthread_mutex_lock(&cache_lock);
3231 g_tree_destroy(cache_tree);
3233 pthread_mutex_lock(&journal_lock);
3236 RRDD_LOG(LOG_INFO, "goodbye");
3240 free(config_pid_file);
3243 } /* }}} int cleanup */
3245 static int read_options (int argc, char **argv) /* {{{ */
3250 socket_permission_clear (&default_socket);
3252 default_socket.socket_group = (gid_t)-1;
3253 default_socket.socket_permissions = (mode_t)-1;
3255 while ((option = getopt(argc, argv, "Ogl:s:m:P:f:w:z:t:Bb:p:Fj:a:h?")) != -1)
3260 opt_no_overwrite = 1;
3269 listen_socket_t *new;
3271 new = malloc(sizeof(listen_socket_t));
3274 fprintf(stderr, "read_options: malloc failed.\n");
3277 memset(new, 0, sizeof(listen_socket_t));
3279 strncpy(new->addr, optarg, sizeof(new->addr)-1);
3281 /* Add permissions to the socket {{{ */
3282 if (default_socket.permissions != 0)
3284 socket_permission_copy (new, &default_socket);
3286 else /* if (default_socket.permissions == 0) */
3288 /* Add permission for ALL commands to the socket. */
3289 socket_permission_set_all (new);
3291 /* }}} Done adding permissions. */
3293 new->socket_group = default_socket.socket_group;
3294 new->socket_permissions = default_socket.socket_permissions;
3296 if (!rrd_add_ptr((void ***)&config_listen_address_list,
3297 &config_listen_address_list_len, new))
3299 fprintf(stderr, "read_options: rrd_add_ptr failed.\n");
3305 /* set socket group permissions */
3311 group_gid = strtoul(optarg, NULL, 10);
3312 if (errno != EINVAL && group_gid>0)
3314 /* we were passed a number */
3315 grp = getgrgid(group_gid);
3319 grp = getgrnam(optarg);
3324 default_socket.socket_group = grp->gr_gid;
3328 /* no idea what the user wanted... */
3329 fprintf (stderr, "read_options: couldn't map \"%s\" to a group, Sorry\n", optarg);
3335 /* set socket file permissions */
3339 char *endptr = NULL;
3341 tmp = strtol (optarg, &endptr, 8);
3342 if ((endptr == optarg) || (! endptr) || (*endptr != '\0')
3343 || (tmp > 07777) || (tmp < 0)) {
3344 fprintf (stderr, "read_options: Invalid file mode \"%s\".\n",
3349 default_socket.socket_permissions = (mode_t)tmp;
3360 socket_permission_clear (&default_socket);
3362 optcopy = strdup (optarg);
3365 while ((ptr = strtok_r (dummy, ", ", &saveptr)) != NULL)
3368 status = socket_permission_add (&default_socket, ptr);
3371 fprintf (stderr, "read_options: Adding permission \"%s\" to "
3372 "socket failed. Most likely, this permission doesn't "
3373 "exist. Check your command line.\n", ptr);
3386 temp = atoi (optarg);
3388 config_flush_interval = temp;
3391 fprintf (stderr, "Invalid flush interval: %s\n", optarg);
3401 temp = atoi (optarg);
3403 config_write_interval = temp;
3406 fprintf (stderr, "Invalid write interval: %s\n", optarg);
3416 temp = atoi(optarg);
3418 config_write_jitter = temp;
3421 fprintf (stderr, "Invalid write jitter: -z %s\n", optarg);
3431 threads = atoi(optarg);
3433 config_queue_threads = threads;
3436 fprintf (stderr, "Invalid thread count: -t %s\n", optarg);
3443 config_write_base_only = 1;
3449 char base_realpath[PATH_MAX];
3451 if (config_base_dir != NULL)
3452 free (config_base_dir);
3453 config_base_dir = strdup (optarg);
3454 if (config_base_dir == NULL)
3456 fprintf (stderr, "read_options: strdup failed.\n");
3460 if (rrd_mkdir_p (config_base_dir, 0777) != 0)
3462 fprintf (stderr, "Failed to create base directory '%s': %s\n",
3463 config_base_dir, rrd_strerror (errno));
3467 /* make sure that the base directory is not resolved via
3468 * symbolic links. this makes some performance-enhancing
3469 * assumptions possible (we don't have to resolve paths
3470 * that start with a "/")
3472 if (realpath(config_base_dir, base_realpath) == NULL)
3474 fprintf (stderr, "Failed to canonicalize the base directory '%s': "
3475 "%s\n", config_base_dir, rrd_strerror(errno));
3479 len = strlen (config_base_dir);
3480 while ((len > 0) && (config_base_dir[len - 1] == '/'))
3482 config_base_dir[len - 1] = 0;
3488 fprintf (stderr, "Invalid base directory: %s\n", optarg);
3492 _config_base_dir_len = len;
3494 len = strlen (base_realpath);
3495 while ((len > 0) && (base_realpath[len - 1] == '/'))
3497 base_realpath[len - 1] = '\0';
3501 if (strncmp(config_base_dir,
3502 base_realpath, sizeof(base_realpath)) != 0)
3505 "Base directory (-b) resolved via file system links!\n"
3506 "Please consult rrdcached '-b' documentation!\n"
3507 "Consider specifying the real directory (%s)\n",
3516 if (config_pid_file != NULL)
3517 free (config_pid_file);
3518 config_pid_file = strdup (optarg);
3519 if (config_pid_file == NULL)
3521 fprintf (stderr, "read_options: strdup failed.\n");
3528 config_flush_at_shutdown = 1;
3533 char journal_dir_actual[PATH_MAX];
3534 journal_dir = realpath((const char *)optarg, journal_dir_actual);
3537 // if we were able to properly resolve the path, lets have a copy
3538 // for use outside this block.
3539 journal_dir = strdup(journal_dir);
3540 status = rrd_mkdir_p(journal_dir, 0777);
3543 fprintf(stderr, "Failed to create journal directory '%s': %s\n",
3544 journal_dir, rrd_strerror(errno));
3547 if (access(journal_dir, R_OK|W_OK|X_OK) != 0)
3549 fprintf(stderr, "Must specify a writable directory with -j! (%s)\n",
3550 errno ? rrd_strerror(errno) : "");
3554 fprintf(stderr, "Unable to resolve journal path (%s,%s)\n", optarg,
3555 errno ? rrd_strerror(errno) : "");
3563 int temp = atoi(optarg);
3565 config_alloc_chunk = temp;
3568 fprintf(stderr, "Invalid allocation size: %s\n", optarg);
3576 printf ("RRDCacheD %s\n"
3577 "Copyright (C) 2008,2009 Florian octo Forster and Kevin Brintnall\n"
3579 "Usage: rrdcached [options]\n"
3581 "Valid options are:\n"
3582 " -l <address> Socket address to listen to.\n"
3583 " Default: "RRDCACHED_DEFAULT_ADDRESS"\n"
3584 " -P <perms> Sets the permissions to assign to all following "
3586 " -w <seconds> Interval in which to write data.\n"
3587 " -z <delay> Delay writes up to <delay> seconds to spread load\n"
3588 " -t <threads> Number of write threads.\n"
3589 " -f <seconds> Interval in which to flush dead data.\n"
3590 " -p <file> Location of the PID-file.\n"
3591 " -b <dir> Base directory to change to.\n"
3592 " -B Restrict file access to paths within -b <dir>\n"
3593 " -g Do not fork and run in the foreground.\n"
3594 " -j <dir> Directory in which to create the journal files.\n"
3595 " -F Always flush all updates at shutdown\n"
3596 " -s <id|name> Group owner of all following UNIX sockets\n"
3597 " (the socket will also have read/write permissions "
3599 " -m <mode> File permissions (octal) of all following UNIX "
3601 " -a <size> Memory allocation chunk size. Default is 1.\n"
3602 " -O Do not allow CREATE commands to overwrite existing\n"
3603 " files, even if asked to.\n"
3605 "For more information and a detailed description of all options "
3607 "to the rrdcached(1) manual page.\n",
3614 } /* switch (option) */
3615 } /* while (getopt) */
3617 /* advise the user when values are not sane */
3618 if (config_flush_interval < 2 * config_write_interval)
3619 fprintf(stderr, "WARNING: flush interval (-f) should be at least"
3620 " 2x write interval (-w) !\n");
3621 if (config_write_jitter > config_write_interval)
3622 fprintf(stderr, "WARNING: write delay (-z) should NOT be larger than"
3623 " write interval (-w) !\n");
3625 if (config_write_base_only && config_base_dir == NULL)
3626 fprintf(stderr, "WARNING: -B does not make sense without -b!\n"
3627 " Consult the rrdcached documentation\n");
3629 if (journal_dir == NULL)
3630 config_flush_at_shutdown = 1;
3633 } /* }}} int read_options */
3635 int main (int argc, char **argv)
3639 status = read_options (argc, argv);
3647 status = daemonize ();
3650 fprintf (stderr, "rrdcached: daemonize failed, exiting.\n");
3656 /* start the queue threads */
3657 queue_threads = calloc(config_queue_threads, sizeof(*queue_threads));
3658 if (queue_threads == NULL)
3660 RRDD_LOG (LOG_ERR, "FATAL: cannot calloc queue threads");
3664 for (int i = 0; i < config_queue_threads; i++)
3666 memset (&queue_threads[i], 0, sizeof (*queue_threads));
3667 status = pthread_create (&queue_threads[i], NULL, queue_thread_main, NULL);
3670 RRDD_LOG (LOG_ERR, "FATAL: cannot create queue thread");
3676 /* start the flush thread */
3677 memset(&flush_thread, 0, sizeof(flush_thread));
3678 status = pthread_create (&flush_thread, NULL, flush_thread_main, NULL);
3681 RRDD_LOG (LOG_ERR, "FATAL: cannot create flush thread");
3686 listen_thread_main (NULL);
3693 * vim: set sw=2 sts=2 ts=8 et fdm=marker :