Merge pull request #417 from udoprog/udoprog/riemann

write_riemann: Add extra meta strings as attributes in notifications

diff --git a/configure.in b/configure.in
index c23df17..153bac4 100644 (file)
--- a/configure.in
+++ b/configure.in
@@ -1121,6 +1121,17 @@ else
   AC_MSG_ERROR([Didn't find out how doubles are stored in memory. Sorry.])
 fi; fi; fi
 
+# --with-useragent {{{
+AC_ARG_WITH(useragent, [AS_HELP_STRING([--with-useragent@<:@=AGENT@:>@], [User agent to use on http requests])],
+[
+    if test "x$withval" != "xno" && test "x$withval" != "xyes"
+    then
+        AC_DEFINE_UNQUOTED(COLLECTD_USERAGENT, ["$withval"], [User agent for http requests])
+    fi
+])
+
+# }}}
+
 have_getfsstat="no"
 AC_CHECK_FUNCS(getfsstat, [have_getfsstat="yes"])
 have_getvfsstat="no"

diff --git a/src/apache.c b/src/apache.c
index 8458ce1..c5f099f 100644 (file)
--- a/src/apache.c
+++ b/src/apache.c
@@ -402,7 +402,7 @@ static int init_host (apache_t *st) /* {{{ */
                curl_easy_setopt (st->curl, CURLOPT_WRITEHEADER, st);
        }
 
-       curl_easy_setopt (st->curl, CURLOPT_USERAGENT, PACKAGE_NAME"/"PACKAGE_VERSION);
+       curl_easy_setopt (st->curl, CURLOPT_USERAGENT, COLLECTD_USERAGENT);
        curl_easy_setopt (st->curl, CURLOPT_ERRORBUFFER, st->apache_curl_error);
 
        if (st->user != NULL)

diff --git a/src/ascent.c b/src/ascent.c
index 94a3938..6809bac 100644 (file)
--- a/src/ascent.c
+++ b/src/ascent.c
@@ -541,7 +541,7 @@ static int ascent_init (void) /* {{{ */
 
   curl_easy_setopt (curl, CURLOPT_NOSIGNAL, 1L);
   curl_easy_setopt (curl, CURLOPT_WRITEFUNCTION, ascent_curl_callback);
-  curl_easy_setopt (curl, CURLOPT_USERAGENT, PACKAGE_NAME"/"PACKAGE_VERSION);
+  curl_easy_setopt (curl, CURLOPT_USERAGENT, COLLECTD_USERAGENT);
   curl_easy_setopt (curl, CURLOPT_ERRORBUFFER, ascent_curl_error);
 
   if (user != NULL)

diff --git a/src/bind.c b/src/bind.c
index ddde840..2f990a2 100644 (file)
--- a/src/bind.c
+++ b/src/bind.c
@@ -1395,7 +1395,7 @@ static int bind_init (void) /* {{{ */
 
   curl_easy_setopt (curl, CURLOPT_NOSIGNAL, 1L);
   curl_easy_setopt (curl, CURLOPT_WRITEFUNCTION, bind_curl_callback);
-  curl_easy_setopt (curl, CURLOPT_USERAGENT, PACKAGE_NAME"/"PACKAGE_VERSION);
+  curl_easy_setopt (curl, CURLOPT_USERAGENT, COLLECTD_USERAGENT);
   curl_easy_setopt (curl, CURLOPT_ERRORBUFFER, bind_curl_error);
   curl_easy_setopt (curl, CURLOPT_URL, (url != NULL) ? url : BIND_DEFAULT_URL);
   curl_easy_setopt (curl, CURLOPT_FOLLOWLOCATION, 1L);

diff --git a/src/collectd.h b/src/collectd.h
index 7cb405c..969aeda 100644 (file)
--- a/src/collectd.h
+++ b/src/collectd.h
@@ -262,6 +262,10 @@ typedef int _Bool;
 # define COLLECTD_DEFAULT_INTERVAL 10.0
 #endif
 
+ #ifndef COLLECTD_USERAGENT
+ # define COLLECTD_USERAGENT PACKAGE_NAME"/"PACKAGE_VERSION
+ #endif
+
 /* Remove GNU specific __attribute__ settings when using another compiler */
 #if !__GNUC__
 # define __attribute__(x) /**/

diff --git a/src/curl.c b/src/curl.c
index 280e61c..5990368 100644 (file)
--- a/src/curl.c
+++ b/src/curl.c
@@ -364,8 +364,7 @@ static int cc_page_init_curl (web_page_t *wp) /* {{{ */
   curl_easy_setopt (wp->curl, CURLOPT_NOSIGNAL, 1L);
   curl_easy_setopt (wp->curl, CURLOPT_WRITEFUNCTION, cc_curl_callback);
   curl_easy_setopt (wp->curl, CURLOPT_WRITEDATA, wp);
-  curl_easy_setopt (wp->curl, CURLOPT_USERAGENT,
-      PACKAGE_NAME"/"PACKAGE_VERSION);
+  curl_easy_setopt (wp->curl, CURLOPT_USERAGENT, COLLECTD_USERAGENT);
   curl_easy_setopt (wp->curl, CURLOPT_ERRORBUFFER, wp->curl_errbuf);
   curl_easy_setopt (wp->curl, CURLOPT_URL, wp->url);
   curl_easy_setopt (wp->curl, CURLOPT_FOLLOWLOCATION, 1L);

diff --git a/src/curl_json.c b/src/curl_json.c
index 511863e..251fee8 100644 (file)
--- a/src/curl_json.c
+++ b/src/curl_json.c
@@ -586,8 +586,7 @@ static int cj_init_curl (cj_t *db) /* {{{ */
   curl_easy_setopt (db->curl, CURLOPT_NOSIGNAL, 1L);
   curl_easy_setopt (db->curl, CURLOPT_WRITEFUNCTION, cj_curl_callback);
   curl_easy_setopt (db->curl, CURLOPT_WRITEDATA, db);
-  curl_easy_setopt (db->curl, CURLOPT_USERAGENT,
-                    PACKAGE_NAME"/"PACKAGE_VERSION);
+  curl_easy_setopt (db->curl, CURLOPT_USERAGENT, COLLECTD_USERAGENT);
   curl_easy_setopt (db->curl, CURLOPT_ERRORBUFFER, db->curl_errbuf);
   curl_easy_setopt (db->curl, CURLOPT_URL, db->url);
 

diff --git a/src/curl_xml.c b/src/curl_xml.c
index 5adaf06..3554d7e 100644 (file)
--- a/src/curl_xml.c
+++ b/src/curl_xml.c
@@ -837,8 +837,7 @@ static int cx_init_curl (cx_t *db) /* {{{ */
   curl_easy_setopt (db->curl, CURLOPT_NOSIGNAL, 1L);
   curl_easy_setopt (db->curl, CURLOPT_WRITEFUNCTION, cx_curl_callback);
   curl_easy_setopt (db->curl, CURLOPT_WRITEDATA, db);
-  curl_easy_setopt (db->curl, CURLOPT_USERAGENT,
-                    PACKAGE_NAME"/"PACKAGE_VERSION);
+  curl_easy_setopt (db->curl, CURLOPT_USERAGENT, COLLECTD_USERAGENT);
   curl_easy_setopt (db->curl, CURLOPT_ERRORBUFFER, db->curl_errbuf);
   curl_easy_setopt (db->curl, CURLOPT_URL, db->url);
 

diff --git a/src/nginx.c b/src/nginx.c
index de52262..c356353 100644 (file)
--- a/src/nginx.c
+++ b/src/nginx.c
@@ -122,7 +122,7 @@ static int init (void)
 
   curl_easy_setopt (curl, CURLOPT_NOSIGNAL, 1L);
   curl_easy_setopt (curl, CURLOPT_WRITEFUNCTION, nginx_curl_callback);
-  curl_easy_setopt (curl, CURLOPT_USERAGENT, PACKAGE_NAME"/"PACKAGE_VERSION);
+  curl_easy_setopt (curl, CURLOPT_USERAGENT, COLLECTD_USERAGENT);
   curl_easy_setopt (curl, CURLOPT_ERRORBUFFER, nginx_curl_error);
 
   if (user != NULL)

diff --git a/src/write_http.c b/src/write_http.c
index 7f5943a..25133c3 100644 (file)
--- a/src/write_http.c
+++ b/src/write_http.c
@@ -112,7 +112,7 @@ static int wh_callback_init (wh_callback_t *cb) /* {{{ */
         }
 
         curl_easy_setopt (cb->curl, CURLOPT_NOSIGNAL, 1L);
-        curl_easy_setopt (cb->curl, CURLOPT_USERAGENT, PACKAGE_NAME"/"PACKAGE_VERSION);
+        curl_easy_setopt (cb->curl, CURLOPT_USERAGENT, COLLECTD_USERAGENT);
 
         headers = NULL;
         headers = curl_slist_append (headers, "Accept:  */*");

diff --git a/src/write_riemann.c b/src/write_riemann.c
index 42e11f9..0bd8e3a 100644 (file)
--- a/src/write_riemann.c
+++ b/src/write_riemann.c
@@ -178,32 +178,30 @@ riemann_disconnect (struct riemann_host *host)
        return (0);
 }
 
-static int
-riemann_send(struct riemann_host *host, Msg const *msg)
+static inline int
+riemann_send_msg(struct riemann_host *host, const Msg *msg)
 {
-       u_char *buffer;
+       int status = 0;
+       u_char *buffer = NULL;
        size_t  buffer_len;
-       int status;
-
-       pthread_mutex_lock (&host->lock);
 
        status = riemann_connect (host);
+
        if (status != 0)
-       {
-               pthread_mutex_unlock (&host->lock);
                return status;
-       }
 
        buffer_len = msg__get_packed_size(msg);
+
        if (host->use_tcp)
                buffer_len += 4;
 
        buffer = malloc (buffer_len);
+
        if (buffer == NULL) {
-               pthread_mutex_unlock (&host->lock);
                ERROR ("write_riemann plugin: malloc failed.");
                return ENOMEM;
        }
+
        memset (buffer, 0, buffer_len);
 
        if (host->use_tcp)
@@ -218,26 +216,105 @@ riemann_send(struct riemann_host *host, Msg const *msg)
        }
 
        status = (int) swrite (host->s, buffer, buffer_len);
+
        if (status != 0)
        {
                char errbuf[1024];
 
-               riemann_disconnect (host);
-               pthread_mutex_unlock (&host->lock);
-
                ERROR ("write_riemann plugin: Sending to Riemann at %s:%s failed: %s",
                                (host->node != NULL) ? host->node : RIEMANN_HOST,
                                (host->service != NULL) ? host->service : RIEMANN_PORT,
                                sstrerror (errno, errbuf, sizeof (errbuf)));
+
                sfree (buffer);
                return -1;
        }
 
-       pthread_mutex_unlock (&host->lock);
        sfree (buffer);
        return 0;
 }
 
+static inline int
+riemann_recv_ack(struct riemann_host *host)
+{
+       int status = 0;
+       Msg *msg = NULL;
+       uint32_t header;
+
+       status = (int) sread (host->s, &header, 4);
+
+       if (status != 0)
+               return -1;
+
+       size_t size = ntohl(header);
+
+       // Buffer on the stack since acknowledges are typically small.
+       u_char buffer[size];
+       memset (buffer, 0, size);
+
+       status = (int) sread (host->s, buffer, size);
+
+       if (status != 0)
+               return status;
+
+       msg = msg__unpack (NULL, size, buffer);
+
+       if (msg == NULL)
+               return -1;
+
+       if (!msg->ok)
+       {
+               ERROR ("write_riemann plugin: Sending to Riemann at %s:%s acknowledgement message reported error: %s",
+                               (host->node != NULL) ? host->node : RIEMANN_HOST,
+                               (host->service != NULL) ? host->service : RIEMANN_PORT,
+                               msg->error);
+
+               msg__free_unpacked(msg, NULL);
+               return -1;
+       }
+
+       msg__free_unpacked (msg, NULL);
+       return 0;
+}
+
+/**
+ * Function to send messages (Msg) to riemann.
+ *
+ * Acquires the host lock, disconnects on errors.
+ */
+static int
+riemann_send(struct riemann_host *host, Msg const *msg)
+{
+       int status = 0;
+       pthread_mutex_lock (&host->lock);
+
+       status = riemann_send_msg(host, msg);
+
+       if (status != 0) {
+               riemann_disconnect (host);
+               pthread_mutex_unlock (&host->lock);
+               return status;
+       }
+
+       /*
+        * For TCP we need to receive message acknowledgemenent.
+        */
+       if (host->use_tcp)
+       {
+               status = riemann_recv_ack(host);
+
+               if (status != 0)
+               {
+                       riemann_disconnect (host);
+                       pthread_mutex_unlock (&host->lock);
+                       return status;
+               }
+       }
+
+       pthread_mutex_unlock (&host->lock);
+       return 0;
+}
+
 static int riemann_event_add_tag (Event *event, char const *tag) /* {{{ */
 {
        return (strarray_add (&event->tags, &event->n_tags, tag));