X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_daemon.c;h=36d418b1c0b63cedf787a596e6081d3913e367d0;hp=e2726e39ebd9eb9257446c0689d00412e5f09b99;hb=6d9b8ecd5a3e2f9ce9df0f46d0818bc36a93ec5c;hpb=a14a7d3d517356874606e9973144ac3f6d15bd97

diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c
index e2726e3..36d418b 100644
--- a/src/rrd_daemon.c
+++ b/src/rrd_daemon.c
@@ -170,6 +170,7 @@ typedef enum queue_side_e queue_side_t;
  * Variables
  */
 static int stay_foreground = 0;
+static uid_t daemon_uid;
 
 static listen_socket_t *listen_fds = NULL;
 static size_t listen_fds_num = 0;
@@ -539,6 +540,34 @@ static void remove_from_queue(cache_item_t *ci) /* {{{ */
   ci->flags &= ~CI_FLAGS_IN_QUEUE;
 } /* }}} 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)
+{
+  cache_item_t *ci;
+
+  ci = g_tree_lookup(cache_tree, file);
+  if (ci == NULL)
+    return ENOENT;
+
+  g_tree_remove (cache_tree, file);
+  remove_from_queue(ci);
+
+  for (int i=0; i < ci->values_num; i++)
+    free(ci->values[i]);
+
+  free (ci->values);
+  free (ci->file);
+
+  /* in case anyone is waiting */
+  pthread_cond_broadcast(&ci->flushed);
+
+  free (ci);
+
+  return 0;
+} /* }}} static int forget_file */
+
 /*
  * enqueue_cache_item:
  * `cache_lock' must be acquired before calling this function!
@@ -673,26 +702,10 @@ static int flush_old_values (int max_age)
 
   for (k = 0; k < cfd.keys_num; k++)
   {
-    cache_item_t *ci;
-
-    /* This must not fail. */
-    ci = (cache_item_t *) g_tree_lookup (cache_tree, cfd.keys[k]);
-    assert (ci != NULL);
-
-    /* If we end up here with values available, something's seriously
-     * messed up. */
-    assert (ci->values_num == 0);
-
-    /* Remove the node from the tree */
-    g_tree_remove (cache_tree, cfd.keys[k]);
-    cfd.keys[k] = NULL;
-
-    /* Now free and clean up `ci'. */
-    free (ci->file);
-    ci->file = NULL;
-    free (ci);
-    ci = NULL;
-  } /* for (k = 0; k < cfd.keys_num; k++) */
+    /* should never fail, since we have held the cache_lock
+     * the entire time */
+    assert( forget_file(cfd.keys[k]) == 0 );
+  }
 
   if (cfd.keys != NULL)
   {
@@ -942,6 +955,20 @@ 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)
+{
+  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;
@@ -962,6 +989,9 @@ static int flush_file (const char *filename) /* {{{ */
     pthread_cond_wait(&ci->flushed, &cache_lock);
   }
 
+  /* DO NOT DO ANYTHING WITH ci HERE!!  The entry
+   * may have been purged during our cond_wait() */
+
   pthread_mutex_unlock(&cache_lock);
 
   return (0);
@@ -978,9 +1008,11 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */
   {
     "Command overview\n"
     ,
+    "HELP [<command>]\n"
     "FLUSH <filename>\n"
     "FLUSHALL\n"
-    "HELP [<command>]\n"
+    "PENDING <filename>\n"
+    "FORGET <filename>\n"
     "UPDATE <filename> <values> [<values> ...]\n"
     "BATCH\n"
     "STATS\n"
@@ -1005,6 +1037,26 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */
     "Triggers writing of all pending updates.  Returns immediately.\n"
   };
 
+  char *help_pending[2] =
+  {
+    "Help for PENDING\n"
+    ,
+    "Usage: PENDING <filename>\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 <filename>\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"
@@ -1063,6 +1115,10 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */
       help_text = help_flush;
     else if (strcasecmp (command, "flushall") == 0)
       help_text = help_flushall;
+    else if (strcasecmp (command, "pending") == 0)
+      help_text = help_pending;
+    else if (strcasecmp (command, "forget") == 0)
+      help_text = help_forget;
     else if (strcasecmp (command, "stats") == 0)
       help_text = help_stats;
     else if (strcasecmp (command, "batch") == 0)
@@ -1168,6 +1224,11 @@ static int handle_request_flush (listen_socket_t *sock, /* {{{ */
 
 static int handle_request_flushall(listen_socket_t *sock) /* {{{ */
 {
+  int status;
+
+  status = has_privilege(sock, PRIV_HIGH);
+  if (status <= 0)
+    return status;
 
   RRDD_LOG(LOG_DEBUG, "Received FLUSHALL");
 
@@ -1178,18 +1239,93 @@ static int handle_request_flushall(listen_socket_t *sock) /* {{{ */
   return send_response(sock, RESP_OK, "Started flush.\n");
 } /* }}} static int handle_request_flushall */
 
+static int handle_request_pending(listen_socket_t *sock, /* {{{ */
+                                  char *buffer, size_t buffer_size)
+{
+  int status;
+  char *file;
+  cache_item_t *ci;
+
+  status = buffer_get_field(&buffer, &buffer_size, &file);
+  if (status != 0)
+    return send_response(sock, RESP_ERR,
+                         "Usage: PENDING <filename>\n");
+
+  status = has_privilege(sock, PRIV_HIGH);
+  if (status <= 0)
+    return status;
+
+  pthread_mutex_lock(&cache_lock);
+  ci = g_tree_lookup(cache_tree, file);
+  if (ci == NULL)
+  {
+    pthread_mutex_unlock(&cache_lock);
+    return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(ENOENT));
+  }
+
+  for (int i=0; i < ci->values_num; i++)
+    add_response_info(sock, "%s\n", ci->values[i]);
+
+  pthread_mutex_unlock(&cache_lock);
+  return send_response(sock, RESP_OK, "updates pending\n");
+} /* }}} static int handle_request_pending */
+
+static int handle_request_forget(listen_socket_t *sock, /* {{{ */
+                                 char *buffer, size_t buffer_size)
+{
+  int status;
+  char *file;
+
+  status = buffer_get_field(&buffer, &buffer_size, &file);
+  if (status != 0)
+    return send_response(sock, RESP_ERR,
+                         "Usage: FORGET <filename>\n");
+
+  status = has_privilege(sock, PRIV_HIGH);
+  if (status <= 0)
+    return status;
+
+  if (!check_file_access(file, sock)) return 0;
+
+  pthread_mutex_lock(&cache_lock);
+  status = forget_file(file);
+  pthread_mutex_unlock(&cache_lock);
+
+  if (status == 0)
+  {
+    if (sock != NULL)
+      journal_write("forget", file);
+
+    return send_response(sock, RESP_OK, "Gone!\n");
+  }
+  else
+    return send_response(sock, RESP_ERR, "cannot forget: %s\n",
+                         status < 0 ? "Internal error" : rrd_strerror(status));
+
+  /* NOTREACHED */
+  assert(1==0);
+} /* }}} static int handle_request_forget */
+
 static int handle_request_update (listen_socket_t *sock, /* {{{ */
     char *buffer, size_t buffer_size)
 {
   char *file;
   int values_num = 0;
   int status;
+  char orig_buf[CMD_MAX];
 
   time_t now;
   cache_item_t *ci;
 
   now = time (NULL);
 
+  status = has_privilege(sock, PRIV_HIGH);
+  if (status <= 0)
+    return status;
+
+  /* save it for the journal later */
+  strncpy(orig_buf, buffer, sizeof(orig_buf)-1);
+
   status = buffer_get_field (&buffer, &buffer_size, &file);
   if (status != 0)
     return send_response(sock, RESP_ERR,
@@ -1257,6 +1393,10 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */
   } /* }}} */
   assert (ci != NULL);
 
+  /* don't re-write updates in replay mode */
+  if (sock != NULL)
+    journal_write("update", orig_buf);
+
   while (buffer_size > 0)
   {
     char **temp;
@@ -1365,19 +1505,6 @@ static int batch_done (listen_socket_t *sock) /* {{{ */
   return send_response(sock, RESP_OK, "errors\n");
 } /* }}} static int batch_done */
 
-/* returns 1 if we have the required privilege level */
-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 */
-
 /* if sock==NULL, we are in journal replay mode */
 static int handle_request (listen_socket_t *sock, /* {{{ */
                            char *buffer, size_t buffer_size)
@@ -1401,17 +1528,7 @@ static int handle_request (listen_socket_t *sock, /* {{{ */
     sock->batch_cmd++;
 
   if (strcasecmp (command, "update") == 0)
-  {
-    status = has_privilege(sock, PRIV_HIGH);
-    if (status <= 0)
-      return status;
-
-    /* don't re-write updates in replay mode */
-    if (sock != NULL)
-      journal_write(command, buffer_ptr);
-
     return (handle_request_update (sock, buffer_ptr, buffer_size));
-  }
   else if (strcasecmp (command, "wrote") == 0 && sock == NULL)
   {
     /* this is only valid in replay mode */
@@ -1420,13 +1537,11 @@ static int handle_request (listen_socket_t *sock, /* {{{ */
   else if (strcasecmp (command, "flush") == 0)
     return (handle_request_flush (sock, buffer_ptr, buffer_size));
   else if (strcasecmp (command, "flushall") == 0)
-  {
-    status = has_privilege(sock, PRIV_HIGH);
-    if (status <= 0)
-      return status;
-
     return (handle_request_flushall(sock));
-  }
+  else if (strcasecmp (command, "pending") == 0)
+    return (handle_request_pending(sock, buffer_ptr, buffer_size));
+  else if (strcasecmp (command, "forget") == 0)
+    return (handle_request_forget(sock, buffer_ptr, buffer_size));
   else if (strcasecmp (command, "stats") == 0)
     return (handle_request_stats (sock));
   else if (strcasecmp (command, "help") == 0)
@@ -1446,6 +1561,7 @@ static int handle_request (listen_socket_t *sock, /* {{{ */
 static void journal_rotate(void) /* {{{ */
 {
   FILE *old_fh = NULL;
+  int new_fd;
 
   if (journal_cur == NULL || journal_old == NULL)
     return;
@@ -1460,11 +1576,20 @@ static void journal_rotate(void) /* {{{ */
   if (journal_fh != NULL)
   {
     old_fh = journal_fh;
+    journal_fh = NULL;
     rename(journal_cur, journal_old);
     ++stats_journal_rotate;
   }
 
-  journal_fh = fopen(journal_cur, "a");
+  new_fd = open(journal_cur, O_WRONLY|O_CREAT|O_APPEND,
+                S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
+  if (new_fd >= 0)
+  {
+    journal_fh = fdopen(new_fd, "a");
+    if (journal_fh == NULL)
+      close(new_fd);
+  }
+
   pthread_mutex_unlock(&journal_lock);
 
   if (old_fh != NULL)
@@ -1542,6 +1667,44 @@ static int journal_replay (const char *file) /* {{{ */
 
   if (file == NULL) return 0;
 
+  {
+    char *reason;
+    int status = 0;
+    struct stat statbuf;
+
+    memset(&statbuf, 0, sizeof(statbuf));
+    if (stat(file, &statbuf) != 0)
+    {
+      if (errno == ENOENT)
+        return 0;
+
+      reason = "stat error";
+      status = errno;
+    }
+    else if (!S_ISREG(statbuf.st_mode))
+    {
+      reason = "not a regular file";
+      status = EPERM;
+    }
+    if (statbuf.st_uid != daemon_uid)
+    {
+      reason = "not owned by daemon user";
+      status = EACCES;
+    }
+    if (statbuf.st_mode & (S_IWGRP|S_IWOTH))
+    {
+      reason = "must not be user/group writable";
+      status = EACCES;
+    }
+
+    if (status != 0)
+    {
+      RRDD_LOG(LOG_ERR, "journal_replay: %s : %s (%s)",
+               file, rrd_strerror(status), reason);
+      return 0;
+    }
+  }
+
   fh = fopen(file, "r");
   if (fh == NULL)
   {
@@ -1582,17 +1745,36 @@ static int journal_replay (const char *file) /* {{{ */
 
   fclose(fh);
 
-  if (entry_cnt > 0)
-  {
-    RRDD_LOG(LOG_INFO, "Replayed %d entries (%d failures)",
-             entry_cnt, fail_cnt);
-    return 1;
-  }
-  else
-    return 0;
+  RRDD_LOG(LOG_INFO, "Replayed %d entries (%d failures)",
+           entry_cnt, fail_cnt);
 
+  return entry_cnt > 0 ? 1 : 0;
 } /* }}} static int journal_replay */
 
+static void journal_init(void) /* {{{ */
+{
+  int had_journal = 0;
+
+  if (journal_cur == NULL) return;
+
+  pthread_mutex_lock(&journal_lock);
+
+  RRDD_LOG(LOG_INFO, "checking for journal files");
+
+  had_journal += journal_replay(journal_old);
+  had_journal += journal_replay(journal_cur);
+
+  /* it must have been a crash.  start a flush */
+  if (had_journal && config_flush_at_shutdown)
+    flush_old_values(-1);
+
+  pthread_mutex_unlock(&journal_lock);
+  journal_rotate();
+
+  RRDD_LOG(LOG_INFO, "journal processing complete");
+
+} /* }}} static void journal_init */
+
 static void close_connection(listen_socket_t *sock)
 {
   close(sock->fd) ;  sock->fd   = -1;
@@ -1662,23 +1844,18 @@ static void *connection_thread_main (void *args) /* {{{ */
     else if (status < 0) /* error */
     {
       status = errno;
-      if (status == EINTR)
-        continue;
-      RRDD_LOG (LOG_ERR, "connection_thread_main: poll(2) failed.");
+      if (status != EINTR)
+        RRDD_LOG (LOG_ERR, "connection_thread_main: poll(2) failed.");
       continue;
     }
 
     if ((pollfd.revents & POLLHUP) != 0) /* normal shutdown */
-    {
-      close_connection(sock);
       break;
-    }
     else if ((pollfd.revents & (POLLIN | POLLPRI)) == 0)
     {
       RRDD_LOG (LOG_WARNING, "connection_thread_main: "
           "poll(2) returned something unexpected: %#04hx",
           pollfd.revents);
-      close_connection(sock);
       break;
     }
 
@@ -2075,6 +2252,8 @@ static int daemonize (void) /* {{{ */
   int fd;
   char *base_dir;
 
+  daemon_uid = geteuid();
+
   fd = open_pidfile();
   if (fd < 0) return fd;
 
@@ -2399,25 +2578,7 @@ int main (int argc, char **argv)
     return (1);
   }
 
-  if (journal_cur != NULL)
-  {
-    int had_journal = 0;
-
-    pthread_mutex_lock(&journal_lock);
-
-    RRDD_LOG(LOG_INFO, "checking for journal files");
-
-    had_journal += journal_replay(journal_old);
-    had_journal += journal_replay(journal_cur);
-
-    if (had_journal)
-      flush_old_values(-1);
-
-    pthread_mutex_unlock(&journal_lock);
-    journal_rotate();
-
-    RRDD_LOG(LOG_INFO, "journal processing complete");
-  }
+  journal_init();
 
   /* start the queue thread */
   memset (&queue_thread, 0, sizeof (queue_thread));