src/rrd_daemon.c: Implement PID-file and basedir options.
[rrdtool.git] / src / rrd_daemon.c
index d738c73..ac26a27 100644 (file)
@@ -116,6 +116,13 @@ struct cache_item_s
   cache_item_t *next;
 };
 
+enum queue_side_e
+{
+  HEAD,
+  TAIL
+};
+typedef enum queue_side_e queue_side_t;
+
 /*
  * Variables
  */
@@ -137,8 +144,12 @@ 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 pthread_cond_t  flush_cond = PTHREAD_COND_INITIALIZER;
+
 static int config_write_interval = 300;
 static int config_flush_interval = 3600;
+static char *config_pid_file = NULL;
+static char *config_base_dir = NULL;
 
 static char **config_listen_address_list = NULL;
 static int config_listen_address_list_len = 0;
@@ -156,11 +167,52 @@ static void sig_term_handler (int s __attribute__((unused))) /* {{{ */
   do_shutdown++;
 } /* }}} void sig_term_handler */
 
+static int write_pidfile (void) /* {{{ */
+{
+  pid_t pid;
+  char *file;
+  FILE *fh;
+
+  pid = getpid ();
+  
+  file = (config_pid_file != NULL)
+    ? config_pid_file
+    : LOCALSTATEDIR "/run/rrdcached.pid";
+
+  fh = fopen (file, "w");
+  if (fh == NULL)
+  {
+    RRDD_LOG (LOG_ERR, "write_pidfile: Opening `%s' failed.", file);
+    return (-1);
+  }
+
+  fprintf (fh, "%i\n", (int) pid);
+  fclose (fh);
+
+  return (0);
+} /* }}} int write_pidfile */
+
+static int remove_pidfile (void) /* {{{ */
+{
+  char *file;
+  int status;
+
+  file = (config_pid_file != NULL)
+    ? config_pid_file
+    : LOCALSTATEDIR "/run/rrdcached.pid";
+
+  status = unlink (file);
+  if (status == 0)
+    return (0);
+  return (errno);
+} /* }}} int remove_pidfile */
+
 /*
  * enqueue_cache_item:
  * `cache_lock' must be acquired before calling this function!
  */
-static int enqueue_cache_item (cache_item_t *ci) /* {{{ */
+static int enqueue_cache_item (cache_item_t *ci, /* {{{ */
+    queue_side_t side)
 {
   RRDD_LOG (LOG_DEBUG, "enqueue_cache_item: Adding %s to the update queue.",
       ci->file);
@@ -168,21 +220,58 @@ static int enqueue_cache_item (cache_item_t *ci) /* {{{ */
   if (ci == NULL)
     return (-1);
 
-  if ((ci->flags & CI_FLAGS_IN_QUEUE) != 0)
-    return (-1);
-
-  assert (ci->next == NULL);
-
   if (ci->values_num == 0)
     return (0);
 
-  if (cache_queue_tail == NULL)
-    cache_queue_head = ci;
-  else
-    cache_queue_tail->next = ci;
-  cache_queue_tail = ci;
+  if (side == HEAD)
+  {
+    if ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
+    {
+      assert (ci->next == NULL);
+      ci->next = cache_queue_head;
+      cache_queue_head = ci;
 
-  ci->flags |= CI_FLAGS_IN_QUEUE;
+      if (cache_queue_tail == NULL)
+        cache_queue_tail = cache_queue_head;
+    }
+    else if (cache_queue_head == ci)
+    {
+      /* do nothing */
+    }
+    else /* enqueued, but not first entry */
+    {
+      cache_item_t *prev;
+
+      /* find previous entry */
+      for (prev = cache_queue_head; prev != NULL; prev = prev->next)
+        if (prev->next == ci)
+          break;
+      assert (prev != NULL);
+
+      /* move to the front */
+      prev->next = ci->next;
+      ci->next = cache_queue_head;
+      cache_queue_head = ci;
+
+      /* check if we need to adapt the tail */
+      if (cache_queue_tail == ci)
+        cache_queue_tail = prev;
+    }
+  }
+  else /* (side == TAIL) */
+  {
+    /* We don't move values back in the list.. */
+    if ((ci->flags & CI_FLAGS_IN_QUEUE) != 0)
+      return (0);
+
+    assert (ci->next == NULL);
+
+    if (cache_queue_tail == NULL)
+      cache_queue_head = ci;
+    else
+      cache_queue_tail->next = ci;
+    cache_queue_tail = ci;
+  }
 
   ci->flags |= CI_FLAGS_IN_QUEUE;
 
@@ -208,7 +297,7 @@ static gboolean tree_callback_flush (gpointer key /* {{{ */
   if (((now - ci->last_flush_time) >= config_write_interval)
       && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
       && (ci->values_num > 0))
-    enqueue_cache_item (ci);
+    enqueue_cache_item (ci, TAIL);
 
   return (TRUE);
 } /* }}} gboolean tree_callback_flush */
@@ -310,6 +399,7 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
       free (values[i]);
 
     pthread_mutex_lock (&cache_lock);
+    pthread_cond_broadcast (&flush_cond);
   } /* while (do_shutdown == 0) */
   pthread_mutex_unlock (&cache_lock);
 
@@ -378,6 +468,79 @@ static int buffer_get_field (char **buffer_ret, /* {{{ */
   return (0);
 } /* }}} int buffer_get_field */
 
+static int flush_file (const char *filename) /* {{{ */
+{
+  cache_item_t *ci;
+
+  pthread_mutex_lock (&cache_lock);
+
+  ci = g_tree_lookup (cache_tree, filename);
+  if (ci == NULL)
+  {
+    pthread_mutex_unlock (&cache_lock);
+    return (ENOENT);
+  }
+
+  /* Enqueue at head */
+  enqueue_cache_item (ci, HEAD);
+  pthread_cond_signal (&cache_cond);
+
+  while ((ci->flags & CI_FLAGS_IN_QUEUE) != 0)
+  {
+    ci = NULL;
+
+    pthread_cond_wait (&flush_cond, &cache_lock);
+
+    ci = g_tree_lookup (cache_tree, filename);
+    if (ci == NULL)
+    {
+      RRDD_LOG (LOG_ERR, "flush_file: Tree node went away "
+          "while waiting for flush.");
+      pthread_mutex_unlock (&cache_lock);
+      return (-1);
+    }
+  }
+
+  pthread_mutex_unlock (&cache_lock);
+  return (0);
+} /* }}} int flush_file */
+
+static int handle_request_flush (int fd, /* {{{ */
+    char *buffer, size_t buffer_size)
+{
+  char *file;
+  int status;
+  char result[4096];
+
+  status = buffer_get_field (&buffer, &buffer_size, &file);
+  if (status != 0)
+  {
+    RRDD_LOG (LOG_INFO, "handle_request_flush: Cannot get file name.");
+    return (-1);
+  }
+
+  status = flush_file (file);
+  if (status == 0)
+    snprintf (result, sizeof (result), "0 Successfully flushed %s.\n", file);
+  else if (status == ENOENT)
+    snprintf (result, sizeof (result), "-1 No such file: %s.\n", file);
+  else if (status < 0)
+    strncpy (result, "-1 Internal error.\n", sizeof (result));
+  else
+    snprintf (result, sizeof (result), "-1 Failed with status %i.\n", status);
+  result[sizeof (result) - 1] = 0;
+
+  status = write (fd, result, strlen (result));
+  if (status < 0)
+  {
+    status = errno;
+    RRDD_LOG (LOG_INFO, "handle_request_flush: write(2) returned an error.");
+    return (status);
+  }
+
+  return (0);
+} /* }}} int handle_request_flush */
+
 static int handle_request_update (int fd, /* {{{ */
     char *buffer, size_t buffer_size)
 {
@@ -470,7 +633,7 @@ static int handle_request_update (int fd, /* {{{ */
       && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
       && (ci->values_num > 0))
   {
-    enqueue_cache_item (ci);
+    enqueue_cache_item (ci, TAIL);
     pthread_cond_signal (&cache_cond);
   }
 
@@ -499,7 +662,11 @@ static int handle_request (int fd) /* {{{ */
   int status;
 
   status = read (fd, buffer, sizeof (buffer));
-  if (status < 1)
+  if (status == 0)
+  {
+    return (1);
+  }
+  else if (status < 0)
   {
     RRDD_LOG (LOG_ERR, "handle_request: read(2) failed.");
     return (-1);
@@ -512,6 +679,14 @@ static int handle_request (int fd) /* {{{ */
     RRDD_LOG (LOG_INFO, "handle_request: malformed request.");
     return (-1);
   }
+
+  /* Accept Windows style line endings, too */
+  if ((buffer_size > 2) && (buffer[buffer_size - 2] == '\r'))
+  {
+    buffer_size--;
+    buffer[buffer_size - 1] = '\n';
+  }
+
   /* Place the normal field separator at the end to simplify
    * `buffer_get_field's work. */
   buffer[buffer_size - 1] = ' ';
@@ -529,6 +704,10 @@ static int handle_request (int fd) /* {{{ */
   {
     return (handle_request_update (fd, buffer_ptr, buffer_size));
   }
+  else if (strcmp (command, "flush") == 0)
+  {
+    return (handle_request_flush (fd, buffer_ptr, buffer_size));
+  }
   else
   {
     RRDD_LOG (LOG_INFO, "handle_request: unknown command: %s.", buffer);
@@ -709,7 +888,7 @@ static int open_listen_socket (const char *addr) /* {{{ */
   ai_hints.ai_socktype = SOCK_STREAM;
 
   ai_res = NULL;
-  status = getaddrinfo (addr, DEFAULT_PORT, &ai_hints, &ai_res);
+  status = getaddrinfo (addr, RRDCACHED_DEFAULT_PORT, &ai_hints, &ai_res);
   if (status != 0)
   {
     RRDD_LOG (LOG_ERR, "open_listen_socket: getaddrinfo(%s) failed: "
@@ -797,7 +976,7 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
   }
 
   if (config_listen_address_list_len < 1)
-    open_listen_socket (RRDD_SOCK_PATH);
+    open_listen_socket (RRDCACHED_DEFAULT_ADDRESS);
 
   if (listen_fds_num < 1)
   {
@@ -919,7 +1098,15 @@ static int daemonize (void) /* {{{ */
   }
 
   /* Change into the /tmp directory. */
-  chdir ("/tmp");
+  base_dir = (config_base_dir != NULL)
+    ? config_base_dir
+    : "/tmp";
+  status = chdir (base_dir);
+  if (status != 0)
+  {
+    fprintf (stderr, "daemonize: chdir (%s) failed.\n", base_dir);
+    return (-1);
+  }
 
   /* Become session leader */
   setsid ();
@@ -967,6 +1154,8 @@ static int daemonize (void) /* {{{ */
     return (-1);
   }
 
+  write_pidfile ();
+
   return (0);
 } /* }}} int daemonize */
 
@@ -981,6 +1170,8 @@ static int cleanup (void) /* {{{ */
   pthread_join (queue_thread, /* return = */ NULL);
   RRDD_LOG (LOG_DEBUG, "cleanup: done");
 
+  remove_pidfile ();
+
   closelog ();
 
   return (0);
@@ -991,7 +1182,7 @@ static int read_options (int argc, char **argv) /* {{{ */
   int option;
   int status = 0;
 
-  while ((option = getopt(argc, argv, "l:f:w:h?")) != -1)
+  while ((option = getopt(argc, argv, "l:f:w:b:p:h?")) != -1)
   {
     switch (option)
     {
@@ -1048,6 +1239,47 @@ static int read_options (int argc, char **argv) /* {{{ */
       }
       break;
 
+      case 'b':
+      {
+        size_t len;
+
+        if (config_base_dir != NULL)
+          free (config_base_dir);
+        config_base_dir = strdup (optarg);
+        if (config_base_dir == NULL)
+        {
+          fprintf (stderr, "read_options: strdup failed.\n");
+          return (3);
+        }
+
+        len = strlen (config_base_dir);
+        while ((len > 0) && (config_base_dir[len - 1] == '/'))
+        {
+          config_base_dir[len - 1] = 0;
+          len--;
+        }
+
+        if (len < 1)
+        {
+          fprintf (stderr, "Invalid base directory: %s\n", optarg);
+          return (4);
+        }
+      }
+      break;
+
+      case 'p':
+      {
+        if (config_pid_file != NULL)
+          free (config_pid_file);
+        config_pid_file = strdup (optarg);
+        if (config_pid_file == NULL)
+        {
+          fprintf (stderr, "read_options: strdup failed.\n");
+          return (3);
+        }
+      }
+      break;
+
       case 'h':
       case '?':
         printf ("RRDd %s  Copyright (C) 2008 Florian octo Forster\n"
@@ -1058,6 +1290,8 @@ static int read_options (int argc, char **argv) /* {{{ */
             "  -l <address>  Socket address to listen to.\n"
             "  -w <seconds>  Interval in which to write data.\n"
             "  -f <seconds>  Interval in which to flush dead data.\n"
+            "  -p <file>     Location of the PID-file.\n"
+            "  -b <dir>      Base directory to change to.\n"
             "\n"
             "For more information and a detailed description of all options "
             "please refer\n"