X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=a179eeffd1f72f5cde0d1fcb013e70ae2ac4b7e2;hp=5906cdca5901358fd339cf24f6de366eb99d7369;hb=9bbfb88c5a3b6b581c81d2c9a9db14246f84675a;hpb=d68c9a04537b683991a7355b812b0af954008cf1 diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c index 5906cdca5..a179eeffd 100644 --- a/src/bin/lttng-sessiond/consumer.c +++ b/src/bin/lttng-sessiond/consumer.c @@ -1,5 +1,6 @@ /* * Copyright (C) 2012 - David Goulet + * 2018 - Jérémie Galarneau * * This program is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License, version 2 only, as @@ -735,7 +736,8 @@ error: * * The consumer socket lock must be held by the caller. */ -int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd) +int consumer_send_fds(struct consumer_socket *sock, const int *fds, + size_t nb_fd) { int ret; @@ -833,7 +835,8 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, uint32_t ust_app_uid, int64_t blocking_timeout, const char *root_shm_path, - const char *shm_path) + const char *shm_path, + uint64_t trace_archive_id) { assert(msg); @@ -862,6 +865,7 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, msg->u.ask_channel.monitor = monitor; msg->u.ask_channel.ust_app_uid = ust_app_uid; msg->u.ask_channel.blocking_timeout = blocking_timeout; + msg->u.ask_channel.trace_archive_id = trace_archive_id; memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid)); @@ -889,8 +893,7 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, /* * Init channel communication message structure. */ -void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, - enum lttng_consumer_command cmd, +void consumer_init_add_channel_comm_msg(struct lttcomm_consumer_msg *msg, uint64_t channel_key, uint64_t session_id, const char *pathname, @@ -913,7 +916,7 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, memset(msg, 0, sizeof(struct lttcomm_consumer_msg)); /* Send channel */ - msg->cmd_type = cmd; + msg->cmd_type = LTTNG_CONSUMER_ADD_CHANNEL; msg->u.channel.channel_key = channel_key; msg->u.channel.session_id = session_id; msg->u.channel.uid = uid; @@ -939,20 +942,21 @@ void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, /* * Init stream communication message structure. */ -void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg, - enum lttng_consumer_command cmd, +void consumer_init_add_stream_comm_msg(struct lttcomm_consumer_msg *msg, uint64_t channel_key, uint64_t stream_key, - int cpu) + int32_t cpu, + uint64_t trace_archive_id) { assert(msg); memset(msg, 0, sizeof(struct lttcomm_consumer_msg)); - msg->cmd_type = cmd; + msg->cmd_type = LTTNG_CONSUMER_ADD_STREAM; msg->u.stream.channel_key = channel_key; msg->u.stream.stream_key = stream_key; msg->u.stream.cpu = cpu; + msg->u.stream.trace_archive_id = trace_archive_id; } void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg, @@ -973,7 +977,7 @@ void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg, */ int consumer_send_stream(struct consumer_socket *sock, struct consumer_output *dst, struct lttcomm_consumer_msg *msg, - int *fds, size_t nb_fd) + const int *fds, size_t nb_fd) { int ret; @@ -1078,10 +1082,6 @@ int consumer_send_pipe(struct consumer_socket *consumer_sock, pipe_name = "channel monitor"; command_name = "SET_CHANNEL_MONITOR_PIPE"; break; - case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE: - pipe_name = "channel rotate"; - command_name = "SET_CHANNEL_ROTATE_PIPE"; - break; default: ERR("Unexpected command received in %s (cmd = %d)", __func__, (int) cmd); @@ -1121,13 +1121,6 @@ int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock, LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe); } -int consumer_send_channel_rotate_pipe(struct consumer_socket *consumer_sock, - int pipe) -{ - return consumer_send_pipe(consumer_sock, - LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE, pipe); -} - /* * Set consumer subdirectory using the session name and a generated datetime if * needed. This is appended to the current subdirectory. @@ -1429,7 +1422,8 @@ end: */ 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, uint64_t nb_packets_per_stream) + const char *session_path, int wait, uint64_t nb_packets_per_stream, + uint64_t trace_archive_id) { int ret; struct lttcomm_consumer_msg msg; @@ -1445,6 +1439,7 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, msg.u.snapshot_channel.key = key; msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream; msg.u.snapshot_channel.metadata = metadata; + msg.u.snapshot_channel.trace_archive_id = trace_archive_id; if (output->consumer->type == CONSUMER_DST_NET) { msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index; @@ -1460,6 +1455,16 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, if (ret < 0) { ret = -LTTNG_ERR_NOMEM; goto error; + } else if (ret >= sizeof(msg.u.snapshot_channel.pathname)) { + ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%i bytes required) with path \"%s/%s/%s-%s-%" PRIu64 "%s\"", + sizeof(msg.u.snapshot_channel.pathname), + ret, output->consumer->dst.net.base_dir, + output->consumer->subdir, + output->name, output->datetime, + output->nb_snapshot, + session_path); + ret = -LTTNG_ERR_SNAPSHOT_FAIL; + goto error; } } else { ret = snprintf(msg.u.snapshot_channel.pathname, @@ -1472,7 +1477,16 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, if (ret < 0) { ret = -LTTNG_ERR_NOMEM; goto error; + } else if (ret >= sizeof(msg.u.snapshot_channel.pathname)) { + ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%i bytes required) with path \"%s/%s-%s-%" PRIu64 "%s\"", + sizeof(msg.u.snapshot_channel.pathname), + ret, output->consumer->dst.session_root_path, + output->name, output->datetime, output->nb_snapshot, + session_path); + ret = -LTTNG_ERR_SNAPSHOT_FAIL; + goto error; } + msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL; /* Create directory. Ignore if exist. */ @@ -1491,6 +1505,14 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key, ret = consumer_send_msg(socket, &msg); pthread_mutex_unlock(socket->lock); if (ret < 0) { + switch (-ret) { + case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND: + ret = -LTTNG_ERR_CHAN_NOT_FOUND; + break; + default: + ret = -LTTNG_ERR_SNAPSHOT_FAIL; + break; + } goto error; } @@ -1625,8 +1647,7 @@ end: int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key, uid_t uid, gid_t gid, struct consumer_output *output, char *domain_path, bool is_metadata_channel, - uint64_t new_chunk_id, - bool *rotate_pending_relay) + uint64_t new_chunk_id) { int ret; struct lttcomm_consumer_msg msg; @@ -1650,10 +1671,9 @@ int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key, output->chunk_path, domain_path); if (ret < 0 || ret == sizeof(msg.u.rotate_channel.pathname)) { ERR("Failed to format channel path name when asking consumer to rotate channel"); - ret = -1; + ret = -LTTNG_ERR_INVALID; goto error; } - *rotate_pending_relay = true; } else { msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL; ret = snprintf(msg.u.rotate_channel.pathname, @@ -1662,7 +1682,7 @@ int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key, output->chunk_path, domain_path); if (ret < 0 || ret == sizeof(msg.u.rotate_channel.pathname)) { ERR("Failed to format channel path name when asking consumer to rotate channel"); - ret = -1; + ret = -LTTNG_ERR_INVALID; goto error; } } @@ -1670,9 +1690,16 @@ int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key, health_code_update(); ret = consumer_send_msg(socket, &msg); if (ret < 0) { + switch (-ret) { + case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND: + ret = -LTTNG_ERR_CHAN_NOT_FOUND; + break; + default: + ret = -LTTNG_ERR_ROTATION_FAIL_CONSUMER; + break; + } goto error; } - error: pthread_mutex_unlock(socket->lock); health_code_update(); @@ -1698,7 +1725,7 @@ int consumer_rotate_rename(struct consumer_socket *socket, uint64_t session_id, if (old_path_length >= sizeof(msg.u.rotate_rename.old_path)) { ERR("consumer_rotate_rename: old path length (%zu bytes) exceeds the maximal length allowed by the consumer protocol (%zu bytes)", old_path_length + 1, sizeof(msg.u.rotate_rename.old_path)); - ret = -1; + ret = -LTTNG_ERR_INVALID; goto error; } @@ -1706,7 +1733,7 @@ int consumer_rotate_rename(struct consumer_socket *socket, uint64_t session_id, if (new_path_length >= sizeof(msg.u.rotate_rename.new_path)) { ERR("consumer_rotate_rename: new path length (%zu bytes) exceeds the maximal length allowed by the consumer protocol (%zu bytes)", new_path_length + 1, sizeof(msg.u.rotate_rename.new_path)); - ret = -1; + ret = -LTTNG_ERR_INVALID; goto error; } @@ -1726,24 +1753,68 @@ int consumer_rotate_rename(struct consumer_socket *socket, uint64_t session_id, health_code_update(); ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + ret = -LTTNG_ERR_ROTATE_RENAME_FAIL_CONSUMER; + goto error; + } + +error: + health_code_update(); + return ret; +} + +/* + * Ask the consumer if a rotation is locally pending. Must be called with the + * socket lock held. + * + * Return 1 if the rotation is still pending, 0 if finished, a negative value + * on error. + */ +int consumer_check_rotation_pending_local(struct consumer_socket *socket, + uint64_t session_id, uint64_t chunk_id) +{ + int ret; + struct lttcomm_consumer_msg msg; + uint32_t pending = 0; + + assert(socket); + + DBG("Asking consumer to locally check for pending rotation for session %" PRIu64 ", chunk id %" PRIu64, + session_id, chunk_id); + + memset(&msg, 0, sizeof(msg)); + msg.cmd_type = LTTNG_CONSUMER_CHECK_ROTATION_PENDING_LOCAL; + msg.u.check_rotation_pending_local.session_id = session_id; + msg.u.check_rotation_pending_local.chunk_id = chunk_id; + + health_code_update(); + ret = consumer_send_msg(socket, &msg); + if (ret < 0) { + ret = -LTTNG_ERR_ROTATION_PENDING_LOCAL_FAIL_CONSUMER; + goto error; + } + + ret = consumer_socket_recv(socket, &pending, sizeof(pending)); if (ret < 0) { goto error; } + ret = pending; + error: health_code_update(); return ret; } /* - * Ask the relay if a rotation is still pending. Must be called with the socket - * lock held. + * Ask the consumer if a rotation is pending on the relayd. Must be called with + * the socket lock held. * * Return 1 if the rotation is still pending, 0 if finished, a negative value * on error. */ -int consumer_rotate_pending_relay(struct consumer_socket *socket, - struct consumer_output *output, uint64_t session_id, +int consumer_check_rotation_pending_relay(struct consumer_socket *socket, + const struct consumer_output *output, uint64_t session_id, uint64_t chunk_id) { int ret; @@ -1752,19 +1823,20 @@ int consumer_rotate_pending_relay(struct consumer_socket *socket, assert(socket); - DBG("Consumer rotate pending on relay for session %" PRIu64 ", chunk id %" PRIu64, + DBG("Asking consumer to check for pending rotation on relay for session %" PRIu64 ", chunk id %" PRIu64, session_id, chunk_id); assert(output->type == CONSUMER_DST_NET); memset(&msg, 0, sizeof(msg)); - msg.cmd_type = LTTNG_CONSUMER_ROTATE_PENDING_RELAY; - msg.u.rotate_pending_relay.session_id = session_id; - msg.u.rotate_pending_relay.relayd_id = output->net_seq_index; - msg.u.rotate_pending_relay.chunk_id = chunk_id; + msg.cmd_type = LTTNG_CONSUMER_CHECK_ROTATION_PENDING_RELAY; + msg.u.check_rotation_pending_relay.session_id = session_id; + msg.u.check_rotation_pending_relay.relayd_id = output->net_seq_index; + msg.u.check_rotation_pending_relay.chunk_id = chunk_id; health_code_update(); ret = consumer_send_msg(socket, &msg); if (ret < 0) { + ret = -LTTNG_ERR_ROTATION_PENDING_RELAY_FAIL_CONSUMER; goto error; } @@ -1804,7 +1876,7 @@ int consumer_mkdir(struct consumer_socket *socket, uint64_t session_id, ret = snprintf(msg.u.mkdir.path, sizeof(msg.u.mkdir.path), "%s", path); if (ret < 0 || ret >= sizeof(msg.u.mkdir.path)) { ERR("Format path"); - ret = -1; + ret = -LTTNG_ERR_INVALID; goto error; } @@ -1817,6 +1889,7 @@ int consumer_mkdir(struct consumer_socket *socket, uint64_t session_id, health_code_update(); ret = consumer_send_msg(socket, &msg); if (ret < 0) { + ret = -LTTNG_ERR_MKDIR_FAIL_CONSUMER; goto error; }