src/zeromq.c

   1 /**
   2  * collectd - src/zeromq.c
   3  * Copyright (C) 2005-2009  Florian octo Forster
   4  * Copyright (C) 2009       Aman Gupta
   5  * Copyright (C) 2010       Julien Ammous
   6  *
   7  * This program is free software; you can redistribute it and/or modify it
   8  * under the terms of the GNU General Public License as published by the
   9  * Free Software Foundation; only version 2 of the License is applicable.
  10  *
  11  * This program is distributed in the hope that it will be useful, but
  12  * WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License along
  17  * with this program; if not, write to the Free Software Foundation, Inc.,
  18  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  19  *
  20  * Authors:
  21  *   Florian octo Forster <octo at verplant.org>
  22  *   Aman Gupta <aman at tmm1.net>
  23  *   Julien Ammous
  24  **/
  25
  26 #include "collectd.h"
  27 #include "common.h" /* auxiliary functions */
  28 #include "plugin.h" /* plugin_register_*, plugin_dispatch_values */
  29 #include "utils_cache.h"
  30 #include "network.h"
  31
  32 /* for htons() */
  33 #if HAVE_ARPA_INET_H
  34 # include <arpa/inet.h>
  35 #endif
  36 #include <pthread.h>
  37 #include <zmq.h>
  38
  39 // copy/paste from network.c ...
  40
  41 static value_list_t     send_buffer_vl = VALUE_LIST_STATIC;
  42
  43 static uint64_t stats_values_not_dispatched = 0;
  44 static uint64_t stats_values_dispatched = 0;
  45
  46 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
  47  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
  48  * +-------+-----------------------+-------------------------------+
  49  * ! Ver.  !                       ! Length                        !
  50  * +-------+-----------------------+-------------------------------+
  51  */
  52 struct part_header_s
  53 {
  54         uint16_t type;
  55         uint16_t length;
  56 };
  57 typedef struct part_header_s part_header_t;
  58
  59 // we do not want to crypt here
  60 #undef HAVE_LIBGCRYPT
  61
  62 static _Bool check_receive_okay (const value_list_t *vl) /* {{{ */
  63 {
  64   uint64_t time_sent = 0;
  65   int status;
  66
  67   status = uc_meta_data_get_unsigned_int (vl,
  68       "network:time_sent", &time_sent);
  69
  70   /* This is a value we already sent. Don't allow it to be received again in
  71    * order to avoid looping. */
  72   if ((status == 0) && (time_sent >= ((uint64_t) vl->time)))
  73     return (0);
  74
  75   return (1);
  76 } /* }}} _Bool check_receive_okay */
  77
  78 static int network_dispatch_values (value_list_t *vl, /* {{{ */
  79     const char *username)
  80 {
  81   int status;
  82
  83   // DEBUG("host: %s", vl->host);
  84   // DEBUG("plugin: %s", vl->plugin);
  85   // DEBUG("plugin_instance: %s", vl->plugin_instance);
  86   // DEBUG("type: %s", vl->type);
  87   // DEBUG("type_instance: %s", vl->type_instance);
  88
  89   if ((vl->time <= 0)
  90       || (strlen (vl->host) <= 0)
  91       || (strlen (vl->plugin) <= 0)
  92       || (strlen (vl->type) <= 0))
  93     return (-EINVAL);
  94
  95   if (!check_receive_okay (vl))
  96   {
  97 #if COLLECT_DEBUG
  98     char name[6*DATA_MAX_NAME_LEN];
  99     FORMAT_VL (name, sizeof (name), vl);
 100     name[sizeof (name) - 1] = 0;
 101     DEBUG ("network plugin: network_dispatch_values: "
 102         "NOT dispatching %s.", name);
 103 #endif
 104     stats_values_not_dispatched++;
 105     return (0);
 106   }
 107
 108   assert (vl->meta == NULL);
 109
 110   vl->meta = meta_data_create ();
 111   if (vl->meta == NULL)
 112   {
 113     ERROR ("network plugin: meta_data_create failed.");
 114     return (-ENOMEM);
 115   }
 116
 117   status = meta_data_add_boolean (vl->meta, "network:received", 1);
 118   if (status != 0)
 119   {
 120     ERROR ("network plugin: meta_data_add_boolean failed.");
 121     meta_data_destroy (vl->meta);
 122     vl->meta = NULL;
 123     return (status);
 124   }
 125
 126   if (username != NULL)
 127   {
 128     status = meta_data_add_string (vl->meta, "network:username", username);
 129     if (status != 0)
 130     {
 131       ERROR ("network plugin: meta_data_add_string failed.");
 132       meta_data_destroy (vl->meta);
 133       vl->meta = NULL;
 134       return (status);
 135     }
 136   }
 137
 138   // DEBUG("dispatching %d values", vl->values_len);
 139   plugin_dispatch_values (vl);
 140   stats_values_dispatched++;
 141
 142   meta_data_destroy (vl->meta);
 143   vl->meta = NULL;
 144
 145   return (0);
 146 } /* }}} int network_dispatch_values */
 147
 148 static int write_part_values (char **ret_buffer, int *ret_buffer_len, const data_set_t *ds, const value_list_t *vl)
 149 {
 150   char *packet_ptr;
 151   int packet_len;
 152   int num_values;
 153
 154   part_header_t pkg_ph;
 155   uint16_t      pkg_num_values;
 156   uint8_t      *pkg_values_types;
 157   value_t      *pkg_values;
 158
 159   int offset;
 160   int i;
 161
 162   num_values = vl->values_len;
 163   packet_len = sizeof (part_header_t) + sizeof (uint16_t)
 164     + (num_values * sizeof (uint8_t))
 165     + (num_values * sizeof (value_t));
 166
 167   if (*ret_buffer_len < packet_len)
 168     return (-1);
 169
 170   pkg_values_types = (uint8_t *) malloc (num_values * sizeof (uint8_t));
 171   if (pkg_values_types == NULL)
 172   {
 173     ERROR ("network plugin: write_part_values: malloc failed.");
 174     return (-1);
 175   }
 176
 177   pkg_values = (value_t *) malloc (num_values * sizeof (value_t));
 178   if (pkg_values == NULL)
 179   {
 180     free (pkg_values_types);
 181     ERROR ("network plugin: write_part_values: malloc failed.");
 182     return (-1);
 183   }
 184
 185   pkg_ph.type = htons (TYPE_VALUES);
 186   pkg_ph.length = htons (packet_len);
 187
 188   pkg_num_values = htons ((uint16_t) vl->values_len);
 189
 190   for (i = 0; i < num_values; i++)
 191   {
 192     pkg_values_types[i] = (uint8_t) ds->ds[i].type;
 193     switch (ds->ds[i].type)
 194     {
 195       case DS_TYPE_COUNTER:
 196         pkg_values[i].counter = htonll (vl->values[i].counter);
 197         break;
 198
 199       case DS_TYPE_GAUGE:
 200         pkg_values[i].gauge = htond (vl->values[i].gauge);
 201         break;
 202
 203       case DS_TYPE_DERIVE:
 204         pkg_values[i].derive = htonll (vl->values[i].derive);
 205         break;
 206
 207       case DS_TYPE_ABSOLUTE:
 208         pkg_values[i].absolute = htonll (vl->values[i].absolute);
 209         break;
 210
 211       default:
 212         free (pkg_values_types);
 213         free (pkg_values);
 214         ERROR ("network plugin: write_part_values: "
 215             "Unknown data source type: %i",
 216             ds->ds[i].type);
 217         return (-1);
 218     } /* switch (ds->ds[i].type) */
 219   } /* for (num_values) */
 220
 221   /*
 222    * Use `memcpy' to write everything to the buffer, because the pointer
 223    * may be unaligned and some architectures, such as SPARC, can't handle
 224    * that.
 225    */
 226   packet_ptr = *ret_buffer;
 227   offset = 0;
 228   memcpy (packet_ptr + offset, &pkg_ph, sizeof (pkg_ph));
 229   offset += sizeof (pkg_ph);
 230   memcpy (packet_ptr + offset, &pkg_num_values, sizeof (pkg_num_values));
 231   offset += sizeof (pkg_num_values);
 232   memcpy (packet_ptr + offset, pkg_values_types, num_values * sizeof (uint8_t));
 233   offset += num_values * sizeof (uint8_t);
 234   memcpy (packet_ptr + offset, pkg_values, num_values * sizeof (value_t));
 235   offset += num_values * sizeof (value_t);
 236
 237   assert (offset == packet_len);
 238
 239   *ret_buffer = packet_ptr + packet_len;
 240   *ret_buffer_len -= packet_len;
 241
 242   free (pkg_values_types);
 243   free (pkg_values);
 244
 245   return (0);
 246 } /* int write_part_values */
 247
 248 static int write_part_number (char **ret_buffer, int *ret_buffer_len,
 249     int type, uint64_t value)
 250 {
 251   char *packet_ptr;
 252   int packet_len;
 253
 254   part_header_t pkg_head;
 255   uint64_t pkg_value;
 256
 257   int offset;
 258
 259   packet_len = sizeof (pkg_head) + sizeof (pkg_value);
 260
 261   if (*ret_buffer_len < packet_len)
 262     return (-1);
 263
 264   pkg_head.type = htons (type);
 265   pkg_head.length = htons (packet_len);
 266   pkg_value = htonll (value);
 267
 268   packet_ptr = *ret_buffer;
 269   offset = 0;
 270   memcpy (packet_ptr + offset, &pkg_head, sizeof (pkg_head));
 271   offset += sizeof (pkg_head);
 272   memcpy (packet_ptr + offset, &pkg_value, sizeof (pkg_value));
 273   offset += sizeof (pkg_value);
 274
 275   assert (offset == packet_len);
 276
 277   *ret_buffer = packet_ptr + packet_len;
 278   *ret_buffer_len -= packet_len;
 279
 280   return (0);
 281 } /* int write_part_number */
 282
 283 static int write_part_string (char **ret_buffer, int *ret_buffer_len,
 284     int type, const char *str, int str_len)
 285 {
 286   char *buffer;
 287   int buffer_len;
 288
 289   uint16_t pkg_type;
 290   uint16_t pkg_length;
 291
 292   int offset;
 293
 294   buffer_len = 2 * sizeof (uint16_t) + str_len + 1;
 295   if (*ret_buffer_len < buffer_len)
 296     return (-1);
 297
 298   pkg_type = htons (type);
 299   pkg_length = htons (buffer_len);
 300
 301   buffer = *ret_buffer;
 302   offset = 0;
 303   memcpy (buffer + offset, (void *) &pkg_type, sizeof (pkg_type));
 304   offset += sizeof (pkg_type);
 305   memcpy (buffer + offset, (void *) &pkg_length, sizeof (pkg_length));
 306   offset += sizeof (pkg_length);
 307   memcpy (buffer + offset, str, str_len);
 308   offset += str_len;
 309   memset (buffer + offset, '\0', 1);
 310   offset += 1;
 311
 312   assert (offset == buffer_len);
 313
 314   *ret_buffer = buffer + buffer_len;
 315   *ret_buffer_len -= buffer_len;
 316
 317   return (0);
 318 } /* int write_part_string */
 319
 320 static int parse_part_values (void **ret_buffer, size_t *ret_buffer_len,
 321     value_t **ret_values, int *ret_num_values)
 322 {
 323   char *buffer = *ret_buffer;
 324   size_t buffer_len = *ret_buffer_len;
 325
 326   uint16_t tmp16;
 327   size_t exp_size;
 328   int   i;
 329
 330   uint16_t pkg_length;
 331   uint16_t pkg_type;
 332   uint16_t pkg_numval;
 333
 334   uint8_t *pkg_types;
 335   value_t *pkg_values;
 336
 337   if (buffer_len < 15)
 338   {
 339     NOTICE ("network plugin: packet is too short: "
 340         "buffer_len = %zu", buffer_len);
 341     return (-1);
 342   }
 343
 344   memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 345   buffer += sizeof (tmp16);
 346   pkg_type = ntohs (tmp16);
 347
 348   memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 349   buffer += sizeof (tmp16);
 350   pkg_length = ntohs (tmp16);
 351
 352   memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 353   buffer += sizeof (tmp16);
 354   pkg_numval = ntohs (tmp16);
 355
 356   assert (pkg_type == TYPE_VALUES);
 357
 358   exp_size = 3 * sizeof (uint16_t)
 359     + pkg_numval * (sizeof (uint8_t) + sizeof (value_t));
 360   if ((buffer_len < 0) || (buffer_len < exp_size))
 361   {
 362     WARNING ("network plugin: parse_part_values: "
 363         "Packet too short: "
 364         "Chunk of size %zu expected, "
 365         "but buffer has only %zu bytes left.",
 366         exp_size, buffer_len);
 367     return (-1);
 368   }
 369
 370   if (pkg_length != exp_size)
 371   {
 372     WARNING ("network plugin: parse_part_values: "
 373         "Length and number of values "
 374         "in the packet don't match.");
 375     return (-1);
 376   }
 377
 378   pkg_types = (uint8_t *) malloc (pkg_numval * sizeof (uint8_t));
 379   pkg_values = (value_t *) malloc (pkg_numval * sizeof (value_t));
 380   if ((pkg_types == NULL) || (pkg_values == NULL))
 381   {
 382     sfree (pkg_types);
 383     sfree (pkg_values);
 384     ERROR ("network plugin: parse_part_values: malloc failed.");
 385     return (-1);
 386   }
 387
 388   memcpy ((void *) pkg_types, (void *) buffer, pkg_numval * sizeof (uint8_t));
 389   buffer += pkg_numval * sizeof (uint8_t);
 390   memcpy ((void *) pkg_values, (void *) buffer, pkg_numval * sizeof (value_t));
 391   buffer += pkg_numval * sizeof (value_t);
 392
 393   for (i = 0; i < pkg_numval; i++)
 394   {
 395     switch (pkg_types[i])
 396     {
 397       case DS_TYPE_COUNTER:
 398         pkg_values[i].counter = (counter_t) ntohll (pkg_values[i].counter);
 399         break;
 400
 401       case DS_TYPE_GAUGE:
 402         pkg_values[i].gauge = (gauge_t) ntohd (pkg_values[i].gauge);
 403         break;
 404
 405       case DS_TYPE_DERIVE:
 406         pkg_values[i].derive = (derive_t) ntohll (pkg_values[i].derive);
 407         break;
 408
 409       case DS_TYPE_ABSOLUTE:
 410         pkg_values[i].absolute = (absolute_t) ntohll (pkg_values[i].absolute);
 411         break;
 412
 413       default:
 414         NOTICE ("network plugin: parse_part_values: "
 415       "Don't know how to handle data source type %"PRIu8,
 416       pkg_types[i]);
 417         sfree (pkg_types);
 418         sfree (pkg_values);
 419         return (-1);
 420     } /* switch (pkg_types[i]) */
 421   }
 422
 423   *ret_buffer     = buffer;
 424   *ret_buffer_len = buffer_len - pkg_length;
 425   *ret_num_values = pkg_numval;
 426   *ret_values     = pkg_values;
 427
 428   sfree (pkg_types);
 429
 430   return (0);
 431 } /* int parse_part_values */
 432
 433 static int add_to_buffer (char *buffer, int buffer_size, /* {{{ */
 434     value_list_t *vl_def,
 435     const data_set_t *ds, const value_list_t *vl)
 436 {
 437   char *buffer_orig = buffer;
 438
 439   if (write_part_string (&buffer, &buffer_size, TYPE_HOST, vl->host, strlen (vl->host)) != 0)
 440     return (-1);
 441
 442   if (write_part_number (&buffer, &buffer_size, TYPE_TIME, (uint64_t) vl->time))
 443     return (-1);
 444
 445   if (write_part_number (&buffer, &buffer_size, TYPE_INTERVAL, (uint64_t) vl->interval))
 446     return (-1);
 447
 448   if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN, vl->plugin, strlen (vl->plugin)) != 0)
 449     return (-1);
 450
 451   if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN_INSTANCE, vl->plugin_instance, strlen (vl->plugin_instance)) != 0)
 452     return (-1);
 453
 454   if (write_part_string (&buffer, &buffer_size, TYPE_TYPE, vl->type, strlen (vl->type)) != 0)
 455     return (-1);
 456
 457   if (write_part_string (&buffer, &buffer_size, TYPE_TYPE_INSTANCE, vl->type_instance, strlen (vl->type_instance)) != 0)
 458     return (-1);
 459
 460   if (write_part_values (&buffer, &buffer_size, ds, vl) != 0)
 461     return (-1);
 462
 463   return (buffer - buffer_orig);
 464 } /* }}} int add_to_buffer */
 465
 466 static int parse_part_number (void **ret_buffer, size_t *ret_buffer_len,
 467     uint64_t *value)
 468 {
 469   char *buffer = *ret_buffer;
 470   size_t buffer_len = *ret_buffer_len;
 471
 472   uint16_t tmp16;
 473   uint64_t tmp64;
 474   size_t exp_size = 2 * sizeof (uint16_t) + sizeof (uint64_t);
 475
 476   uint16_t pkg_length;
 477   uint16_t pkg_type;
 478
 479   if ((buffer_len < 0) || ((size_t) buffer_len < exp_size))
 480   {
 481     WARNING ("network plugin: parse_part_number: "
 482         "Packet too short: "
 483         "Chunk of size %zu expected, "
 484         "but buffer has only %zu bytes left.",
 485         exp_size, buffer_len);
 486     return (-1);
 487   }
 488
 489   memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 490   buffer += sizeof (tmp16);
 491   pkg_type = ntohs (tmp16);
 492
 493   memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 494   buffer += sizeof (tmp16);
 495   pkg_length = ntohs (tmp16);
 496
 497   memcpy ((void *) &tmp64, buffer, sizeof (tmp64));
 498   buffer += sizeof (tmp64);
 499   *value = ntohll (tmp64);
 500
 501   *ret_buffer = buffer;
 502   *ret_buffer_len = buffer_len - pkg_length;
 503
 504   return (0);
 505 } /* int parse_part_number */
 506
 507 static int parse_part_string (void **ret_buffer, size_t *ret_buffer_len,
 508     char *output, int output_len)
 509 {
 510   char *buffer = *ret_buffer;
 511   size_t buffer_len = *ret_buffer_len;
 512
 513   uint16_t tmp16;
 514   size_t header_size = 2 * sizeof (uint16_t);
 515
 516   uint16_t pkg_length;
 517   uint16_t pkg_type;
 518
 519   if ((buffer_len < 0) || (buffer_len < header_size))
 520   {
 521     WARNING ("network plugin: parse_part_string: "
 522         "Packet too short: "
 523         "Chunk of at least size %zu expected, "
 524         "but buffer has only %zu bytes left.",
 525         header_size, buffer_len);
 526     return (-1);
 527   }
 528
 529   memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 530   buffer += sizeof (tmp16);
 531   pkg_type = ntohs (tmp16);
 532
 533   memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 534   buffer += sizeof (tmp16);
 535   pkg_length = ntohs (tmp16);
 536
 537   /* Check that packet fits in the input buffer */
 538   if (pkg_length > buffer_len)
 539   {
 540     WARNING ("network plugin: parse_part_string: "
 541         "Packet too big: "
 542         "Chunk of size %"PRIu16" received, "
 543         "but buffer has only %zu bytes left.",
 544         pkg_length, buffer_len);
 545     return (-1);
 546   }
 547
 548   /* Check that pkg_length is in the valid range */
 549   if (pkg_length <= header_size)
 550   {
 551     WARNING ("network plugin: parse_part_string: "
 552         "Packet too short: "
 553         "Header claims this packet is only %hu "
 554         "bytes long.", pkg_length);
 555     return (-1);
 556   }
 557
 558   /* Check that the package data fits into the output buffer.
 559    * The previous if-statement ensures that:
 560    * `pkg_length > header_size' */
 561   if ((output_len < 0)
 562       || ((size_t) output_len < ((size_t) pkg_length - header_size)))
 563   {
 564     WARNING ("network plugin: parse_part_string: "
 565         "Output buffer too small.");
 566     return (-1);
 567   }
 568
 569   /* All sanity checks successfull, let's copy the data over */
 570   output_len = pkg_length - header_size;
 571   memcpy ((void *) output, (void *) buffer, output_len);
 572   buffer += output_len;
 573
 574   /* For some very weird reason '\0' doesn't do the trick on SPARC in
 575    * this statement. */
 576   if (output[output_len - 1] != 0)
 577   {
 578     WARNING ("network plugin: parse_part_string: "
 579         "Received string does not end "
 580         "with a NULL-byte.");
 581     return (-1);
 582   }
 583
 584   *ret_buffer = buffer;
 585   *ret_buffer_len = buffer_len - pkg_length;
 586
 587   return (0);
 588 } /* int parse_part_string */
 589
 590
 591 static int parse_packet (void *se, /* {{{ */
 592     void *buffer, size_t buffer_size, int flags,
 593     const char *username)
 594 {
 595   int status;
 596
 597   value_list_t vl = VALUE_LIST_INIT;
 598   notification_t n;
 599
 600 #if HAVE_LIBGCRYPT
 601   int packet_was_signed = (flags & PP_SIGNED);
 602         int packet_was_encrypted = (flags & PP_ENCRYPTED);
 603   int printed_ignore_warning = 0;
 604 #endif /* HAVE_LIBGCRYPT */
 605
 606
 607   memset (&vl, '\0', sizeof (vl));
 608   memset (&n, '\0', sizeof (n));
 609   status = 0;
 610
 611   while ((status == 0) && (0 < buffer_size)
 612       && ((unsigned int) buffer_size > sizeof (part_header_t)))
 613   {
 614     uint16_t pkg_length;
 615     uint16_t pkg_type;
 616
 617     memcpy ((void *) &pkg_type,
 618         (void *) buffer,
 619         sizeof (pkg_type));
 620     memcpy ((void *) &pkg_length,
 621         (void *) (buffer + sizeof (pkg_type)),
 622         sizeof (pkg_length));
 623
 624     pkg_length = ntohs (pkg_length);
 625     pkg_type = ntohs (pkg_type);
 626
 627     if (pkg_length > buffer_size)
 628       break;
 629     /* Ensure that this loop terminates eventually */
 630     if (pkg_length < (2 * sizeof (uint16_t)))
 631       break;
 632
 633     if (pkg_type == TYPE_ENCR_AES256)
 634     {
 635       // status = parse_part_encr_aes256 (se,
 636       //     &buffer, &buffer_size, flags);
 637       // if (status != 0)
 638       // {
 639         ERROR ("network plugin: Decrypting AES256 "
 640             "part failed "
 641             "with status %i.", status);
 642         break;
 643       // }
 644     }
 645 #if HAVE_LIBGCRYPT
 646     else if ((se->data.server.security_level == SECURITY_LEVEL_ENCRYPT)
 647         && (packet_was_encrypted == 0))
 648     {
 649       if (printed_ignore_warning == 0)
 650       {
 651         INFO ("network plugin: Unencrypted packet or "
 652             "part has been ignored.");
 653         printed_ignore_warning = 1;
 654       }
 655       buffer = ((char *) buffer) + pkg_length;
 656       continue;
 657     }
 658 #endif /* HAVE_LIBGCRYPT */
 659     else if (pkg_type == TYPE_SIGN_SHA256)
 660     {
 661       // status = parse_part_sign_sha256 (se,
 662       //                                   &buffer, &buffer_size, flags);
 663       // if (status != 0)
 664       // {
 665         ERROR ("network plugin: Verifying HMAC-SHA-256 "
 666             "signature failed "
 667             "with status %i.", status);
 668         break;
 669       // }
 670     }
 671 #if HAVE_LIBGCRYPT
 672     else if ((se->data.server.security_level == SECURITY_LEVEL_SIGN)
 673         && (packet_was_encrypted == 0)
 674         && (packet_was_signed == 0))
 675     {
 676       if (printed_ignore_warning == 0)
 677       {
 678         INFO ("network plugin: Unsigned packet or "
 679             "part has been ignored.");
 680         printed_ignore_warning = 1;
 681       }
 682       buffer = ((char *) buffer) + pkg_length;
 683       continue;
 684     }
 685 #endif /* HAVE_LIBGCRYPT */
 686     else if (pkg_type == TYPE_VALUES)
 687     {
 688       status = parse_part_values (&buffer, &buffer_size,
 689           &vl.values, &vl.values_len);
 690       if (status != 0)
 691         break;
 692
 693       network_dispatch_values (&vl, username);
 694
 695       sfree (vl.values);
 696     }
 697     else if (pkg_type == TYPE_TIME)
 698     {
 699       uint64_t tmp = 0;
 700       status = parse_part_number (&buffer, &buffer_size,
 701           &tmp);
 702       if (status == 0)
 703       {
 704         vl.time = (time_t) tmp;
 705         n.time = (time_t) tmp;
 706       }
 707     }
 708     else if (pkg_type == TYPE_INTERVAL)
 709     {
 710       uint64_t tmp = 0;
 711       status = parse_part_number (&buffer, &buffer_size,
 712           &tmp);
 713       if (status == 0)
 714         vl.interval = (int) tmp;
 715     }
 716     else if (pkg_type == TYPE_HOST)
 717     {
 718       status = parse_part_string (&buffer, &buffer_size,
 719           vl.host, sizeof (vl.host));
 720       if (status == 0)
 721         sstrncpy (n.host, vl.host, sizeof (n.host));
 722     }
 723     else if (pkg_type == TYPE_PLUGIN)
 724     {
 725       status = parse_part_string (&buffer, &buffer_size,
 726           vl.plugin, sizeof (vl.plugin));
 727       if (status == 0)
 728         sstrncpy (n.plugin, vl.plugin,
 729             sizeof (n.plugin));
 730     }
 731     else if (pkg_type == TYPE_PLUGIN_INSTANCE)
 732     {
 733       status = parse_part_string (&buffer, &buffer_size,
 734           vl.plugin_instance,
 735           sizeof (vl.plugin_instance));
 736       if (status == 0)
 737         sstrncpy (n.plugin_instance,
 738             vl.plugin_instance,
 739             sizeof (n.plugin_instance));
 740     }
 741     else if (pkg_type == TYPE_TYPE)
 742     {
 743       status = parse_part_string (&buffer, &buffer_size,
 744           vl.type, sizeof (vl.type));
 745       if (status == 0)
 746         sstrncpy (n.type, vl.type, sizeof (n.type));
 747     }
 748     else if (pkg_type == TYPE_TYPE_INSTANCE)
 749     {
 750       status = parse_part_string (&buffer, &buffer_size,
 751           vl.type_instance,
 752           sizeof (vl.type_instance));
 753       if (status == 0)
 754         sstrncpy (n.type_instance, vl.type_instance,
 755             sizeof (n.type_instance));
 756     }
 757     else if (pkg_type == TYPE_MESSAGE)
 758     {
 759       status = parse_part_string (&buffer, &buffer_size,
 760           n.message, sizeof (n.message));
 761
 762       if (status != 0)
 763       {
 764         /* do nothing */
 765       }
 766       else if ((n.severity != NOTIF_FAILURE)
 767           && (n.severity != NOTIF_WARNING)
 768           && (n.severity != NOTIF_OKAY))
 769       {
 770         INFO ("network plugin: "
 771             "Ignoring notification with "
 772             "unknown severity %i.",
 773             n.severity);
 774       }
 775       else if (n.time <= 0)
 776       {
 777         INFO ("network plugin: "
 778             "Ignoring notification with "
 779             "time == 0.");
 780       }
 781       else if (strlen (n.message) <= 0)
 782       {
 783         INFO ("network plugin: "
 784             "Ignoring notification with "
 785             "an empty message.");
 786       }
 787       else
 788       {
 789         plugin_dispatch_notification (&n);
 790       }
 791     }
 792     else if (pkg_type == TYPE_SEVERITY)
 793     {
 794       uint64_t tmp = 0;
 795       status = parse_part_number (&buffer, &buffer_size,
 796           &tmp);
 797       if (status == 0)
 798         n.severity = (int) tmp;
 799     }
 800     else
 801     {
 802       DEBUG ("network plugin: parse_packet: Unknown part"
 803           " type: 0x%04hx", pkg_type);
 804       buffer = ((char *) buffer) + pkg_length;
 805     }
 806   } /* while (buffer_size > sizeof (part_header_t)) */
 807
 808   if (status == 0 && buffer_size > 0)
 809     WARNING ("network plugin: parse_packet: Received truncated "
 810         "packet, try increasing `MaxPacketSize'");
 811
 812   return (status);
 813 } /* }}} int parse_packet */
 814
 815
 816
 817
 818
 819
 820
 821
 822 ////////////////////////////
 823 //// END OF COPY / PASTE ///
 824 ////////////////////////////
 825
 826
 827
 828
 829 static const char *config_keys[] =
 830 {
 831   "ListenOn",
 832   "SendDataTo"
 833 };
 834 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
 835
 836 // config data
 837 static char *zmq_listen_on = NULL;
 838 static char *zmq_send_to = NULL;
 839
 840 // private data
 841 static int thread_running = 1;
 842 static pthread_t listen_thread_id;
 843 static void *zmq_context = NULL;
 844 static void *push_socket = NULL;
 845 static void *pull_socket = NULL;
 846
 847 // return values:
 848 // 0  => success
 849 // -1 => error
 850 //
 851
 852 static int my_config (const char *key, const char *value)
 853 {
 854   if( strcasecmp(key, "ListenOn") == 0 ) {
 855     if( zmq_listen_on != NULL ) {
 856       free(zmq_listen_on);
 857       zmq_listen_on = NULL;
 858     }
 859
 860     if( (zmq_listen_on = strdup(value)) == NULL ) {
 861       return 1;
 862     }
 863   }
 864   else if( strcasecmp(key, "SendDataTo") == 0 ) {
 865     if( zmq_send_to != NULL ) {
 866       free(zmq_send_to);
 867       zmq_send_to = NULL;
 868     }
 869
 870     if( (zmq_send_to = strdup(value)) == NULL ) {
 871       return 1;
 872     }
 873   }
 874
 875
 876   return 0;
 877 }
 878
 879 static void *receive_thread (void __attribute__((unused)) *arg)
 880 {
 881   int status;
 882   zmq_msg_t msg;
 883   char *data = NULL;
 884   size_t data_size;
 885
 886   while( thread_running ) {
 887     status = zmq_msg_init(&msg);
 888     assert( status == 0 );
 889     if( zmq_recv(pull_socket, &msg, 0) != 0 ) {
 890       WARNING("zmq_recv : %s", zmq_strerror(errno));
 891       continue;
 892     }
 893
 894     data = zmq_msg_data(&msg);
 895     data_size = zmq_msg_size(&msg);
 896
 897     parse_packet(NULL, data, data_size,
 898       /* flags = */ 0,
 899       /* username = */ NULL);
 900
 901     DEBUG("ZeroMQ: received data, parse returned %d", status);
 902
 903     if( zmq_msg_close(&msg) != 0 ) {
 904       ERROR("zmq_msg_close : %s", zmq_strerror(errno));
 905     }
 906
 907   }
 908
 909   return NULL;
 910 }
 911
 912 static int plugin_init (void)
 913 {
 914   int major, minor, patch, status;
 915   zmq_version (&major, &minor, &patch);
 916
 917
 918   // init zeromq (1 I/O thread)
 919   zmq_context = zmq_init(1);
 920   if( zmq_context == NULL ) {
 921     ERROR("zmq_init : %s", zmq_strerror(errno));
 922     return 1;
 923   }
 924
 925   // start listening socket
 926   if( zmq_listen_on != NULL ) {
 927     pull_socket = zmq_socket(zmq_context, ZMQ_PULL);
 928     if( pull_socket == NULL ) {
 929       ERROR("zmq_socket : %s", zmq_strerror(errno));
 930       return 1;
 931     }
 932
 933     // and bind it to the inbound queue
 934     if( zmq_bind(pull_socket, zmq_listen_on) != 0 ) {
 935       ERROR("zmq_bind : %s", zmq_strerror(errno));
 936       return 1;
 937     }
 938
 939     // listenining thread
 940     status = pthread_create(&listen_thread_id,
 941       NULL /* no attributes */,
 942       receive_thread,
 943       NULL /* no argument */);
 944
 945     if( status != 0 ) {
 946       ERROR("network: pthread_create failed: %s", strerror(errno));
 947       return 1;
 948     }
 949
 950     INFO("ZeroMQ listening on %s", zmq_listen_on);
 951   }
 952
 953   // start send socket
 954   if( zmq_send_to != NULL ) {
 955     push_socket = zmq_socket(zmq_context, ZMQ_PUSH);
 956
 957     if( push_socket == NULL ) {
 958       ERROR("zmq_socket : %s", zmq_strerror(errno));
 959       return 1;
 960     }
 961
 962     // and connect to remote host
 963     if( zmq_connect(push_socket, zmq_send_to) != 0 ) {
 964       ERROR("zmq_connect : %s", zmq_strerror(errno));
 965       return 1;
 966     }
 967
 968     INFO("ZeroMQ pushing to %s", zmq_send_to);
 969   }
 970
 971
 972
 973   INFO("ZeroMQ plugin initialized (zeromq v%d.%d.%d).", major, minor, patch);
 974   return 0;
 975 }
 976
 977
 978
 979 static int write_notification (const notification_t *n, user_data_t __attribute__((unused)) *user_data)
 980 {
 981   DEBUG("ZeroMQ: received notification, not implemented yet");
 982   return 0;
 983 }
 984
 985 void free_data(void *data, void *hint)
 986 {
 987   free(data);
 988 }
 989
 990 #define PACKET_SIZE   512
 991
 992 static int write_value (const data_set_t *ds, const value_list_t *vl,
 993                 user_data_t __attribute__((unused)) *user_data)
 994 {
 995   zmq_msg_t msg;
 996   char      *send_buffer;
 997   int       send_buffer_size = PACKET_SIZE, real_size;
 998
 999   // do nothing if no socket open
1000   if( zmq_send_to == NULL )
1001     return 0;
1002
1003   send_buffer = malloc(PACKET_SIZE);
1004   if( send_buffer == NULL ) {
1005     ERROR("Unable to allocate memory for send_buffer, aborting write");
1006     return 1;
1007   }
1008
1009   // empty buffer
1010   memset(send_buffer, 0, PACKET_SIZE);
1011
1012   real_size = add_to_buffer(send_buffer, send_buffer_size, &send_buffer_vl, ds, vl);
1013
1014   // zeromq will free the memory when needed by calling the free_data function
1015   if( zmq_msg_init_data(&msg, send_buffer, real_size, free_data, NULL) != 0 ) {
1016     ERROR("zmq_msg_init : %s", zmq_strerror(errno));
1017     return 1;
1018   }
1019
1020   // try to send the message
1021   if( zmq_send(push_socket, &msg, ZMQ_NOBLOCK) != 0 ) {
1022     if( errno == EAGAIN ) {
1023       WARNING("ZeroMQ: Cannot send message, queue is full");
1024     }
1025     else {
1026       ERROR("zmq_send : %s", zmq_strerror(errno));
1027       return 1;
1028     }
1029   }
1030
1031   DEBUG("ZeroMQ: data sent");
1032
1033   return 0;
1034 }
1035
1036 static int my_shutdown (void)
1037 {
1038   if( zmq_context ) {
1039
1040     thread_running = 0;
1041
1042     DEBUG("ZeroMQ: shutting down");
1043
1044     if( zmq_term(zmq_context) != 0 ) {
1045       ERROR("zmq_term : %s", zmq_strerror(errno));
1046       return 1;
1047     }
1048
1049     pthread_join(listen_thread_id, NULL);
1050   }
1051
1052   return 0;
1053 }
1054
1055 void module_register (void)
1056 {
1057   plugin_register_config("zeromq", my_config, config_keys, config_keys_num);
1058   plugin_register_init("zeromq", plugin_init);
1059   plugin_register_write("zeromq", write_value,
1060       /* user_data = */ NULL);
1061   plugin_register_notification ("network", write_notification,
1062       /* user_data = */ NULL);
1063   plugin_register_shutdown ("zeromq", my_shutdown);
1064 }