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);
free(msg);
return ret;
}
-
-int relayd_rotate_rename(struct lttcomm_relayd_sock *rsock,
- const char *old_path, const char *new_path)
-{
- int ret;
- struct lttcomm_relayd_rotate_rename *msg = NULL;
- struct lttcomm_relayd_generic_reply reply;
- size_t old_path_length, new_path_length;
- size_t msg_length;
-
- /* Code flow error. Safety net. */
- assert(rsock);
-
- DBG("Relayd rename chunk %s to %s", old_path, new_path);
-
- /* The two paths are sent with a '\0' delimiter between them. */
- old_path_length = strlen(old_path) + 1;
- new_path_length = strlen(new_path) + 1;
-
- msg_length = sizeof(*msg) + old_path_length + new_path_length;
- msg = zmalloc(msg_length);
- if (!msg) {
- PERROR("zmalloc rotate-rename command message");
- ret = -1;
- goto error;
- }
-
- assert(old_path_length <= UINT32_MAX);
- msg->old_path_length = htobe32(old_path_length);
-
- assert(new_path_length <= UINT32_MAX);
- msg->new_path_length = htobe32(new_path_length);
-
- strcpy(msg->paths, old_path);
- strcpy(msg->paths + old_path_length, new_path);
-
- /* Send command */
- ret = send_command(rsock, RELAYD_ROTATE_RENAME, (const void *) msg,
- msg_length, 0);
- if (ret < 0) {
- goto error;
- }
-
- /* Receive response */
- ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
- if (ret < 0) {
- goto error;
- }
-
- reply.ret_code = be32toh(reply.ret_code);
-
- /* Return session id or negative ret code. */
- if (reply.ret_code != LTTNG_OK) {
- ret = -1;
- ERR("Relayd rotate rename replied error %d", reply.ret_code);
- } else {
- /* Success */
- ret = 0;
- }
-
- DBG("Relayd rotate rename completed successfully");
-
-error:
- free(msg);
- return ret;
-}
-
-int relayd_rotate_pending(struct lttcomm_relayd_sock *rsock, uint64_t chunk_id)
-{
- int ret;
- struct lttcomm_relayd_rotate_pending msg;
- struct lttcomm_relayd_rotate_pending_reply reply;
-
- /* Code flow error. Safety net. */
- assert(rsock);
-
- DBG("Querying relayd for rotate pending with chunk_id %" PRIu64,
- chunk_id);
-
- memset(&msg, 0, sizeof(msg));
- msg.chunk_id = htobe64(chunk_id);
-
- /* Send command */
- ret = send_command(rsock, RELAYD_ROTATE_PENDING, (void *) &msg,
- sizeof(msg), 0);
- if (ret < 0) {
- goto error;
- }
-
- /* Receive response */
- ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
- if (ret < 0) {
- goto error;
- }
-
- reply.generic.ret_code = be32toh(reply.generic.ret_code);
-
- /* Return session id or negative ret code. */
- if (reply.generic.ret_code != LTTNG_OK) {
- ret = -reply.generic.ret_code;
- ERR("Relayd rotate pending replied with error %d", ret);
- goto error;
- } else {
- /* No error, just rotate pending state */
- if (reply.is_pending == 0 || reply.is_pending == 1) {
- ret = reply.is_pending;
- DBG("Relayd rotate pending command completed successfully with result \"%s\"",
- ret ? "rotation pending" : "rotation NOT pending");
- } else {
- ret = -LTTNG_ERR_UNK;
- }
- }
-
-error:
- return ret;
-}
-
-int relayd_mkdir(struct lttcomm_relayd_sock *rsock, const char *path)
-{
- int ret;
- struct lttcomm_relayd_mkdir *msg;
- struct lttcomm_relayd_generic_reply reply;
- size_t len;
-
- /* Code flow error. Safety net. */
- assert(rsock);
-
- DBG("Relayd mkdir path %s", path);
-
- len = strlen(path) + 1;
- msg = zmalloc(sizeof(msg->length) + len);
- if (!msg) {
- PERROR("Alloc mkdir msg");
- ret = -1;
- goto error;
- }
- msg->length = htobe32((uint32_t) len);
-
- if (lttng_strncpy(msg->path, path, len)) {
- ret = -1;
- goto error;
- }
-
- /* Send command */
- ret = send_command(rsock, RELAYD_MKDIR, (void *) msg,
- sizeof(msg->length) + len, 0);
- if (ret < 0) {
- goto error;
- }
-
- /* Receive response */
- ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
- if (ret < 0) {
- goto error;
- }
-
- reply.ret_code = be32toh(reply.ret_code);
-
- /* Return session id or negative ret code. */
- if (reply.ret_code != LTTNG_OK) {
- ret = -1;
- ERR("Relayd mkdir replied error %d", reply.ret_code);
- } else {
- /* Success */
- ret = 0;
- }
-
- DBG("Relayd mkdir completed successfully");
-
-error:
- free(msg);
- return ret;
-}
projects / lttng-tools.git / commitdiff
