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