X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=dddc2a2b4f77b25c9ba968826460147302f25e0a;hp=11277f29346e99104cfbedd6453ace99a76493f6;hb=348a81dcf7b6944b10a813d93dcaf86fdb5194f6;hpb=e5add6d004793894ef4c7e047bc0f8885763b205

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index 11277f293..dddc2a2b4 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -1104,6 +1104,7 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr,
 	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 */
@@ -1114,21 +1115,27 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr,
 			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,
-				&current_chunk_id.value);
+		ret = cmd_create_session_2_11(payload, session_name, hostname,
+				&live_timer, &snapshot, &id_sessiond_value,
+				sessiond_uuid, &has_current_chunk,
+				&current_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(&current_chunk_id,
+					current_chunk_id_value);
+		}
 	}
 
 	if (ret < 0) {
@@ -1139,6 +1146,7 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr,
 			snapshot, sessiond_uuid,
 			id_sessiond.is_set ? &id_sessiond.value : NULL,
 			current_chunk_id.is_set ? &current_chunk_id.value : NULL,
+			creation_time.is_set ? &creation_time.value : NULL,
 			conn->major, conn->minor);
 	if (!session) {
 		ret = -1;
@@ -1199,6 +1207,34 @@ static void publish_connection_local_streams(struct relay_connection *conn)
 	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
  */
@@ -1215,7 +1251,7 @@ static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr,
 	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");
@@ -1243,6 +1279,10 @@ static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr,
 		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;
@@ -1255,8 +1295,7 @@ static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr,
 
 	/* 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;
 
@@ -1552,11 +1591,13 @@ end:
  */
 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);
@@ -1564,12 +1605,26 @@ int create_rotate_index_file(struct relay_stream *stream,
 	}
 	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;
@@ -2610,9 +2665,6 @@ static int relay_rotate_session_stream(const struct lttcomm_relayd_hdr *recv_hdr
 		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
@@ -2671,10 +2723,21 @@ static int init_session_output_directory_handle(struct relay_session *session,
 	 * 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");
@@ -2853,6 +2916,107 @@ end_no_reply:
 	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);
 
@@ -2923,6 +3087,10 @@ static int relay_process_control_command(struct relay_connection *conn,
 		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);