X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=79fdbf1c8b632bc9f049e27511f7b593314bd0f7;hb=6a00837f8cb0431a2ad90974d67fae138ea97dd5;hp=0cf43d2ca3da002ffb7d68aef558316c75b9006a;hpb=6151a90fe7fa3dea52c57771df9083e56de7a60b;p=lttng-tools.git diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index 0cf43d2ca..79fdbf1c8 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); @@ -405,8 +408,32 @@ error: return output; } +/* + * Iterate over the consumer output socket hash table and destroy them. The + * socket file descriptor are only closed if the consumer output was + * registered meaning it's an external consumer. + */ +void consumer_destroy_output_sockets(struct consumer_output *obj) +{ + struct lttng_ht_iter iter; + struct consumer_socket *socket; + + if (!obj->socks) { + return; + } + + rcu_read_lock(); + cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) { + consumer_del_socket(socket, obj); + consumer_destroy_socket(socket); + } + rcu_read_unlock(); +} + /* * 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) { @@ -414,19 +441,11 @@ void consumer_destroy_output(struct consumer_output *obj) return; } - if (obj->socks) { - struct lttng_ht_iter iter; - struct consumer_socket *socket; - - rcu_read_lock(); - cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) { - consumer_del_socket(socket, obj); - consumer_destroy_socket(socket); - } - rcu_read_unlock(); + consumer_destroy_output_sockets(obj); + if (obj->socks) { /* Finally destroy HT */ - lttng_ht_destroy(obj->socks); + ht_cleanup_push(obj->socks); } free(obj); @@ -434,12 +453,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 +476,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 +560,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 +575,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 +622,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 +634,7 @@ int consumer_set_network_uri(struct consumer_output *obj, equal: return 1; error: - return -1; + return ret; } /* @@ -665,7 +735,9 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, unsigned char *uuid, uint32_t chan_id, uint64_t tracefile_size, - uint64_t tracefile_count) + uint64_t tracefile_count, + uint64_t session_id_per_pid, + unsigned int monitor) { assert(msg); @@ -681,6 +753,7 @@ 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; @@ -688,12 +761,15 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, 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'; @@ -715,7 +791,8 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, enum lttng_event_output output, int type, uint64_t tracefile_size, - uint64_t tracefile_count) + uint64_t tracefile_count, + unsigned int monitor) { assert(msg); @@ -734,6 +811,7 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, 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)); @@ -1107,7 +1185,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) { @@ -1124,3 +1202,74 @@ 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 max_stream_size) +{ + 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_stream_size = max_stream_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-%" PRIu64 "%s", output->consumer->subdir, + output->name, output->datetime, output->nb_snapshot, + 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-%" PRIu64 "%s", output->consumer->dst.trace_path, + output->name, output->datetime, output->nb_snapshot, + 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; +}