#include "connection.h"
#include "tracefile-array.h"
+static const char *help_msg =
+#ifdef LTTNG_EMBED_HELP
+#include <lttng-relayd.8.h>
+#else
+NULL
+#endif
+;
+
/* command line options */
char *opt_output_path;
static int opt_daemon, opt_background;
}
break;
case 'h':
- ret = utils_show_man_page(8, "lttng-relayd");
+ ret = utils_show_help(8, "lttng-relayd", help_msg);
if (ret) {
- ERR("Cannot view man page lttng-relayd(8)");
+ ERR("Cannot show --help for `lttng-relayd`");
perror("exec");
}
exit(EXIT_FAILURE);
health_code_update();
- while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
+ for (;;) {
health_code_update();
/* Atomically prepare the queue futex */
futex_nto1_prepare(&relay_conn_queue.futex);
+ if (CMM_LOAD_SHARED(dispatch_thread_exit)) {
+ break;
+ }
+
do {
health_code_update();
&channel_name);
break;
case 2: /* LTTng sessiond 2.2. Allocates path_name and channel_name. */
- default:
+ case 3: /* LTTng sessiond 2.3. Allocates path_name and channel_name. */
+ case 4: /* LTTng sessiond 2.4. Allocates path_name and channel_name. */
+ case 5: /* LTTng sessiond 2.5. Allocates path_name and channel_name. */
+ case 6: /* LTTng sessiond 2.6. Allocates path_name and channel_name. */
+ case 7: /* LTTng sessiond 2.7. Allocates path_name and channel_name. */
+ case 8: /* LTTng sessiond 2.8. Allocates path_name and channel_name. */
+ case 9: /* LTTng sessiond 2.9. Allocates path_name and channel_name. */
+ case 10: /* LTTng sessiond 2.10. Allocates path_name and channel_name. */
ret = cmd_recv_stream_2_2(conn, &path_name,
&channel_name, &tracefile_size, &tracefile_count);
break;
+ case 11: /* LTTng sessiond 2.11. Allocates path_name and channel_name. */
+ default:
+ ret = cmd_recv_stream_2_11(conn, &path_name,
+ &channel_name, &tracefile_size, &tracefile_count,
+ session);
+ break;
}
if (ret < 0) {
goto send_reply;
return ret;
}
+static
+int 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 = rotate_index_file(stream);
+ if (ret < 0) {
+ ERR("Failed to rotate index file");
+ goto end;
+ }
+ }
+
+ stream->rotate_at_seq_num = -1ULL;
+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 check_rotate_stream(struct relay_stream *stream)
+{
+ int ret;
+
+ /* No rotation expected */
+ if (stream->rotate_at_seq_num == -1ULL) {
+ ret = 0;
+ goto end;
+ }
+
+ if (stream->prev_seq < stream->rotate_at_seq_num) {
+ DBG("Stream %" PRIu64 " no yet ready for rotation",
+ stream->stream_handle);
+ ret = 0;
+ 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);
+ fprintf(stderr, "Rotation after too much data has been written in tracefile "
+ "for stream %" PRIu64 ", need to truncate before "
+ "rotating\n", stream->stream_handle);
+ /* TODO */
+ } else {
+ 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 = check_rotate_stream(metadata_stream);
+ if (ret < 0) {
+ ERR("Check rotate stream");
+ goto end_put;
+ }
+
end_put:
pthread_mutex_unlock(&metadata_stream->lock);
stream_put(metadata_stream);
struct lttcomm_relayd_generic_reply reply;
struct relay_stream *stream;
uint64_t net_seq_num;
+ size_t msg_len;
assert(conn);
goto end_no_session;
}
+ msg_len = lttcomm_relayd_index_len(
+ lttng_to_index_major(conn->major, conn->minor),
+ lttng_to_index_minor(conn->major, conn->minor));
ret = conn->sock->ops->recvmsg(conn->sock, &index_info,
- sizeof(index_info), 0);
- if (ret < sizeof(index_info)) {
+ msg_len, 0);
+ if (ret < msg_len) {
if (ret == 0) {
/* Orderly shutdown. Not necessary to print an error. */
DBG("Socket %d did an orderly shutdown", conn->sock->fd);
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;
+
+ DBG("Rotate stream received");
+
+ if (!session || conn->version_check_done == 0) {
+ ERR("Trying to rotate a stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ if (session->minor < 11) {
+ ERR("Unsupported feature before 2.11");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ ret = conn->sock->ops->recvmsg(conn->sock, &stream_info,
+ sizeof(stream_info), 0);
+ if (ret < sizeof(stream_info)) {
+ 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_session;
+ }
+
+ stream = stream_get_by_id(be64toh(stream_info.stream_id));
+ if (!stream) {
+ ret = -1;
+ goto end;
+ }
+
+ len = lttng_strnlen(stream_info.new_pathname,
+ sizeof(stream_info.new_pathname));
+ /* Ensure that NULL-terminated and fits in local filename length. */
+ if (len == sizeof(stream_info.new_pathname) || len >= LTTNG_NAME_MAX) {
+ ret = -ENAMETOOLONG;
+ ERR("Path name too long");
+ goto end;
+ }
+
+ 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(stream_info.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->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 = check_rotate_stream(stream);
+ }
+ if (ret < 0) {
+ ERR("Check rotate stream");
+ 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("Relay sending stream id");
+ ret = send_ret;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * relay_rotate_rename: rename the trace folder after the rotation is
+ * complete. 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, send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_rotate_rename stream_info;
+ struct lttcomm_relayd_generic_reply reply;
+ size_t len;
+ char *old = NULL, *new = NULL;
+
+ DBG("Rotate rename received");
+
+ if (!session || conn->version_check_done == 0) {
+ ERR("Trying to rename before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ if (session->minor < 11) {
+ ERR("Unsupported feature before 2.11");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ ret = conn->sock->ops->recvmsg(conn->sock, &stream_info,
+ sizeof(stream_info), 0);
+ if (ret < sizeof(stream_info)) {
+ 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_rename struct size : %d", ret);
+ }
+ ret = -1;
+ goto end_no_session;
+ }
+
+ len = lttng_strnlen(stream_info.current_path,
+ sizeof(stream_info.current_path));
+ /* Ensure that NULL-terminated and fits in local filename length. */
+ if (len == sizeof(stream_info.current_path) || len >= LTTNG_NAME_MAX) {
+ ret = -ENAMETOOLONG;
+ ERR("Path name too long");
+ goto end;
+ }
+
+ len = lttng_strnlen(stream_info.new_path,
+ sizeof(stream_info.new_path));
+ /* Ensure that NULL-terminated and fits in local filename length. */
+ if (len == sizeof(stream_info.new_path) || len >= LTTNG_NAME_MAX) {
+ ret = -ENAMETOOLONG;
+ ERR("Path name too long");
+ goto end;
+ }
+
+ old = create_output_path(stream_info.current_path);
+ if (!old) {
+ ERR("Failed to create current output path");
+ ret = -1;
+ goto end;
+ }
+
+ new = create_output_path(stream_info.new_path);
+ if (!new) {
+ ERR("Failed to create new output path");
+ ret = -1;
+ goto end;
+ }
+
+ ret = utils_mkdir_recursive(new, S_IRWXU | S_IRWXG,
+ -1, -1);
+ if (ret < 0) {
+ ERR("relay creating output directory");
+ goto end;
+ }
+
+ ret = rename(old, new);
+ if (ret < 0 && errno != ENOENT) {
+ PERROR("Rename completed rotation chunk");
+ 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(struct lttcomm_relayd_generic_reply), 0);
+ if (send_ret < 0) {
+ ERR("Relay sending stream id");
+ ret = send_ret;
+ }
+
+end_no_session:
+ free(old);
+ free(new);
+ return ret;
+}
+
+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;
+ uint64_t chunk_id;
+ uint32_t rotate_pending;
+
+ DBG("Rotate pending command received");
+
+ if (!session || conn->version_check_done == 0) {
+ ERR("Trying to check for data before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ if (session->minor < 11) {
+ ERR("Unsupported feature before 2.11");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ ret = conn->sock->ops->recvmsg(conn->sock, &msg, sizeof(msg), 0);
+ if (ret < sizeof(msg)) {
+ 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_pending struct size : %d",
+ ret);
+ }
+ ret = -1;
+ goto end_no_session;
+ }
+
+ chunk_id = be64toh(msg.chunk_id);
+
+ rotate_pending = 0;
+
+ 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) {
+ rotate_pending = 1;
+ DBG("Stream %" PRIu64 " is still rotating",
+ stream->stream_handle);
+ } else if (stream->chunk_id < chunk_id) {
+ rotate_pending = 1;
+ 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;
+ }
+ }
+ rcu_read_unlock();
+
+send_reply:
+ memset(&reply, 0, sizeof(reply));
+ reply.ret_code = htobe32(rotate_pending);
+ ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (ret < 0) {
+ ERR("Relay rotate pending ret code failed");
+ }
+
+end_no_session:
+ return ret;
+}
+
/*
* Process the commands received on the control socket
*/
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;
+ case RELAYD_ROTATE_PENDING:
+ ret = relay_rotate_pending(recv_hdr, conn);
+ break;
case RELAYD_UPDATE_SYNC_INFO:
default:
ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
goto end;
}
- if (rotate_index || !stream->index_fd) {
- int fd;
-
- /* Put ref on previous index_fd. */
- if (stream->index_fd) {
- stream_fd_put(stream->index_fd);
- stream->index_fd = NULL;
- }
-
- fd = index_create_file(stream->path_name, stream->channel_name,
- -1, -1, stream->tracefile_size,
- tracefile_array_get_file_index_head(stream->tfa));
- if (fd < 0) {
- ret = -1;
- /* Put self-ref for this index due to error. */
- relay_index_put(index);
- goto end;
- }
- stream->index_fd = stream_fd_create(fd);
- if (!stream->index_fd) {
- ret = -1;
- if (close(fd)) {
- PERROR("Error closing FD %d", fd);
- }
+ if (rotate_index || !stream->index_file) {
+ ret = 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);
- /* Will put the local ref. */
+ index = NULL;
goto end;
}
}
- if (relay_index_set_fd(index, stream->index_fd, data_offset)) {
+ if (relay_index_set_file(index, stream->index_file, data_offset)) {
ret = -1;
/* Put self-ref for this index due to error. */
relay_index_put(index);
+ index = NULL;
goto end;
}
} else {
/* Put self-ref for this index due to error. */
relay_index_put(index);
+ index = NULL;
ret = -1;
}
end:
stream->prev_seq = net_seq_num;
+ ret = check_rotate_stream(stream);
+ if (ret < 0) {
+ ERR("Check rotate stream");
+ goto end_stream_unlock;
+ }
+
end_stream_unlock:
close_requested = stream->close_requested;
pthread_mutex_unlock(&stream->lock);
destroy_conn,
sock_n.node) {
health_code_update();
+
+ if (session_abort(destroy_conn->session)) {
+ assert(0);
+ }
+
/*
* No need to grab another ref, because we own
* destroy_conn.
health_app_destroy(health_relayd);
exit_health_app_create:
exit_options:
+ /*
+ * Wait for all pending call_rcu work to complete before tearing
+ * down data structures. call_rcu worker may be trying to
+ * perform lookups in those structures.
+ */
+ rcu_barrier();
relayd_cleanup();
/* Ensure all prior call_rcu are done. */
projects / deliverable / lttng-tools.git / blobdiff