int ret = 0;
ssize_t send_ret;
struct relay_session *session = NULL;
- struct lttcomm_relayd_status_session reply;
- char session_name[LTTNG_NAME_MAX];
- char hostname[LTTNG_HOST_NAME_MAX];
+ struct lttcomm_relayd_status_session reply = {};
+ char session_name[LTTNG_NAME_MAX] = {};
+ char hostname[LTTNG_HOST_NAME_MAX] = {};
uint32_t live_timer = 0;
bool snapshot = false;
/* Left nil for peers < 2.11. */
lttng_uuid sessiond_uuid = {};
-
- memset(session_name, 0, LTTNG_NAME_MAX);
- memset(hostname, 0, LTTNG_HOST_NAME_MAX);
-
- memset(&reply, 0, sizeof(reply));
+ LTTNG_OPTIONAL(uint64_t) id_sessiond = {};
+ LTTNG_OPTIONAL(uint64_t) current_chunk_id = {};
if (conn->minor < 4) {
/* From 2.1 to 2.3 */
ret = cmd_create_session_2_4(payload, session_name,
hostname, &live_timer, &snapshot);
} else {
+ bool has_current_chunk;
+
/* From 2.11 to ... */
ret = cmd_create_session_2_11(payload, session_name,
hostname, &live_timer, &snapshot,
- sessiond_uuid);
+ &id_sessiond.value, sessiond_uuid,
+ &has_current_chunk,
+ ¤t_chunk_id.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;
}
if (ret < 0) {
}
session = session_create(session_name, hostname, live_timer,
- snapshot, sessiond_uuid, conn->major, conn->minor);
+ snapshot, sessiond_uuid,
+ id_sessiond.is_set ? &id_sessiond.value : NULL,
+ current_chunk_id.is_set ? ¤t_chunk_id.value : NULL,
+ conn->major, conn->minor);
if (!session) {
ret = -1;
goto send_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);
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);
projects / lttng-tools.git / blobdiff