nDPIsrvd collectd-exec overhaul.

 * Install targets updated.
 * Removed nDPIsrvd.h token validation function (done automatically by token_get).

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

8 files changed, 474 insertions, 71 deletions

diff --git a/Makefile b/Makefile
index 61ea2a6e9..829eeb41f 100644
--- a/Makefile
+++ b/Makefile
@@ -95,10 +95,12 @@ endif
 
 install: all
 	$(INSTALL) -d '$(DESTDIR)$(PREFIX)/bin' '$(DESTDIR)$(PREFIX)/sbin'
-	$(INSTALL) $(INSTALL_ARGS) -t '$(DESTDIR)$(PREFIX)/bin' nDPIsrvd
-	$(INSTALL) $(INSTALL_ARGS) -t '$(DESTDIR)$(PREFIX)/sbin' nDPId
-	$(INSTALL) $(INSTALL_ARGS) -t '$(DESTDIR)$(PREFIX)/bin' examples/c-captured/c-captured
-	$(INSTALL) $(INSTALL_ARGS) -t '$(DESTDIR)$(PREFIX)/bin' examples/c-json-stdout/c-json-stdout
+	$(INSTALL) $(INSTALL_ARGS) ./nDPIsrvd '$(DESTDIR)$(PREFIX)/bin'
+	$(INSTALL) $(INSTALL_ARGS) ./nDPId '$(DESTDIR)$(PREFIX)/sbin'
+	$(INSTALL) $(INSTALL_ARGS) ./examples/c-captured/c-captured '$(DESTDIR)$(PREFIX)/bin/nDPIsrvd-captured'
+	$(INSTALL) $(INSTALL_ARGS) ./examples/c-json-stdout/c-json-stdout '$(DESTDIR)$(PREFIX)/bin/nDPIsrvd-json-dump'
+	$(INSTALL) $(INSTALL_ARGS) ./examples/c-collectd/c-collectd '$(DESTDIR)$(PREFIX)/bin/nDPIsrvd-collectd'
+	$(INSTALL) $(INSTALL_ARGS) ./examples/py-flow-info/flow-info.py '$(DESTDIR)$(PREFIX)/bin/nDPIsrvd-flow-info.py'
 ifneq ($(GOCC),)
 	$(INSTALL) $(INSTALL_ARGS) -t '$(DESTDIR)$(PREFIX)/bin' examples/go-dashboard/go-dashboard
 endif
diff --git a/README.md b/README.md
index 8b511179b..04078d332 100644
--- a/README.md
+++ b/README.md
@@ -1,7 +1,7 @@
 # abstract
 
 nDPId is a set of daemons and tools to capture, process and classify network flows.
-It's only dependencies (besides a half-way modern c library and POSIX threads) are libnDPI (>= 3.3.0) and libpcap.
+It's only dependencies (besides a half-way modern c library and POSIX threads) are libnDPI (>= 3.4.0, current github dev branch) and libpcap.
 
 The core daemon nDPId uses pthread but does use mutexes for performance reasons.
 Instead synchronization is achieved by a packet distribution mechanism.
diff --git a/dependencies/nDPIsrvd.h b/dependencies/nDPIsrvd.h
index e38cd8a66..c2fee80d7 100644
--- a/dependencies/nDPIsrvd.h
+++ b/dependencies/nDPIsrvd.h
@@ -495,8 +495,14 @@ static inline struct nDPIsrvd_json_token const *
 token_get(struct nDPIsrvd_socket const * const sock, char const * const key, size_t key_length)
 {
     struct nDPIsrvd_json_token * token = NULL;
+
     HASH_FIND(hh, sock->json.token_table, key, key_length, token);
-    return token;
+    if (token != NULL && token->value_length > 0 && token->value != NULL)
+    {
+        return token;
+    }
+
+    return NULL;
 }
 
 static inline char const *
@@ -515,14 +521,9 @@ token_get_value(struct nDPIsrvd_socket const * const sock, char const * const ke
     return NULL;
 }
 
-static inline int is_token_valid(struct nDPIsrvd_json_token const * const token)
-{
-    return token != NULL && token->value_length > 0 && token->value != NULL;
-}
-
 static inline int token_value_equals(struct nDPIsrvd_json_token const * const token, char const * const value, size_t value_length)
 {
-    if (is_token_valid(token) == 0)
+    if (token == NULL)
     {
         return 0;
     }
@@ -552,7 +553,7 @@ str_value_to_ull(char const * const value_as_string, nDPIsrvd_ull_ptr const valu
 static inline enum nDPIsrvd_conversion_return
 token_value_to_ull(struct nDPIsrvd_json_token const * const token, nDPIsrvd_ull_ptr const value)
 {
-    if (is_token_valid(token) == 0)
+    if (token == NULL)
     {
         return CONVERISON_KEY_NOT_FOUND;
     }
@@ -563,8 +564,7 @@ token_value_to_ull(struct nDPIsrvd_json_token const * const token, nDPIsrvd_ull_
 static inline int nDPIsrvd_build_flow_key(struct nDPIsrvd_flow_key * const key,
                                           struct nDPIsrvd_json_token const * const tokens[nDPIsrvd_FLOW_KEY_TOKENS])
 {
-    if (is_token_valid(tokens[0]) == 0 || is_token_valid(tokens[1]) == 0 ||
-        is_token_valid(tokens[2]) == 0)
+    if (tokens[0] == NULL || tokens[1] == NULL || tokens[2] == NULL)
     {
         return 1;
     }
@@ -741,7 +741,7 @@ static inline enum nDPIsrvd_parse_return nDPIsrvd_parse(struct nDPIsrvd_socket *
 
         struct nDPIsrvd_json_token const * const flow_id = TOKEN_GET_SZ(sock, "flow_id");
         struct nDPIsrvd_flow * flow = NULL;
-        if (is_token_valid(flow_id) != 0)
+        if (flow_id != NULL)
         {
             flow = nDPIsrvd_get_flow(sock, flow_id);
             if (flow == NULL)
@@ -754,7 +754,7 @@ static inline enum nDPIsrvd_parse_return nDPIsrvd_parse(struct nDPIsrvd_socket *
         {
             ret = PARSE_JSON_CALLBACK_ERROR;
         }
-        if (is_token_valid(flow_id) != 0 && nDPIsrvd_check_flow_end(sock, flow) != 0)
+        if (flow_id != NULL && nDPIsrvd_check_flow_end(sock, flow) != 0)
         {
             ret = PARSE_FLOW_MGMT_ERROR;
         }
diff --git a/examples/c-captured/c-captured.c b/examples/c-captured/c-captured.c
index 18d37056d..b5f7646a3 100644
--- a/examples/c-captured/c-captured.c
+++ b/examples/c-captured/c-captured.c
@@ -16,7 +16,6 @@
 #include <time.h>
 #include <unistd.h>
 
-#include "config.h"
 #include "nDPIsrvd.h"
 #include "utarray.h"
 #include "utils.h"
@@ -49,7 +48,7 @@ struct flow_user_data
     UT_array * packets;
 };
 
-struct nDPIsrvd_socket * sock = NULL;
+static struct nDPIsrvd_socket * sock = NULL;
 static int main_thread_shutdown = 0;
 
 static char * pidfile = NULL;
@@ -570,6 +569,29 @@ static int parse_options(int argc, char ** argv)
     return 0;
 }
 
+static int mainloop(void)
+{
+    while (main_thread_shutdown == 0)
+    {
+        errno = 0;
+        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));
+            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));
+            return 1;
+        }
+    }
+
+    return 0;
+}
+
 int main(int argc, char ** argv)
 {
     sock = nDPIsrvd_init(0, sizeof(struct flow_user_data), captured_json_callback, captured_flow_end_callback);
@@ -587,6 +609,14 @@ int main(int argc, char ** argv)
     printf("Recv buffer size: %u\n", NETWORK_BUFFER_MAX_SIZE);
     printf("Connecting to `%s'..\n", serv_optarg);
 
+    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);
+        nDPIsrvd_free(&sock);
+        return 1;
+    }
+
     signal(SIGINT, sighandler);
     signal(SIGTERM, sighandler);
     signal(SIGPIPE, sighandler);
@@ -612,40 +642,10 @@ int main(int argc, char ** argv)
     }
     chmod(datadir, S_IRWXU);
 
-    enum nDPIsrvd_connect_return connect_ret = nDPIsrvd_connect(sock);
-    if (connect_ret != CONNECT_OK)
-    {
-        syslog(LOG_DAEMON | LOG_ERR, "%s: nDPIsrvd socket connect to %s failed!", argv[0], serv_optarg);
-        nDPIsrvd_free(&sock);
-        return 1;
-    }
-
-    while (main_thread_shutdown == 0)
-    {
-        errno = 0;
-        enum nDPIsrvd_read_return read_ret = nDPIsrvd_read(sock);
-        if (read_ret != READ_OK)
-        {
-            syslog(LOG_DAEMON | LOG_ERR,
-                   "%s: nDPIsrvd read failed with: %s",
-                   argv[0],
-                   nDPIsrvd_enum_to_string(read_ret));
-            break;
-        }
-
-        enum nDPIsrvd_parse_return parse_ret = nDPIsrvd_parse(sock);
-        if (parse_ret != PARSE_OK)
-        {
-            syslog(LOG_DAEMON | LOG_ERR,
-                   "%s: nDPIsrvd parse failed with: %s",
-                   argv[0],
-                   nDPIsrvd_enum_to_string(parse_ret));
-            break;
-        }
-    }
+    int retval = mainloop();
 
     nDPIsrvd_free(&sock);
     closelog();
 
-    return 0;
+    return retval;
 }
diff --git a/examples/c-collectd/c-collectd.c b/examples/c-collectd/c-collectd.c
index 5418f11a2..291676817 100644
--- a/examples/c-collectd/c-collectd.c
+++ b/examples/c-collectd/c-collectd.c
@@ -40,6 +40,56 @@ static struct
     uint64_t flow_detected_count;
     uint64_t flow_detection_update_count;
     uint64_t flow_not_detected_count;
+
+    uint64_t flow_packet_count;
+    uint64_t flow_total_bytes;
+    uint64_t flow_risky_count;
+
+    uint64_t flow_breed_safe_count;
+    uint64_t flow_breed_acceptable_count;
+    uint64_t flow_breed_fun_count;
+    uint64_t flow_breed_unsafe_count;
+    uint64_t flow_breed_potentially_dangerous_count;
+    uint64_t flow_breed_dangerous_count;
+    uint64_t flow_breed_unrated_count;
+    uint64_t flow_breed_unknown_count;
+
+    uint64_t flow_category_media_count;
+    uint64_t flow_category_vpn_count;
+    uint64_t flow_category_email_count;
+    uint64_t flow_category_data_transfer_count;
+    uint64_t flow_category_web_count;
+    uint64_t flow_category_social_network_count;
+    uint64_t flow_category_download_count;
+    uint64_t flow_category_game_count;
+    uint64_t flow_category_chat_count;
+    uint64_t flow_category_voip_count;
+    uint64_t flow_category_database_count;
+    uint64_t flow_category_remote_access_count;
+    uint64_t flow_category_cloud_count;
+    uint64_t flow_category_network_count;
+    uint64_t flow_category_collaborative_count;
+    uint64_t flow_category_rpc_count;
+    uint64_t flow_category_streaming_count;
+    uint64_t flow_category_system_count;
+    uint64_t flow_category_software_update_count;
+    uint64_t flow_category_music_count;
+    uint64_t flow_category_video_count;
+    uint64_t flow_category_shopping_count;
+    uint64_t flow_category_productivity_count;
+    uint64_t flow_category_file_sharing_count;
+    uint64_t flow_category_mining_count;
+    uint64_t flow_category_malware_count;
+    uint64_t flow_category_advertisment_count;
+    uint64_t flow_category_other_count;
+    uint64_t flow_category_unknown_count;
+
+    uint64_t flow_l3_ip4_count;
+    uint64_t flow_l3_ip6_count;
+    uint64_t flow_l3_other_count;
+    uint64_t flow_l4_tcp_count;
+    uint64_t flow_l4_udp_count;
+    uint64_t flow_l4_other_count;
 } collectd_statistics = {};
 
 static int set_collectd_timer(void)
@@ -177,7 +227,8 @@ 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),
+                       COLLECTD_PUTVAL_N_FORMAT(flow_not_detected_count) COLLECTD_PUTVAL_N_FORMAT(flow_packet_count)
+                           COLLECTD_PUTVAL_N_FORMAT(flow_total_bytes) COLLECTD_PUTVAL_N_FORMAT(flow_risky_count),
 
            COLLECTD_PUTVAL_N(flow_new_count),
            COLLECTD_PUTVAL_N(flow_end_count),
@@ -185,7 +236,91 @@ static void print_collectd_exec_output(void)
            COLLECTD_PUTVAL_N(flow_guessed_count),
            COLLECTD_PUTVAL_N(flow_detected_count),
            COLLECTD_PUTVAL_N(flow_detection_update_count),
-           COLLECTD_PUTVAL_N(flow_not_detected_count));
+           COLLECTD_PUTVAL_N(flow_not_detected_count),
+           COLLECTD_PUTVAL_N(flow_packet_count),
+           COLLECTD_PUTVAL_N(flow_total_bytes),
+           COLLECTD_PUTVAL_N(flow_risky_count));
+
+    printf(COLLECTD_PUTVAL_N_FORMAT(flow_breed_safe_count) COLLECTD_PUTVAL_N_FORMAT(flow_breed_acceptable_count)
+               COLLECTD_PUTVAL_N_FORMAT(flow_breed_fun_count) COLLECTD_PUTVAL_N_FORMAT(flow_breed_unsafe_count)
+                   COLLECTD_PUTVAL_N_FORMAT(flow_breed_potentially_dangerous_count)
+                       COLLECTD_PUTVAL_N_FORMAT(flow_breed_dangerous_count)
+                           COLLECTD_PUTVAL_N_FORMAT(flow_breed_unrated_count)
+                               COLLECTD_PUTVAL_N_FORMAT(flow_breed_unknown_count),
+
+           COLLECTD_PUTVAL_N(flow_breed_safe_count),
+           COLLECTD_PUTVAL_N(flow_breed_acceptable_count),
+           COLLECTD_PUTVAL_N(flow_breed_fun_count),
+           COLLECTD_PUTVAL_N(flow_breed_unsafe_count),
+           COLLECTD_PUTVAL_N(flow_breed_potentially_dangerous_count),
+           COLLECTD_PUTVAL_N(flow_breed_dangerous_count),
+           COLLECTD_PUTVAL_N(flow_breed_unrated_count),
+           COLLECTD_PUTVAL_N(flow_breed_unknown_count));
+
+    printf(
+        COLLECTD_PUTVAL_N_FORMAT(flow_category_media_count) COLLECTD_PUTVAL_N_FORMAT(
+            flow_category_vpn_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_email_count)
+            COLLECTD_PUTVAL_N_FORMAT(flow_category_data_transfer_count) COLLECTD_PUTVAL_N_FORMAT(
+                flow_category_web_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_social_network_count)
+                COLLECTD_PUTVAL_N_FORMAT(flow_category_download_count) COLLECTD_PUTVAL_N_FORMAT(
+                    flow_category_game_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_chat_count)
+                    COLLECTD_PUTVAL_N_FORMAT(flow_category_voip_count) COLLECTD_PUTVAL_N_FORMAT(
+                        flow_category_database_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_remote_access_count)
+                        COLLECTD_PUTVAL_N_FORMAT(flow_category_cloud_count) COLLECTD_PUTVAL_N_FORMAT(
+                            flow_category_network_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_collaborative_count)
+                            COLLECTD_PUTVAL_N_FORMAT(flow_category_rpc_count) COLLECTD_PUTVAL_N_FORMAT(
+                                flow_category_streaming_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_system_count)
+                                COLLECTD_PUTVAL_N_FORMAT(flow_category_software_update_count) COLLECTD_PUTVAL_N_FORMAT(
+                                    flow_category_music_count) COLLECTD_PUTVAL_N_FORMAT(flow_category_video_count)
+                                    COLLECTD_PUTVAL_N_FORMAT(flow_category_shopping_count)
+                                        COLLECTD_PUTVAL_N_FORMAT(flow_category_productivity_count)
+                                            COLLECTD_PUTVAL_N_FORMAT(flow_category_file_sharing_count)
+                                                COLLECTD_PUTVAL_N_FORMAT(flow_category_mining_count)
+                                                    COLLECTD_PUTVAL_N_FORMAT(flow_category_malware_count)
+                                                        COLLECTD_PUTVAL_N_FORMAT(flow_category_advertisment_count)
+                                                            COLLECTD_PUTVAL_N_FORMAT(flow_category_other_count)
+                                                                COLLECTD_PUTVAL_N_FORMAT(flow_category_unknown_count),
+
+        COLLECTD_PUTVAL_N(flow_category_media_count),
+        COLLECTD_PUTVAL_N(flow_category_vpn_count),
+        COLLECTD_PUTVAL_N(flow_category_email_count),
+        COLLECTD_PUTVAL_N(flow_category_data_transfer_count),
+        COLLECTD_PUTVAL_N(flow_category_web_count),
+        COLLECTD_PUTVAL_N(flow_category_social_network_count),
+        COLLECTD_PUTVAL_N(flow_category_download_count),
+        COLLECTD_PUTVAL_N(flow_category_game_count),
+        COLLECTD_PUTVAL_N(flow_category_chat_count),
+        COLLECTD_PUTVAL_N(flow_category_voip_count),
+        COLLECTD_PUTVAL_N(flow_category_database_count),
+        COLLECTD_PUTVAL_N(flow_category_remote_access_count),
+        COLLECTD_PUTVAL_N(flow_category_cloud_count),
+        COLLECTD_PUTVAL_N(flow_category_network_count),
+        COLLECTD_PUTVAL_N(flow_category_collaborative_count),
+        COLLECTD_PUTVAL_N(flow_category_rpc_count),
+        COLLECTD_PUTVAL_N(flow_category_streaming_count),
+        COLLECTD_PUTVAL_N(flow_category_system_count),
+        COLLECTD_PUTVAL_N(flow_category_software_update_count),
+        COLLECTD_PUTVAL_N(flow_category_music_count),
+        COLLECTD_PUTVAL_N(flow_category_video_count),
+        COLLECTD_PUTVAL_N(flow_category_shopping_count),
+        COLLECTD_PUTVAL_N(flow_category_productivity_count),
+        COLLECTD_PUTVAL_N(flow_category_file_sharing_count),
+        COLLECTD_PUTVAL_N(flow_category_mining_count),
+        COLLECTD_PUTVAL_N(flow_category_malware_count),
+        COLLECTD_PUTVAL_N(flow_category_advertisment_count),
+        COLLECTD_PUTVAL_N(flow_category_other_count),
+        COLLECTD_PUTVAL_N(flow_category_unknown_count));
+
+    printf(COLLECTD_PUTVAL_N_FORMAT(flow_l3_ip4_count) COLLECTD_PUTVAL_N_FORMAT(flow_l3_ip6_count)
+               COLLECTD_PUTVAL_N_FORMAT(flow_l3_other_count) COLLECTD_PUTVAL_N_FORMAT(flow_l4_tcp_count)
+                   COLLECTD_PUTVAL_N_FORMAT(flow_l4_udp_count) COLLECTD_PUTVAL_N_FORMAT(flow_l4_other_count),
+
+           COLLECTD_PUTVAL_N(flow_l3_ip4_count),
+           COLLECTD_PUTVAL_N(flow_l3_ip6_count),
+           COLLECTD_PUTVAL_N(flow_l3_other_count),
+           COLLECTD_PUTVAL_N(flow_l4_tcp_count),
+           COLLECTD_PUTVAL_N(flow_l4_udp_count),
+           COLLECTD_PUTVAL_N(flow_l4_other_count));
 
     memset(&collectd_statistics, 0, sizeof(collectd_statistics));
 }
@@ -248,6 +383,20 @@ static int mainloop(int epollfd)
     return 0;
 }
 
+static uint64_t get_total_flow_bytes(struct nDPIsrvd_socket * const sock)
+{
+    nDPIsrvd_ull total_bytes_ull = 0;
+
+    if (TOKEN_VALUE_TO_ULL(TOKEN_GET_SZ(sock, "flow_tot_l4_data_len"), &total_bytes_ull) == CONVERSION_OK)
+    {
+        return total_bytes_ull;
+    }
+    else
+    {
+        return 0;
+    }
+}
+
 static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_socket * const sock,
                                                             struct nDPIsrvd_flow * const flow)
 {
@@ -259,14 +408,44 @@ 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++;
+
+        struct nDPIsrvd_json_token const * const l3_proto = TOKEN_GET_SZ(sock, "l3_proto");
+        if (TOKEN_VALUE_EQUALS_SZ(l3_proto, "ip4") != 0)
+        {
+            collectd_statistics.flow_l3_ip4_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(l3_proto, "ip6") != 0)
+        {
+            collectd_statistics.flow_l3_ip6_count++;
+        }
+        else if (l3_proto != NULL)
+        {
+            collectd_statistics.flow_l3_other_count++;
+        }
+
+        struct nDPIsrvd_json_token const * const l4_proto = TOKEN_GET_SZ(sock, "l4_proto");
+        if (TOKEN_VALUE_EQUALS_SZ(l3_proto, "tcp") != 0)
+        {
+            collectd_statistics.flow_l4_tcp_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(l3_proto, "tcp") != 0)
+        {
+            collectd_statistics.flow_l4_udp_count++;
+        }
+        else if (l4_proto != NULL)
+        {
+            collectd_statistics.flow_l4_other_count++;
+        }
     }
     else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "end") != 0)
     {
         collectd_statistics.flow_end_count++;
+        collectd_statistics.flow_total_bytes += get_total_flow_bytes(sock);
     }
     else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "idle") != 0)
     {
         collectd_statistics.flow_idle_count++;
+        collectd_statistics.flow_total_bytes += get_total_flow_bytes(sock);
     }
     else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "guessed") != 0)
     {
@@ -275,6 +454,163 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
     else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "detected") != 0)
     {
         collectd_statistics.flow_detected_count++;
+
+        if (TOKEN_GET_SZ(sock, "flow_risk") != NULL)
+        {
+            collectd_statistics.flow_risky_count++;
+        }
+
+        struct nDPIsrvd_json_token const * const breed = TOKEN_GET_SZ(sock, "breed");
+        if (TOKEN_VALUE_EQUALS_SZ(breed, "Safe") != 0)
+        {
+            collectd_statistics.flow_breed_safe_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(breed, "Acceptable") != 0)
+        {
+            collectd_statistics.flow_breed_acceptable_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(breed, "Fun") != 0)
+        {
+            collectd_statistics.flow_breed_fun_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(breed, "Unsafe") != 0)
+        {
+            collectd_statistics.flow_breed_unsafe_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(breed, "Potentially Dangerous") != 0)
+        {
+            collectd_statistics.flow_breed_potentially_dangerous_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(breed, "Dangerous") != 0)
+        {
+            collectd_statistics.flow_breed_dangerous_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(breed, "Unrated") != 0)
+        {
+            collectd_statistics.flow_breed_unrated_count++;
+        }
+        else
+        {
+            collectd_statistics.flow_breed_unknown_count++;
+        }
+
+        struct nDPIsrvd_json_token const * const category = TOKEN_GET_SZ(sock, "category");
+        if (TOKEN_VALUE_EQUALS_SZ(category, "Media") != 0)
+        {
+            collectd_statistics.flow_category_media_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "VPN") != 0)
+        {
+            collectd_statistics.flow_category_vpn_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Email") != 0)
+        {
+            collectd_statistics.flow_category_email_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "DataTransfer") != 0)
+        {
+            collectd_statistics.flow_category_data_transfer_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Web") != 0)
+        {
+            collectd_statistics.flow_category_web_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "SocialNetwork") != 0)
+        {
+            collectd_statistics.flow_category_social_network_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Download-FileTransfer-FileSharing") != 0)
+        {
+            collectd_statistics.flow_category_download_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Game") != 0)
+        {
+            collectd_statistics.flow_category_game_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Chat") != 0)
+        {
+            collectd_statistics.flow_category_chat_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "VoIP") != 0)
+        {
+            collectd_statistics.flow_category_voip_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Database") != 0)
+        {
+            collectd_statistics.flow_category_database_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "RemoteAccess") != 0)
+        {
+            collectd_statistics.flow_category_remote_access_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Cloud") != 0)
+        {
+            collectd_statistics.flow_category_cloud_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Network") != 0)
+        {
+            collectd_statistics.flow_category_network_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Collaborative") != 0)
+        {
+            collectd_statistics.flow_category_collaborative_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "RPC") != 0)
+        {
+            collectd_statistics.flow_category_rpc_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Streaming") != 0)
+        {
+            collectd_statistics.flow_category_streaming_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "System") != 0)
+        {
+            collectd_statistics.flow_category_system_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "SoftwareUpdate") != 0)
+        {
+            collectd_statistics.flow_category_software_update_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Music") != 0)
+        {
+            collectd_statistics.flow_category_music_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Video") != 0)
+        {
+            collectd_statistics.flow_category_video_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Shopping") != 0)
+        {
+            collectd_statistics.flow_category_shopping_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Productivity") != 0)
+        {
+            collectd_statistics.flow_category_productivity_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "FileSharing") != 0)
+        {
+            collectd_statistics.flow_category_file_sharing_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Mining") != 0)
+        {
+            collectd_statistics.flow_category_mining_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Malware") != 0)
+        {
+            collectd_statistics.flow_category_malware_count++;
+        }
+        else if (TOKEN_VALUE_EQUALS_SZ(category, "Advertisement") != 0)
+        {
+            collectd_statistics.flow_category_advertisment_count++;
+        }
+        else if (category != NULL)
+        {
+            collectd_statistics.flow_category_other_count++;
+        }
+        else
+        {
+            collectd_statistics.flow_category_unknown_count++;
+        }
     }
     else if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "detection-update") != 0)
     {
@@ -285,6 +621,11 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
         collectd_statistics.flow_not_detected_count++;
     }
 
+    if (TOKEN_GET_SZ(sock, "packet_event_name") != NULL)
+    {
+        collectd_statistics.flow_packet_count++;
+    }
+
     return CALLBACK_OK;
 }
 
diff --git a/examples/c-collectd/plugin_nDPIsrvd.conf b/examples/c-collectd/plugin_nDPIsrvd.conf
index 59688b915..eedc9b6c9 100644
--- a/examples/c-collectd/plugin_nDPIsrvd.conf
+++ b/examples/c-collectd/plugin_nDPIsrvd.conf
@@ -1,13 +1,14 @@
 # nDPIsrvd collectd config file
 LoadPlugin exec
 <Plugin exec>
-  Exec "toni"  "/usr/bin/nDPIsrvd-collectd"
-#  Exec "toni" "/usr/bin/nDPIsrvd-collectd" "-s" "127.0.0.1:7000"
+  Exec "ndpi"  "/usr/bin/nDPIsrvd-collectd"
+#  Exec "ndpi" "/usr/bin/nDPIsrvd-collectd" "-s" "/tmp/ndpid-distributor.sock"
+#  Exec "ndpi" "/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>
+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
index 9fbdcd93b..7211939c3 100644
--- a/examples/c-collectd/plugin_nDPIsrvd_types.db
+++ b/examples/c-collectd/plugin_nDPIsrvd_types.db
@@ -1,4 +1,7 @@
 # Add those types to collectd types.db
+# e.g. `cat plugin_nDPIsrvd_types.db >>/usr/share/collectd/types.db'
+
+# flow event counters
 flow_new_count              value:GAUGE:0:U
 flow_end_count              value:GAUGE:0:U
 flow_idle_count             value:GAUGE:0:U
@@ -6,3 +9,57 @@ 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
+
+# flow additional counters
+flow_packet_count           value:GAUGE:0:U
+flow_total_bytes            value:GAUGE:0:U
+flow_risky_count            value:GAUGE:0:U
+
+# flow breed counters
+flow_breed_safe_count                   value:GAUGE:0:U
+flow_breed_acceptable_count             value:GAUGE:0:U
+flow_breed_fun_count                    value:GAUGE:0:U
+flow_breed_unsafe_count                 value:GAUGE:0:U
+flow_breed_potentially_dangerous_count  value:GAUGE:0:U
+flow_breed_dangerous_count              value:GAUGE:0:U
+flow_breed_unrated_count                value:GAUGE:0:U
+flow_breed_unknown_count                value:GAUGE:0:U
+
+# flow category counters
+flow_category_media_count               value:GAUGE:0:U
+flow_category_vpn_count                 value:GAUGE:0:U
+flow_category_email_count               value:GAUGE:0:U
+flow_category_data_transfer_count       value:GAUGE:0:U
+flow_category_web_count                 value:GAUGE:0:U
+flow_category_social_network_count      value:GAUGE:0:U
+flow_category_download_count            value:GAUGE:0:U
+flow_category_game_count                value:GAUGE:0:U
+flow_category_chat_count                value:GAUGE:0:U
+flow_category_voip_count                value:GAUGE:0:U
+flow_category_database_count            value:GAUGE:0:U
+flow_category_remote_access_count       value:GAUGE:0:U
+flow_category_cloud_count               value:GAUGE:0:U
+flow_category_network_count             value:GAUGE:0:U
+flow_category_collaborative_count       value:GAUGE:0:U
+flow_category_rpc_count                 value:GAUGE:0:U
+flow_category_streaming_count           value:GAUGE:0:U
+flow_category_system_count              value:GAUGE:0:U
+flow_category_software_update_count     value:GAUGE:0:U
+flow_category_music_count               value:GAUGE:0:U
+flow_category_video_count               value:GAUGE:0:U
+flow_category_shopping_count            value:GAUGE:0:U
+flow_category_productivity_count        value:GAUGE:0:U
+flow_category_file_sharing_count        value:GAUGE:0:U
+flow_category_mining_count              value:GAUGE:0:U
+flow_category_malware_count             value:GAUGE:0:U
+flow_category_advertisment_count        value:GAUGE:0:U
+flow_category_other_count               value:GAUGE:0:U
+flow_category_unknown_count             value:GAUGE:0:U
+
+# flow l3 / l4 counters
+flow_l3_ip4_count       value:GAUGE:0:U
+flow_l3_ip6_count       value:GAUGE:0:U
+flow_l3_other_count     value:GAUGE:0:U
+flow_l4_tcp_count       value:GAUGE:0:U
+flow_l4_udp_count       value:GAUGE:0:U
+flow_l4_other_count     value:GAUGE:0:U
diff --git a/nDPIsrvd.c b/nDPIsrvd.c
index 082a5c53a..0d51bce00 100644
--- a/nDPIsrvd.c
+++ b/nDPIsrvd.c
@@ -719,20 +719,22 @@ int main(int argc, char ** argv)
         goto error;
     }
 
-    struct epoll_event accept_event = {};
-    accept_event.data.fd = json_sockfd;
-    accept_event.events = EPOLLIN;
-    if (epoll_ctl(epollfd, EPOLL_CTL_ADD, json_sockfd, &accept_event) < 0)
     {
-        syslog(LOG_DAEMON | LOG_ERR, "Error adding JSON fd to epoll: %s", strerror(errno));
-        goto error;
+        struct epoll_event accept_event = {.data.fd = json_sockfd, .events = EPOLLIN};
+        if (epoll_ctl(epollfd, EPOLL_CTL_ADD, json_sockfd, &accept_event) < 0)
+        {
+            syslog(LOG_DAEMON | LOG_ERR, "Error adding JSON fd to epoll: %s", strerror(errno));
+            goto error;
+        }
     }
-    accept_event.data.fd = serv_sockfd;
-    accept_event.events = EPOLLIN;
-    if (epoll_ctl(epollfd, EPOLL_CTL_ADD, serv_sockfd, &accept_event) < 0)
+
     {
-        syslog(LOG_DAEMON | LOG_ERR, "Error adding INET fd to epoll: %s", strerror(errno));
-        goto error;
+        struct epoll_event accept_event = {.data.fd = serv_sockfd, .events = EPOLLIN};
+        if (epoll_ctl(epollfd, EPOLL_CTL_ADD, serv_sockfd, &accept_event) < 0)
+        {
+            syslog(LOG_DAEMON | LOG_ERR, "Error adding INET fd to epoll: %s", strerror(errno));
+            goto error;
+        }
     }
 
     retval = mainloop(epollfd);