X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Frelayd%2Frelayd.c;h=995e4880197550575667d6e18ca613645b1916b5;hb=75cfe9e675b3180eaecefac4eaed5b756f53aae8;hp=c3ad597ed8eb50d69dd9a57e65399a996ff423b3;hpb=76b9afaa436496319456775d7ba7f20eeac09d5d;p=lttng-tools.git diff --git a/src/common/relayd/relayd.c b/src/common/relayd/relayd.c index c3ad597ed..995e48801 100644 --- a/src/common/relayd/relayd.c +++ b/src/common/relayd/relayd.c @@ -72,14 +72,14 @@ static int send_command(struct lttcomm_relayd_sock *rsock, memcpy(buf + sizeof(header), data, size); } + DBG3("Relayd sending command %d of size %" PRIu64, (int) cmd, buf_size); ret = rsock->sock.ops->sendmsg(&rsock->sock, buf, buf_size, flags); if (ret < 0) { + PERROR("Failed to send command %d of size %" PRIu64, + (int) cmd, buf_size); ret = -errno; goto error; } - - DBG3("Relayd sending command %d of size %" PRIu64, cmd, buf_size); - error: free(buf); alloc_error: @@ -942,6 +942,200 @@ error: return ret; } +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) +{ + int ret; + struct lttcomm_relayd_rotate_stream *msg = NULL; + struct lttcomm_relayd_generic_reply reply; + size_t len; + int msg_len; + + /* Code flow error. Safety net. */ + assert(rsock); + + DBG("Sending rotate stream id %" PRIu64 " command to relayd", stream_id); + + /* Account for the trailing NULL. */ + len = 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; + goto error; + } + + msg_len = offsetof(struct lttcomm_relayd_rotate_stream, new_pathname) + len; + msg = zmalloc(msg_len); + if (!msg) { + PERROR("Failed to allocate relayd rotate stream command of %d bytes", + msg_len); + ret = -1; + goto error; + } + + if (lttng_strncpy(msg->new_pathname, new_pathname, len)) { + ret = -1; + ERR("Failed to copy relayd rotate stream command's new path name"); + goto error; + } + + msg->pathname_length = htobe32(len); + msg->stream_id = htobe64(stream_id); + msg->new_chunk_id = htobe64(new_chunk_id); + /* + * The seq_num is invalid for metadata streams, but it is ignored on + * the relay. + */ + msg->rotate_at_seq_num = htobe64(seq_num); + + /* Send command. */ + ret = send_command(rsock, RELAYD_ROTATE_STREAM, (void *) msg, msg_len, 0); + if (ret < 0) { + ERR("Send rotate command"); + goto error; + } + + /* Receive response. */ + ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); + if (ret < 0) { + ERR("Receive rotate reply"); + goto error; + } + + reply.ret_code = be32toh(reply.ret_code); + + /* Return session id or negative ret code. */ + if (reply.ret_code != LTTNG_OK) { + ret = -1; + ERR("Relayd rotate stream replied error %d", reply.ret_code); + } else { + /* Success. */ + ret = 0; + DBG("Relayd rotated stream id %" PRIu64 " successfully", stream_id); + } + +error: + free(msg); + return ret; +} + +int relayd_rotate_rename(struct lttcomm_relayd_sock *rsock, + const char *old_path, const char *new_path) +{ + int ret; + struct lttcomm_relayd_rotate_rename *msg = NULL; + struct lttcomm_relayd_generic_reply reply; + size_t old_path_length, new_path_length; + size_t msg_length; + + /* Code flow error. Safety net. */ + assert(rsock); + + DBG("Relayd rename chunk %s to %s", old_path, new_path); + + /* The two paths are sent with a '\0' delimiter between them. */ + old_path_length = strlen(old_path) + 1; + new_path_length = strlen(new_path) + 1; + + msg_length = sizeof(*msg) + old_path_length + new_path_length; + msg = zmalloc(msg_length); + if (!msg) { + PERROR("zmalloc rotate-rename command message"); + ret = -1; + goto error; + } + + assert(old_path_length <= UINT32_MAX); + msg->old_path_length = htobe32(old_path_length); + + assert(new_path_length <= UINT32_MAX); + msg->new_path_length = htobe32(new_path_length); + + strcpy(msg->paths, old_path); + strcpy(msg->paths + old_path_length, new_path); + + /* Send command */ + ret = send_command(rsock, RELAYD_ROTATE_RENAME, (const void *) msg, + msg_length, 0); + if (ret < 0) { + goto error; + } + + /* Receive response */ + ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); + if (ret < 0) { + goto error; + } + + reply.ret_code = be32toh(reply.ret_code); + + /* Return session id or negative ret code. */ + if (reply.ret_code != LTTNG_OK) { + ret = -1; + ERR("Relayd rotate rename replied error %d", reply.ret_code); + } else { + /* Success */ + ret = 0; + } + + DBG("Relayd rotate rename completed successfully"); + +error: + free(msg); + return ret; +} + +int relayd_rotate_pending(struct lttcomm_relayd_sock *rsock, uint64_t chunk_id) +{ + int ret; + struct lttcomm_relayd_rotate_pending msg; + struct lttcomm_relayd_rotate_pending_reply reply; + + /* Code flow error. Safety net. */ + assert(rsock); + + DBG("Querying relayd for rotate pending with chunk_id %" PRIu64, + chunk_id); + + memset(&msg, 0, sizeof(msg)); + msg.chunk_id = htobe64(chunk_id); + + /* Send command */ + ret = send_command(rsock, RELAYD_ROTATE_PENDING, (void *) &msg, + sizeof(msg), 0); + if (ret < 0) { + goto error; + } + + /* Receive response */ + ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); + if (ret < 0) { + goto error; + } + + reply.generic.ret_code = be32toh(reply.generic.ret_code); + + /* Return session id or negative ret code. */ + if (reply.generic.ret_code != LTTNG_OK) { + ret = -reply.generic.ret_code; + ERR("Relayd rotate pending replied with error %d", ret); + goto error; + } else { + /* No error, just rotate pending state */ + if (reply.is_pending == 0 || reply.is_pending == 1) { + ret = reply.is_pending; + DBG("Relayd rotate pending command completed successfully with result \"%s\"", + ret ? "rotation pending" : "rotation NOT pending"); + } else { + ret = -LTTNG_ERR_UNK; + } + } + +error: + return ret; +} + int relayd_mkdir(struct lttcomm_relayd_sock *rsock, const char *path) { int ret; @@ -997,5 +1191,4 @@ int relayd_mkdir(struct lttcomm_relayd_sock *rsock, const char *path) error: free(msg); return ret; - }