src/rrdd.c

   1 /**
   2  * collectd - src/rrdd.c
   3  * Copyright (C) 2008 Florian octo Forster
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; only version 2 of the License is applicable.
   8  *
   9  * This program is distributed in the hope that it will be useful, but
  10  * WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  17  *
  18  * Authors:
  19  *   Florian octo Forster <octo at verplant.org>
  20  **/
  21
  22 #define RRDD_DEBUG 1
  23
  24 #include "rrdd.h"
  25
  26 #if RRDD_DEBUG
  27 # define RRDD_LOG(severity, ...) do { fprintf (stderr, __VA_ARGS__); fprintf (stderr, "\n"); } while (0)
  28 #else
  29 # define RRDD_LOG(severity, ...) syslog ((severity), __VA_ARGS__)
  30 #endif
  31
  32 /*
  33  * Types
  34  */
  35 struct listen_socket_s
  36 {
  37   int fd;
  38   char path[PATH_MAX + 1];
  39 };
  40 typedef struct listen_socket_s listen_socket_t;
  41
  42 struct cache_item_s;
  43 typedef struct cache_item_s cache_item_t;
  44 struct cache_item_s
  45 {
  46   char *file;
  47   char **values;
  48   int values_num;
  49   time_t last_flush_time;
  50 #define CI_FLAGS_IN_TREE  0x01
  51 #define CI_FLAGS_IN_QUEUE 0x02
  52   int flags;
  53
  54   cache_item_t *next;
  55 };
  56
  57 /*
  58  * Variables
  59  */
  60 static listen_socket_t *listen_fds = NULL;
  61 static size_t listen_fds_num = 0;
  62
  63 static int do_shutdown = 0;
  64
  65 static pthread_t queue_thread;
  66
  67 static pthread_t *connetion_threads = NULL;
  68 static pthread_mutex_t connetion_threads_lock = PTHREAD_MUTEX_INITIALIZER;
  69 static int connetion_threads_num = 0;
  70
  71 /* Cache stuff */
  72 static avl_tree_t     *cache_tree = NULL;
  73 static cache_item_t   *cache_queue_head = NULL;
  74 static cache_item_t   *cache_queue_tail = NULL;
  75 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
  76 static pthread_cond_t  cache_cond = PTHREAD_COND_INITIALIZER;
  77
  78 /*
  79  * Functions
  80  */
  81 static void sig_int_handler (int signal) /* {{{ */
  82 {
  83   do_shutdown++;
  84 } /* }}} void sig_int_handler */
  85
  86 static void sig_term_handler (int signal) /* {{{ */
  87 {
  88   do_shutdown++;
  89 } /* }}} void sig_term_handler */
  90
  91 static int cache_tree_compare (const void *v0, const void *v1) /* {{{ */
  92 {
  93   cache_item_t *c0 = (cache_item_t *) v0;
  94   cache_item_t *c1 = (cache_item_t *) v1;
  95
  96   assert (c0->file != NULL);
  97   assert (c1->file != NULL);
  98
  99   return (strcmp (c0->file, c1->file));
 100 } /* }}} int cache_tree_compare */
 101
 102 static void cache_tree_free (void *v) /* {{{ */
 103 {
 104   cache_item_t *c = (cache_item_t *) v;
 105
 106   assert (c->values_num == 0);
 107   assert ((c->flags & CI_FLAGS_IN_TREE) != 0);
 108   assert ((c->flags & CI_FLAGS_IN_QUEUE) == 0);
 109
 110   free (c->file);
 111   c->file = NULL;
 112   free (c);
 113 } /* }}} void cache_tree_free */
 114
 115 static void *queue_thread_main (void *args) /* {{{ */
 116 {
 117   pthread_mutex_lock (&cache_lock);
 118   while ((do_shutdown == 0) || (cache_queue_head != NULL))
 119   {
 120     cache_item_t *ci;
 121
 122     char *file;
 123     char **values;
 124     int values_num;
 125     int i;
 126
 127     if (cache_queue_head == NULL)
 128       pthread_cond_wait (&cache_cond, &cache_lock);
 129
 130     if (cache_queue_head == NULL)
 131       continue;
 132
 133     ci = cache_queue_head;
 134
 135     /* copy the relevant parts */
 136     file = strdup (ci->file);
 137     if (file == NULL)
 138     {
 139       RRDD_LOG (LOG_ERR, "queue_thread_main: strdup failed.");
 140       continue;
 141     }
 142
 143     values = ci->values;
 144     values_num = ci->values_num;
 145
 146     ci->values = NULL;
 147     ci->values_num = 0;
 148
 149     ci->last_flush_time = time (NULL);
 150     ci->flags &= ~(CI_FLAGS_IN_QUEUE);
 151
 152     cache_queue_head = ci->next;
 153     if (cache_queue_head == NULL)
 154       cache_queue_tail = NULL;
 155     ci->next = NULL;
 156
 157     pthread_mutex_unlock (&cache_lock);
 158
 159     RRDD_LOG (LOG_DEBUG, "queue_thread_main: rrd_update (%s, %i, %p)",
 160         file, values_num, (void *) values);
 161
 162     free (file);
 163     for (i = 0; i < values_num; i++)
 164       free (values[i]);
 165
 166     pthread_mutex_lock (&cache_lock);
 167   } /* while (do_shutdown == 0) */
 168   pthread_mutex_unlock (&cache_lock);
 169
 170   RRDD_LOG (LOG_DEBUG, "queue_thread_main: Exiting.");
 171
 172   return (NULL);
 173 } /* }}} void *queue_thread_main */
 174
 175 static int handle_request_update (int fd, /* {{{ */
 176     char *buffer, int buffer_size)
 177 {
 178   char *file;
 179   char *value;
 180   char *buffer_ptr;
 181   int values_num = 0;
 182
 183   time_t now;
 184
 185   avl_node_t *node;
 186   cache_item_t ci_temp;
 187   cache_item_t *ci;
 188
 189   char answer[4096];
 190
 191   now = time (NULL);
 192
 193   RRDD_LOG (LOG_DEBUG, "handle_request_update (%i, %p, %i)",
 194       fd, (void *) buffer, buffer_size);
 195
 196   buffer_ptr = buffer;
 197
 198   file = buffer_ptr;
 199   buffer_ptr += strlen (file) + 1;
 200
 201   ci_temp.file = file;
 202
 203   pthread_mutex_lock (&cache_lock);
 204
 205   node = avl_search (cache_tree, (void *) &ci_temp);
 206   if (node == NULL)
 207   {
 208     ci = (cache_item_t *) malloc (sizeof (cache_item_t));
 209     if (ci == NULL)
 210     {
 211       pthread_mutex_unlock (&cache_lock);
 212       RRDD_LOG (LOG_ERR, "handle_request_update: malloc failed.");
 213       return (-1);
 214     }
 215     memset (ci, 0, sizeof (cache_item_t));
 216
 217     ci->file = strdup (file);
 218     if (ci->file == NULL)
 219     {
 220       pthread_mutex_unlock (&cache_lock);
 221       RRDD_LOG (LOG_ERR, "handle_request_update: malloc failed.");
 222       free (ci);
 223       return (-1);
 224     }
 225
 226     ci->values = NULL;
 227     ci->values_num = 0;
 228     ci->last_flush_time = now;
 229     ci->flags = CI_FLAGS_IN_TREE;
 230
 231     if (avl_insert (cache_tree, (void *) ci) == NULL)
 232     {
 233       pthread_mutex_unlock (&cache_lock);
 234       RRDD_LOG (LOG_ERR, "handle_request_update: avl_insert failed.");
 235       free (ci->file);
 236       free (ci);
 237       return (-1);
 238     }
 239
 240     RRDD_LOG (LOG_DEBUG, "handle_request_update: Created new AVL node %s.",
 241         ci->file);
 242   }
 243   else /* if (ci != NULL) */
 244   {
 245     ci = (cache_item_t *) node->item;
 246   }
 247   assert (ci != NULL);
 248
 249   while (*buffer_ptr != 0)
 250   {
 251     char **temp;
 252
 253     value = buffer_ptr;
 254     buffer_ptr += strlen (value) + 1;
 255
 256     temp = (char **) realloc (ci->values,
 257         sizeof (char *) * (ci->values_num + 1));
 258     if (temp == NULL)
 259     {
 260       RRDD_LOG (LOG_ERR, "handle_request_update: realloc failed.");
 261       continue;
 262     }
 263     ci->values = temp;
 264
 265     ci->values[ci->values_num] = strdup (value);
 266     if (ci->values[ci->values_num] == NULL)
 267     {
 268       RRDD_LOG (LOG_ERR, "handle_request_update: strdup failed.");
 269       continue;
 270     }
 271     ci->values_num++;
 272
 273     values_num++;
 274   }
 275
 276   /* FIXME: Timeout should not be hard-coded. */
 277   if (((now - ci->last_flush_time) > 300)
 278       && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0))
 279   {
 280     RRDD_LOG (LOG_DEBUG, "handle_request_update: Adding %s to the update queue.",
 281         ci->file);
 282
 283     assert (ci->next == NULL);
 284
 285     if (cache_queue_tail == NULL)
 286       cache_queue_head = ci;
 287     else
 288       cache_queue_tail->next = ci;
 289     cache_queue_tail = ci;
 290
 291     pthread_cond_signal (&cache_cond);
 292   }
 293
 294   pthread_mutex_unlock (&cache_lock);
 295
 296   snprintf (answer, sizeof (answer), "0 Enqueued %i value(s)\n", values_num);
 297   answer[sizeof (answer) - 1] = 0;
 298
 299   write (fd, answer, sizeof (answer));
 300
 301   return (0);
 302 } /* }}} int handle_request_update */
 303
 304 static int handle_request (int fd) /* {{{ */
 305 {
 306   char buffer[4096];
 307   int buffer_size;
 308
 309   RRDD_LOG (LOG_DEBUG, "handle_request (%i)", fd);
 310
 311   buffer_size = read (fd, buffer, sizeof (buffer));
 312   if (buffer_size < 1)
 313   {
 314     RRDD_LOG (LOG_ERR, "handle_request: read(2) failed.");
 315     return (-1);
 316   }
 317   assert (((size_t) buffer_size) <= sizeof (buffer));
 318
 319   if ((buffer[buffer_size - 2] != 0)
 320       || (buffer[buffer_size - 1] != 0))
 321   {
 322     RRDD_LOG (LOG_INFO, "handle_request: malformed request.");
 323     return (-1);
 324   }
 325
 326   /* fields in the buffer a separated by null bytes. */
 327   if (strcmp (buffer, "update") == 0)
 328   {
 329     int offset = strlen ("update") + 1;
 330     return (handle_request_update (fd, buffer + offset,
 331           buffer_size - offset));
 332   }
 333   else
 334   {
 335     RRDD_LOG (LOG_INFO, "handle_request: unknown command: %s.", buffer);
 336     return (-1);
 337   }
 338 } /* }}} int handle_request */
 339
 340 static void *connection_thread_main (void *args) /* {{{ */
 341 {
 342   pthread_t self;
 343   int i;
 344   int fd;
 345
 346   fd = *((int *) args);
 347
 348   while (do_shutdown == 0)
 349   {
 350     struct pollfd pollfd;
 351     int status;
 352
 353     pollfd.fd = fd;
 354     pollfd.events = POLLIN | POLLPRI;
 355     pollfd.revents = 0;
 356
 357     status = poll (&pollfd, 1, /* timeout = */ 500);
 358     if (status == 0) /* timeout */
 359       continue;
 360     else if (status < 0) /* error */
 361     {
 362       status = errno;
 363       if (status == EINTR)
 364         continue;
 365       RRDD_LOG (LOG_ERR, "connection_thread_main: poll(2) failed.");
 366       continue;
 367     }
 368
 369     if ((pollfd.revents & POLLHUP) != 0) /* normal shutdown */
 370     {
 371       close (fd);
 372       break;
 373     }
 374     else if ((pollfd.revents & (POLLIN | POLLPRI)) == 0)
 375     {
 376       RRDD_LOG (LOG_WARNING, "connection_thread_main: poll(2) returned "
 377           "something unexpected.");
 378       close (fd);
 379       break;
 380     }
 381
 382     status = handle_request (fd);
 383     if (status != 0)
 384     {
 385       close (fd);
 386       break;
 387     }
 388   }
 389
 390   self = pthread_self ();
 391   /* Remove this thread from the connection threads list */
 392   pthread_mutex_lock (&connetion_threads_lock);
 393   /* Find out own index in the array */
 394   for (i = 0; i < connetion_threads_num; i++)
 395     if (pthread_equal (connetion_threads[i], self) != 0)
 396       break;
 397   assert (i < connetion_threads_num);
 398
 399   /* Move the trailing threads forward. */
 400   if (i < (connetion_threads_num - 1))
 401   {
 402     memmove (connetion_threads + i,
 403         connetion_threads + i + 1,
 404         sizeof (pthread_t) * (connetion_threads_num - i - 1));
 405   }
 406
 407   connetion_threads_num--;
 408   pthread_mutex_unlock (&connetion_threads_lock);
 409
 410   free (args);
 411   return (NULL);
 412 } /* }}} void *connection_thread_main */
 413
 414 static int open_listen_socket (const char *path) /* {{{ */
 415 {
 416   int fd;
 417   struct sockaddr_un sa;
 418   listen_socket_t *temp;
 419   int status;
 420
 421   temp = (listen_socket_t *) realloc (listen_fds,
 422       sizeof (listen_fds[0]) * (listen_fds_num + 1));
 423   if (temp == NULL)
 424   {
 425     RRDD_LOG (LOG_ERR, "open_listen_socket: realloc failed.");
 426     return (-1);
 427   }
 428   listen_fds = temp;
 429   memset (listen_fds + listen_fds_num, 0, sizeof (listen_fds[0]));
 430
 431   fd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
 432   if (fd < 0)
 433   {
 434     RRDD_LOG (LOG_ERR, "open_listen_socket: socket(2) failed.");
 435     return (-1);
 436   }
 437
 438   memset (&sa, 0, sizeof (sa));
 439   sa.sun_family = AF_UNIX;
 440   strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
 441
 442   status = bind (fd, (struct sockaddr *) &sa, sizeof (sa));
 443   if (status != 0)
 444   {
 445     RRDD_LOG (LOG_ERR, "open_listen_socket: bind(2) failed.");
 446     close (fd);
 447     unlink (path);
 448     return (-1);
 449   }
 450
 451   status = listen (fd, /* backlog = */ 10);
 452   if (status != 0)
 453   {
 454     RRDD_LOG (LOG_ERR, "open_listen_socket: listen(2) failed.");
 455     close (fd);
 456     unlink (path);
 457     return (-1);
 458   }
 459
 460   listen_fds[listen_fds_num].fd = fd;
 461   strncpy (listen_fds[listen_fds_num].path, path,
 462       sizeof (listen_fds[listen_fds_num].path) - 1);
 463   listen_fds_num++;
 464
 465   return (0);
 466 } /* }}} int open_listen_socket */
 467
 468 static int close_listen_sockets (void) /* {{{ */
 469 {
 470   size_t i;
 471
 472   for (i = 0; i < listen_fds_num; i++)
 473   {
 474     close (listen_fds[i].fd);
 475     unlink (listen_fds[i].path);
 476   }
 477
 478   free (listen_fds);
 479   listen_fds = NULL;
 480   listen_fds_num = 0;
 481
 482   return (0);
 483 } /* }}} int close_listen_sockets */
 484
 485 static void *listen_thread_main (void *args) /* {{{ */
 486 {
 487   char buffer[4096];
 488   int status;
 489   int i;
 490
 491   status = open_listen_socket (RRDD_SOCK_PATH);
 492   if (status != 0)
 493   {
 494     RRDD_LOG (LOG_ERR, "listen_thread_main: open_listen_socket failed.");
 495     return (NULL);
 496   }
 497
 498   while (do_shutdown == 0)
 499   {
 500     int *client_sd;
 501     struct sockaddr_un client_sa;
 502     socklen_t client_sa_size;
 503     pthread_t tid;
 504
 505     client_sd = (int *) malloc (sizeof (int));
 506     if (client_sd == NULL)
 507     {
 508       RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
 509       sleep (120);
 510       continue;
 511     }
 512
 513     client_sa_size = sizeof (client_sa);
 514     /* FIXME: Don't implement listen_fds as a list or use poll(2) here! */
 515     *client_sd = accept (listen_fds[0].fd,
 516         (struct sockaddr *) &client_sa, &client_sa_size);
 517     if (*client_sd < 0)
 518     {
 519       RRDD_LOG (LOG_ERR, "listen_thread_main: accept(2) failed.");
 520       continue;
 521     }
 522
 523     RRDD_LOG (LOG_DEBUG, "listen_thread_main: accept(2) returned fd #%i.",
 524         *client_sd);
 525
 526     status = pthread_create (&tid, /* attr = */ NULL, connection_thread_main,
 527         /* args = */ (void *) client_sd);
 528     if (status != 0)
 529     {
 530       RRDD_LOG (LOG_ERR, "listen_thread_main: pthread_create failed.");
 531       close (*client_sd);
 532       free (client_sd);
 533       continue;
 534     }
 535
 536     /* FIXME: bug: if the new thread is run first, it may run into the
 537      *   assert (i < connetion_threads_num);
 538      * assertion. */
 539     pthread_mutex_lock (&connetion_threads_lock);
 540     {
 541       pthread_t *temp;
 542
 543       temp = (pthread_t *) realloc (connetion_threads,
 544           sizeof (pthread_t) * (connetion_threads_num + 1));
 545       if (temp == NULL)
 546       {
 547         RRDD_LOG (LOG_ERR, "listen_thread_main: realloc failed.");
 548       }
 549       else
 550       {
 551         connetion_threads = temp;
 552         connetion_threads[connetion_threads_num] = tid;
 553         connetion_threads_num++;
 554       }
 555     }
 556     pthread_mutex_unlock (&connetion_threads_lock);
 557   } /* while (do_shutdown == 0) */
 558
 559   close_listen_sockets ();
 560
 561   pthread_mutex_lock (&connetion_threads_lock);
 562   while (connetion_threads_num > 0)
 563   {
 564     pthread_t wait_for;
 565
 566     wait_for = connetion_threads[0];
 567
 568     pthread_mutex_unlock (&connetion_threads_lock);
 569     pthread_join (wait_for, /* retval = */ NULL);
 570     pthread_mutex_lock (&connetion_threads_lock);
 571   }
 572   pthread_mutex_unlock (&connetion_threads_lock);
 573
 574   RRDD_LOG (LOG_DEBUG, "listen_thread_main: Exiting.");
 575
 576   return (NULL);
 577 } /* }}} void *listen_thread_main */
 578
 579 static int daemonize (void) /* {{{ */
 580 {
 581   pid_t child;
 582   int status;
 583
 584 #if !RRDD_DEBUG
 585   child = fork ();
 586   if (child < 0)
 587   {
 588     fprintf (stderr, "daemonize: fork(2) failed.\n");
 589     return (-1);
 590   }
 591   else if (child > 0)
 592   {
 593     return (1);
 594   }
 595
 596   /* Change into the /tmp directory. */
 597   chdir ("/tmp");
 598
 599   /* Become session leader */
 600   setsid ();
 601
 602   /* Open the first three file descriptors to /dev/null */
 603   close (2);
 604   close (1);
 605   close (0);
 606
 607   open ("/dev/null", O_RDWR);
 608   dup (0);
 609   dup (0);
 610 #endif /* RRDD_DEBUG */
 611
 612   {
 613     struct sigaction sa;
 614
 615     memset (&sa, 0, sizeof (sa));
 616     sa.sa_handler = sig_int_handler;
 617     sigaction (SIGINT, &sa, NULL);
 618
 619     memset (&sa, 0, sizeof (sa));
 620     sa.sa_handler = sig_term_handler;
 621     sigaction (SIGINT, &sa, NULL);
 622   }
 623
 624   openlog ("rrdd", LOG_PID, LOG_DAEMON);
 625
 626   cache_tree = avl_alloc_tree (cache_tree_compare, cache_tree_free);
 627   if (cache_tree == NULL)
 628   {
 629     RRDD_LOG (LOG_ERR, "daemonize: avl_alloc_tree failed.");
 630     return (-1);
 631   }
 632
 633   memset (&queue_thread, 0, sizeof (queue_thread));
 634   status = pthread_create (&queue_thread, /* attr = */ NULL,
 635       queue_thread_main, /* args = */ NULL);
 636   if (status != 0)
 637   {
 638     RRDD_LOG (LOG_ERR, "daemonize: pthread_create failed.");
 639     return (-1);
 640   }
 641
 642   return (0);
 643 } /* }}} int daemonize */
 644
 645 static int cleanup (void) /* {{{ */
 646 {
 647   RRDD_LOG (LOG_DEBUG, "cleanup ()");
 648
 649   do_shutdown++;
 650
 651   pthread_join (queue_thread, /* return = */ NULL);
 652
 653   closelog ();
 654
 655   return (0);
 656 } /* }}} int cleanup */
 657
 658 int main (int argc, char **argv)
 659 {
 660   int status;
 661
 662   printf ("%s by Florian Forster, Version %s\n",
 663       PACKAGE_NAME, PACKAGE_VERSION);
 664
 665   status = daemonize ();
 666   if (status == 1)
 667   {
 668     struct sigaction sigchld;
 669
 670     memset (&sigchld, 0, sizeof (sigchld));
 671     sigchld.sa_handler = SIG_IGN;
 672     sigaction (SIGCHLD, &sigchld, NULL);
 673
 674     return (0);
 675   }
 676   else if (status != 0)
 677   {
 678     fprintf (stderr, "daemonize failed, exiting.\n");
 679     return (1);
 680   }
 681
 682   listen_thread_main (NULL);
 683
 684   cleanup ();
 685
 686   return (0);
 687 } /* int main */
 688
 689 /*
 690  * vim: set sw=2 sts=2 ts=8 et fdm=marker :
 691  */