* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
-#define _GNU_SOURCE
#define _LGPL_SOURCE
#include <getopt.h>
#include <grp.h>
#include <urcu/uatomic.h>
#include <unistd.h>
#include <fcntl.h>
-#include <config.h>
#include <lttng/lttng.h>
#include <common/common.h>
#include <common/sessiond-comm/relayd.h>
#include <common/uri.h>
#include <common/utils.h>
-#include <common/config/config.h>
+#include <common/config/session-config.h>
#include <urcu/rculist.h>
#include "cmd.h"
*/
#define NR_LTTNG_RELAY_READY 3
static int lttng_relay_ready = NR_LTTNG_RELAY_READY;
+
+/* Size of receive buffer. */
+#define RECV_DATA_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. */
{ "output", 1, 0, 'o', },
{ "verbose", 0, 0, 'v', },
{ "config", 1, 0, 'f' },
+ { "version", 0, 0, 'V' },
{ NULL, 0, 0, 0, },
};
-static const char *config_ignore_options[] = { "help", "config" };
-
-/*
- * usage function on stderr
- */
-static void usage(void)
-{
- fprintf(stderr, "Usage: %s OPTIONS\n\nOptions:\n", progname);
- fprintf(stderr, " -h, --help Display this usage.\n");
- fprintf(stderr, " -d, --daemonize Start as a daemon.\n");
- fprintf(stderr, " -b, --background Start as a daemon, keeping console open.\n");
- fprintf(stderr, " -C, --control-port URL Control port listening.\n");
- fprintf(stderr, " -D, --data-port URL Data port listening.\n");
- fprintf(stderr, " -L, --live-port URL Live view port listening.\n");
- fprintf(stderr, " -o, --output PATH Output path for traces. Must use an absolute path.\n");
- fprintf(stderr, " -v, --verbose Verbose mode. Activate DBG() macro.\n");
- fprintf(stderr, " -g, --group NAME Specify the tracing group name. (default: tracing)\n");
- fprintf(stderr, " -f --config Load daemon configuration file\n");
-}
+static const char *config_ignore_options[] = { "help", "config", "version" };
/*
* Take an option from the getopt output and set it in the right variable to be
}
break;
case 'h':
- usage();
+ ret = utils_show_man_page(8, "lttng-relayd");
+ if (ret) {
+ ERR("Cannot view man page lttng-relayd(8)");
+ perror("exec");
+ }
exit(EXIT_FAILURE);
+ case 'V':
+ fprintf(stdout, "%s\n", VERSION);
+ exit(EXIT_SUCCESS);
case 'o':
if (lttng_is_setuid_setgid()) {
WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
/*
* config_entry_handler_cb used to handle options read from a config file.
- * See config_entry_handler_cb comment in common/config/config.h for the
+ * See config_entry_handler_cb comment in common/config/session-config.h for the
* return value conventions.
*/
static int config_entry_handler(const struct config_entry *entry, void *unused)
static void sighandler(int sig)
{
switch (sig) {
- case SIGPIPE:
- DBG("SIGPIPE caught");
- return;
case SIGINT:
DBG("SIGINT caught");
if (lttng_relay_stop_threads()) {
return ret;
}
- sa.sa_handler = sighandler;
sa.sa_mask = sigset;
sa.sa_flags = 0;
+
+ sa.sa_handler = sighandler;
if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
PERROR("sigaction");
return ret;
return ret;
}
- if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
+ if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
PERROR("sigaction");
return ret;
}
- if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
+ sa.sa_handler = SIG_IGN;
+ if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
PERROR("sigaction");
return ret;
}
* Set index data from the control port to a given index object.
*/
static int set_index_control_data(struct relay_index *index,
- struct lttcomm_relayd_index *data)
+ struct lttcomm_relayd_index *data,
+ struct relay_connection *conn)
{
struct ctf_packet_index index_data;
index_data.timestamp_end = data->timestamp_end;
index_data.events_discarded = data->events_discarded;
index_data.stream_id = data->stream_id;
+
+ if (conn->minor >= 8) {
+ index->index_data.stream_instance_id = data->stream_instance_id;
+ index->index_data.packet_seq_num = data->packet_seq_num;
+ }
+
return relay_index_set_data(index, &index_data);
}
int ret = 0, send_ret;
struct relay_session *session;
struct lttcomm_relayd_status_session reply;
- char session_name[NAME_MAX];
- char hostname[HOST_NAME_MAX];
+ char session_name[LTTNG_NAME_MAX];
+ char hostname[LTTNG_HOST_NAME_MAX];
uint32_t live_timer = 0;
bool snapshot = false;
- memset(session_name, 0, NAME_MAX);
- memset(hostname, 0, HOST_NAME_MAX);
+ memset(session_name, 0, LTTNG_NAME_MAX);
+ memset(hostname, 0, LTTNG_HOST_NAME_MAX);
memset(&reply, 0, sizeof(reply));
return ret;
}
+/*
+ * relay_reset_metadata: reset a metadata stream
+ */
+static
+int relay_reset_metadata(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret, send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_reset_metadata stream_info;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+
+ DBG("Reset metadata received");
+
+ if (!session || conn->version_check_done == 0) {
+ ERR("Trying to reset a metadata stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ ret = conn->sock->ops->recvmsg(conn->sock, &stream_info,
+ sizeof(struct lttcomm_relayd_reset_metadata), 0);
+ if (ret < sizeof(struct lttcomm_relayd_reset_metadata)) {
+ 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 reset_metadata struct "
+ "size : %d", ret);
+ }
+ ret = -1;
+ goto end_no_session;
+ }
+ DBG("Update metadata to version %" PRIu64, be64toh(stream_info.version));
+
+ /* Unsupported for live sessions for now. */
+ if (session->live_timer != 0) {
+ ret = -1;
+ goto end;
+ }
+
+ stream = stream_get_by_id(be64toh(stream_info.stream_id));
+ if (!stream) {
+ ret = -1;
+ goto end;
+ }
+ pthread_mutex_lock(&stream->lock);
+ if (!stream->is_metadata) {
+ ret = -1;
+ goto end_unlock;
+ }
+
+ ret = utils_rotate_stream_file(stream->path_name, stream->channel_name,
+ 0, 0, -1, -1, stream->stream_fd->fd, NULL,
+ &stream->stream_fd->fd);
+ if (ret < 0) {
+ ERR("Failed to rotate metadata file %s of channel %s",
+ stream->path_name, stream->channel_name);
+ goto end_unlock;
+ }
+
+end_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 reset metadata reply");
+ ret = send_ret;
+ }
+
+end_no_session:
+ return ret;
+}
+
/*
* relay_unknown_command: send -1 if received unknown command
*/
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);
ERR("relay_index_get_by_id_or_create index NULL");
goto end_stream_put;
}
- if (set_index_control_data(index, &index_info)) {
+ if (set_index_control_data(index, &index_info, conn)) {
ERR("set_index_control_data error");
relay_index_put(index);
ret = -1;
case RELAYD_STREAMS_SENT:
ret = relay_streams_sent(recv_hdr, conn);
break;
+ case RELAYD_RESET_METADATA:
+ ret = relay_reset_metadata(recv_hdr, conn);
+ break;
case RELAYD_UPDATE_SYNC_INFO:
default:
ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
/* Get data offset because we are about to update the index. */
data_offset = htobe64(stream->tracefile_size_current);
- DBG("handle_index_data: stream %" PRIu64 " data offset %" PRIu64,
- stream->stream_handle, stream->tracefile_size_current);
+ DBG("handle_index_data: stream %" PRIu64 " net_seq_num %" PRIu64 " data offset %" PRIu64,
+ stream->stream_handle, net_seq_num, stream->tracefile_size_current);
/*
* Lookup for an existing index for that stream id/sequence
goto end;
}
- if (rotate_index || !stream->index_fd) {
- int fd;
+ if (rotate_index || !stream->index_file) {
+ uint32_t major, minor;
- /* Put ref on previous index_fd. */
- if (stream->index_fd) {
- stream_fd_put(stream->index_fd);
- stream->index_fd = NULL;
+ /* Put ref on previous index_file. */
+ if (stream->index_file) {
+ lttng_index_file_put(stream->index_file);
+ stream->index_file = NULL;
}
-
- fd = index_create_file(stream->path_name, stream->channel_name,
+ 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));
- 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) {
+ 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;
- if (close(fd)) {
- PERROR("Error closing FD %d", fd);
- }
/* 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:
uint32_t data_size;
struct relay_session *session;
bool new_stream = false, close_requested = false;
+ size_t chunk_size = RECV_DATA_BUFFER_SIZE;
+ size_t recv_off = 0;
+ char data_buffer[chunk_size];
ret = conn->sock->ops->recvmsg(conn->sock, &data_hdr,
sizeof(struct lttcomm_relayd_data_hdr), 0);
stream_id = be64toh(data_hdr.stream_id);
stream = stream_get_by_id(stream_id);
if (!stream) {
+ ERR("relay_process_data: Cannot find stream %" PRIu64, stream_id);
ret = -1;
goto end;
}
session = stream->trace->session;
data_size = be32toh(data_hdr.data_size);
- if (data_buffer_size < data_size) {
- char *tmp_data_ptr;
-
- tmp_data_ptr = realloc(data_buffer, data_size);
- if (!tmp_data_ptr) {
- ERR("Allocating data buffer");
- free(data_buffer);
- ret = -1;
- goto end_stream_put;
- }
- data_buffer = tmp_data_ptr;
- data_buffer_size = data_size;
- }
- memset(data_buffer, 0, data_size);
net_seq_num = be64toh(data_hdr.net_seq_num);
DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
data_size, stream_id, net_seq_num);
- ret = conn->sock->ops->recvmsg(conn->sock, data_buffer, data_size, 0);
- if (ret <= 0) {
- if (ret == 0) {
- /* Orderly shutdown. Not necessary to print an error. */
- DBG("Socket %d did an orderly shutdown", conn->sock->fd);
- }
- ret = -1;
- goto end_stream_put;
- }
pthread_mutex_lock(&stream->lock);
if (session->minor >= 4 && !session->snapshot) {
ret = handle_index_data(stream, net_seq_num, rotate_index);
if (ret < 0) {
+ ERR("handle_index_data: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d",
+ stream->stream_handle, net_seq_num, ret);
goto end_stream_unlock;
}
}
- /* Write data to stream output fd. */
- size_ret = lttng_write(stream->stream_fd->fd, data_buffer, data_size);
- if (size_ret < data_size) {
- ERR("Relay error writing data to file");
- ret = -1;
- goto end_stream_unlock;
- }
+ for (recv_off = 0; recv_off < data_size; recv_off += chunk_size) {
+ size_t recv_size = min(data_size - recv_off, chunk_size);
- DBG2("Relay wrote %zd bytes to tracefile for stream id %" PRIu64,
- size_ret, stream->stream_handle);
+ ret = conn->sock->ops->recvmsg(conn->sock, data_buffer, recv_size, 0);
+ if (ret <= 0) {
+ if (ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown", conn->sock->fd);
+ } else {
+ ERR("Socket %d error %d", conn->sock->fd, ret);
+ }
+ ret = -1;
+ goto end_stream_unlock;
+ }
+
+ /* Write data to stream output fd. */
+ size_ret = lttng_write(stream->stream_fd->fd, data_buffer,
+ recv_size);
+ if (size_ret < recv_size) {
+ ERR("Relay error writing data to file");
+ ret = -1;
+ goto end_stream_unlock;
+ }
+
+ DBG2("Relay wrote %zd bytes to tracefile for stream id %" PRIu64,
+ size_ret, stream->stream_handle);
+ }
ret = write_padding_to_file(stream->stream_fd->fd,
be32toh(data_hdr.padding_size));
if (ret < 0) {
+ ERR("write_padding_to_file: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d",
+ stream->stream_handle, net_seq_num, ret);
goto end_stream_unlock;
}
stream->tracefile_size_current +=
uatomic_set(&session->new_streams, 1);
pthread_mutex_unlock(&session->lock);
}
-end_stream_put:
stream_put(stream);
end:
return ret;
DBG("Thread exited with error");
}
DBG("Worker thread cleanup complete");
- free(data_buffer);
error_testpoint:
if (err) {
health_error();
}
}
-
/* Initialize thread health monitoring */
health_relayd = health_app_create(NR_HEALTH_RELAYD_TYPES);
if (!health_relayd) {
goto exit_init_data;
}
- /* Check if daemon is UID = 0 */
- if (!getuid()) {
- if (control_uri->port < 1024 || data_uri->port < 1024 || live_uri->port < 1024) {
- ERR("Need to be root to use ports < 1024");
- retval = -1;
- goto exit_init_data;
- }
- }
-
/* Setup the thread apps communication pipe. */
if (create_relay_conn_pipe()) {
retval = -1;
/* Init relay command queue. */
cds_wfcq_init(&relay_conn_queue.head, &relay_conn_queue.tail);
- /* Set up max poll set size */
- if (lttng_poll_set_max_size()) {
- retval = -1;
- goto exit_init_data;
- }
-
/* Initialize communication library */
lttcomm_init();
lttcomm_inet_init();
}
/* Create thread to manage the client socket */
- ret = pthread_create(&health_thread, NULL,
+ ret = pthread_create(&health_thread, default_pthread_attr(),
thread_manage_health, (void *) NULL);
if (ret) {
errno = ret;
}
/* Setup the dispatcher thread */
- ret = pthread_create(&dispatcher_thread, NULL,
+ ret = pthread_create(&dispatcher_thread, default_pthread_attr(),
relay_thread_dispatcher, (void *) NULL);
if (ret) {
errno = ret;
}
/* Setup the worker thread */
- ret = pthread_create(&worker_thread, NULL,
+ ret = pthread_create(&worker_thread, default_pthread_attr(),
relay_thread_worker, NULL);
if (ret) {
errno = ret;
}
/* Setup the listener thread */
- ret = pthread_create(&listener_thread, NULL,
+ ret = pthread_create(&listener_thread, default_pthread_attr(),
relay_thread_listener, (void *) NULL);
if (ret) {
errno = ret;
projects / lttng-tools.git / blobdiff