X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=e745702f85cd9cea776a1063d7dea6025849bd4a;hp=155f2b053aca8dfdfc3653c4c9bfdd7eb8f71ada;hb=51a4828f21385847870ec2b2499e2f3e3842c26e;hpb=bbc4768c20f1c552222e1746f9475d145d7bf04e diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index 155f2b053..e745702f8 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -38,6 +38,63 @@ #include "utils.h" #include "lttng-sessiond.h" +/* + * Return allocated full pathname of the session using the consumer trace path + * and subdir if available. + * + * The caller can safely free(3) the returned value. On error, NULL is + * returned. + */ +char *setup_channel_trace_path(struct consumer_output *consumer, + const char *session_path, size_t *consumer_path_offset) +{ + int ret; + char *pathname; + + assert(consumer); + assert(session_path); + + health_code_update(); + + /* + * Allocate the string ourself to make sure we never exceed + * LTTNG_PATH_MAX. + */ + pathname = zmalloc(LTTNG_PATH_MAX); + if (!pathname) { + goto error; + } + + /* Get correct path name destination */ + if (consumer->type == CONSUMER_DST_NET && + consumer->relay_major_version == 2 && + consumer->relay_minor_version < 11) { + ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s/%s/%s", + consumer->dst.net.base_dir, + consumer->chunk_path, consumer->domain_subdir, + session_path); + *consumer_path_offset = 0; + } else { + ret = snprintf(pathname, LTTNG_PATH_MAX, "%s/%s", + consumer->domain_subdir, session_path); + *consumer_path_offset = strlen(consumer->domain_subdir) + 1; + } + DBG3("Consumer trace path relative to current trace chunk: \"%s\"", + pathname); + if (ret < 0) { + PERROR("Failed to format channel path"); + goto error; + } else if (ret >= LTTNG_PATH_MAX) { + ERR("Truncation occurred while formatting channel path"); + goto error; + } + + return pathname; +error: + free(pathname); + return NULL; +} + /* * Send a data payload using a given consumer socket of size len. * @@ -315,7 +372,7 @@ error: * object reference is not needed anymore. */ struct consumer_socket *consumer_find_socket_by_bitness(int bits, - struct consumer_output *consumer) + const struct consumer_output *consumer) { int consumer_fd; struct consumer_socket *socket = NULL; @@ -348,7 +405,7 @@ end: * returned consumer_socket. */ struct consumer_socket *consumer_find_socket(int key, - struct consumer_output *consumer) + const struct consumer_output *consumer) { struct lttng_ht_iter iter; struct lttng_ht_node_ulong *node; @@ -405,7 +462,7 @@ void consumer_add_socket(struct consumer_socket *sock, } /* - * Delte consumer socket to consumer output object. Read side lock must be + * Delete consumer socket to consumer output object. Read side lock must be * acquired before calling this function. */ void consumer_del_socket(struct consumer_socket *sock, @@ -569,6 +626,7 @@ struct consumer_output *consumer_copy_output(struct consumer_output *src) output->snapshot = src->snapshot; output->relay_major_version = src->relay_major_version; output->relay_minor_version = src->relay_minor_version; + output->relay_allows_clear = src->relay_allows_clear; memcpy(&output->dst, &src->dst, sizeof(output->dst)); ret = consumer_copy_sockets(output, src); if (ret < 0) { @@ -893,10 +951,6 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.ask_channel.buffer_credentials.uid = UINT32_MAX; msg->u.ask_channel.buffer_credentials.gid = UINT32_MAX; - if (monitor) { - assert(trace_chunk); - } - if (trace_chunk) { uint64_t chunk_id; enum lttng_trace_chunk_status chunk_status; @@ -1079,7 +1133,9 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer, enum lttng_stream_type type, uint64_t session_id, const char *session_name, const char *hostname, - int session_live_timer, const uint64_t *current_chunk_id) + const char *base_path, int session_live_timer, + const uint64_t *current_chunk_id, time_t session_creation_time, + bool session_name_contains_creation_time) { int ret; struct lttcomm_consumer_msg msg; @@ -1097,16 +1153,26 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, } if (type == LTTNG_STREAM_CONTROL) { + char output_path[LTTNG_PATH_MAX] = {}; + uint64_t relayd_session_id; + ret = relayd_create_session(rsock, - &msg.u.relayd_sock.relayd_session_id, - session_name, hostname, session_live_timer, + &relayd_session_id, + session_name, hostname, base_path, + session_live_timer, consumer->snapshot, session_id, - sessiond_uuid, current_chunk_id); + sessiond_uuid, current_chunk_id, + session_creation_time, + session_name_contains_creation_time, + output_path); if (ret < 0) { /* Close the control socket. */ (void) relayd_close(rsock); goto error; } + msg.u.relayd_sock.relayd_session_id = relayd_session_id; + DBG("Created session on relay, output path reply: %s", + output_path); } msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET; @@ -1127,7 +1193,7 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, } DBG3("Sending relayd socket file descriptor to consumer"); - ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1); + ret = consumer_send_fds(consumer_sock, ALIGNED_CONST_PTR(rsock->sock.fd), 1); if (ret < 0) { goto error; } @@ -1439,7 +1505,7 @@ end: * Returns LTTNG_OK on success or else an LTTng error code. */ enum lttng_error_code consumer_snapshot_channel(struct consumer_socket *socket, - uint64_t key, const struct snapshot_output *output, int metadata, + uint64_t key, const struct consumer_output *output, int metadata, uid_t uid, gid_t gid, const char *channel_path, int wait, uint64_t nb_packets_per_stream) { @@ -1449,7 +1515,6 @@ enum lttng_error_code consumer_snapshot_channel(struct consumer_socket *socket, assert(socket); assert(output); - assert(output->consumer); DBG("Consumer snapshot channel key %" PRIu64, key); @@ -1459,9 +1524,9 @@ enum lttng_error_code consumer_snapshot_channel(struct consumer_socket *socket, msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream; msg.u.snapshot_channel.metadata = metadata; - if (output->consumer->type == CONSUMER_DST_NET) { + if (output->type == CONSUMER_DST_NET) { msg.u.snapshot_channel.relayd_id = - output->consumer->net_seq_index; + output->net_seq_index; msg.u.snapshot_channel.use_relayd = 1; } else { msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL; @@ -1662,6 +1727,34 @@ error: return ret; } +int consumer_clear_channel(struct consumer_socket *socket, uint64_t key) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(socket); + + DBG("Consumer clear channel %" PRIu64, key); + + memset(&msg, 0, sizeof(msg)); + msg.cmd_type = LTTNG_CONSUMER_CLEAR_CHANNEL; + msg.u.clear_channel.key = key; + + health_code_update(); + + pthread_mutex_lock(socket->lock); + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto error_socket; + } + +error_socket: + pthread_mutex_unlock(socket->lock); + + health_code_update(); + return ret; +} + int consumer_init(struct consumer_socket *socket, const lttng_uuid sessiond_uuid) { @@ -1693,15 +1786,17 @@ error: */ int consumer_create_trace_chunk(struct consumer_socket *socket, uint64_t relayd_id, uint64_t session_id, - struct lttng_trace_chunk *chunk) + struct lttng_trace_chunk *chunk, + const char *domain_subdir) { int ret; enum lttng_trace_chunk_status chunk_status; struct lttng_credentials chunk_credentials; - const struct lttng_directory_handle *chunk_directory_handle; - int chunk_dirfd; + const struct lttng_directory_handle *chunk_directory_handle = NULL; + struct lttng_directory_handle *domain_handle = NULL; + int domain_dirfd; const char *chunk_name; - bool chunk_name_overriden; + bool chunk_name_overridden; uint64_t chunk_id; time_t creation_timestamp; char creation_timestamp_buffer[ISO8601_STR_LEN]; @@ -1721,14 +1816,14 @@ int consumer_create_trace_chunk(struct consumer_socket *socket, } chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name, - &chunk_name_overriden); + &chunk_name_overridden); if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK && chunk_status != LTTNG_TRACE_CHUNK_STATUS_NONE) { ERR("Failed to get name of trace chunk"); ret = -LTTNG_ERR_FATAL; goto error; } - if (chunk_name_overriden) { + if (chunk_name_overridden) { ret = lttng_strncpy(msg.u.create_trace_chunk.override_name, chunk_name, sizeof(msg.u.create_trace_chunk.override_name)); @@ -1768,25 +1863,12 @@ int consumer_create_trace_chunk(struct consumer_socket *socket, msg.u.create_trace_chunk.chunk_id = chunk_id; if (chunk_has_local_output) { - chunk_status = lttng_trace_chunk_get_chunk_directory_handle( + chunk_status = lttng_trace_chunk_borrow_chunk_directory_handle( chunk, &chunk_directory_handle); if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { ret = -LTTNG_ERR_FATAL; goto error; } - - /* - * This will only compile on platforms that support - * dirfd (POSIX.2008). This is fine as the session daemon - * is only built for such platforms. - * - * The ownership of the chunk directory handle's is maintained - * by the trace chunk. - */ - chunk_dirfd = lttng_directory_handle_get_dirfd( - chunk_directory_handle); - assert(chunk_dirfd >= 0); - chunk_status = lttng_trace_chunk_get_credentials( chunk, &chunk_credentials); if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { @@ -1797,6 +1879,37 @@ int consumer_create_trace_chunk(struct consumer_socket *socket, ret = -LTTNG_ERR_FATAL; goto error; } + ret = lttng_directory_handle_create_subdirectory_as_user( + chunk_directory_handle, + domain_subdir, + S_IRWXU | S_IRWXG, + &chunk_credentials); + if (ret) { + PERROR("Failed to create chunk domain output directory \"%s\"", + domain_subdir); + ret = -LTTNG_ERR_FATAL; + goto error; + } + domain_handle = lttng_directory_handle_create_from_handle( + domain_subdir, + chunk_directory_handle); + if (!domain_handle) { + ret = -LTTNG_ERR_FATAL; + goto error; + } + + /* + * This will only compile on platforms that support + * dirfd (POSIX.2008). This is fine as the session daemon + * is only built for such platforms. + * + * The ownership of the chunk directory handle's is maintained + * by the trace chunk. + */ + domain_dirfd = lttng_directory_handle_get_dirfd( + domain_handle); + assert(domain_dirfd >= 0); + msg.u.create_trace_chunk.credentials.value.uid = chunk_credentials.uid; msg.u.create_trace_chunk.credentials.value.gid = @@ -1819,9 +1932,9 @@ int consumer_create_trace_chunk(struct consumer_socket *socket, } if (chunk_has_local_output) { - DBG("Sending trace chunk directory fd to consumer"); + DBG("Sending trace chunk domain directory fd to consumer"); health_code_update(); - ret = consumer_send_fds(socket, &chunk_dirfd, 1); + ret = consumer_send_fds(socket, &domain_dirfd, 1); health_code_update(); if (ret < 0) { ERR("Trace chunk creation error on consumer"); @@ -1830,6 +1943,7 @@ int consumer_create_trace_chunk(struct consumer_socket *socket, } } error: + lttng_directory_handle_put(domain_handle); return ret; } @@ -1840,7 +1954,8 @@ error: */ int consumer_close_trace_chunk(struct consumer_socket *socket, uint64_t relayd_id, uint64_t session_id, - struct lttng_trace_chunk *chunk) + struct lttng_trace_chunk *chunk, + char *closed_trace_chunk_path) { int ret; enum lttng_trace_chunk_status chunk_status; @@ -1848,12 +1963,15 @@ int consumer_close_trace_chunk(struct consumer_socket *socket, .cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK, .u.close_trace_chunk.session_id = session_id, }; + struct lttcomm_consumer_close_trace_chunk_reply reply; uint64_t chunk_id; time_t close_timestamp; enum lttng_trace_chunk_command_type close_command; const char *close_command_name = "none"; + struct lttng_dynamic_buffer path_reception_buffer; assert(socket); + lttng_dynamic_buffer_init(&path_reception_buffer); if (relayd_id != -1ULL) { LTTNG_OPTIONAL_SET( @@ -1904,13 +2022,51 @@ int consumer_close_trace_chunk(struct consumer_socket *socket, relayd_id, session_id, chunk_id, close_command_name); health_code_update(); - ret = consumer_send_msg(socket, &msg); + ret = consumer_socket_send(socket, &msg, sizeof(struct lttcomm_consumer_msg)); if (ret < 0) { ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER; goto error; } - + ret = consumer_socket_recv(socket, &reply, sizeof(reply)); + if (ret < 0) { + ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER; + goto error; + } + if (reply.path_length >= LTTNG_PATH_MAX) { + ERR("Invalid path returned by relay daemon: %" PRIu32 "bytes exceeds maximal allowed length of %d bytes", + reply.path_length, LTTNG_PATH_MAX); + ret = -LTTNG_ERR_INVALID_PROTOCOL; + goto error; + } + ret = lttng_dynamic_buffer_set_size(&path_reception_buffer, + reply.path_length); + if (ret) { + ERR("Failed to allocate reception buffer of path returned by the \"close trace chunk\" command"); + ret = -LTTNG_ERR_NOMEM; + goto error; + } + ret = consumer_socket_recv(socket, path_reception_buffer.data, + path_reception_buffer.size); + if (ret < 0) { + ERR("Communication error while receiving path of closed trace chunk"); + ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER; + goto error; + } + if (path_reception_buffer.data[path_reception_buffer.size - 1] != '\0') { + ERR("Invalid path returned by relay daemon: not null-terminated"); + ret = -LTTNG_ERR_INVALID_PROTOCOL; + goto error; + } + if (closed_trace_chunk_path) { + /* + * closed_trace_chunk_path is assumed to have a length >= + * LTTNG_PATH_MAX + */ + memcpy(closed_trace_chunk_path, path_reception_buffer.data, + path_reception_buffer.size); + } error: + lttng_dynamic_buffer_reset(&path_reception_buffer); health_code_update(); return ret; }