X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Frelayd%2Frelayd.c;h=fb459699d32a68ddbf93d001df1829d93798181f;hb=d295668767ac8234e83984e1812d342d03293d88;hp=45f965c7941e9c3be6d13880a3933f9e40a86106;hpb=f86f6389a47137d2f5e8fc707916b6650fd1de00;p=lttng-tools.git diff --git a/src/common/relayd/relayd.c b/src/common/relayd/relayd.c index 45f965c79..fb459699d 100644 --- a/src/common/relayd/relayd.c +++ b/src/common/relayd/relayd.c @@ -26,8 +26,10 @@ #include #include #include +#include #include #include +#include #include "relayd.h" @@ -125,8 +127,9 @@ error: * payload size is introduced. */ static int relayd_create_session_2_11(struct lttcomm_relayd_sock *rsock, - char *session_name, char *hostname, - int session_live_timer, unsigned int snapshot) + const char *session_name, const char *hostname, + int session_live_timer, unsigned int snapshot, + uint64_t sessiond_session_id, const lttng_uuid sessiond_uuid) { int ret; struct lttcomm_relayd_create_session_2_11 *msg = NULL; @@ -164,6 +167,9 @@ static int relayd_create_session_2_11(struct lttcomm_relayd_sock *rsock, msg->live_timer = htobe32(session_live_timer); msg->snapshot = !!snapshot; + lttng_uuid_copy(msg->sessiond_uuid, sessiond_uuid); + msg->session_id = htobe64(sessiond_session_id); + /* Send command */ ret = send_command(rsock, RELAYD_CREATE_SESSION, msg, msg_length, 0); if (ret < 0) { @@ -178,8 +184,8 @@ error: * support the live reading capability. */ static int relayd_create_session_2_4(struct lttcomm_relayd_sock *rsock, - char *session_name, char *hostname, int session_live_timer, - unsigned int snapshot) + const char *session_name, const char *hostname, + int session_live_timer, unsigned int snapshot) { int ret; struct lttcomm_relayd_create_session_2_4 msg; @@ -230,15 +236,18 @@ error: * On success, return 0 else a negative value which is either an errno error or * a lttng error code from the relayd. */ -int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_id, - char *session_name, char *hostname, int session_live_timer, - unsigned int snapshot) +int relayd_create_session(struct lttcomm_relayd_sock *rsock, + uint64_t *relayd_session_id, + const char *session_name, const char *hostname, + int session_live_timer, + unsigned int snapshot, uint64_t sessiond_session_id, + const lttng_uuid sessiond_uuid) { int ret; struct lttcomm_relayd_status_session reply; assert(rsock); - assert(session_id); + assert(relayd_session_id); DBG("Relayd create session"); @@ -252,7 +261,8 @@ int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_i } else { /* From 2.11 to ... */ ret = relayd_create_session_2_11(rsock, session_name, - hostname, session_live_timer, snapshot); + hostname, session_live_timer, snapshot, + sessiond_session_id, sessiond_uuid); } if (ret < 0) { @@ -275,7 +285,7 @@ int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_i goto error; } else { ret = 0; - *session_id = reply.session_id; + *relayd_session_id = reply.session_id; } DBG("Relayd session created with id %" PRIu64, reply.session_id); @@ -284,6 +294,121 @@ error: return ret; } +static int relayd_add_stream_2_1(struct lttcomm_relayd_sock *rsock, + const char *channel_name, const char *pathname) +{ + int ret; + struct lttcomm_relayd_add_stream msg; + + memset(&msg, 0, sizeof(msg)); + if (lttng_strncpy(msg.channel_name, channel_name, + sizeof(msg.channel_name))) { + ret = -1; + goto error; + } + + if (lttng_strncpy(msg.pathname, pathname, + sizeof(msg.pathname))) { + ret = -1; + goto error; + } + + /* Send command */ + ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0); + if (ret < 0) { + ret = -1; + goto error; + } + ret = 0; +error: + return ret; +} + +static int relayd_add_stream_2_2(struct lttcomm_relayd_sock *rsock, + const char *channel_name, const char *pathname, + uint64_t tracefile_size, uint64_t tracefile_count) +{ + int ret; + struct lttcomm_relayd_add_stream_2_2 msg; + + memset(&msg, 0, sizeof(msg)); + /* Compat with relayd 2.2 to 2.10 */ + if (lttng_strncpy(msg.channel_name, channel_name, + sizeof(msg.channel_name))) { + ret = -1; + goto error; + } + if (lttng_strncpy(msg.pathname, pathname, + sizeof(msg.pathname))) { + ret = -1; + goto error; + } + msg.tracefile_size = htobe64(tracefile_size); + msg.tracefile_count = htobe64(tracefile_count); + + /* Send command */ + ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0); + if (ret < 0) { + goto error; + } + ret = 0; +error: + return ret; +} + +static int relayd_add_stream_2_11(struct lttcomm_relayd_sock *rsock, + const char *channel_name, const char *pathname, + uint64_t tracefile_size, uint64_t tracefile_count, + uint64_t trace_archive_id) +{ + int ret; + struct lttcomm_relayd_add_stream_2_11 *msg = NULL; + size_t channel_name_len; + size_t pathname_len; + size_t msg_length; + + /* The two names are sent with a '\0' delimiter between them. */ + channel_name_len = strlen(channel_name) + 1; + pathname_len = strlen(pathname) + 1; + + msg_length = sizeof(*msg) + channel_name_len + pathname_len; + msg = zmalloc(msg_length); + if (!msg) { + PERROR("zmalloc add_stream_2_11 command message"); + ret = -1; + goto error; + } + + assert(channel_name_len <= UINT32_MAX); + msg->channel_name_len = htobe32(channel_name_len); + + assert(pathname_len <= UINT32_MAX); + msg->pathname_len = htobe32(pathname_len); + + if (lttng_strncpy(msg->names, channel_name, channel_name_len)) { + ret = -1; + goto error; + } + if (lttng_strncpy(msg->names + channel_name_len, pathname, pathname_len)) { + ret = -1; + goto error; + } + + msg->tracefile_size = htobe64(tracefile_size); + msg->tracefile_count = htobe64(tracefile_count); + msg->trace_archive_id = htobe64(trace_archive_id); + + /* Send command */ + ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) msg, msg_length, 0); + if (ret < 0) { + goto error; + } + ret = 0; +error: + free(msg); + return ret; +} + /* * Add stream on the relayd and assign stream handle to the stream_id argument. * @@ -291,11 +416,10 @@ error: */ int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_name, const char *pathname, uint64_t *stream_id, - uint64_t tracefile_size, uint64_t tracefile_count) + uint64_t tracefile_size, uint64_t tracefile_count, + struct lttng_trace_chunk *trace_chunk) { int ret; - struct lttcomm_relayd_add_stream msg; - struct lttcomm_relayd_add_stream_2_2 msg_2_2; struct lttcomm_relayd_status_stream reply; /* Code flow error. Safety net. */ @@ -307,44 +431,33 @@ int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_nam /* Compat with relayd 2.1 */ if (rsock->minor == 1) { - memset(&msg, 0, sizeof(msg)); - if (lttng_strncpy(msg.channel_name, channel_name, - sizeof(msg.channel_name))) { - ret = -1; - goto error; - } - if (lttng_strncpy(msg.pathname, pathname, - sizeof(msg.pathname))) { - ret = -1; - goto error; - } - - /* Send command */ - ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0); - if (ret < 0) { - goto error; - } + /* For 2.1 */ + assert(!trace_chunk); + ret = relayd_add_stream_2_1(rsock, channel_name, pathname); + + } else if (rsock->minor > 1 && rsock->minor < 11) { + /* From 2.2 to 2.10 */ + assert(!trace_chunk); + ret = relayd_add_stream_2_2(rsock, channel_name, pathname, + tracefile_size, tracefile_count); } else { - memset(&msg_2_2, 0, sizeof(msg_2_2)); - /* Compat with relayd 2.2+ */ - if (lttng_strncpy(msg_2_2.channel_name, channel_name, - sizeof(msg_2_2.channel_name))) { - ret = -1; - goto error; - } - if (lttng_strncpy(msg_2_2.pathname, pathname, - sizeof(msg_2_2.pathname))) { - ret = -1; - goto error; - } - msg_2_2.tracefile_size = htobe64(tracefile_size); - msg_2_2.tracefile_count = htobe64(tracefile_count); + enum lttng_trace_chunk_status chunk_status; + uint64_t chunk_id; - /* Send command */ - ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg_2_2, sizeof(msg_2_2), 0); - if (ret < 0) { - goto error; - } + assert(trace_chunk); + chunk_status = lttng_trace_chunk_get_id(trace_chunk, + &chunk_id); + assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK); + + /* From 2.11 to ...*/ + ret = relayd_add_stream_2_11(rsock, channel_name, pathname, + tracefile_size, tracefile_count, + chunk_id); + } + + if (ret) { + ret = -1; + goto error; } /* Waiting for reply */ @@ -368,7 +481,7 @@ int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_nam } DBG("Relayd stream added successfully with handle %" PRIu64, - reply.handle); + reply.handle); error: return ret; @@ -997,14 +1110,15 @@ error: } int relayd_rotate_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id, - const char *new_pathname, uint64_t new_chunk_id, - uint64_t seq_num) + uint64_t new_chunk_id, uint64_t seq_num) { int ret; struct lttcomm_relayd_rotate_stream *msg = NULL; struct lttcomm_relayd_generic_reply reply; size_t len; int msg_len; + /* FIXME */ + char *new_pathname = NULL; /* Code flow error. Safety net. */ assert(rsock); @@ -1012,7 +1126,7 @@ int relayd_rotate_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id, DBG("Sending rotate stream id %" PRIu64 " command to relayd", stream_id); /* Account for the trailing NULL. */ - len = strnlen(new_pathname, LTTNG_PATH_MAX) + 1; + len = lttng_strnlen(new_pathname, LTTNG_PATH_MAX) + 1; if (len > LTTNG_PATH_MAX) { ERR("Path used in relayd rotate stream command exceeds the maximal allowed length"); ret = -1;