X-Git-Url: http://git.lttng.org./?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=c0e9d76dfd79550d920a021e3f2b3018cf387483;hb=e9b7c75317d46e1c4f8fb93afbb6ae6ed7672179;hp=11277f29346e99104cfbedd6453ace99a76493f6;hpb=0ebdafe0e5c235d2e4ce6d4dcff48bfd9a77a479;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 11277f293..c0e9d76df 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -1104,6 +1104,7 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, lttng_uuid sessiond_uuid = {}; LTTNG_OPTIONAL(uint64_t) id_sessiond = {}; LTTNG_OPTIONAL(uint64_t) current_chunk_id = {}; + LTTNG_OPTIONAL(time_t) creation_time = {}; if (conn->minor < 4) { /* From 2.1 to 2.3 */ @@ -1114,21 +1115,27 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, hostname, &live_timer, &snapshot); } else { bool has_current_chunk; + uint64_t current_chunk_id_value; + time_t creation_time_value; + uint64_t id_sessiond_value; /* From 2.11 to ... */ - ret = cmd_create_session_2_11(payload, session_name, - hostname, &live_timer, &snapshot, - &id_sessiond.value, sessiond_uuid, - &has_current_chunk, - ¤t_chunk_id.value); + ret = cmd_create_session_2_11(payload, session_name, hostname, + &live_timer, &snapshot, &id_sessiond_value, + sessiond_uuid, &has_current_chunk, + ¤t_chunk_id_value, &creation_time_value); 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"); ret = -1; goto send_reply; } - id_sessiond.is_set = true; - current_chunk_id.is_set = has_current_chunk; + LTTNG_OPTIONAL_SET(&id_sessiond, id_sessiond_value); + LTTNG_OPTIONAL_SET(&creation_time, creation_time_value); + if (has_current_chunk) { + LTTNG_OPTIONAL_SET(¤t_chunk_id, + current_chunk_id_value); + } } if (ret < 0) { @@ -1139,6 +1146,7 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, 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); if (!session) { ret = -1; @@ -2853,6 +2861,107 @@ end_no_reply: return ret; } +/* + * relay_close_trace_chunk: close a trace chunk + */ +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; + ssize_t send_ret; + struct relay_session *session = conn->session; + struct lttcomm_relayd_close_trace_chunk *msg; + struct lttcomm_relayd_generic_reply reply = {}; + struct lttng_buffer_view header_view; + struct lttng_trace_chunk *chunk = NULL; + enum lttng_error_code reply_code = LTTNG_OK; + enum lttng_trace_chunk_status chunk_status; + uint64_t chunk_id; + LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command; + time_t close_timestamp; + + if (!session || !conn->version_check_done) { + ERR("Trying to close a trace chunk before version check"); + ret = -1; + goto end_no_reply; + } + + if (session->major == 2 && session->minor < 11) { + ERR("Chunk close command is unsupported before 2.11"); + ret = -1; + goto end_no_reply; + } + + header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg)); + if (!header_view.data) { + ERR("Failed to receive payload of chunk close command"); + ret = -1; + goto end_no_reply; + } + + /* Convert to host endianness. */ + msg = (typeof(msg)) header_view.data; + chunk_id = be64toh(msg->chunk_id); + close_timestamp = (time_t) be64toh(msg->close_timestamp); + close_command = (typeof(close_command)){ + .value = be32toh(msg->close_command.value), + .is_set = msg->close_command.is_set, + }; + + chunk = sessiond_trace_chunk_registry_get_chunk( + sessiond_trace_chunk_registry, + conn->session->sessiond_uuid, + conn->session->id, + chunk_id); + if (!chunk) { + char uuid_str[UUID_STR_LEN]; + + lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str); + ERR("Failed to find chunk to close: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64, + uuid_str, + conn->session->id, + msg->chunk_id); + ret = -1; + reply_code = LTTNG_ERR_NOMEM; + goto end; + } + + chunk_status = lttng_trace_chunk_set_close_timestamp( + chunk, close_timestamp); + if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { + ERR("Failed to set trace chunk close timestamp"); + ret = -1; + reply_code = LTTNG_ERR_UNK; + goto end; + } + + if (close_command.is_set) { + chunk_status = lttng_trace_chunk_set_close_command( + chunk, close_command.value); + if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { + ret = -1; + reply_code = LTTNG_ERR_INVALID; + goto end; + } + } + +end: + reply.ret_code = htobe32((uint32_t) reply_code); + send_ret = conn->sock->ops->sendmsg(conn->sock, + &reply, + sizeof(struct lttcomm_relayd_generic_reply), + 0); + if (send_ret < (ssize_t) sizeof(reply)) { + ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)", + send_ret); + ret = -1; + } +end_no_reply: + lttng_trace_chunk_put(chunk); + return ret; +} + #define DBG_CMD(cmd_name, conn) \ DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd); @@ -2923,6 +3032,10 @@ static int relay_process_control_command(struct relay_connection *conn, DBG_CMD("RELAYD_CREATE_TRACE_CHUNK", conn); ret = relay_create_trace_chunk(header, conn, payload); break; + case RELAYD_CLOSE_TRACE_CHUNK: + DBG_CMD("RELAYD_CLOSE_TRACE_CHUNK", conn); + ret = relay_close_trace_chunk(header, conn, payload); + break; case RELAYD_UPDATE_SYNC_INFO: default: ERR("Received unknown command (%u)", header->cmd);