scottk/nDPId
1
0
Fork 0
mirror of https://github.com/optim-enterprises-bv/nDPId.git synced 2025-10-30 17:57:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fixed collectd-exec issues.

Browse Source 
* Added collectd example config and types.db

Signed-off-by: Toni Uhlig <matzeton@googlemail.com>
This commit is contained in:
Toni Uhlig
2021-03-12 17:28:10 +01:00
parent 772b67b767
commit 9a06b97473
3 changed files with 98 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
examples/c-collectd/c-collectd.c
View File

@@ -10,6 +10,17 @@
#include "nDPIsrvd.h" #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 struct nDPIsrvd_socket * sock = NULL;
static int main_thread_shutdown = 0; static int main_thread_shutdown = 0;
static int collectd_timerfd = -1; static int collectd_timerfd = -1;
@@ -18,8 +29,10 @@ static char * serv_optarg = NULL;
static char * collectd_hostname = NULL; static char * collectd_hostname = NULL;
static char * collectd_interval = NULL; static char * collectd_interval = NULL;
static nDPIsrvd_ull collectd_interval_ull = 0uL; static nDPIsrvd_ull collectd_interval_ull = 0uL;
static int quiet = 0;
static struct { static struct
{
uint64_t flow_new_count; uint64_t flow_new_count;
uint64_t flow_end_count; uint64_t flow_end_count;
uint64_t flow_idle_count; uint64_t flow_idle_count;
@@ -55,17 +68,13 @@ static int create_collectd_timer(void)
static void sighandler(int signum) 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) if (main_thread_shutdown == 0)
{ {
syslog(LOG_DAEMON | LOG_NOTICE, "Shutting down .."); LOG(LOG_DAEMON | LOG_NOTICE, "%s", "Shutting down ..");
main_thread_shutdown = 1; 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) static int parse_options(int argc, char ** argv)
@@ -74,14 +83,16 @@ static int parse_options(int argc, char ** argv)
static char const usage[] = static char const usage[] =
"Usage: %s " "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-s\tDestination where nDPIsrvd is listening on.\n"
"\t-c\tCollectd hostname.\n" "\t-c\tCollectd hostname.\n"
"\t \tThis value defaults to the environment variable COLLECTD_HOSTNAME.\n" "\t \tThis value defaults to the environment variable COLLECTD_HOSTNAME.\n"
"\t-i\tInterval between print statistics to stdout.\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) switch (opt)
{ {
@@ -91,14 +102,17 @@ static int parse_options(int argc, char ** argv)
break; break;
case 'c': case 'c':
free(collectd_hostname); free(collectd_hostname);
collectd_hostname = strdup(optarg); collectd_hostname = strdup(optarg);
break; break;
case 'i': case 'i':
free(collectd_interval); free(collectd_interval);
collectd_interval = strdup(optarg); collectd_interval = strdup(optarg);
break; break;
case 'q':
quiet = 1;
break;
default: default:
fprintf(stderr, usage, argv[0]); LOG(LOG_DAEMON | LOG_ERR, usage, argv[0]);
return 1; 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) 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; return 1;
} }
if (nDPIsrvd_setup_address(&sock->address, serv_optarg) != 0) 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; return 1;
} }
if (optind < argc) if (optind < argc)
{ {
fprintf(stderr, "Unexpected argument after options\n\n"); LOG(LOG_DAEMON | LOG_ERR, "%s", "Unexpected argument after options");
fprintf(stderr, usage, argv[0]); if (quiet == 0)
{
LOG(0, "%s", "");
LOG(0, usage, argv[0]);
}
return 1; return 1;
} }
return 0; return 0;
} }
#define COLLECTD_PUTVAL_N_FORMAT(name) "PUTVAL %s/nDPId/" #name " interval=%llu N:%llu\n" #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)collectd_statistics.value #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) static void print_collectd_exec_output(void)
{ {
printf(COLLECTD_PUTVAL_N_FORMAT(flow_new_count) time_t now = time(NULL);
COLLECTD_PUTVAL_N_FORMAT(flow_end_count)
COLLECTD_PUTVAL_N_FORMAT(flow_idle_count) printf(COLLECTD_PUTVAL_N_FORMAT(flow_new_count) COLLECTD_PUTVAL_N_FORMAT(flow_end_count)
COLLECTD_PUTVAL_N_FORMAT(flow_guessed_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_detected_count) COLLECTD_PUTVAL_N_FORMAT(flow_detection_update_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(flow_new_count), COLLECTD_PUTVAL_N(flow_new_count),
COLLECTD_PUTVAL_N(flow_end_count), COLLECTD_PUTVAL_N(flow_end_count),
@@ -184,7 +203,7 @@ static int mainloop(int epollfd)
{ {
if (events[i].events & EPOLLERR) 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; break;
} }
@@ -195,12 +214,12 @@ static int mainloop(int epollfd)
errno = 0; errno = 0;
if (read(collectd_timerfd, &expirations, sizeof(expirations)) != sizeof(expirations)) 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; return 1;
} }
if (set_collectd_timer() != 0) 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; return 1;
} }
@@ -212,14 +231,14 @@ static int mainloop(int epollfd)
enum nDPIsrvd_read_return read_ret = nDPIsrvd_read(sock); enum nDPIsrvd_read_return read_ret = nDPIsrvd_read(sock);
if (read_ret != READ_OK) 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; return 1;
} }
enum nDPIsrvd_parse_return parse_ret = nDPIsrvd_parse(sock); enum nDPIsrvd_parse_return parse_ret = nDPIsrvd_parse(sock);
if (parse_ret != PARSE_OK) 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; 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) if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "new") != 0)
{ {
collectd_statistics.flow_new_count++; 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++; 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++; 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++; 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++; 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++; 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++; collectd_statistics.flow_not_detected_count++;
} }
@@ -267,10 +292,12 @@ int main(int argc, char ** argv)
{ {
int retval = 1; int retval = 1;
openlog("nDPIsrvd-collectd", LOG_CONS, LOG_DAEMON);
sock = nDPIsrvd_init(0, 0, captured_json_callback, NULL); sock = nDPIsrvd_init(0, 0, captured_json_callback, NULL);
if (sock == 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; return 1;
} }
@@ -279,13 +306,22 @@ int main(int argc, char ** argv)
return 1; return 1;
} }
printf("Recv buffer size: %u\n", NETWORK_BUFFER_MAX_SIZE); if (getenv("COLLECTD_HOSTNAME") == NULL && getenv("COLLECTD_INTERVAL") == NULL)
printf("Connecting to `%s'..\n", serv_optarg); {
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); enum nDPIsrvd_connect_return connect_ret = nDPIsrvd_connect(sock);
if (connect_ret != CONNECT_OK) 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); nDPIsrvd_free(&sock);
return 1; return 1;
} }
@@ -294,18 +330,16 @@ int main(int argc, char ** argv)
signal(SIGTERM, sighandler); signal(SIGTERM, sighandler);
signal(SIGPIPE, SIG_IGN); signal(SIGPIPE, SIG_IGN);
openlog("nDPIsrvd-collectd", LOG_CONS, LOG_DAEMON);
int epollfd = epoll_create1(0); int epollfd = epoll_create1(0);
if (epollfd < 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; return 1;
} }
if (create_collectd_timer() != 0) 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; return 1;
} }
@@ -313,7 +347,7 @@ int main(int argc, char ** argv)
struct epoll_event timer_event = {.data.fd = collectd_timerfd, .events = EPOLLIN}; struct epoll_event timer_event = {.data.fd = collectd_timerfd, .events = EPOLLIN};
if (epoll_ctl(epollfd, EPOLL_CTL_ADD, collectd_timerfd, &timer_event) < 0) 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; return 1;
} }
} }
@@ -322,12 +356,12 @@ int main(int argc, char ** argv)
struct epoll_event socket_event = {.data.fd = sock->fd, .events = EPOLLIN}; struct epoll_event socket_event = {.data.fd = sock->fd, .events = EPOLLIN};
if (epoll_ctl(epollfd, EPOLL_CTL_ADD, sock->fd, &socket_event) < 0) 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; return 1;
} }
} }
syslog(LOG_DAEMON | LOG_NOTICE, "%s", "Initialization succeeded."); LOG(LOG_DAEMON | LOG_NOTICE, "%s", "Initialization succeeded.");
retval = mainloop(epollfd); retval = mainloop(epollfd);
nDPIsrvd_free(&sock); nDPIsrvd_free(&sock);

10
examples/c-collectd/plugin_nDPIsrvd.conf
View File

@@ -1,5 +1,13 @@
# nDPIsrvd collectd config file # nDPIsrvd collectd config file
LoadPlugin exec LoadPlugin exec
<Plugin 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> </Plugin>
# Uncomment for testing
#LoadPlugin write_log
#LoadPlugin rrdtool
#<Plugin rrdtool>
# DataDir "nDPIsrvd-collectd"
#</Plugin>

8
examples/c-collectd/plugin_nDPIsrvd_types.db Normal file
View File

@@ -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