X-Git-Url: https://git.verplant.org/?a=blobdiff_plain;f=src%2Fwrite_riemann.c;h=2936dfa46f5ab91f158d3092baa164cdeab39140;hb=0d80072344ea5322c34f716ed77a5319a62bfc96;hp=3429976fd0faff0d8670178f55e0beb6711634dd;hpb=0ed69e797f1e173dd4386f419f1f65bf2afdcf51;p=collectd.git diff --git a/src/write_riemann.c b/src/write_riemann.c index 3429976f..2936dfa4 100644 --- a/src/write_riemann.c +++ b/src/write_riemann.c @@ -37,6 +37,7 @@ #include "common.h" #include "configfile.h" #include "utils_cache.h" +#include "utils_complain.h" #include "write_riemann_threshold.h" #define RIEMANN_HOST "localhost" @@ -45,6 +46,7 @@ #define RIEMANN_BATCH_MAX 8192 struct riemann_host { + c_complain_t init_complaint; char *name; char *event_service_prefix; pthread_mutex_t lock; @@ -60,11 +62,13 @@ struct riemann_host { double ttl_factor; cdtime_t batch_init; int batch_max; + int batch_timeout; int reference_count; riemann_message_t *batch_msg; char *tls_ca_file; char *tls_cert_file; char *tls_key_file; + struct timeval timeout; }; static char **riemann_tags; @@ -92,12 +96,25 @@ static int wrr_connect(struct riemann_host *host) /* {{{ */ RIEMANN_CLIENT_OPTION_TLS_KEY_FILE, host->tls_key_file, RIEMANN_CLIENT_OPTION_NONE); if (host->client == NULL) { - WARNING("write_riemann plugin: Unable to connect to Riemann at %s:%d", - node, port); + c_complain (LOG_ERR, &host->init_complaint, + "write_riemann plugin: Unable to connect to Riemann at %s:%d", + node, port); return -1; } - DEBUG("write_riemann plugin: got a successful connection for: %s:%d", - node, port); + if (host->timeout.tv_sec != 0) { + if (riemann_client_set_timeout(host->client, &host->timeout) != 0) { + riemann_client_free(host->client); + host->client = NULL; + c_complain (LOG_ERR, &host->init_complaint, + "write_riemann plugin: Unable to connect to Riemann at %s:%d", + node, port); + return -1; + } + } + + c_release (LOG_INFO, &host->init_complaint, + "write_riemann plugin: Successfully connected to %s:%d", + node, port); return 0; } /* }}} int wrr_connect */ @@ -119,19 +136,18 @@ static int wrr_disconnect(struct riemann_host *host) /* {{{ */ * * Acquires the host lock, disconnects on errors. */ -static int wrr_send(struct riemann_host *host, riemann_message_t *msg) /* {{{ */ +static int wrr_send_nolock(struct riemann_host *host, riemann_message_t *msg) /* {{{ */ { int status = 0; - pthread_mutex_lock (&host->lock); status = wrr_connect(host); - if (status != 0) + if (status != 0) { return status; + } status = riemann_client_send_message(host->client, msg); if (status != 0) { wrr_disconnect(host); - pthread_mutex_unlock(&host->lock); return status; } @@ -147,16 +163,24 @@ static int wrr_send(struct riemann_host *host, riemann_message_t *msg) /* {{{ */ if (response == NULL) { wrr_disconnect(host); - pthread_mutex_unlock(&host->lock); return errno; } riemann_message_free(response); } - pthread_mutex_unlock (&host->lock); return 0; } /* }}} int wrr_send */ +static int wrr_send(struct riemann_host *host, riemann_message_t *msg) +{ + int status = 0; + + pthread_mutex_lock (&host->lock); + status = wrr_send_nolock(host, msg); + pthread_mutex_unlock (&host->lock); + return status; +} + static riemann_message_t *wrr_notification_to_message(struct riemann_host *host, /* {{{ */ notification_t const *n) { @@ -437,30 +461,16 @@ static int wrr_batch_flush_nolock(cdtime_t timeout, cdtime_t now; int status = 0; + now = cdtime(); if (timeout > 0) { - now = cdtime(); - if ((host->batch_init + timeout) > now) + if ((host->batch_init + timeout) > now) { return status; + } } - wrr_send(host, host->batch_msg); + wrr_send_nolock(host, host->batch_msg); riemann_message_free(host->batch_msg); - if (host->client_type != RIEMANN_CLIENT_UDP) - { - riemann_message_t *response; - - response = riemann_client_recv_message(host->client); - - if (!response) - { - wrr_disconnect(host); - return errno; - } - - riemann_message_free(response); - } - - host->batch_init = cdtime(); + host->batch_init = now; host->batch_msg = NULL; return status; } @@ -479,8 +489,11 @@ static int wrr_batch_flush(cdtime_t timeout, pthread_mutex_lock(&host->lock); status = wrr_batch_flush_nolock(timeout, host); if (status != 0) - ERROR("write_riemann plugin: riemann_client_send failed with status %i", - status); + c_complain (LOG_ERR, &host->init_complaint, + "write_riemann plugin: riemann_client_send failed with status %i", + status); + else + c_release (LOG_DEBUG, &host->init_complaint, "write_riemann plugin: batch sent."); pthread_mutex_unlock(&host->lock); return status; @@ -494,6 +507,7 @@ static int wrr_batch_add_value_list(struct riemann_host *host, /* {{{ */ riemann_message_t *msg; size_t len; int ret; + cdtime_t timeout; msg = wrr_value_list_to_message(host, ds, vl, statuses); if (msg == NULL) @@ -525,7 +539,12 @@ static int wrr_batch_add_value_list(struct riemann_host *host, /* {{{ */ ret = 0; if ((host->batch_max < 0) || (((size_t) host->batch_max) <= len)) { ret = wrr_batch_flush_nolock(0, host); - } + } else { + if (host->batch_timeout > 0) { + timeout = TIME_T_TO_CDTIME_T((time_t)host->batch_timeout); + ret = wrr_batch_flush_nolock(timeout, host); + } + } pthread_mutex_unlock(&host->lock); return ret; @@ -549,8 +568,12 @@ static int wrr_notification(const notification_t *n, user_data_t *ud) /* {{{ */ status = wrr_send(host, msg); if (status != 0) - ERROR("write_riemann plugin: riemann_client_send failed with status %i", - status); + c_complain (LOG_ERR, &host->init_complaint, + "write_riemann plugin: riemann_client_send failed with status %i", + status); + else + c_release (LOG_DEBUG, &host->init_complaint, + "write_riemann plugin: riemann_client_send succeeded"); riemann_message_free(msg); return (status); @@ -574,16 +597,13 @@ static int wrr_write(const data_set_t *ds, /* {{{ */ } if (host->client_type != RIEMANN_CLIENT_UDP && host->batch_mode) { - wrr_batch_add_value_list(host, ds, vl, statuses); + wrr_batch_add_value_list(host, ds, vl, statuses); } else { msg = wrr_value_list_to_message(host, ds, vl, statuses); if (msg == NULL) return (-1); status = wrr_send(host, msg); - if (status != 0) - ERROR("write_riemann plugin: riemann_client_send failed with status %i", - status); riemann_message_free(msg); } @@ -626,6 +646,7 @@ static int wrr_config_node(oconfig_item_t *ci) /* {{{ */ return ENOMEM; } pthread_mutex_init(&host->lock, NULL); + C_COMPLAIN_INIT (&host->init_complaint); host->reference_count = 1; host->node = NULL; host->port = 0; @@ -636,9 +657,12 @@ static int wrr_config_node(oconfig_item_t *ci) /* {{{ */ host->batch_mode = 1; host->batch_max = RIEMANN_BATCH_MAX; /* typical MSS */ host->batch_init = cdtime(); + host->batch_timeout = 0; host->ttl_factor = RIEMANN_TTL_FACTOR; host->client = NULL; host->client_type = RIEMANN_CLIENT_TCP; + host->timeout.tv_sec = 0; + host->timeout.tv_usec = 0; status = cf_util_get_string(ci, &host->name); if (status != 0) { @@ -679,6 +703,14 @@ static int wrr_config_node(oconfig_item_t *ci) /* {{{ */ status = cf_util_get_int(child, &host->batch_max); if (status != 0) break; + } else if (strcasecmp("BatchFlushTimeout", child->key) == 0) { + status = cf_util_get_int(child, &host->batch_timeout); + if (status != 0) + break; + } else if (strcasecmp("Timeout", child->key) == 0) { + status = cf_util_get_int(child, (int *)&host->timeout.tv_sec); + if (status != 0) + break; } else if (strcasecmp("Port", child->key) == 0) { host->port = cf_util_get_port_number(child); if (host->port == -1) {