#include <common/sessiond-comm/relayd.h>
#include <common/uri.h>
#include <common/utils.h>
+#include <common/align.h>
#include <common/config/session-config.h>
#include <urcu/rculist.h>
/* Size of receive buffer. */
#define RECV_DATA_BUFFER_SIZE 65536
+#define FILE_COPY_BUFFER_SIZE 65536
static int recv_child_signal; /* Set to 1 when a SIGUSR1 signal is received. */
static pid_t child_ppid; /* Internal parent PID use with daemonize. */
return ret;
}
+/*
+ * Close the current index file if it is open, and create a new one.
+ *
+ * Return 0 on success, -1 on error.
+ */
+static
+int create_rotate_index_file(struct relay_stream *stream)
+{
+ int ret;
+ uint32_t major, minor;
+
+ /* Put ref on previous index_file. */
+ if (stream->index_file) {
+ lttng_index_file_put(stream->index_file);
+ stream->index_file = NULL;
+ }
+ major = stream->trace->session->major;
+ minor = stream->trace->session->minor;
+ stream->index_file = lttng_index_file_create(stream->path_name,
+ stream->channel_name,
+ -1, -1, stream->tracefile_size,
+ tracefile_array_get_file_index_head(stream->tfa),
+ lttng_to_index_major(major, minor),
+ lttng_to_index_minor(major, minor));
+ if (!stream->index_file) {
+ ret = -1;
+ goto end;
+ }
+
+ ret = 0;
+
+end:
+ return ret;
+}
+
+static
+int do_rotate_stream(struct relay_stream *stream)
+{
+ int ret;
+
+ /* Perform the stream rotation. */
+ ret = utils_rotate_stream_file(stream->path_name,
+ stream->channel_name, stream->tracefile_size,
+ stream->tracefile_count, -1,
+ -1, stream->stream_fd->fd,
+ NULL, &stream->stream_fd->fd);
+ if (ret < 0) {
+ ERR("Rotating stream output file");
+ goto end;
+ }
+ stream->tracefile_size_current = 0;
+
+ /* Rotate also the index if the stream is not a metadata stream. */
+ if (!stream->is_metadata) {
+ ret = create_rotate_index_file(stream);
+ if (ret < 0) {
+ ERR("Failed to rotate index file");
+ goto end;
+ }
+ }
+
+ stream->rotate_at_seq_num = -1ULL;
+ stream->pos_after_last_complete_data_index = 0;
+
+end:
+ return ret;
+}
+
+/*
+ * If too much data has been written in a tracefile before we received the
+ * rotation command, we have to move the excess data to the new tracefile and
+ * perform the rotation. This can happen because the control and data
+ * connections are separate, the indexes as well as the commands arrive from
+ * the control connection and we have no control over the order so we could be
+ * in a situation where too much data has been received on the data connection
+ * before the rotation command on the control connection arrives. We don't need
+ * to update the index because its order is guaranteed with the rotation
+ * command message.
+ */
+static
+int rotate_truncate_stream(struct relay_stream *stream)
+{
+ int ret, new_fd;
+ uint64_t diff, pos = 0;
+ char buf[FILE_COPY_BUFFER_SIZE];
+
+ assert(!stream->is_metadata);
+
+ assert(stream->tracefile_size_current >
+ stream->pos_after_last_complete_data_index);
+ diff = stream->tracefile_size_current -
+ stream->pos_after_last_complete_data_index;
+
+ /* Create the new tracefile. */
+ new_fd = utils_create_stream_file(stream->path_name,
+ stream->channel_name,
+ stream->tracefile_size, stream->tracefile_count,
+ /* uid */ -1, /* gid */ -1, /* suffix */ NULL);
+ if (new_fd < 0) {
+ ERR("Failed to create new stream file at path %s for channel %s",
+ stream->path_name, stream->channel_name);
+ ret = -1;
+ goto end;
+ }
+
+ /*
+ * Rewind the current tracefile to the position at which the rotation
+ * should have occured.
+ */
+ ret = lseek(stream->stream_fd->fd,
+ stream->pos_after_last_complete_data_index, SEEK_SET);
+ if (ret < 0) {
+ PERROR("seek truncate stream");
+ goto end;
+ }
+
+ /* Move data from the old file to the new file. */
+ while (pos < diff) {
+ uint64_t count, bytes_left;
+ ssize_t io_ret;
+
+ bytes_left = diff - pos;
+ count = bytes_left > sizeof(buf) ? sizeof(buf) : bytes_left;
+ assert(count <= SIZE_MAX);
+
+ io_ret = lttng_read(stream->stream_fd->fd, buf, count);
+ if (io_ret < (ssize_t) count) {
+ char error_string[256];
+
+ snprintf(error_string, sizeof(error_string),
+ "Failed to read %" PRIu64 " bytes from fd %i in rotate_truncate_stream(), returned %zi",
+ count, stream->stream_fd->fd, io_ret);
+ if (io_ret == -1) {
+ PERROR("%s", error_string);
+ } else {
+ ERR("%s", error_string);
+ }
+ ret = -1;
+ goto end;
+ }
+
+ io_ret = lttng_write(new_fd, buf, count);
+ if (io_ret < (ssize_t) count) {
+ char error_string[256];
+
+ snprintf(error_string, sizeof(error_string),
+ "Failed to write %" PRIu64 " bytes from fd %i in rotate_truncate_stream(), returned %zi",
+ count, new_fd, io_ret);
+ if (io_ret == -1) {
+ PERROR("%s", error_string);
+ } else {
+ ERR("%s", error_string);
+ }
+ ret = -1;
+ goto end;
+ }
+
+ pos += count;
+ }
+
+ /* Truncate the file to get rid of the excess data. */
+ ret = ftruncate(stream->stream_fd->fd,
+ stream->pos_after_last_complete_data_index);
+ if (ret) {
+ PERROR("ftruncate");
+ goto end;
+ }
+
+ ret = close(stream->stream_fd->fd);
+ if (ret < 0) {
+ PERROR("Closing tracefile");
+ goto end;
+ }
+
+ ret = create_rotate_index_file(stream);
+ if (ret < 0) {
+ ERR("Rotate stream index file");
+ goto end;
+ }
+
+ /*
+ * Update the offset and FD of all the eventual indexes created by the
+ * data connection before the rotation command arrived.
+ */
+ ret = relay_index_switch_all_files(stream);
+ if (ret < 0) {
+ ERR("Failed to rotate index file");
+ goto end;
+ }
+
+ stream->stream_fd->fd = new_fd;
+ stream->tracefile_size_current = diff;
+ stream->pos_after_last_complete_data_index = 0;
+ stream->rotate_at_seq_num = -1ULL;
+
+ ret = 0;
+
+end:
+ return ret;
+}
+
+/*
+ * Check if a stream should perform a rotation (for session rotation).
+ * Must be called with the stream lock held.
+ *
+ * Return 0 on success, a negative value on error.
+ */
+static
+int try_rotate_stream(struct relay_stream *stream)
+{
+ int ret = 0;
+
+ /* No rotation expected. */
+ if (stream->rotate_at_seq_num == -1ULL) {
+ goto end;
+ }
+
+ if (stream->prev_seq < stream->rotate_at_seq_num) {
+ DBG("Stream %" PRIu64 " no yet ready for rotation",
+ stream->stream_handle);
+ goto end;
+ } else if (stream->prev_seq > stream->rotate_at_seq_num) {
+ DBG("Rotation after too much data has been written in tracefile "
+ "for stream %" PRIu64 ", need to truncate before "
+ "rotating", stream->stream_handle);
+ ret = rotate_truncate_stream(stream);
+ if (ret) {
+ ERR("Failed to truncate stream");
+ goto end;
+ }
+ } else {
+ /* stream->prev_seq == stream->rotate_at_seq_num */
+ DBG("Stream %" PRIu64 " ready for rotation",
+ stream->stream_handle);
+ ret = do_rotate_stream(stream);
+ }
+
+end:
+ return ret;
+}
+
/*
* relay_recv_metadata: receive the metadata for the session.
*/
DBG2("Relay metadata written. Updated metadata_received %" PRIu64,
metadata_stream->metadata_received);
+ ret = try_rotate_stream(metadata_stream);
+ if (ret < 0) {
+ goto end_put;
+ }
+
end_put:
pthread_mutex_unlock(&metadata_stream->lock);
stream_put(metadata_stream);
if (ret == 0) {
tracefile_array_commit_seq(stream->tfa);
stream->index_received_seqcount++;
+ stream->pos_after_last_complete_data_index += index->total_size;
} else if (ret > 0) {
/* no flush. */
ret = 0;
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.
*/
case RELAYD_RESET_METADATA:
ret = relay_reset_metadata(recv_hdr, conn);
break;
+ case RELAYD_ROTATE_STREAM:
+ ret = relay_rotate_session_stream(recv_hdr, conn);
+ break;
case RELAYD_ROTATE_RENAME:
ret = relay_rotate_rename(recv_hdr, conn);
break;
* Return 0 on success else a negative value.
*/
static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
- int rotate_index)
+ int rotate_index, bool *flushed, uint64_t total_size)
{
int ret = 0;
uint64_t data_offset;
}
if (rotate_index || !stream->index_file) {
- uint32_t major, minor;
-
- /* Put ref on previous index_file. */
- if (stream->index_file) {
- lttng_index_file_put(stream->index_file);
- stream->index_file = NULL;
- }
- major = stream->trace->session->major;
- minor = stream->trace->session->minor;
- stream->index_file = lttng_index_file_create(stream->path_name,
- stream->channel_name,
- -1, -1, stream->tracefile_size,
- tracefile_array_get_file_index_head(stream->tfa),
- lttng_to_index_major(major, minor),
- lttng_to_index_minor(major, minor));
- if (!stream->index_file) {
- ret = -1;
+ ret = create_rotate_index_file(stream);
+ if (ret < 0) {
+ ERR("Failed to rotate index");
/* Put self-ref for this index due to error. */
relay_index_put(index);
index = NULL;
if (ret == 0) {
tracefile_array_commit_seq(stream->tfa);
stream->index_received_seqcount++;
+ *flushed = true;
} else if (ret > 0) {
+ index->total_size = total_size;
/* No flush. */
ret = 0;
} else {
size_t chunk_size = RECV_DATA_BUFFER_SIZE;
size_t recv_off = 0;
char data_buffer[chunk_size];
+ bool index_flushed = false;
ret = conn->sock->ops->recvmsg(conn->sock, &data_hdr,
sizeof(struct lttcomm_relayd_data_hdr), 0);
* snapshot and index are NOT supported.
*/
if (session->minor >= 4 && !session->snapshot) {
- ret = handle_index_data(stream, net_seq_num, rotate_index);
+ ret = handle_index_data(stream, net_seq_num, rotate_index,
+ &index_flushed,
+ data_size + be32toh(data_hdr.padding_size));
if (ret < 0) {
ERR("handle_index_data: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d",
stream->stream_handle, net_seq_num, ret);
if (stream->prev_seq == -1ULL) {
new_stream = true;
}
+ if (index_flushed) {
+ stream->pos_after_last_complete_data_index =
+ stream->tracefile_size_current;
+ }
stream->prev_seq = net_seq_num;
+ ret = try_rotate_stream(stream);
+ if (ret < 0) {
+ goto end_stream_unlock;
+ }
+
end_stream_unlock:
close_requested = stream->close_requested;
pthread_mutex_unlock(&stream->lock);
projects / lttng-tools.git / commitdiff
