X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_daemon.c;h=2c81424a34d0426d8680b751f13221b9837d268c;hp=36d418b1c0b63cedf787a596e6081d3913e367d0;hb=b63a6268ac7c3668f6731c0a0972e4575c3f6dcf;hpb=6d9b8ecd5a3e2f9ce9df0f46d0818bc36a93ec5c diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c index 36d418b..2c81424 100644 --- a/src/rrd_daemon.c +++ b/src/rrd_daemon.c @@ -1,7 +1,7 @@ /** * RRDTool - src/rrd_daemon.c - * Copyright (C) 2008 Florian octo Forster - * Copyright (C) 2008 Kevin Brintnall + * Copyright (C) 2008-2010 Florian octo Forster + * Copyright (C) 2008,2009 Kevin Brintnall * * This program is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License as published by the @@ -62,23 +62,34 @@ /* * Now for some includes.. */ -#include "rrd.h" /* {{{ */ +/* {{{ */ + +#include "rrd_tool.h" #include "rrd_client.h" +#include "unused.h" #include -#include -#include + +#ifndef WIN32 +#ifdef HAVE_STDINT_H +# include +#endif #include -#include #include -#include #include +#include + +#else + +#endif +#include +#include #include #include +#include #include #include -#include #include #include #include @@ -88,25 +99,27 @@ #include #include #include +#include +#include + +#ifdef HAVE_LIBWRAP +#include +#endif /* HAVE_LIBWRAP */ #include /* }}} */ -#define RRDD_LOG(severity, ...) syslog ((severity), __VA_ARGS__) - -#ifndef __GNUC__ -# define __attribute__(x) /**/ -#endif +#define RRDD_LOG(severity, ...) \ + do { \ + if (stay_foreground) { \ + fprintf(stderr, __VA_ARGS__); \ + fprintf(stderr, "\n"); } \ + syslog ((severity), __VA_ARGS__); \ + } while (0) /* * Types */ -typedef enum -{ - PRIV_LOW, - PRIV_HIGH -} socket_privilege; - typedef enum { RESP_ERR = -1, RESP_OK = 0 } response_code; struct listen_socket_s @@ -114,10 +127,9 @@ struct listen_socket_s int fd; char addr[PATH_MAX + 1]; int family; - socket_privilege privilege; /* state for BATCH processing */ - int batch_mode; + time_t batch_start; int batch_cmd; /* buffered IO */ @@ -127,17 +139,49 @@ struct listen_socket_s char *wbuf; ssize_t wbuf_len; + + uint32_t permissions; + + gid_t socket_group; + mode_t socket_permissions; }; typedef struct listen_socket_s listen_socket_t; +struct command_s; +typedef struct command_s command_t; +/* note: guard against "unused" warnings in the handlers */ +#define DISPATCH_PROTO listen_socket_t UNUSED(*sock),\ + time_t UNUSED(now),\ + char UNUSED(*buffer),\ + size_t UNUSED(buffer_size) + +#define HANDLER_PROTO command_t UNUSED(*cmd),\ + DISPATCH_PROTO + +struct command_s { + char *cmd; + int (*handler)(HANDLER_PROTO); + + char context; /* where we expect to see it */ +#define CMD_CONTEXT_CLIENT (1<<0) +#define CMD_CONTEXT_BATCH (1<<1) +#define CMD_CONTEXT_JOURNAL (1<<2) +#define CMD_CONTEXT_ANY (0x7f) + + char *syntax; + char *help; +}; + struct cache_item_s; typedef struct cache_item_s cache_item_t; struct cache_item_s { char *file; char **values; - int values_num; + size_t values_num; /* number of valid pointers */ + size_t values_alloc; /* number of allocated pointers */ time_t last_flush_time; + double last_update_stamp; #define CI_FLAGS_IN_TREE (1<<0) #define CI_FLAGS_IN_QUEUE (1<<1) int flags; @@ -162,9 +206,13 @@ enum queue_side_e }; typedef enum queue_side_e queue_side_t; -/* max length of socket command or response */ -#define CMD_MAX 4096 -#define RBUF_SIZE (CMD_MAX*2) +/* describe a set of journal files */ +typedef struct { + char **files; + size_t files_num; +} journal_set; + +#define RBUF_SIZE (RRD_CMD_MAX*2) /* * Variables @@ -175,12 +223,23 @@ static uid_t daemon_uid; static listen_socket_t *listen_fds = NULL; static size_t listen_fds_num = 0; -static int do_shutdown = 0; +static listen_socket_t default_socket; + +enum { + RUNNING, /* normal operation */ + FLUSHING, /* flushing remaining values */ + SHUTDOWN /* shutting down */ +} state = RUNNING; -static pthread_t queue_thread; +static pthread_t *queue_threads; +static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER; +static int config_queue_threads = 4; + +static pthread_t flush_thread; +static pthread_cond_t flush_cond = PTHREAD_COND_INITIALIZER; -static pthread_t *connection_threads = NULL; static pthread_mutex_t connection_threads_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t connection_threads_done = PTHREAD_COND_INITIALIZER; static int connection_threads_num = 0; /* Cache stuff */ @@ -188,7 +247,6 @@ static GTree *cache_tree = NULL; static cache_item_t *cache_queue_head = NULL; static cache_item_t *cache_queue_tail = NULL; static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; -static pthread_cond_t cache_cond = PTHREAD_COND_INITIALIZER; static int config_write_interval = 300; static int config_write_jitter = 0; @@ -198,9 +256,10 @@ static char *config_pid_file = NULL; static char *config_base_dir = NULL; static size_t _config_base_dir_len = 0; static int config_write_base_only = 0; +static size_t config_alloc_chunk = 1; static listen_socket_t **config_listen_address_list = NULL; -static int config_listen_address_list_len = 0; +static size_t config_listen_address_list_len = 0; static uint64_t stats_queue_length = 0; static uint64_t stats_updates_received = 0; @@ -211,42 +270,55 @@ static uint64_t stats_journal_bytes = 0; static uint64_t stats_journal_rotate = 0; static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER; +static int opt_no_overwrite = 0; /* default for the daemon */ + /* Journaled updates */ -static char *journal_cur = NULL; -static char *journal_old = NULL; -static FILE *journal_fh = NULL; +#define JOURNAL_REPLAY(s) ((s) == NULL) +#define JOURNAL_BASE "rrd.journal" +static journal_set *journal_cur = NULL; +static journal_set *journal_old = NULL; +static char *journal_dir = NULL; +static FILE *journal_fh = NULL; /* current journal file handle */ +static long journal_size = 0; /* current journal size */ +#define JOURNAL_MAX (1 * 1024 * 1024 * 1024) static pthread_mutex_t journal_lock = PTHREAD_MUTEX_INITIALIZER; static int journal_write(char *cmd, char *args); static void journal_done(void); static void journal_rotate(void); +/* prototypes for forward refernces */ +static int handle_request_help (HANDLER_PROTO); + /* * Functions */ static void sig_common (const char *sig) /* {{{ */ { RRDD_LOG(LOG_NOTICE, "caught SIG%s", sig); - do_shutdown++; - pthread_cond_broadcast(&cache_cond); + if (state == RUNNING) { + state = FLUSHING; + } + pthread_cond_broadcast(&flush_cond); + pthread_cond_broadcast(&queue_cond); } /* }}} void sig_common */ -static void sig_int_handler (int s __attribute__((unused))) /* {{{ */ +static void sig_int_handler (int UNUSED(s)) /* {{{ */ { sig_common("INT"); } /* }}} void sig_int_handler */ -static void sig_term_handler (int s __attribute__((unused))) /* {{{ */ +static void sig_term_handler (int UNUSED(s)) /* {{{ */ { sig_common("TERM"); } /* }}} void sig_term_handler */ -static void sig_usr1_handler (int s __attribute__((unused))) /* {{{ */ +static void sig_usr1_handler (int UNUSED(s)) /* {{{ */ { config_flush_at_shutdown = 1; sig_common("USR1"); } /* }}} void sig_usr1_handler */ -static void sig_usr2_handler (int s __attribute__((unused))) /* {{{ */ +static void sig_usr2_handler (int UNUSED(s)) /* {{{ */ { config_flush_at_shutdown = 0; sig_common("USR2"); @@ -285,23 +357,87 @@ static void install_signal_handlers(void) /* {{{ */ } /* }}} void install_signal_handlers */ -static int open_pidfile(void) /* {{{ */ +static int open_pidfile(char *action, int oflag) /* {{{ */ { int fd; - char *file; + const char *file; + char *file_copy, *dir; file = (config_pid_file != NULL) ? config_pid_file : LOCALSTATEDIR "/run/rrdcached.pid"; - fd = open(file, O_CREAT|O_EXCL|O_WRONLY, S_IRUSR|S_IRGRP|S_IROTH); + /* dirname may modify its argument */ + file_copy = strdup(file); + if (file_copy == NULL) + { + fprintf(stderr, "rrdcached: strdup(): %s\n", + rrd_strerror(errno)); + return -1; + } + + dir = dirname(file_copy); + if (rrd_mkdir_p(dir, 0777) != 0) + { + fprintf(stderr, "Failed to create pidfile directory '%s': %s\n", + dir, rrd_strerror(errno)); + return -1; + } + + free(file_copy); + + fd = open(file, oflag, S_IWUSR|S_IRUSR|S_IRGRP|S_IROTH); if (fd < 0) - fprintf(stderr, "FATAL: cannot create '%s' (%s)\n", - file, rrd_strerror(errno)); + fprintf(stderr, "rrdcached: can't %s pid file '%s' (%s)\n", + action, file, rrd_strerror(errno)); return(fd); } /* }}} static int open_pidfile */ +/* check existing pid file to see whether a daemon is running */ +static int check_pidfile(void) +{ + int pid_fd; + pid_t pid; + char pid_str[16]; + + pid_fd = open_pidfile("open", O_RDWR); + if (pid_fd < 0) + return pid_fd; + + if (read(pid_fd, pid_str, sizeof(pid_str)) <= 0) + return -1; + + pid = atoi(pid_str); + if (pid <= 0) + return -1; + + /* another running process that we can signal COULD be + * a competing rrdcached */ + if (pid != getpid() && kill(pid, 0) == 0) + { + fprintf(stderr, + "FATAL: Another rrdcached daemon is running?? (pid %d)\n", pid); + close(pid_fd); + return -1; + } + + lseek(pid_fd, 0, SEEK_SET); + if (ftruncate(pid_fd, 0) == -1) + { + fprintf(stderr, + "FATAL: Faild to truncate stale PID file. (pid %d)\n", pid); + close(pid_fd); + return -1; + } + + fprintf(stderr, + "rrdcached: removed stale PID file (no rrdcached on pid %d)\n" + "rrdcached: starting normally.\n", pid); + + return pid_fd; +} /* }}} static int check_pidfile */ + static int write_pidfile (int fd) /* {{{ */ { pid_t pid; @@ -372,7 +508,7 @@ static char *next_cmd (listen_socket_t *sock, ssize_t *len) /* {{{ */ /* NOTREACHED */ assert(1==0); -} +} /* }}} char *next_cmd */ /* add the characters directly to the write buffer */ static int add_to_wbuf(listen_socket_t *sock, char *str, size_t len) /* {{{ */ @@ -381,7 +517,7 @@ static int add_to_wbuf(listen_socket_t *sock, char *str, size_t len) /* {{{ */ assert(sock != NULL); - new_buf = realloc(sock->wbuf, sock->wbuf_len + len + 1); + new_buf = rrd_realloc(sock->wbuf, sock->wbuf_len + len + 1); if (new_buf == NULL) { RRDD_LOG(LOG_ERR, "add_to_wbuf: realloc failed"); @@ -400,15 +536,15 @@ static int add_to_wbuf(listen_socket_t *sock, char *str, size_t len) /* {{{ */ static int add_response_info(listen_socket_t *sock, char *fmt, ...) /* {{{ */ { va_list argp; - char buffer[CMD_MAX]; + char buffer[RRD_CMD_MAX]; int len; - if (sock == NULL) return 0; /* journal replay mode */ - if (sock->batch_mode) return 0; /* no extra info returned when in BATCH */ + if (JOURNAL_REPLAY(sock)) return 0; + if (sock->batch_start) return 0; /* no extra info returned when in BATCH */ va_start(argp, fmt); #ifdef HAVE_VSNPRINTF - len = vsnprintf(buffer, sizeof(buffer)-1, fmt, argp); + len = vsnprintf(buffer, sizeof(buffer), fmt, argp); #else len = vsprintf(buffer, fmt, argp); #endif @@ -445,14 +581,14 @@ static int send_response (listen_socket_t *sock, response_code rc, char *fmt, ...) /* {{{ */ { va_list argp; - char buffer[CMD_MAX]; + char buffer[RRD_CMD_MAX]; int lines; ssize_t wrote; int rclen, len; - if (sock == NULL) return rc; /* journal replay mode */ + if (JOURNAL_REPLAY(sock)) return rc; - if (sock->batch_mode) + if (sock->batch_start) { if (rc == RESP_OK) return rc; /* no response on success during BATCH */ @@ -466,7 +602,7 @@ static int send_response (listen_socket_t *sock, response_code rc, rclen = sprintf(buffer, "%d ", lines); va_start(argp, fmt); #ifdef HAVE_VSNPRINTF - len = vsnprintf(buffer+rclen, sizeof(buffer)-rclen-1, fmt, argp); + len = vsnprintf(buffer+rclen, sizeof(buffer)-rclen, fmt, argp); #else len = vsprintf(buffer+rclen, fmt, argp); #endif @@ -477,7 +613,7 @@ static int send_response (listen_socket_t *sock, response_code rc, len += rclen; /* append the result to the wbuf, don't write to the user */ - if (sock->batch_mode) + if (sock->batch_start) return add_to_wbuf(sock, buffer, len); /* first write must be complete */ @@ -487,7 +623,7 @@ static int send_response (listen_socket_t *sock, response_code rc, return -1; } - if (sock->wbuf != NULL) + if (sock->wbuf != NULL && rc == RESP_OK) { wrote = 0; while (wrote < sock->wbuf_len) @@ -512,10 +648,11 @@ static void wipe_ci_values(cache_item_t *ci, time_t when) { ci->values = NULL; ci->values_num = 0; + ci->values_alloc = 0; ci->last_flush_time = when; if (config_write_jitter > 0) - ci->last_flush_time += (random() % config_write_jitter); + ci->last_flush_time += (rrd_random() % config_write_jitter); } /* remove_from_queue @@ -525,6 +662,7 @@ static void wipe_ci_values(cache_item_t *ci, time_t when) static void remove_from_queue(cache_item_t *ci) /* {{{ */ { if (ci == NULL) return; + if ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) return; /* not queued */ if (ci->prev == NULL) cache_queue_head = ci->next; /* reset head */ @@ -538,23 +676,24 @@ static void remove_from_queue(cache_item_t *ci) /* {{{ */ ci->next = ci->prev = NULL; ci->flags &= ~CI_FLAGS_IN_QUEUE; + + pthread_mutex_lock (&stats_lock); + assert (stats_queue_length > 0); + stats_queue_length--; + pthread_mutex_unlock (&stats_lock); + } /* }}} static void remove_from_queue */ -/* remove an entry from the tree and free all its resources. - * must hold 'cache lock' while calling this. - * returns 0 on success, otherwise errno */ -static int forget_file(const char *file) +/* free the resources associated with the cache_item_t + * must hold cache_lock when calling this function + */ +static void *free_cache_item(cache_item_t *ci) /* {{{ */ { - cache_item_t *ci; - - ci = g_tree_lookup(cache_tree, file); - if (ci == NULL) - return ENOENT; + if (ci == NULL) return NULL; - g_tree_remove (cache_tree, file); remove_from_queue(ci); - for (int i=0; i < ci->values_num; i++) + for (size_t i=0; i < ci->values_num; i++) free(ci->values[i]); free (ci->values); @@ -562,11 +701,12 @@ static int forget_file(const char *file) /* in case anyone is waiting */ pthread_cond_broadcast(&ci->flushed); + pthread_cond_destroy(&ci->flushed); free (ci); - return 0; -} /* }}} static int forget_file */ + return NULL; +} /* }}} static void *free_cache_item */ /* * enqueue_cache_item: @@ -586,9 +726,8 @@ static int enqueue_cache_item (cache_item_t *ci, /* {{{ */ if (cache_queue_head == ci) return 0; - /* remove from the double linked list */ - if (ci->flags & CI_FLAGS_IN_QUEUE) - remove_from_queue(ci); + /* remove if further down in queue */ + remove_from_queue(ci); ci->prev = NULL; ci->next = cache_queue_head; @@ -620,7 +759,7 @@ static int enqueue_cache_item (cache_item_t *ci, /* {{{ */ ci->flags |= CI_FLAGS_IN_QUEUE; - pthread_cond_broadcast(&cache_cond); + pthread_cond_signal(&queue_cond); pthread_mutex_lock (&stats_lock); stats_queue_length++; pthread_mutex_unlock (&stats_lock); @@ -630,7 +769,7 @@ static int enqueue_cache_item (cache_item_t *ci, /* {{{ */ /* * tree_callback_flush: - * Called via `g_tree_foreach' in `queue_thread_main'. `cache_lock' is held + * Called via `g_tree_foreach' in `flush_thread_main'. `cache_lock' is held * while this is in progress. */ static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */ @@ -642,36 +781,23 @@ static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */ ci = (cache_item_t *) value; cfd = (callback_flush_data_t *) data; - if ((ci->last_flush_time <= cfd->abs_timeout) - && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) - && (ci->values_num > 0)) - { - enqueue_cache_item (ci, TAIL); - } - else if ((do_shutdown != 0) - && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) - && (ci->values_num > 0)) + if (ci->flags & CI_FLAGS_IN_QUEUE) + return FALSE; + + if (ci->values_num > 0 + && (ci->last_flush_time <= cfd->abs_timeout || state != RUNNING)) { enqueue_cache_item (ci, TAIL); } else if (((cfd->now - ci->last_flush_time) >= config_flush_interval) - && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) && (ci->values_num <= 0)) { - char **temp; - - temp = (char **) realloc (cfd->keys, - sizeof (char *) * (cfd->keys_num + 1)); - if (temp == NULL) + assert ((char *) key == ci->file); + if (!rrd_add_ptr((void ***)&cfd->keys, &cfd->keys_num, (void *)key)) { - RRDD_LOG (LOG_ERR, "tree_callback_flush: realloc failed."); + RRDD_LOG (LOG_ERR, "tree_callback_flush: rrd_add_ptrs failed."); return (FALSE); } - cfd->keys = temp; - /* Make really sure this points to the _same_ place */ - assert ((char *) key == ci->file); - cfd->keys[cfd->keys_num] = (char *) key; - cfd->keys_num++; } return (FALSE); @@ -702,9 +828,10 @@ static int flush_old_values (int max_age) for (k = 0; k < cfd.keys_num; k++) { + gboolean status = g_tree_remove(cache_tree, cfd.keys[k]); /* should never fail, since we have held the cache_lock * the entire time */ - assert( forget_file(cfd.keys[k]) == 0 ); + assert(status == TRUE); } if (cfd.keys != NULL) @@ -716,40 +843,34 @@ static int flush_old_values (int max_age) return (0); } /* int flush_old_values */ -static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ +static void *flush_thread_main (void UNUSED(*args)) /* {{{ */ { struct timeval now; struct timespec next_flush; - int final_flush = 0; /* make sure we only flush once on shutdown */ + int status; gettimeofday (&now, NULL); next_flush.tv_sec = now.tv_sec + config_flush_interval; next_flush.tv_nsec = 1000 * now.tv_usec; - pthread_mutex_lock (&cache_lock); - while ((do_shutdown == 0) || (cache_queue_head != NULL)) - { - cache_item_t *ci; - char *file; - char **values; - int values_num; - int status; - int i; + pthread_mutex_lock(&cache_lock); - /* First, check if it's time to do the cache flush. */ + while (state == RUNNING) + { gettimeofday (&now, NULL); if ((now.tv_sec > next_flush.tv_sec) || ((now.tv_sec == next_flush.tv_sec) && ((1000 * now.tv_usec) > next_flush.tv_nsec))) { + RRDD_LOG(LOG_DEBUG, "flushing old values"); + + /* Determine the time of the next cache flush. */ + next_flush.tv_sec = now.tv_sec + config_flush_interval; + /* Flush all values that haven't been written in the last * `config_write_interval' seconds. */ flush_old_values (config_write_interval); - /* Determine the time of the next cache flush. */ - while (next_flush.tv_sec <= now.tv_sec) - next_flush.tv_sec += config_flush_interval; - /* unlock the cache while we rotate so we don't block incoming * updates if the fsync() blocks on disk I/O */ pthread_mutex_unlock(&cache_lock); @@ -757,28 +878,49 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ pthread_mutex_lock(&cache_lock); } + status = pthread_cond_timedwait(&flush_cond, &cache_lock, &next_flush); + if (status != 0 && status != ETIMEDOUT) + { + RRDD_LOG (LOG_ERR, "flush_thread_main: " + "pthread_cond_timedwait returned %i.", status); + } + } + + if (config_flush_at_shutdown) + flush_old_values (-1); /* flush everything */ + + state = SHUTDOWN; + + pthread_mutex_unlock(&cache_lock); + + return NULL; +} /* void *flush_thread_main */ + +static void *queue_thread_main (void UNUSED(*args)) /* {{{ */ +{ + pthread_mutex_lock (&cache_lock); + + while (state != SHUTDOWN + || (cache_queue_head != NULL && config_flush_at_shutdown)) + { + cache_item_t *ci; + char *file; + char **values; + size_t values_num; + int status; + /* Now, check if there's something to store away. If not, wait until - * something comes in or it's time to do the cache flush. if we are - * shutting down, do not wait around. */ - if (cache_queue_head == NULL && !do_shutdown) + * something comes in. */ + if (cache_queue_head == NULL) { - status = pthread_cond_timedwait (&cache_cond, &cache_lock, &next_flush); + status = pthread_cond_wait (&queue_cond, &cache_lock); if ((status != 0) && (status != ETIMEDOUT)) { RRDD_LOG (LOG_ERR, "queue_thread_main: " - "pthread_cond_timedwait returned %i.", status); + "pthread_cond_wait returned %i.", status); } } - /* We're about to shut down */ - if (do_shutdown != 0 && !final_flush++) - { - if (config_flush_at_shutdown) - flush_old_values (-1); /* flush everything */ - else - break; - } - /* Check if a value has arrived. This may be NULL if we timed out or there * was an interrupt such as a signal. */ if (cache_queue_head == NULL) @@ -803,15 +945,10 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ wipe_ci_values(ci, time(NULL)); remove_from_queue(ci); - pthread_mutex_lock (&stats_lock); - assert (stats_queue_length > 0); - stats_queue_length--; - pthread_mutex_unlock (&stats_lock); - pthread_mutex_unlock (&cache_lock); rrd_clear_error (); - status = rrd_update_r (file, NULL, values_num, (void *) values); + status = rrd_update_r (file, NULL, (int) values_num, (void *) values); if (status != 0) { RRDD_LOG (LOG_NOTICE, "queue_thread_main: " @@ -820,13 +957,14 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ } journal_write("wrote", file); - pthread_cond_broadcast(&ci->flushed); - for (i = 0; i < values_num; i++) - free (values[i]); - - free(values); - free(file); + /* Search again in the tree. It's possible someone issued a "FORGET" + * while we were writing the update values. */ + pthread_mutex_lock(&cache_lock); + ci = (cache_item_t *) g_tree_lookup(cache_tree, file); + if (ci) + pthread_cond_broadcast(&ci->flushed); + pthread_mutex_unlock(&cache_lock); if (status == 0) { @@ -836,26 +974,12 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ pthread_mutex_unlock (&stats_lock); } - pthread_mutex_lock (&cache_lock); - - /* We're about to shut down */ - if (do_shutdown != 0 && !final_flush++) - { - if (config_flush_at_shutdown) - flush_old_values (-1); /* flush everything */ - else - break; - } - } /* while ((do_shutdown == 0) || (cache_queue_head != NULL)) */ - pthread_mutex_unlock (&cache_lock); + rrd_free_ptrs((void ***) &values, &values_num); + free(file); - if (config_flush_at_shutdown) - { - assert(cache_queue_head == NULL); - RRDD_LOG(LOG_INFO, "clean shutdown; all RRDs flushed"); + pthread_mutex_lock (&cache_lock); } - - journal_done(); + pthread_mutex_unlock (&cache_lock); return (NULL); } /* }}} void *queue_thread_main */ @@ -932,7 +1056,7 @@ static int check_file_access (const char *file, listen_socket_t *sock) /* {{{ */ assert(file != NULL); if (!config_write_base_only - || sock == NULL /* journal replay */ + || JOURNAL_REPLAY(sock) || config_base_dir == NULL) return 1; @@ -955,19 +1079,25 @@ err: return 0; } /* }}} static int check_file_access */ -/* returns 1 if we have the required privilege level, - * otherwise issue an error to the user on sock */ -static int has_privilege (listen_socket_t *sock, /* {{{ */ - socket_privilege priv) +/* when using a base dir, convert relative paths to absolute paths. + * if necessary, modifies the "filename" pointer to point + * to the new path created in "tmp". "tmp" is provided + * by the caller and sizeof(tmp) must be >= PATH_MAX. + * + * this allows us to optimize for the expected case (absolute path) + * with a no-op. + */ +static void get_abs_path(char **filename, char *tmp) { - if (sock == NULL) /* journal replay */ - return 1; + assert(tmp != NULL); + assert(filename != NULL && *filename != NULL); - if (sock->privilege >= priv) - return 1; + if (config_base_dir == NULL || **filename == '/') + return; - return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(EACCES)); -} /* }}} static int has_privilege */ + snprintf(tmp, PATH_MAX, "%s/%s", config_base_dir, *filename); + *filename = tmp; +} /* }}} static int get_abs_path */ static int flush_file (const char *filename) /* {{{ */ { @@ -997,141 +1127,17 @@ static int flush_file (const char *filename) /* {{{ */ return (0); } /* }}} int flush_file */ -static int handle_request_help (listen_socket_t *sock, /* {{{ */ - char *buffer, size_t buffer_size) +static int syntax_error(listen_socket_t *sock, command_t *cmd) /* {{{ */ { - int status; - char **help_text; - char *command; - - char *help_help[2] = - { - "Command overview\n" - , - "HELP []\n" - "FLUSH \n" - "FLUSHALL\n" - "PENDING \n" - "FORGET \n" - "UPDATE [ ...]\n" - "BATCH\n" - "STATS\n" - }; - - char *help_flush[2] = - { - "Help for FLUSH\n" - , - "Usage: FLUSH \n" - "\n" - "Adds the given filename to the head of the update queue and returns\n" - "after is has been dequeued.\n" - }; - - char *help_flushall[2] = - { - "Help for FLUSHALL\n" - , - "Usage: FLUSHALL\n" - "\n" - "Triggers writing of all pending updates. Returns immediately.\n" - }; - - char *help_pending[2] = - { - "Help for PENDING\n" - , - "Usage: PENDING \n" - "\n" - "Shows any 'pending' updates for a file, in order.\n" - "The updates shown have not yet been written to the underlying RRD file.\n" - }; - - char *help_forget[2] = - { - "Help for FORGET\n" - , - "Usage: FORGET \n" - "\n" - "Removes the file completely from the cache.\n" - "Any pending updates for the file will be lost.\n" - }; - - char *help_update[2] = - { - "Help for UPDATE\n" - , - "Usage: UPDATE [ ...]\n" - "\n" - "Adds the given file to the internal cache if it is not yet known and\n" - "appends the given value(s) to the entry. See the rrdcached(1) manpage\n" - "for details.\n" - "\n" - "Each has the following form:\n" - " =