X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=d164bbd8cde4c5c176859bb6728433eef455d479;hp=82e2603544c9b770ec0f60a1344f456835ddb76e;hb=6fa5fe7cc78bea0b0bba154a0f911d3df530e18f;hpb=c35f9726a22f1d93e14589688d830efccda196f3 diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 82e260354..d164bbd8c 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -37,6 +37,7 @@ #include #include #include +#include #include #include @@ -58,7 +59,7 @@ #include #include #include -#include +#include #include "cmd.h" #include "ctf-trace.h" @@ -1073,6 +1074,7 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, uint32_t live_timer = 0; bool snapshot = false; /* Left nil for peers < 2.11. */ + char base_path[LTTNG_PATH_MAX] = {}; lttng_uuid sessiond_uuid = {}; LTTNG_OPTIONAL(uint64_t) id_sessiond = {}; LTTNG_OPTIONAL(uint64_t) current_chunk_id = {}; @@ -1093,7 +1095,7 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, /* From 2.11 to ... */ ret = cmd_create_session_2_11(payload, session_name, hostname, - &live_timer, &snapshot, &id_sessiond_value, + base_path, &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)) { @@ -1114,7 +1116,7 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr, goto send_reply; } - session = session_create(session_name, hostname, live_timer, + session = session_create(session_name, hostname, base_path, live_timer, snapshot, sessiond_uuid, id_sessiond.is_set ? &id_sessiond.value : NULL, current_chunk_id.is_set ? ¤t_chunk_id.value : NULL, @@ -2116,7 +2118,7 @@ static int relay_rotate_session_streams( struct relay_connection *conn, const struct lttng_buffer_view *payload) { - int ret; + int ret = 0; uint32_t i; ssize_t send_ret; enum lttng_error_code reply_code = LTTNG_ERR_UNK; @@ -2127,6 +2129,8 @@ static int relay_rotate_session_streams( const size_t header_len = sizeof(struct lttcomm_relayd_rotate_streams); struct lttng_trace_chunk *next_trace_chunk = NULL; struct lttng_buffer_view stream_positions; + char chunk_id_buf[MAX_INT_DEC_LEN(uint64_t)]; + const char *chunk_id_str = "none"; if (!session || !conn->version_check_done) { ERR("Trying to rotate a stream before version check"); @@ -2180,8 +2184,20 @@ static int relay_rotate_session_streams( ret = -1; goto end; } + + ret = snprintf(chunk_id_buf, sizeof(chunk_id_buf), "%" PRIu64, + rotate_streams.new_chunk_id.value); + if (ret < 0 || ret >= sizeof(chunk_id_buf)) { + chunk_id_str = "formatting error"; + } else { + chunk_id_str = chunk_id_buf; + } } + DBG("Rotate %" PRIu32 " streams of session \"%s\" to chunk \"%s\"", + rotate_streams.stream_count, session->session_name, + chunk_id_str); + stream_positions = lttng_buffer_view_from_view(payload, sizeof(rotate_streams), -1); if (!stream_positions.data || @@ -2237,6 +2253,7 @@ end: ret = -1; } + ret = 0; end_no_reply: lttng_trace_chunk_put(next_trace_chunk); return ret; @@ -2246,29 +2263,49 @@ static int init_session_output_directory_handle(struct relay_session *session, struct lttng_directory_handle *handle) { int ret; - /* hostname/session_name */ + /* + * session_directory: + * + * if base_path is NULL + * hostname/session_name + * else + * hostname/base_path + */ char *session_directory = NULL; /* * base path + session_directory * 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); + /* + * If base path is set, it overrides the session name for the + * session relative base path. No timestamp is appended if the + * base path is overridden. + */ + if (session->base_path[0] == '\0') { + char creation_time_str[16]; + struct tm *timeinfo; - pthread_mutex_lock(&session->lock); - ret = asprintf(&session_directory, "%s/%s-%s", session->hostname, - session->session_name, creation_time_str); - pthread_mutex_unlock(&session->lock); + 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-%s", session->hostname, + session->session_name, creation_time_str); + pthread_mutex_unlock(&session->lock); + } else { + pthread_mutex_lock(&session->lock); + ret = asprintf(&session_directory, "%s/%s", session->hostname, + session->base_path); + pthread_mutex_unlock(&session->lock); + } if (ret < 0) { PERROR("Failed to format session directory name"); goto end; @@ -2423,11 +2460,22 @@ static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, } pthread_mutex_lock(&conn->session->lock); - lttng_trace_chunk_put(conn->session->current_trace_chunk); + if (conn->session->pending_closure_trace_chunk) { + /* + * Invalid; this means a second create_trace_chunk command was + * received before a close_trace_chunk. + */ + ERR("Invalid trace chunk close command received; a trace chunk is already waiting for a trace chunk close command"); + reply_code = LTTNG_ERR_INVALID_PROTOCOL; + ret = -1; + goto end_unlock_session; + } + conn->session->pending_closure_trace_chunk = + conn->session->current_trace_chunk; conn->session->current_trace_chunk = published_chunk; published_chunk = NULL; +end_unlock_session: pthread_mutex_unlock(&conn->session->lock); - end: reply.ret_code = htobe32((uint32_t) reply_code); send_ret = conn->sock->ops->sendmsg(conn->sock, @@ -2512,13 +2560,23 @@ static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, goto end; } + pthread_mutex_lock(&session->lock); + if (session->pending_closure_trace_chunk && + session->pending_closure_trace_chunk != chunk) { + ERR("Trace chunk close command for session \"%s\" does not target the trace chunk pending closure", + session->session_name); + reply_code = LTTNG_ERR_INVALID_PROTOCOL; + ret = -1; + goto end_unlock_session; + } + 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; + goto end_unlock_session; } if (close_command.is_set) { @@ -2527,11 +2585,10 @@ static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) { ret = -1; reply_code = LTTNG_ERR_INVALID; - goto end; + goto end_unlock_session; } } - pthread_mutex_lock(&session->lock); if (session->current_trace_chunk == chunk) { /* * After a trace chunk close command, no new streams @@ -2544,6 +2601,9 @@ static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr, lttng_trace_chunk_put(session->current_trace_chunk); session->current_trace_chunk = NULL; } + lttng_trace_chunk_put(session->pending_closure_trace_chunk); + session->pending_closure_trace_chunk = NULL; +end_unlock_session: pthread_mutex_unlock(&session->lock); end: