X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=e745702f85cd9cea776a1063d7dea6025849bd4a;hp=50d19b074cf09542eb1a757f68fa3aaaa135a1db;hb=51a4828f21385847870ec2b2499e2f3e3842c26e;hpb=d295668767ac8234e83984e1812d342d03293d88 diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index 50d19b074..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) { @@ -883,7 +941,8 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, int64_t blocking_timeout, const char *root_shm_path, const char *shm_path, - struct lttng_trace_chunk *trace_chunk) + struct lttng_trace_chunk *trace_chunk, + const struct lttng_credentials *buffer_credentials) { assert(msg); @@ -892,27 +951,16 @@ 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; - struct lttng_credentials chunk_credentials; chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id); assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK); LTTNG_OPTIONAL_SET(&msg->u.ask_channel.chunk_id, chunk_id); - - chunk_status = lttng_trace_chunk_get_credentials(trace_chunk, - &chunk_credentials); - assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK); - msg->u.ask_channel.buffer_credentials.uid = - chunk_credentials.uid; - msg->u.ask_channel.buffer_credentials.gid = - chunk_credentials.gid; } + msg->u.ask_channel.buffer_credentials.uid = buffer_credentials->uid; + msg->u.ask_channel.buffer_credentials.gid = buffer_credentials->gid; msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION; msg->u.ask_channel.subbuf_size = subbuf_size; @@ -1085,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 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; @@ -1103,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); + 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; @@ -1133,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; } @@ -1445,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) { @@ -1455,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); @@ -1465,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; @@ -1668,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) { @@ -1699,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]; @@ -1727,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)); @@ -1772,15 +1861,42 @@ int consumer_create_trace_chunk(struct consumer_socket *socket, goto error; } msg.u.create_trace_chunk.chunk_id = chunk_id; - /* Only used for logging purposes. */ 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; } + chunk_status = lttng_trace_chunk_get_credentials( + chunk, &chunk_credentials); + if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { + /* + * Not associating credentials to a sessiond chunk is a + * fatal internal error. + */ + 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 @@ -1790,23 +1906,16 @@ int consumer_create_trace_chunk(struct consumer_socket *socket, * 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); - } + domain_dirfd = lttng_directory_handle_get_dirfd( + domain_handle); + assert(domain_dirfd >= 0); - chunk_status = lttng_trace_chunk_get_credentials(chunk, - &chunk_credentials); - if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { - /* - * Not associating credentials to a sessiond chunk is a fatal - * internal error. - */ - ret = -LTTNG_ERR_FATAL; - goto error; + msg.u.create_trace_chunk.credentials.value.uid = + chunk_credentials.uid; + msg.u.create_trace_chunk.credentials.value.gid = + chunk_credentials.gid; + msg.u.create_trace_chunk.credentials.is_set = 1; } - msg.u.create_trace_chunk.credentials.uid = chunk_credentials.uid; - msg.u.create_trace_chunk.credentials.gid = chunk_credentials.gid; DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64 @@ -1823,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"); @@ -1834,6 +1943,7 @@ int consumer_create_trace_chunk(struct consumer_socket *socket, } } error: + lttng_directory_handle_put(domain_handle); return ret; } @@ -1844,22 +1954,43 @@ 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; struct lttcomm_consumer_msg msg = { - .cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK, - .u.close_trace_chunk.session_id = session_id, + .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(&msg.u.close_trace_chunk.relayd_id, - relayd_id); + LTTNG_OPTIONAL_SET( + &msg.u.close_trace_chunk.relayd_id, relayd_id); + } + + chunk_status = lttng_trace_chunk_get_close_command( + chunk, &close_command); + switch (chunk_status) { + case LTTNG_TRACE_CHUNK_STATUS_OK: + LTTNG_OPTIONAL_SET(&msg.u.close_trace_chunk.close_command, + (uint32_t) close_command); + break; + case LTTNG_TRACE_CHUNK_STATUS_NONE: + break; + default: + ERR("Failed to get trace chunk close command"); + ret = -1; + goto error; } chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id); @@ -1881,19 +2012,61 @@ int consumer_close_trace_chunk(struct consumer_socket *socket, assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK); msg.u.close_trace_chunk.close_timestamp = (uint64_t) close_timestamp; + if (msg.u.close_trace_chunk.close_command.is_set) { + close_command_name = lttng_trace_chunk_command_type_get_name( + close_command); + } DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64 - ", session_id = %" PRIu64 - ", chunk_id = %" PRIu64, - relayd_id, session_id, chunk_id); + ", session_id = %" PRIu64 ", chunk_id = %" PRIu64 + ", close command = \"%s\"", + 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; }