Backport: relayd: track relayd control connection sockets
[lttng-tools.git] / src / bin / lttng-relayd / main.c
index 7f0b413bcbb789b83422255aad252e71c3b8e0d5..123731e171c662cef4e69f4bacedbd42681be508 100644 (file)
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <sys/wait.h>
+#include <sys/resource.h>
 #include <inttypes.h>
 #include <urcu/futex.h>
 #include <urcu/uatomic.h>
 #include <unistd.h>
 #include <fcntl.h>
+#include <ctype.h>
 
 #include <lttng/lttng.h>
 #include <common/common.h>
 #include <common/config/session-config.h>
 #include <common/dynamic-buffer.h>
 #include <common/buffer-view.h>
+#include <common/fd-tracker/utils.h>
 #include <urcu/rculist.h>
 
+#include "version.h"
 #include "cmd.h"
 #include "ctf-trace.h"
 #include "index.h"
@@ -84,7 +88,9 @@ enum relay_connection_status {
 
 /* command line options */
 char *opt_output_path, *opt_working_directory;
-static int opt_daemon, opt_background;
+static int opt_daemon, opt_background, opt_print_version;
+int opt_group_output_by_session;
+int opt_group_output_by_host;
 
 /*
  * We need to wait for listener and live listener threads, as well as
@@ -145,6 +151,9 @@ static uint64_t last_relay_stream_id;
  */
 static struct relay_conn_queue relay_conn_queue;
 
+/* Cap of file desriptors to be in simultaneous use by the relay daemon. */
+static unsigned int lttng_opt_fd_cap;
+
 /* Global relay stream hash table. */
 struct lttng_ht *relay_streams_ht;
 
@@ -157,6 +166,9 @@ struct lttng_ht *sessions_ht;
 /* Relayd health monitoring */
 struct health_app *health_relayd;
 
+/* Global fd tracker. */
+struct fd_tracker *the_fd_tracker;
+
 static struct option long_options[] = {
        { "control-port", 1, 0, 'C', },
        { "data-port", 1, 0, 'D', },
@@ -164,17 +176,37 @@ static struct option long_options[] = {
        { "daemonize", 0, 0, 'd', },
        { "background", 0, 0, 'b', },
        { "group", 1, 0, 'g', },
+       { "fd-cap", 1, 0, '\0', },
        { "help", 0, 0, 'h', },
        { "output", 1, 0, 'o', },
        { "verbose", 0, 0, 'v', },
        { "config", 1, 0, 'f' },
        { "version", 0, 0, 'V' },
        { "working-directory", 1, 0, 'w', },
+       { "group-output-by-session", 0, 0, 's', },
+       { "group-output-by-host", 0, 0, 'p', },
        { NULL, 0, 0, 0, },
 };
 
 static const char *config_ignore_options[] = { "help", "config", "version" };
 
+static void print_version(void) {
+       fprintf(stdout, "%s\n", VERSION);
+}
+
+static void relayd_config_log(void)
+{
+       DBG("LTTng-relayd " VERSION " - " VERSION_NAME "%s%s",
+                       GIT_VERSION[0] == '\0' ? "" : " - " GIT_VERSION,
+                       EXTRA_VERSION_NAME[0] == '\0' ? "" : " - " EXTRA_VERSION_NAME);
+       if (EXTRA_VERSION_DESCRIPTION[0] != '\0') {
+               DBG("LTTng-relayd extra version description:\n\t" EXTRA_VERSION_DESCRIPTION "\n");
+       }
+       if (EXTRA_VERSION_PATCHES[0] != '\0') {
+               DBG("LTTng-relayd extra patches:\n\t" EXTRA_VERSION_PATCHES "\n");
+       }
+}
+
 /*
  * Take an option from the getopt output and set it in the right variable to be
  * used later.
@@ -187,9 +219,33 @@ static int set_option(int opt, const char *arg, const char *optname)
 
        switch (opt) {
        case 0:
-               fprintf(stderr, "option %s", optname);
-               if (arg) {
-                       fprintf(stderr, " with arg %s\n", arg);
+               if (!strcmp(optname, "fd-cap")) {
+                       unsigned long v;
+
+                       errno = 0;
+                       v = strtoul(arg, NULL, 0);
+                       if (errno != 0 || !isdigit(arg[0])) {
+                               ERR("Wrong value in --fd-cap parameter: %s", arg);
+                               ret = -1;
+                               goto end;
+                       }
+                       if (v < DEFAULT_RELAYD_MINIMAL_FD_CAP) {
+                               ERR("File descriptor cap must be set to at least %d",
+                                               DEFAULT_RELAYD_MINIMAL_FD_CAP);
+                       }
+                       if (v >= UINT_MAX) {
+                               ERR("File descriptor cap overflow in --fd-cap parameter: %s", arg);
+                               ret = -1;
+                               goto end;
+                       }
+                       lttng_opt_fd_cap = (unsigned int) v;
+                       DBG3("File descriptor cap set to %u", lttng_opt_fd_cap);
+
+               } else {
+                       fprintf(stderr, "unknown option %s", optname);
+                       if (arg) {
+                               fprintf(stderr, " with arg %s\n", arg);
+                       }
                }
                break;
        case 'C':
@@ -265,8 +321,8 @@ static int set_option(int opt, const char *arg, const char *optname)
                }
                exit(EXIT_FAILURE);
        case 'V':
-               fprintf(stdout, "%s\n", VERSION);
-               exit(EXIT_SUCCESS);
+               opt_print_version = 1;
+               break;
        case 'o':
                if (lttng_is_setuid_setgid()) {
                        WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
@@ -305,6 +361,20 @@ static int set_option(int opt, const char *arg, const char *optname)
                        }
                }
                break;
+       case 's':
+               if (opt_group_output_by_host) {
+                       ERR("Cannot set --group-output-by-session, --group-output-by-host already defined");
+                       exit(EXIT_FAILURE);
+               }
+               opt_group_output_by_session = 1;
+               break;
+       case 'p':
+               if (opt_group_output_by_session) {
+                       ERR("Cannot set --group-output-by-host, --group-output-by-session already defined");
+                       exit(EXIT_FAILURE);
+               }
+               opt_group_output_by_host = 1;
+               break;
        default:
                /* Unknown option or other error.
                 * Error is printed by getopt, just return */
@@ -374,6 +444,16 @@ end:
        return ret;
 }
 
+static void parse_env_options(void)
+{
+       char *value = NULL;
+
+       value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_WORKING_DIRECTORY_ENV);
+       if (value) {
+               opt_working_directory = value;
+       }
+}
+
 static int set_options(int argc, char **argv)
 {
        int c, ret = 0, option_index = 0, retval = 0;
@@ -491,6 +571,23 @@ static int set_options(int argc, char **argv)
                        goto exit;
                }
        }
+       if (lttng_opt_fd_cap == 0) {
+               int ret;
+               struct rlimit rlimit;
+
+               ret = getrlimit(RLIMIT_NOFILE, &rlimit);
+               if (ret) {
+                       PERROR("Failed to get file descriptor limit");
+                       retval = -1;
+               }
+
+               lttng_opt_fd_cap = rlimit.rlim_cur;
+       }
+
+       if (!opt_group_output_by_session && !opt_group_output_by_host) {
+               /* Group by host by default */
+               opt_group_output_by_host = 1;
+       }
 
 exit:
        free(optstring);
@@ -499,13 +596,9 @@ exit:
 
 static void print_global_objects(void)
 {
-       rcu_register_thread();
-
        print_viewer_streams();
        print_relay_streams();
        print_sessions();
-
-       rcu_unregister_thread();
 }
 
 /*
@@ -528,7 +621,7 @@ static void relayd_cleanup(void)
        free(opt_output_path);
 
        /* Close thread quit pipes */
-       utils_close_pipe(thread_quit_pipe);
+       (void) fd_tracker_util_pipe_close(the_fd_tracker, thread_quit_pipe);
 
        uri_free(control_uri);
        uri_free(data_uri);
@@ -690,17 +783,26 @@ void lttng_relay_notify_ready(void)
  */
 static int init_thread_quit_pipe(void)
 {
-       int ret;
-
-       ret = utils_create_pipe_cloexec(thread_quit_pipe);
+       return fd_tracker_util_pipe_open_cloexec(the_fd_tracker,
+                       "Quit pipe", thread_quit_pipe);
+}
 
-       return ret;
+/*
+ * Init health quit pipe.
+ *
+ * Return -1 on error or 0 if all pipes are created.
+ */
+static int init_health_quit_pipe(void)
+{
+       return fd_tracker_util_pipe_open_cloexec(the_fd_tracker,
+                       "Health quit pipe", health_quit_pipe);
 }
 
 /*
  * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
  */
-static int create_thread_poll_set(struct lttng_poll_event *events, int size)
+static int create_named_thread_poll_set(struct lttng_poll_event *events,
+               int size, const char *name)
 {
        int ret;
 
@@ -709,10 +811,8 @@ static int create_thread_poll_set(struct lttng_poll_event *events, int size)
                goto error;
        }
 
-       ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
-       if (ret < 0) {
-               goto error;
-       }
+       ret = fd_tracker_util_poll_create(the_fd_tracker,
+                       name, events, 1, LTTNG_CLOEXEC);
 
        /* Add quit pipe */
        ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
@@ -740,13 +840,55 @@ static int check_thread_quit_pipe(int fd, uint32_t events)
        return 0;
 }
 
+static int create_sock(void *data, int *out_fd)
+{
+       int ret;
+       struct lttcomm_sock *sock = data;
+
+       ret = lttcomm_create_sock(sock);
+       if (ret < 0) {
+               goto end;
+       }
+
+       *out_fd = sock->fd;
+end:
+       return ret;
+}
+
+static int close_sock(void *data, int *in_fd)
+{
+       struct lttcomm_sock *sock = data;
+
+       return sock->ops->close(sock);
+}
+
+static int accept_sock(void *data, int *out_fd)
+{
+       int ret = 0;
+       /* Socks is an array of in_sock, out_sock. */
+       struct lttcomm_sock **socks = data;
+       struct lttcomm_sock *in_sock = socks[0];
+
+        socks[1] = in_sock->ops->accept(in_sock);
+       if (!socks[1]) {
+               ret = -1;
+               goto end;
+       }
+       *out_fd = socks[1]->fd;
+end:
+       return ret;
+}
+
 /*
  * Create and init socket from uri.
  */
-static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri)
+static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri,
+               const char *name)
 {
-       int ret;
+       int ret, sock_fd;
        struct lttcomm_sock *sock = NULL;
+       char uri_str[PATH_MAX];
+       char *formated_name = NULL;
 
        sock = lttcomm_alloc_sock_from_uri(uri);
        if (sock == NULL) {
@@ -754,11 +896,25 @@ static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri)
                goto error;
        }
 
-       ret = lttcomm_create_sock(sock);
-       if (ret < 0) {
-               goto error;
+       /*
+        * Don't fail to create the socket if the name can't be built as it is
+        * only used for debugging purposes.
+        */
+       ret = uri_to_str_url(uri, uri_str, sizeof(uri_str));
+       uri_str[sizeof(uri_str) - 1] = '\0';
+       if (ret >= 0) {
+               ret = asprintf(&formated_name, "%s socket @ %s", name,
+                               uri_str);
+               if (ret < 0) {
+                       formated_name = NULL;
+               }
        }
-       DBG("Listening on sock %d", sock->fd);
+
+       ret = fd_tracker_open_unsuspendable_fd(the_fd_tracker, &sock_fd,
+                       (const char **) (formated_name ? &formated_name : NULL),
+                       1, create_sock, sock);
+       free(formated_name);
+       DBG("Listening on %s socket %d", name, sock->fd);
 
        ret = sock->ops->bind(sock);
        if (ret < 0) {
@@ -780,6 +936,27 @@ error:
        return NULL;
 }
 
+static
+struct lttcomm_sock *accept_relayd_sock(struct lttcomm_sock *listening_sock,
+               const char *name)
+{
+       int out_fd, ret;
+       struct lttcomm_sock *socks[2] = { listening_sock, NULL };
+       struct lttcomm_sock *new_sock = NULL;
+
+        ret = fd_tracker_open_unsuspendable_fd(
+                       the_fd_tracker, &out_fd,
+                       (const char **) &name,
+                       1, accept_sock, &socks);
+       if (ret) {
+               goto end;
+       }
+       new_sock = socks[1];
+       DBG("%s accepted, socket %d", name, new_sock->fd);
+end:
+       return new_sock;
+}
+
 /*
  * This thread manages the listening for new connections on the network
  */
@@ -796,12 +973,12 @@ static void *relay_thread_listener(void *data)
 
        health_code_update();
 
-       control_sock = relay_socket_create(control_uri);
+       control_sock = relay_socket_create(control_uri, "Control listener");
        if (!control_sock) {
                goto error_sock_control;
        }
 
-       data_sock = relay_socket_create(data_uri);
+       data_sock = relay_socket_create(data_uri, "Data listener");
        if (!data_sock) {
                goto error_sock_relay;
        }
@@ -810,7 +987,7 @@ static void *relay_thread_listener(void *data)
         * Pass 3 as size here for the thread quit pipe, control and
         * data socket.
         */
-       ret = create_thread_poll_set(&events, 3);
+       ret = create_named_thread_poll_set(&events, 3, "Listener thread epoll");
        if (ret < 0) {
                goto error_create_poll;
        }
@@ -886,20 +1063,18 @@ restart:
                                 */
                                int val = 1;
                                struct relay_connection *new_conn;
-                               struct lttcomm_sock *newsock;
+                               struct lttcomm_sock *newsock = NULL;
                                enum connection_type type;
 
                                if (pollfd == data_sock->fd) {
                                        type = RELAY_DATA;
-                                       newsock = data_sock->ops->accept(data_sock);
-                                       DBG("Relay data connection accepted, socket %d",
-                                                       newsock->fd);
+                                       newsock = accept_relayd_sock(data_sock,
+                                                       "Data socket to relayd");
                                } else {
                                        assert(pollfd == control_sock->fd);
                                        type = RELAY_CONTROL;
-                                       newsock = control_sock->ops->accept(control_sock);
-                                       DBG("Relay control connection accepted, socket %d",
-                                                       newsock->fd);
+                                       newsock = accept_relayd_sock(control_sock,
+                                                       "Control socket to relayd");
                                }
                                if (!newsock) {
                                        PERROR("accepting sock");
@@ -952,10 +1127,12 @@ exit:
 error:
 error_poll_add:
 error_testpoint:
-       lttng_poll_clean(&events);
+       (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
 error_create_poll:
        if (data_sock->fd >= 0) {
-               ret = data_sock->ops->close(data_sock);
+               ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker,
+                               &data_sock->fd, 1, close_sock,
+                               data_sock);
                if (ret) {
                        PERROR("close");
                }
@@ -963,7 +1140,9 @@ error_create_poll:
        lttcomm_destroy_sock(data_sock);
 error_sock_relay:
        if (control_sock->fd >= 0) {
-               ret = control_sock->ops->close(control_sock);
+               ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker,
+                               &control_sock->fd, 1, close_sock,
+                               control_sock);
                if (ret) {
                        PERROR("close");
                }
@@ -1209,12 +1388,13 @@ static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr,
        switch (session->minor) {
        case 1: /* LTTng sessiond 2.1. Allocates path_name and channel_name. */
                ret = cmd_recv_stream_2_1(payload, &path_name,
-                       &channel_name);
+                       &channel_name, session);
                break;
        case 2: /* LTTng sessiond 2.2. Allocates path_name and channel_name. */
        default:
                ret = cmd_recv_stream_2_2(payload, &path_name,
-                       &channel_name, &tracefile_size, &tracefile_count);
+                       &channel_name, &tracefile_size, &tracefile_count,
+                       session);
                break;
        }
        if (ret < 0) {
@@ -2780,7 +2960,8 @@ static void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollf
 
        (void) lttng_poll_del(events, pollfd);
 
-       ret = close(pollfd);
+       ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker, &pollfd, 1,
+                       fd_tracker_util_close_fd, NULL);
        if (ret < 0) {
                ERR("Closing pollfd %d", pollfd);
        }
@@ -2842,7 +3023,7 @@ static void *relay_thread_worker(void *data)
                goto relay_connections_ht_error;
        }
 
-       ret = create_thread_poll_set(&events, 2);
+       ret = create_named_thread_poll_set(&events, 2, "Worker thread epoll");
        if (ret < 0) {
                goto error_poll_create;
        }
@@ -3114,12 +3295,13 @@ error:
        }
        rcu_read_unlock();
 
-       lttng_poll_clean(&events);
+       (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
 error_poll_create:
        lttng_ht_destroy(relay_connections_ht);
 relay_connections_ht_error:
        /* Close relay conn pipes */
-       utils_close_pipe(relay_conn_pipe);
+       (void) fd_tracker_util_pipe_close(the_fd_tracker,
+                       relay_conn_pipe);
        if (err) {
                DBG("Thread exited with error");
        }
@@ -3141,11 +3323,45 @@ error_testpoint:
  */
 static int create_relay_conn_pipe(void)
 {
-       int ret;
+       return fd_tracker_util_pipe_open_cloexec(the_fd_tracker,
+                       "Relayd connection pipe", relay_conn_pipe);
+}
 
-       ret = utils_create_pipe_cloexec(relay_conn_pipe);
+static
+int stdio_open(void *data, int *fds)
+{
+       fds[0] = fileno(stdout);
+       fds[1] = fileno(stderr);
+       return 0;
+}
 
-       return ret;
+static
+int noop_close(void *data, int *fds)
+{
+       return 0;
+}
+
+static
+int track_stdio(void)
+{
+       int fds[2];
+       const char *names[] = { "stdout", "stderr" };
+
+       return fd_tracker_open_unsuspendable_fd(the_fd_tracker, fds,
+                       names, 2, stdio_open, NULL);
+}
+
+static
+void untrack_stdio(void)
+{
+       int fds[] = { fileno(stdout), fileno(stderr) };
+
+       /*
+        * noop_close is used since we don't really want to close
+        * the stdio output fds; we merely want to stop tracking them.
+        */
+       (void) fd_tracker_close_unsuspendable_fd(the_fd_tracker,
+                       fds, 2, noop_close, NULL);
 }
 
 /*
@@ -3156,7 +3372,13 @@ int main(int argc, char **argv)
        int ret = 0, retval = 0;
        void *status;
 
-       /* Parse arguments */
+       /* Parse environment variables */
+       parse_env_options();
+
+       /*
+        * Parse arguments.
+        * Command line arguments overwrite environment.
+        */
        progname = argv[0];
        if (set_options(argc, argv)) {
                retval = -1;
@@ -3168,6 +3390,19 @@ int main(int argc, char **argv)
                goto exit_options;
        }
 
+       relayd_config_log();
+
+       if (opt_print_version) {
+               print_version();
+               retval = 0;
+               goto exit_options;
+       }
+
+       ret = fclose(stdin);
+       if (ret) {
+               PERROR("Failed to close stdin");
+               goto exit_options;
+       }
        /* Try to create directory if -o, --output is specified. */
        if (opt_output_path) {
                if (*opt_output_path != '/') {
@@ -3187,26 +3422,14 @@ int main(int argc, char **argv)
 
        /* Daemonize */
        if (opt_daemon || opt_background) {
-               int i;
-
                ret = lttng_daemonize(&child_ppid, &recv_child_signal,
                        !opt_background);
                if (ret < 0) {
                        retval = -1;
                        goto exit_options;
                }
-
-               /*
-                * We are in the child. Make sure all other file
-                * descriptors are closed, in case we are called with
-                * more opened file descriptors than the standard ones.
-                */
-               for (i = 3; i < sysconf(_SC_OPEN_MAX); i++) {
-                       (void) close(i);
-               }
        }
 
-
        if (opt_working_directory) {
                ret = utils_change_working_dir(opt_working_directory);
                if (ret) {
@@ -3214,6 +3437,25 @@ int main(int argc, char **argv)
                        goto exit_options;
                }
        }
+       /*
+        * The RCU thread registration (and use, through the fd-tracker's
+        * creation) is done after the daemonization to allow us to not
+        * deal with liburcu's fork() management as the call RCU needs to
+        * be restored.
+        */
+       rcu_register_thread();
+
+       the_fd_tracker = fd_tracker_create(lttng_opt_fd_cap);
+       if (!the_fd_tracker) {
+               retval = -1;
+               goto exit_options;
+       }
+
+       ret = track_stdio();
+       if (ret) {
+               retval = -1;
+               goto exit_options;
+       }
 
        /* Initialize thread health monitoring */
        health_relayd = health_app_create(NR_HEALTH_RELAYD_TYPES);
@@ -3263,7 +3505,7 @@ int main(int argc, char **argv)
                goto exit_init_data;
        }
 
-       ret = utils_create_pipe(health_quit_pipe);
+       ret = init_health_quit_pipe();
        if (ret) {
                retval = -1;
                goto exit_health_quit_pipe;
@@ -3359,7 +3601,7 @@ exit_dispatcher_thread:
        }
 exit_health_thread:
 
-       utils_close_pipe(health_quit_pipe);
+       (void) fd_tracker_util_pipe_close(the_fd_tracker, health_quit_pipe);
 exit_health_quit_pipe:
 
 exit_init_data:
@@ -3377,6 +3619,14 @@ exit_options:
        /* Ensure all prior call_rcu are done. */
        rcu_barrier();
 
+       untrack_stdio();
+       /*
+        * fd_tracker_destroy() will log the contents of the fd-tracker
+        * if a leak is detected.
+        */
+       fd_tracker_destroy(the_fd_tracker);
+       rcu_unregister_thread();
+
        if (!retval) {
                exit(EXIT_SUCCESS);
        } else {
This page took 0.048971 seconds and 5 git commands to generate.