+
+ chunk_status = lttng_trace_chunk_override_name(
+ chunk, chunk_name_view.data);
+ switch (chunk_status) {
+ case LTTNG_TRACE_CHUNK_STATUS_OK:
+ break;
+ case LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT:
+ ERR("Failed to set the name of new trace chunk in trace chunk creation command (invalid name)");
+ reply_code = LTTNG_ERR_INVALID;
+ ret = -1;
+ goto end;
+ default:
+ ERR("Failed to set the name of new trace chunk in trace chunk creation command (unknown error)");
+ reply_code = LTTNG_ERR_UNK;
+ ret = -1;
+ 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;
+ ret = -1;
+ goto end;
+ }
+
+ session_output = session_create_output_directory_handle(
+ conn->session);
+ if (!session_output) {
+ reply_code = LTTNG_ERR_CREATE_DIR_FAIL;
+ goto end;
+ }
+ chunk_status = lttng_trace_chunk_set_as_owner(chunk, session_output);
+ lttng_directory_handle_put(session_output);
+ session_output = NULL;
+ if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ reply_code = LTTNG_ERR_UNK;
+ ret = -1;
+ goto end;
+ }
+
+ published_chunk = sessiond_trace_chunk_registry_publish_chunk(
+ sessiond_trace_chunk_registry,
+ conn->session->sessiond_uuid,
+ conn->session->id,
+ chunk);
+ if (!published_chunk) {
+ char uuid_str[LTTNG_UUID_STR_LEN];
+
+ lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
+ ERR("Failed to publish chunk: 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;
+ }
+
+ pthread_mutex_lock(&conn->session->lock);
+ if (conn->session->pending_closure_trace_chunk) {
+ /*
+ * Invalid; this means a second create_trace_chunk command was
+ * received before a close_trace_chunk.
+ */
+ ERR("Invalid trace chunk close command received; a trace chunk is already waiting for a trace chunk close command");
+ reply_code = LTTNG_ERR_INVALID_PROTOCOL;
+ ret = -1;
+ goto end_unlock_session;
+ }
+ conn->session->pending_closure_trace_chunk =
+ conn->session->current_trace_chunk;
+ conn->session->current_trace_chunk = published_chunk;
+ published_chunk = NULL;
+end_unlock_session:
+ pthread_mutex_unlock(&conn->session->lock);
+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);
+ lttng_trace_chunk_put(published_chunk);
+ lttng_directory_handle_put(session_output);
+ 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, buf_ret;
+ ssize_t send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_close_trace_chunk *msg;
+ 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;
+ 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;
+ 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");
+ 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[LTTNG_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;
+ }
+
+ pthread_mutex_lock(&session->lock);
+ if (session->pending_closure_trace_chunk &&
+ session->pending_closure_trace_chunk != chunk) {
+ ERR("Trace chunk close command for session \"%s\" does not target the trace chunk pending closure",
+ session->session_name);
+ reply_code = LTTNG_ERR_INVALID_PROTOCOL;
+ ret = -1;
+ goto end_unlock_session;
+ }
+
+ 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_unlock_session;
+ }
+
+ 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_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) {
+ /*
+ * After a trace chunk close command, no new streams
+ * referencing the chunk may be created. Hence, on the
+ * event that no new trace chunk have been created for
+ * the session, the reference to the current trace chunk
+ * is released in order to allow it to be reclaimed when
+ * the last stream releases its reference to it.
+ */
+ lttng_trace_chunk_put(session->current_trace_chunk);
+ session->current_trace_chunk = NULL;
+ }
+ lttng_trace_chunk_put(session->pending_closure_trace_chunk);
+ session->pending_closure_trace_chunk = NULL;
+end_unlock_session:
+ pthread_mutex_unlock(&session->lock);
+
+end:
+ 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_payload.data,
+ reply_payload.size,
+ 0);
+ 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;
+}
+
+/*
+ * relay_trace_chunk_exists: check if a trace chunk exists
+ */
+static int relay_trace_chunk_exists(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_trace_chunk_exists *msg;
+ struct lttcomm_relayd_trace_chunk_exists_reply reply = {};
+ struct lttng_buffer_view header_view;
+ uint64_t chunk_id;
+ bool chunk_exists;
+
+ 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);
+
+ ret = sessiond_trace_chunk_registry_chunk_exists(
+ sessiond_trace_chunk_registry,
+ conn->session->sessiond_uuid,
+ conn->session->id,
+ chunk_id, &chunk_exists);
+ /*
+ * If ret is not 0, send the reply and report the error to the caller.
+ * It is a protocol (or internal) error and the session/connection
+ * should be torn down.
+ */
+ reply = (typeof(reply)){
+ .generic.ret_code = htobe32((uint32_t)
+ (ret == 0 ? LTTNG_OK : LTTNG_ERR_INVALID_PROTOCOL)),
+ .trace_chunk_exists = ret == 0 ? chunk_exists : 0,
+ };
+ send_ret = conn->sock->ops->sendmsg(
+ conn->sock, &reply, sizeof(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:
+ return ret;
+}
+
+#define DBG_CMD(cmd_name, conn) \
+ DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd);
+
+static int relay_process_control_command(struct relay_connection *conn,
+ const struct lttcomm_relayd_hdr *header,
+ const struct lttng_buffer_view *payload)
+{
+ int ret = 0;
+
+ switch (header->cmd) {
+ case RELAYD_CREATE_SESSION:
+ DBG_CMD("RELAYD_CREATE_SESSION", conn);
+ ret = relay_create_session(header, conn, payload);
+ break;
+ case RELAYD_ADD_STREAM:
+ DBG_CMD("RELAYD_ADD_STREAM", conn);
+ ret = relay_add_stream(header, conn, payload);
+ break;
+ case RELAYD_START_DATA:
+ DBG_CMD("RELAYD_START_DATA", conn);
+ ret = relay_start(header, conn, payload);
+ break;
+ case RELAYD_SEND_METADATA:
+ DBG_CMD("RELAYD_SEND_METADATA", conn);
+ ret = relay_recv_metadata(header, conn, payload);
+ break;
+ case RELAYD_VERSION:
+ DBG_CMD("RELAYD_VERSION", conn);
+ ret = relay_send_version(header, conn, payload);
+ break;
+ case RELAYD_CLOSE_STREAM:
+ DBG_CMD("RELAYD_CLOSE_STREAM", conn);
+ ret = relay_close_stream(header, conn, payload);
+ break;
+ case RELAYD_DATA_PENDING:
+ DBG_CMD("RELAYD_DATA_PENDING", conn);
+ ret = relay_data_pending(header, conn, payload);
+ break;
+ case RELAYD_QUIESCENT_CONTROL:
+ DBG_CMD("RELAYD_QUIESCENT_CONTROL", conn);
+ ret = relay_quiescent_control(header, conn, payload);
+ break;
+ case RELAYD_BEGIN_DATA_PENDING:
+ DBG_CMD("RELAYD_BEGIN_DATA_PENDING", conn);
+ ret = relay_begin_data_pending(header, conn, payload);
+ break;
+ case RELAYD_END_DATA_PENDING:
+ DBG_CMD("RELAYD_END_DATA_PENDING", conn);
+ ret = relay_end_data_pending(header, conn, payload);
+ break;
+ case RELAYD_SEND_INDEX:
+ DBG_CMD("RELAYD_SEND_INDEX", conn);
+ ret = relay_recv_index(header, conn, payload);
+ break;
+ case RELAYD_STREAMS_SENT:
+ DBG_CMD("RELAYD_STREAMS_SENT", conn);
+ ret = relay_streams_sent(header, conn, payload);
+ break;
+ case RELAYD_RESET_METADATA:
+ DBG_CMD("RELAYD_RESET_METADATA", conn);
+ ret = relay_reset_metadata(header, conn, payload);
+ break;
+ case RELAYD_ROTATE_STREAMS:
+ DBG_CMD("RELAYD_ROTATE_STREAMS", conn);
+ ret = relay_rotate_session_streams(header, conn, payload);
+ break;
+ case RELAYD_CREATE_TRACE_CHUNK:
+ 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_TRACE_CHUNK_EXISTS:
+ DBG_CMD("RELAYD_TRACE_CHUNK_EXISTS", conn);
+ ret = relay_trace_chunk_exists(header, conn, payload);
+ break;
+ case RELAYD_UPDATE_SYNC_INFO:
+ default:
+ ERR("Received unknown command (%u)", header->cmd);
+ relay_unknown_command(conn);
+ ret = -1;
+ goto end;
+ }
+
+end:
+ return ret;
+}
+
+static enum relay_connection_status relay_process_control_receive_payload(
+ struct relay_connection *conn)
+{
+ int ret = 0;
+ enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
+ struct lttng_dynamic_buffer *reception_buffer =
+ &conn->protocol.ctrl.reception_buffer;
+ struct ctrl_connection_state_receive_payload *state =
+ &conn->protocol.ctrl.state.receive_payload;
+ struct lttng_buffer_view payload_view;
+
+ if (state->left_to_receive == 0) {
+ /* Short-circuit for payload-less commands. */
+ goto reception_complete;
+ }
+
+ ret = conn->sock->ops->recvmsg(conn->sock,
+ reception_buffer->data + state->received,
+ state->left_to_receive, MSG_DONTWAIT);
+ if (ret < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK) {
+ PERROR("Unable to receive command payload on sock %d",
+ conn->sock->fd);
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ }
+ goto end;
+ } else if (ret == 0) {
+ DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
+ status = RELAY_CONNECTION_STATUS_CLOSED;
+ goto end;
+ }
+
+ assert(ret > 0);
+ assert(ret <= state->left_to_receive);
+
+ state->left_to_receive -= ret;
+ state->received += ret;
+
+ if (state->left_to_receive > 0) {
+ /*
+ * Can't transition to the protocol's next state, wait to
+ * receive the rest of the header.
+ */
+ DBG3("Partial reception of control connection protocol payload (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
+ state->received, state->left_to_receive,
+ conn->sock->fd);
+ goto end;
+ }
+
+reception_complete:
+ DBG("Done receiving control command payload: fd = %i, payload size = %" PRIu64 " bytes",
+ conn->sock->fd, state->received);
+ /*
+ * The payload required to process the command has been received.
+ * A view to the reception buffer is forwarded to the various
+ * commands and the state of the control is reset on success.
+ *
+ * Commands are responsible for sending their reply to the peer.
+ */
+ payload_view = lttng_buffer_view_from_dynamic_buffer(reception_buffer,
+ 0, -1);
+ ret = relay_process_control_command(conn,
+ &state->header, &payload_view);
+ if (ret < 0) {
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ goto end;
+ }
+
+ ret = connection_reset_protocol_state(conn);
+ if (ret) {
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ }
+end:
+ return status;
+}
+
+static enum relay_connection_status relay_process_control_receive_header(
+ struct relay_connection *conn)
+{
+ int ret = 0;
+ enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
+ struct lttcomm_relayd_hdr header;
+ struct lttng_dynamic_buffer *reception_buffer =
+ &conn->protocol.ctrl.reception_buffer;
+ struct ctrl_connection_state_receive_header *state =
+ &conn->protocol.ctrl.state.receive_header;
+
+ assert(state->left_to_receive != 0);
+
+ ret = conn->sock->ops->recvmsg(conn->sock,
+ reception_buffer->data + state->received,
+ state->left_to_receive, MSG_DONTWAIT);
+ if (ret < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK) {
+ PERROR("Unable to receive control command header on sock %d",
+ conn->sock->fd);
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ }
+ goto end;
+ } else if (ret == 0) {
+ DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
+ status = RELAY_CONNECTION_STATUS_CLOSED;
+ goto end;
+ }
+
+ assert(ret > 0);
+ assert(ret <= state->left_to_receive);
+
+ state->left_to_receive -= ret;
+ state->received += ret;
+
+ if (state->left_to_receive > 0) {
+ /*
+ * Can't transition to the protocol's next state, wait to
+ * receive the rest of the header.
+ */
+ DBG3("Partial reception of control connection protocol header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
+ state->received, state->left_to_receive,
+ conn->sock->fd);
+ goto end;
+ }
+
+ /* Transition to next state: receiving the command's payload. */
+ conn->protocol.ctrl.state_id =
+ CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD;
+ memcpy(&header, reception_buffer->data, sizeof(header));
+ header.circuit_id = be64toh(header.circuit_id);
+ header.data_size = be64toh(header.data_size);
+ header.cmd = be32toh(header.cmd);
+ header.cmd_version = be32toh(header.cmd_version);
+ memcpy(&conn->protocol.ctrl.state.receive_payload.header,
+ &header, sizeof(header));
+
+ DBG("Done receiving control command header: fd = %i, cmd = %" PRIu32 ", cmd_version = %" PRIu32 ", payload size = %" PRIu64 " bytes",
+ conn->sock->fd, header.cmd, header.cmd_version,
+ header.data_size);
+
+ if (header.data_size > DEFAULT_NETWORK_RELAYD_CTRL_MAX_PAYLOAD_SIZE) {
+ ERR("Command header indicates a payload (%" PRIu64 " bytes) that exceeds the maximal payload size allowed on a control connection.",
+ header.data_size);
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ goto end;