Merge branch 'collectd-5.4' into collectd-5.5
[collectd.git] / src / tcpconns.c
index 3c8fc72..b6a3fdd 100644 (file)
@@ -17,7 +17,7 @@
  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  *
  * Author:
- *   Florian octo Forster <octo at verplant.org>
+ *   Florian octo Forster <octo at collectd.org>
  *   Michael Stapelberg <michael+git at stapelberg.de>
  **/
 
 #endif
 
 #if KERNEL_LINUX
+# include <asm/types.h>
+/* sys/socket.h is necessary to compile when using netlink on older systems. */
+# include <sys/socket.h>
+# include <linux/netlink.h>
+#if HAVE_LINUX_INET_DIAG_H
+# include <linux/inet_diag.h>
+#endif
+# include <sys/socket.h>
+# include <arpa/inet.h>
 /* #endif KERNEL_LINUX */
 
 #elif HAVE_SYSCTLBYNAME
 #endif /* KERNEL_AIX */
 
 #if KERNEL_LINUX
+#if HAVE_STRUCT_LINUX_INET_DIAG_REQ
+struct nlreq {
+  struct nlmsghdr nlh;
+  struct inet_diag_req r;
+};
+#endif
+
 static const char *tcp_state[] =
 {
   "", /* 0 */
@@ -203,13 +219,13 @@ static const char *tcp_state[] =
   "CLOSED",
   "LISTEN",
   "SYN_SENT",
-  "SYN_RCVD",
+  "SYN_RECV",
   "ESTABLISHED",
   "CLOSE_WAIT",
-  "FIN_WAIT_1",
+  "FIN_WAIT1",
   "CLOSING",
   "LAST_ACK",
-  "FIN_WAIT_2",
+  "FIN_WAIT2",
   "TIME_WAIT"
 };
 
@@ -256,12 +272,40 @@ static const char *config_keys[] =
 {
   "ListeningPorts",
   "LocalPort",
-  "RemotePort"
+  "RemotePort",
+  "AllPortsSummary"
 };
 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
 
 static int port_collect_listening = 0;
+static int port_collect_total = 0;
 static port_entry_t *port_list_head = NULL;
+static uint32_t count_total[TCP_STATE_MAX + 1];
+
+#if KERNEL_LINUX
+#if HAVE_STRUCT_LINUX_INET_DIAG_REQ
+/* This depends on linux inet_diag_req because if this structure is missing,
+ * sequence_number is useless and we get a compilation warning.
+ */
+static uint32_t sequence_number = 0;
+#endif
+
+enum
+{
+  SRC_DUNNO,
+  SRC_NETLINK,
+  SRC_PROC
+} linux_source = SRC_DUNNO;
+#endif
+
+static void conn_prepare_vl (value_list_t *vl, value_t *values)
+{
+  vl->values = values;
+  vl->values_len = 1;
+  sstrncpy (vl->host, hostname_g, sizeof (vl->host));
+  sstrncpy (vl->plugin, "tcpconns", sizeof (vl->plugin));
+  sstrncpy (vl->type, "tcp_connections", sizeof (vl->type));
+}
 
 static void conn_submit_port_entry (port_entry_t *pe)
 {
@@ -269,11 +313,7 @@ static void conn_submit_port_entry (port_entry_t *pe)
   value_list_t vl = VALUE_LIST_INIT;
   int i;
 
-  vl.values = values;
-  vl.values_len = 1;
-  sstrncpy (vl.host, hostname_g, sizeof (vl.host));
-  sstrncpy (vl.plugin, "tcpconns", sizeof (vl.plugin));
-  sstrncpy (vl.type, "tcp_connections", sizeof (vl.type));
+  conn_prepare_vl (&vl, values);
 
   if (((port_collect_listening != 0) && (pe->flags & PORT_IS_LISTENING))
       || (pe->flags & PORT_COLLECT_LOCAL))
@@ -307,10 +347,33 @@ static void conn_submit_port_entry (port_entry_t *pe)
   }
 } /* void conn_submit */
 
+static void conn_submit_port_total (void)
+{
+  value_t values[1];
+  value_list_t vl = VALUE_LIST_INIT;
+  int i;
+
+  conn_prepare_vl (&vl, values);
+
+  sstrncpy (vl.plugin_instance, "all", sizeof (vl.plugin_instance));
+
+  for (i = 1; i <= TCP_STATE_MAX; i++)
+  {
+    vl.values[0].gauge = count_total[i];
+
+    sstrncpy (vl.type_instance, tcp_state[i], sizeof (vl.type_instance));
+
+    plugin_dispatch_values (&vl);
+  }
+}
+
 static void conn_submit_all (void)
 {
   port_entry_t *pe;
 
+  if (port_collect_total)
+    conn_submit_port_total ();
+
   for (pe = port_list_head; pe != NULL; pe = pe->next)
     conn_submit_port_entry (pe);
 } /* void conn_submit_all */
@@ -349,23 +412,25 @@ static void conn_reset_port_entry (void)
   port_entry_t *prev = NULL;
   port_entry_t *pe = port_list_head;
 
+  memset (&count_total, '\0', sizeof(count_total));
+
   while (pe != NULL)
   {
     /* If this entry was created while reading the files (ant not when handling
      * the configuration) remove it now. */
     if ((pe->flags & (PORT_COLLECT_LOCAL
-           | PORT_COLLECT_REMOTE
-           | PORT_IS_LISTENING)) == 0)
+            | PORT_COLLECT_REMOTE
+            | PORT_IS_LISTENING)) == 0)
     {
       port_entry_t *next = pe->next;
 
       DEBUG ("tcpconns plugin: Removing temporary entry "
-         "for listening port %"PRIu16, pe->port);
+          "for listening port %"PRIu16, pe->port);
 
       if (prev == NULL)
-       port_list_head = next;
+        port_list_head = next;
       else
-       prev->next = next;
+        prev->next = next;
 
       sfree (pe);
       pe = next;
@@ -377,6 +442,7 @@ static void conn_reset_port_entry (void)
     memset (pe->count_remote, '\0', sizeof (pe->count_remote));
     pe->flags &= ~PORT_IS_LISTENING;
 
+    prev = pe;
     pe = pe->next;
   }
 } /* void conn_reset_port_entry */
@@ -396,6 +462,8 @@ static int conn_handle_ports (uint16_t port_local, uint16_t port_remote, uint8_t
     return (-1);
   }
 
+  count_total[state]++;
+
   /* Listening sockets */
   if ((state == TCP_STATE_LISTEN) && (port_collect_listening != 0))
   {
@@ -419,6 +487,144 @@ static int conn_handle_ports (uint16_t port_local, uint16_t port_remote, uint8_t
 } /* int conn_handle_ports */
 
 #if KERNEL_LINUX
+/* Returns zero on success, less than zero on socket error and greater than
+ * zero on other errors. */
+static int conn_read_netlink (void)
+{
+#if HAVE_STRUCT_LINUX_INET_DIAG_REQ
+  int fd;
+  struct sockaddr_nl nladdr;
+  struct nlreq req;
+  struct msghdr msg;
+  struct iovec iov;
+  struct inet_diag_msg *r;
+  char buf[8192];
+
+  /* If this fails, it's likely a permission problem. We'll fall back to
+   * reading this information from files below. */
+  fd = socket(AF_NETLINK, SOCK_RAW, NETLINK_INET_DIAG);
+  if (fd < 0)
+  {
+    ERROR ("tcpconns plugin: conn_read_netlink: socket(AF_NETLINK, SOCK_RAW, "
+       "NETLINK_INET_DIAG) failed: %s",
+       sstrerror (errno, buf, sizeof (buf)));
+    return (-1);
+  }
+
+  memset(&nladdr, 0, sizeof(nladdr));
+  nladdr.nl_family = AF_NETLINK;
+
+  memset(&req, 0, sizeof(req));
+  req.nlh.nlmsg_len = sizeof(req);
+  req.nlh.nlmsg_type = TCPDIAG_GETSOCK;
+  /* NLM_F_ROOT: return the complete table instead of a single entry.
+   * NLM_F_MATCH: return all entries matching criteria (not implemented)
+   * NLM_F_REQUEST: must be set on all request messages */
+  req.nlh.nlmsg_flags = NLM_F_ROOT | NLM_F_MATCH | NLM_F_REQUEST;
+  req.nlh.nlmsg_pid = 0;
+  /* The sequence_number is used to track our messages. Since netlink is not
+   * reliable, we don't want to end up with a corrupt or incomplete old
+   * message in case the system is/was out of memory. */
+  req.nlh.nlmsg_seq = ++sequence_number;
+  req.r.idiag_family = AF_INET;
+  req.r.idiag_states = 0xfff;
+  req.r.idiag_ext = 0;
+
+  memset(&iov, 0, sizeof(iov));
+  iov.iov_base = &req;
+  iov.iov_len = sizeof(req);
+
+  memset(&msg, 0, sizeof(msg));
+  msg.msg_name = (void*)&nladdr;
+  msg.msg_namelen = sizeof(nladdr);
+  msg.msg_iov = &iov;
+  msg.msg_iovlen = 1;
+
+  if (sendmsg (fd, &msg, 0) < 0)
+  {
+    ERROR ("tcpconns plugin: conn_read_netlink: sendmsg(2) failed: %s",
+       sstrerror (errno, buf, sizeof (buf)));
+    close (fd);
+    return (-1);
+  }
+
+  iov.iov_base = buf;
+  iov.iov_len = sizeof(buf);
+
+  while (1)
+  {
+    int status;
+    struct nlmsghdr *h;
+
+    memset(&msg, 0, sizeof(msg));
+    msg.msg_name = (void*)&nladdr;
+    msg.msg_namelen = sizeof(nladdr);
+    msg.msg_iov = &iov;
+    msg.msg_iovlen = 1;
+
+    status = recvmsg(fd, (void *) &msg, /* flags = */ 0);
+    if (status < 0)
+    {
+      if ((errno == EINTR) || (errno == EAGAIN))
+        continue;
+
+      ERROR ("tcpconns plugin: conn_read_netlink: recvmsg(2) failed: %s",
+         sstrerror (errno, buf, sizeof (buf)));
+      close (fd);
+      return (-1);
+    }
+    else if (status == 0)
+    {
+      close (fd);
+      DEBUG ("tcpconns plugin: conn_read_netlink: Unexpected zero-sized "
+         "reply from netlink socket.");
+      return (0);
+    }
+
+    h = (struct nlmsghdr*)buf;
+    while (NLMSG_OK(h, status))
+    {
+      if (h->nlmsg_seq != sequence_number)
+      {
+       h = NLMSG_NEXT(h, status);
+       continue;
+      }
+
+      if (h->nlmsg_type == NLMSG_DONE)
+      {
+       close (fd);
+       return (0);
+      }
+      else if (h->nlmsg_type == NLMSG_ERROR)
+      {
+       struct nlmsgerr *msg_error;
+
+       msg_error = NLMSG_DATA(h);
+       WARNING ("tcpconns plugin: conn_read_netlink: Received error %i.",
+           msg_error->error);
+
+       close (fd);
+       return (1);
+      }
+
+      r = NLMSG_DATA(h);
+
+      /* This code does not (need to) distinguish between IPv4 and IPv6. */
+      conn_handle_ports (ntohs(r->id.idiag_sport),
+         ntohs(r->id.idiag_dport),
+         r->idiag_state);
+
+      h = NLMSG_NEXT(h, status);
+    } /* while (NLMSG_OK) */
+  } /* while (1) */
+
+  /* Not reached because the while() loop above handles the exit condition. */
+  return (0);
+#else
+  return (1);
+#endif /* HAVE_STRUCT_LINUX_INET_DIAG_REQ */
+} /* int conn_read_netlink */
+
 static int conn_handle_line (char *buffer)
 {
   char *fields[32];
@@ -534,6 +740,13 @@ static int conn_config (const char *key, const char *value)
       else
        pe->flags |= PORT_COLLECT_REMOTE;
   }
+  else if (strcasecmp (key, "AllPortsSummary") == 0)
+  {
+    if (IS_TRUE (value))
+      port_collect_total = 1;
+    else
+      port_collect_total = 0;
+  }
   else
   {
     return (-1);
@@ -545,7 +758,7 @@ static int conn_config (const char *key, const char *value)
 #if KERNEL_LINUX
 static int conn_init (void)
 {
-  if (port_list_head == NULL)
+  if (port_collect_total == 0 && port_list_head == NULL)
     port_collect_listening = 1;
 
   return (0);
@@ -553,25 +766,54 @@ static int conn_init (void)
 
 static int conn_read (void)
 {
-  int errors_num = 0;
+  int status;
 
   conn_reset_port_entry ();
 
-  if (conn_read_file ("/proc/net/tcp") != 0)
-    errors_num++;
-  if (conn_read_file ("/proc/net/tcp6") != 0)
-    errors_num++;
-
-  if (errors_num < 2)
+  if (linux_source == SRC_NETLINK)
   {
-    conn_submit_all ();
+    status = conn_read_netlink ();
   }
-  else
+  else if (linux_source == SRC_PROC)
   {
-    ERROR ("tcpconns plugin: Neither /proc/net/tcp nor /proc/net/tcp6 "
-       "coult be read.");
-    return (-1);
+    int errors_num = 0;
+
+    if (conn_read_file ("/proc/net/tcp") != 0)
+      errors_num++;
+    if (conn_read_file ("/proc/net/tcp6") != 0)
+      errors_num++;
+
+    if (errors_num < 2)
+      status = 0;
+    else
+      status = ENOENT;
   }
+  else /* if (linux_source == SRC_DUNNO) */
+  {
+    /* Try to use netlink for getting this data, it is _much_ faster on systems
+     * with a large amount of connections. */
+    status = conn_read_netlink ();
+    if (status == 0)
+    {
+      INFO ("tcpconns plugin: Reading from netlink succeeded. "
+         "Will use the netlink method from now on.");
+      linux_source = SRC_NETLINK;
+    }
+    else
+    {
+      INFO ("tcpconns plugin: Reading from netlink failed. "
+         "Will read from /proc from now on.");
+      linux_source = SRC_PROC;
+
+      /* return success here to avoid the "plugin failed" message. */
+      return (0);
+    }
+  }
+
+  if (status == 0)
+    conn_submit_all ();
+  else
+    return (status);
 
   return (0);
 } /* int conn_read */
@@ -710,7 +952,9 @@ static int conn_init (void)
 static int conn_read (void)
 {
   struct inpcbtable table;
+#if !defined(__OpenBSD__) && (defined(__NetBSD_Version__) && __NetBSD_Version__ <= 699002700)
   struct inpcb *head;
+#endif
   struct inpcb *next;
   struct inpcb inpcb;
   struct tcpcb tcpcb;
@@ -723,18 +967,30 @@ static int conn_read (void)
   if (status != 0)
     return (-1);
 
+#if defined(__OpenBSD__) || (defined(__NetBSD_Version__) && __NetBSD_Version__ > 699002700)
+  /* inpt_queue is a TAILQ on OpenBSD */
+  /* Get the first pcb */
+  next = (struct inpcb *)TAILQ_FIRST (&table.inpt_queue);
+  while (next)
+#else
   /* Get the `head' pcb */
   head = (struct inpcb *) &(inpcbtable_ptr->inpt_queue);
   /* Get the first pcb */
   next = (struct inpcb *)CIRCLEQ_FIRST (&table.inpt_queue);
 
   while (next != head)
+#endif
   {
     /* Read the pcb pointed to by `next' into `inpcb' */
     kread ((u_long) next, &inpcb, sizeof (inpcb));
 
     /* Advance `next' */
+#if defined(__OpenBSD__) || (defined(__NetBSD_Version__) && __NetBSD_Version__ > 699002700)
+    /* inpt_queue is a TAILQ on OpenBSD */
+    next = (struct inpcb *)TAILQ_NEXT (&inpcb, inp_queue);
+#else
     next = (struct inpcb *)CIRCLEQ_NEXT (&inpcb, inp_queue);
+#endif
 
     /* Ignore sockets, that are not connected. */
 #ifdef __NetBSD__