network plugin: Improve client connecting behavior.
authorFlorian Forster <octo@collectd.org>
Sat, 6 Sep 2014 08:58:45 +0000 (10:58 +0200)
committerFlorian Forster <octo@collectd.org>
Sat, 6 Sep 2014 09:21:53 +0000 (11:21 +0200)
This moves the socket creation logic so it's called from
networt_send_buffer_plain(). This allows us to recover after network
failures or when collectd was started before the network was available.

Fixes: #627

src/network.c

index 1b6cf1e..ce9b0cc 100644 (file)
@@ -395,7 +395,7 @@ static _Bool check_send_notify_okay (const notification_t *n) /* {{{ */
   {
     c_complain_once (LOG_ERR, &complain_forwarding,
         "network plugin: A notification has been received via the network "
-        "forwarding if enabled. Forwarding of notifications is currently "
+        "and forwarding is enabled. Forwarding of notifications is currently "
         "not supported, because there is not loop-deteciton available. "
         "Please contact the collectd mailing list if you need this "
         "feature.");
@@ -1992,14 +1992,19 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const int int
 
 /* Initialize a sockent structure. `type' must be either `SOCKENT_TYPE_CLIENT'
  * or `SOCKENT_TYPE_SERVER' */
-static int sockent_init (sockent_t *se, int type) /* {{{ */
+static sockent_t *sockent_create (int type) /* {{{ */
 {
-       if (se == NULL)
-               return (-1);
+       sockent_t *se;
+
+       if ((type != SOCKENT_TYPE_CLIENT) || (type != SOCKENT_TYPE_SERVER))
+               return (NULL);
 
+       se = malloc (sizeof (*se));
+       if (se == NULL)
+               return (NULL);
        memset (se, 0, sizeof (*se));
 
-       se->type = SOCKENT_TYPE_CLIENT;
+       se->type = type;
        se->node = NULL;
        se->service = NULL;
        se->interface = 0;
@@ -2007,7 +2012,6 @@ static int sockent_init (sockent_t *se, int type) /* {{{ */
 
        if (type == SOCKENT_TYPE_SERVER)
        {
-               se->type = SOCKENT_TYPE_SERVER;
                se->data.server.fd = NULL;
 #if HAVE_LIBGCRYPT
                se->data.server.security_level = SECURITY_LEVEL_NONE;
@@ -2028,23 +2032,11 @@ static int sockent_init (sockent_t *se, int type) /* {{{ */
 #endif
        }
 
-       return (0);
-} /* }}} int sockent_init */
+       return (se);
+} /* }}} sockent_t *sockent_create */
 
-/* Open the file descriptors for a initialized sockent structure. */
-static int sockent_open (sockent_t *se) /* {{{ */
+static int sockent_init_crypto (sockent_t *se) /* {{{ */
 {
-       struct addrinfo  ai_hints;
-       struct addrinfo *ai_list, *ai_ptr;
-       int              ai_return;
-
-        const char *node;
-        const char *service;
-
-       if (se == NULL)
-               return (-1);
-
-       /* Set up the security structures. */
 #if HAVE_LIBGCRYPT /* {{{ */
        if (se->type == SOCKENT_TYPE_CLIENT)
        {
@@ -2095,13 +2087,134 @@ static int sockent_open (sockent_t *se) /* {{{ */
        }
 #endif /* }}} HAVE_LIBGCRYPT */
 
+       return (0);
+} /* }}} int sockent_init_crypto */
+
+static int sockent_client_connect (sockent_t *se) /* {{{ */
+{
+       static c_complain_t complaint = C_COMPLAIN_INIT_STATIC;
+
+       struct sockent_client *client;
+       struct addrinfo  ai_hints;
+       struct addrinfo *ai_list = NULL, *ai_ptr;
+       int status;
+
+       if ((se == NULL) || (se->type != SOCKENT_TYPE_CLIENT))
+               return (EINVAL);
+
+       client = &se->data.client;
+       if (client->fd >= 0) /* already connected */
+               return (0);
+
+       memset (&ai_hints, 0, sizeof (ai_hints));
+#ifdef AI_ADDRCONFIG
+       ai_hints.ai_flags |= AI_ADDRCONFIG;
+#endif
+       ai_hints.ai_family   = AF_UNSPEC;
+       ai_hints.ai_socktype = SOCK_DGRAM;
+       ai_hints.ai_protocol = IPPROTO_UDP;
+
+       status = getaddrinfo (se->node,
+                       (se->service != NULL) ? se->service : NET_DEFAULT_PORT,
+                       &ai_hints, &ai_list);
+       if (status != 0)
+       {
+               c_complain (LOG_ERR, &complaint,
+                               "network plugin: getaddrinfo (%s, %s) failed: %s",
+                               (se->node == NULL) ? "(null)" : se->node,
+                               (se->service == NULL) ? "(null)" : se->service,
+                               gai_strerror (status));
+               return (-1);
+       }
+       else
+       {
+               c_release (LOG_NOTICE, &complaint,
+                               "network plugin: Successfully resolved \"%s\".",
+                               se->node);
+       }
+
+       for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
+       {
+               client->fd = socket (ai_ptr->ai_family,
+                               ai_ptr->ai_socktype,
+                               ai_ptr->ai_protocol);
+               if (client->fd < 0)
+               {
+                       char errbuf[1024];
+                       ERROR ("network plugin: socket(2) failed: %s",
+                                       sstrerror (errno, errbuf,
+                                               sizeof (errbuf)));
+                       continue;
+               }
+
+               client->addr = malloc (sizeof (*client->addr));
+               if (client->addr == NULL)
+               {
+                       ERROR ("network plugin: malloc failed.");
+                       close (client->fd);
+                       client->fd = -1;
+                       continue;
+               }
+
+               memset (client->addr, 0, sizeof (*client->addr));
+               assert (sizeof (*client->addr) >= ai_ptr->ai_addrlen);
+               memcpy (client->addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
+               client->addrlen = ai_ptr->ai_addrlen;
+
+               network_set_ttl (se, ai_ptr);
+               network_set_interface (se, ai_ptr);
+
+               /* We don't open more than one write-socket per
+                * node/service pair.. */
+               break;
+       }
+
+       freeaddrinfo (ai_list);
+       if (client->fd < 0)
+               return (-1);
+       return (0);
+} /* }}} int sockent_client_connect */
+
+static int sockent_client_disconnect (sockent_t *se) /* {{{ */
+{
+       struct sockent_client *client;
+
+       if ((se == NULL) || (se->type != SOCKENT_TYPE_CLIENT))
+               return (EINVAL);
+
+       client = &se->data.client;
+       if (client->fd >= 0) /* connected */
+       {
+               close (client->fd);
+               client->fd = -1;
+       }
+
+       sfree (client->addr);
+       client->addrlen = 0;
+
+       return (0);
+} /* }}} int sockent_client_disconnect */
+
+/* Open the file descriptors for a initialized sockent structure. */
+static int sockent_server_listen (sockent_t *se) /* {{{ */
+{
+       struct addrinfo  ai_hints;
+       struct addrinfo *ai_list, *ai_ptr;
+       int              status;
+
+        const char *node;
+        const char *service;
+
+       if (se == NULL)
+               return (-1);
+
         node = se->node;
         service = se->service;
 
         if (service == NULL)
           service = NET_DEFAULT_PORT;
 
-        DEBUG ("network plugin: sockent_open: node = %s; service = %s;",
+        DEBUG ("network plugin: sockent_server_listen: node = %s; service = %s;",
             node, service);
 
        memset (&ai_hints, 0, sizeof (ai_hints));
@@ -2116,109 +2229,59 @@ static int sockent_open (sockent_t *se) /* {{{ */
        ai_hints.ai_socktype = SOCK_DGRAM;
        ai_hints.ai_protocol = IPPROTO_UDP;
 
-       ai_return = getaddrinfo (node, service, &ai_hints, &ai_list);
-       if (ai_return != 0)
+       status = getaddrinfo (node, service, &ai_hints, &ai_list);
+       if (status != 0)
        {
                ERROR ("network plugin: getaddrinfo (%s, %s) failed: %s",
                                (se->node == NULL) ? "(null)" : se->node,
                                (se->service == NULL) ? "(null)" : se->service,
-                               gai_strerror (ai_return));
+                               gai_strerror (status));
                return (-1);
        }
 
        for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
        {
-               int status;
+               int *tmp;
 
-               if (se->type == SOCKENT_TYPE_SERVER) /* {{{ */
+               tmp = realloc (se->data.server.fd,
+                               sizeof (*tmp) * (se->data.server.fd_num + 1));
+               if (tmp == NULL)
                {
-                       int *tmp;
-
-                       tmp = realloc (se->data.server.fd,
-                                       sizeof (*tmp) * (se->data.server.fd_num + 1));
-                       if (tmp == NULL)
-                       {
-                               ERROR ("network plugin: realloc failed.");
-                               continue;
-                       }
-                       se->data.server.fd = tmp;
-                       tmp = se->data.server.fd + se->data.server.fd_num;
-
-                       *tmp = socket (ai_ptr->ai_family, ai_ptr->ai_socktype,
-                                       ai_ptr->ai_protocol);
-                       if (*tmp < 0)
-                       {
-                               char errbuf[1024];
-                               ERROR ("network plugin: socket(2) failed: %s",
-                                               sstrerror (errno, errbuf,
-                                                       sizeof (errbuf)));
-                               continue;
-                       }
-
-                       status = network_bind_socket (*tmp, ai_ptr, se->interface);
-                       if (status != 0)
-                       {
-                               close (*tmp);
-                               *tmp = -1;
-                               continue;
-                       }
-
-                       se->data.server.fd_num++;
+                       ERROR ("network plugin: realloc failed.");
                        continue;
-               } /* }}} if (se->type == SOCKENT_TYPE_SERVER) */
-               else /* if (se->type == SOCKENT_TYPE_CLIENT) {{{ */
-               {
-                       se->data.client.fd = socket (ai_ptr->ai_family,
-                                       ai_ptr->ai_socktype,
-                                       ai_ptr->ai_protocol);
-                       if (se->data.client.fd < 0)
-                       {
-                               char errbuf[1024];
-                               ERROR ("network plugin: socket(2) failed: %s",
-                                               sstrerror (errno, errbuf,
-                                                       sizeof (errbuf)));
-                               continue;
-                       }
-
-                       se->data.client.addr = malloc (sizeof (*se->data.client.addr));
-                       if (se->data.client.addr == NULL)
-                       {
-                               ERROR ("network plugin: malloc failed.");
-                               close (se->data.client.fd);
-                               se->data.client.fd = -1;
-                               continue;
-                       }
+               }
+               se->data.server.fd = tmp;
+               tmp = se->data.server.fd + se->data.server.fd_num;
 
-                       memset (se->data.client.addr, 0, sizeof (*se->data.client.addr));
-                       assert (sizeof (*se->data.client.addr) >= ai_ptr->ai_addrlen);
-                       memcpy (se->data.client.addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
-                       se->data.client.addrlen = ai_ptr->ai_addrlen;
+               *tmp = socket (ai_ptr->ai_family, ai_ptr->ai_socktype,
+                               ai_ptr->ai_protocol);
+               if (*tmp < 0)
+               {
+                       char errbuf[1024];
+                       ERROR ("network plugin: socket(2) failed: %s",
+                                       sstrerror (errno, errbuf,
+                                               sizeof (errbuf)));
+                       continue;
+               }
 
-                       network_set_ttl (se, ai_ptr);
-                       network_set_interface (se, ai_ptr);
+               status = network_bind_socket (*tmp, ai_ptr, se->interface);
+               if (status != 0)
+               {
+                       close (*tmp);
+                       *tmp = -1;
+                       continue;
+               }
 
-                       /* We don't open more than one write-socket per
-                        * node/service pair.. */
-                       break;
-               } /* }}} if (se->type == SOCKENT_TYPE_CLIENT) */
+               se->data.server.fd_num++;
+               continue;
        } /* for (ai_list) */
 
        freeaddrinfo (ai_list);
 
-       /* Check if all went well. */
-       if (se->type == SOCKENT_TYPE_SERVER)
-       {
-               if (se->data.server.fd_num <= 0)
-                       return (-1);
-       }
-       else /* if (se->type == SOCKENT_TYPE_CLIENT) */
-       {
-               if (se->data.client.fd < 0)
-                       return (-1);
-       }
-
+       if (se->data.server.fd_num <= 0)
+               return (-1);
        return (0);
-} /* }}} int sockent_open */
+} /* }}} int sockent_server_listen */
 
 /* Add a sockent to the global list of sockets */
 static int sockent_add (sockent_t *se) /* {{{ */
@@ -2486,26 +2549,32 @@ static void network_init_buffer (void)
        memset (&send_buffer_vl, 0, sizeof (send_buffer_vl));
 } /* int network_init_buffer */
 
-static void networt_send_buffer_plain (const sockent_t *se, /* {{{ */
+static void networt_send_buffer_plain (sockent_t *se, /* {{{ */
                const char *buffer, size_t buffer_size)
 {
        int status;
 
        while (42)
        {
+               status = sockent_client_connect (se);
+               if (status != 0)
+                       return;
+
                status = sendto (se->data.client.fd, buffer, buffer_size,
-                    /* flags = */ 0,
-                    (struct sockaddr *) se->data.client.addr,
-                    se->data.client.addrlen);
-                if (status < 0)
+                               /* flags = */ 0,
+                               (struct sockaddr *) se->data.client.addr,
+                               se->data.client.addrlen);
+               if (status < 0)
                {
                        char errbuf[1024];
-                       if (errno == EINTR)
+
+                       if ((errno == EINTR) || (errno == EAGAIN))
                                continue;
-                       ERROR ("network plugin: sendto failed: %s",
-                                       sstrerror (errno, errbuf,
-                                               sizeof (errbuf)));
-                       break;
+
+                       ERROR ("network plugin: sendto failed: %s. Closing sending socket.",
+                                       sstrerror (errno, errbuf, sizeof (errbuf)));
+                       sockent_client_disconnect (se);
+                       return;
                }
 
                break;
@@ -2518,7 +2587,7 @@ static void networt_send_buffer_plain (const sockent_t *se, /* {{{ */
   buffer_offset += (s); \
 } while (0)
 
-static void networt_send_buffer_signed (const sockent_t *se, /* {{{ */
+static void networt_send_buffer_signed (sockent_t *se, /* {{{ */
                const char *in_buffer, size_t in_buffer_size)
 {
   part_signature_sha256_t ps;
@@ -3014,13 +3083,12 @@ static int network_config_add_listen (const oconfig_item_t *ci) /* {{{ */
     return (-1);
   }
 
-  se = malloc (sizeof (*se));
+  se = sockent_create (SOCKENT_TYPE_SERVER);
   if (se == NULL)
   {
-    ERROR ("network plugin: malloc failed.");
+    ERROR ("network plugin: sockent_create failed.");
     return (-1);
   }
-  sockent_init (se, SOCKENT_TYPE_SERVER);
 
   se->node = strdup (ci->values[0].value.string);
   if (ci->values_num >= 2)
@@ -3060,10 +3128,18 @@ static int network_config_add_listen (const oconfig_item_t *ci) /* {{{ */
   }
 #endif /* HAVE_LIBGCRYPT */
 
-  status = sockent_open (se);
+  status = sockent_init_crypto (se);
   if (status != 0)
   {
-    ERROR ("network plugin: network_config_add_listen: sockent_open failed.");
+    ERROR ("network plugin: network_config_add_listen: sockent_init_crypto() failed.");
+    sockent_destroy (se);
+    return (-1);
+  }
+
+  status = sockent_server_listen (se);
+  if (status != 0)
+  {
+    ERROR ("network plugin: network_config_add_server: sockent_server_listen failed.");
     sockent_destroy (se);
     return (-1);
   }
@@ -3094,13 +3170,12 @@ static int network_config_add_server (const oconfig_item_t *ci) /* {{{ */
     return (-1);
   }
 
-  se = malloc (sizeof (*se));
+  se = sockent_create (SOCKENT_TYPE_CLIENT);
   if (se == NULL)
   {
-    ERROR ("network plugin: malloc failed.");
+    ERROR ("network plugin: sockent_create failed.");
     return (-1);
   }
-  sockent_init (se, SOCKENT_TYPE_CLIENT);
 
   se->node = strdup (ci->values[0].value.string);
   if (ci->values_num >= 2)
@@ -3143,14 +3218,17 @@ static int network_config_add_server (const oconfig_item_t *ci) /* {{{ */
   }
 #endif /* HAVE_LIBGCRYPT */
 
-  status = sockent_open (se);
+  status = sockent_init_crypto (se);
   if (status != 0)
   {
-    ERROR ("network plugin: network_config_add_server: sockent_open failed.");
+    ERROR ("network plugin: network_config_add_server: sockent_init_crypto() failed.");
     sockent_destroy (se);
     return (-1);
   }
 
+  /* No call to sockent_client_connect() here -- it is called from
+   * networt_send_buffer_plain(). */
+
   status = sockent_add (se);
   if (status != 0)
   {
@@ -3268,6 +3346,8 @@ static int network_notification (const notification_t *n,
 
 static int network_shutdown (void)
 {
+       sockent_t *se;
+
        listen_loop++;
 
        /* Kill the listening thread */
@@ -3298,7 +3378,9 @@ static int network_shutdown (void)
 
        sfree (send_buffer);
 
-       /* TODO: Close `sending_sockets' */
+       for (se = sending_sockets; se != NULL; se = se->next)
+               sockent_client_disconnect (se);
+       sockent_destroy (sending_sockets);
 
        plugin_unregister_config ("network");
        plugin_unregister_init ("network");