lttng_uuid sessiond_uuid = {};
LTTNG_OPTIONAL(uint64_t) id_sessiond = {};
LTTNG_OPTIONAL(uint64_t) current_chunk_id = {};
+ LTTNG_OPTIONAL(time_t) creation_time = {};
if (conn->minor < 4) {
/* From 2.1 to 2.3 */
hostname, &live_timer, &snapshot);
} else {
bool has_current_chunk;
+ uint64_t current_chunk_id_value;
+ time_t creation_time_value;
+ uint64_t id_sessiond_value;
/* From 2.11 to ... */
- ret = cmd_create_session_2_11(payload, session_name,
- hostname, &live_timer, &snapshot,
- &id_sessiond.value, sessiond_uuid,
- &has_current_chunk,
- ¤t_chunk_id.value);
+ ret = cmd_create_session_2_11(payload, session_name, hostname,
+ &live_timer, &snapshot, &id_sessiond_value,
+ sessiond_uuid, &has_current_chunk,
+ ¤t_chunk_id_value, &creation_time_value);
if (lttng_uuid_is_nil(sessiond_uuid)) {
/* The nil UUID is reserved for pre-2.11 clients. */
ERR("Illegal nil UUID announced by peer in create session command");
ret = -1;
goto send_reply;
}
- id_sessiond.is_set = true;
- current_chunk_id.is_set = has_current_chunk;
+ LTTNG_OPTIONAL_SET(&id_sessiond, id_sessiond_value);
+ LTTNG_OPTIONAL_SET(&creation_time, creation_time_value);
+ if (has_current_chunk) {
+ LTTNG_OPTIONAL_SET(¤t_chunk_id,
+ current_chunk_id_value);
+ }
}
if (ret < 0) {
snapshot, sessiond_uuid,
id_sessiond.is_set ? &id_sessiond.value : NULL,
current_chunk_id.is_set ? ¤t_chunk_id.value : NULL,
+ creation_time.is_set ? &creation_time.value : NULL,
conn->major, conn->minor);
if (!session) {
ret = -1;
pthread_mutex_unlock(&session->lock);
}
+static int conform_channel_path(char *channel_path)
+{
+ int ret = 0;
+
+ if (strstr("../", channel_path)) {
+ ERR("Refusing channel path as it walks up the path hierarchy: \"%s\"",
+ channel_path);
+ ret = -1;
+ goto end;
+ }
+
+ if (*channel_path == '/') {
+ const size_t len = strlen(channel_path);
+
+ /*
+ * Channel paths from peers prior to 2.11 are expressed as an
+ * absolute path that is, in reality, relative to the relay
+ * daemon's output directory. Remove the leading slash so it
+ * is correctly interpreted as a relative path later on.
+ *
+ * len (and not len - 1) is used to copy the trailing NULL.
+ */
+ bcopy(channel_path + 1, channel_path, len);
+ }
+end:
+ return ret;
+}
+
/*
* relay_add_stream: allocate a new stream for a session
*/
uint64_t stream_handle = -1ULL;
char *path_name = NULL, *channel_name = NULL;
uint64_t tracefile_size = 0, tracefile_count = 0;
- struct relay_stream_chunk_id stream_chunk_id = { 0 };
+ LTTNG_OPTIONAL(uint64_t) stream_chunk_id = {};
if (!session || !conn->version_check_done) {
ERR("Trying to add a stream before version check");
goto send_reply;
}
+ if (conform_channel_path(path_name)) {
+ goto send_reply;
+ }
+
trace = ctf_trace_get_by_path_or_create(session, path_name);
if (!trace) {
goto send_reply;
/* We pass ownership of path_name and channel_name. */
stream = stream_create(trace, stream_handle, path_name,
- channel_name, tracefile_size, tracefile_count,
- &stream_chunk_id);
+ channel_name, tracefile_size, tracefile_count);
path_name = NULL;
channel_name = NULL;
*/
static
int create_rotate_index_file(struct relay_stream *stream,
- const char *stream_path)
+ const char *channel_path)
{
int ret;
uint32_t major, minor;
+ ASSERT_LOCKED(stream->lock);
+
/* Put ref on previous index_file. */
if (stream->index_file) {
lttng_index_file_put(stream->index_file);
}
major = stream->trace->session->major;
minor = stream->trace->session->minor;
- stream->index_file = lttng_index_file_create(stream_path,
- stream->channel_name,
- -1, -1, stream->tracefile_size,
- tracefile_array_get_file_index_head(stream->tfa),
+ if (!stream->trace->index_folder_created) {
+ char *index_subpath = NULL;
+
+ ret = asprintf(&index_subpath, "%s/%s", channel_path, DEFAULT_INDEX_DIR);
+ if (ret < 0) {
+ goto end;
+ }
+
+ ret = lttng_trace_chunk_create_subdirectory(stream->trace_chunk, index_subpath);
+ free(index_subpath);
+ if (ret) {
+ goto end;
+ }
+ stream->trace->index_folder_created = true;
+ }
+ stream->index_file = lttng_index_file_create_from_trace_chunk(
+ stream->trace_chunk, channel_path, stream->channel_name,
+ stream->tracefile_size, stream->tracefile_count,
lttng_to_index_major(major, minor),
- lttng_to_index_minor(major, minor));
+ lttng_to_index_minor(major, minor), true);
if (!stream->index_file) {
ret = -1;
goto end;
goto end_stream_unlock;
}
- assert(stream->current_chunk_id.is_set);
- stream->current_chunk_id.value = stream_info.new_chunk_id;
-
if (stream->is_metadata) {
/*
* Metadata streams have no index; consider its rotation
* e.g. /home/user/lttng-traces/hostname/session_name
*/
char *full_session_path = NULL;
+ char creation_time_str[16];
+ struct tm *timeinfo;
+
+ assert(session->creation_time.is_set);
+ timeinfo = localtime(&session->creation_time.value);
+ if (!timeinfo) {
+ ret = -1;
+ goto end;
+ }
+ strftime(creation_time_str, sizeof(creation_time_str), "%Y%m%d-%H%M%S",
+ timeinfo);
pthread_mutex_lock(&session->lock);
- ret = asprintf(&session_directory, "%s/%s", session->hostname,
- session->session_name);
+ ret = asprintf(&session_directory, "%s/%s-%s", session->hostname,
+ session->session_name, creation_time_str);
pthread_mutex_unlock(&session->lock);
if (ret < 0) {
PERROR("Failed to format session directory name");
return ret;
}
+/*
+ * relay_close_trace_chunk: close a trace chunk
+ */
+static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn,
+ const struct lttng_buffer_view *payload)
+{
+ int ret = 0;
+ ssize_t send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_close_trace_chunk *msg;
+ struct lttcomm_relayd_generic_reply reply = {};
+ struct lttng_buffer_view header_view;
+ struct lttng_trace_chunk *chunk = NULL;
+ enum lttng_error_code reply_code = LTTNG_OK;
+ enum lttng_trace_chunk_status chunk_status;
+ uint64_t chunk_id;
+ LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command;
+ time_t close_timestamp;
+
+ if (!session || !conn->version_check_done) {
+ ERR("Trying to close a trace chunk before version check");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ if (session->major == 2 && session->minor < 11) {
+ ERR("Chunk close command is unsupported before 2.11");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
+ if (!header_view.data) {
+ ERR("Failed to receive payload of chunk close command");
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ /* Convert to host endianness. */
+ msg = (typeof(msg)) header_view.data;
+ chunk_id = be64toh(msg->chunk_id);
+ close_timestamp = (time_t) be64toh(msg->close_timestamp);
+ close_command = (typeof(close_command)){
+ .value = be32toh(msg->close_command.value),
+ .is_set = msg->close_command.is_set,
+ };
+
+ chunk = sessiond_trace_chunk_registry_get_chunk(
+ sessiond_trace_chunk_registry,
+ conn->session->sessiond_uuid,
+ conn->session->id,
+ chunk_id);
+ if (!chunk) {
+ char uuid_str[UUID_STR_LEN];
+
+ lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
+ ERR("Failed to find chunk to close: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
+ uuid_str,
+ conn->session->id,
+ msg->chunk_id);
+ ret = -1;
+ reply_code = LTTNG_ERR_NOMEM;
+ goto end;
+ }
+
+ chunk_status = lttng_trace_chunk_set_close_timestamp(
+ chunk, close_timestamp);
+ if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ ERR("Failed to set trace chunk close timestamp");
+ ret = -1;
+ reply_code = LTTNG_ERR_UNK;
+ goto end;
+ }
+
+ if (close_command.is_set) {
+ chunk_status = lttng_trace_chunk_set_close_command(
+ chunk, close_command.value);
+ if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ ret = -1;
+ reply_code = LTTNG_ERR_INVALID;
+ goto end;
+ }
+ }
+
+end:
+ reply.ret_code = htobe32((uint32_t) reply_code);
+ send_ret = conn->sock->ops->sendmsg(conn->sock,
+ &reply,
+ sizeof(struct lttcomm_relayd_generic_reply),
+ 0);
+ if (send_ret < (ssize_t) sizeof(reply)) {
+ ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
+ send_ret);
+ ret = -1;
+ }
+end_no_reply:
+ lttng_trace_chunk_put(chunk);
+ return ret;
+}
+
#define DBG_CMD(cmd_name, conn) \
DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd);
DBG_CMD("RELAYD_CREATE_TRACE_CHUNK", conn);
ret = relay_create_trace_chunk(header, conn, payload);
break;
+ case RELAYD_CLOSE_TRACE_CHUNK:
+ DBG_CMD("RELAYD_CLOSE_TRACE_CHUNK", conn);
+ ret = relay_close_trace_chunk(header, conn, payload);
+ break;
case RELAYD_UPDATE_SYNC_INFO:
default:
ERR("Received unknown command (%u)", header->cmd);
projects / lttng-tools.git / blobdiff