/*
- * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
- * 2018 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
+ * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
+ * Copyright (C) 2018 Jérémie Galarneau <jeremie.galarneau@efficios.com>
*
- * 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
- * published by the Free Software Foundation.
+ * SPDX-License-Identifier: GPL-2.0-only
*
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
- * more details.
- *
- * You should have received a copy of the GNU General Public License along with
- * this program; if not, write to the Free Software Foundation, Inc., 51
- * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
#define _LGPL_SOURCE
#include "utils.h"
#include "lttng-sessiond.h"
+/*
+ * Return allocated full pathname of the session using the consumer trace path
+ * and subdir if available.
+ *
+ * The caller can safely free(3) the returned value. On error, NULL is
+ * returned.
+ */
+char *setup_channel_trace_path(struct consumer_output *consumer,
+ const char *session_path, size_t *consumer_path_offset)
+{
+ int ret;
+ char *pathname;
+
+ assert(consumer);
+ assert(session_path);
+
+ health_code_update();
+
+ /*
+ * Allocate the string ourself to make sure we never exceed
+ * LTTNG_PATH_MAX.
+ */
+ pathname = zmalloc(LTTNG_PATH_MAX);
+ if (!pathname) {
+ goto error;
+ }
+
+ /* Get correct path name destination */
+ if (consumer->type == CONSUMER_DST_NET &&
+ consumer->relay_major_version == 2 &&
+ consumer->relay_minor_version < 11) {
+ ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s/%s/%s",
+ consumer->dst.net.base_dir,
+ consumer->chunk_path, consumer->domain_subdir,
+ session_path);
+ *consumer_path_offset = 0;
+ } else {
+ ret = snprintf(pathname, LTTNG_PATH_MAX, "%s/%s",
+ consumer->domain_subdir, session_path);
+ *consumer_path_offset = strlen(consumer->domain_subdir) + 1;
+ }
+ DBG3("Consumer trace path relative to current trace chunk: \"%s\"",
+ pathname);
+ if (ret < 0) {
+ PERROR("Failed to format channel path");
+ goto error;
+ } else if (ret >= LTTNG_PATH_MAX) {
+ ERR("Truncation occurred while formatting channel path");
+ goto error;
+ }
+
+ return pathname;
+error:
+ free(pathname);
+ return NULL;
+}
+
/*
* Send a data payload using a given consumer socket of size len.
*
*
* Return 0 on success else a negative value on error.
*/
-int consumer_socket_send(struct consumer_socket *socket, void *msg, size_t len)
+int consumer_socket_send(
+ struct consumer_socket *socket, const void *msg, size_t len)
{
int fd;
ssize_t size;
switch (bits) {
case 64:
- consumer_fd = uatomic_read(&ust_consumerd64_fd);
+ consumer_fd = uatomic_read(&the_ust_consumerd64_fd);
break;
case 32:
- consumer_fd = uatomic_read(&ust_consumerd32_fd);
+ consumer_fd = uatomic_read(&the_ust_consumerd32_fd);
break;
default:
assert(0);
output->snapshot = src->snapshot;
output->relay_major_version = src->relay_major_version;
output->relay_minor_version = src->relay_minor_version;
+ output->relay_allows_clear = src->relay_allows_clear;
memcpy(&output->dst, &src->dst, sizeof(output->dst));
ret = consumer_copy_sockets(output, src);
if (ret < 0) {
* The consumer socket lock must be held by the caller.
*/
int consumer_send_msg(struct consumer_socket *sock,
- struct lttcomm_consumer_msg *msg)
+ const struct lttcomm_consumer_msg *msg)
{
int ret;
unsigned int switch_timer_interval,
unsigned int read_timer_interval,
unsigned int live_timer_interval,
+ bool is_in_live_session,
unsigned int monitor_timer_interval,
int output,
int type,
{
assert(msg);
- /* Zeroed structure */
+ /* Zeroed structure */
memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
msg->u.ask_channel.buffer_credentials.uid = UINT32_MAX;
msg->u.ask_channel.buffer_credentials.gid = UINT32_MAX;
- if (trace_chunk) {
+ if (trace_chunk) {
uint64_t chunk_id;
enum lttng_trace_chunk_status chunk_status;
chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
LTTNG_OPTIONAL_SET(&msg->u.ask_channel.chunk_id, chunk_id);
- }
- msg->u.ask_channel.buffer_credentials.uid = buffer_credentials->uid;
- msg->u.ask_channel.buffer_credentials.gid = buffer_credentials->gid;
+ }
+ msg->u.ask_channel.buffer_credentials.uid =
+ lttng_credentials_get_uid(buffer_credentials);
+ msg->u.ask_channel.buffer_credentials.gid =
+ lttng_credentials_get_gid(buffer_credentials);
msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
msg->u.ask_channel.subbuf_size = subbuf_size;
msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
msg->u.ask_channel.read_timer_interval = read_timer_interval;
msg->u.ask_channel.live_timer_interval = live_timer_interval;
+ msg->u.ask_channel.is_live = is_in_live_session;
msg->u.ask_channel.monitor_timer_interval = monitor_timer_interval;
msg->u.ask_channel.output = output;
msg->u.ask_channel.type = type;
uint64_t tracefile_count,
unsigned int monitor,
unsigned int live_timer_interval,
+ bool is_in_live_session,
unsigned int monitor_timer_interval,
struct lttng_trace_chunk *trace_chunk)
{
/* Zeroed structure */
memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
- if (trace_chunk) {
+ if (trace_chunk) {
uint64_t chunk_id;
enum lttng_trace_chunk_status chunk_status;
chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
LTTNG_OPTIONAL_SET(&msg->u.channel.chunk_id, chunk_id);
- }
+ }
/* Send channel */
msg->cmd_type = LTTNG_CONSUMER_ADD_CHANNEL;
msg->u.channel.tracefile_count = tracefile_count;
msg->u.channel.monitor = monitor;
msg->u.channel.live_timer_interval = live_timer_interval;
+ msg->u.channel.is_live = is_in_live_session;
msg->u.channel.monitor_timer_interval = monitor_timer_interval;
strncpy(msg->u.channel.pathname, pathname,
struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
enum lttng_stream_type type, uint64_t session_id,
const char *session_name, const char *hostname,
- int session_live_timer, const uint64_t *current_chunk_id,
- time_t session_creation_time)
+ const char *base_path, int session_live_timer,
+ const uint64_t *current_chunk_id, time_t session_creation_time,
+ bool session_name_contains_creation_time)
{
int ret;
struct lttcomm_consumer_msg msg;
}
if (type == LTTNG_STREAM_CONTROL) {
- ret = relayd_create_session(rsock,
- &msg.u.relayd_sock.relayd_session_id,
- session_name, hostname, session_live_timer,
- consumer->snapshot, session_id,
- sessiond_uuid, current_chunk_id,
- session_creation_time);
+ char output_path[LTTNG_PATH_MAX] = {};
+ uint64_t relayd_session_id;
+
+ ret = relayd_create_session(rsock, &relayd_session_id,
+ session_name, hostname, base_path,
+ session_live_timer, consumer->snapshot,
+ session_id, the_sessiond_uuid, current_chunk_id,
+ session_creation_time,
+ session_name_contains_creation_time,
+ output_path);
if (ret < 0) {
/* Close the control socket. */
(void) relayd_close(rsock);
goto error;
}
+ msg.u.relayd_sock.relayd_session_id = relayd_session_id;
+ DBG("Created session on relay, output path reply: %s",
+ output_path);
}
msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
}
DBG3("Sending relayd socket file descriptor to consumer");
- ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
+ ret = consumer_send_fds(consumer_sock, ALIGNED_CONST_PTR(rsock->sock.fd), 1);
if (ret < 0) {
goto error;
}
return ret;
}
+int consumer_open_channel_packets(struct consumer_socket *socket, uint64_t key)
+{
+ int ret;
+ const struct lttcomm_consumer_msg msg = {
+ .cmd_type = LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS,
+ .u.open_channel_packets.key = key,
+ };
+
+ assert(socket);
+
+ DBG("Consumer open channel packets: channel key = %" PRIu64, key);
+
+ health_code_update();
+
+ pthread_mutex_lock(socket->lock);
+ ret = consumer_send_msg(socket, &msg);
+ pthread_mutex_unlock(socket->lock);
+ if (ret < 0) {
+ goto error_socket;
+ }
+
+error_socket:
+ health_code_update();
+ return ret;
+}
+
+int consumer_clear_channel(struct consumer_socket *socket, uint64_t key)
+{
+ int ret;
+ struct lttcomm_consumer_msg msg;
+
+ assert(socket);
+
+ DBG("Consumer clear channel %" PRIu64, key);
+
+ memset(&msg, 0, sizeof(msg));
+ msg.cmd_type = LTTNG_CONSUMER_CLEAR_CHANNEL;
+ msg.u.clear_channel.key = key;
+
+ health_code_update();
+
+ pthread_mutex_lock(socket->lock);
+ ret = consumer_send_msg(socket, &msg);
+ if (ret < 0) {
+ goto error_socket;
+ }
+
+error_socket:
+ pthread_mutex_unlock(socket->lock);
+
+ health_code_update();
+ return ret;
+}
+
int consumer_init(struct consumer_socket *socket,
const lttng_uuid sessiond_uuid)
{
*/
int consumer_create_trace_chunk(struct consumer_socket *socket,
uint64_t relayd_id, uint64_t session_id,
- struct lttng_trace_chunk *chunk)
+ struct lttng_trace_chunk *chunk,
+ const char *domain_subdir)
{
int ret;
enum lttng_trace_chunk_status chunk_status;
struct lttng_credentials chunk_credentials;
- const struct lttng_directory_handle *chunk_directory_handle;
- int chunk_dirfd;
+ const struct lttng_directory_handle *chunk_directory_handle = NULL;
+ struct lttng_directory_handle *domain_handle = NULL;
+ int domain_dirfd;
const char *chunk_name;
bool chunk_name_overridden;
uint64_t chunk_id;
char creation_timestamp_buffer[ISO8601_STR_LEN];
const char *creation_timestamp_str = "(none)";
const bool chunk_has_local_output = relayd_id == -1ULL;
+ enum lttng_trace_chunk_status tc_status;
struct lttcomm_consumer_msg msg = {
.cmd_type = LTTNG_CONSUMER_CREATE_TRACE_CHUNK,
.u.create_trace_chunk.session_id = session_id,
msg.u.create_trace_chunk.chunk_id = chunk_id;
if (chunk_has_local_output) {
- chunk_status = lttng_trace_chunk_get_chunk_directory_handle(
+ chunk_status = lttng_trace_chunk_borrow_chunk_directory_handle(
chunk, &chunk_directory_handle);
if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
ret = -LTTNG_ERR_FATAL;
goto error;
}
-
- /*
- * This will only compile on platforms that support
- * dirfd (POSIX.2008). This is fine as the session daemon
- * is only built for such platforms.
- *
- * The ownership of the chunk directory handle's is maintained
- * by the trace chunk.
- */
- chunk_dirfd = lttng_directory_handle_get_dirfd(
- chunk_directory_handle);
- assert(chunk_dirfd >= 0);
-
chunk_status = lttng_trace_chunk_get_credentials(
chunk, &chunk_credentials);
if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
ret = -LTTNG_ERR_FATAL;
goto error;
}
+ tc_status = lttng_trace_chunk_create_subdirectory(
+ chunk, domain_subdir);
+ if (tc_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ PERROR("Failed to create chunk domain output directory \"%s\"",
+ domain_subdir);
+ ret = -LTTNG_ERR_FATAL;
+ goto error;
+ }
+ domain_handle = lttng_directory_handle_create_from_handle(
+ domain_subdir,
+ chunk_directory_handle);
+ if (!domain_handle) {
+ ret = -LTTNG_ERR_FATAL;
+ goto error;
+ }
+
+ /*
+ * This will only compile on platforms that support
+ * dirfd (POSIX.2008). This is fine as the session daemon
+ * is only built for such platforms.
+ *
+ * The ownership of the chunk directory handle's is maintained
+ * by the trace chunk.
+ */
+ domain_dirfd = lttng_directory_handle_get_dirfd(
+ domain_handle);
+ assert(domain_dirfd >= 0);
+
msg.u.create_trace_chunk.credentials.value.uid =
- chunk_credentials.uid;
+ lttng_credentials_get_uid(&chunk_credentials);
msg.u.create_trace_chunk.credentials.value.gid =
- chunk_credentials.gid;
+ lttng_credentials_get_gid(&chunk_credentials);
msg.u.create_trace_chunk.credentials.is_set = 1;
}
}
if (chunk_has_local_output) {
- DBG("Sending trace chunk directory fd to consumer");
+ DBG("Sending trace chunk domain directory fd to consumer");
health_code_update();
- ret = consumer_send_fds(socket, &chunk_dirfd, 1);
+ ret = consumer_send_fds(socket, &domain_dirfd, 1);
health_code_update();
if (ret < 0) {
ERR("Trace chunk creation error on consumer");
}
}
error:
+ lttng_directory_handle_put(domain_handle);
return ret;
}
*/
int consumer_close_trace_chunk(struct consumer_socket *socket,
uint64_t relayd_id, uint64_t session_id,
- struct lttng_trace_chunk *chunk)
+ struct lttng_trace_chunk *chunk,
+ char *closed_trace_chunk_path)
{
int ret;
enum lttng_trace_chunk_status chunk_status;
.cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK,
.u.close_trace_chunk.session_id = session_id,
};
+ struct lttcomm_consumer_close_trace_chunk_reply reply;
uint64_t chunk_id;
time_t close_timestamp;
enum lttng_trace_chunk_command_type close_command;
const char *close_command_name = "none";
+ struct lttng_dynamic_buffer path_reception_buffer;
assert(socket);
+ lttng_dynamic_buffer_init(&path_reception_buffer);
if (relayd_id != -1ULL) {
LTTNG_OPTIONAL_SET(
relayd_id, session_id, chunk_id, close_command_name);
health_code_update();
- ret = consumer_send_msg(socket, &msg);
+ ret = consumer_socket_send(socket, &msg, sizeof(struct lttcomm_consumer_msg));
if (ret < 0) {
ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
goto error;
}
-
+ ret = consumer_socket_recv(socket, &reply, sizeof(reply));
+ if (ret < 0) {
+ ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
+ goto error;
+ }
+ if (reply.path_length >= LTTNG_PATH_MAX) {
+ ERR("Invalid path returned by relay daemon: %" PRIu32 "bytes exceeds maximal allowed length of %d bytes",
+ reply.path_length, LTTNG_PATH_MAX);
+ ret = -LTTNG_ERR_INVALID_PROTOCOL;
+ goto error;
+ }
+ ret = lttng_dynamic_buffer_set_size(&path_reception_buffer,
+ reply.path_length);
+ if (ret) {
+ ERR("Failed to allocate reception buffer of path returned by the \"close trace chunk\" command");
+ ret = -LTTNG_ERR_NOMEM;
+ goto error;
+ }
+ ret = consumer_socket_recv(socket, path_reception_buffer.data,
+ path_reception_buffer.size);
+ if (ret < 0) {
+ ERR("Communication error while receiving path of closed trace chunk");
+ ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
+ goto error;
+ }
+ if (path_reception_buffer.data[path_reception_buffer.size - 1] != '\0') {
+ ERR("Invalid path returned by relay daemon: not null-terminated");
+ ret = -LTTNG_ERR_INVALID_PROTOCOL;
+ goto error;
+ }
+ if (closed_trace_chunk_path) {
+ /*
+ * closed_trace_chunk_path is assumed to have a length >=
+ * LTTNG_PATH_MAX
+ */
+ memcpy(closed_trace_chunk_path, path_reception_buffer.data,
+ path_reception_buffer.size);
+ }
error:
+ lttng_dynamic_buffer_reset(&path_reception_buffer);
health_code_update();
return ret;
}