X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=8e1879f50ae3875dd90581c1fe3a43465d8057af;hb=6addfa379ee608b20cfe5e15d135bcb6a9724e90;hp=35725b61e6bfe16a90a971a202cedf7d40698465;hpb=178a055717baca3641cecbb45fe3c0d5d3286a3a;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 35725b61e..8e1879f50 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -46,6 +46,7 @@ #include #include #include +#include #include #include #include @@ -197,33 +198,48 @@ int set_option(int opt, const char *arg, const char *optname) } break; case 'C': - ret = uri_parse(arg, &control_uri); - if (ret < 0) { - ERR("Invalid control URI specified"); - goto end; - } - if (control_uri->port == 0) { - control_uri->port = DEFAULT_NETWORK_CONTROL_PORT; + if (lttng_is_setuid_setgid()) { + WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.", + "-C, --control-port"); + } else { + ret = uri_parse(arg, &control_uri); + if (ret < 0) { + ERR("Invalid control URI specified"); + goto end; + } + if (control_uri->port == 0) { + control_uri->port = DEFAULT_NETWORK_CONTROL_PORT; + } } break; case 'D': - ret = uri_parse(arg, &data_uri); - if (ret < 0) { - ERR("Invalid data URI specified"); - goto end; - } - if (data_uri->port == 0) { - data_uri->port = DEFAULT_NETWORK_DATA_PORT; + if (lttng_is_setuid_setgid()) { + WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.", + "-D, -data-port"); + } else { + ret = uri_parse(arg, &data_uri); + if (ret < 0) { + ERR("Invalid data URI specified"); + goto end; + } + if (data_uri->port == 0) { + data_uri->port = DEFAULT_NETWORK_DATA_PORT; + } } break; case 'L': - ret = uri_parse(arg, &live_uri); - if (ret < 0) { - ERR("Invalid live URI specified"); - goto end; - } - if (live_uri->port == 0) { - live_uri->port = DEFAULT_NETWORK_VIEWER_PORT; + if (lttng_is_setuid_setgid()) { + WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.", + "-L, -live-port"); + } else { + ret = uri_parse(arg, &live_uri); + if (ret < 0) { + ERR("Invalid live URI specified"); + goto end; + } + if (live_uri->port == 0) { + live_uri->port = DEFAULT_NETWORK_VIEWER_PORT; + } } break; case 'd': @@ -233,23 +249,33 @@ int set_option(int opt, const char *arg, const char *optname) opt_background = 1; break; case 'g': - tracing_group_name = strdup(arg); - if (tracing_group_name == NULL) { - ret = -errno; - PERROR("strdup"); - goto end; + if (lttng_is_setuid_setgid()) { + WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.", + "-g, --group"); + } else { + tracing_group_name = strdup(arg); + if (tracing_group_name == NULL) { + ret = -errno; + PERROR("strdup"); + goto end; + } + tracing_group_name_override = 1; } - tracing_group_name_override = 1; break; case 'h': usage(); exit(EXIT_FAILURE); case 'o': - ret = asprintf(&opt_output_path, "%s", arg); - if (ret < 0) { - ret = -errno; - PERROR("asprintf opt_output_path"); - goto end; + if (lttng_is_setuid_setgid()) { + WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.", + "-o, --output"); + } else { + ret = asprintf(&opt_output_path, "%s", arg); + if (ret < 0) { + ret = -errno; + PERROR("asprintf opt_output_path"); + goto end; + } } break; case 'v': @@ -359,9 +385,14 @@ int set_options(int argc, char **argv) continue; } - config_path = utils_expand_path(optarg); - if (!config_path) { - ERR("Failed to resolve path: %s", optarg); + if (lttng_is_setuid_setgid()) { + WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.", + "-f, --config"); + } else { + config_path = utils_expand_path(optarg); + if (!config_path) { + ERR("Failed to resolve path: %s", optarg); + } } } @@ -494,46 +525,55 @@ int notify_thread_pipe(int wpipe) ret = lttng_write(wpipe, "!", 1); if (ret < 1) { PERROR("write poll pipe"); + goto end; } - + ret = 0; +end: return ret; } -static void notify_health_quit_pipe(int *pipe) +static +int notify_health_quit_pipe(int *pipe) { ssize_t ret; ret = lttng_write(pipe[1], "4", 1); if (ret < 1) { PERROR("write relay health quit"); + goto end; } + ret = 0; +end: + return ret; } /* - * Stop all threads by closing the thread quit pipe. + * Stop all relayd and relayd-live threads. */ -static -void stop_threads(void) +int lttng_relay_stop_threads(void) { - int ret; + int retval = 0; /* Stopping all threads */ DBG("Terminating all threads"); - ret = notify_thread_pipe(thread_quit_pipe[1]); - if (ret < 0) { + if (notify_thread_pipe(thread_quit_pipe[1])) { ERR("write error on thread quit pipe"); + retval = -1; } - notify_health_quit_pipe(health_quit_pipe); + if (notify_health_quit_pipe(health_quit_pipe)) { + ERR("write error on health quit pipe"); + } /* Dispatch thread */ CMM_STORE_SHARED(dispatch_thread_exit, 1); futex_nto1_wake(&relay_conn_queue.futex); - ret = relayd_live_stop(); - if (ret) { + if (relayd_live_stop()) { ERR("Error stopping live threads"); + retval = -1; } + return retval; } /* @@ -551,11 +591,15 @@ void sighandler(int sig) return; case SIGINT: DBG("SIGINT caught"); - stop_threads(); + if (lttng_relay_stop_threads()) { + ERR("Error stopping threads"); + } break; case SIGTERM: DBG("SIGTERM caught"); - stop_threads(); + if (lttng_relay_stop_threads()) { + ERR("Error stopping threads"); + } break; case SIGUSR1: CMM_STORE_SHARED(recv_child_signal, 1); @@ -855,6 +899,11 @@ restart: revents = LTTNG_POLL_GETEV(&events, i); pollfd = LTTNG_POLL_GETFD(&events, i); + if (!revents) { + /* No activity for this FD (poll implementation). */ + continue; + } + /* Thread quit pipe has been closed. Killing thread. */ ret = check_thread_quit_pipe(pollfd, revents); if (ret) { @@ -948,7 +997,7 @@ error_sock_control: } health_unregister(health_relayd); DBG("Relay listener thread cleanup complete"); - stop_threads(); + lttng_relay_stop_threads(); return NULL; } @@ -1024,7 +1073,7 @@ error_testpoint: } health_unregister(health_relayd); DBG("Dispatch thread dying"); - stop_threads(); + lttng_relay_stop_threads(); return NULL; } @@ -1199,7 +1248,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, struct relay_session *session = conn->session; struct relay_stream *stream = NULL; struct lttcomm_relayd_status_stream reply; - struct ctf_trace *trace; + struct ctf_trace *trace = NULL; if (!session || conn->version_check_done == 0) { ERR("Trying to add a stream before version check"); @@ -1227,7 +1276,6 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, goto err_free_stream; } - rcu_read_lock(); stream->stream_handle = ++last_relay_stream_id; stream->prev_seq = -1ULL; stream->session_id = session->id; @@ -1240,7 +1288,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, ret = utils_mkdir_recursive(stream->path_name, S_IRWXU | S_IRWXG); if (ret < 0) { ERR("relay creating output directory"); - goto end; + goto err_free_stream; } /* @@ -1251,7 +1299,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, stream->tracefile_size, 0, relayd_uid, relayd_gid, NULL); if (ret < 0) { ERR("Create output file"); - goto end; + goto err_free_stream; } stream->fd = ret; if (stream->tracefile_size) { @@ -1260,6 +1308,8 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, DBG("Tracefile %s/%s created", stream->path_name, stream->channel_name); } + /* Protect access to "trace" */ + rcu_read_lock(); trace = ctf_trace_find_by_path(session->ctf_traces_ht, stream->path_name); if (!trace) { trace = ctf_trace_create(stream->path_name); @@ -1287,6 +1337,9 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, /* * Both in the ctf_trace object and the global stream ht since the data * side of the relayd does not have the concept of session. + * + * rcu_read_lock() is kept to protect the stream which is now part of + * the relay_streams_ht. */ lttng_ht_add_unique_u64(relay_streams_ht, &stream->node); cds_list_add_tail(&stream->trace_list, &trace->stream_list); @@ -1303,7 +1356,7 @@ end: if (ret < 0) { reply.ret_code = htobe32(LTTNG_ERR_UNK); /* stream was not properly added to the ht, so free it */ - free(stream); + stream_destroy(stream); } else { reply.ret_code = htobe32(LTTNG_OK); } @@ -1314,15 +1367,19 @@ end: ERR("Relay sending stream id"); ret = send_ret; } + /* + * rcu_read_lock() was held to protect either "trace" OR the "stream" at + * this point. + */ rcu_read_unlock(); + trace = NULL; + stream = NULL; end_no_session: return ret; err_free_stream: - free(stream->path_name); - free(stream->channel_name); - free(stream); + stream_destroy(stream); return ret; } @@ -2341,7 +2398,7 @@ int relay_process_data(struct relay_connection *conn) pthread_mutex_lock(&vstream->overwrite_lock); vstream->abort_flag = 1; pthread_mutex_unlock(&vstream->overwrite_lock); - DBG("Streaming side setting abort_flag on stream %s_%lu\n", + DBG("Streaming side setting abort_flag on stream %s_%" PRIu64 "\n", stream->channel_name, new_id); } else if (vstream->tracefile_count_current == stream->tracefile_count_current) { @@ -2445,7 +2502,6 @@ void *relay_thread_worker(void *data) { int ret, err = -1, last_seen_data_fd = -1; uint32_t nb_fd; - struct relay_connection *conn; struct lttng_poll_event events; struct lttng_ht *relay_connections_ht; struct lttng_ht_iter iter; @@ -2453,6 +2509,7 @@ void *relay_thread_worker(void *data) struct relay_local_data *relay_ctx = (struct relay_local_data *) data; struct lttng_ht *sessions_ht = relay_ctx->sessions_ht; struct relay_index *index; + struct relay_connection *destroy_conn = NULL; DBG("[thread] Relay worker started"); @@ -2512,8 +2569,8 @@ restart: nb_fd = ret; /* - * Process control. The control connection is prioritised so we don't - * starve it with high throughout put tracing data on the data + * Process control. The control connection is prioritised so we + * don't starve it with high throughput tracing data on the data * connection. */ for (i = 0; i < nb_fd; i++) { @@ -2523,6 +2580,11 @@ restart: health_code_update(); + if (!revents) { + /* No activity for this FD (poll implementation). */ + continue; + } + /* Thread quit pipe has been closed. Killing thread. */ ret = check_thread_quit_pipe(pollfd, revents); if (ret) { @@ -2536,6 +2598,8 @@ restart: ERR("Relay connection pipe error"); goto error; } else if (revents & LPOLLIN) { + struct relay_connection *conn; + ret = lttng_read(relay_conn_pipe[0], &conn, sizeof(conn)); if (ret < 0) { goto error; @@ -2551,32 +2615,34 @@ restart: DBG("Connection socket %d added", conn->sock->fd); } } else { + struct relay_connection *ctrl_conn; + rcu_read_lock(); - conn = connection_find_by_sock(relay_connections_ht, pollfd); + ctrl_conn = connection_find_by_sock(relay_connections_ht, pollfd); /* If not found, there is a synchronization issue. */ - assert(conn); + assert(ctrl_conn); if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { cleanup_connection_pollfd(&events, pollfd); - destroy_connection(relay_connections_ht, conn); + destroy_connection(relay_connections_ht, ctrl_conn); if (last_seen_data_fd == pollfd) { last_seen_data_fd = last_notdel_data_fd; } } else if (revents & LPOLLIN) { - if (conn->type == RELAY_CONTROL) { - ret = conn->sock->ops->recvmsg(conn->sock, &recv_hdr, + if (ctrl_conn->type == RELAY_CONTROL) { + ret = ctrl_conn->sock->ops->recvmsg(ctrl_conn->sock, &recv_hdr, sizeof(recv_hdr), 0); if (ret <= 0) { /* Connection closed */ cleanup_connection_pollfd(&events, pollfd); - destroy_connection(relay_connections_ht, conn); + destroy_connection(relay_connections_ht, ctrl_conn); DBG("Control connection closed with %d", pollfd); } else { - ret = relay_process_control(&recv_hdr, conn); + ret = relay_process_control(&recv_hdr, ctrl_conn); if (ret < 0) { /* Clear the session on error. */ cleanup_connection_pollfd(&events, pollfd); - destroy_connection(relay_connections_ht, conn); + destroy_connection(relay_connections_ht, ctrl_conn); DBG("Connection closed with %d", pollfd); } seen_control = 1; @@ -2622,49 +2688,53 @@ restart: /* Fetch the poll data. */ uint32_t revents = LTTNG_POLL_GETEV(&events, i); int pollfd = LTTNG_POLL_GETFD(&events, i); + struct relay_connection *data_conn; health_code_update(); + if (!revents) { + /* No activity for this FD (poll implementation). */ + continue; + } + /* Skip the command pipe. It's handled in the first loop. */ if (pollfd == relay_conn_pipe[0]) { continue; } - if (revents) { - rcu_read_lock(); - conn = connection_find_by_sock(relay_connections_ht, pollfd); - if (!conn) { - /* Skip it. Might be removed before. */ + rcu_read_lock(); + data_conn = connection_find_by_sock(relay_connections_ht, pollfd); + if (!data_conn) { + /* Skip it. Might be removed before. */ + rcu_read_unlock(); + continue; + } + + if (revents & LPOLLIN) { + if (data_conn->type != RELAY_DATA) { rcu_read_unlock(); continue; } - if (revents & LPOLLIN) { - if (conn->type != RELAY_DATA) { - rcu_read_unlock(); - continue; - } - - ret = relay_process_data(conn); - /* Connection closed */ - if (ret < 0) { - cleanup_connection_pollfd(&events, pollfd); - destroy_connection(relay_connections_ht, conn); - DBG("Data connection closed with %d", pollfd); - /* - * Every goto restart call sets the last seen fd where - * here we don't really care since we gracefully - * continue the loop after the connection is deleted. - */ - } else { - /* Keep last seen port. */ - last_seen_data_fd = pollfd; - rcu_read_unlock(); - goto restart; - } + ret = relay_process_data(data_conn); + /* Connection closed */ + if (ret < 0) { + cleanup_connection_pollfd(&events, pollfd); + destroy_connection(relay_connections_ht, data_conn); + DBG("Data connection closed with %d", pollfd); + /* + * Every goto restart call sets the last seen fd where + * here we don't really care since we gracefully + * continue the loop after the connection is deleted. + */ + } else { + /* Keep last seen port. */ + last_seen_data_fd = pollfd; + rcu_read_unlock(); + goto restart; } - rcu_read_unlock(); } + rcu_read_unlock(); } last_seen_data_fd = -1; } @@ -2678,10 +2748,11 @@ error: /* Cleanup reamaining connection object. */ rcu_read_lock(); - cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, conn, + cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, + destroy_conn, sock_n.node) { health_code_update(); - destroy_connection(relay_connections_ht, conn); + destroy_connection(relay_connections_ht, destroy_conn); } rcu_read_unlock(); error_poll_create: @@ -2711,7 +2782,7 @@ error_testpoint: } health_unregister(health_relayd); rcu_unregister_thread(); - stop_threads(); + lttng_relay_stop_threads(); return NULL; } @@ -2824,7 +2895,10 @@ int main(int argc, char **argv) cds_wfcq_init(&relay_conn_queue.head, &relay_conn_queue.tail); /* Set up max poll set size */ - lttng_poll_set_max_size(); + if (lttng_poll_set_max_size()) { + retval = -1; + goto exit_init_data; + } /* Initialize communication library */ lttcomm_init();