added new ForceReconnectTimeout to write_graphite plugin

diff --git a/src/collectd.conf.pod b/src/collectd.conf.pod
index d7d1501..6844e10 100644 (file)
--- a/src/collectd.conf.pod
+++ b/src/collectd.conf.pod
@@ -7271,6 +7271,10 @@ Service name or port number to connect to. Defaults to C<2003>.
 
 Protocol to use when connecting to I<Graphite>. Defaults to C<tcp>.
 
+=item B<ForceReconnectTimeout> I<Timeout Seconds>
+
+This parameter enables a forced close and reopen established connection with graphite backend after the specified Timeout seconds. This behaviour is desired in environments where the connection to the backend of graphite is done through load balancers. Default to 0 seconds ( 0 seconds = disabled ).
+
 =item B<LogSendErrors> B<false>|B<true>
 
 If set to B<true> (the default), logs errors when sending data to I<Graphite>.

diff --git a/src/write_graphite.c b/src/write_graphite.c
index 7fc9eb9..836f619 100644 (file)
--- a/src/write_graphite.c
+++ b/src/write_graphite.c
@@ -94,8 +94,34 @@ struct wg_callback
     pthread_mutex_t send_lock;
     c_complain_t init_complaint;
     cdtime_t last_connect_time;
+
+    /*Force reconnect useful for load balanced environments*/
+    cdtime_t last_force_reconnect_time;
+    int force_reconnect_timeout;
 };
 
+/*
+* Force Reconnect functions
+*/
+
+static void wg_force_reconnect_check(struct wg_callback *cb)
+{
+    cdtime_t now;
+    if(!cb->force_reconnect_timeout) return;
+    //check if address changes if addr_timeout
+    now = cdtime ();
+    DEBUG("wg_force_reconnect_check: now %ld last: %ld ",CDTIME_T_TO_TIME_T(now),CDTIME_T_TO_TIME_T(cb->last_force_reconnect_time));
+    if ((now - cb->last_force_reconnect_time) < TIME_T_TO_CDTIME_T(cb->force_reconnect_timeout)){
+       return;
+    }
+    //here we should close connection on next
+    close (cb->sock_fd);
+    cb->sock_fd = -1;
+    INFO("Connection Forced closed after %ld seconds ",CDTIME_T_TO_TIME_T(now - cb->last_force_reconnect_time));
+    cb->last_force_reconnect_time = now;
+}
+
+
 
 /*
  * Functions
@@ -326,6 +352,8 @@ static int wg_send_message (char const *message, struct wg_callback *cb)
 
     pthread_mutex_lock (&cb->send_lock);
 
+    wg_force_reconnect_check(cb);
+
     if (cb->sock_fd < 0)
     {
         status = wg_callback_init (cb);
@@ -462,6 +490,8 @@ static int wg_config_node (oconfig_item_t *ci)
     cb->node = strdup (WG_DEFAULT_NODE);
     cb->service = strdup (WG_DEFAULT_SERVICE);
     cb->protocol = strdup (WG_DEFAULT_PROTOCOL);
+    cb->last_force_reconnect_time=cdtime();
+    cb->force_reconnect_timeout=0;
     cb->log_send_errors = WG_DEFAULT_LOG_SEND_ERRORS;
     cb->prefix = NULL;
     cb->postfix = NULL;
@@ -502,6 +532,8 @@ static int wg_config_node (oconfig_item_t *ci)
                 status = -1;
             }
         }
+        else if (strcasecmp ("ForceReconnectTimeout", child->key) == 0)
+            cf_util_get_int (child,&cb->force_reconnect_timeout);
         else if (strcasecmp ("LogSendErrors", child->key) == 0)
             cf_util_get_boolean (child, &cb->log_send_errors);
         else if (strcasecmp ("Prefix", child->key) == 0)