Second pass at direct send support for graphite

gmetad/conf.c.in

@@ -261,6 +261,14 @@ static DOTCONF_CB(cb_carbon_port)
    return NULL;
 }
 
+static DOTCONF_CB(cb_carbon_timeout)
+{
+   gmetad_config_t *c = (gmetad_config_t*) cmd->option->info;
+   debug_msg("Setting carbon timeout to %d", cmd->data.value);
+   c->carbon_timeout = cmd->data.value;
+   return NULL;
+}
+
 static DOTCONF_CB(cb_graphite_prefix)
 {
    gmetad_config_t *c = (gmetad_config_t*) cmd->option->info;
@@ -295,6 +303,7 @@ static configoption_t gmetad_options[] =
       {"case_sensitive_hostnames", ARG_INT, cb_case_sensitive_hostnames, &gmetad_config, 0},
       {"carbon_server", ARG_STR, cb_carbon_server, &gmetad_config, 0},
       {"carbon_port", ARG_INT, cb_carbon_port, &gmetad_config, 0},
+      {"carbon_timeout", ARG_INT, cb_carbon_timeout, &gmetad_config, 0},
       {"graphite_prefix", ARG_STR, cb_graphite_prefix, &gmetad_config, 0},
       LAST_OPTION
    };

gmetad/conf.h

@@ -19,6 +19,7 @@ typedef struct
       char *rrd_rootdir;
       char *carbon_server;
       int carbon_port;
+      int carbon_timeout;
       char *graphite_prefix;
       int scalable_mode;
       int all_trusted;

gmetad/process_xml.c

@@ -1,4 +1,4 @@
-/* $Id$ */
+/* $Id: process_xml.c 2626 2011-07-07 15:44:35Z rufustfirefly $ */
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -643,17 +643,14 @@ startElement_METRIC(void *data, const char *el, const char **attr)
                   xmldata->rval = write_data_to_rrd(xmldata->sourcename,
                         xmldata->hostname, name, metricval, NULL,
                         xmldata->ds->step, xmldata->source.localtime, slope);
-
-                  /* If the user has specified a carbon server, send the metric
-                   * to carbon as well.
-                   */
-                  if (gmetad_config.carbon_server)
-                        carbon_ret = write_data_to_carbon(xmldata->sourcename, xmldata->hostname, name, metricval, xmldata->source.localtime);
+						if (gmetad_config.carbon_server) // if the user has specified a carbon server, send the metric to carbon as well
+                  carbon_ret=write_data_to_carbon(xmldata->sourcename, xmldata->hostname, name, metricval,xmldata->source.localtime);
             }
          metric->id = METRIC_NODE;
          metric->report_start = metric_report_start;
          metric->report_end = metric_report_end;
 
+
          edge = metric->stringslen;
          metric->name = addstring(metric->strings, &edge, name);
          metric->stringslen = edge;

gmetad/rrd_helpers.c

@@ -1,11 +1,12 @@
-/* $Id$ */
+/* $Id: rrd_helpers.c 2200 2010-01-08 17:17:00Z d_pocock $ */
 #include <ctype.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <unistd.h>
+#include <fcntl.h>
 #include <rrd.h>
 #include <gmetad.h>
 #include <errno.h>
@@ -14,6 +15,7 @@
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <netdb.h>
+#include <sys/poll.h>
 
 #include "rrd_helpers.h"
 
@@ -236,44 +238,70 @@ static int
 push_data_to_carbon( char *graphite_msg)
 {
   int port;
-  int sock;
+  int carbon_socket;
   struct sockaddr_in server;
+  int carbon_timeout = 500;
   int nbytes;
+  struct pollfd carbon_struct_poll;
+  int poll_rval;
+  int fl;
+
+  if (gmetad_config.carbon_port)
+     port=gmetad_config.carbon_port;
+  else
+     port=2003;
 
-  port = gmetad_config.carbon_port ? gmetad_config.carbon_port : 2003;
 
   debug_msg("Carbon Proxy:: sending \'%s\' to %s", graphite_msg, gmetad_config.carbon_server);
 
-  /* Create a socket. */
-  sock = socket (PF_INET, SOCK_STREAM, 0);
-  if (sock < 0)
+      /* Create a socket. */
+  carbon_socket = socket (PF_INET, SOCK_STREAM, 0);
+  if (carbon_socket < 0)
     {
       perror ("socket (client)");
-      close (sock);
+      close (carbon_socket);
       return EXIT_FAILURE;
     }
 
+  /* Set the socket to not block */
+  fl = fcntl(carbon_socket,F_GETFL,0);
+  fcntl(carbon_socket,F_SETFL,fl | O_NONBLOCK);
+
   /* Connect to the server. */
   init_sockaddr (&server, gmetad_config.carbon_server, port);
-  if (0 > connect (sock,
-                   (struct sockaddr *) &server,
-                   sizeof (server)))
-    {
-      perror ("connect (client)");
-      close (sock);
-      return EXIT_FAILURE;
-    }
-
-  /* Send data to the server. */
-  nbytes = write (sock, graphite_msg, strlen(graphite_msg) + 1);
-  if (nbytes < 0)
-    {
-      perror ("write");
-      close (sock);
+  connect (carbon_socket, (struct sockaddr *) &server, sizeof (server));
+
+  /* Start Poll */
+   carbon_struct_poll.fd=carbon_socket;
+   carbon_struct_poll.events = POLLOUT;
+   poll_rval = poll( &carbon_struct_poll, 1, carbon_timeout ); // default timeout .5s
+
+  /* Send data to the server when the socket becomes ready */
+  if( poll_rval < 0 ) {
+    debug_msg("carbon proxy:: poll() error");
+  } else if ( poll_rval == 0 ) {
+    debug_msg("carbon proxy:: Timeout connecting to %s",gmetad_config.carbon_server);
+  } else {
+    if( carbon_struct_poll.revents & POLLOUT ) {
+      /* Ready to send data to the server. */
+      debug_msg("carbon proxy:: %s is ready to receive",gmetad_config.carbon_server);
+      nbytes = write (carbon_socket, graphite_msg, strlen(graphite_msg) + 1);
+      if (nbytes < 0) {
+        perror ("write");
+        close(carbon_socket);
+        return EXIT_FAILURE;
+      }
+    } else if ( carbon_struct_poll.revents & POLLHUP ) {
+      debug_msg("carbon proxy:: Recvd an RST from %s during transmission",gmetad_config.carbon_server);
+     close(carbon_socket);
+     return EXIT_FAILURE;
+    } else if ( carbon_struct_poll.revents & POLLERR ) {
+      debug_msg("carbon proxy:: Recvd an POLLERR from %s during transmission",gmetad_config.carbon_server);
+      close(carbon_socket);
       return EXIT_FAILURE;
     }
-
-  close (sock);
+  }
+  close (carbon_socket);
   return EXIT_SUCCESS;
 }
 
@@ -282,15 +310,16 @@ write_data_to_carbon ( const char *source, const char *host, const char *metric,
                     const char *sum, unsigned int process_time )
 {
 
-   char s_process_time[15];
+	char s_process_time[15];
    char graphite_msg[ PATHSIZE + 1 ];
    int i;
 
    /*  if process_time is undefined, we set it to the current time */
    if (!process_time)
       process_time = time(0);
 
-   sprintf(s_process_time, "%u", process_time);
+
+	sprintf(s_process_time, "%u", process_time);
 
    /* Build the path */
    strncpy(graphite_msg, gmetad_config.graphite_prefix, PATHSIZE);
@@ -302,14 +331,18 @@ write_data_to_carbon ( const char *source, const char *host, const char *metric,
 
 
    if (host) {
-      int hostlen=strlen(host);
-      char hostcp[hostlen+1]; 
-
-      /* find and replace . for _ in the hostname*/
-      for (i=0; i<=hostlen; i++) {
-         hostcp[i] = ( host[i] == '.') ? '_' : host[i];
+		int hostlen=strlen(host);
+		char hostcp[hostlen+1]; 
+
+		/* find and replace . for _ in the hostname*/
+		for(i=0; i<=hostlen; i++){
+         if ( host[i] == '.') {
+           hostcp[i]='_';
+         }else{
+           hostcp[i]=host[i];
+         }
       }
-      hostcp[i+1]=0;
+		hostcp[i+1]=0;
 
       strncat(graphite_msg, ".", PATHSIZE-strlen(graphite_msg));
 
@@ -321,6 +354,7 @@ write_data_to_carbon ( const char *source, const char *host, const char *metric,
          for( ; graphite_msg[i] != 0; i++)
             graphite_msg[i] = tolower(graphite_msg[i]);
       }
+
    }
 
    strncat(graphite_msg, ".", PATHSIZE-strlen(graphite_msg));
@@ -331,8 +365,7 @@ write_data_to_carbon ( const char *source, const char *host, const char *metric,
    strncat(graphite_msg, s_process_time, PATHSIZE-strlen(graphite_msg));
    strncat(graphite_msg, "\n", PATHSIZE-strlen(graphite_msg));
 
-   graphite_msg[strlen(graphite_msg)+1] = 0;
+	graphite_msg[strlen(graphite_msg)+1] = 0;
 
    return push_data_to_carbon( graphite_msg );
 }
-