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