X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_daemon.c;h=154f0dee366bff3208a5a1ba9b6cc2d20664202c;hp=e2a1813a1efa3c3ffd0cf0ce449ce0452e1422b9;hb=116616fe64adf308a6bd11677245cca0bc8c5e05;hpb=46a1f600e785a8cf32fa5a4482363afe6a3e736f diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c index e2a1813..154f0de 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,2009 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 @@ -75,19 +75,27 @@ #include "rrd_client.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 @@ -97,11 +105,18 @@ #include #include #include +#include +#include #include /* }}} */ -#define RRDD_LOG(severity, ...) syslog ((severity), __VA_ARGS__) +#define RRDD_LOG(severity, ...) \ + do { \ + if (stay_foreground) \ + fprintf(stderr, __VA_ARGS__); \ + syslog ((severity), __VA_ARGS__); \ + } while (0) #ifndef __GNUC__ # define __attribute__(x) /**/ @@ -110,12 +125,6 @@ /* * Types */ -typedef enum -{ - PRIV_LOW, - PRIV_HIGH -} socket_privilege; - typedef enum { RESP_ERR = -1, RESP_OK = 0 } response_code; struct listen_socket_s @@ -123,7 +132,6 @@ struct listen_socket_s int fd; char addr[PATH_MAX + 1]; int family; - socket_privilege privilege; /* state for BATCH processing */ time_t batch_start; @@ -136,16 +144,46 @@ 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 *sock __attribute__((unused)),\ + time_t now __attribute__((unused)),\ + char *buffer __attribute__((unused)),\ + size_t buffer_size __attribute__((unused)) + +#define HANDLER_PROTO command_t *cmd __attribute__((unused)),\ + 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; time_t last_flush_time; time_t last_update_stamp; #define CI_FLAGS_IN_TREE (1<<0) @@ -172,6 +210,12 @@ enum queue_side_e }; typedef enum queue_side_e queue_side_t; +/* describe a set of journal files */ +typedef struct { + char **files; + size_t files_num; +} journal_set; + /* max length of socket command or response */ #define CMD_MAX 4096 #define RBUF_SIZE (CMD_MAX*2) @@ -185,12 +229,21 @@ static uid_t daemon_uid; static listen_socket_t *listen_fds = NULL; static size_t listen_fds_num = 0; -static int do_shutdown = 0; +enum { + RUNNING, /* normal operation */ + FLUSHING, /* flushing remaining values */ + SHUTDOWN /* shutting down */ +} state = RUNNING; + +static pthread_t *queue_threads; +static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER; +static int config_queue_threads = 4; -static pthread_t queue_thread; +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 */ @@ -198,7 +251,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; @@ -210,7 +262,7 @@ static size_t _config_base_dir_len = 0; static int config_write_base_only = 0; 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; @@ -222,22 +274,31 @@ static uint64_t stats_journal_rotate = 0; static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER; /* 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); + state = FLUSHING; + pthread_cond_broadcast(&flush_cond); + pthread_cond_broadcast(&queue_cond); } /* }}} void sig_common */ static void sig_int_handler (int s __attribute__((unused))) /* {{{ */ @@ -298,12 +359,32 @@ static void install_signal_handlers(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"; + /* 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, "rrdcached: can't %s pid file '%s' (%s)\n", @@ -341,7 +422,13 @@ static int check_pidfile(void) } lseek(pid_fd, 0, SEEK_SET); - ftruncate(pid_fd, 0); + 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" @@ -420,7 +507,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) /* {{{ */ @@ -429,7 +516,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"); @@ -451,12 +538,12 @@ static int add_response_info(listen_socket_t *sock, char *fmt, ...) /* {{{ */ char buffer[CMD_MAX]; int len; - if (sock == NULL) return 0; /* journal replay mode */ + 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 @@ -498,7 +585,7 @@ static int send_response (listen_socket_t *sock, response_code rc, ssize_t wrote; int rclen, len; - if (sock == NULL) return rc; /* journal replay mode */ + if (JOURNAL_REPLAY(sock)) return rc; if (sock->batch_start) { @@ -514,7 +601,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 @@ -563,7 +650,7 @@ static void wipe_ci_values(cache_item_t *ci, time_t when) 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 @@ -587,6 +674,12 @@ 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 */ /* free the resources associated with the cache_item_t @@ -598,7 +691,7 @@ static void *free_cache_item(cache_item_t *ci) /* {{{ */ 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); @@ -606,6 +699,7 @@ static void *free_cache_item(cache_item_t *ci) /* {{{ */ /* in case anyone is waiting */ pthread_cond_broadcast(&ci->flushed); + pthread_cond_destroy(&ci->flushed); free (ci); @@ -663,7 +757,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); @@ -673,7 +767,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, /* {{{ */ @@ -688,33 +782,20 @@ static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */ if (ci->flags & CI_FLAGS_IN_QUEUE) return FALSE; - if ((ci->last_flush_time <= cfd->abs_timeout) - && (ci->values_num > 0)) - { - enqueue_cache_item (ci, TAIL); - } - else if ((do_shutdown != 0) - && (ci->values_num > 0)) + 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->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); @@ -745,9 +826,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( g_tree_remove(cache_tree, cfd.keys[k]) == TRUE ); + assert(status == TRUE); } if (cfd.keys != NULL) @@ -759,40 +841,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 *args __attribute__((unused))) /* {{{ */ { 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. */ - 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); @@ -800,28 +876,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 *args __attribute__((unused))) /* {{{ */ +{ + 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) @@ -846,15 +943,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: " @@ -863,13 +955,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) { @@ -879,26 +972,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 */ @@ -975,7 +1054,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; @@ -1018,20 +1097,6 @@ static void get_abs_path(char **filename, char *tmp) *filename = tmp; } /* }}} static int get_abs_path */ -/* 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) -{ - if (sock == NULL) /* journal replay */ - return 1; - - if (sock->privilege >= priv) - return 1; - - return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(EACCES)); -} /* }}} static int has_privilege */ - static int flush_file (const char *filename) /* {{{ */ { cache_item_t *ci; @@ -1060,141 +1125,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" - " =