X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=d3c561c9ec6932531fc3c46a1a464835dc45e12f;hp=f0fb2dc00f92967efdf7d22df4c7c61e98abd73b;hb=10a5031171c7bca5b4498c871b119e5a88b6a3fb;hpb=331744e34f56a5aec69b05d356d6901e67926acc diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index f0fb2dc00..d3c561c9e 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -32,6 +32,7 @@ #include "consumer.h" #include "health.h" #include "ust-app.h" +#include "utils.h" /* * Receive a reply command status message from the consumer. Consumer socket @@ -169,7 +170,7 @@ void consumer_output_send_destroy_relayd(struct consumer_output *consumer) assert(consumer); /* Destroy any relayd connection */ - if (consumer && consumer->type == CONSUMER_DST_NET) { + if (consumer->type == CONSUMER_DST_NET) { rcu_read_lock(); cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket, node.node) { @@ -225,6 +226,8 @@ int consumer_create_socket(struct consumer_data *data, rcu_read_unlock(); } + socket->type = data->type; + DBG3("Consumer socket created (fd: %d) and added to output", data->cmd_sock); @@ -407,6 +410,8 @@ error: /* * Delete the consumer_output object from the list and free the ptr. + * + * Should *NOT* be called with RCU read-side lock held. */ void consumer_destroy_output(struct consumer_output *obj) { @@ -426,7 +431,7 @@ void consumer_destroy_output(struct consumer_output *obj) rcu_read_unlock(); /* Finally destroy HT */ - lttng_ht_destroy(obj->socks); + ht_cleanup_push(obj->socks); } free(obj); @@ -434,12 +439,13 @@ void consumer_destroy_output(struct consumer_output *obj) /* * Copy consumer output and returned the newly allocated copy. + * + * Should *NOT* be called with RCU read-side lock held. */ struct consumer_output *consumer_copy_output(struct consumer_output *obj) { + int ret; struct lttng_ht *tmp_ht_ptr; - struct lttng_ht_iter iter; - struct consumer_socket *socket, *copy_sock; struct consumer_output *output; assert(obj); @@ -456,27 +462,63 @@ struct consumer_output *consumer_copy_output(struct consumer_output *obj) /* Putting back the HT pointer and start copying socket(s). */ output->socks = tmp_ht_ptr; + ret = consumer_copy_sockets(output, obj); + if (ret < 0) { + goto malloc_error; + } + +error: + return output; + +malloc_error: + consumer_destroy_output(output); + return NULL; +} + +/* + * Copy consumer sockets from src to dst. + * + * Return 0 on success or else a negative value. + */ +int consumer_copy_sockets(struct consumer_output *dst, + struct consumer_output *src) +{ + int ret = 0; + struct lttng_ht_iter iter; + struct consumer_socket *socket, *copy_sock; + + assert(dst); + assert(src); + rcu_read_lock(); - cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) { + cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) { + /* Ignore socket that are already there. */ + copy_sock = consumer_find_socket(socket->fd, dst); + if (copy_sock) { + continue; + } + /* Create new socket object. */ copy_sock = consumer_allocate_socket(socket->fd); if (copy_sock == NULL) { rcu_read_unlock(); - goto malloc_error; + ret = -ENOMEM; + goto error; } copy_sock->registered = socket->registered; + /* + * This is valid because this lock is shared accross all consumer + * object being the global lock of the consumer data structure of the + * session daemon. + */ copy_sock->lock = socket->lock; - consumer_add_socket(copy_sock, output); + consumer_add_socket(copy_sock, dst); } rcu_read_unlock(); error: - return output; - -malloc_error: - consumer_destroy_output(output); - return NULL; + return ret; } /* @@ -504,6 +546,12 @@ int consumer_set_network_uri(struct consumer_output *obj, if (uri->port == 0) { /* Assign default port. */ uri->port = DEFAULT_NETWORK_CONTROL_PORT; + } else { + if (obj->dst.net.data_isset && uri->port == + obj->dst.net.data.port) { + ret = -LTTNG_ERR_INVALID; + goto error; + } } DBG3("Consumer control URI set with port %d", uri->port); break; @@ -513,11 +561,18 @@ int consumer_set_network_uri(struct consumer_output *obj, if (uri->port == 0) { /* Assign default port. */ uri->port = DEFAULT_NETWORK_DATA_PORT; + } else { + if (obj->dst.net.control_isset && uri->port == + obj->dst.net.control.port) { + ret = -LTTNG_ERR_INVALID; + goto error; + } } DBG3("Consumer data URI set with port %d", uri->port); break; default: ERR("Set network uri type unknown %d", uri->stype); + ret = -LTTNG_ERR_INVALID; goto error; } @@ -553,6 +608,7 @@ int consumer_set_network_uri(struct consumer_output *obj, } if (ret < 0) { PERROR("snprintf set consumer uri subdir"); + ret = -LTTNG_ERR_NOMEM; goto error; } @@ -564,7 +620,7 @@ int consumer_set_network_uri(struct consumer_output *obj, equal: return 1; error: - return -1; + return ret; } /* @@ -663,7 +719,11 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, uint64_t relayd_id, uint64_t key, unsigned char *uuid, - uint32_t chan_id) + uint32_t chan_id, + uint64_t tracefile_size, + uint64_t tracefile_count, + uint64_t session_id_per_pid, + unsigned int monitor) { assert(msg); @@ -679,17 +739,23 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.ask_channel.output = output; msg->u.ask_channel.type = type; msg->u.ask_channel.session_id = session_id; + msg->u.ask_channel.session_id_per_pid = session_id_per_pid; msg->u.ask_channel.uid = uid; msg->u.ask_channel.gid = gid; msg->u.ask_channel.relayd_id = relayd_id; msg->u.ask_channel.key = key; msg->u.ask_channel.chan_id = chan_id; + msg->u.ask_channel.tracefile_size = tracefile_size; + msg->u.ask_channel.tracefile_count = tracefile_count; + msg->u.ask_channel.monitor = monitor; memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid)); - strncpy(msg->u.ask_channel.pathname, pathname, - sizeof(msg->u.ask_channel.pathname)); - msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0'; + if (pathname) { + strncpy(msg->u.ask_channel.pathname, pathname, + sizeof(msg->u.ask_channel.pathname)); + msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0'; + } strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name)); msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0'; @@ -709,7 +775,10 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, const char *name, unsigned int nb_init_streams, enum lttng_event_output output, - int type) + int type, + uint64_t tracefile_size, + uint64_t tracefile_count, + unsigned int monitor) { assert(msg); @@ -726,6 +795,9 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.channel.nb_init_streams = nb_init_streams; msg->u.channel.output = output; msg->u.channel.type = type; + msg->u.channel.tracefile_size = tracefile_size; + msg->u.channel.tracefile_count = tracefile_count; + msg->u.channel.monitor = monitor; strncpy(msg->u.channel.pathname, pathname, sizeof(msg->u.channel.pathname)); @@ -797,14 +869,14 @@ error: * On success return positive value. On error, negative value. */ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, - struct lttcomm_sock *sock, struct consumer_output *consumer, + struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer, enum lttng_stream_type type, uint64_t session_id) { int ret; struct lttcomm_consumer_msg msg; /* Code flow error. Safety net. */ - assert(sock); + assert(rsock); assert(consumer); assert(consumer_sock); @@ -823,13 +895,13 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, msg.u.relayd_sock.net_index = consumer->net_seq_index; msg.u.relayd_sock.type = type; msg.u.relayd_sock.session_id = session_id; - memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock)); + memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock)); DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd); ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg)); if (ret < 0) { /* The above call will print a PERROR on error. */ - DBG("Error when sending relayd sockets on sock %d", sock->fd); + DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd); goto error; } @@ -839,7 +911,7 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, } DBG3("Sending relayd socket file descriptor to consumer"); - ret = consumer_send_fds(consumer_sock, &sock->fd, 1); + ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1); if (ret < 0) { goto error; } @@ -1099,7 +1171,7 @@ int consumer_push_metadata(struct consumer_socket *socket, goto end; } - DBG3("Consumer pushing metadata on sock %d of len %lu", socket->fd, len); + DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len); ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len); if (ret < 0) { @@ -1116,3 +1188,72 @@ end: health_code_update(); return ret; } + +/* + * Ask the consumer to snapshot a specific channel using the key. + * + * Return 0 on success or else a negative error. + */ +int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, + struct snapshot_output *output, int metadata, uid_t uid, gid_t gid, + const char *session_path, int wait) +{ + int ret; + struct lttcomm_consumer_msg msg; + + assert(socket); + assert(socket->fd >= 0); + assert(output); + assert(output->consumer); + + DBG("Consumer snapshot channel key %" PRIu64, key); + + memset(&msg, 0, sizeof(msg)); + msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL; + msg.u.snapshot_channel.key = key; + msg.u.snapshot_channel.max_size = output->max_size; + msg.u.snapshot_channel.metadata = metadata; + + if (output->consumer->type == CONSUMER_DST_NET) { + msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index; + msg.u.snapshot_channel.use_relayd = 1; + ret = snprintf(msg.u.snapshot_channel.pathname, + sizeof(msg.u.snapshot_channel.pathname), "%s/%s-%s%s", + output->consumer->subdir, output->name, output->datetime, + session_path); + if (ret < 0) { + ret = -LTTNG_ERR_NOMEM; + goto error; + } + } else { + ret = snprintf(msg.u.snapshot_channel.pathname, + sizeof(msg.u.snapshot_channel.pathname), "%s/%s-%s%s", + output->consumer->dst.trace_path, output->name, + output->datetime, session_path); + if (ret < 0) { + ret = -LTTNG_ERR_NOMEM; + goto error; + } + msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL; + + /* Create directory. Ignore if exist. */ + ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname, + S_IRWXU | S_IRWXG, uid, gid); + if (ret < 0) { + if (ret != -EEXIST) { + ERR("Trace directory creation error"); + goto error; + } + } + } + + health_code_update(); + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + goto error; + } + +error: + health_code_update(); + return ret; +}