src/write_syslog.c

   1 /**
   2  * collectd - src/write_syslog.c
   3  * Copyright (C) 2012       Pierre-Yves Ritschard
   4  * Copyright (C) 2011       Scott Sanders
   5  * Copyright (C) 2009       Paul Sadauskas
   6  * Copyright (C) 2009       Doug MacEachern
   7  * Copyright (C) 2007-2012  Florian octo Forster
   8  * Copyright (C) 2013-2014  Limelight Networks, Inc.
   9  * Copyright (C) 2019       Shirly Radco
  10  * This program is free software; you can redistribute it and/or modify it
  11  * under the terms of the GNU General Public License as published by the
  12  * Free Software Foundation; only version 2 of the License is applicable.
  13  *
  14  * This program is distributed in the hope that it will be useful, but
  15  * WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU General Public License along
  20  * with this program; if not, write to the Free Software Foundation, Inc.,
  21  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  22  *
  23  * Based on the write_graphite plugin. Authors:
  24  *   Florian octo Forster <octo at collectd.org>
  25  *   Doug MacEachern <dougm at hyperic.com>
  26  *   Paul Sadauskas <psadauskas at gmail.com>
  27  *   Scott Sanders <scott at jssjr.com>
  28  *   Pierre-Yves Ritschard <pyr at spootnik.org>
  29  * Based on the write_tsdb plugin. Authors:
  30  *   Brett Hawn <bhawn at llnw.com>
  31  *   Kevin Bowling <kbowling@llnw.com>
  32  * write_syslog. Authors:
  33  *   Shirly Radco <sradco@redhat.com>
  34  **/
  35
  36 /* write_syslog plugin configuration example
  37  *
  38  * <Plugin write_syslog>
  39  *   <Node>
  40  *     Host "localhost"
  41  *     Port "44514"
  42  *     Prefix "collectd"
  43  *     MessageFormat "human"
  44  *     HostTags "["prefix1" "example1"="example1_v"]
  45  *   </Node>
  46  * </Plugin>
  47  *
  48  */
  49
  50 #include "collectd.h"
  51 #include "utils/common/common.h"
  52
  53 #include "plugin.h"
  54 #include "utils_cache.h"
  55 #include "utils_random.h"
  56
  57 #include <netdb.h>
  58
  59 #define WS_DEFAULT_NODE "localhost"
  60
  61 #define WS_DEFAULT_SERVICE "44514"
  62
  63 #define WS_DEFAULT_FORMAT "human"
  64
  65 #define WS_DEFAULT_PREFIX "collectd"
  66
  67 #define WS_DEFAULT_ESCAPE '.'
  68
  69 /* Ethernet - (IPv6 + TCP) = 1500 - (40 + 32) = 1428 */
  70 #define WS_SEND_BUF_SIZE 1428
  71
  72 /*
  73  * Private variables
  74  */
  75 struct ws_callback {
  76   struct addrinfo *ai;
  77   cdtime_t ai_last_update;
  78   int sock_fd;
  79
  80   char *node;
  81   char *service;
  82   char *host_tags;
  83   char *msg_format;
  84   char *metrics_prefix;
  85   bool store_rates;
  86   bool always_append_ds;
  87
  88   char send_buf[WS_SEND_BUF_SIZE];
  89   size_t send_buf_free;
  90   size_t send_buf_fill;
  91   cdtime_t send_buf_init_time;
  92
  93   pthread_mutex_t send_lock;
  94
  95   bool connect_failed_log_enabled;
  96   int connect_dns_failed_attempts_remaining;
  97   cdtime_t next_random_ttl;
  98 };
  99
 100 static cdtime_t resolve_interval;
 101 static cdtime_t resolve_jitter;
 102
 103 /*
 104  * Functions
 105  */
 106 static void ws_reset_buffer(struct ws_callback *cb) {
 107   memset(cb->send_buf, 0, sizeof(cb->send_buf));
 108   cb->send_buf_free = sizeof(cb->send_buf);
 109   cb->send_buf_fill = 0;
 110   cb->send_buf_init_time = cdtime();
 111 }
 112
 113 static int ws_send_buffer(struct ws_callback *cb) {
 114   ssize_t status = 0;
 115
 116   status = swrite(cb->sock_fd, cb->send_buf, strlen(cb->send_buf));
 117   if (status != 0) {
 118     ERROR("write_syslog plugin: send failed with status %zi (%s)", status,
 119           STRERRNO);
 120
 121     if (cb->sock_fd > 0) {
 122       close(cb->sock_fd);
 123       cb->sock_fd = -1;
 124     }
 125
 126     return -1;
 127   }
 128
 129   return 0;
 130 }
 131
 132 /* NOTE: You must hold cb->send_lock when calling this function! */
 133 static int ws_flush_nolock(cdtime_t timeout, struct ws_callback *cb) {
 134   int status;
 135
 136   DEBUG("write_syslog plugin: ws_flush_nolock: timeout = %.3f; "
 137         "send_buf_fill = %" PRIsz ";",
 138         (double)timeout, cb->send_buf_fill);
 139
 140   /* timeout == 0  => flush unconditionally */
 141   if (timeout > 0) {
 142     cdtime_t now;
 143
 144     now = cdtime();
 145     if ((cb->send_buf_init_time + timeout) > now)
 146       return 0;
 147   }
 148
 149   if (cb->send_buf_fill == 0) {
 150     cb->send_buf_init_time = cdtime();
 151     return 0;
 152   }
 153
 154   status = ws_send_buffer(cb);
 155   ws_reset_buffer(cb);
 156
 157   return status;
 158 }
 159
 160 static cdtime_t new_random_ttl(void) {
 161   if (resolve_jitter == 0)
 162     return 0;
 163
 164   return (cdtime_t)cdrand_range(0, (long)resolve_jitter);
 165 }
 166
 167 static int ws_callback_init(struct ws_callback *cb) {
 168   int status;
 169   cdtime_t now;
 170
 171   const char *node = cb->node ? cb->node : WS_DEFAULT_NODE;
 172   const char *service = cb->service ? cb->service : WS_DEFAULT_SERVICE;
 173
 174   if (cb->sock_fd > 0)
 175     return 0;
 176
 177   now = cdtime();
 178   if (cb->ai) {
 179     /* When we are here, we still have the IP in cache.
 180      * If we have remaining attempts without calling the DNS, we update the
 181      * last_update date so we keep the info until next time.
 182      * If there is no more attempts, we need to flush the cache.
 183      */
 184
 185     if ((cb->ai_last_update + resolve_interval + cb->next_random_ttl) < now) {
 186       cb->next_random_ttl = new_random_ttl();
 187       if (cb->connect_dns_failed_attempts_remaining > 0) {
 188         /* Warning : this is run under send_lock mutex.
 189          * This is why we do not use another mutex here.
 190          * */
 191         cb->ai_last_update = now;
 192         cb->connect_dns_failed_attempts_remaining--;
 193       } else {
 194         freeaddrinfo(cb->ai);
 195         cb->ai = NULL;
 196       }
 197     }
 198   }
 199
 200   if (cb->ai == NULL) {
 201     if ((cb->ai_last_update + resolve_interval + cb->next_random_ttl) >= now) {
 202       DEBUG("write_syslog plugin: too many getaddrinfo(%s, %s) failures", node,
 203             service);
 204       return -1;
 205     }
 206     cb->ai_last_update = now;
 207     cb->next_random_ttl = new_random_ttl();
 208
 209     struct addrinfo ai_hints = {
 210         .ai_family = AF_UNSPEC,
 211         .ai_flags = AI_ADDRCONFIG,
 212         .ai_socktype = SOCK_STREAM,
 213     };
 214
 215     status = getaddrinfo(node, service, &ai_hints, &cb->ai);
 216     if (status != 0) {
 217       if (cb->ai) {
 218         freeaddrinfo(cb->ai);
 219         cb->ai = NULL;
 220       }
 221       if (cb->connect_failed_log_enabled) {
 222         ERROR("write_syslog plugin: getaddrinfo(%s, %s) failed: %s", node,
 223               service, gai_strerror(status));
 224         cb->connect_failed_log_enabled = 0;
 225       }
 226       return -1;
 227     }
 228   }
 229
 230   assert(cb->ai != NULL);
 231   for (struct addrinfo *ai = cb->ai; ai != NULL; ai = ai->ai_next) {
 232     cb->sock_fd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
 233     if (cb->sock_fd < 0)
 234       continue;
 235
 236     set_sock_opts(cb->sock_fd);
 237
 238     status = connect(cb->sock_fd, ai->ai_addr, ai->ai_addrlen);
 239     if (status != 0) {
 240       close(cb->sock_fd);
 241       cb->sock_fd = -1;
 242       continue;
 243     }
 244
 245     break;
 246   }
 247
 248   if (cb->sock_fd < 0) {
 249     ERROR("write_syslog plugin: Connecting to %s:%s failed. "
 250           "The last error was: %s",
 251           node, service, STRERRNO);
 252     return -1;
 253   }
 254
 255   if (cb->connect_failed_log_enabled == 0) {
 256     INFO("write_syslog plugin: Connecting to %s:%s succeeded.", node, service);
 257     cb->connect_failed_log_enabled = 1;
 258   }
 259   cb->connect_dns_failed_attempts_remaining = 1;
 260
 261   ws_reset_buffer(cb);
 262
 263   return 0;
 264 }
 265
 266 static void ws_callback_free(void *data) {
 267   struct ws_callback *cb;
 268
 269   if (data == NULL)
 270     return;
 271
 272   cb = data;
 273
 274   pthread_mutex_lock(&cb->send_lock);
 275
 276   ws_flush_nolock(0, cb);
 277
 278   close(cb->sock_fd);
 279   cb->sock_fd = -1;
 280
 281   sfree(cb->node);
 282   sfree(cb->service);
 283   sfree(cb->host_tags);
 284   sfree(cb->msg_format);
 285   sfree(cb->metrics_prefix);
 286
 287   pthread_mutex_unlock(&cb->send_lock);
 288   pthread_mutex_destroy(&cb->send_lock);
 289
 290   sfree(cb);
 291 }
 292
 293 static int ws_flush(cdtime_t timeout,
 294                     const char *identifier __attribute__((unused)),
 295                     user_data_t *user_data) {
 296   struct ws_callback *cb;
 297   int status;
 298
 299   if (user_data == NULL)
 300     return -EINVAL;
 301
 302   cb = user_data->data;
 303
 304   pthread_mutex_lock(&cb->send_lock);
 305
 306   if (cb->sock_fd < 0) {
 307     status = ws_callback_init(cb);
 308     if (status != 0) {
 309       ERROR("write_syslog plugin: ws_callback_init failed.");
 310       pthread_mutex_unlock(&cb->send_lock);
 311       return -1;
 312     }
 313   }
 314
 315   status = ws_flush_nolock(timeout, cb);
 316   pthread_mutex_unlock(&cb->send_lock);
 317
 318   return status;
 319 }
 320
 321 static int ws_format_values(char *ret, size_t ret_len, int ds_num,
 322                             const data_set_t *ds, const value_list_t *vl,
 323                             bool store_rates) {
 324   size_t offset = 0;
 325   int status;
 326   gauge_t *rates = NULL;
 327
 328   assert(strcmp(ds->type, vl->type) == 0);
 329
 330   memset(ret, 0, ret_len);
 331
 332 #define BUFFER_ADD(...)                                                        \
 333   do {                                                                         \
 334     status = snprintf(ret + offset, ret_len - offset, __VA_ARGS__);            \
 335     if (status < 1) {                                                          \
 336       sfree(rates);                                                            \
 337       return -1;                                                               \
 338     } else if (((size_t)status) >= (ret_len - offset)) {                       \
 339       sfree(rates);                                                            \
 340       return -1;                                                               \
 341     } else                                                                     \
 342       offset += ((size_t)status);                                              \
 343   } while (0)
 344
 345   if (ds->ds[ds_num].type == DS_TYPE_GAUGE)
 346     BUFFER_ADD(GAUGE_FORMAT, vl->values[ds_num].gauge);
 347   else if (store_rates) {
 348     if (rates == NULL)
 349       rates = uc_get_rate(ds, vl);
 350     if (rates == NULL) {
 351       WARNING("format_values: "
 352               "uc_get_rate failed.");
 353       return -1;
 354     }
 355     BUFFER_ADD(GAUGE_FORMAT, rates[ds_num]);
 356   } else if (ds->ds[ds_num].type == DS_TYPE_COUNTER)
 357     BUFFER_ADD("%" PRIu64, (uint64_t)vl->values[ds_num].counter);
 358   else if (ds->ds[ds_num].type == DS_TYPE_DERIVE)
 359     BUFFER_ADD("%" PRIi64, vl->values[ds_num].derive);
 360   else if (ds->ds[ds_num].type == DS_TYPE_ABSOLUTE)
 361     BUFFER_ADD("%" PRIu64, vl->values[ds_num].absolute);
 362   else {
 363     ERROR("format_values plugin: Unknown data source type: %i",
 364           ds->ds[ds_num].type);
 365     sfree(rates);
 366     return -1;
 367   }
 368
 369 #undef BUFFER_ADD
 370
 371   sfree(rates);
 372   return 0;
 373 }
 374
 375 static int ws_format_name(char *ret, int ret_len, const value_list_t *vl,
 376                           const struct ws_callback *cb, const char *ds_name) {
 377
 378   if (ds_name != NULL) {
 379     snprintf(ret, ret_len, "%s.%s", vl->type, ds_name);
 380   } else { /* ds_name == NULL */
 381     snprintf(ret, ret_len, "%s", vl->type);
 382   }
 383
 384   return 0;
 385 }
 386
 387 static int ws_send_message(const char *key, const char *value, cdtime_t time,
 388                            struct ws_callback *cb, const char *plugin,
 389                            const char *plugin_instance,
 390                            const char *type_instance, const char *type,
 391                            const char *ds_name, cdtime_t interval,
 392                            const char *host) {
 393   int status;
 394   size_t message_len;
 395   char message[1024];
 396   char rfc3339_timestamp[64];
 397   const char *host_tags = cb->host_tags ? cb->host_tags : "";
 398   const char *host_tags_json_prefix = "";
 399   const char *metrics_prefix =
 400       cb->metrics_prefix ? cb->metrics_prefix : WS_DEFAULT_PREFIX;
 401   const char *msg_format = cb->msg_format ? cb->msg_format : WS_DEFAULT_FORMAT;
 402   int pid;
 403
 404   pid = getpid();
 405
 406   rfc3339_local(rfc3339_timestamp, sizeof(rfc3339_timestamp), time);
 407
 408   /* skip if value is NaN */
 409   if (value[0] == 'n')
 410     return 0;
 411
 412   if (strcasecmp("JSON", msg_format) == 0) {
 413     if (cb->host_tags) {
 414       host_tags_json_prefix = ",";
 415     }
 416     status = snprintf(
 417         /* The metric key-values are are part of the syslog msg, in json
 418            format */
 419         message, sizeof(message),
 420         "<166>1 %s %s collectd %d - - {\"time\":%.0f, \"%s\":{ \"%s\":{ "
 421         "\"%s\":%s }, "
 422         "\"plugin\":\"%s\", \"plugin_instance\":\"%s\", "
 423         "\"type_instance\":\"%s\","
 424         " \"type\":\"%s\", \"interval\":%.0f }, \"hostname\":\"%s\" %s "
 425         "%s}\n",
 426         rfc3339_timestamp, host, pid, CDTIME_T_TO_DOUBLE(time), metrics_prefix,
 427         plugin, key, value, plugin, plugin_instance, type_instance, type,
 428         CDTIME_T_TO_DOUBLE(interval), host, host_tags_json_prefix, host_tags);
 429   } else {
 430     status = snprintf(
 431         /* The metric key-values are part of the syslog structrude data,
 432          * MessageFormat = "human" */
 433         message, sizeof(message),
 434         "<166>1 %s %s collectd %d - [%s value=\"%s\""
 435         " plugin=\"%s\" plugin_instance=\"%s\""
 436         " type_instance=\"%s\" type=\"%s\""
 437         " ds_name=\"%s\" interval=\"%.0f\"] %s %s.%s=\"%s\"\n",
 438         rfc3339_timestamp, host, pid, metrics_prefix, value, plugin,
 439         plugin_instance, type_instance, type, ds_name,
 440         CDTIME_T_TO_DOUBLE(interval), host_tags, plugin, key, value);
 441   }
 442   if (status < 0)
 443     return -1;
 444   message_len = (size_t)status;
 445
 446   if (message_len >= sizeof(message)) {
 447     ERROR("write_syslog plugin: message buffer too small: "
 448           "Need %" PRIsz " bytes.",
 449           message_len + 1);
 450     return -1;
 451   }
 452
 453   pthread_mutex_lock(&cb->send_lock);
 454
 455   if (cb->sock_fd < 0) {
 456     status = ws_callback_init(cb);
 457     if (status != 0) {
 458       ERROR("write_syslog plugin: ws_callback_init failed.");
 459       pthread_mutex_unlock(&cb->send_lock);
 460       return -1;
 461     }
 462   }
 463
 464   if (message_len >= cb->send_buf_free) {
 465     status = ws_flush_nolock(0, cb);
 466     if (status != 0) {
 467       pthread_mutex_unlock(&cb->send_lock);
 468       return status;
 469     }
 470   }
 471
 472   /* Assert that we have enough space for this message. */
 473   assert(message_len < cb->send_buf_free);
 474
 475   /* `message_len + 1' because `message_len' does not include the
 476    * trailing null byte. Neither does `send_buffer_fill'. */
 477   memcpy(cb->send_buf + cb->send_buf_fill, message, message_len + 1);
 478   cb->send_buf_fill += message_len;
 479   cb->send_buf_free -= message_len;
 480
 481   DEBUG("write_syslog plugin: [%s]:%s buf %" PRIsz "/%" PRIsz
 482         " (%.1f %%) \"%s\"",
 483         cb->node, cb->service, cb->send_buf_fill, sizeof(cb->send_buf),
 484         100.0 * ((double)cb->send_buf_fill) / ((double)sizeof(cb->send_buf)),
 485         message);
 486
 487   pthread_mutex_unlock(&cb->send_lock);
 488
 489   return 0;
 490 }
 491
 492 static int ws_write_messages(const data_set_t *ds, const value_list_t *vl,
 493                              struct ws_callback *cb) {
 494   char key[10 * DATA_MAX_NAME_LEN];
 495   char values[512];
 496
 497   int status;
 498
 499   if (0 != strcmp(ds->type, vl->type)) {
 500     ERROR("write_syslog plugin: DS type does not match "
 501           "value list type");
 502     return -1;
 503   }
 504
 505   for (size_t i = 0; i < ds->ds_num; i++) {
 506     const char *ds_name = NULL;
 507
 508     if (cb->always_append_ds || (ds->ds_num > 1))
 509       ds_name = ds->ds[i].name;
 510
 511     /* Copy the identifier to 'key' and escape it. */
 512     status = ws_format_name(key, sizeof(key), vl, cb, ds_name);
 513     if (status != 0) {
 514       ERROR("write_syslog plugin: error with format_name");
 515       return status;
 516     }
 517
 518     escape_string(key, sizeof(key));
 519     /* Convert the values to an ASCII representation and put that into
 520      * 'values'. */
 521     status =
 522         ws_format_values(values, sizeof(values), i, ds, vl, cb->store_rates);
 523     if (status != 0) {
 524       ERROR("write_syslog plugin: error with "
 525             "ws_format_values");
 526       return status;
 527     }
 528
 529     /* Send the message to tcp */
 530     status = ws_send_message(key, values, vl->time, cb, vl->plugin,
 531                              vl->plugin_instance, vl->type_instance, vl->type,
 532                              ds_name, vl->interval, vl->host);
 533     if (status != 0) {
 534       ERROR("write_syslog plugin: error with "
 535             "ws_send_message");
 536       return status;
 537     }
 538   }
 539
 540   return 0;
 541 }
 542
 543 static int ws_write(const data_set_t *ds, const value_list_t *vl,
 544                     user_data_t *user_data) {
 545   struct ws_callback *cb;
 546   int status;
 547
 548   if (user_data == NULL)
 549     return EINVAL;
 550
 551   cb = user_data->data;
 552
 553   status = ws_write_messages(ds, vl, cb);
 554
 555   return status;
 556 }
 557
 558 static int ws_config_tsd(oconfig_item_t *ci) {
 559   struct ws_callback *cb;
 560   char callback_name[DATA_MAX_NAME_LEN];
 561
 562   cb = calloc(1, sizeof(*cb));
 563   if (cb == NULL) {
 564     ERROR("write_syslog plugin: calloc failed.");
 565     return -1;
 566   }
 567   cb->sock_fd = -1;
 568   cb->connect_failed_log_enabled = 1;
 569   cb->next_random_ttl = new_random_ttl();
 570
 571   pthread_mutex_init(&cb->send_lock, NULL);
 572
 573   for (int i = 0; i < ci->children_num; i++) {
 574     oconfig_item_t *child = ci->children + i;
 575
 576     if (strcasecmp("Host", child->key) == 0)
 577       cf_util_get_string(child, &cb->node);
 578     else if (strcasecmp("Port", child->key) == 0)
 579       cf_util_get_service(child, &cb->service);
 580     else if (strcasecmp("MessageFormat", child->key) == 0)
 581       cf_util_get_string(child, &cb->msg_format);
 582     else if (strcasecmp("HostTags", child->key) == 0)
 583       cf_util_get_string(child, &cb->host_tags);
 584     else if (strcasecmp("StoreRates", child->key) == 0)
 585       cf_util_get_boolean(child, &cb->store_rates);
 586     else if (strcasecmp("AlwaysAppendDS", child->key) == 0)
 587       cf_util_get_boolean(child, &cb->always_append_ds);
 588     else if (strcasecmp("Prefix", child->key) == 0)
 589       cf_util_get_string(child, &cb->metrics_prefix);
 590     else {
 591       ERROR("write_syslog plugin: Invalid configuration "
 592             "option: %s.",
 593             child->key);
 594       return -1;
 595     }
 596   }
 597
 598   snprintf(callback_name, sizeof(callback_name), "write_syslog/%s/%s",
 599            cb->node != NULL ? cb->node : WS_DEFAULT_NODE,
 600            cb->service != NULL ? cb->service : WS_DEFAULT_SERVICE);
 601
 602   user_data_t user_data = {.data = cb, .free_func = ws_callback_free};
 603
 604   plugin_register_write(callback_name, ws_write, &user_data);
 605
 606   user_data.free_func = NULL;
 607   plugin_register_flush(callback_name, ws_flush, &user_data);
 608
 609   return 0;
 610 }
 611
 612 static int ws_config(oconfig_item_t *ci) {
 613   if ((resolve_interval == 0) && (resolve_jitter == 0))
 614     resolve_interval = resolve_jitter = plugin_get_interval();
 615
 616   for (int i = 0; i < ci->children_num; i++) {
 617     oconfig_item_t *child = ci->children + i;
 618
 619     if (strcasecmp("Node", child->key) == 0) {
 620       if (ws_config_tsd(child) < 0)
 621         return -1;
 622     } else if (strcasecmp("ResolveInterval", child->key) == 0)
 623       cf_util_get_cdtime(child, &resolve_interval);
 624     else if (strcasecmp("ResolveJitter", child->key) == 0)
 625       cf_util_get_cdtime(child, &resolve_jitter);
 626     else {
 627       ERROR("write_syslog plugin: Invalid configuration "
 628             "option: %s.",
 629             child->key);
 630       return -1;
 631     }
 632   }
 633
 634   return 0;
 635 }
 636
 637 void module_register(void) {
 638   plugin_register_complex_config("write_syslog", ws_config);
 639 }