mirror of
https://github.com/optim-enterprises-bv/nDPId-2.git
synced 2025-10-31 10:17:51 +00:00
1937 lines
63 KiB
C
1937 lines
63 KiB
C
#if defined(__FreeBSD__) || defined(__APPLE__)
|
|
#include <sys/stat.h>
|
|
#endif
|
|
#include <arpa/inet.h>
|
|
#include <errno.h>
|
|
#include <fcntl.h>
|
|
#include <netdb.h>
|
|
#include <netinet/tcp.h>
|
|
#include <pwd.h>
|
|
#include <signal.h>
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#include <stdint.h>
|
|
#include <string.h>
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
#include <sys/signalfd.h>
|
|
#endif
|
|
#include <sys/socket.h>
|
|
#include <sys/types.h>
|
|
#include <unistd.h>
|
|
|
|
#include "config.h"
|
|
#include "nDPIsrvd.h"
|
|
#include "nio.h"
|
|
#include "utils.h"
|
|
|
|
enum sock_type
|
|
{
|
|
COLLECTOR_UN,
|
|
DISTRIBUTOR_UN,
|
|
DISTRIBUTOR_IN,
|
|
};
|
|
|
|
struct nDPIsrvd_write_buffer
|
|
{
|
|
struct nDPIsrvd_buffer buf;
|
|
size_t written;
|
|
};
|
|
|
|
struct remote_desc
|
|
{
|
|
enum sock_type sock_type;
|
|
int fd;
|
|
|
|
union
|
|
{
|
|
struct
|
|
{
|
|
struct sockaddr_un peer;
|
|
unsigned long long int json_bytes;
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
pid_t pid;
|
|
#endif
|
|
|
|
struct nDPIsrvd_json_buffer main_read_buffer;
|
|
} event_collector_un;
|
|
struct
|
|
{
|
|
struct sockaddr_un peer;
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
pid_t pid;
|
|
char * user_name;
|
|
#endif
|
|
|
|
struct nDPIsrvd_write_buffer main_write_buffer;
|
|
UT_array * additional_write_buffers;
|
|
} event_distributor_un; /* UNIX socket */
|
|
struct
|
|
{
|
|
struct sockaddr_in peer;
|
|
char peer_addr[INET_ADDRSTRLEN];
|
|
|
|
struct nDPIsrvd_write_buffer main_write_buffer;
|
|
UT_array * additional_write_buffers;
|
|
} event_distributor_in; /* TCP/IP socket */
|
|
};
|
|
};
|
|
|
|
static struct
|
|
{
|
|
struct remote_desc * desc;
|
|
nDPIsrvd_ull desc_size;
|
|
nDPIsrvd_ull desc_used;
|
|
} remotes = {NULL, 0, 0};
|
|
|
|
static int nDPIsrvd_main_thread_shutdown = 0;
|
|
static int collector_un_sockfd = -1;
|
|
static int distributor_un_sockfd = -1;
|
|
static int distributor_in_sockfd = -1;
|
|
static struct nDPIsrvd_address distributor_in_address = {
|
|
.raw.sa_family = (sa_family_t)0xFFFF,
|
|
};
|
|
|
|
static struct
|
|
{
|
|
struct cmdarg config_file;
|
|
struct cmdarg pidfile;
|
|
struct cmdarg collector_un_sockpath;
|
|
struct cmdarg distributor_un_sockpath;
|
|
struct cmdarg distributor_in_address;
|
|
struct cmdarg user;
|
|
struct cmdarg group;
|
|
struct cmdarg collector_group;
|
|
struct cmdarg distributor_group;
|
|
struct cmdarg max_remote_descriptors;
|
|
struct cmdarg max_write_buffers;
|
|
struct cmdarg bufferbloat_fallback_to_blocking;
|
|
#ifdef ENABLE_EPOLL
|
|
struct cmdarg use_poll;
|
|
#endif
|
|
} nDPIsrvd_options = {.config_file = CMDARG_STR(NULL),
|
|
.pidfile = CMDARG_STR(nDPIsrvd_PIDFILE),
|
|
.collector_un_sockpath = CMDARG_STR(COLLECTOR_UNIX_SOCKET),
|
|
.distributor_un_sockpath = CMDARG_STR(DISTRIBUTOR_UNIX_SOCKET),
|
|
.distributor_in_address = CMDARG_STR(NULL),
|
|
.user = CMDARG_STR(DEFAULT_CHUSER),
|
|
.group = CMDARG_STR(NULL),
|
|
.collector_group = CMDARG_STR(NULL),
|
|
.distributor_group = CMDARG_STR(NULL),
|
|
.max_remote_descriptors = CMDARG_ULL(nDPIsrvd_MAX_REMOTE_DESCRIPTORS),
|
|
.max_write_buffers = CMDARG_ULL(nDPIsrvd_MAX_WRITE_BUFFERS),
|
|
.bufferbloat_fallback_to_blocking = CMDARG_BOOL(1)
|
|
#ifdef ENABLE_EPOLL
|
|
,
|
|
.use_poll = CMDARG_BOOL(0)
|
|
#endif
|
|
};
|
|
struct confopt config_map[] = {CONFOPT("pidfile", &nDPIsrvd_options.pidfile),
|
|
CONFOPT("collector", &nDPIsrvd_options.collector_un_sockpath),
|
|
CONFOPT("distributor-unix", &nDPIsrvd_options.distributor_un_sockpath),
|
|
CONFOPT("distributor-in", &nDPIsrvd_options.distributor_in_address),
|
|
CONFOPT("user", &nDPIsrvd_options.user),
|
|
CONFOPT("group", &nDPIsrvd_options.group),
|
|
CONFOPT("collector-group", &nDPIsrvd_options.collector_group),
|
|
CONFOPT("distributor-group", &nDPIsrvd_options.distributor_group),
|
|
CONFOPT("max-remote-descriptors", &nDPIsrvd_options.max_remote_descriptors),
|
|
CONFOPT("max-write-buffers", &nDPIsrvd_options.max_write_buffers),
|
|
CONFOPT("blocking-io-fallback", &nDPIsrvd_options.bufferbloat_fallback_to_blocking)
|
|
#ifdef ENABLE_EPOLL
|
|
,
|
|
CONFOPT("poll", &nDPIsrvd_options.use_poll)
|
|
#endif
|
|
};
|
|
|
|
static void logger_nDPIsrvd(struct remote_desc const * const remote,
|
|
char const * const prefix,
|
|
char const * const format,
|
|
...);
|
|
static int fcntl_add_flags(int fd, int flags);
|
|
static int fcntl_del_flags(int fd, int flags);
|
|
static int add_in_event_fd(struct nio * const io, int fd);
|
|
static int add_in_event(struct nio * const io, struct remote_desc * const remote);
|
|
static int del_event(struct nio * const io, int fd);
|
|
static int set_in_event(struct nio * const io, struct remote_desc * const remote);
|
|
static void disconnect_client(struct nio * const io, struct remote_desc * const current);
|
|
static int drain_write_buffers_blocking(struct remote_desc * const remote);
|
|
|
|
static void nDPIsrvd_buffer_array_copy(void * dst, const void * src)
|
|
{
|
|
struct nDPIsrvd_write_buffer * const buf_dst = (struct nDPIsrvd_write_buffer *)dst;
|
|
struct nDPIsrvd_write_buffer const * const buf_src = (struct nDPIsrvd_write_buffer *)src;
|
|
|
|
buf_dst->buf.ptr.raw = NULL;
|
|
if (nDPIsrvd_buffer_init(&buf_dst->buf, buf_src->buf.used) != 0)
|
|
{
|
|
logger(1, "Additional write buffer init failed, size: %zu bytes", buf_src->buf.used);
|
|
return;
|
|
}
|
|
|
|
buf_dst->written = buf_src->written;
|
|
buf_dst->buf.used = buf_src->buf.used;
|
|
memcpy(buf_dst->buf.ptr.raw, buf_src->buf.ptr.raw, buf_src->buf.used);
|
|
}
|
|
|
|
static void nDPIsrvd_buffer_array_dtor(void * elt)
|
|
{
|
|
struct nDPIsrvd_write_buffer * const buf_dst = (struct nDPIsrvd_write_buffer *)elt;
|
|
|
|
nDPIsrvd_buffer_free(&buf_dst->buf);
|
|
buf_dst->written = 0;
|
|
}
|
|
|
|
static const UT_icd nDPIsrvd_buffer_array_icd = {sizeof(struct nDPIsrvd_write_buffer),
|
|
NULL,
|
|
nDPIsrvd_buffer_array_copy,
|
|
nDPIsrvd_buffer_array_dtor};
|
|
|
|
#ifndef NO_MAIN
|
|
#ifdef ENABLE_MEMORY_PROFILING
|
|
void nDPIsrvd_memprof_log_alloc(size_t alloc_size)
|
|
{
|
|
(void)alloc_size;
|
|
}
|
|
|
|
void nDPIsrvd_memprof_log_free(size_t free_size)
|
|
{
|
|
(void)free_size;
|
|
}
|
|
|
|
void nDPIsrvd_memprof_log(char const * const format, ...)
|
|
{
|
|
va_list ap;
|
|
|
|
va_start(ap, format);
|
|
vlogger(0, format, ap);
|
|
va_end(ap);
|
|
}
|
|
#endif
|
|
#endif
|
|
|
|
static struct nDPIsrvd_json_buffer * get_read_buffer(struct remote_desc * const remote)
|
|
{
|
|
switch (remote->sock_type)
|
|
{
|
|
case COLLECTOR_UN:
|
|
return &remote->event_collector_un.main_read_buffer;
|
|
|
|
case DISTRIBUTOR_UN:
|
|
case DISTRIBUTOR_IN:
|
|
return NULL;
|
|
}
|
|
|
|
return NULL;
|
|
}
|
|
|
|
static struct nDPIsrvd_write_buffer * get_write_buffer(struct remote_desc * const remote)
|
|
{
|
|
switch (remote->sock_type)
|
|
{
|
|
case COLLECTOR_UN:
|
|
return NULL;
|
|
|
|
case DISTRIBUTOR_UN:
|
|
return &remote->event_distributor_un.main_write_buffer;
|
|
|
|
case DISTRIBUTOR_IN:
|
|
return &remote->event_distributor_in.main_write_buffer;
|
|
}
|
|
|
|
return NULL;
|
|
}
|
|
|
|
static UT_array * get_additional_write_buffers(struct remote_desc * const remote)
|
|
{
|
|
switch (remote->sock_type)
|
|
{
|
|
case COLLECTOR_UN:
|
|
return NULL;
|
|
|
|
case DISTRIBUTOR_UN:
|
|
return remote->event_distributor_un.additional_write_buffers;
|
|
|
|
case DISTRIBUTOR_IN:
|
|
return remote->event_distributor_in.additional_write_buffers;
|
|
}
|
|
|
|
return NULL;
|
|
}
|
|
|
|
static int add_to_additional_write_buffers(struct remote_desc * const remote,
|
|
uint8_t * const buf,
|
|
nDPIsrvd_ull json_message_length)
|
|
{
|
|
struct nDPIsrvd_write_buffer buf_src = {};
|
|
UT_array * const additional_write_buffers = get_additional_write_buffers(remote);
|
|
|
|
if (additional_write_buffers == NULL)
|
|
{
|
|
return -1;
|
|
}
|
|
|
|
if (utarray_len(additional_write_buffers) >= GET_CMDARG_ULL(nDPIsrvd_options.max_write_buffers))
|
|
{
|
|
if (GET_CMDARG_BOOL(nDPIsrvd_options.bufferbloat_fallback_to_blocking) == 0)
|
|
{
|
|
logger_nDPIsrvd(remote,
|
|
"Buffer limit for",
|
|
"for reached, remote too slow: %u lines",
|
|
utarray_len(additional_write_buffers));
|
|
logger_nDPIsrvd(remote, "%s", "You can try to increase buffer limits with `-C'.");
|
|
return -1;
|
|
}
|
|
else
|
|
{
|
|
logger_nDPIsrvd(remote,
|
|
"Buffer limit for",
|
|
"reached, falling back to blocking I/O: %u lines",
|
|
utarray_len(additional_write_buffers));
|
|
if (drain_write_buffers_blocking(remote) != 0)
|
|
{
|
|
return -1;
|
|
}
|
|
}
|
|
}
|
|
|
|
buf_src.buf.ptr.raw = buf;
|
|
buf_src.buf.used = buf_src.buf.max = json_message_length;
|
|
utarray_push_back(additional_write_buffers, &buf_src);
|
|
|
|
return 0;
|
|
}
|
|
|
|
static void logger_nDPIsrvd(struct remote_desc const * const remote,
|
|
char const * const prefix,
|
|
char const * const format,
|
|
...)
|
|
{
|
|
char logbuf[512];
|
|
va_list ap;
|
|
|
|
va_start(ap, format);
|
|
vsnprintf(logbuf, sizeof(logbuf), format, ap);
|
|
|
|
switch (remote->sock_type)
|
|
{
|
|
case DISTRIBUTOR_UN:
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
logger(1,
|
|
"%s PID %d (User: %s) %s",
|
|
prefix,
|
|
remote->event_distributor_un.pid,
|
|
remote->event_distributor_un.user_name,
|
|
logbuf);
|
|
#else
|
|
logger(1, "%s %s", prefix, logbuf);
|
|
#endif
|
|
break;
|
|
case DISTRIBUTOR_IN:
|
|
logger(1,
|
|
"%s %.*s:%u %s",
|
|
prefix,
|
|
(int)sizeof(remote->event_distributor_in.peer_addr),
|
|
remote->event_distributor_in.peer_addr,
|
|
ntohs(remote->event_distributor_in.peer.sin_port),
|
|
logbuf);
|
|
break;
|
|
case COLLECTOR_UN:
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
logger(1, "%s PID %d %s", prefix, remote->event_collector_un.pid, logbuf);
|
|
#else
|
|
logger(1, "%s %s", prefix, logbuf);
|
|
#endif
|
|
break;
|
|
}
|
|
|
|
va_end(ap);
|
|
}
|
|
|
|
static int drain_main_buffer(struct remote_desc * const remote)
|
|
{
|
|
ssize_t bytes_written;
|
|
struct nDPIsrvd_write_buffer * const write_buffer = get_write_buffer(remote);
|
|
|
|
if (write_buffer == NULL)
|
|
{
|
|
return -1;
|
|
}
|
|
|
|
if (write_buffer->buf.used == 0)
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
errno = 0;
|
|
while ((bytes_written = write(remote->fd, write_buffer->buf.ptr.raw, write_buffer->buf.used)) < 0 && errno == EINTR)
|
|
{
|
|
errno = 0;
|
|
}
|
|
if (errno == EAGAIN)
|
|
{
|
|
return 0;
|
|
}
|
|
if (bytes_written < 0 || errno != 0)
|
|
{
|
|
logger_nDPIsrvd(remote, "Distributor connection", "closed, send failed: %s", strerror(errno));
|
|
return -1;
|
|
}
|
|
if (bytes_written == 0)
|
|
{
|
|
logger_nDPIsrvd(remote, "Distributor connection", "closed");
|
|
return -1;
|
|
}
|
|
if ((size_t)bytes_written < write_buffer->buf.used)
|
|
{
|
|
#if 0
|
|
logger_nDPIsrvd(
|
|
remote, "Distributor", "wrote less than expected: %zd < %zu", bytes_written, remote->buf.used);
|
|
#endif
|
|
memmove(write_buffer->buf.ptr.raw,
|
|
write_buffer->buf.ptr.raw + bytes_written,
|
|
write_buffer->buf.used - bytes_written);
|
|
}
|
|
|
|
write_buffer->buf.used -= bytes_written;
|
|
return 0;
|
|
}
|
|
|
|
static int drain_write_buffers(struct remote_desc * const remote)
|
|
{
|
|
UT_array * const additional_write_buffers = get_additional_write_buffers(remote);
|
|
|
|
errno = 0;
|
|
|
|
if (drain_main_buffer(remote) != 0 || additional_write_buffers == NULL)
|
|
{
|
|
return -1;
|
|
}
|
|
|
|
while (utarray_len(additional_write_buffers) > 0)
|
|
{
|
|
struct nDPIsrvd_write_buffer * buf = (struct nDPIsrvd_write_buffer *)utarray_front(additional_write_buffers);
|
|
ssize_t written;
|
|
|
|
while ((written = write(remote->fd, buf->buf.ptr.raw + buf->written, buf->buf.used - buf->written)) < 0 &&
|
|
errno == EINTR)
|
|
{
|
|
// Retry if interrupted by a signal.
|
|
}
|
|
switch (written)
|
|
{
|
|
case -1:
|
|
if (errno == EAGAIN)
|
|
{
|
|
return 0;
|
|
}
|
|
return -1;
|
|
case 0:
|
|
return -1;
|
|
default:
|
|
buf->written += written;
|
|
if (buf->written == buf->buf.max)
|
|
{
|
|
utarray_erase(additional_write_buffers, 0, 1);
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int drain_write_buffers_blocking(struct remote_desc * const remote)
|
|
{
|
|
int retval = 0;
|
|
|
|
if (fcntl_del_flags(remote->fd, O_NONBLOCK) != 0)
|
|
{
|
|
logger_nDPIsrvd(remote, "Error setting distributor", "fd flags to blocking mode: %s", strerror(errno));
|
|
return -1;
|
|
}
|
|
if (drain_write_buffers(remote) != 0)
|
|
{
|
|
logger_nDPIsrvd(remote, "Could not drain buffers for", "in blocking I/O: %s", strerror(errno));
|
|
retval = -1;
|
|
}
|
|
if (fcntl_add_flags(remote->fd, O_NONBLOCK) != 0)
|
|
{
|
|
logger_nDPIsrvd(remote, "Error setting distributor", "fd flags to non-blocking mode: %s", strerror(errno));
|
|
return -1;
|
|
}
|
|
|
|
return retval;
|
|
}
|
|
|
|
static int handle_outgoing_data(struct nio * const io, struct remote_desc * const remote)
|
|
{
|
|
UT_array * const additional_write_buffers = get_additional_write_buffers(remote);
|
|
|
|
if (additional_write_buffers == NULL)
|
|
{
|
|
return -1;
|
|
}
|
|
if (drain_write_buffers(remote) != 0)
|
|
{
|
|
logger_nDPIsrvd(remote, "Could not drain buffers for", ": %s", strerror(errno));
|
|
disconnect_client(io, remote);
|
|
return -1;
|
|
}
|
|
if (utarray_len(additional_write_buffers) == 0)
|
|
{
|
|
struct nDPIsrvd_write_buffer const * const write_buffer = get_write_buffer(remote);
|
|
|
|
if (write_buffer->buf.used == 0)
|
|
{
|
|
return set_in_event(io, remote);
|
|
}
|
|
else
|
|
{
|
|
return drain_main_buffer(remote);
|
|
}
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int fcntl_add_flags(int fd, int flags)
|
|
{
|
|
int cur_flags = fcntl(fd, F_GETFL, 0);
|
|
|
|
if (cur_flags == -1)
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
return fcntl(fd, F_SETFL, cur_flags | flags);
|
|
}
|
|
|
|
static int fcntl_del_flags(int fd, int flags)
|
|
{
|
|
int cur_flags = fcntl(fd, F_GETFL, 0);
|
|
|
|
if (cur_flags == -1)
|
|
{
|
|
return -1;
|
|
}
|
|
|
|
return fcntl(fd, F_SETFL, cur_flags & ~flags);
|
|
}
|
|
|
|
static int create_listen_sockets(void)
|
|
{
|
|
collector_un_sockfd = socket(AF_UNIX, SOCK_STREAM, 0);
|
|
distributor_un_sockfd = socket(AF_UNIX, SOCK_STREAM, 0);
|
|
if (collector_un_sockfd < 0 || distributor_un_sockfd < 0 || set_fd_cloexec(collector_un_sockfd) < 0 ||
|
|
set_fd_cloexec(distributor_un_sockfd) < 0)
|
|
{
|
|
logger(1, "Error creating UNIX socket: %s", strerror(errno));
|
|
return 1;
|
|
}
|
|
|
|
if (IS_CMDARG_SET(nDPIsrvd_options.distributor_in_address) != 0)
|
|
{
|
|
distributor_in_sockfd = socket(distributor_in_address.raw.sa_family, SOCK_STREAM, 0);
|
|
if (distributor_in_sockfd < 0 || set_fd_cloexec(distributor_in_sockfd) < 0)
|
|
{
|
|
logger(1, "Error creating TCP/IP socket: %s", strerror(errno));
|
|
return 1;
|
|
}
|
|
int opt = 1;
|
|
if (setsockopt(distributor_in_sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0)
|
|
{
|
|
logger(1, "Setting TCP/IP socket option SO_REUSEADDR failed: %s", strerror(errno));
|
|
}
|
|
}
|
|
|
|
{
|
|
int opt = 1;
|
|
if (setsockopt(collector_un_sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0 ||
|
|
setsockopt(distributor_un_sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0)
|
|
{
|
|
logger(1, "Setting UNIX socket option SO_REUSEADDR failed: %s", strerror(errno));
|
|
}
|
|
}
|
|
|
|
{
|
|
struct sockaddr_un collector_addr;
|
|
collector_addr.sun_family = AF_UNIX;
|
|
int written = snprintf(collector_addr.sun_path,
|
|
sizeof(collector_addr.sun_path),
|
|
"%s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath));
|
|
if (written < 0)
|
|
{
|
|
logger(1, "snprintf failed: %s", strerror(errno));
|
|
return 1;
|
|
}
|
|
else if (written == sizeof(collector_addr.sun_path))
|
|
{
|
|
logger(1, "Collector UNIX socket path too long, max: %zu characters", sizeof(collector_addr.sun_path) - 1);
|
|
return 1;
|
|
}
|
|
|
|
if (bind(collector_un_sockfd, (struct sockaddr *)&collector_addr, sizeof(collector_addr)) < 0)
|
|
{
|
|
logger(1,
|
|
"Error binding Collector UNIX socket to `%s': %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath),
|
|
strerror(errno));
|
|
return 1;
|
|
}
|
|
}
|
|
|
|
{
|
|
struct sockaddr_un distributor_addr;
|
|
distributor_addr.sun_family = AF_UNIX;
|
|
int written = snprintf(distributor_addr.sun_path,
|
|
sizeof(distributor_addr.sun_path),
|
|
"%s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath));
|
|
if (written < 0)
|
|
{
|
|
logger(1, "snprintf failed: %s", strerror(errno));
|
|
return 2;
|
|
}
|
|
else if (written == sizeof(distributor_addr.sun_path))
|
|
{
|
|
logger(1,
|
|
"Distributor UNIX socket path too long, max: %zu characters",
|
|
sizeof(distributor_addr.sun_path) - 1);
|
|
return 2;
|
|
}
|
|
|
|
if (bind(distributor_un_sockfd, (struct sockaddr *)&distributor_addr, sizeof(distributor_addr)) < 0)
|
|
{
|
|
logger(1,
|
|
"Error binding Distributor socket to `%s': %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath),
|
|
strerror(errno));
|
|
return 2;
|
|
}
|
|
}
|
|
|
|
if (IS_CMDARG_SET(nDPIsrvd_options.distributor_in_address) != 0)
|
|
{
|
|
if (bind(distributor_in_sockfd, &distributor_in_address.raw, distributor_in_address.size) < 0)
|
|
{
|
|
logger(1,
|
|
"Error binding Distributor TCP/IP socket to %s: %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_in_address),
|
|
strerror(errno));
|
|
return 3;
|
|
}
|
|
if (listen(distributor_in_sockfd, 16) < 0)
|
|
{
|
|
logger(1,
|
|
"Error listening Distributor TCP/IP socket to %s: %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_in_address),
|
|
strerror(errno));
|
|
return 3;
|
|
}
|
|
if (fcntl_add_flags(distributor_in_sockfd, O_NONBLOCK) != 0)
|
|
{
|
|
logger(1,
|
|
"Error setting Distributor TCP/IP socket %s to non-blocking mode: %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_in_address),
|
|
strerror(errno));
|
|
return 3;
|
|
}
|
|
}
|
|
|
|
if (listen(collector_un_sockfd, 16) < 0 || listen(distributor_un_sockfd, 16) < 0)
|
|
{
|
|
logger(1, "Error listening UNIX socket: %s", strerror(errno));
|
|
return 3;
|
|
}
|
|
|
|
if (fcntl_add_flags(collector_un_sockfd, O_NONBLOCK) != 0)
|
|
{
|
|
logger(1,
|
|
"Error setting Collector UNIX socket `%s' to non-blocking mode: %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath),
|
|
strerror(errno));
|
|
return 3;
|
|
}
|
|
|
|
if (fcntl_add_flags(distributor_un_sockfd, O_NONBLOCK) != 0)
|
|
{
|
|
logger(1,
|
|
"Error setting Distributor UNIX socket `%s' to non-blocking mode: %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath),
|
|
strerror(errno));
|
|
return 3;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static struct remote_desc * get_remote_descriptor(enum sock_type type, int remote_fd, size_t max_buffer_size)
|
|
{
|
|
if (remotes.desc_used == remotes.desc_size)
|
|
{
|
|
logger(1, "Max number of connections reached: %llu", remotes.desc_used);
|
|
return NULL;
|
|
}
|
|
|
|
for (size_t i = 0; i < remotes.desc_size; ++i)
|
|
{
|
|
if (remotes.desc[i].fd == -1)
|
|
{
|
|
remotes.desc_used++;
|
|
|
|
struct nDPIsrvd_write_buffer * write_buffer = NULL;
|
|
UT_array ** additional_write_buffers = NULL;
|
|
|
|
switch (type)
|
|
{
|
|
case COLLECTOR_UN:
|
|
if (nDPIsrvd_json_buffer_init(&remotes.desc[i].event_collector_un.main_read_buffer,
|
|
max_buffer_size) != 0)
|
|
{
|
|
logger(1, "Read/JSON buffer init failed, size: %zu bytes", max_buffer_size);
|
|
return NULL;
|
|
}
|
|
break;
|
|
case DISTRIBUTOR_UN:
|
|
write_buffer = &remotes.desc[i].event_distributor_un.main_write_buffer;
|
|
additional_write_buffers = &remotes.desc[i].event_distributor_un.additional_write_buffers;
|
|
break;
|
|
case DISTRIBUTOR_IN:
|
|
write_buffer = &remotes.desc[i].event_distributor_in.main_write_buffer;
|
|
additional_write_buffers = &remotes.desc[i].event_distributor_in.additional_write_buffers;
|
|
break;
|
|
}
|
|
|
|
if (additional_write_buffers != NULL && *additional_write_buffers == NULL)
|
|
{
|
|
utarray_new(*additional_write_buffers, &nDPIsrvd_buffer_array_icd);
|
|
if (*additional_write_buffers == NULL)
|
|
{
|
|
logger(1, "%s", "Could not create additional write buffers");
|
|
return NULL;
|
|
}
|
|
}
|
|
if (write_buffer != NULL && nDPIsrvd_buffer_init(&write_buffer->buf, max_buffer_size) != 0)
|
|
{
|
|
logger(1, "Write buffer init failed, size: %zu bytes", max_buffer_size);
|
|
return NULL;
|
|
}
|
|
|
|
remotes.desc[i].sock_type = type;
|
|
remotes.desc[i].fd = remote_fd;
|
|
return &remotes.desc[i];
|
|
}
|
|
}
|
|
|
|
logger(1, "%s", "BUG: Unknown error while finding the remote descriptor");
|
|
return NULL;
|
|
}
|
|
|
|
static void free_remote(struct nio * const io, struct remote_desc * remote)
|
|
{
|
|
if (remote->fd > -1)
|
|
{
|
|
errno = 0;
|
|
if (del_event(io, remote->fd) != 0)
|
|
{
|
|
logger_nDPIsrvd(remote,
|
|
"Could not delete event from queue for connection",
|
|
": %s",
|
|
(errno != 0 ? strerror(errno) : "Internal Error"));
|
|
}
|
|
errno = 0;
|
|
close(remote->fd);
|
|
|
|
switch (remote->sock_type)
|
|
{
|
|
case COLLECTOR_UN:
|
|
if (errno != 0)
|
|
{
|
|
logger_nDPIsrvd(remote, "Error closing collector connection", ": %s", strerror(errno));
|
|
}
|
|
nDPIsrvd_json_buffer_free(&remote->event_collector_un.main_read_buffer);
|
|
break;
|
|
case DISTRIBUTOR_UN:
|
|
if (errno != 0)
|
|
{
|
|
logger_nDPIsrvd(remote, "Error closing distributor connection", ": %s", strerror(errno));
|
|
}
|
|
if (remote->event_distributor_un.additional_write_buffers != NULL)
|
|
{
|
|
utarray_free(remote->event_distributor_un.additional_write_buffers);
|
|
}
|
|
nDPIsrvd_buffer_free(&remote->event_distributor_un.main_write_buffer.buf);
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
free(remote->event_distributor_un.user_name);
|
|
#endif
|
|
break;
|
|
case DISTRIBUTOR_IN:
|
|
if (errno != 0)
|
|
{
|
|
logger_nDPIsrvd(remote, "Error closing distributor connection", ": %s", strerror(errno));
|
|
}
|
|
if (remote->event_distributor_in.additional_write_buffers != NULL)
|
|
{
|
|
utarray_free(remote->event_distributor_in.additional_write_buffers);
|
|
}
|
|
nDPIsrvd_buffer_free(&remote->event_distributor_in.main_write_buffer.buf);
|
|
break;
|
|
}
|
|
|
|
memset(remote, 0, sizeof(*remote));
|
|
remote->fd = -1;
|
|
remotes.desc_used--;
|
|
}
|
|
}
|
|
|
|
static void free_remotes(struct nio * const io)
|
|
{
|
|
for (size_t i = 0; i < remotes.desc_size; ++i)
|
|
{
|
|
free_remote(io, &remotes.desc[i]);
|
|
}
|
|
nDPIsrvd_free(remotes.desc);
|
|
remotes.desc = NULL;
|
|
remotes.desc_used = 0;
|
|
remotes.desc_size = 0;
|
|
}
|
|
|
|
static int add_in_event_fd(struct nio * const io, int fd)
|
|
{
|
|
return nio_add_fd(io, fd, NIO_EVENT_INPUT, NULL) != NIO_SUCCESS;
|
|
}
|
|
|
|
static int add_in_event(struct nio * const io, struct remote_desc * const remote)
|
|
{
|
|
return nio_add_fd(io, remote->fd, NIO_EVENT_INPUT, remote) != NIO_SUCCESS;
|
|
}
|
|
|
|
static int set_out_event(struct nio * const io, struct remote_desc * const remote)
|
|
{
|
|
return nio_mod_fd(io, remote->fd, NIO_EVENT_OUTPUT, remote) != NIO_SUCCESS;
|
|
}
|
|
|
|
static int set_in_event(struct nio * const io, struct remote_desc * const remote)
|
|
{
|
|
return nio_mod_fd(io, remote->fd, NIO_EVENT_INPUT, remote) != NIO_SUCCESS;
|
|
}
|
|
|
|
static int del_event(struct nio * const io, int fd)
|
|
{
|
|
return nio_del_fd(io, fd) != NIO_SUCCESS;
|
|
}
|
|
|
|
static void disconnect_client(struct nio * const io, struct remote_desc * const remote)
|
|
{
|
|
free_remote(io, remote);
|
|
}
|
|
|
|
static int nDPIsrvd_parse_options(int argc, char ** argv)
|
|
{
|
|
int opt;
|
|
|
|
while ((opt = getopt(argc, argv, "f:lL:c:dp:s:S:G:m:u:g:C:Dvh")) != -1)
|
|
{
|
|
switch (opt)
|
|
{
|
|
case 'f':
|
|
set_cmdarg_string(&nDPIsrvd_options.config_file, optarg);
|
|
break;
|
|
case 'l':
|
|
enable_console_logger();
|
|
break;
|
|
case 'L':
|
|
if (enable_file_logger(optarg) != 0)
|
|
{
|
|
return 1;
|
|
}
|
|
break;
|
|
case 'c':
|
|
set_cmdarg_string(&nDPIsrvd_options.collector_un_sockpath, optarg);
|
|
break;
|
|
case 'e':
|
|
#ifdef ENABLE_EPOLL
|
|
set_cmdarg_boolean(&nDPIsrvd_options.use_poll, 1);
|
|
#else
|
|
logger_early(1, "%s", "nDPIsrvd was built w/o epoll() support, poll() is already the default");
|
|
#endif
|
|
break;
|
|
case 'd':
|
|
daemonize_enable();
|
|
break;
|
|
case 'p':
|
|
set_cmdarg_string(&nDPIsrvd_options.pidfile, optarg);
|
|
break;
|
|
case 's':
|
|
set_cmdarg_string(&nDPIsrvd_options.distributor_un_sockpath, optarg);
|
|
break;
|
|
case 'S':
|
|
set_cmdarg_string(&nDPIsrvd_options.distributor_in_address, optarg);
|
|
break;
|
|
case 'G':
|
|
{
|
|
char const * const sep = strchr(optarg, ':');
|
|
char group[256];
|
|
|
|
if (sep == NULL)
|
|
{
|
|
fprintf(stderr, "%s: Argument for `-G' is not in the format group:group\n", argv[0]);
|
|
return 1;
|
|
}
|
|
|
|
if (snprintf(group, sizeof(group), "%.*s", (int)(sep - optarg), optarg) > 0)
|
|
{
|
|
set_cmdarg_string(&nDPIsrvd_options.collector_group, group);
|
|
}
|
|
if (snprintf(group, sizeof(group), "%s", sep + 1) > 0)
|
|
{
|
|
set_cmdarg_string(&nDPIsrvd_options.distributor_group, group);
|
|
}
|
|
break;
|
|
}
|
|
case 'm':
|
|
{
|
|
nDPIsrvd_ull tmp;
|
|
|
|
if (str_value_to_ull(optarg, &tmp) != CONVERSION_OK)
|
|
{
|
|
fprintf(stderr, "%s: Argument for `-C' is not a number: %s\n", argv[0], optarg);
|
|
return 1;
|
|
}
|
|
set_cmdarg_ull(&nDPIsrvd_options.max_remote_descriptors, tmp);
|
|
break;
|
|
}
|
|
case 'u':
|
|
set_cmdarg_string(&nDPIsrvd_options.user, optarg);
|
|
break;
|
|
case 'g':
|
|
set_cmdarg_string(&nDPIsrvd_options.group, optarg);
|
|
break;
|
|
case 'C':
|
|
{
|
|
nDPIsrvd_ull tmp;
|
|
|
|
if (str_value_to_ull(optarg, &tmp) != CONVERSION_OK)
|
|
{
|
|
fprintf(stderr, "%s: Argument for `-C' is not a number: %s\n", argv[0], optarg);
|
|
return 1;
|
|
}
|
|
set_cmdarg_ull(&nDPIsrvd_options.max_write_buffers, tmp);
|
|
break;
|
|
}
|
|
case 'D':
|
|
set_cmdarg_boolean(&nDPIsrvd_options.bufferbloat_fallback_to_blocking, 0);
|
|
break;
|
|
case 'v':
|
|
fprintf(stderr, "%s", get_nDPId_version());
|
|
return 1;
|
|
case 'h':
|
|
default:
|
|
fprintf(stderr, "%s\n", get_nDPId_version());
|
|
fprintf(stderr,
|
|
"Usage: %s [-f config-file] [-l] [-L logfile]\n"
|
|
"\t[-c path-to-unix-sock] [-e] [-d] [-p pidfile]\n"
|
|
"\t[-s path-to-distributor-unix-socket] [-S distributor-host:port]\n"
|
|
"\t[-G collector-unix-socket-group:distributor-unix-socket-group]\n"
|
|
"\t[-m max-remote-descriptors] [-u user] [-g group]\n"
|
|
"\t[-C max-buffered-json-lines] [-D]\n"
|
|
"\t[-v] [-h]\n\n"
|
|
"\t-f\tLoad nDPIsrvd options from a configuration file.\n"
|
|
"\t-l\tLog all messages to stderr.\n"
|
|
"\t-L\tLog all messages to a log file.\n"
|
|
"\t-c\tPath to a listening UNIX socket (nDPIsrvd Collector).\n"
|
|
"\t \tDefault: %s\n"
|
|
"\t-e\tUse poll() instead of epoll().\n"
|
|
"\t \tDefault: epoll() on Linux, poll() otherwise\n"
|
|
"\t-d\tFork into background after initialization.\n"
|
|
"\t-p\tWrite the daemon PID to the given file path.\n"
|
|
"\t \tDefault: %s\n"
|
|
"\t-m\tMax accepted (Collector and Distributor) clients.\n"
|
|
"\t-u\tChange UID to the numeric value of user.\n"
|
|
"\t \tDefault: %s\n"
|
|
"\t-g\tChange GID to the numeric value of group.\n"
|
|
"\t-C\tMax buffered JSON lines before nDPIsrvd disconnects/blocking-IO a client.\n"
|
|
"\t-D\tDisconnect a slow client instead of falling back to blocking-IO.\n"
|
|
"\t-s\tPath to a listening UNIX socket (nDPIsrvd Distributor).\n"
|
|
"\t \tDefault: %s\n"
|
|
"\t-S\tAddress:Port of the listening TCP/IP socket (nDPIsrvd Distributor).\n"
|
|
"\t-G\tGroup owner of the UNIX collector/distributor socket.\n"
|
|
"\t \tDefault: Either the group set via `-g', otherwise the primary group of `-u'\n"
|
|
"\t-v\tversion\n"
|
|
"\t-h\tthis\n\n",
|
|
argv[0],
|
|
nDPIsrvd_options.collector_un_sockpath.string.default_value,
|
|
nDPIsrvd_options.pidfile.string.default_value,
|
|
nDPIsrvd_options.user.string.default_value,
|
|
nDPIsrvd_options.distributor_un_sockpath.string.default_value);
|
|
return 1;
|
|
}
|
|
}
|
|
|
|
set_config_defaults(&config_map[0], nDPIsrvd_ARRAY_LENGTH(config_map));
|
|
|
|
if (is_path_absolute("Pidfile", GET_CMDARG_STR(nDPIsrvd_options.pidfile)) != 0)
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
if (is_path_absolute("Collector UNIX socket", GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath)) != 0)
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
if (is_path_absolute("Distributor UNIX socket", GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath)) != 0)
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
if (IS_CMDARG_SET(nDPIsrvd_options.distributor_in_address) != 0)
|
|
{
|
|
if (nDPIsrvd_setup_address(&distributor_in_address, GET_CMDARG_STR(nDPIsrvd_options.distributor_in_address)) !=
|
|
0)
|
|
{
|
|
logger_early(1,
|
|
"%s: Could not parse address %s",
|
|
argv[0],
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_in_address));
|
|
return 1;
|
|
}
|
|
if (distributor_in_address.raw.sa_family == AF_UNIX)
|
|
{
|
|
logger_early(1,
|
|
"%s: You've requested to setup another UNIX socket `%s', but there is already one at `%s'",
|
|
argv[0],
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_in_address),
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath));
|
|
return 1;
|
|
}
|
|
}
|
|
|
|
if (optind < argc)
|
|
{
|
|
logger_early(1, "%s: Unexpected argument after options", argv[0]);
|
|
return 1;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static struct remote_desc * accept_remote(int server_fd,
|
|
enum sock_type socktype,
|
|
struct sockaddr * const sockaddr,
|
|
socklen_t * const addrlen)
|
|
{
|
|
int client_fd;
|
|
|
|
while ((client_fd = accept(server_fd, sockaddr, addrlen)) < 0 && errno == EINTR) {}
|
|
if (client_fd < 0 || set_fd_cloexec(client_fd) < 0)
|
|
{
|
|
logger(1, "Accept failed: %s", strerror(errno));
|
|
return NULL;
|
|
}
|
|
|
|
struct remote_desc * current = get_remote_descriptor(socktype, client_fd, NETWORK_BUFFER_MAX_SIZE);
|
|
if (current == NULL)
|
|
{
|
|
return NULL;
|
|
}
|
|
|
|
return current;
|
|
}
|
|
|
|
static int new_connection(struct nio * const io, int eventfd)
|
|
{
|
|
union
|
|
{
|
|
struct sockaddr_un saddr_collector_un;
|
|
struct sockaddr_un saddr_distributor_un;
|
|
struct sockaddr_in saddr_distributor_in;
|
|
} sockaddr;
|
|
|
|
socklen_t peer_addr_len;
|
|
enum sock_type stype;
|
|
int server_fd;
|
|
if (eventfd == collector_un_sockfd)
|
|
{
|
|
peer_addr_len = sizeof(sockaddr.saddr_collector_un);
|
|
stype = COLLECTOR_UN;
|
|
server_fd = collector_un_sockfd;
|
|
}
|
|
else if (eventfd == distributor_un_sockfd)
|
|
{
|
|
peer_addr_len = sizeof(sockaddr.saddr_distributor_un);
|
|
stype = DISTRIBUTOR_UN;
|
|
server_fd = distributor_un_sockfd;
|
|
}
|
|
else if (eventfd == distributor_in_sockfd)
|
|
{
|
|
peer_addr_len = sizeof(sockaddr.saddr_distributor_in);
|
|
stype = DISTRIBUTOR_IN;
|
|
server_fd = distributor_in_sockfd;
|
|
}
|
|
else
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
struct remote_desc * const current = accept_remote(server_fd, stype, (struct sockaddr *)&sockaddr, &peer_addr_len);
|
|
if (current == NULL)
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
int sockopt;
|
|
switch (current->sock_type)
|
|
{
|
|
case COLLECTOR_UN:
|
|
current->event_collector_un.peer = sockaddr.saddr_collector_un;
|
|
current->event_collector_un.json_bytes = 0;
|
|
|
|
sockopt = NETWORK_BUFFER_MAX_SIZE;
|
|
if (setsockopt(current->fd, SOL_SOCKET, SO_RCVBUF, &sockopt, sizeof(sockopt)) < 0)
|
|
{
|
|
logger(1, "Error setting socket option SO_RCVBUF: %s", strerror(errno));
|
|
return 1;
|
|
}
|
|
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
struct ucred ucred = {};
|
|
socklen_t ucred_len = sizeof(ucred);
|
|
if (getsockopt(current->fd, SOL_SOCKET, SO_PEERCRED, &ucred, &ucred_len) == -1)
|
|
{
|
|
logger(1, "Error getting credentials from UNIX socket: %s", strerror(errno));
|
|
return 1;
|
|
}
|
|
current->event_collector_un.pid = ucred.pid;
|
|
#endif
|
|
|
|
logger_nDPIsrvd(current, "New collector connection from", "");
|
|
break;
|
|
case DISTRIBUTOR_UN:
|
|
case DISTRIBUTOR_IN:
|
|
if (current->sock_type == DISTRIBUTOR_UN)
|
|
{
|
|
current->event_distributor_un.peer = sockaddr.saddr_distributor_un;
|
|
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
struct ucred ucred = {};
|
|
socklen_t ucred_len = sizeof(ucred);
|
|
if (getsockopt(current->fd, SOL_SOCKET, SO_PEERCRED, &ucred, &ucred_len) == -1)
|
|
{
|
|
logger(1, "Error getting credentials from UNIX socket: %s", strerror(errno));
|
|
return 1;
|
|
}
|
|
|
|
struct passwd pwnam = {};
|
|
struct passwd * pwres = NULL;
|
|
ssize_t pwsiz = sysconf(_SC_GETPW_R_SIZE_MAX);
|
|
if (pwsiz == -1)
|
|
{
|
|
pwsiz = BUFSIZ;
|
|
}
|
|
char buf[pwsiz];
|
|
if (getpwuid_r(ucred.uid, &pwnam, &buf[0], pwsiz, &pwres) != 0 || pwres == NULL)
|
|
{
|
|
logger(1, "Could not get passwd entry for user id %u", ucred.uid);
|
|
return 1;
|
|
}
|
|
|
|
current->event_distributor_un.pid = ucred.pid;
|
|
current->event_distributor_un.user_name = strdup(pwres->pw_name);
|
|
#endif
|
|
}
|
|
else
|
|
{
|
|
current->event_distributor_in.peer = sockaddr.saddr_distributor_in;
|
|
|
|
sockopt = 1;
|
|
if (setsockopt(current->fd, SOL_SOCKET, SO_RCVBUF, &sockopt, sizeof(sockopt)) < 0)
|
|
{
|
|
logger(1, "Error setting socket option SO_RCVBUF: %s", strerror(errno));
|
|
return 1;
|
|
}
|
|
|
|
if (inet_ntop(current->event_distributor_in.peer.sin_family,
|
|
¤t->event_distributor_in.peer.sin_addr,
|
|
¤t->event_distributor_in.peer_addr[0],
|
|
sizeof(current->event_distributor_in.peer_addr)) == NULL)
|
|
{
|
|
logger(1, "Error converting an internet address: %s", strerror(errno));
|
|
return 1;
|
|
}
|
|
}
|
|
|
|
sockopt = NETWORK_BUFFER_MAX_SIZE;
|
|
if (setsockopt(current->fd, SOL_SOCKET, SO_SNDBUF, &sockopt, sizeof(sockopt)) < 0)
|
|
{
|
|
logger(1, "Error setting socket option SO_SNDBUF: %s", strerror(errno));
|
|
return 1;
|
|
}
|
|
|
|
{
|
|
struct timeval send_timeout = {1, 0};
|
|
if (setsockopt(current->fd, SOL_SOCKET, SO_SNDTIMEO, (char *)&send_timeout, sizeof(send_timeout)) != 0)
|
|
{
|
|
logger(1, "Error setting socket option send timeout: %s", strerror(errno));
|
|
}
|
|
}
|
|
|
|
logger_nDPIsrvd(current, "New distributor connection from", "");
|
|
break;
|
|
}
|
|
|
|
/* nonblocking fd is mandatory */
|
|
if (fcntl_add_flags(current->fd, O_NONBLOCK) != 0)
|
|
{
|
|
logger(1, "Error setting fd flags to non-blocking mode: %s", strerror(errno));
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
|
|
/* shutdown writing end for collector clients */
|
|
if (current->sock_type == COLLECTOR_UN)
|
|
{
|
|
shutdown(current->fd, SHUT_WR); // collector
|
|
/* shutdown reading end for distributor clients does not work due to epoll usage */
|
|
}
|
|
|
|
/* setup event I/O */
|
|
errno = 0;
|
|
if (add_in_event(io, current) != NIO_SUCCESS)
|
|
{
|
|
logger(1, "Error adding input event to %d: %s", current->fd, (errno != 0 ? strerror(errno) : "Internal Error"));
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int handle_collector_protocol(struct nio * const io, struct remote_desc * const current)
|
|
{
|
|
struct nDPIsrvd_json_buffer * const json_read_buffer = get_read_buffer(current);
|
|
char * json_msg_start = NULL;
|
|
|
|
if (json_read_buffer == NULL)
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
if (json_read_buffer->buf.ptr.text[NETWORK_BUFFER_LENGTH_DIGITS] != '{')
|
|
{
|
|
logger_nDPIsrvd(current,
|
|
"BUG: Collector connection",
|
|
"JSON invalid opening character: '%c'",
|
|
json_read_buffer->buf.ptr.text[NETWORK_BUFFER_LENGTH_DIGITS]);
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
|
|
errno = 0;
|
|
current->event_collector_un.json_bytes = strtoull(json_read_buffer->buf.ptr.text, &json_msg_start, 10);
|
|
current->event_collector_un.json_bytes += json_msg_start - json_read_buffer->buf.ptr.text;
|
|
|
|
if (errno == ERANGE)
|
|
{
|
|
logger_nDPIsrvd(current, "BUG: Collector connection", "JSON message length exceeds numceric limits");
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
|
|
if (json_msg_start == json_read_buffer->buf.ptr.text)
|
|
{
|
|
logger_nDPIsrvd(current,
|
|
"BUG: Collector connection",
|
|
"missing JSON message length in protocol preamble: \"%.*s\"",
|
|
NETWORK_BUFFER_LENGTH_DIGITS,
|
|
json_read_buffer->buf.ptr.text);
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
|
|
if (json_msg_start - json_read_buffer->buf.ptr.text != NETWORK_BUFFER_LENGTH_DIGITS)
|
|
{
|
|
logger_nDPIsrvd(current,
|
|
"BUG: Collector connection",
|
|
"invalid collector protocol data received. Expected protocol preamble of size %u bytes, got "
|
|
"%ld "
|
|
"bytes",
|
|
NETWORK_BUFFER_LENGTH_DIGITS,
|
|
(long int)(json_msg_start - json_read_buffer->buf.ptr.text));
|
|
}
|
|
|
|
if (current->event_collector_un.json_bytes > json_read_buffer->buf.max)
|
|
{
|
|
logger_nDPIsrvd(current,
|
|
"BUG: Collector connection",
|
|
"JSON message too big: %llu > %zu",
|
|
current->event_collector_un.json_bytes,
|
|
json_read_buffer->buf.max);
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
|
|
if (current->event_collector_un.json_bytes > json_read_buffer->buf.used)
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
if (json_read_buffer->buf.ptr.text[current->event_collector_un.json_bytes - 2] != '}' ||
|
|
json_read_buffer->buf.ptr.text[current->event_collector_un.json_bytes - 1] != '\n')
|
|
{
|
|
logger_nDPIsrvd(current,
|
|
"BUG: Collector connection",
|
|
"invalid JSON message: %.*s...",
|
|
(int)current->event_collector_un.json_bytes > 512 ? 512
|
|
: (int)current->event_collector_un.json_bytes,
|
|
json_read_buffer->buf.ptr.text);
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int handle_incoming_data(struct nio * const io, struct remote_desc * const current)
|
|
{
|
|
struct nDPIsrvd_json_buffer * const json_read_buffer = get_read_buffer(current);
|
|
|
|
if (json_read_buffer == NULL)
|
|
{
|
|
unsigned char garbage = 0;
|
|
|
|
if (read(current->fd, &garbage, sizeof(garbage)) == sizeof(garbage))
|
|
{
|
|
logger_nDPIsrvd(current, "Received data from", "who is not allowed to send us some.");
|
|
}
|
|
else
|
|
{
|
|
logger_nDPIsrvd(current, "Distributor connection", "closed");
|
|
}
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
|
|
/* read JSON messages (or parts) from the UNIX socket (collecting) */
|
|
if (json_read_buffer->buf.used == json_read_buffer->buf.max)
|
|
{
|
|
logger_nDPIsrvd(current,
|
|
"Collector connection",
|
|
"read buffer (%zu bytes) full. No more read possible.",
|
|
json_read_buffer->buf.max);
|
|
}
|
|
else
|
|
{
|
|
errno = 0;
|
|
ssize_t bytes_read;
|
|
|
|
while ((bytes_read = read(current->fd,
|
|
json_read_buffer->buf.ptr.raw + json_read_buffer->buf.used,
|
|
json_read_buffer->buf.max - json_read_buffer->buf.used)) < 0 &&
|
|
errno == EINTR)
|
|
{
|
|
// Retry if interrupted by a signal.
|
|
}
|
|
if (bytes_read < 0 || errno != 0)
|
|
{
|
|
logger_nDPIsrvd(current, "Could not read remote", ": %s", strerror(errno));
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
if (bytes_read == 0)
|
|
{
|
|
logger_nDPIsrvd(current, "Collector connection", "closed during read");
|
|
disconnect_client(io, current);
|
|
return 1;
|
|
}
|
|
json_read_buffer->buf.used += bytes_read;
|
|
}
|
|
|
|
while (json_read_buffer->buf.used >= NETWORK_BUFFER_LENGTH_DIGITS + 1)
|
|
{
|
|
if (handle_collector_protocol(io, current) != 0)
|
|
{
|
|
break;
|
|
}
|
|
|
|
for (size_t i = 0; i < remotes.desc_size; ++i)
|
|
{
|
|
struct nDPIsrvd_write_buffer * const write_buffer = get_write_buffer(&remotes.desc[i]);
|
|
UT_array * const additional_write_buffers = get_additional_write_buffers(&remotes.desc[i]);
|
|
|
|
if (remotes.desc[i].fd < 0 || write_buffer == NULL || additional_write_buffers == NULL)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
if (current->event_collector_un.json_bytes > write_buffer->buf.max - write_buffer->buf.used ||
|
|
utarray_len(additional_write_buffers) > 0)
|
|
{
|
|
if (utarray_len(additional_write_buffers) == 0)
|
|
{
|
|
errno = 0;
|
|
if (set_out_event(io, &remotes.desc[i]) != 0)
|
|
{
|
|
logger_nDPIsrvd(&remotes.desc[i],
|
|
"Could not add event to",
|
|
", disconnecting: %s",
|
|
(errno != 0 ? strerror(errno) : "Internal Error"));
|
|
disconnect_client(io, &remotes.desc[i]);
|
|
continue;
|
|
}
|
|
}
|
|
if (add_to_additional_write_buffers(&remotes.desc[i],
|
|
json_read_buffer->buf.ptr.raw,
|
|
current->event_collector_un.json_bytes) != 0)
|
|
{
|
|
disconnect_client(io, &remotes.desc[i]);
|
|
continue;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
memcpy(write_buffer->buf.ptr.raw + write_buffer->buf.used,
|
|
json_read_buffer->buf.ptr.raw,
|
|
current->event_collector_un.json_bytes);
|
|
write_buffer->buf.used += current->event_collector_un.json_bytes;
|
|
}
|
|
|
|
if (drain_main_buffer(&remotes.desc[i]) != 0)
|
|
{
|
|
disconnect_client(io, &remotes.desc[i]);
|
|
}
|
|
}
|
|
|
|
memmove(json_read_buffer->buf.ptr.raw,
|
|
json_read_buffer->buf.ptr.raw + current->event_collector_un.json_bytes,
|
|
json_read_buffer->buf.used - current->event_collector_un.json_bytes);
|
|
json_read_buffer->buf.used -= current->event_collector_un.json_bytes;
|
|
current->event_collector_un.json_bytes = 0;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int handle_data_event(struct nio * const io, int index)
|
|
{
|
|
struct remote_desc * const current = (struct remote_desc *)nio_get_ptr(io, index);
|
|
|
|
if (nio_has_input(io, index) != NIO_SUCCESS && nio_can_output(io, index) != NIO_SUCCESS)
|
|
{
|
|
logger(1, "%s", "Neither input nor output event set.");
|
|
return 1;
|
|
}
|
|
|
|
if (current == NULL)
|
|
{
|
|
logger(1, "%s", "Remote descriptor got from event data invalid.");
|
|
return 1;
|
|
}
|
|
|
|
if (current->fd < 0)
|
|
{
|
|
logger(1, "File descriptor `%d' got from event data invalid.", current->fd);
|
|
return 1;
|
|
}
|
|
|
|
if (nio_has_input(io, index) == NIO_SUCCESS)
|
|
{
|
|
return handle_incoming_data(io, current);
|
|
}
|
|
else
|
|
{
|
|
return handle_outgoing_data(io, current);
|
|
}
|
|
}
|
|
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
static int setup_signalfd(struct nio * const io)
|
|
{
|
|
sigset_t mask;
|
|
int sfd;
|
|
|
|
sigemptyset(&mask);
|
|
sigaddset(&mask, SIGINT);
|
|
sigaddset(&mask, SIGTERM);
|
|
sigaddset(&mask, SIGQUIT);
|
|
|
|
if (sigprocmask(SIG_BLOCK, &mask, NULL) == -1)
|
|
{
|
|
return -1;
|
|
}
|
|
sfd = signalfd(-1, &mask, 0);
|
|
if (sfd == -1)
|
|
{
|
|
return -1;
|
|
}
|
|
|
|
if (add_in_event_fd(io, sfd) != 0)
|
|
{
|
|
return -1;
|
|
}
|
|
|
|
if (fcntl_add_flags(sfd, O_NONBLOCK) != 0)
|
|
{
|
|
return -1;
|
|
}
|
|
|
|
return sfd;
|
|
}
|
|
#endif
|
|
|
|
static int mainloop(struct nio * const io)
|
|
{
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
int signalfd = setup_signalfd(io);
|
|
#endif
|
|
|
|
while (nDPIsrvd_main_thread_shutdown == 0)
|
|
{
|
|
if (nio_run(io, 1000) != NIO_SUCCESS)
|
|
{
|
|
logger(1, "Event I/O returned error: %s", strerror(errno));
|
|
}
|
|
|
|
int nready = nio_get_nready(io);
|
|
|
|
for (int i = 0; i < nready; i++)
|
|
{
|
|
int fd = nio_get_fd(io, i);
|
|
|
|
if (nio_has_error(io, i) == NIO_SUCCESS)
|
|
{
|
|
if (fd != collector_un_sockfd && fd != distributor_un_sockfd && fd != distributor_in_sockfd)
|
|
{
|
|
struct remote_desc * const current = (struct remote_desc *)nio_get_ptr(io, i);
|
|
switch (current->sock_type)
|
|
{
|
|
case COLLECTOR_UN:
|
|
logger_nDPIsrvd(current, "Collector connection", "closed");
|
|
break;
|
|
case DISTRIBUTOR_UN:
|
|
case DISTRIBUTOR_IN:
|
|
logger_nDPIsrvd(current, "Distributor connection", "closed");
|
|
break;
|
|
}
|
|
disconnect_client(io, current);
|
|
}
|
|
else
|
|
{
|
|
logger(1, "Event I/O error: %s", (errno != 0 ? strerror(errno) : "unknown"));
|
|
}
|
|
break;
|
|
}
|
|
|
|
if (fd == collector_un_sockfd || fd == distributor_un_sockfd || fd == distributor_in_sockfd)
|
|
{
|
|
/* New connection to collector / distributor. */
|
|
if (new_connection(io, fd) != 0)
|
|
{
|
|
continue;
|
|
}
|
|
}
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
else if (fd == signalfd)
|
|
{
|
|
struct signalfd_siginfo fdsi;
|
|
ssize_t s;
|
|
|
|
s = read(signalfd, &fdsi, sizeof(struct signalfd_siginfo));
|
|
if (s != sizeof(struct signalfd_siginfo))
|
|
{
|
|
if (s < 0)
|
|
{
|
|
logger(1, "Read from signal fd returned: %s", strerror(errno));
|
|
nDPIsrvd_main_thread_shutdown = 1;
|
|
}
|
|
else
|
|
{
|
|
logger(1,
|
|
"Invalid signal fd read size. Got %zd, wanted %zu bytes.",
|
|
s,
|
|
sizeof(struct signalfd_siginfo));
|
|
}
|
|
continue;
|
|
}
|
|
|
|
if (fdsi.ssi_signo == SIGINT || fdsi.ssi_signo == SIGTERM || fdsi.ssi_signo == SIGQUIT)
|
|
{
|
|
nDPIsrvd_main_thread_shutdown = 1;
|
|
continue;
|
|
}
|
|
}
|
|
#endif
|
|
else
|
|
{
|
|
/* Incoming data / Outoing data ready to receive / send. */
|
|
if (handle_data_event(io, i) != 0)
|
|
{
|
|
/* do nothing */
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
free_remotes(io);
|
|
nio_free(io);
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
close(signalfd);
|
|
#endif
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int setup_event_queue(struct nio * const io)
|
|
{
|
|
#ifdef ENABLE_EPOLL
|
|
if ((GET_CMDARG_BOOL(nDPIsrvd_options.use_poll) == 0 && nio_use_epoll(io, 32) != NIO_SUCCESS) ||
|
|
(GET_CMDARG_BOOL(nDPIsrvd_options.use_poll) != 0 &&
|
|
nio_use_poll(io, nDPIsrvd_MAX_REMOTE_DESCRIPTORS) != NIO_SUCCESS))
|
|
#else
|
|
if (nio_use_poll(io, nDPIsrvd_MAX_REMOTE_DESCRIPTORS) != NIO_SUCCESS)
|
|
#endif
|
|
{
|
|
logger(1, "%s", "Event I/O poll/epoll setup failed");
|
|
return -1;
|
|
}
|
|
|
|
errno = 0;
|
|
if (add_in_event_fd(io, collector_un_sockfd) != 0)
|
|
{
|
|
logger(1,
|
|
"Error adding collector UNIX socket fd to event I/O: %s",
|
|
(errno != 0 ? strerror(errno) : "Internal Error"));
|
|
return -1;
|
|
}
|
|
|
|
errno = 0;
|
|
if (add_in_event_fd(io, distributor_un_sockfd) != 0)
|
|
{
|
|
logger(1,
|
|
"Error adding distributor UNIX socket fd to event I/O: %s",
|
|
(errno != 0 ? strerror(errno) : "Internal Error"));
|
|
return -1;
|
|
}
|
|
|
|
if (distributor_in_sockfd >= 0)
|
|
{
|
|
errno = 0;
|
|
if (add_in_event_fd(io, distributor_in_sockfd) != 0)
|
|
{
|
|
logger(1,
|
|
"Error adding distributor TCP/IP socket fd to event I/O: %s",
|
|
(errno != 0 ? strerror(errno) : "Internal Error"));
|
|
return -1;
|
|
}
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int setup_remote_descriptors(nDPIsrvd_ull max_remote_descriptors)
|
|
{
|
|
remotes.desc_used = 0;
|
|
remotes.desc_size = max_remote_descriptors;
|
|
remotes.desc = (struct remote_desc *)nDPIsrvd_calloc(remotes.desc_size, sizeof(*remotes.desc));
|
|
if (remotes.desc == NULL)
|
|
{
|
|
return -1;
|
|
}
|
|
for (size_t i = 0; i < remotes.desc_size; ++i)
|
|
{
|
|
remotes.desc[i].fd = -1;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
static int nDPIsrvd_parsed_config_line(
|
|
int lineno, char const * const section, char const * const name, char const * const value, void * const user_data)
|
|
{
|
|
(void)user_data;
|
|
|
|
if (strnlen(section, INI_MAX_SECTION) == nDPIsrvd_STRLEN_SZ("general") &&
|
|
strncmp(section, "general", INI_MAX_SECTION) == 0)
|
|
{
|
|
size_t i;
|
|
|
|
for (i = 0; i < nDPIsrvd_ARRAY_LENGTH(config_map); ++i)
|
|
{
|
|
if (strnlen(name, INI_MAX_NAME) == strnlen(config_map[i].key, INI_MAX_NAME) &&
|
|
strncmp(name, config_map[i].key, INI_MAX_NAME) == 0)
|
|
{
|
|
if (IS_CMDARG_SET(*config_map[i].opt) != 0)
|
|
{
|
|
logger_early(1, "General config key `%s' already set, ignoring value `%s'", name, value);
|
|
}
|
|
else
|
|
{
|
|
if (set_config_from(&config_map[i], value) != 0)
|
|
{
|
|
return 0;
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
if (i == nDPIsrvd_ARRAY_LENGTH(config_map))
|
|
{
|
|
logger_early(1, "Invalid general config key `%s' at line %d", name, lineno);
|
|
}
|
|
}
|
|
else
|
|
{
|
|
logger_early(
|
|
1, "Invalid config section `%s' at line %d with key `%s' and value `%s'", section, lineno, name, value);
|
|
}
|
|
|
|
return 1;
|
|
}
|
|
|
|
#ifndef NO_MAIN
|
|
int main(int argc, char ** argv)
|
|
{
|
|
int retval = 1;
|
|
struct nio io;
|
|
|
|
if (argc == 0)
|
|
{
|
|
return 1;
|
|
}
|
|
|
|
nio_init(&io);
|
|
init_logging("nDPIsrvd");
|
|
|
|
if (nDPIsrvd_parse_options(argc, argv) != 0)
|
|
{
|
|
return 1;
|
|
}
|
|
{
|
|
int ret;
|
|
|
|
if (IS_CMDARG_SET(nDPIsrvd_options.config_file) != 0 &&
|
|
(ret =
|
|
parse_config_file(GET_CMDARG_STR(nDPIsrvd_options.config_file), nDPIsrvd_parsed_config_line, NULL)) !=
|
|
0)
|
|
{
|
|
if (ret > 0)
|
|
{
|
|
logger_early(1, "Config file `%s' is malformed", GET_CMDARG_STR(nDPIsrvd_options.config_file));
|
|
}
|
|
else if (ret == -ENOENT)
|
|
{
|
|
logger_early(1, "Path `%s' is not a regular file", GET_CMDARG_STR(nDPIsrvd_options.config_file));
|
|
}
|
|
else
|
|
{
|
|
logger_early(1,
|
|
"Could not open file `%s' for reading: %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.config_file),
|
|
strerror(errno));
|
|
}
|
|
return 1;
|
|
}
|
|
}
|
|
|
|
if (is_daemonize_enabled() != 0 && is_console_logger_enabled() != 0)
|
|
{
|
|
logger_early(1,
|
|
"%s",
|
|
"Daemon mode `-d' and `-l' can not be used together, "
|
|
"because stdout/stderr is beeing redirected to /dev/null");
|
|
return 1;
|
|
}
|
|
|
|
if (access(GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath), F_OK) == 0)
|
|
{
|
|
logger_early(1,
|
|
"UNIX socket `%s' exists; nDPIsrvd already running? "
|
|
"Please remove the socket manually or change socket path.",
|
|
GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath));
|
|
return 1;
|
|
}
|
|
|
|
if (access(GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath), F_OK) == 0)
|
|
{
|
|
logger_early(1,
|
|
"UNIX socket `%s' exists; nDPIsrvd already running? "
|
|
"Please remove the socket manually or change socket path.",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath));
|
|
return 1;
|
|
}
|
|
|
|
log_app_info();
|
|
|
|
if (daemonize_with_pidfile(GET_CMDARG_STR(nDPIsrvd_options.pidfile)) != 0)
|
|
{
|
|
goto error;
|
|
}
|
|
|
|
if (setup_remote_descriptors(GET_CMDARG_ULL(nDPIsrvd_options.max_remote_descriptors)) != 0)
|
|
{
|
|
goto error;
|
|
}
|
|
|
|
switch (create_listen_sockets())
|
|
{
|
|
case 0:
|
|
break;
|
|
case 1:
|
|
goto error;
|
|
case 2:
|
|
if (unlink(GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath)) != 0)
|
|
{
|
|
logger(1,
|
|
"Could not unlink `%s': %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath),
|
|
strerror(errno));
|
|
}
|
|
goto error;
|
|
case 3:
|
|
goto error_unlink_sockets;
|
|
default:
|
|
goto error;
|
|
}
|
|
|
|
logger(0, "collector UNIX socket listen on `%s'", GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath));
|
|
logger(0, "distributor UNIX listen on `%s'", GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath));
|
|
switch (distributor_in_address.raw.sa_family)
|
|
{
|
|
default:
|
|
goto error_unlink_sockets;
|
|
case AF_INET:
|
|
case AF_INET6:
|
|
logger(1,
|
|
"Please keep in mind that using a TCP Socket may leak sensitive information to "
|
|
"everyone with access to the device/network. You've been warned!");
|
|
break;
|
|
case AF_UNIX:
|
|
case (sa_family_t)0xFFFF:
|
|
break;
|
|
}
|
|
|
|
int ret = chmod_chown(GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath),
|
|
S_IRUSR | S_IWUSR | S_IWGRP,
|
|
GET_CMDARG_STR(nDPIsrvd_options.user),
|
|
IS_CMDARG_SET(nDPIsrvd_options.collector_group) != 0
|
|
? GET_CMDARG_STR(nDPIsrvd_options.collector_group)
|
|
: GET_CMDARG_STR(nDPIsrvd_options.group));
|
|
if (ret != 0)
|
|
{
|
|
if (IS_CMDARG_SET(nDPIsrvd_options.collector_group) != 0 || IS_CMDARG_SET(nDPIsrvd_options.group) != 0)
|
|
{
|
|
logger(1,
|
|
"Could not chmod/chown `%s' to user `%s' and group `%s': %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath),
|
|
GET_CMDARG_STR(nDPIsrvd_options.user),
|
|
IS_CMDARG_SET(nDPIsrvd_options.collector_group) != 0
|
|
? GET_CMDARG_STR(nDPIsrvd_options.collector_group)
|
|
: GET_CMDARG_STR(nDPIsrvd_options.group),
|
|
strerror(ret));
|
|
}
|
|
else
|
|
{
|
|
logger(1,
|
|
"Could not chmod/chown `%s' to user `%s': %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath),
|
|
GET_CMDARG_STR(nDPIsrvd_options.user),
|
|
strerror(ret));
|
|
}
|
|
if (ret != EPERM)
|
|
{
|
|
goto error_unlink_sockets;
|
|
}
|
|
}
|
|
|
|
ret = chmod_chown(GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath),
|
|
S_IRUSR | S_IWUSR | S_IWGRP,
|
|
GET_CMDARG_STR(nDPIsrvd_options.user),
|
|
IS_CMDARG_SET(nDPIsrvd_options.distributor_group) != 0
|
|
? GET_CMDARG_STR(nDPIsrvd_options.distributor_group)
|
|
: GET_CMDARG_STR(nDPIsrvd_options.group));
|
|
if (ret != 0)
|
|
{
|
|
if (IS_CMDARG_SET(nDPIsrvd_options.distributor_group) != 0 || IS_CMDARG_SET(nDPIsrvd_options.group) != 0)
|
|
{
|
|
logger(1,
|
|
"Could not chmod/chown `%s' to user `%s' and group `%s': %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath),
|
|
GET_CMDARG_STR(nDPIsrvd_options.user),
|
|
IS_CMDARG_SET(nDPIsrvd_options.distributor_group) != 0
|
|
? GET_CMDARG_STR(nDPIsrvd_options.distributor_group)
|
|
: GET_CMDARG_STR(nDPIsrvd_options.group),
|
|
strerror(ret));
|
|
}
|
|
else
|
|
{
|
|
logger(1,
|
|
"Could not chmod/chown `%s' to user `%s': %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath),
|
|
GET_CMDARG_STR(nDPIsrvd_options.user),
|
|
strerror(ret));
|
|
}
|
|
if (ret != EPERM)
|
|
{
|
|
goto error_unlink_sockets;
|
|
}
|
|
}
|
|
|
|
ret = change_user_group(GET_CMDARG_STR(nDPIsrvd_options.user),
|
|
GET_CMDARG_STR(nDPIsrvd_options.group),
|
|
GET_CMDARG_STR(nDPIsrvd_options.pidfile));
|
|
if (ret != 0 && ret != -EPERM)
|
|
{
|
|
if (GET_CMDARG_STR(nDPIsrvd_options.group) != NULL)
|
|
{
|
|
logger(1,
|
|
"Change user/group to %s/%s failed: %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.user),
|
|
GET_CMDARG_STR(nDPIsrvd_options.group),
|
|
strerror(-ret));
|
|
}
|
|
else
|
|
{
|
|
logger(1, "Change user to %s failed: %s", GET_CMDARG_STR(nDPIsrvd_options.user), strerror(-ret));
|
|
}
|
|
goto error_unlink_sockets;
|
|
}
|
|
|
|
signal(SIGPIPE, SIG_IGN);
|
|
#if !defined(__FreeBSD__) && !defined(__APPLE__)
|
|
signal(SIGINT, SIG_IGN);
|
|
signal(SIGTERM, SIG_IGN);
|
|
signal(SIGQUIT, SIG_IGN);
|
|
#endif
|
|
|
|
if (setup_event_queue(&io) != 0)
|
|
{
|
|
goto error_unlink_sockets;
|
|
}
|
|
|
|
retval = mainloop(&io);
|
|
|
|
error_unlink_sockets:
|
|
if (unlink(GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath)) != 0)
|
|
{
|
|
logger(1, "Could not unlink `%s': %s", GET_CMDARG_STR(nDPIsrvd_options.collector_un_sockpath), strerror(errno));
|
|
}
|
|
if (unlink(GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath)) != 0)
|
|
{
|
|
logger(1,
|
|
"Could not unlink `%s': %s",
|
|
GET_CMDARG_STR(nDPIsrvd_options.distributor_un_sockpath),
|
|
strerror(errno));
|
|
}
|
|
error:
|
|
close(collector_un_sockfd);
|
|
close(distributor_un_sockfd);
|
|
close(distributor_in_sockfd);
|
|
|
|
daemonize_shutdown(GET_CMDARG_STR(nDPIsrvd_options.pidfile));
|
|
logger(0, "Bye.");
|
|
shutdown_logging();
|
|
|
|
return retval;
|
|
}
|
|
#endif
|