+ goto end_no_session;
+ }
+
+ /*
+ * Publish every pending stream in the connection recv list which are
+ * now ready to be used by the viewer.
+ */
+ publish_connection_local_streams(conn);
+
+ memset(&reply, 0, sizeof(reply));
+ reply.ret_code = htobe32(LTTNG_OK);
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < 0) {
+ ERR("Relay sending sent_stream reply");
+ ret = send_ret;
+ } else {
+ /* Success. */
+ ret = 0;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * relay_rotate_stream: rotate a stream to a new tracefile for the session
+ * rotation feature (not the tracefile rotation feature).
+ */
+static int relay_rotate_session_stream(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret, send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_rotate_stream stream_info;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+ size_t len;
+ char *new_pathname = NULL;
+
+ DBG("Rotate stream received");
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to rotate a stream 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;
+ }
+
+ memset(&stream_info, 0, sizeof(struct lttcomm_relayd_rotate_stream));
+
+ /*
+ * Receive the struct up to the new_pathname member since we don't know
+ * its size yet.
+ */
+ ret = conn->sock->ops->recvmsg(conn->sock, &stream_info,
+ sizeof(struct lttcomm_relayd_rotate_stream), 0);
+ if (ret < sizeof(struct lttcomm_relayd_rotate_stream)) {
+ if (ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown", conn->sock->fd);
+ } else {
+ ERR("Relay didn't receive valid rotate_stream struct size : %d", ret);
+ }
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ stream = stream_get_by_id(be64toh(stream_info.stream_id));
+ if (!stream) {
+ ret = -1;
+ goto end;
+ }
+
+ len = be32toh(stream_info.pathname_length);
+ /* Ensure it fits in local filename length. */
+ if (len >= LTTNG_PATH_MAX) {
+ ret = -ENAMETOOLONG;
+ ERR("Length of relay_rotate_session_stream command's path name (%zu bytes) exceeds the maximal allowed length of %i bytes",
+ len, LTTNG_PATH_MAX);
+ goto end;
+ }
+
+ new_pathname = zmalloc(len);
+ if (!new_pathname) {
+ PERROR("Failed to allocation new path name of relay_rotate_session_stream command");
+ ret = -1;
+ goto end;
+ }
+
+ ret = conn->sock->ops->recvmsg(conn->sock, new_pathname, len, 0);
+ if (ret < len) {
+ if (ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown", conn->sock->fd);
+ } else {
+ ERR("Relay didn't receive valid rotate_stream struct size : %d", ret);
+ }
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ pthread_mutex_lock(&stream->lock);
+
+ /*
+ * Update the trace path (just the folder, the stream name does not
+ * change).
+ */
+ free(stream->path_name);
+ stream->path_name = create_output_path(new_pathname);
+ if (!stream->path_name) {
+ ERR("Failed to create a new output path");
+ goto end_stream_unlock;
+ }
+ ret = utils_mkdir_recursive(stream->path_name, S_IRWXU | S_IRWXG,
+ -1, -1);
+ if (ret < 0) {
+ ERR("relay creating output directory");
+ goto end_stream_unlock;
+ }
+ stream->chunk_id = be64toh(stream_info.new_chunk_id);
+
+ if (stream->is_metadata) {
+ /*
+ * The metadata stream is sent only over the control connection
+ * so we know we have all the data to perform the stream
+ * rotation.
+ */
+ ret = do_rotate_stream(stream);
+ } else {
+ stream->rotate_at_seq_num = be64toh(stream_info.rotate_at_seq_num);
+ ret = try_rotate_stream(stream);
+ }
+ if (ret < 0) {
+ goto end_stream_unlock;
+ }
+
+end_stream_unlock:
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+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(struct lttcomm_relayd_generic_reply), 0);
+ if (send_ret < 0) {
+ ERR("Failed to send reply of rotate session stream command");
+ ret = send_ret;
+ }
+
+end_no_reply:
+ free(new_pathname);
+ return ret;
+}
+
+/*
+ * relay_mkdir: Create a folder on the disk.
+ */
+static int relay_mkdir(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret;
+ ssize_t network_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_mkdir path_info_header;
+ struct lttcomm_relayd_mkdir *path_info = NULL;
+ struct lttcomm_relayd_generic_reply reply;
+ char *path = NULL;
+
+ 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;
+ }
+
+ network_ret = conn->sock->ops->recvmsg(conn->sock, &path_info_header,
+ sizeof(path_info_header), 0);
+ if (network_ret < (ssize_t) sizeof(path_info_header)) {
+ if (network_ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown", conn->sock->fd);
+ } else {
+ ERR("Reception of mkdir command argument length failed with ret = %zi, expected %zu",
+ network_ret, sizeof(path_info_header));
+ }
+ ret = -1;
+ goto end_no_session;
+ }
+
+ path_info_header.length = be32toh(path_info_header.length);
+
+ /* 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_info = zmalloc(sizeof(path_info_header) + path_info_header.length);
+ if (!path_info) {
+ PERROR("zmalloc of mkdir command path");
+ ret = -1;
+ goto end;
+ }
+
+ network_ret = conn->sock->ops->recvmsg(conn->sock, path_info->path,
+ path_info_header.length, 0);
+ if (network_ret < (ssize_t) path_info_header.length) {
+ if (network_ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown", conn->sock->fd);
+ } else {
+ ERR("Reception of mkdir path argument failed with ret = %zi, expected %" PRIu32,
+ network_ret, path_info_header.length);
+ }
+ ret = -1;
+ goto end_no_session;
+ }
+
+ path = create_output_path(path_info->path);
+ if (!path) {
+ ERR("Failed to create output path");
+ ret = -1;
+ goto end;
+ }
+
+ 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);
+ }
+ network_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(struct lttcomm_relayd_generic_reply), 0);
+ if (network_ret < (ssize_t) sizeof(struct lttcomm_relayd_generic_reply)) {
+ ERR("Failed to send mkdir command status code with ret = %zi, expected %zu",
+ network_ret,
+ sizeof(struct lttcomm_relayd_generic_reply));
+ ret = -1;
+ }
+
+end_no_session:
+ free(path);
+ free(path_info);
+ 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(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret;
+ ssize_t network_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_generic_reply reply;
+ struct lttcomm_relayd_rotate_rename header;
+ char *received_paths = NULL;
+ size_t received_paths_size;
+ const char *received_old_path, *received_new_path;
+ char *complete_old_path = NULL, *complete_new_path = NULL;
+
+ 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;
+ }
+
+ network_ret = conn->sock->ops->recvmsg(conn->sock, &header,
+ sizeof(header), 0);
+ if (network_ret < (ssize_t) sizeof(header)) {
+ if (network_ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown",
+ conn->sock->fd);
+ } else {
+ ERR("Relay didn't receive a valid rotate_rename command header: expected %zu bytes, recvmsg() returned %zi",
+ sizeof(header), network_ret);
+ }
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ 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;
+
+ /* 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;
+ }
+
+ received_paths = zmalloc(received_paths_size);
+ if (!received_paths) {
+ PERROR("Could not allocate rotate commands paths reception buffer");
+ ret = -1;
+ goto end;
+ }
+
+ network_ret = conn->sock->ops->recvmsg(conn->sock, received_paths,
+ received_paths_size, 0);
+ if (network_ret < (ssize_t) received_paths_size) {
+ if (network_ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown",
+ conn->sock->fd);
+ } else {
+ ERR("Relay failed to received rename command paths (%zu bytes): recvmsg() returned %zi",
+ received_paths_size, network_ret);
+ }
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ /* Validate that both paths received are NULL terminated. */
+ if (received_paths[header.old_path_length - 1] != '\0') {
+ ERR("relay_rotate_rename command's \"old\" path is invalid (not NULL terminated)");
+ ret = -1;
+ goto end;
+ }
+ if (received_paths[received_paths_size - 1] != '\0') {
+ ERR("relay_rotate_rename command's \"new\" path is invalid (not NULL terminated)");
+ ret = -1;
+ goto end;
+ }
+
+ received_old_path = received_paths;
+ received_new_path = received_paths + header.old_path_length;
+
+ complete_old_path = create_output_path(received_old_path);
+ 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(received_new_path);
+ if (!complete_new_path) {
+ ERR("Failed to build new output path in rotate_rename command");
+ ret = -1;
+ goto end;
+ }
+
+ 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);
+ }
+ network_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(struct lttcomm_relayd_generic_reply), 0);
+ if (network_ret < sizeof(struct lttcomm_relayd_generic_reply)) {
+ ERR("Relay sending stream id");
+ ret = -1;
+ }
+
+end_no_reply:
+ free(received_paths);
+ 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(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_rotate_pending msg;
+ struct lttcomm_relayd_generic_reply reply;
+ struct lttng_ht_iter iter;
+ struct relay_stream *stream;
+ int ret = 0;
+ ssize_t network_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;
+ }
+
+ network_ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0);
+ if (network_ret < (ssize_t) sizeof(msg)) {
+ if (network_ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown", conn->sock->fd);
+ } else {
+ ERR("Relay didn't receive valid rotate_pending struct size : %zi",
+ network_ret);
+ }
+ ret = -1;
+ goto end_no_reply;