2 * Copyright (c) - 2013 Julien Desfossez <jdesfossez@efficios.com>
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.
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
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.
26 #include <sys/types.h>
29 #include <sys/socket.h>
30 #include <netinet/in.h>
37 #include <lttng/lttng.h>
39 #include <urcu/list.h>
40 #include <bin/lttng-sessiond/session.h>
41 #include <common/common.h>
43 #include <bin/lttng-relayd/lttng-viewer-abi.h>
44 #include <common/index/ctf-index.h>
46 #include <common/compat/endian.h>
48 #define SESSION1 "test1"
49 #define RELAYD_URL "net://localhost"
50 #define LIVE_TIMER 2000000
52 /* Number of TAP tests in this file */
54 #define mmap_size 524288
56 int ust_consumerd32_fd
;
57 int ust_consumerd64_fd
;
59 static int control_sock
;
60 struct live_session
*session
;
62 static int first_packet_offset
;
63 static int first_packet_len
;
64 static int first_packet_stream_id
= -1;
66 struct viewer_stream
{
68 uint64_t ctf_trace_id
;
77 struct viewer_stream
*streams
;
78 uint64_t live_timer_interval
;
79 uint64_t stream_count
;
83 ssize_t
lttng_live_recv(int fd
, void *buf
, size_t len
)
86 size_t copied
= 0, to_copy
= len
;
89 ret
= recv(fd
, buf
+ copied
, to_copy
, 0);
91 assert(ret
<= to_copy
);
95 } while ((ret
> 0 && to_copy
> 0)
96 || (ret
< 0 && errno
== EINTR
));
99 /* ret = 0 means orderly shutdown, ret < 0 is error. */
104 ssize_t
lttng_live_send(int fd
, const void *buf
, size_t len
)
109 ret
= send(fd
, buf
, len
, MSG_NOSIGNAL
);
110 } while (ret
< 0 && errno
== EINTR
);
115 int connect_viewer(char *hostname
)
117 struct hostent
*host
;
118 struct sockaddr_in server_addr
;
121 host
= gethostbyname(hostname
);
127 if ((control_sock
= socket(AF_INET
, SOCK_STREAM
, 0)) == -1) {
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);
138 if (connect(control_sock
, (struct sockaddr
*) &server_addr
,
139 sizeof(struct sockaddr
)) == -1) {
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);
156 int establish_connection(void)
158 struct lttng_viewer_cmd cmd
;
159 struct lttng_viewer_connect connect
;
162 cmd
.cmd
= htobe32(LTTNG_VIEWER_CONNECT
);
163 cmd
.data_size
= sizeof(connect
);
166 memset(&connect
, 0, sizeof(connect
));
167 connect
.major
= htobe32(VERSION_MAJOR
);
168 connect
.minor
= htobe32(VERSION_MINOR
);
169 connect
.type
= htobe32(LTTNG_VIEWER_CLIENT_COMMAND
);
171 ret_len
= lttng_live_send(control_sock
, &cmd
, sizeof(cmd
));
173 diag("Error sending cmd");
176 ret_len
= lttng_live_send(control_sock
, &connect
, sizeof(connect
));
178 diag("Error sending version");
182 ret_len
= lttng_live_recv(control_sock
, &connect
, sizeof(connect
));
184 diag("[error] Remote side has closed connection");
188 diag("Error receiving version");
198 * Returns the number of sessions, should be 1 during the unit test.
200 int list_sessions(uint64_t *session_id
)
202 struct lttng_viewer_cmd cmd
;
203 struct lttng_viewer_list_sessions list
;
204 struct lttng_viewer_session lsession
;
207 int first_session
= 0;
209 cmd
.cmd
= htobe32(LTTNG_VIEWER_LIST_SESSIONS
);
213 ret_len
= lttng_live_send(control_sock
, &cmd
, sizeof(cmd
));
215 diag("Error sending cmd");
219 ret_len
= lttng_live_recv(control_sock
, &list
, sizeof(list
));
221 diag("[error] Remote side has closed connection");
225 diag("Error receiving session list");
229 for (i
= 0; i
< be32toh(list
.sessions_count
); i
++) {
230 ret_len
= lttng_live_recv(control_sock
, &lsession
, sizeof(lsession
));
232 diag("Error receiving session");
235 if (lsession
.streams
> 0 && first_session
<= 0) {
236 first_session
= be64toh(lsession
.id
);
237 *session_id
= first_session
;
241 return be32toh(list
.sessions_count
);
247 int create_viewer_session(void)
249 struct lttng_viewer_cmd cmd
;
250 struct lttng_viewer_create_session_response resp
;
253 cmd
.cmd
= htobe32(LTTNG_VIEWER_CREATE_SESSION
);
257 ret_len
= lttng_live_send(control_sock
, &cmd
, sizeof(cmd
));
259 diag("[error] Error sending cmd");
262 assert(ret_len
== sizeof(cmd
));
264 ret_len
= lttng_live_recv(control_sock
, &resp
, sizeof(resp
));
266 diag("[error] Remote side has closed connection");
270 diag("[error] Error receiving create session reply");
273 assert(ret_len
== sizeof(resp
));
275 if (be32toh(resp
.status
) != LTTNG_VIEWER_CREATE_SESSION_OK
) {
276 diag("[error] Error creating viewer session");
285 int attach_session(uint64_t id
)
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
;
294 session
= zmalloc(sizeof(struct live_session
));
299 cmd
.cmd
= htobe32(LTTNG_VIEWER_ATTACH_SESSION
);
300 cmd
.data_size
= sizeof(rq
);
303 memset(&rq
, 0, sizeof(rq
));
304 rq
.session_id
= htobe64(id
);
305 rq
.seek
= htobe32(LTTNG_VIEWER_SEEK_BEGINNING
);
307 ret_len
= lttng_live_send(control_sock
, &cmd
, sizeof(cmd
));
309 diag("Error sending cmd LTTNG_VIEWER_ATTACH_SESSION");
312 ret_len
= lttng_live_send(control_sock
, &rq
, sizeof(rq
));
314 diag("Error sending attach request");
318 ret_len
= lttng_live_recv(control_sock
, &rp
, sizeof(rp
));
320 diag("[error] Remote side has closed connection");
324 diag("Error receiving attach response");
327 if (be32toh(rp
.status
) != LTTNG_VIEWER_ATTACH_OK
) {
331 session
->stream_count
= be32toh(rp
.streams_count
);
332 if (session
->stream_count
== 0) {
333 diag("Got session stream count == 0");
336 session
->streams
= zmalloc(session
->stream_count
*
337 sizeof(struct viewer_stream
));
338 if (!session
->streams
) {
342 for (i
= 0; i
< be32toh(rp
.streams_count
); i
++) {
343 ret_len
= lttng_live_recv(control_sock
, &stream
, sizeof(stream
));
345 diag("[error] Remote side has closed connection");
349 diag("Error receiving stream");
352 session
->streams
[i
].id
= be64toh(stream
.id
);
354 session
->streams
[i
].ctf_trace_id
= be64toh(stream
.ctf_trace_id
);
355 session
->streams
[i
].first_read
= 1;
356 session
->streams
[i
].mmap_base
= mmap(NULL
, mmap_size
,
357 PROT_READ
| PROT_WRITE
,
358 MAP_PRIVATE
| MAP_ANONYMOUS
, -1, 0);
359 if (session
->streams
[i
].mmap_base
== MAP_FAILED
) {
364 if (be32toh(stream
.metadata_flag
)) {
365 session
->streams
[i
].metadata_flag
= 1;
368 return session
->stream_count
;
374 int get_metadata(void)
376 struct lttng_viewer_cmd cmd
;
377 struct lttng_viewer_get_metadata rq
;
378 struct lttng_viewer_metadata_packet rp
;
384 int metadata_stream_id
= -1;
386 cmd
.cmd
= htobe32(LTTNG_VIEWER_GET_METADATA
);
387 cmd
.data_size
= sizeof(rq
);
390 for (i
= 0; i
< session
->stream_count
; i
++) {
391 if (session
->streams
[i
].metadata_flag
) {
392 metadata_stream_id
= i
;
397 if (metadata_stream_id
< 0) {
398 diag("No metadata stream found");
402 rq
.stream_id
= htobe64(session
->streams
[metadata_stream_id
].id
);
404 ret_len
= lttng_live_send(control_sock
, &cmd
, sizeof(cmd
));
406 diag("Error sending cmd");
409 ret_len
= lttng_live_send(control_sock
, &rq
, sizeof(rq
));
411 diag("Error sending get_metadata request");
414 ret_len
= lttng_live_recv(control_sock
, &rp
, sizeof(rp
));
416 diag("[error] Remote side has closed connection");
420 diag("Error receiving metadata response");
423 switch (be32toh(rp
.status
)) {
424 case LTTNG_VIEWER_METADATA_OK
:
426 case LTTNG_VIEWER_NO_NEW_METADATA
:
427 diag("Got LTTNG_VIEWER_NO_NEW_METADATA:");
430 case LTTNG_VIEWER_METADATA_ERR
:
431 diag("Got LTTNG_VIEWER_METADATA_ERR:");
434 diag("Got unknown status during LTTNG_VIEWER_GET_METADATA");
438 len
= be64toh(rp
.len
);
445 PERROR("relay data zmalloc");
448 ret_len
= lttng_live_recv(control_sock
, data
, len
);
450 diag("[error] Remote side has closed connection");
451 goto error_free_data
;
454 diag("Error receiving trace packet");
455 goto error_free_data
;
468 int get_next_index(void)
470 struct lttng_viewer_cmd cmd
;
471 struct lttng_viewer_get_next_index rq
;
472 struct lttng_viewer_index rp
;
476 cmd
.cmd
= htobe32(LTTNG_VIEWER_GET_NEXT_INDEX
);
477 cmd
.data_size
= sizeof(rq
);
480 for (id
= 0; id
< session
->stream_count
; id
++) {
481 if (session
->streams
[id
].metadata_flag
) {
484 memset(&rq
, 0, sizeof(rq
));
485 rq
.stream_id
= htobe64(session
->streams
[id
].id
);
488 ret_len
= lttng_live_send(control_sock
, &cmd
, sizeof(cmd
));
490 diag("Error sending cmd");
493 ret_len
= lttng_live_send(control_sock
, &rq
, sizeof(rq
));
495 diag("Error sending get_next_index request");
498 ret_len
= lttng_live_recv(control_sock
, &rp
, sizeof(rp
));
500 diag("[error] Remote side has closed connection");
504 diag("Error receiving index response");
508 rp
.flags
= be32toh(rp
.flags
);
510 switch (be32toh(rp
.status
)) {
511 case LTTNG_VIEWER_INDEX_INACTIVE
:
512 /* Skip this stream. */
513 diag("Got LTTNG_VIEWER_INDEX_INACTIVE");
515 case LTTNG_VIEWER_INDEX_OK
:
517 case LTTNG_VIEWER_INDEX_RETRY
:
520 case LTTNG_VIEWER_INDEX_HUP
:
521 diag("Got LTTNG_VIEWER_INDEX_HUP");
522 session
->streams
[id
].id
= -1ULL;
523 session
->streams
[id
].fd
= -1;
525 case LTTNG_VIEWER_INDEX_ERR
:
526 diag("Got LTTNG_VIEWER_INDEX_ERR");
529 diag("Unknown reply status during LTTNG_VIEWER_GET_NEXT_INDEX (%d)", be32toh(rp
.status
));
532 if (first_packet_stream_id
< 0) {
534 * Initialize the first packet stream id. That is,
535 * the first active stream encoutered.
537 first_packet_offset
= be64toh(rp
.offset
);
538 first_packet_len
= be64toh(rp
.packet_size
) / CHAR_BIT
;
539 first_packet_stream_id
= id
;
540 diag("Got first packet index with offset %d and len %d",
541 first_packet_offset
, first_packet_len
);
551 int get_data_packet(int id
, uint64_t offset
,
554 struct lttng_viewer_cmd cmd
;
555 struct lttng_viewer_get_packet rq
;
556 struct lttng_viewer_trace_packet rp
;
559 cmd
.cmd
= htobe32(LTTNG_VIEWER_GET_PACKET
);
560 cmd
.data_size
= sizeof(rq
);
563 memset(&rq
, 0, sizeof(rq
));
564 rq
.stream_id
= htobe64(session
->streams
[id
].id
);
565 /* Already in big endian. */
567 rq
.len
= htobe32(len
);
569 ret_len
= lttng_live_send(control_sock
, &cmd
, sizeof(cmd
));
571 diag("Error sending cmd");
574 ret_len
= lttng_live_send(control_sock
, &rq
, sizeof(rq
));
576 diag("Error sending get_data_packet request");
579 ret_len
= lttng_live_recv(control_sock
, &rp
, sizeof(rp
));
581 diag("[error] Remote side has closed connection");
585 diag("Error receiving data response");
588 rp
.flags
= be32toh(rp
.flags
);
590 switch (be32toh(rp
.status
)) {
591 case LTTNG_VIEWER_GET_PACKET_OK
:
592 len
= be32toh(rp
.len
);
594 diag("Got LTTNG_VIEWER_GET_PACKET_OK, but len == 0");
598 case LTTNG_VIEWER_GET_PACKET_RETRY
:
599 diag("Got LTTNG_VIEWER_GET_PACKET_RETRY:");
601 case LTTNG_VIEWER_GET_PACKET_ERR
:
602 if (rp
.flags
& LTTNG_VIEWER_FLAG_NEW_METADATA
) {
603 diag("Got LTTNG_VIEWER_GET_PACKET_ERR with NEW_METADATA flag");
606 diag("Got LTTNG_VIEWER_GET_PACKET_ERR:");
609 diag("Got unknown status code during LTTNG_VIEWER_GET_PACKET");
613 if (len
> mmap_size
) {
614 diag("mmap_size not big enough");
618 ret_len
= lttng_live_recv(control_sock
, session
->streams
[id
].mmap_base
, len
);
620 diag("[error] Remote side has closed connection");
624 diag("Error receiving trace packet");
633 int main(int argc
, char **argv
)
638 plan_tests(NUM_TESTS
);
640 diag("Live unit tests");
642 ret
= connect_viewer("localhost");
643 ok(ret
== 0, "Connect viewer to relayd");
645 ret
= establish_connection();
646 ok(ret
== 0, "Established connection and version check with %d.%d",
647 VERSION_MAJOR
, VERSION_MINOR
);
649 ret
= list_sessions(&session_id
);
650 ok(ret
> 0, "List sessions : %d session(s)", ret
);
652 ret
= create_viewer_session();
653 ok(ret
== 0, "Create viewer session");
655 ret
= attach_session(session_id
);
656 ok(ret
> 0, "Attach to session, %d stream(s) received", ret
);
658 ret
= get_metadata();
659 ok(ret
> 0, "Get metadata, received %d bytes", ret
);
661 ret
= get_next_index();
662 ok(ret
== 0, "Get one index per stream");
664 ret
= get_data_packet(first_packet_stream_id
, first_packet_offset
,
667 "Get one data packet for stream %d, offset %d, len %d",
668 first_packet_stream_id
, first_packet_offset
,
671 return exit_status();