Fixed collectd-exec issues.

 * Added collectd example config and types.db

Signed-off-by: Toni Uhlig <matzeton@googlemail.com>

3 files changed, 98 insertions, 48 deletions

diff --git a/examples/c-collectd/c-collectd.c b/examples/c-collectd/c-collectd.c
index c4f6686ad..5418f11a2 100644
--- a/examples/c-collectd/c-collectd.c
+++ b/examples/c-collectd/c-collectd.c
@@ -10,6 +10,17 @@
 
 #include "nDPIsrvd.h"
 
+#define LOG(flags, format, ...)                                                                                        \
+    if (quiet == 0)                                                                                                    \
+    {                                                                                                                  \
+        fprintf(stderr, format, __VA_ARGS__);                                                                          \
+        fprintf(stderr, "%s", "\n");                                                                                   \
+    }                                                                                                                  \
+    else                                                                                                               \
+    {                                                                                                                  \
+        syslog(flags, format, __VA_ARGS__);                                                                            \
+    }
+
 static struct nDPIsrvd_socket * sock = NULL;
 static int main_thread_shutdown = 0;
 static int collectd_timerfd = -1;
@@ -18,8 +29,10 @@ static char * serv_optarg = NULL;
 static char * collectd_hostname = NULL;
 static char * collectd_interval = NULL;
 static nDPIsrvd_ull collectd_interval_ull = 0uL;
+static int quiet = 0;
 
-static struct {
+static struct
+{
     uint64_t flow_new_count;
     uint64_t flow_end_count;
     uint64_t flow_idle_count;
@@ -55,17 +68,13 @@ static int create_collectd_timer(void)
 
 static void sighandler(int signum)
 {
-    syslog(LOG_DAEMON | LOG_NOTICE, "Received SIGNAL %d", signum);
+    LOG(LOG_DAEMON | LOG_NOTICE, "Received SIGNAL %d", signum);
 
     if (main_thread_shutdown == 0)
     {
-        syslog(LOG_DAEMON | LOG_NOTICE, "Shutting down ..");
+        LOG(LOG_DAEMON | LOG_NOTICE, "%s", "Shutting down ..");
         main_thread_shutdown = 1;
     }
-    else
-    {
-        syslog(LOG_DAEMON | LOG_NOTICE, "Reader threads are already shutting down, please be patient.");
-    }
 }
 
 static int parse_options(int argc, char ** argv)
@@ -74,14 +83,16 @@ static int parse_options(int argc, char ** argv)
 
     static char const usage[] =
         "Usage: %s "
-        "[-s host] [-i interval]\n\n"
+        "[-s host] [-c hostname] [-i interval] [-q]\n\n"
         "\t-s\tDestination where nDPIsrvd is listening on.\n"
         "\t-c\tCollectd hostname.\n"
         "\t  \tThis value defaults to the environment variable COLLECTD_HOSTNAME.\n"
         "\t-i\tInterval between print statistics to stdout.\n"
-        "\t  \tThis value defaults to the environment variable COLLECTD_INTERVAL.\n";
+        "\t  \tThis value defaults to the environment variable COLLECTD_INTERVAL.\n"
+        "\t-q\tDo not print anything except collectd statistics.\n"
+        "\t  \tAutomatically enabled if environment variables mentioned above are set.\n";
 
-    while ((opt = getopt(argc, argv, "hs:c:i:")) != -1)
+    while ((opt = getopt(argc, argv, "hs:c:i:q")) != -1)
     {
         switch (opt)
         {
@@ -91,14 +102,17 @@ static int parse_options(int argc, char ** argv)
                 break;
             case 'c':
                 free(collectd_hostname);
-                collectd_hostname =  strdup(optarg);
+                collectd_hostname = strdup(optarg);
                 break;
             case 'i':
                 free(collectd_interval);
                 collectd_interval = strdup(optarg);
                 break;
+            case 'q':
+                quiet = 1;
+                break;
             default:
-                fprintf(stderr, usage, argv[0]);
+                LOG(LOG_DAEMON | LOG_ERR, usage, argv[0]);
                 return 1;
         }
     }
@@ -128,37 +142,42 @@ static int parse_options(int argc, char ** argv)
 
     if (str_value_to_ull(collectd_interval, &collectd_interval_ull) != CONVERSION_OK)
     {
-        fprintf(stderr, "%s: Collectd interval `%s' is not a valid number\n", argv[0], collectd_interval);
+        LOG(LOG_DAEMON | LOG_ERR, "Collectd interval `%s' is not a valid number", collectd_interval);
         return 1;
     }
 
     if (nDPIsrvd_setup_address(&sock->address, serv_optarg) != 0)
     {
-        fprintf(stderr, "%s: Could not parse address `%s'\n", argv[0], serv_optarg);
+        LOG(LOG_DAEMON | LOG_ERR, "Could not parse address `%s'", serv_optarg);
         return 1;
     }
 
     if (optind < argc)
     {
-        fprintf(stderr, "Unexpected argument after options\n\n");
-        fprintf(stderr, usage, argv[0]);
+        LOG(LOG_DAEMON | LOG_ERR, "%s", "Unexpected argument after options");
+        if (quiet == 0)
+        {
+            LOG(0, "%s", "");
+            LOG(0, usage, argv[0]);
+        }
         return 1;
     }
 
     return 0;
 }
 
-#define COLLECTD_PUTVAL_N_FORMAT(name) "PUTVAL %s/nDPId/" #name " interval=%llu N:%llu\n"
-#define COLLECTD_PUTVAL_N(value) collectd_hostname, collectd_interval_ull, (unsigned long long int)collectd_statistics.value
+#define COLLECTD_PUTVAL_N_FORMAT(name) "PUTVAL %s/nDPId/" #name " interval=%llu %llu:%llu\n"
+#define COLLECTD_PUTVAL_N(value)                                                                                       \
+    collectd_hostname, collectd_interval_ull, (unsigned long long int)now,                                             \
+        (unsigned long long int)collectd_statistics.value
 static void print_collectd_exec_output(void)
 {
-    printf(COLLECTD_PUTVAL_N_FORMAT(flow_new_count)
-           COLLECTD_PUTVAL_N_FORMAT(flow_end_count)
-           COLLECTD_PUTVAL_N_FORMAT(flow_idle_count)
-           COLLECTD_PUTVAL_N_FORMAT(flow_guessed_count)
-           COLLECTD_PUTVAL_N_FORMAT(flow_detected_count)
-           COLLECTD_PUTVAL_N_FORMAT(flow_detection_update_count)
-           COLLECTD_PUTVAL_N_FORMAT(flow_not_detected_count),
+    time_t now = time(NULL);
+
+    printf(COLLECTD_PUTVAL_N_FORMAT(flow_new_count) COLLECTD_PUTVAL_N_FORMAT(flow_end_count)
+               COLLECTD_PUTVAL_N_FORMAT(flow_idle_count) COLLECTD_PUTVAL_N_FORMAT(flow_guessed_count)
+                   COLLECTD_PUTVAL_N_FORMAT(flow_detected_count) COLLECTD_PUTVAL_N_FORMAT(flow_detection_update_count)
+                       COLLECTD_PUTVAL_N_FORMAT(flow_not_detected_count),
 
            COLLECTD_PUTVAL_N(flow_new_count),
            COLLECTD_PUTVAL_N(flow_end_count),
@@ -184,7 +203,7 @@ static int mainloop(int epollfd)
         {
             if (events[i].events & EPOLLERR)
             {
-                syslog(LOG_DAEMON | LOG_ERR, "Epoll event error: %s", (errno != 0 ? strerror(errno) : "EPOLLERR"));
+                LOG(LOG_DAEMON | LOG_ERR, "Epoll event error: %s", (errno != 0 ? strerror(errno) : "EPOLLERR"));
                 break;
             }
 
@@ -195,12 +214,12 @@ static int mainloop(int epollfd)
                 errno = 0;
                 if (read(collectd_timerfd, &expirations, sizeof(expirations)) != sizeof(expirations))
                 {
-                    syslog(LOG_DAEMON | LOG_ERR, "Could not read timer expirations: %s", strerror(errno));
+                    LOG(LOG_DAEMON | LOG_ERR, "Could not read timer expirations: %s", strerror(errno));
                     return 1;
                 }
                 if (set_collectd_timer() != 0)
                 {
-                    syslog(LOG_DAEMON | LOG_ERR, "Could not set timer: %s", strerror(errno));
+                    LOG(LOG_DAEMON | LOG_ERR, "Could not set timer: %s", strerror(errno));
                     return 1;
                 }
 
@@ -212,14 +231,14 @@ static int mainloop(int epollfd)
                 enum nDPIsrvd_read_return read_ret = nDPIsrvd_read(sock);
                 if (read_ret != READ_OK)
                 {
-                    syslog(LOG_DAEMON | LOG_ERR, "nDPIsrvd read failed with: %s", nDPIsrvd_enum_to_string(read_ret));
+                    LOG(LOG_DAEMON | LOG_ERR, "nDPIsrvd read failed with: %s", nDPIsrvd_enum_to_string(read_ret));
                     return 1;
                 }
 
                 enum nDPIsrvd_parse_return parse_ret = nDPIsrvd_parse(sock);
                 if (parse_ret != PARSE_OK)
                 {
-                    syslog(LOG_DAEMON | LOG_ERR, "nDPIsrvd parse failed with: %s", nDPIsrvd_enum_to_string(parse_ret));
+                    LOG(LOG_DAEMON | LOG_ERR, "nDPIsrvd parse failed with: %s", nDPIsrvd_enum_to_string(parse_ret));
                     return 1;
                 }
             }
@@ -240,22 +259,28 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
     if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "new") != 0)
     {
         collectd_statistics.flow_new_count++;
-    } else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "end") != 0)
+    }
+    else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "end") != 0)
     {
         collectd_statistics.flow_end_count++;
-    } else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "idle") != 0)
+    }
+    else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "idle") != 0)
     {
         collectd_statistics.flow_idle_count++;
-    } else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "guessed") != 0)
+    }
+    else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "guessed") != 0)
     {
         collectd_statistics.flow_guessed_count++;
-    } else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "detected") != 0)
+    }
+    else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "detected") != 0)
     {
         collectd_statistics.flow_detected_count++;
-    } else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "detection-update") != 0)
+    }
+    else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "detection-update") != 0)
     {
         collectd_statistics.flow_detection_update_count++;
-    } else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "not-detected") != 0)
+    }
+    else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "not-detected") != 0)
     {
         collectd_statistics.flow_not_detected_count++;
     }
@@ -267,10 +292,12 @@ int main(int argc, char ** argv)
 {
     int retval = 1;
 
+    openlog("nDPIsrvd-collectd", LOG_CONS, LOG_DAEMON);
+
     sock = nDPIsrvd_init(0, 0, captured_json_callback, NULL);
     if (sock == NULL)
     {
-        fprintf(stderr, "%s: nDPIsrvd socket memory allocation failed!\n", argv[0]);
+        LOG(LOG_DAEMON | LOG_ERR, "%s", "nDPIsrvd socket memory allocation failed!");
         return 1;
     }
 
@@ -279,13 +306,22 @@ int main(int argc, char ** argv)
         return 1;
     }
 
-    printf("Recv buffer size: %u\n", NETWORK_BUFFER_MAX_SIZE);
-    printf("Connecting to `%s'..\n", serv_optarg);
+    if (getenv("COLLECTD_HOSTNAME") == NULL && getenv("COLLECTD_INTERVAL") == NULL)
+    {
+        LOG(LOG_DAEMON | LOG_NOTICE, "Recv buffer size: %u", NETWORK_BUFFER_MAX_SIZE);
+        LOG(LOG_DAEMON | LOG_NOTICE, "Connecting to `%s'..", serv_optarg);
+    }
+    else
+    {
+        quiet = 1;
+        LOG(LOG_DAEMON | LOG_NOTICE, "Collectd hostname: %s", getenv("COLLECTD_HOSTNAME"));
+        LOG(LOG_DAEMON | LOG_NOTICE, "Collectd interval: %llu", collectd_interval_ull);
+    }
 
     enum nDPIsrvd_connect_return connect_ret = nDPIsrvd_connect(sock);
     if (connect_ret != CONNECT_OK)
     {
-        fprintf(stderr, "%s: nDPIsrvd socket connect to %s failed!\n", argv[0], serv_optarg);
+        LOG(LOG_DAEMON | LOG_ERR, "nDPIsrvd socket connect to %s failed!", serv_optarg);
         nDPIsrvd_free(&sock);
         return 1;
     }
@@ -294,18 +330,16 @@ int main(int argc, char ** argv)
     signal(SIGTERM, sighandler);
     signal(SIGPIPE, SIG_IGN);
 
-    openlog("nDPIsrvd-collectd", LOG_CONS, LOG_DAEMON);
-
     int epollfd = epoll_create1(0);
     if (epollfd < 0)
     {
-        syslog(LOG_DAEMON | LOG_ERR, "Error creating epoll: %s", strerror(errno));
+        LOG(LOG_DAEMON | LOG_ERR, "Error creating epoll: %s", strerror(errno));
         return 1;
     }
 
     if (create_collectd_timer() != 0)
     {
-        syslog(LOG_DAEMON | LOG_ERR, "Error creating timer: %s", strerror(errno));
+        LOG(LOG_DAEMON | LOG_ERR, "Error creating timer: %s", strerror(errno));
         return 1;
     }
 
@@ -313,7 +347,7 @@ int main(int argc, char ** argv)
         struct epoll_event timer_event = {.data.fd = collectd_timerfd, .events = EPOLLIN};
         if (epoll_ctl(epollfd, EPOLL_CTL_ADD, collectd_timerfd, &timer_event) < 0)
         {
-            syslog(LOG_DAEMON | LOG_ERR, "Error adding JSON fd to epoll: %s", strerror(errno));
+            LOG(LOG_DAEMON | LOG_ERR, "Error adding JSON fd to epoll: %s", strerror(errno));
             return 1;
         }
     }
@@ -322,12 +356,12 @@ int main(int argc, char ** argv)
         struct epoll_event socket_event = {.data.fd = sock->fd, .events = EPOLLIN};
         if (epoll_ctl(epollfd, EPOLL_CTL_ADD, sock->fd, &socket_event) < 0)
         {
-            syslog(LOG_DAEMON | LOG_ERR, "Error adding nDPIsrvd socket fd to epoll: %s", strerror(errno));
+            LOG(LOG_DAEMON | LOG_ERR, "Error adding nDPIsrvd socket fd to epoll: %s", strerror(errno));
             return 1;
         }
     }
 
-    syslog(LOG_DAEMON | LOG_NOTICE, "%s", "Initialization succeeded.");
+    LOG(LOG_DAEMON | LOG_NOTICE, "%s", "Initialization succeeded.");
     retval = mainloop(epollfd);
 
     nDPIsrvd_free(&sock);
diff --git a/examples/c-collectd/plugin_nDPIsrvd.conf b/examples/c-collectd/plugin_nDPIsrvd.conf
index 0177e89a8..59688b915 100644
--- a/examples/c-collectd/plugin_nDPIsrvd.conf
+++ b/examples/c-collectd/plugin_nDPIsrvd.conf
@@ -1,5 +1,13 @@
 # nDPIsrvd collectd config file
 LoadPlugin exec
 <Plugin exec>
-  Exec "ndpi" "/usr/bin/nDPIsrvd-collectd"
+  Exec "toni"  "/usr/bin/nDPIsrvd-collectd"
+#  Exec "toni" "/usr/bin/nDPIsrvd-collectd" "-s" "127.0.0.1:7000"
 </Plugin>
+
+# Uncomment for testing
+#LoadPlugin write_log
+#LoadPlugin rrdtool
+#<Plugin rrdtool>
+#   DataDir "nDPIsrvd-collectd"
+#</Plugin>
diff --git a/examples/c-collectd/plugin_nDPIsrvd_types.db b/examples/c-collectd/plugin_nDPIsrvd_types.db
new file mode 100644
index 000000000..9fbdcd93b
--- /dev/null
+++ b/examples/c-collectd/plugin_nDPIsrvd_types.db
@@ -0,0 +1,8 @@
+# Add those types to collectd types.db
+flow_new_count              value:GAUGE:0:U
+flow_end_count              value:GAUGE:0:U
+flow_idle_count             value:GAUGE:0:U
+flow_guessed_count          value:GAUGE:0:U
+flow_detected_count         value:GAUGE:0:U
+flow_detection_update_count value:GAUGE:0:U
+flow_not_detected_count     value:GAUGE:0:U