[lttng-tools.git] / tests / regression / tools / live / live_test.c

/*
 * Copyright (c) - 2013 Julien Desfossez <jdesfossez@efficios.com>
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by as
 * published by the Free Software Foundation; only version 2 of the License.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#define _GNU_SOURCE
#include <assert.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include <sys/types.h>
#include <inttypes.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>

#include <tap/tap.h>
#include <lttng/lttng.h>

#include <urcu/list.h>
#include <bin/lttng-sessiond/session.h>
#include <common/common.h>

#include <bin/lttng-relayd/lttng-viewer-abi.h>
#include <common/index/ctf-index.h>

#include <common/compat/endian.h>

#define SESSION1 "test1"
#define RELAYD_URL "net://localhost"
#define LIVE_TIMER 2000000

/* Number of TAP tests in this file */
#define NUM_TESTS 8
#define mmap_size 524288

int ust_consumerd32_fd;
int ust_consumerd64_fd;

static int control_sock;
struct live_session *session;

static int first_packet_offset;
static int first_packet_len;
static int first_packet_stream_id;

struct viewer_stream {
	uint64_t id;
	uint64_t ctf_trace_id;
	void *mmap_base;
	int fd;
	int metadata_flag;
	int first_read;
	char path[PATH_MAX];
};

struct live_session {
	struct viewer_stream *streams;
	uint64_t live_timer_interval;
	uint64_t stream_count;
};

static
ssize_t lttng_live_recv(int fd, void *buf, size_t len)
{
	ssize_t ret;
	size_t copied = 0, to_copy = len;

	do {
		ret = recv(fd, buf + copied, to_copy, 0);
		if (ret > 0) {
			assert(ret <= to_copy);
			copied += ret;
			to_copy -= ret;
		}
	} while ((ret > 0 && to_copy > 0)
		|| (ret < 0 && errno == EINTR));
	if (ret > 0)
		ret = copied;
	/* ret = 0 means orderly shutdown, ret < 0 is error. */
	return ret;
}

static
ssize_t lttng_live_send(int fd, const void *buf, size_t len)
{
	ssize_t ret;

	do {
		ret = send(fd, buf, len, MSG_NOSIGNAL);
	} while (ret < 0 && errno == EINTR);
	return ret;
}

static
int connect_viewer(char *hostname)
{
	struct hostent *host;
	struct sockaddr_in server_addr;
	int ret;

	host = gethostbyname(hostname);
	if (!host) {
		ret = -1;
		goto end;
	}

	if ((control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
		PERROR("Socket");
		ret = -1;
		goto end;
	}

	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(5344);
	server_addr.sin_addr = *((struct in_addr *) host->h_addr);
	bzero(&(server_addr.sin_zero), 8);

	if (connect(control_sock, (struct sockaddr *) &server_addr,
				sizeof(struct sockaddr)) == -1) {
		PERROR("Connect");
		ret = -1;
		goto end;
	}

	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(5345);
	server_addr.sin_addr = *((struct in_addr *) host->h_addr);
	bzero(&(server_addr.sin_zero), 8);

	ret = 0;

end:
	return ret;
}

int establish_connection(void)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_connect connect;
	ssize_t ret_len;

	cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
	cmd.data_size = sizeof(connect);
	cmd.cmd_version = 0;

	memset(&connect, 0, sizeof(connect));
	connect.major = htobe32(VERSION_MAJOR);
	connect.minor = htobe32(VERSION_MINOR);
	connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);

	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
	if (ret_len < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}
	ret_len = lttng_live_send(control_sock, &connect, sizeof(connect));
	if (ret_len < 0) {
		fprintf(stderr, "Error sending version\n");
		goto error;
	}

	ret_len = lttng_live_recv(control_sock, &connect, sizeof(connect));
	if (ret_len == 0) {
		fprintf(stderr, "[error] Remote side has closed connection\n");
		goto error;
	}
	if (ret_len < 0) {
		fprintf(stderr, "Error receiving version\n");
		goto error;
	}
	return 0;

error:
	return -1;
}

/*
 * Returns the number of sessions, should be 1 during the unit test.
 */
int list_sessions(int *session_id)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_list_sessions list;
	struct lttng_viewer_session lsession;
	int i;
	ssize_t ret_len;
	int first_session = 0;

	cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
	cmd.data_size = 0;
	cmd.cmd_version = 0;

	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
	if (ret_len < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}

	ret_len = lttng_live_recv(control_sock, &list, sizeof(list));
	if (ret_len == 0) {
		fprintf(stderr, "[error] Remote side has closed connection\n");
		goto error;
	}
	if (ret_len < 0) {
		fprintf(stderr, "Error receiving session list\n");
		goto error;
	}

	for (i = 0; i < be32toh(list.sessions_count); i++) {
		ret_len = lttng_live_recv(control_sock, &lsession, sizeof(lsession));
		if (ret_len < 0) {
			fprintf(stderr, "Error receiving session\n");
			goto error;
		}
		if (lsession.streams > 0 && first_session <= 0) {
			first_session = be64toh(lsession.id);
			*session_id = first_session;
		}
	}

	return be32toh(list.sessions_count);

error:
	return -1;
}

int create_viewer_session(void)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_create_session_response resp;
	ssize_t ret_len;

	cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
	cmd.data_size = 0;
	cmd.cmd_version = 0;

	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
	if (ret_len < 0) {
		fprintf(stderr, "[error] Error sending cmd\n");
		goto error;
	}
	assert(ret_len == sizeof(cmd));

	ret_len = lttng_live_recv(control_sock, &resp, sizeof(resp));
	if (ret_len == 0) {
		fprintf(stderr, "[error] Remote side has closed connection\n");
		goto error;
	}
	if (ret_len < 0) {
		fprintf(stderr, "[error] Error receiving create session reply\n");
		goto error;
	}
	assert(ret_len == sizeof(resp));

	if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
		fprintf(stderr, "[error] Error creating viewer session\n");
		goto error;
	}
	return 0;

error:
	return -1;
}

int attach_session(int id)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_attach_session_request rq;
	struct lttng_viewer_attach_session_response rp;
	struct lttng_viewer_stream stream;
	int i;
	ssize_t ret_len;

	session = zmalloc(sizeof(struct live_session));
	if (!session) {
		goto error;
	}

	cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
	cmd.data_size = sizeof(rq);
	cmd.cmd_version = 0;

	memset(&rq, 0, sizeof(rq));
	rq.session_id = htobe64(id);
	rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);

	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
	if (ret_len < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}
	ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
	if (ret_len < 0) {
		fprintf(stderr, "Error sending attach request\n");
		goto error;
	}

	ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
	if (ret_len == 0) {
		fprintf(stderr, "[error] Remote side has closed connection\n");
		goto error;
	}
	if (ret_len < 0) {
		fprintf(stderr, "Error receiving attach response\n");
		goto error;
	}
	if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
		goto error;
	}

	session->stream_count = be32toh(rp.streams_count);
	session->streams = zmalloc(session->stream_count *
			sizeof(struct viewer_stream));
	if (!session->streams) {
		goto error;
	}

	for (i = 0; i < be32toh(rp.streams_count); i++) {
		ret_len = lttng_live_recv(control_sock, &stream, sizeof(stream));
		if (ret_len == 0) {
			fprintf(stderr, "[error] Remote side has closed connection\n");
			goto error;
		}
		if (ret_len < 0) {
			fprintf(stderr, "Error receiving stream\n");
			goto error;
		}
		session->streams[i].id = be64toh(stream.id);

		session->streams[i].ctf_trace_id = be64toh(stream.ctf_trace_id);
		session->streams[i].first_read = 1;
		session->streams[i].mmap_base = mmap(NULL, mmap_size, PROT_READ | PROT_WRITE,
				MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
		if (session->streams[i].mmap_base == MAP_FAILED) {
			fprintf(stderr, "mmap error\n");
			goto error;
		}

		if (be32toh(stream.metadata_flag)) {
			session->streams[i].metadata_flag = 1;
		}
	}
	return session->stream_count;

error:
	return -1;
}

int get_metadata(void)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_get_metadata rq;
	struct lttng_viewer_metadata_packet rp;
	ssize_t ret_len;
	int ret;
	uint64_t i;
	char *data = NULL;
	uint64_t len = 0;
	int metadata_stream_id = -1;

	cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
	cmd.data_size = sizeof(rq);
	cmd.cmd_version = 0;

	for (i = 0; i < session->stream_count; i++) {
		if (session->streams[i].metadata_flag) {
			metadata_stream_id = i;
			break;
		}
	}

	if (metadata_stream_id < 0) {
		fprintf(stderr, "No metadata stream found\n");
		goto error;
	}

	rq.stream_id = htobe64(session->streams[metadata_stream_id].id);

	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
	if (ret_len < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}
	ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
	if (ret_len < 0) {
		fprintf(stderr, "Error sending get_metadata request\n");
		goto error;
	}
	ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
	if (ret_len == 0) {
		fprintf(stderr, "[error] Remote side has closed connection\n");
		goto error;
	}
	if (ret_len < 0) {
		fprintf(stderr, "Error receiving metadata response\n");
		goto error;
	}
	switch (be32toh(rp.status)) {
		case LTTNG_VIEWER_METADATA_OK:
			break;
		case LTTNG_VIEWER_NO_NEW_METADATA:
			fprintf(stderr, "NO NEW\n");
			ret = 0;
			goto end;
		case LTTNG_VIEWER_METADATA_ERR:
			fprintf(stderr, "ERR\n");
			goto error;
		default:
			fprintf(stderr, "UNKNOWN\n");
			goto error;
	}

	len = be64toh(rp.len);
	if (len <= 0) {
		goto error;
	}

	data = zmalloc(len);
	if (!data) {
		PERROR("relay data zmalloc");
		goto error;
	}
	ret_len = lttng_live_recv(control_sock, data, len);
	if (ret_len == 0) {
		fprintf(stderr, "[error] Remote side has closed connection\n");
		goto error_free_data;
	}
	if (ret_len < 0) {
		fprintf(stderr, "Error receiving trace packet\n");
		goto error_free_data;
	}
	free(data);
	ret = len;
end:
	return ret;

error_free_data:
	free(data);
error:
	return -1;
}

int get_next_index(void)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_get_next_index rq;
	struct lttng_viewer_index rp;
	ssize_t ret_len;
	int id;

	cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
	cmd.data_size = sizeof(rq);
	cmd.cmd_version = 0;

	for (id = 0; id < session->stream_count; id++) {
		if (session->streams[id].metadata_flag) {
			continue;
		}
		memset(&rq, 0, sizeof(rq));
		rq.stream_id = htobe64(session->streams[id].id);

retry:
		ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
		if (ret_len < 0) {
			fprintf(stderr, "Error sending cmd\n");
			goto error;
		}
		ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
		if (ret_len < 0) {
			fprintf(stderr, "Error sending get_next_index request\n");
			goto error;
		}
		ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
		if (ret_len == 0) {
			fprintf(stderr, "[error] Remote side has closed connection\n");
			goto error;
		}
		if (ret_len < 0) {
			fprintf(stderr, "Error receiving index response\n");
			goto error;
		}

		rp.flags = be32toh(rp.flags);

		switch (be32toh(rp.status)) {
			case LTTNG_VIEWER_INDEX_INACTIVE:
				fprintf(stderr, "(INACTIVE)\n");
				break;
			case LTTNG_VIEWER_INDEX_OK:
				break;
			case LTTNG_VIEWER_INDEX_RETRY:
				sleep(1);
				goto retry;
			case LTTNG_VIEWER_INDEX_HUP:
				fprintf(stderr, "(HUP)\n");
				session->streams[id].id = -1ULL;
				session->streams[id].fd = -1;
				break;
			case LTTNG_VIEWER_INDEX_ERR:
				fprintf(stderr, "(ERR)\n");
				goto error;
			default:
				fprintf(stderr, "SHOULD NOT HAPPEN\n");
				goto error;
		}
		if (!first_packet_stream_id) {
			first_packet_offset = be64toh(rp.offset);
			first_packet_len = be64toh(rp.packet_size) / CHAR_BIT;
			first_packet_stream_id = id;
		}
	}
	return 0;

error:
	return -1;
}

static
int get_data_packet(int id, uint64_t offset,
		uint64_t len)
{
	struct lttng_viewer_cmd cmd;
	struct lttng_viewer_get_packet rq;
	struct lttng_viewer_trace_packet rp;
	ssize_t ret_len;

	cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
	cmd.data_size = sizeof(rq);
	cmd.cmd_version = 0;

	memset(&rq, 0, sizeof(rq));
	rq.stream_id = htobe64(session->streams[id].id);
	/* Already in big endian. */
	rq.offset = offset;
	rq.len = htobe32(len);

	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
	if (ret_len < 0) {
		fprintf(stderr, "Error sending cmd\n");
		goto error;
	}
	ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
	if (ret_len < 0) {
		fprintf(stderr, "Error sending get_data_packet request\n");
		goto error;
	}
	ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
	if (ret_len == 0) {
		fprintf(stderr, "[error] Remote side has closed connection\n");
		goto error;
	}
	if (ret_len < 0) {
		fprintf(stderr, "Error receiving data response\n");
		goto error;
	}
	rp.flags = be32toh(rp.flags);

	switch (be32toh(rp.status)) {
	case LTTNG_VIEWER_GET_PACKET_OK:
		len = be32toh(rp.len);
		break;
	case LTTNG_VIEWER_GET_PACKET_RETRY:
		fprintf(stderr, "RETRY\n");
		goto error;
	case LTTNG_VIEWER_GET_PACKET_ERR:
		if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
			fprintf(stderr, "NEW_METADATA\n");
			goto end;
		}
		fprintf(stderr, "ERR\n");
		goto error;
	default:
		fprintf(stderr, "UNKNOWN\n");
		goto error;
	}

	if (len == 0) {
		goto error;
	}

	if (len > mmap_size) {
		fprintf(stderr, "mmap_size not big enough\n");
		goto error;
	}

	ret_len = lttng_live_recv(control_sock, session->streams[id].mmap_base, len);
	if (ret_len == 0) {
		fprintf(stderr, "[error] Remote side has closed connection\n");
		goto error;
	}
	if (ret_len < 0) {
		fprintf(stderr, "Error receiving trace packet\n");
		goto error;
	}
end:
	return 0;
error:
	return -1;
}

int main(int argc, char **argv)
{
	int ret;
	int session_id;

	plan_tests(NUM_TESTS);

	diag("Live unit tests");

	ret = connect_viewer("localhost");
	ok(ret == 0, "Connect viewer to relayd");

	ret = establish_connection();
	ok(ret == 0, "Established connection and version check with %d.%d",
			VERSION_MAJOR, VERSION_MINOR);

	ret = list_sessions(&session_id);
	ok(ret > 0, "List sessions : %d session(s)", ret);

	ret = create_viewer_session();
	ok(ret == 0, "Create viewer session");

	ret = attach_session(session_id);
	ok(ret > 0, "Attach to session, %d streams received", ret);

	ret = get_metadata();
	ok(ret > 0, "Get metadata, received %d bytes", ret);

	ret = get_next_index();
	ok(ret == 0, "Get one index per stream");

	ret = get_data_packet(first_packet_stream_id, first_packet_offset,
			first_packet_len);
	ok(ret == 0,
			"Get one data packet for stream %d, offset %d, len %d",
			first_packet_stream_id, first_packet_offset,
			first_packet_len);

	return exit_status();
}
Commit	Line	Data
1b368955 JD	1	/*
	2	* Copyright (c) - 2013 Julien Desfossez <jdesfossez@efficios.com>
	3	*
	4	* This program is free software; you can redistribute it and/or modify it
	5	* under the terms of the GNU General Public License as published by as
	6	* published by the Free Software Foundation; only version 2 of the License.
	7	*
	8	* This program is distributed in the hope that it will be useful, but WITHOUT
	9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	10	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
	11	* more details.
	12	*
	13	* You should have received a copy of the GNU General Public License along with
	14	* this program; if not, write to the Free Software Foundation, Inc., 51
	15	* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	16	*/
	17
	18	#define _GNU_SOURCE
	19	#include <assert.h>
	20	#include <errno.h>
	21	#include <stdio.h>
	22	#include <stdlib.h>
	23	#include <string.h>
	24	#include <unistd.h>
	25	#include <time.h>
	26	#include <sys/types.h>
	27	#include <inttypes.h>
	28	#include <stdlib.h>
	29	#include <sys/socket.h>
	30	#include <netinet/in.h>
	31	#include <netdb.h>
	32	#include <fcntl.h>
	33	#include <sys/mman.h>
	34	#include <sys/stat.h>
	35
	36	#include <tap/tap.h>
	37	#include <lttng/lttng.h>
	38
	39	#include <urcu/list.h>
	40	#include <bin/lttng-sessiond/session.h>
	41	#include <common/common.h>
	42
2f8f53af	43	#include <bin/lttng-relayd/lttng-viewer-abi.h>
50adc264	44	#include <common/index/ctf-index.h>
1b368955	45
f263b7fd JD	46	#include <common/compat/endian.h>
f263b7fd JD	47
1b368955 JD	48	#define SESSION1 "test1"
	49	#define RELAYD_URL "net://localhost"
	50	#define LIVE_TIMER 2000000
	51
	52	/* Number of TAP tests in this file */
25b50f8f	53	#define NUM_TESTS 8
1b368955 JD	54	#define mmap_size 524288
	55
	56	int ust_consumerd32_fd;
	57	int ust_consumerd64_fd;
	58
	59	static int control_sock;
	60	struct live_session *session;
	61
	62	static int first_packet_offset;
	63	static int first_packet_len;
	64	static int first_packet_stream_id;
	65
	66	struct viewer_stream {
	67	uint64_t id;
	68	uint64_t ctf_trace_id;
	69	void *mmap_base;
	70	int fd;
	71	int metadata_flag;
	72	int first_read;
	73	char path[PATH_MAX];
	74	};
	75
	76	struct live_session {
	77	struct viewer_stream *streams;
	78	uint64_t live_timer_interval;
	79	uint64_t stream_count;
	80	};
	81
10236918 MD	82	static
	83	ssize_t lttng_live_recv(int fd, void *buf, size_t len)
	84	{
	85	ssize_t ret;
	86	size_t copied = 0, to_copy = len;
	87
	88	do {
	89	ret = recv(fd, buf + copied, to_copy, 0);
	90	if (ret > 0) {
	91	assert(ret <= to_copy);
	92	copied += ret;
	93	to_copy -= ret;
	94	}
	95	} while ((ret > 0 && to_copy > 0)
	96	\|\| (ret < 0 && errno == EINTR));
	97	if (ret > 0)
	98	ret = copied;
	99	/* ret = 0 means orderly shutdown, ret < 0 is error. */
	100	return ret;
	101	}
	102
	103	static
	104	ssize_t lttng_live_send(int fd, const void *buf, size_t len)
	105	{
	106	ssize_t ret;
	107
	108	do {
	109	ret = send(fd, buf, len, MSG_NOSIGNAL);
	110	} while (ret < 0 && errno == EINTR);
	111	return ret;
	112	}
	113
1b368955 JD	114	static
	115	int connect_viewer(char *hostname)
	116	{
	117	struct hostent *host;
	118	struct sockaddr_in server_addr;
	119	int ret;
	120
	121	host = gethostbyname(hostname);
	122	if (!host) {
	123	ret = -1;
	124	goto end;
	125	}
	126
	127	if ((control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
6f04ed72	128	PERROR("Socket");
1b368955 JD	129	ret = -1;
	130	goto end;
	131	}
	132
	133	server_addr.sin_family = AF_INET;
	134	server_addr.sin_port = htons(5344);
	135	server_addr.sin_addr = ((struct in_addr ) host->h_addr);
	136	bzero(&(server_addr.sin_zero), 8);
	137
	138	if (connect(control_sock, (struct sockaddr *) &server_addr,
	139	sizeof(struct sockaddr)) == -1) {
6f04ed72	140	PERROR("Connect");
1b368955 JD	141	ret = -1;
	142	goto end;
	143	}
	144
	145	server_addr.sin_family = AF_INET;
	146	server_addr.sin_port = htons(5345);
	147	server_addr.sin_addr = ((struct in_addr ) host->h_addr);
	148	bzero(&(server_addr.sin_zero), 8);
	149
	150	ret = 0;
	151
	152	end:
	153	return ret;
	154	}
	155
	156	int establish_connection(void)
	157	{
	158	struct lttng_viewer_cmd cmd;
	159	struct lttng_viewer_connect connect;
10236918	160	ssize_t ret_len;
1b368955	161
c4e361a4	162	cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
1b368955 JD	163	cmd.data_size = sizeof(connect);
	164	cmd.cmd_version = 0;
	165
dcf5c25e	166	memset(&connect, 0, sizeof(connect));
1b368955 JD	167	connect.major = htobe32(VERSION_MAJOR);
1b368955 JD	168	connect.minor = htobe32(VERSION_MINOR);
c4e361a4	169	connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
1b368955	170
10236918 MD	171	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
10236918 MD	172	if (ret_len < 0) {
1b368955 JD	173	fprintf(stderr, "Error sending cmd\n");
	174	goto error;
	175	}
10236918 MD	176	ret_len = lttng_live_send(control_sock, &connect, sizeof(connect));
10236918 MD	177	if (ret_len < 0) {
1b368955 JD	178	fprintf(stderr, "Error sending version\n");
	179	goto error;
	180	}
	181
10236918 MD	182	ret_len = lttng_live_recv(control_sock, &connect, sizeof(connect));
	183	if (ret_len == 0) {
	184	fprintf(stderr, "[error] Remote side has closed connection\n");
	185	goto error;
	186	}
	187	if (ret_len < 0) {
1b368955 JD	188	fprintf(stderr, "Error receiving version\n");
	189	goto error;
	190	}
10236918	191	return 0;
1b368955 JD	192
1b368955 JD	193	error:
10236918	194	return -1;
1b368955 JD	195	}
	196
	197	/*
	198	* Returns the number of sessions, should be 1 during the unit test.
	199	*/
	200	int list_sessions(int *session_id)
	201	{
	202	struct lttng_viewer_cmd cmd;
	203	struct lttng_viewer_list_sessions list;
	204	struct lttng_viewer_session lsession;
10236918 MD	205	int i;
10236918 MD	206	ssize_t ret_len;
1b368955 JD	207	int first_session = 0;
1b368955 JD	208
c4e361a4	209	cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
1b368955 JD	210	cmd.data_size = 0;
	211	cmd.cmd_version = 0;
	212
10236918 MD	213	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
10236918 MD	214	if (ret_len < 0) {
1b368955 JD	215	fprintf(stderr, "Error sending cmd\n");
	216	goto error;
	217	}
	218
10236918 MD	219	ret_len = lttng_live_recv(control_sock, &list, sizeof(list));
	220	if (ret_len == 0) {
	221	fprintf(stderr, "[error] Remote side has closed connection\n");
	222	goto error;
	223	}
	224	if (ret_len < 0) {
1b368955 JD	225	fprintf(stderr, "Error receiving session list\n");
	226	goto error;
	227	}
	228
	229	for (i = 0; i < be32toh(list.sessions_count); i++) {
10236918 MD	230	ret_len = lttng_live_recv(control_sock, &lsession, sizeof(lsession));
10236918 MD	231	if (ret_len < 0) {
1b368955 JD	232	fprintf(stderr, "Error receiving session\n");
	233	goto error;
	234	}
	235	if (lsession.streams > 0 && first_session <= 0) {
	236	first_session = be64toh(lsession.id);
	237	*session_id = first_session;
	238	}
	239	}
	240
10236918	241	return be32toh(list.sessions_count);
1b368955 JD	242
1b368955 JD	243	error:
10236918	244	return -1;
1b368955 JD	245	}
1b368955 JD	246
10236918	247	int create_viewer_session(void)
25b50f8f JD	248	{
	249	struct lttng_viewer_cmd cmd;
	250	struct lttng_viewer_create_session_response resp;
25b50f8f JD	251	ssize_t ret_len;
	252
	253	cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
	254	cmd.data_size = 0;
	255	cmd.cmd_version = 0;
	256
10236918	257	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
25b50f8f JD	258	if (ret_len < 0) {
25b50f8f JD	259	fprintf(stderr, "[error] Error sending cmd\n");
25b50f8f JD	260	goto error;
	261	}
	262	assert(ret_len == sizeof(cmd));
	263
10236918 MD	264	ret_len = lttng_live_recv(control_sock, &resp, sizeof(resp));
	265	if (ret_len == 0) {
	266	fprintf(stderr, "[error] Remote side has closed connection\n");
	267	goto error;
	268	}
25b50f8f JD	269	if (ret_len < 0) {
25b50f8f JD	270	fprintf(stderr, "[error] Error receiving create session reply\n");
25b50f8f JD	271	goto error;
	272	}
	273	assert(ret_len == sizeof(resp));
	274
	275	if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
	276	fprintf(stderr, "[error] Error creating viewer session\n");
25b50f8f JD	277	goto error;
25b50f8f JD	278	}
10236918	279	return 0;
25b50f8f JD	280
25b50f8f JD	281	error:
10236918	282	return -1;
25b50f8f JD	283	}
25b50f8f JD	284
1b368955 JD	285	int attach_session(int id)
	286	{
	287	struct lttng_viewer_cmd cmd;
	288	struct lttng_viewer_attach_session_request rq;
	289	struct lttng_viewer_attach_session_response rp;
	290	struct lttng_viewer_stream stream;
10236918 MD	291	int i;
10236918 MD	292	ssize_t ret_len;
1b368955 JD	293
	294	session = zmalloc(sizeof(struct live_session));
	295	if (!session) {
1b368955 JD	296	goto error;
	297	}
	298
c4e361a4	299	cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
1b368955 JD	300	cmd.data_size = sizeof(rq);
	301	cmd.cmd_version = 0;
	302
dcf5c25e	303	memset(&rq, 0, sizeof(rq));
1b368955	304	rq.session_id = htobe64(id);
c4e361a4	305	rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
1b368955	306
10236918 MD	307	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
10236918 MD	308	if (ret_len < 0) {
1b368955 JD	309	fprintf(stderr, "Error sending cmd\n");
	310	goto error;
	311	}
10236918 MD	312	ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
10236918 MD	313	if (ret_len < 0) {
1b368955 JD	314	fprintf(stderr, "Error sending attach request\n");
	315	goto error;
	316	}
	317
10236918 MD	318	ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
	319	if (ret_len == 0) {
	320	fprintf(stderr, "[error] Remote side has closed connection\n");
	321	goto error;
	322	}
	323	if (ret_len < 0) {
1b368955 JD	324	fprintf(stderr, "Error receiving attach response\n");
	325	goto error;
	326	}
c4e361a4	327	if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
10236918	328	goto error;
1b368955 JD	329	}
	330
	331	session->stream_count = be32toh(rp.streams_count);
	332	session->streams = zmalloc(session->stream_count *
	333	sizeof(struct viewer_stream));
	334	if (!session->streams) {
1b368955 JD	335	goto error;
	336	}
	337
	338	for (i = 0; i < be32toh(rp.streams_count); i++) {
10236918 MD	339	ret_len = lttng_live_recv(control_sock, &stream, sizeof(stream));
	340	if (ret_len == 0) {
	341	fprintf(stderr, "[error] Remote side has closed connection\n");
	342	goto error;
	343	}
	344	if (ret_len < 0) {
1b368955 JD	345	fprintf(stderr, "Error receiving stream\n");
	346	goto error;
	347	}
	348	session->streams[i].id = be64toh(stream.id);
	349
	350	session->streams[i].ctf_trace_id = be64toh(stream.ctf_trace_id);
	351	session->streams[i].first_read = 1;
	352	session->streams[i].mmap_base = mmap(NULL, mmap_size, PROT_READ \| PROT_WRITE,
	353	MAP_PRIVATE \| MAP_ANONYMOUS, -1, 0);
	354	if (session->streams[i].mmap_base == MAP_FAILED) {
	355	fprintf(stderr, "mmap error\n");
1b368955 JD	356	goto error;
	357	}
	358
	359	if (be32toh(stream.metadata_flag)) {
	360	session->streams[i].metadata_flag = 1;
	361	}
	362	}
10236918	363	return session->stream_count;
1b368955	364
1b368955	365	error:
10236918	366	return -1;
1b368955 JD	367	}
	368
	369	int get_metadata(void)
	370	{
	371	struct lttng_viewer_cmd cmd;
	372	struct lttng_viewer_get_metadata rq;
	373	struct lttng_viewer_metadata_packet rp;
10236918	374	ssize_t ret_len;
1b368955 JD	375	int ret;
	376	uint64_t i;
	377	char *data = NULL;
	378	uint64_t len = 0;
	379	int metadata_stream_id = -1;
	380
c4e361a4	381	cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
1b368955 JD	382	cmd.data_size = sizeof(rq);
	383	cmd.cmd_version = 0;
	384
	385	for (i = 0; i < session->stream_count; i++) {
	386	if (session->streams[i].metadata_flag) {
	387	metadata_stream_id = i;
	388	break;
	389	}
	390	}
	391
	392	if (metadata_stream_id < 0) {
	393	fprintf(stderr, "No metadata stream found\n");
1b368955 JD	394	goto error;
	395	}
	396
	397	rq.stream_id = htobe64(session->streams[metadata_stream_id].id);
	398
10236918 MD	399	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
10236918 MD	400	if (ret_len < 0) {
1b368955 JD	401	fprintf(stderr, "Error sending cmd\n");
	402	goto error;
	403	}
10236918 MD	404	ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
10236918 MD	405	if (ret_len < 0) {
1b368955 JD	406	fprintf(stderr, "Error sending get_metadata request\n");
	407	goto error;
	408	}
10236918 MD	409	ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
	410	if (ret_len == 0) {
	411	fprintf(stderr, "[error] Remote side has closed connection\n");
	412	goto error;
	413	}
	414	if (ret_len < 0) {
1b368955 JD	415	fprintf(stderr, "Error receiving metadata response\n");
	416	goto error;
	417	}
	418	switch (be32toh(rp.status)) {
c4e361a4	419	case LTTNG_VIEWER_METADATA_OK:
1b368955	420	break;
c4e361a4	421	case LTTNG_VIEWER_NO_NEW_METADATA:
1b368955	422	fprintf(stderr, "NO NEW\n");
10236918	423	ret = 0;
1b368955	424	goto end;
c4e361a4	425	case LTTNG_VIEWER_METADATA_ERR:
1b368955	426	fprintf(stderr, "ERR\n");
10236918	427	goto error;
1b368955 JD	428	default:
1b368955 JD	429	fprintf(stderr, "UNKNOWN\n");
10236918	430	goto error;
1b368955 JD	431	}
	432
	433	len = be64toh(rp.len);
	434	if (len <= 0) {
10236918	435	goto error;
1b368955 JD	436	}
	437
	438	data = zmalloc(len);
	439	if (!data) {
6f04ed72	440	PERROR("relay data zmalloc");
1b368955 JD	441	goto error;
1b368955 JD	442	}
10236918 MD	443	ret_len = lttng_live_recv(control_sock, data, len);
	444	if (ret_len == 0) {
	445	fprintf(stderr, "[error] Remote side has closed connection\n");
	446	goto error_free_data;
	447	}
	448	if (ret_len < 0) {
1b368955	449	fprintf(stderr, "Error receiving trace packet\n");
10236918	450	goto error_free_data;
1b368955 JD	451	}
1b368955 JD	452	free(data);
10236918	453	ret = len;
1b368955	454	end:
1b368955	455	return ret;
10236918 MD	456
	457	error_free_data:
	458	free(data);
	459	error:
	460	return -1;
1b368955 JD	461	}
	462
	463	int get_next_index(void)
	464	{
	465	struct lttng_viewer_cmd cmd;
	466	struct lttng_viewer_get_next_index rq;
	467	struct lttng_viewer_index rp;
10236918	468	ssize_t ret_len;
1b368955 JD	469	int id;
1b368955 JD	470
c4e361a4	471	cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
1b368955 JD	472	cmd.data_size = sizeof(rq);
	473	cmd.cmd_version = 0;
	474
	475	for (id = 0; id < session->stream_count; id++) {
	476	if (session->streams[id].metadata_flag) {
	477	continue;
	478	}
10236918	479	memset(&rq, 0, sizeof(rq));
1b368955 JD	480	rq.stream_id = htobe64(session->streams[id].id);
	481
	482	retry:
10236918 MD	483	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
10236918 MD	484	if (ret_len < 0) {
1b368955 JD	485	fprintf(stderr, "Error sending cmd\n");
	486	goto error;
	487	}
10236918 MD	488	ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
10236918 MD	489	if (ret_len < 0) {
1b368955 JD	490	fprintf(stderr, "Error sending get_next_index request\n");
	491	goto error;
	492	}
10236918 MD	493	ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
	494	if (ret_len == 0) {
	495	fprintf(stderr, "[error] Remote side has closed connection\n");
	496	goto error;
	497	}
	498	if (ret_len < 0) {
1b368955 JD	499	fprintf(stderr, "Error receiving index response\n");
	500	goto error;
	501	}
	502
	503	rp.flags = be32toh(rp.flags);
	504
	505	switch (be32toh(rp.status)) {
c4e361a4	506	case LTTNG_VIEWER_INDEX_INACTIVE:
1b368955 JD	507	fprintf(stderr, "(INACTIVE)\n");
1b368955 JD	508	break;
c4e361a4	509	case LTTNG_VIEWER_INDEX_OK:
1b368955	510	break;
c4e361a4	511	case LTTNG_VIEWER_INDEX_RETRY:
1b368955 JD	512	sleep(1);
1b368955 JD	513	goto retry;
c4e361a4	514	case LTTNG_VIEWER_INDEX_HUP:
1b368955 JD	515	fprintf(stderr, "(HUP)\n");
	516	session->streams[id].id = -1ULL;
	517	session->streams[id].fd = -1;
	518	break;
c4e361a4	519	case LTTNG_VIEWER_INDEX_ERR:
1b368955	520	fprintf(stderr, "(ERR)\n");
1b368955 JD	521	goto error;
	522	default:
	523	fprintf(stderr, "SHOULD NOT HAPPEN\n");
1b368955 JD	524	goto error;
	525	}
	526	if (!first_packet_stream_id) {
	527	first_packet_offset = be64toh(rp.offset);
	528	first_packet_len = be64toh(rp.packet_size) / CHAR_BIT;
	529	first_packet_stream_id = id;
	530	}
	531	}
10236918	532	return 0;
1b368955 JD	533
1b368955 JD	534	error:
10236918	535	return -1;
1b368955 JD	536	}
	537
	538	static
	539	int get_data_packet(int id, uint64_t offset,
	540	uint64_t len)
	541	{
	542	struct lttng_viewer_cmd cmd;
	543	struct lttng_viewer_get_packet rq;
	544	struct lttng_viewer_trace_packet rp;
10236918	545	ssize_t ret_len;
1b368955	546
c4e361a4	547	cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
1b368955 JD	548	cmd.data_size = sizeof(rq);
	549	cmd.cmd_version = 0;
	550
10236918	551	memset(&rq, 0, sizeof(rq));
1b368955 JD	552	rq.stream_id = htobe64(session->streams[id].id);
	553	/* Already in big endian. */
	554	rq.offset = offset;
	555	rq.len = htobe32(len);
	556
10236918 MD	557	ret_len = lttng_live_send(control_sock, &cmd, sizeof(cmd));
10236918 MD	558	if (ret_len < 0) {
1b368955 JD	559	fprintf(stderr, "Error sending cmd\n");
	560	goto error;
	561	}
10236918 MD	562	ret_len = lttng_live_send(control_sock, &rq, sizeof(rq));
10236918 MD	563	if (ret_len < 0) {
1b368955 JD	564	fprintf(stderr, "Error sending get_data_packet request\n");
	565	goto error;
	566	}
10236918 MD	567	ret_len = lttng_live_recv(control_sock, &rp, sizeof(rp));
	568	if (ret_len == 0) {
	569	fprintf(stderr, "[error] Remote side has closed connection\n");
	570	goto error;
	571	}
	572	if (ret_len < 0) {
1b368955 JD	573	fprintf(stderr, "Error receiving data response\n");
	574	goto error;
	575	}
	576	rp.flags = be32toh(rp.flags);
	577
	578	switch (be32toh(rp.status)) {
c4e361a4	579	case LTTNG_VIEWER_GET_PACKET_OK:
10236918	580	len = be32toh(rp.len);
1b368955	581	break;
c4e361a4	582	case LTTNG_VIEWER_GET_PACKET_RETRY:
1b368955	583	fprintf(stderr, "RETRY\n");
10236918	584	goto error;
c4e361a4	585	case LTTNG_VIEWER_GET_PACKET_ERR:
1b368955 JD	586	if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1b368955 JD	587	fprintf(stderr, "NEW_METADATA\n");
1b368955 JD	588	goto end;
	589	}
	590	fprintf(stderr, "ERR\n");
10236918	591	goto error;
1b368955 JD	592	default:
1b368955 JD	593	fprintf(stderr, "UNKNOWN\n");
10236918	594	goto error;
1b368955 JD	595	}
1b368955 JD	596
10236918 MD	597	if (len == 0) {
10236918 MD	598	goto error;
1b368955 JD	599	}
	600
	601	if (len > mmap_size) {
	602	fprintf(stderr, "mmap_size not big enough\n");
1b368955 JD	603	goto error;
	604	}
	605
10236918 MD	606	ret_len = lttng_live_recv(control_sock, session->streams[id].mmap_base, len);
	607	if (ret_len == 0) {
	608	fprintf(stderr, "[error] Remote side has closed connection\n");
	609	goto error;
	610	}
	611	if (ret_len < 0) {
1b368955 JD	612	fprintf(stderr, "Error receiving trace packet\n");
	613	goto error;
	614	}
1b368955	615	end:
10236918	616	return 0;
1b368955	617	error:
10236918	618	return -1;
1b368955 JD	619	}
	620
	621	int main(int argc, char **argv)
	622	{
	623	int ret;
	624	int session_id;
	625
	626	plan_tests(NUM_TESTS);
	627
	628	diag("Live unit tests");
	629
	630	ret = connect_viewer("localhost");
	631	ok(ret == 0, "Connect viewer to relayd");
	632
	633	ret = establish_connection();
	634	ok(ret == 0, "Established connection and version check with %d.%d",
	635	VERSION_MAJOR, VERSION_MINOR);
	636
	637	ret = list_sessions(&session_id);
	638	ok(ret > 0, "List sessions : %d session(s)", ret);
	639
25b50f8f JD	640	ret = create_viewer_session();
	641	ok(ret == 0, "Create viewer session");
	642
1b368955 JD	643	ret = attach_session(session_id);
	644	ok(ret > 0, "Attach to session, %d streams received", ret);
	645
	646	ret = get_metadata();
	647	ok(ret > 0, "Get metadata, received %d bytes", ret);
	648
	649	ret = get_next_index();
	650	ok(ret == 0, "Get one index per stream");
	651
	652	ret = get_data_packet(first_packet_stream_id, first_packet_offset,
	653	first_packet_len);
10236918	654	ok(ret == 0,
1b368955 JD	655	"Get one data packet for stream %d, offset %d, len %d",
	656	first_packet_stream_id, first_packet_offset,
	657	first_packet_len);
	658
	659	return exit_status();
	660	}