X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=62a5bd7cefc16026ff05f3e3741e4b4258837fd1;hp=9589b6f52bdeed68ea62111ad7e7d2f88d7c2b77;hb=9898f786493c5e60dcf99fa3102bb95752ad73ed;hpb=30b9d5ab8a79cfa37c4be6b5bf3193e2a8501bb8 diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 9589b6f52..62a5bd7ce 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -40,6 +40,7 @@ #include #include #include +#include #include #include @@ -1068,16 +1069,21 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, int ret = 0; ssize_t send_ret; struct relay_session *session = NULL; - struct lttcomm_relayd_status_session reply = {}; + struct lttcomm_relayd_create_session_reply_2_11 reply = {}; char session_name[LTTNG_NAME_MAX] = {}; char hostname[LTTNG_HOST_NAME_MAX] = {}; uint32_t live_timer = 0; bool snapshot = false; + bool session_name_contains_creation_timestamp = false; /* Left nil for peers < 2.11. */ + char base_path[LTTNG_PATH_MAX] = {}; lttng_uuid sessiond_uuid = {}; LTTNG_OPTIONAL(uint64_t) id_sessiond = {}; LTTNG_OPTIONAL(uint64_t) current_chunk_id = {}; LTTNG_OPTIONAL(time_t) creation_time = {}; + struct lttng_dynamic_buffer reply_payload; + + lttng_dynamic_buffer_init(&reply_payload); if (conn->minor < 4) { /* From 2.1 to 2.3 */ @@ -1094,9 +1100,10 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, /* From 2.11 to ... */ ret = cmd_create_session_2_11(payload, session_name, hostname, - &live_timer, &snapshot, &id_sessiond_value, + base_path, &live_timer, &snapshot, &id_sessiond_value, sessiond_uuid, &has_current_chunk, - ¤t_chunk_id_value, &creation_time_value); + ¤t_chunk_id_value, &creation_time_value, + &session_name_contains_creation_timestamp); if (lttng_uuid_is_nil(sessiond_uuid)) { /* The nil UUID is reserved for pre-2.11 clients. */ ERR("Illegal nil UUID announced by peer in create session command"); @@ -1115,12 +1122,13 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, goto send_reply; } - session = session_create(session_name, hostname, live_timer, + session = session_create(session_name, hostname, base_path, live_timer, snapshot, sessiond_uuid, id_sessiond.is_set ? &id_sessiond.value : NULL, current_chunk_id.is_set ? ¤t_chunk_id.value : NULL, creation_time.is_set ? &creation_time.value : NULL, - conn->major, conn->minor); + conn->major, conn->minor, + session_name_contains_creation_timestamp); if (!session) { ret = -1; goto send_reply; @@ -1129,24 +1137,59 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, conn->session = session; DBG("Created session %" PRIu64, session->id); - reply.session_id = htobe64(session->id); + reply.generic.session_id = htobe64(session->id); send_reply: if (ret < 0) { - reply.ret_code = htobe32(LTTNG_ERR_FATAL); + reply.generic.ret_code = htobe32(LTTNG_ERR_FATAL); } else { - reply.ret_code = htobe32(LTTNG_OK); + reply.generic.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 \"create session\" command reply (ret = %zd)", - send_ret); + if (conn->minor < 11) { + /* From 2.1 to 2.10 */ + ret = lttng_dynamic_buffer_append(&reply_payload, + &reply.generic, sizeof(reply.generic)); + if (ret) { + ERR("Failed to append \"create session\" command reply header to payload buffer"); + ret = -1; + goto end; + } + } else { + const uint32_t output_path_length = + session ? strlen(session->output_path) + 1 : 0; + + reply.output_path_length = htobe32(output_path_length); + ret = lttng_dynamic_buffer_append( + &reply_payload, &reply, sizeof(reply)); + if (ret) { + ERR("Failed to append \"create session\" command reply header to payload buffer"); + goto end; + } + + if (output_path_length) { + ret = lttng_dynamic_buffer_append(&reply_payload, + session->output_path, + output_path_length); + if (ret) { + ERR("Failed to append \"create session\" command reply path to payload buffer"); + goto end; + } + } + } + + send_ret = conn->sock->ops->sendmsg(conn->sock, reply_payload.data, + reply_payload.size, 0); + if (send_ret < (ssize_t) reply_payload.size) { + ERR("Failed to send \"create session\" command reply of %zu bytes (ret = %zd)", + reply_payload.size, send_ret); ret = -1; } +end: if (ret < 0 && session) { session_put(session); } + lttng_dynamic_buffer_reset(&reply_payload); return ret; } @@ -2191,6 +2234,7 @@ static int relay_rotate_session_streams( } else { chunk_id_str = chunk_id_buf; } + session->has_rotated = true; } DBG("Rotate %" PRIu32 " streams of session \"%s\" to chunk \"%s\"", @@ -2258,61 +2302,7 @@ end_no_reply: return ret; } -static int init_session_output_directory_handle(struct relay_session *session, - struct lttng_directory_handle *handle) -{ - int ret; - /* hostname/session_name */ - char *session_directory = NULL; - /* - * base path + session_directory - * e.g. /home/user/lttng-traces/hostname/session_name - */ - char *full_session_path = NULL; - char creation_time_str[16]; - struct tm *timeinfo; - - assert(session->creation_time.is_set); - timeinfo = localtime(&session->creation_time.value); - if (!timeinfo) { - ret = -1; - goto end; - } - strftime(creation_time_str, sizeof(creation_time_str), "%Y%m%d-%H%M%S", - timeinfo); - pthread_mutex_lock(&session->lock); - ret = asprintf(&session_directory, "%s/%s-%s", session->hostname, - session->session_name, creation_time_str); - pthread_mutex_unlock(&session->lock); - if (ret < 0) { - PERROR("Failed to format session directory name"); - goto end; - } - - full_session_path = create_output_path(session_directory); - if (!full_session_path) { - ret = -1; - goto end; - } - - ret = utils_mkdir_recursive( - full_session_path, S_IRWXU | S_IRWXG, -1, -1); - if (ret) { - ERR("Failed to create session output path \"%s\"", - full_session_path); - goto end; - } - - ret = lttng_directory_handle_init(handle, full_session_path); - if (ret) { - goto end; - } -end: - free(session_directory); - free(full_session_path); - return ret; -} /* * relay_create_trace_chunk: create a new trace chunk @@ -2399,13 +2389,6 @@ static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, } } - ret = init_session_output_directory_handle( - conn->session, &session_output); - if (ret) { - reply_code = LTTNG_ERR_CREATE_DIR_FAIL; - goto end; - } - chunk_status = lttng_trace_chunk_set_credentials_current_user(chunk); if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { reply_code = LTTNG_ERR_UNK; @@ -2413,7 +2396,14 @@ static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, goto end; } + ret = session_init_output_directory_handle( + conn->session, &session_output); + if (ret) { + reply_code = LTTNG_ERR_CREATE_DIR_FAIL; + goto end; + } chunk_status = lttng_trace_chunk_set_as_owner(chunk, &session_output); + lttng_directory_handle_fini(&session_output); if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { reply_code = LTTNG_ERR_UNK; ret = -1; @@ -2469,7 +2459,6 @@ end: end_no_reply: lttng_trace_chunk_put(chunk); lttng_trace_chunk_put(published_chunk); - lttng_directory_handle_fini(&session_output); return ret; } @@ -2480,11 +2469,11 @@ static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, struct relay_connection *conn, const struct lttng_buffer_view *payload) { - int ret = 0; + int ret = 0, buf_ret; ssize_t send_ret; struct relay_session *session = conn->session; struct lttcomm_relayd_close_trace_chunk *msg; - struct lttcomm_relayd_generic_reply reply = {}; + struct lttcomm_relayd_close_trace_chunk_reply reply = {}; struct lttng_buffer_view header_view; struct lttng_trace_chunk *chunk = NULL; enum lttng_error_code reply_code = LTTNG_OK; @@ -2492,6 +2481,12 @@ static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, uint64_t chunk_id; LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command = {}; time_t close_timestamp; + char closed_trace_chunk_path[LTTNG_PATH_MAX]; + size_t path_length = 0; + const char *chunk_name = NULL; + struct lttng_dynamic_buffer reply_payload; + + lttng_dynamic_buffer_init(&reply_payload); if (!session || !conn->version_check_done) { ERR("Trying to close a trace chunk before version check"); @@ -2567,6 +2562,53 @@ static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, goto end_unlock_session; } } + chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name, NULL); + if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { + ERR("Failed to get chunk name"); + ret = -1; + reply_code = LTTNG_ERR_UNK; + goto end_unlock_session; + } + if (!session->has_rotated && !session->snapshot) { + ret = lttng_strncpy(closed_trace_chunk_path, + session->output_path, + sizeof(closed_trace_chunk_path)); + if (ret) { + ERR("Failed to send trace chunk path: path length of %zu bytes exceeds the maximal allowed length of %zu bytes", + strlen(session->output_path), + sizeof(closed_trace_chunk_path)); + reply_code = LTTNG_ERR_NOMEM; + ret = -1; + goto end_unlock_session; + } + } else { + if (session->snapshot) { + ret = snprintf(closed_trace_chunk_path, + sizeof(closed_trace_chunk_path), + "%s/%s", session->output_path, + chunk_name); + } else { + ret = snprintf(closed_trace_chunk_path, + sizeof(closed_trace_chunk_path), + "%s/" DEFAULT_ARCHIVED_TRACE_CHUNKS_DIRECTORY + "/%s", + session->output_path, chunk_name); + } + if (ret < 0 || ret == sizeof(closed_trace_chunk_path)) { + ERR("Failed to format closed trace chunk resulting path"); + reply_code = ret < 0 ? LTTNG_ERR_UNK : LTTNG_ERR_NOMEM; + ret = -1; + goto end_unlock_session; + } + } + DBG("Reply chunk path on close: %s", closed_trace_chunk_path); + path_length = strlen(closed_trace_chunk_path) + 1; + if (path_length > UINT32_MAX) { + ERR("Closed trace chunk path exceeds the maximal length allowed by the protocol"); + ret = -1; + reply_code = LTTNG_ERR_INVALID_PROTOCOL; + goto end_unlock_session; + } if (session->current_trace_chunk == chunk) { /* @@ -2586,18 +2628,37 @@ end_unlock_session: pthread_mutex_unlock(&session->lock); end: - reply.ret_code = htobe32((uint32_t) reply_code); + reply.generic.ret_code = htobe32((uint32_t) reply_code); + reply.path_length = htobe32((uint32_t) path_length); + buf_ret = lttng_dynamic_buffer_append( + &reply_payload, &reply, sizeof(reply)); + if (buf_ret) { + ERR("Failed to append \"close trace chunk\" command reply header to payload buffer"); + goto end_no_reply; + } + + if (reply_code == LTTNG_OK) { + buf_ret = lttng_dynamic_buffer_append(&reply_payload, + closed_trace_chunk_path, path_length); + if (buf_ret) { + ERR("Failed to append \"close trace chunk\" command reply path to payload buffer"); + goto end_no_reply; + } + } + send_ret = conn->sock->ops->sendmsg(conn->sock, - &reply, - sizeof(struct lttcomm_relayd_generic_reply), + reply_payload.data, + reply_payload.size, 0); - if (send_ret < (ssize_t) sizeof(reply)) { - ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)", - send_ret); + if (send_ret < reply_payload.size) { + ERR("Failed to send \"close trace chunk\" command reply of %zu bytes (ret = %zd)", + reply_payload.size, send_ret); ret = -1; + goto end_no_reply; } end_no_reply: lttng_trace_chunk_put(chunk); + lttng_dynamic_buffer_reset(&reply_payload); return ret; }