X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=db021f43fc5b28a9ac0de322752880a3c9a171c9;hb=96b0ecff3ade3c7970d1cc5fb12a50a94c67ee67;hp=282bb8461b1f97a595958cd75504fdf86fb3b322;hpb=639ddf685b134bb075b92819f647f9f3c462df54;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 282bb8461..db021f43f 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -2665,365 +2665,6 @@ end_no_reply: return ret; } -/* - * relay_mkdir: Create a folder on the disk. - */ -static int relay_mkdir(const struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn, - const struct lttng_buffer_view *payload) -{ - int ret; - struct relay_session *session = conn->session; - struct lttcomm_relayd_mkdir path_info_header; - struct lttcomm_relayd_generic_reply reply; - char *path = NULL; - size_t header_len; - ssize_t send_ret; - struct lttng_buffer_view path_view; - - if (!session || !conn->version_check_done) { - ERR("Trying to create a directory before version check"); - ret = -1; - goto end_no_session; - } - - if (session->major == 2 && session->minor < 11) { - /* - * This client is not supposed to use this command since - * it predates its introduction. - */ - ERR("relay_mkdir command is unsupported before LTTng 2.11"); - ret = -1; - goto end_no_session; - } - - header_len = sizeof(path_info_header); - if (payload->size < header_len) { - ERR("Unexpected payload size in \"relay_mkdir\": expected >= %zu bytes, got %zu bytes", - header_len, payload->size); - ret = -1; - goto end_no_session; - } - - memcpy(&path_info_header, payload->data, header_len); - - path_info_header.length = be32toh(path_info_header.length); - - if (payload->size < header_len + path_info_header.length) { - ERR("Unexpected payload size in \"relay_mkdir\" including path: expected >= %zu bytes, got %zu bytes", - header_len + path_info_header.length, payload->size); - ret = -1; - goto end_no_session; - } - - /* Ensure that it fits in local path length. */ - if (path_info_header.length >= LTTNG_PATH_MAX) { - ret = -ENAMETOOLONG; - ERR("Path name argument of mkdir command (%" PRIu32 " bytes) exceeds the maximal length allowed (%d bytes)", - path_info_header.length, LTTNG_PATH_MAX); - goto end; - } - - path_view = lttng_buffer_view_from_view(payload, header_len, - path_info_header.length); - - path = create_output_path(path_view.data); - if (!path) { - ERR("Failed to create output path"); - ret = -1; - goto end; - } - - DBG("MKDIR command has path \"%s\", changed to \"%s\"", path_view.data, path); - ret = utils_mkdir_recursive(path, S_IRWXU | S_IRWXG, -1, -1); - if (ret < 0) { - ERR("relay creating output directory"); - goto end; - } - - ret = 0; - -end: - memset(&reply, 0, sizeof(reply)); - if (ret < 0) { - reply.ret_code = htobe32(LTTNG_ERR_UNK); - } else { - reply.ret_code = htobe32(LTTNG_OK); - } - send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0); - if (send_ret < (ssize_t) sizeof(reply)) { - ERR("Failed to send \"mkdir\" command reply (ret = %zd)", send_ret); - ret = -1; - } - -end_no_session: - free(path); - return ret; -} - -static int validate_rotate_rename_path_length(const char *path_type, - uint32_t path_length) -{ - int ret = 0; - - if (path_length > LTTNG_PATH_MAX) { - ret = -ENAMETOOLONG; - ERR("rotate rename \"%s\" path name length (%" PRIu32 " bytes) exceeds the allowed size of %i bytes", - path_type, path_length, LTTNG_PATH_MAX); - } else if (path_length == 0) { - ret = -EINVAL; - ERR("rotate rename \"%s\" path name has an illegal length of 0", path_type); - } - return ret; -} - -/* - * relay_rotate_rename: rename the trace folder after a rotation is - * completed. We are not closing any fd here, just moving the folder, so it - * works even if data is still in-flight. - */ -static int relay_rotate_rename(const struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn, - const struct lttng_buffer_view *payload) -{ - int ret; - ssize_t send_ret; - struct relay_session *session = conn->session; - struct lttcomm_relayd_generic_reply reply; - struct lttcomm_relayd_rotate_rename header; - size_t header_len; - size_t received_paths_size; - char *complete_old_path = NULL, *complete_new_path = NULL; - struct lttng_buffer_view old_path_view; - struct lttng_buffer_view new_path_view; - - if (!session || !conn->version_check_done) { - ERR("Trying to rename a trace folder before version check"); - ret = -1; - goto end_no_reply; - } - - if (session->major == 2 && session->minor < 11) { - ERR("relay_rotate_rename command is unsupported before LTTng 2.11"); - ret = -1; - goto end_no_reply; - } - - header_len = sizeof(header); - if (payload->size < header_len) { - ERR("Unexpected payload size in \"relay_rotate_rename\": expected >= %zu bytes, got %zu bytes", - header_len, payload->size); - ret = -1; - goto end_no_reply; - } - - memcpy(&header, payload->data, header_len); - - header.old_path_length = be32toh(header.old_path_length); - header.new_path_length = be32toh(header.new_path_length); - received_paths_size = header.old_path_length + header.new_path_length; - - if (payload->size < header_len + received_paths_size) { - ERR("Unexpected payload size in \"relay_rotate_rename\" including paths: expected >= %zu bytes, got %zu bytes", - header_len, payload->size); - ret = -1; - goto end_no_reply; - } - - /* Ensure the paths don't exceed their allowed size. */ - ret = validate_rotate_rename_path_length("old", header.old_path_length); - if (ret) { - goto end; - } - ret = validate_rotate_rename_path_length("new", header.new_path_length); - if (ret) { - goto end; - } - - old_path_view = lttng_buffer_view_from_view(payload, header_len, - header.old_path_length); - new_path_view = lttng_buffer_view_from_view(payload, - header_len + header.old_path_length, - header.new_path_length); - - /* Validate that both paths received are NULL terminated. */ - if (old_path_view.data[old_path_view.size - 1] != '\0') { - ERR("relay_rotate_rename command's \"old\" path is invalid (not NULL terminated)"); - ret = -1; - goto end; - } - if (new_path_view.data[new_path_view.size - 1] != '\0') { - ERR("relay_rotate_rename command's \"new\" path is invalid (not NULL terminated)"); - ret = -1; - goto end; - } - - DBG("ROTATE_RENAME command has argument old path = \"%s\", new_path = \"%s\"", - old_path_view.data, new_path_view.data); - complete_old_path = create_output_path(old_path_view.data); - if (!complete_old_path) { - ERR("Failed to build old output path in rotate_rename command"); - ret = -1; - goto end; - } - - complete_new_path = create_output_path(new_path_view.data); - if (!complete_new_path) { - ERR("Failed to build new output path in rotate_rename command"); - ret = -1; - goto end; - } - DBG("Expanded ROTATE_RENAME arguments to old path = \"%s\", new_path = \"%s\"", - complete_old_path, complete_new_path); - - ret = utils_mkdir_recursive(complete_new_path, S_IRWXU | S_IRWXG, - -1, -1); - if (ret < 0) { - ERR("Failed to mkdir() rotate_rename's \"new\" output directory at \"%s\"", - complete_new_path); - goto end; - } - - /* - * If a domain has not yet created its channel, the domain-specific - * folder might not exist, but this is not an error. - */ - ret = rename(complete_old_path, complete_new_path); - if (ret < 0 && errno != ENOENT) { - PERROR("Renaming chunk in rotate_rename command from \"%s\" to \"%s\"", - complete_old_path, complete_new_path); - goto end; - } - ret = 0; - -end: - memset(&reply, 0, sizeof(reply)); - if (ret < 0) { - reply.ret_code = htobe32(LTTNG_ERR_UNK); - } else { - reply.ret_code = htobe32(LTTNG_OK); - } - send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, - sizeof(reply), 0); - if (send_ret < sizeof(reply)) { - ERR("Failed to send \"rotate rename\" command reply (ret = %zd)", - send_ret); - ret = -1; - } - -end_no_reply: - free(complete_old_path); - free(complete_new_path); - return ret; -} - -/* - * Check if all the streams in the session have completed the last rotation. - * The chunk_id value is used to distinguish the cases where a stream was - * closed on the consumerd before the rotation started but it still active on - * the relayd, and the case where a stream appeared on the consumerd/relayd - * after the last rotation started (in that case, it is already writing in the - * new chunk folder). - */ -static -int relay_rotate_pending(const struct lttcomm_relayd_hdr *recv_hdr, - struct relay_connection *conn, - const struct lttng_buffer_view *payload) -{ - struct relay_session *session = conn->session; - struct lttcomm_relayd_rotate_pending msg; - struct lttcomm_relayd_rotate_pending_reply reply; - struct lttng_ht_iter iter; - struct relay_stream *stream; - int ret = 0; - ssize_t send_ret; - uint64_t chunk_id; - bool rotate_pending = false; - - DBG("Rotate pending command received"); - - if (!session || !conn->version_check_done) { - ERR("Trying to check for data before version check"); - ret = -1; - goto end_no_reply; - } - - if (session->major == 2 && session->minor < 11) { - ERR("Unsupported feature before 2.11"); - ret = -1; - goto end_no_reply; - } - - if (payload->size < sizeof(msg)) { - ERR("Unexpected payload size in \"relay_rotate_pending\": expected >= %zu bytes, got %zu bytes", - sizeof(msg), payload->size); - ret = -1; - goto end_no_reply; - } - - memcpy(&msg, payload->data, sizeof(msg)); - - chunk_id = be64toh(msg.chunk_id); - - DBG("Evaluating rotate pending for session \"%s\" and chunk id %" PRIu64, - session->session_name, chunk_id); - - /* - * Iterate over all the streams in the session and check if they are - * still waiting for data to perform their rotation. - */ - rcu_read_lock(); - cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream, - node.node) { - if (!stream_get(stream)) { - continue; - } - if (stream->trace->session != session) { - stream_put(stream); - continue; - } - pthread_mutex_lock(&stream->lock); - if (stream->rotate_at_seq_num != -1ULL) { - /* We have not yet performed the rotation. */ - rotate_pending = true; - DBG("Stream %" PRIu64 " is still rotating", - stream->stream_handle); - } else if (stream->current_chunk_id.value <= chunk_id) { - /* - * Stream closed on the consumer but still active on the - * relay. - */ - rotate_pending = true; - DBG("Stream %" PRIu64 " did not exist on the consumer " - "when the last rotation started, but is" - "still waiting for data before getting" - "closed", - stream->stream_handle); - } - pthread_mutex_unlock(&stream->lock); - stream_put(stream); - if (rotate_pending) { - goto send_reply; - } - } - -send_reply: - rcu_read_unlock(); - memset(&reply, 0, sizeof(reply)); - reply.generic.ret_code = htobe32((uint32_t) LTTNG_OK); - reply.is_pending = (uint8_t) !!rotate_pending; - send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, - sizeof(reply), 0); - if (send_ret < (ssize_t) sizeof(reply)) { - ERR("Failed to send \"rotate pending\" command reply (ret = %zd)", - send_ret); - ret = -1; - } - -end_no_reply: - return ret; -} - #define DBG_CMD(cmd_name, conn) \ DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd); @@ -3090,18 +2731,6 @@ static int relay_process_control_command(struct relay_connection *conn, DBG_CMD("RELAYD_ROTATE_STREAM", conn); ret = relay_rotate_session_stream(header, conn, payload); break; - case RELAYD_ROTATE_RENAME: - DBG_CMD("RELAYD_ROTATE_RENAME", conn); - ret = relay_rotate_rename(header, conn, payload); - break; - case RELAYD_ROTATE_PENDING: - DBG_CMD("RELAYD_ROTATE_PENDING", conn); - ret = relay_rotate_pending(header, conn, payload); - break; - case RELAYD_MKDIR: - DBG_CMD("RELAYD_MKDIR", conn); - ret = relay_mkdir(header, conn, payload); - break; case RELAYD_UPDATE_SYNC_INFO: default: ERR("Received unknown command (%u)", header->cmd);