src/plugin.c

   1 /**
   2  * collectd - src/plugin.c
   3  * Copyright (C) 2005,2006  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 #include "collectd.h"
  23
  24 #include <ltdl.h>
  25
  26 #if HAVE_PTHREAD_H
  27 # include <pthread.h>
  28 #endif
  29
  30 #include "common.h"
  31 #include "plugin.h"
  32 #include "configfile.h"
  33 #include "utils_llist.h"
  34 #include "utils_cache.h"
  35
  36 /*
  37  * Private structures
  38  */
  39 struct read_func_s
  40 {
  41         int wait_time;
  42         int wait_left;
  43         int (*callback) (void);
  44         enum { DONE = 0, TODO = 1, ACTIVE = 2 } needs_read;
  45 };
  46 typedef struct read_func_s read_func_t;
  47
  48 /*
  49  * Private variables
  50  */
  51 static llist_t *list_init;
  52 static llist_t *list_read;
  53 static llist_t *list_write;
  54 static llist_t *list_shutdown;
  55 static llist_t *list_data_set;
  56 static llist_t *list_log;
  57
  58 static char *plugindir = NULL;
  59
  60 static int             read_loop = 1;
  61 static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER;
  62 static pthread_cond_t  read_cond = PTHREAD_COND_INITIALIZER;
  63 static pthread_t      *read_threads = NULL;
  64 static int             read_threads_num = 0;
  65
  66 /*
  67  * Static functions
  68  */
  69 static const char *plugin_get_dir (void)
  70 {
  71         if (plugindir == NULL)
  72                 return (PLUGINDIR);
  73         else
  74                 return (plugindir);
  75 }
  76
  77 static int register_callback (llist_t **list, const char *name, void *callback)
  78 {
  79         llentry_t *le;
  80
  81         if ((*list == NULL)
  82                         && ((*list = llist_create ()) == NULL))
  83                 return (-1);
  84
  85         le = llist_search (*list, name);
  86         if (le == NULL)
  87         {
  88                 le = llentry_create (name, callback);
  89                 if (le == NULL)
  90                         return (-1);
  91
  92                 llist_append (*list, le);
  93         }
  94         else
  95         {
  96                 le->value = callback;
  97         }
  98
  99         return (0);
 100 } /* int register_callback */
 101
 102 static int plugin_unregister (llist_t *list, const char *name)
 103 {
 104         llentry_t *e;
 105
 106         e = llist_search (list, name);
 107
 108         if (e == NULL)
 109                 return (-1);
 110
 111         llist_remove (list, e);
 112         llentry_destroy (e);
 113
 114         return (0);
 115 } /* int plugin_unregister */
 116
 117 /*
 118  * (Try to) load the shared object `file'. Won't complain if it isn't a shared
 119  * object, but it will bitch about a shared object not having a
 120  * ``module_register'' symbol..
 121  */
 122 static int plugin_load_file (char *file)
 123 {
 124         lt_dlhandle dlh;
 125         void (*reg_handle) (void);
 126
 127         DEBUG ("file = %s", file);
 128
 129         lt_dlinit ();
 130         lt_dlerror (); /* clear errors */
 131
 132         if ((dlh = lt_dlopen (file)) == NULL)
 133         {
 134                 const char *error = lt_dlerror ();
 135
 136                 ERROR ("lt_dlopen failed: %s", error);
 137                 fprintf (stderr, "lt_dlopen failed: %s\n", error);
 138                 return (1);
 139         }
 140
 141         if ((reg_handle = (void (*) (void)) lt_dlsym (dlh, "module_register")) == NULL)
 142         {
 143                 WARNING ("Couldn't find symbol ``module_register'' in ``%s'': %s\n",
 144                                 file, lt_dlerror ());
 145                 lt_dlclose (dlh);
 146                 return (-1);
 147         }
 148
 149         (*reg_handle) ();
 150
 151         return (0);
 152 }
 153
 154 static void *plugin_read_thread (void *args)
 155 {
 156         llentry_t   *le;
 157         read_func_t *rf;
 158         int          status;
 159         int          done;
 160
 161         pthread_mutex_lock (&read_lock);
 162
 163         while (read_loop != 0)
 164         {
 165                 le = llist_head (list_read);
 166                 done = 0;
 167
 168                 while ((read_loop != 0) && (le != NULL))
 169                 {
 170                         rf = (read_func_t *) le->value;
 171
 172                         if (rf->needs_read != TODO)
 173                         {
 174                                 le = le->next;
 175                                 continue;
 176                         }
 177
 178                         /* We will do this read function */
 179                         rf->needs_read = ACTIVE;
 180
 181                         DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Handling %s",
 182                                         (unsigned long int) pthread_self (), le->key);
 183                         pthread_mutex_unlock (&read_lock);
 184
 185                         status = rf->callback ();
 186                         done++;
 187
 188                         if (status != 0)
 189                         {
 190                                 if (rf->wait_time < interval_g)
 191                                         rf->wait_time = interval_g;
 192                                 rf->wait_left = rf->wait_time;
 193                                 rf->wait_time = rf->wait_time * 2;
 194                                 if (rf->wait_time > 86400)
 195                                         rf->wait_time = 86400;
 196
 197                                 NOTICE ("read-function of plugin `%s' "
 198                                                 "failed. Will suspend it for %i "
 199                                                 "seconds.", le->key, rf->wait_left);
 200                         }
 201                         else
 202                         {
 203                                 rf->wait_left = 0;
 204                                 rf->wait_time = interval_g;
 205                         }
 206
 207                         pthread_mutex_lock (&read_lock);
 208
 209                         rf->needs_read = DONE;
 210                         le = le->next;
 211                 } /* while (le != NULL) */
 212
 213                 if ((read_loop != 0) && (done == 0))
 214                 {
 215                         DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Waiting on read_cond.",
 216                                         (unsigned long int) pthread_self ());
 217                         pthread_cond_wait (&read_cond, &read_lock);
 218                 }
 219         } /* while (read_loop) */
 220
 221         pthread_mutex_unlock (&read_lock);
 222
 223         pthread_exit (NULL);
 224 } /* void *plugin_read_thread */
 225
 226 static void start_threads (int num)
 227 {
 228         int i;
 229
 230         if (read_threads != NULL)
 231                 return;
 232
 233         read_threads = (pthread_t *) calloc (num, sizeof (pthread_t));
 234         if (read_threads == NULL)
 235         {
 236                 ERROR ("plugin: start_threads: calloc failed.");
 237                 return;
 238         }
 239
 240         read_threads_num = 0;
 241         for (i = 0; i < num; i++)
 242         {
 243                 if (pthread_create (read_threads + read_threads_num, NULL,
 244                                         plugin_read_thread, NULL) == 0)
 245                 {
 246                         read_threads_num++;
 247                 }
 248                 else
 249                 {
 250                         ERROR ("plugin: start_threads: pthread_create failed.");
 251                         return;
 252                 }
 253         } /* for (i) */
 254 } /* void start_threads */
 255
 256 static void stop_threads (void)
 257 {
 258         int i;
 259
 260         pthread_mutex_lock (&read_lock);
 261         read_loop = 0;
 262         DEBUG ("plugin: stop_threads: Signalling `read_cond'");
 263         pthread_cond_broadcast (&read_cond);
 264         pthread_mutex_unlock (&read_lock);
 265
 266         for (i = 0; i < read_threads_num; i++)
 267         {
 268                 if (pthread_join (read_threads[i], NULL) != 0)
 269                 {
 270                         ERROR ("plugin: stop_threads: pthread_join failed.");
 271                 }
 272                 read_threads[i] = (pthread_t) 0;
 273         }
 274         sfree (read_threads);
 275         read_threads_num = 0;
 276 } /* void stop_threads */
 277
 278 /*
 279  * Public functions
 280  */
 281 void plugin_set_dir (const char *dir)
 282 {
 283         if (plugindir != NULL)
 284                 free (plugindir);
 285
 286         if (dir == NULL)
 287                 plugindir = NULL;
 288         else if ((plugindir = strdup (dir)) == NULL)
 289         {
 290                 char errbuf[1024];
 291                 ERROR ("strdup failed: %s",
 292                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 293         }
 294 }
 295
 296 #define BUFSIZE 512
 297 int plugin_load (const char *type)
 298 {
 299         DIR  *dh;
 300         const char *dir;
 301         char  filename[BUFSIZE];
 302         char  typename[BUFSIZE];
 303         int   typename_len;
 304         int   ret;
 305         struct stat    statbuf;
 306         struct dirent *de;
 307
 308         DEBUG ("type = %s", type);
 309
 310         dir = plugin_get_dir ();
 311         ret = 1;
 312
 313         /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
 314          * type when matching the filename */
 315         if (snprintf (typename, BUFSIZE, "%s.so", type) >= BUFSIZE)
 316         {
 317                 WARNING ("snprintf: truncated: `%s.so'", type);
 318                 return (-1);
 319         }
 320         typename_len = strlen (typename);
 321
 322         if ((dh = opendir (dir)) == NULL)
 323         {
 324                 char errbuf[1024];
 325                 ERROR ("opendir (%s): %s", dir,
 326                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 327                 return (-1);
 328         }
 329
 330         while ((de = readdir (dh)) != NULL)
 331         {
 332                 if (strncasecmp (de->d_name, typename, typename_len))
 333                         continue;
 334
 335                 if (snprintf (filename, BUFSIZE, "%s/%s", dir, de->d_name) >= BUFSIZE)
 336                 {
 337                         WARNING ("snprintf: truncated: `%s/%s'", dir, de->d_name);
 338                         continue;
 339                 }
 340
 341                 if (lstat (filename, &statbuf) == -1)
 342                 {
 343                         char errbuf[1024];
 344                         WARNING ("stat %s: %s", filename,
 345                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 346                         continue;
 347                 }
 348                 else if (!S_ISREG (statbuf.st_mode))
 349                 {
 350                         /* don't follow symlinks */
 351                         continue;
 352                 }
 353
 354                 if (plugin_load_file (filename) == 0)
 355                 {
 356                         /* success */
 357                         ret = 0;
 358                         break;
 359                 }
 360                 else
 361                 {
 362                         fprintf (stderr, "Unable to load plugin %s.\n", type);
 363                 }
 364         }
 365
 366         closedir (dh);
 367
 368         return (ret);
 369 }
 370
 371 /*
 372  * The `register_*' functions follow
 373  */
 374 int plugin_register_config (const char *name,
 375                 int (*callback) (const char *key, const char *val),
 376                 const char **keys, int keys_num)
 377 {
 378         cf_register (name, callback, keys, keys_num);
 379         return (0);
 380 } /* int plugin_register_config */
 381
 382 int plugin_register_complex_config (const char *type,
 383                 int (*callback) (oconfig_item_t *))
 384 {
 385         return (cf_register_complex (type, callback));
 386 } /* int plugin_register_complex_config */
 387
 388 int plugin_register_init (const char *name,
 389                 int (*callback) (void))
 390 {
 391         return (register_callback (&list_init, name, (void *) callback));
 392 } /* plugin_register_init */
 393
 394 int plugin_register_read (const char *name,
 395                 int (*callback) (void))
 396 {
 397         read_func_t *rf;
 398
 399         rf = (read_func_t *) malloc (sizeof (read_func_t));
 400         if (rf == NULL)
 401         {
 402                 char errbuf[1024];
 403                 ERROR ("plugin_register_read: malloc failed: %s",
 404                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 405                 return (-1);
 406         }
 407
 408         memset (rf, '\0', sizeof (read_func_t));
 409         rf->wait_time = interval_g;
 410         rf->wait_left = 0;
 411         rf->callback = callback;
 412         rf->needs_read = DONE;
 413
 414         return (register_callback (&list_read, name, (void *) rf));
 415 } /* int plugin_register_read */
 416
 417 int plugin_register_write (const char *name,
 418                 int (*callback) (const data_set_t *ds, const value_list_t *vl))
 419 {
 420         return (register_callback (&list_write, name, (void *) callback));
 421 } /* int plugin_register_write */
 422
 423 int plugin_register_shutdown (char *name,
 424                 int (*callback) (void))
 425 {
 426         return (register_callback (&list_shutdown, name, (void *) callback));
 427 } /* int plugin_register_shutdown */
 428
 429 int plugin_register_data_set (const data_set_t *ds)
 430 {
 431         data_set_t *ds_copy;
 432         int i;
 433
 434         if ((list_data_set != NULL)
 435                         && (llist_search (list_data_set, ds->type) != NULL))
 436         {
 437                 NOTICE ("Replacing DS `%s' with another version.", ds->type);
 438                 plugin_unregister_data_set (ds->type);
 439         }
 440
 441         ds_copy = (data_set_t *) malloc (sizeof (data_set_t));
 442         if (ds_copy == NULL)
 443                 return (-1);
 444         memcpy(ds_copy, ds, sizeof (data_set_t));
 445
 446         ds_copy->ds = (data_source_t *) malloc (sizeof (data_source_t)
 447                         * ds->ds_num);
 448         if (ds_copy->ds == NULL)
 449         {
 450                 free (ds_copy);
 451                 return (-1);
 452         }
 453
 454         for (i = 0; i < ds->ds_num; i++)
 455                 memcpy (ds_copy->ds + i, ds->ds + i, sizeof (data_source_t));
 456
 457         return (register_callback (&list_data_set, ds->type, (void *) ds_copy));
 458 } /* int plugin_register_data_set */
 459
 460 int plugin_register_log (char *name,
 461                 void (*callback) (int priority, const char *msg))
 462 {
 463         return (register_callback (&list_log, name, (void *) callback));
 464 } /* int plugin_register_log */
 465
 466 int plugin_unregister_config (const char *name)
 467 {
 468         cf_unregister (name);
 469         return (0);
 470 } /* int plugin_unregister_config */
 471
 472 int plugin_unregister_complex_config (const char *name)
 473 {
 474         cf_unregister_complex (name);
 475         return (0);
 476 } /* int plugin_unregister_complex_config */
 477
 478 int plugin_unregister_init (const char *name)
 479 {
 480         return (plugin_unregister (list_init, name));
 481 }
 482
 483 int plugin_unregister_read (const char *name)
 484 {
 485         llentry_t *e;
 486
 487         e = llist_search (list_read, name);
 488
 489         if (e == NULL)
 490                 return (-1);
 491
 492         llist_remove (list_read, e);
 493         free (e->value);
 494         llentry_destroy (e);
 495
 496         return (0);
 497 }
 498
 499 int plugin_unregister_write (const char *name)
 500 {
 501         return (plugin_unregister (list_write, name));
 502 }
 503
 504 int plugin_unregister_shutdown (const char *name)
 505 {
 506         return (plugin_unregister (list_shutdown, name));
 507 }
 508
 509 int plugin_unregister_data_set (const char *name)
 510 {
 511         llentry_t  *e;
 512         data_set_t *ds;
 513
 514         if (list_data_set == NULL)
 515                 return (-1);
 516
 517         e = llist_search (list_data_set, name);
 518
 519         if (e == NULL)
 520                 return (-1);
 521
 522         llist_remove (list_data_set, e);
 523         ds = (data_set_t *) e->value;
 524         llentry_destroy (e);
 525
 526         sfree (ds->ds);
 527         sfree (ds);
 528
 529         return (0);
 530 } /* int plugin_unregister_data_set */
 531
 532 int plugin_unregister_log (const char *name)
 533 {
 534         return (plugin_unregister (list_log, name));
 535 }
 536
 537 void plugin_init_all (void)
 538 {
 539         int (*callback) (void);
 540         llentry_t *le;
 541         int status;
 542
 543         /* Start read-threads */
 544         if (list_read != NULL)
 545         {
 546                 const char *rt;
 547                 int num;
 548                 rt = global_option_get ("ReadThreads");
 549                 num = atoi (rt);
 550                 start_threads ((num > 0) ? num : 5);
 551         }
 552
 553         /* Init the value cache */
 554         uc_init ();
 555
 556         if (list_init == NULL)
 557                 return;
 558
 559         le = llist_head (list_init);
 560         while (le != NULL)
 561         {
 562                 callback = (int (*) (void)) le->value;
 563                 status = (*callback) ();
 564
 565                 if (status != 0)
 566                 {
 567                         ERROR ("Initialization of plugin `%s' "
 568                                         "failed with status %i. "
 569                                         "Plugin will be unloaded.",
 570                                         le->key, status);
 571                         /* FIXME: Unload _all_ functions */
 572                         plugin_unregister_read (le->key);
 573                 }
 574
 575                 le = le->next;
 576         }
 577 } /* void plugin_init_all */
 578
 579 void plugin_read_all (const int *loop)
 580 {
 581         llentry_t   *le;
 582         read_func_t *rf;
 583
 584         if (list_read == NULL)
 585                 return;
 586
 587         pthread_mutex_lock (&read_lock);
 588
 589         le = llist_head (list_read);
 590         while (le != NULL)
 591         {
 592                 rf = (read_func_t *) le->value;
 593
 594                 if (rf->needs_read != DONE)
 595                 {
 596                         le = le->next;
 597                         continue;
 598                 }
 599
 600                 if (rf->wait_left > 0)
 601                         rf->wait_left -= interval_g;
 602
 603                 if (rf->wait_left <= 0)
 604                 {
 605                         rf->needs_read = TODO;
 606                 }
 607
 608                 le = le->next;
 609         }
 610
 611         DEBUG ("plugin: plugin_read_all: Signalling `read_cond'");
 612         pthread_cond_broadcast (&read_cond);
 613         pthread_mutex_unlock (&read_lock);
 614 } /* void plugin_read_all */
 615
 616 void plugin_shutdown_all (void)
 617 {
 618         int (*callback) (void);
 619         llentry_t *le;
 620
 621         stop_threads ();
 622
 623         if (list_shutdown == NULL)
 624                 return;
 625
 626         le = llist_head (list_shutdown);
 627         while (le != NULL)
 628         {
 629                 callback = (int (*) (void)) le->value;
 630
 631                 /* Advance the pointer before calling the callback allows
 632                  * shutdown functions to unregister themselves. If done the
 633                  * other way around the memory `le' points to will be freed
 634                  * after callback returns. */
 635                 le = le->next;
 636
 637                 (*callback) ();
 638         }
 639 } /* void plugin_shutdown_all */
 640
 641 int plugin_dispatch_values (const char *name, value_list_t *vl)
 642 {
 643         int (*callback) (const data_set_t *, const value_list_t *);
 644         data_set_t *ds;
 645         llentry_t *le;
 646
 647         if ((list_write == NULL) || (list_data_set == NULL))
 648                 return (-1);
 649
 650         le = llist_search (list_data_set, name);
 651         if (le == NULL)
 652         {
 653                 DEBUG ("No such dataset registered: %s", name);
 654                 return (-1);
 655         }
 656
 657         ds = (data_set_t *) le->value;
 658
 659         DEBUG ("plugin: plugin_dispatch_values: time = %u; interval = %i; "
 660                         "host = %s; "
 661                         "plugin = %s; plugin_instance = %s; "
 662                         "type = %s; type_instance = %s;",
 663                         (unsigned int) vl->time, vl->interval,
 664                         vl->host,
 665                         vl->plugin, vl->plugin_instance,
 666                         ds->type, vl->type_instance);
 667
 668 #if COLLECT_DEBUG
 669         assert (ds->ds_num == vl->values_len);
 670 #else
 671         if (ds->ds_num != vl->values_len)
 672         {
 673                 ERROR ("plugin: ds->type = %s: (ds->ds_num = %i) != "
 674                                 "(vl->values_len = %i)",
 675                                 ds->type, ds->ds_num, vl->values_len);
 676                 return (-1);
 677         }
 678 #endif
 679
 680         escape_slashes (vl->host, sizeof (vl->host));
 681         escape_slashes (vl->plugin, sizeof (vl->plugin));
 682         escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance));
 683         escape_slashes (vl->type_instance, sizeof (vl->type_instance));
 684
 685         /* Update the value cache */
 686         uc_update (ds, vl);
 687
 688         le = llist_head (list_write);
 689         while (le != NULL)
 690         {
 691                 callback = (int (*) (const data_set_t *, const value_list_t *)) le->value;
 692                 (*callback) (ds, vl);
 693
 694                 le = le->next;
 695         }
 696
 697         return (0);
 698 } /* int plugin_dispatch_values */
 699
 700 void plugin_log (int level, const char *format, ...)
 701 {
 702         char msg[512];
 703         va_list ap;
 704
 705         void (*callback) (int, const char *);
 706         llentry_t *le;
 707
 708         if (list_log == NULL)
 709                 return;
 710
 711 #if !COLLECT_DEBUG
 712         if (level >= LOG_DEBUG)
 713                 return;
 714 #endif
 715
 716         va_start (ap, format);
 717         vsnprintf (msg, 512, format, ap);
 718         msg[511] = '\0';
 719         va_end (ap);
 720
 721         le = llist_head (list_log);
 722         while (le != NULL)
 723         {
 724                 callback = (void (*) (int, const char *)) le->value;
 725                 (*callback) (level, msg);
 726
 727                 le = le->next;
 728         }
 729 } /* void plugin_log */
 730
 731 void plugin_complain (int level, complain_t *c, const char *format, ...)
 732 {
 733         char message[512];
 734         va_list ap;
 735
 736         if (c->delay > 0)
 737         {
 738                 c->delay--;
 739                 return;
 740         }
 741
 742         if (c->interval < interval_g)
 743                 c->interval = interval_g;
 744         else
 745                 c->interval *= 2;
 746
 747         if (c->interval > 86400)
 748                 c->interval = 86400;
 749
 750         c->delay = c->interval / interval_g;
 751
 752         va_start (ap, format);
 753         vsnprintf (message, 512, format, ap);
 754         message[511] = '\0';
 755         va_end (ap);
 756
 757         plugin_log (level, message);
 758 }
 759
 760 void plugin_relief (int level, complain_t *c, const char *format, ...)
 761 {
 762         char message[512];
 763         va_list ap;
 764
 765         if (c->interval == 0)
 766                 return;
 767
 768         c->interval = 0;
 769
 770         va_start (ap, format);
 771         vsnprintf (message, 512, format, ap);
 772         message[511] = '\0';
 773         va_end (ap);
 774
 775         plugin_log (level, message);
 776 }
 777
 778 const data_set_t *plugin_get_ds (const char *name)
 779 {
 780         data_set_t *ds;
 781         llentry_t *le;
 782
 783         le = llist_search (list_data_set, name);
 784         if (le == NULL)
 785         {
 786                 DEBUG ("No such dataset registered: %s", name);
 787                 return (NULL);
 788         }
 789
 790         ds = (data_set_t *) le->value;
 791
 792         return (ds);
 793 } /* data_set_t *plugin_get_ds */