src/email.c

   1 /**
   2  * collectd - src/email.c
   3  * Copyright (C) 2006,2007  Sebastian Harl
   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; either version 2 of the License, or (at your
   8  * option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  18  *
  19  * Author:
  20  *   Sebastian Harl <sh at tokkee.org>
  21  **/
  22
  23 /*
  24  * This plugin communicates with a spam filter, a virus scanner or similar
  25  * software using a UNIX socket and a very simple protocol:
  26  *
  27  * e-mail type (e.g. ham, spam, virus, ...) and size
  28  * e:<type>:<bytes>
  29  *
  30  * spam score
  31  * s:<value>
  32  *
  33  * successful spam checks
  34  * c:<type1>[,<type2>,...]
  35  */
  36
  37 #include "collectd.h"
  38 #include "common.h"
  39 #include "plugin.h"
  40
  41 #include "configfile.h"
  42
  43 #if HAVE_LIBPTHREAD
  44 # include <pthread.h>
  45 #endif
  46
  47 #if HAVE_SYS_SELECT_H
  48 #       include <sys/select.h>
  49 #endif /* HAVE_SYS_SELECT_H */
  50
  51 #if HAVE_SYS_SOCKET_H
  52 #       include <sys/socket.h>
  53 #endif /* HAVE_SYS_SOCKET_H */
  54
  55 /* *sigh* glibc does not define UNIX_PATH_MAX in sys/un.h ... */
  56 #if HAVE_LINUX_UN_H
  57 #       include <linux/un.h>
  58 #elif HAVE_SYS_UN_H
  59 #       include <sys/un.h>
  60 #endif /* HAVE_LINUX_UN_H | HAVE_SYS_UN_H */
  61
  62 /* some systems (e.g. Darwin) seem to not define UNIX_PATH_MAX at all */
  63 #ifndef UNIX_PATH_MAX
  64 # define UNIX_PATH_MAX sizeof (((struct sockaddr_un *)0)->sun_path)
  65 #endif /* UNIX_PATH_MAX */
  66
  67 #if HAVE_GRP_H
  68 #       include <grp.h>
  69 #endif /* HAVE_GRP_H */
  70
  71 #define MODULE_NAME "email"
  72
  73 /* 256 bytes ought to be enough for anybody ;-) */
  74 #define BUFSIZE 256
  75
  76 #ifndef COLLECTD_SOCKET_PREFIX
  77 # define COLLECTD_SOCKET_PREFIX "/tmp/.collectd-"
  78 #endif /* COLLECTD_SOCKET_PREFIX */
  79
  80 #define SOCK_PATH COLLECTD_SOCKET_PREFIX"email"
  81 #define MAX_CONNS 5
  82 #define MAX_CONNS_LIMIT 16384
  83
  84 #define log_err(...) ERROR (MODULE_NAME": "__VA_ARGS__)
  85 #define log_warn(...) WARNING (MODULE_NAME": "__VA_ARGS__)
  86
  87 /*
  88  * Private data structures
  89  */
  90 /* linked list of email and check types */
  91 typedef struct type {
  92         char        *name;
  93         int         value;
  94         struct type *next;
  95 } type_t;
  96
  97 typedef struct {
  98         type_t *head;
  99         type_t *tail;
 100 } type_list_t;
 101
 102 /* collector thread control information */
 103 typedef struct collector {
 104         pthread_t thread;
 105
 106         /* socket descriptor of the current/last connection */
 107         int socket;
 108 } collector_t;
 109
 110 /* linked list of pending connections */
 111 typedef struct conn {
 112         /* socket to read data from */
 113         int socket;
 114
 115         /* buffer to read data to */
 116         char *buffer;
 117         int  idx; /* current write position in buffer */
 118         int  length; /* length of the current line, i.e. index of '\0' */
 119
 120         struct conn *next;
 121 } conn_t;
 122
 123 typedef struct {
 124         conn_t *head;
 125         conn_t *tail;
 126 } conn_list_t;
 127
 128 /*
 129  * Private variables
 130  */
 131 /* valid configuration file keys */
 132 static const char *config_keys[] =
 133 {
 134         "SocketGroup",
 135         "SocketPerms",
 136         "MaxConns"
 137 };
 138 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
 139
 140 /* socket configuration */
 141 static char *sock_group = COLLECTD_GRP_NAME;
 142 static int  sock_perms  = S_IRWXU | S_IRWXG;
 143 static int  max_conns   = MAX_CONNS;
 144
 145 /* state of the plugin */
 146 static int disabled = 0;
 147
 148 /* thread managing "client" connections */
 149 static pthread_t connector = (pthread_t) 0;
 150 static int connector_socket = -1;
 151
 152 /* tell the collector threads that a new connection is available */
 153 static pthread_cond_t conn_available = PTHREAD_COND_INITIALIZER;
 154
 155 /* connections that are waiting to be processed */
 156 static pthread_mutex_t conns_mutex = PTHREAD_MUTEX_INITIALIZER;
 157 static conn_list_t conns;
 158
 159 /* tell the connector thread that a collector is available */
 160 static pthread_cond_t collector_available = PTHREAD_COND_INITIALIZER;
 161
 162 /* collector threads */
 163 static collector_t **collectors = NULL;
 164
 165 static pthread_mutex_t available_mutex = PTHREAD_MUTEX_INITIALIZER;
 166 static int available_collectors;
 167
 168 static pthread_mutex_t count_mutex = PTHREAD_MUTEX_INITIALIZER;
 169 static type_list_t count;
 170
 171 static pthread_mutex_t size_mutex = PTHREAD_MUTEX_INITIALIZER;
 172 static type_list_t size;
 173
 174 static pthread_mutex_t score_mutex = PTHREAD_MUTEX_INITIALIZER;
 175 static double score;
 176 static int score_count;
 177
 178 static pthread_mutex_t check_mutex = PTHREAD_MUTEX_INITIALIZER;
 179 static type_list_t check;
 180
 181 /*
 182  * Private functions
 183  */
 184 static int email_config (const char *key, const char *value)
 185 {
 186         if (0 == strcasecmp (key, "SocketGroup")) {
 187                 sock_group = sstrdup (value);
 188         }
 189         else if (0 == strcasecmp (key, "SocketPerms")) {
 190                 /* the user is responsible for providing reasonable values */
 191                 sock_perms = (int)strtol (value, NULL, 8);
 192         }
 193         else if (0 == strcasecmp (key, "MaxConns")) {
 194                 long int tmp = strtol (value, NULL, 0);
 195
 196                 if (tmp < 1) {
 197                         fprintf (stderr, "email plugin: `MaxConns' was set to invalid "
 198                                         "value %li, will use default %i.\n",
 199                                         tmp, MAX_CONNS);
 200                         max_conns = MAX_CONNS;
 201                 }
 202                 else if (tmp > MAX_CONNS_LIMIT) {
 203                         fprintf (stderr, "email plugin: `MaxConns' was set to invalid "
 204                                         "value %li, will use hardcoded limit %i.\n",
 205                                         tmp, MAX_CONNS_LIMIT);
 206                         max_conns = MAX_CONNS_LIMIT;
 207                 }
 208                 else {
 209                         max_conns = (int)tmp;
 210                 }
 211         }
 212         else {
 213                 return -1;
 214         }
 215         return 0;
 216 } /* static int email_config (char *, char *) */
 217
 218 /* Increment the value of the given name in the given list by incr. */
 219 static void type_list_incr (type_list_t *list, char *name, int incr)
 220 {
 221         if (NULL == list->head) {
 222                 list->head = (type_t *)smalloc (sizeof (type_t));
 223
 224                 list->head->name  = sstrdup (name);
 225                 list->head->value = incr;
 226                 list->head->next  = NULL;
 227
 228                 list->tail = list->head;
 229         }
 230         else {
 231                 type_t *ptr;
 232
 233                 for (ptr = list->head; NULL != ptr; ptr = ptr->next) {
 234                         if (0 == strcmp (name, ptr->name))
 235                                 break;
 236                 }
 237
 238                 if (NULL == ptr) {
 239                         list->tail->next = (type_t *)smalloc (sizeof (type_t));
 240                         list->tail = list->tail->next;
 241
 242                         list->tail->name  = sstrdup (name);
 243                         list->tail->value = incr;
 244                         list->tail->next  = NULL;
 245                 }
 246                 else {
 247                         ptr->value += incr;
 248                 }
 249         }
 250         return;
 251 } /* static void type_list_incr (type_list_t *, char *) */
 252
 253 /* Read a single character from the socket. If an error occurs or end-of-file
 254  * is reached return '\0'. */
 255 static char read_char (conn_t *src)
 256 {
 257         char ret = '\0';
 258
 259         fd_set fdset;
 260
 261         FD_ZERO (&fdset);
 262         FD_SET (src->socket, &fdset);
 263
 264         if (-1 == select (src->socket + 1, &fdset, NULL, NULL, NULL)) {
 265                 char errbuf[1024];
 266                 log_err ("select() failed: %s",
 267                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 268                 return '\0';
 269         }
 270
 271         assert (FD_ISSET (src->socket, &fdset));
 272
 273         do {
 274                 ssize_t len = 0;
 275
 276                 errno = 0;
 277                 if (0 > (len = read (src->socket, (void *)&ret, 1))) {
 278                         if (EINTR != errno) {
 279                                 char errbuf[1024];
 280                                 log_err ("read() failed: %s",
 281                                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 282                                 return '\0';
 283                         }
 284                 }
 285
 286                 if (0 == len)
 287                         return '\0';
 288         } while (EINTR == errno);
 289         return ret;
 290 } /* static char read_char (conn_t *) */
 291
 292 /* Read a single line (terminated by '\n') from the the socket.
 293  *
 294  * The return value is zero terminated and does not contain any newline
 295  * characters.
 296  *
 297  * If an error occurs or end-of-file is reached return NULL.
 298  *
 299  * IMPORTANT NOTE: If there is no newline character found in BUFSIZE
 300  * characters of the input stream, the line will will be ignored! By
 301  * definition we should not get any longer input lines, thus this is
 302  * acceptable in this case ;-) */
 303 static char *read_line (conn_t *src)
 304 {
 305         int i = 0;
 306
 307         assert ((BUFSIZE >= src->idx) && (src->idx >= 0));
 308         assert ((src->idx > src->length) || (src->length == 0));
 309
 310         if (src->length > 0) { /* remove old line */
 311                 src->idx -= (src->length + 1);
 312                 memmove (src->buffer, src->buffer + src->length + 1, src->idx);
 313                 src->length = 0;
 314         }
 315
 316         for (i = 0; i < src->idx; ++i) {
 317                 if ('\n' == src->buffer[i])
 318                         break;
 319         }
 320
 321         if (i == src->idx) {
 322                 fd_set fdset;
 323
 324                 ssize_t len = 0;
 325
 326                 FD_ZERO (&fdset);
 327                 FD_SET (src->socket, &fdset);
 328
 329                 if (-1 == select (src->socket + 1, &fdset, NULL, NULL, NULL)) {
 330                         char errbuf[1024];
 331                         log_err ("select() failed: %s",
 332                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 333                         return NULL;
 334                 }
 335
 336                 assert (FD_ISSET (src->socket, &fdset));
 337
 338                 do {
 339                         errno = 0;
 340                         if (0 > (len = read (src->socket,
 341                                                         (void *)(src->buffer + src->idx),
 342                                                         BUFSIZE - src->idx))) {
 343                                 if (EINTR != errno) {
 344                                         char errbuf[1024];
 345                                         log_err ("read() failed: %s",
 346                                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 347                                         return NULL;
 348                                 }
 349                         }
 350
 351                         if (0 == len)
 352                                 return NULL;
 353                 } while (EINTR == errno);
 354
 355                 src->idx += len;
 356
 357                 for (i = src->idx - len; i < src->idx; ++i) {
 358                         if ('\n' == src->buffer[i])
 359                                 break;
 360                 }
 361
 362                 if (i == src->idx) {
 363                         src->length = 0;
 364
 365                         if (BUFSIZE == src->idx) { /* no space left in buffer */
 366                                 while ('\n' != read_char (src))
 367                                         /* ignore complete line */;
 368
 369                                 src->idx = 0;
 370                         }
 371                         return read_line (src);
 372                 }
 373         }
 374
 375         src->buffer[i] = '\0';
 376         src->length    = i;
 377
 378         return src->buffer;
 379 } /* static char *read_line (conn_t *) */
 380
 381 static void *collect (void *arg)
 382 {
 383         collector_t *this = (collector_t *)arg;
 384
 385         char *buffer = (char *)smalloc (BUFSIZE);
 386
 387         while (1) {
 388                 int loop = 1;
 389
 390                 conn_t *connection;
 391
 392                 pthread_mutex_lock (&conns_mutex);
 393
 394                 while (NULL == conns.head) {
 395                         pthread_cond_wait (&conn_available, &conns_mutex);
 396                 }
 397
 398                 connection = conns.head;
 399                 conns.head = conns.head->next;
 400
 401                 if (NULL == conns.head) {
 402                         conns.tail = NULL;
 403                 }
 404
 405                 this->socket = connection->socket;
 406
 407                 pthread_mutex_unlock (&conns_mutex);
 408
 409                 connection->buffer = buffer;
 410                 connection->idx    = 0;
 411                 connection->length = 0;
 412
 413                 { /* put the socket in non-blocking mode */
 414                         int flags = 0;
 415
 416                         errno = 0;
 417                         if (-1 == fcntl (connection->socket, F_GETFL, &flags)) {
 418                                 char errbuf[1024];
 419                                 log_err ("fcntl() failed: %s",
 420                                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 421                                 loop = 0;
 422                         }
 423
 424                         errno = 0;
 425                         if (-1 == fcntl (connection->socket, F_SETFL, flags | O_NONBLOCK)) {
 426                                 char errbuf[1024];
 427                                 log_err ("fcntl() failed: %s",
 428                                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 429                                 loop = 0;
 430                         }
 431                 }
 432
 433                 while (loop) {
 434                         char *line = read_line (connection);
 435
 436                         if (NULL == line) {
 437                                 loop = 0;
 438                                 break;
 439                         }
 440
 441                         if (':' != line[1]) {
 442                                 log_err ("syntax error in line '%s'", line);
 443                                 continue;
 444                         }
 445
 446                         if ('e' == line[0]) { /* e:<type>:<bytes> */
 447                                 char *ptr  = NULL;
 448                                 char *type = strtok_r (line + 2, ":", &ptr);
 449                                 char *tmp  = strtok_r (NULL, ":", &ptr);
 450                                 int  bytes = 0;
 451
 452                                 if (NULL == tmp) {
 453                                         log_err ("syntax error in line '%s'", line);
 454                                         continue;
 455                                 }
 456
 457                                 bytes = atoi (tmp);
 458
 459                                 pthread_mutex_lock (&count_mutex);
 460                                 type_list_incr (&count, type, 1);
 461                                 pthread_mutex_unlock (&count_mutex);
 462
 463                                 if (bytes > 0) {
 464                                         pthread_mutex_lock (&size_mutex);
 465                                         type_list_incr (&size, type, bytes);
 466                                         pthread_mutex_unlock (&size_mutex);
 467                                 }
 468                         }
 469                         else if ('s' == line[0]) { /* s:<value> */
 470                                 pthread_mutex_lock (&score_mutex);
 471                                 score = (score * (double)score_count + atof (line + 2))
 472                                                 / (double)(score_count + 1);
 473                                 ++score_count;
 474                                 pthread_mutex_unlock (&score_mutex);
 475                         }
 476                         else if ('c' == line[0]) { /* c:<type1>[,<type2>,...] */
 477                                 char *ptr  = NULL;
 478                                 char *type = strtok_r (line + 2, ",", &ptr);
 479
 480                                 do {
 481                                         pthread_mutex_lock (&check_mutex);
 482                                         type_list_incr (&check, type, 1);
 483                                         pthread_mutex_unlock (&check_mutex);
 484                                 } while (NULL != (type = strtok_r (NULL, ",", &ptr)));
 485                         }
 486                         else {
 487                                 log_err ("unknown type '%c'", line[0]);
 488                         }
 489                 } /* while (loop) */
 490
 491                 close (connection->socket);
 492                 free (connection);
 493
 494                 this->socket = -1;
 495
 496                 pthread_mutex_lock (&available_mutex);
 497                 ++available_collectors;
 498                 pthread_mutex_unlock (&available_mutex);
 499
 500                 pthread_cond_signal (&collector_available);
 501         } /* while (1) */
 502
 503         free (buffer);
 504         pthread_exit ((void *)0);
 505 } /* static void *collect (void *) */
 506
 507 static void *open_connection (void *arg)
 508 {
 509         struct sockaddr_un addr;
 510
 511         /* create UNIX socket */
 512         errno = 0;
 513         if (-1 == (connector_socket = socket (PF_UNIX, SOCK_STREAM, 0))) {
 514                 char errbuf[1024];
 515                 disabled = 1;
 516                 log_err ("socket() failed: %s",
 517                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 518                 pthread_exit ((void *)1);
 519         }
 520
 521         addr.sun_family = AF_UNIX;
 522
 523         strncpy (addr.sun_path, SOCK_PATH, (size_t)(UNIX_PATH_MAX - 1));
 524         addr.sun_path[UNIX_PATH_MAX - 1] = '\0';
 525         unlink (addr.sun_path);
 526
 527         errno = 0;
 528         if (-1 == bind (connector_socket, (struct sockaddr *)&addr,
 529                                 offsetof (struct sockaddr_un, sun_path)
 530                                         + strlen(addr.sun_path))) {
 531                 char errbuf[1024];
 532                 disabled = 1;
 533                 connector_socket = -1; /* TODO: close? */
 534                 log_err ("bind() failed: %s",
 535                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 536                 pthread_exit ((void *)1);
 537         }
 538
 539         errno = 0;
 540         if (-1 == listen (connector_socket, 5)) {
 541                 char errbuf[1024];
 542                 disabled = 1;
 543                 connector_socket = -1; /* TODO: close? */
 544                 log_err ("listen() failed: %s",
 545                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 546                 pthread_exit ((void *)1);
 547         }
 548
 549         if ((uid_t) 0 == geteuid ())
 550         {
 551                 struct group sg;
 552                 struct group *grp;
 553                 char grbuf[2048];
 554                 int status;
 555
 556                 grp = NULL;
 557                 status = getgrnam_r (sock_group, &sg, grbuf, sizeof (grbuf), &grp);
 558                 if (status != 0)
 559                 {
 560                         char errbuf[1024];
 561                         log_warn ("getgrnam_r (%s) failed: %s", sock_group,
 562                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 563                 }
 564                 else if (grp == NULL)
 565                 {
 566                         log_warn ("No such group: `%s'", sock_group);
 567                 }
 568                 else
 569                 {
 570                         status = chown (SOCK_PATH, (uid_t) -1, grp->gr_gid);
 571                         if (status != 0)
 572                         {
 573                                 char errbuf[1024];
 574                                 log_warn ("chown (%s, -1, %i) failed: %s",
 575                                                 SOCK_PATH, (int) grp->gr_gid,
 576                                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 577                         }
 578                 }
 579         }
 580         else /* geteuid != 0 */
 581         {
 582                 log_warn ("not running as root");
 583         }
 584
 585         errno = 0;
 586         if (0 != chmod (SOCK_PATH, sock_perms)) {
 587                 char errbuf[1024];
 588                 log_warn ("chmod() failed: %s",
 589                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 590         }
 591
 592         { /* initialize collector threads */
 593                 int i   = 0;
 594                 int err = 0;
 595
 596                 pthread_attr_t ptattr;
 597
 598                 conns.head = NULL;
 599                 conns.tail = NULL;
 600
 601                 pthread_attr_init (&ptattr);
 602                 pthread_attr_setdetachstate (&ptattr, PTHREAD_CREATE_DETACHED);
 603
 604                 available_collectors = max_conns;
 605
 606                 collectors =
 607                         (collector_t **)smalloc (max_conns * sizeof (collector_t *));
 608
 609                 for (i = 0; i < max_conns; ++i) {
 610                         collectors[i] = (collector_t *)smalloc (sizeof (collector_t));
 611                         collectors[i]->socket = -1;
 612
 613                         if (0 != (err = pthread_create (&collectors[i]->thread, &ptattr,
 614                                                         collect, collectors[i]))) {
 615                                 char errbuf[1024];
 616                                 log_err ("pthread_create() failed: %s",
 617                                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 618                                 collectors[i]->thread = (pthread_t) 0;
 619                         }
 620                 }
 621
 622                 pthread_attr_destroy (&ptattr);
 623         }
 624
 625         while (1) {
 626                 int remote = 0;
 627
 628                 conn_t *connection;
 629
 630                 pthread_mutex_lock (&available_mutex);
 631
 632                 while (0 == available_collectors) {
 633                         pthread_cond_wait (&collector_available, &available_mutex);
 634                 }
 635
 636                 --available_collectors;
 637
 638                 pthread_mutex_unlock (&available_mutex);
 639
 640                 do {
 641                         errno = 0;
 642                         if (-1 == (remote = accept (connector_socket, NULL, NULL))) {
 643                                 if (EINTR != errno) {
 644                                         char errbuf[1024];
 645                                         disabled = 1;
 646                                         connector_socket = -1; /* TODO: close? */
 647                                         log_err ("accept() failed: %s",
 648                                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 649                                         pthread_exit ((void *)1);
 650                                 }
 651                         }
 652                 } while (EINTR == errno);
 653
 654                 connection = (conn_t *)smalloc (sizeof (conn_t));
 655
 656                 connection->socket = remote;
 657                 connection->next   = NULL;
 658
 659                 pthread_mutex_lock (&conns_mutex);
 660
 661                 if (NULL == conns.head) {
 662                         conns.head = connection;
 663                         conns.tail = connection;
 664                 }
 665                 else {
 666                         conns.tail->next = connection;
 667                         conns.tail = conns.tail->next;
 668                 }
 669
 670                 pthread_mutex_unlock (&conns_mutex);
 671
 672                 pthread_cond_signal (&conn_available);
 673         }
 674         pthread_exit ((void *)0);
 675 } /* static void *open_connection (void *) */
 676
 677 static int email_init (void)
 678 {
 679         int err = 0;
 680
 681         if (0 != (err = pthread_create (&connector, NULL,
 682                                 open_connection, NULL))) {
 683                 char errbuf[1024];
 684                 disabled = 1;
 685                 log_err ("pthread_create() failed: %s",
 686                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 687                 return (-1);
 688         }
 689
 690         return (0);
 691 } /* int email_init */
 692
 693 static int email_shutdown (void)
 694 {
 695         int i = 0;
 696
 697         if (connector != ((pthread_t) 0)) {
 698                 pthread_kill (connector, SIGTERM);
 699                 connector = (pthread_t) 0;
 700         }
 701
 702         if (connector_socket >= 0) {
 703                 close (connector_socket);
 704                 connector_socket = -1;
 705         }
 706
 707         /* don't allow any more connections to be processed */
 708         pthread_mutex_lock (&conns_mutex);
 709
 710         if (collectors != NULL) {
 711                 for (i = 0; i < max_conns; ++i) {
 712                         if (collectors[i] == NULL)
 713                                 continue;
 714
 715                         if (collectors[i]->thread != ((pthread_t) 0)) {
 716                                 pthread_kill (collectors[i]->thread, SIGTERM);
 717                                 collectors[i]->thread = (pthread_t) 0;
 718                         }
 719
 720                         if (collectors[i]->socket >= 0) {
 721                                 close (collectors[i]->socket);
 722                                 collectors[i]->socket = -1;
 723                         }
 724                 }
 725         } /* if (collectors != NULL) */
 726
 727         pthread_mutex_unlock (&conns_mutex);
 728
 729         unlink (SOCK_PATH);
 730         errno = 0;
 731
 732         return (0);
 733 } /* static void email_shutdown (void) */
 734
 735 static void email_submit (const char *type, const char *type_instance, gauge_t value)
 736 {
 737         value_t values[1];
 738         value_list_t vl = VALUE_LIST_INIT;
 739
 740         values[0].gauge = value;
 741
 742         vl.values = values;
 743         vl.values_len = 1;
 744         vl.time = time (NULL);
 745         strcpy (vl.host, hostname_g);
 746         strcpy (vl.plugin, "email");
 747         strncpy (vl.type_instance, type_instance, sizeof (vl.type_instance));
 748
 749         plugin_dispatch_values (type, &vl);
 750 } /* void email_submit */
 751
 752 /* Copy list l1 to list l2. l2 may partly exist already, but it is assumed
 753  * that neither the order nor the name of any element of either list is
 754  * changed and no elements are deleted. The values of l1 are reset to zero
 755  * after they have been copied to l2. */
 756 static void copy_type_list (type_list_t *l1, type_list_t *l2)
 757 {
 758         type_t *ptr1;
 759         type_t *ptr2;
 760
 761         type_t *last = NULL;
 762
 763         for (ptr1 = l1->head, ptr2 = l2->head; NULL != ptr1;
 764                         ptr1 = ptr1->next, last = ptr2, ptr2 = ptr2->next) {
 765                 if (NULL == ptr2) {
 766                         ptr2 = (type_t *)smalloc (sizeof (type_t));
 767                         ptr2->name = NULL;
 768                         ptr2->next = NULL;
 769
 770                         if (NULL == last) {
 771                                 l2->head = ptr2;
 772                         }
 773                         else {
 774                                 last->next = ptr2;
 775                         }
 776
 777                         l2->tail = ptr2;
 778                 }
 779
 780                 if (NULL == ptr2->name) {
 781                         ptr2->name = sstrdup (ptr1->name);
 782                 }
 783
 784                 ptr2->value = ptr1->value;
 785                 ptr1->value = 0;
 786         }
 787         return;
 788 }
 789
 790 static int email_read (void)
 791 {
 792         type_t *ptr;
 793
 794         double sc;
 795
 796         static type_list_t *cnt;
 797         static type_list_t *sz;
 798         static type_list_t *chk;
 799
 800         if (disabled)
 801                 return (-1);
 802
 803         if (NULL == cnt) {
 804                 cnt = (type_list_t *)smalloc (sizeof (type_list_t));
 805                 cnt->head = NULL;
 806         }
 807
 808         if (NULL == sz) {
 809                 sz = (type_list_t *)smalloc (sizeof (type_list_t));
 810                 sz->head = NULL;
 811         }
 812
 813         if (NULL == chk) {
 814                 chk = (type_list_t *)smalloc (sizeof (type_list_t));
 815                 chk->head = NULL;
 816         }
 817
 818         /* email count */
 819         pthread_mutex_lock (&count_mutex);
 820
 821         copy_type_list (&count, cnt);
 822
 823         pthread_mutex_unlock (&count_mutex);
 824
 825         for (ptr = cnt->head; NULL != ptr; ptr = ptr->next) {
 826                 email_submit ("email_count", ptr->name, ptr->value);
 827         }
 828
 829         /* email size */
 830         pthread_mutex_lock (&size_mutex);
 831
 832         copy_type_list (&size, sz);
 833
 834         pthread_mutex_unlock (&size_mutex);
 835
 836         for (ptr = sz->head; NULL != ptr; ptr = ptr->next) {
 837                 email_submit ("email_size", ptr->name, ptr->value);
 838         }
 839
 840         /* spam score */
 841         pthread_mutex_lock (&score_mutex);
 842
 843         sc = score;
 844         score = 0.0;
 845         score_count = 0;
 846
 847         pthread_mutex_unlock (&score_mutex);
 848
 849         email_submit ("spam_score", "", sc);
 850
 851         /* spam checks */
 852         pthread_mutex_lock (&check_mutex);
 853
 854         copy_type_list (&check, chk);
 855
 856         pthread_mutex_unlock (&check_mutex);
 857
 858         for (ptr = chk->head; NULL != ptr; ptr = ptr->next)
 859                 email_submit ("spam_check", ptr->name, ptr->value);
 860
 861         return (0);
 862 } /* int email_read */
 863
 864 void module_register (void)
 865 {
 866         plugin_register_config ("email", email_config, config_keys, config_keys_num);
 867         plugin_register_init ("email", email_init);
 868         plugin_register_read ("email", email_read);
 869         plugin_register_shutdown ("email", email_shutdown);
 870 } /* void module_register */
 871
 872 /* vim: set sw=4 ts=4 tw=78 noexpandtab : */