write_http plugin: Implement the "BufferSize" option.

Fixes: #722

diff --git a/src/collectd.conf.in b/src/collectd.conf.in
index eed7bc818a99cb26fdf871892005f9e24ffda04d..279e80450135de5b2d744784e459325ab41236ee 100644 (file)
--- a/src/collectd.conf.in
+++ b/src/collectd.conf.in
@@ -1212,6 +1212,7 @@
 #              SSLVersion "TLSv1"
 #              Format "Command"
 #              StoreRates false
+#              BufferSize 4096
 #      </URL>
 #</Plugin>
 

diff --git a/src/collectd.conf.pod b/src/collectd.conf.pod
index 9293445ed39c1af9a039397ab0f14d23421d91c0..03483672f868073bb3ada20d4bc0af60665db590 100644 (file)
--- a/src/collectd.conf.pod
+++ b/src/collectd.conf.pod
@@ -6659,8 +6659,9 @@ want to use authentication all three fields must be set.
 
 =head2 Plugin C<write_http>
 
-This output plugin submits values to an http server by POST them using the
-PUTVAL plain-text protocol. Each destination you want to post data to needs to
+This output plugin submits values to an HTTP server using POST requests and
+encoding metircs with JSON or using the C<PUTVAL> command described in
+L<collectd-unixsock(5)>. Each destination you want to post data to needs to
 have one B<URL> block, within which the destination can be configured further,
 for example by specifying authentication data.
 
@@ -6670,6 +6671,7 @@ Synopsis:
    <URL "http://example.com/post-collectd">
      User "collectd"
      Password "weCh3ik0"
+     Format JSON
    </URL>
  </Plugin>
 
@@ -6746,6 +6748,15 @@ If set to B<true>, convert counter values to rates. If set to B<false> (the
 default) counter values are stored as is, i.E<nbsp>e. as an increasing integer
 number.
 
+=item B<BufferSize> I<Bytes>
+
+Sets the send buffer size to I<Bytes>. By increasing this buffer, less HTTP
+requests will be generated, but more metrics will be batched / metrics are
+cached for longer before being sent, introducing additional delay until they
+are available on the server side. I<Bytes> must be at least 1024 and cannot
+exceed the size of an C<int>, i.e. 2E<nbsp>GByte.
+Defaults to C<4096>.
+
 =back
 
 =head2 Plugin C<write_kafka>

diff --git a/src/write_http.c b/src/write_http.c
index eee5a1c66c9837bdf7165f61be09203f6f5b1c88..198fb48d32d5c242027df88a0ae1f40e0de589cb 100644 (file)
--- a/src/write_http.c
+++ b/src/write_http.c
@@ -2,7 +2,7 @@
  * collectd - src/write_http.c
  * Copyright (C) 2009       Paul Sadauskas
  * Copyright (C) 2009       Doug MacEachern
- * Copyright (C) 2007-2009  Florian octo Forster
+ * Copyright (C) 2007-2014  Florian octo Forster
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License as published by the
@@ -36,6 +36,10 @@
 
 #include <curl/curl.h>
 
+#ifndef WRITE_HTTP_DEFAULT_BUFFER_SIZE
+# define WRITE_HTTP_DEFAULT_BUFFER_SIZE 4096
+#endif
+
 /*
  * Private variables
  */
@@ -63,7 +67,8 @@ struct wh_callback_s
         CURL *curl;
         char curl_errbuf[CURL_ERROR_SIZE];
 
-        char   send_buffer[4096];
+        char  *send_buffer;
+        size_t send_buffer_size;
         size_t send_buffer_free;
         size_t send_buffer_fill;
         cdtime_t send_buffer_init_time;
@@ -74,8 +79,8 @@ typedef struct wh_callback_s wh_callback_t;
 
 static void wh_reset_buffer (wh_callback_t *cb)  /* {{{ */
 {
-        memset (cb->send_buffer, 0, sizeof (cb->send_buffer));
-        cb->send_buffer_free = sizeof (cb->send_buffer);
+        memset (cb->send_buffer, 0, cb->send_buffer_size);
+        cb->send_buffer_free = cb->send_buffer_size;
         cb->send_buffer_fill = 0;
         cb->send_buffer_init_time = cdtime ();
 
@@ -280,7 +285,11 @@ static void wh_callback_free (void *data) /* {{{ */
 
         wh_flush_nolock (/* timeout = */ 0, cb);
 
-        curl_easy_cleanup (cb->curl);
+        if (cb->curl != NULL)
+        {
+                curl_easy_cleanup (cb->curl);
+                cb->curl = NULL;
+        }
         sfree (cb->location);
         sfree (cb->user);
         sfree (cb->pass);
@@ -290,6 +299,7 @@ static void wh_callback_free (void *data) /* {{{ */
         sfree (cb->clientkey);
         sfree (cb->clientcert);
         sfree (cb->clientkeypass);
+        sfree (cb->send_buffer);
 
         sfree (cb);
 } /* }}} void wh_callback_free */
@@ -371,8 +381,8 @@ static int wh_write_command (const data_set_t *ds, const value_list_t *vl, /* {{
 
         DEBUG ("write_http plugin: <%s> buffer %zu/%zu (%g%%) \"%s\"",
                         cb->location,
-                        cb->send_buffer_fill, sizeof (cb->send_buffer),
-                        100.0 * ((double) cb->send_buffer_fill) / ((double) sizeof (cb->send_buffer)),
+                        cb->send_buffer_fill, cb->send_buffer_size,
+                        100.0 * ((double) cb->send_buffer_fill) / ((double) cb->send_buffer_size),
                         command);
 
         /* Check if we have enough space for this command. */
@@ -426,8 +436,8 @@ static int wh_write_json (const data_set_t *ds, const value_list_t *vl, /* {{{ *
 
         DEBUG ("write_http plugin: <%s> buffer %zu/%zu (%g%%)",
                         cb->location,
-                        cb->send_buffer_fill, sizeof (cb->send_buffer),
-                        100.0 * ((double) cb->send_buffer_fill) / ((double) sizeof (cb->send_buffer)));
+                        cb->send_buffer_fill, cb->send_buffer_size,
+                        100.0 * ((double) cb->send_buffer_fill) / ((double) cb->send_buffer_size));
 
         /* Check if we have enough space for this command. */
         pthread_mutex_unlock (&cb->send_lock);
@@ -485,6 +495,7 @@ static int config_set_format (wh_callback_t *cb, /* {{{ */
 static int wh_config_url (oconfig_item_t *ci) /* {{{ */
 {
         wh_callback_t *cb;
+        int buffer_size = 0;
         user_data_t user_data;
         int i;
 
@@ -560,6 +571,8 @@ static int wh_config_url (oconfig_item_t *ci) /* {{{ */
                         config_set_format (cb, child);
                 else if (strcasecmp ("StoreRates", child->key) == 0)
                         cf_util_get_boolean (child, &cb->store_rates);
+                else if (strcasecmp ("BufferSize", child->key) == 0)
+                        cf_util_get_int (child, &buffer_size);
                 else
                 {
                         ERROR ("write_http plugin: Invalid configuration "
@@ -567,6 +580,25 @@ static int wh_config_url (oconfig_item_t *ci) /* {{{ */
                 }
         }
 
+        /* Determine send_buffer_size. */
+        cb->send_buffer_size = WRITE_HTTP_DEFAULT_BUFFER_SIZE;
+        if (buffer_size >= 1024)
+                cb->send_buffer_size = (size_t) buffer_size;
+        else if (buffer_size != 0)
+                ERROR ("write_http plugin: Ignoring invalid BufferSize setting (%d).",
+                                buffer_size);
+
+        /* Allocate the buffer. */
+        cb->send_buffer = malloc (cb->send_buffer_size);
+        if (cb->send_buffer == 0)
+        {
+                ERROR ("write_http plugin: malloc(%zu) failed.", cb->send_buffer_size);
+                wh_callback_free (cb);
+                return (-1);
+        }
+        /* Nulls the buffer and sets ..._free and ..._fill. */
+        wh_reset_buffer (cb);
+
         DEBUG ("write_http: Registering write callback with URL %s",
                         cb->location);