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.h>
44 #include <common/index/lttng-index.h>
46 #define SESSION1 "test1"
47 #define RELAYD_URL "net://localhost"
48 #define LIVE_TIMER 2000000
50 /* Number of TAP tests in this file */
52 #define mmap_size 524288
54 int ust_consumerd32_fd
;
55 int ust_consumerd64_fd
;
57 static int control_sock
;
58 struct live_session
*session
;
60 static int first_packet_offset
;
61 static int first_packet_len
;
62 static int first_packet_stream_id
;
64 struct viewer_stream
{
66 uint64_t ctf_trace_id
;
75 struct viewer_stream
*streams
;
76 uint64_t live_timer_interval
;
77 uint64_t stream_count
;
81 int connect_viewer(char *hostname
)
84 struct sockaddr_in server_addr
;
87 host
= gethostbyname(hostname
);
93 if ((control_sock
= socket(AF_INET
, SOCK_STREAM
, 0)) == -1) {
99 server_addr
.sin_family
= AF_INET
;
100 server_addr
.sin_port
= htons(5344);
101 server_addr
.sin_addr
= *((struct in_addr
*) host
->h_addr
);
102 bzero(&(server_addr
.sin_zero
), 8);
104 if (connect(control_sock
, (struct sockaddr
*) &server_addr
,
105 sizeof(struct sockaddr
)) == -1) {
111 server_addr
.sin_family
= AF_INET
;
112 server_addr
.sin_port
= htons(5345);
113 server_addr
.sin_addr
= *((struct in_addr
*) host
->h_addr
);
114 bzero(&(server_addr
.sin_zero
), 8);
122 int establish_connection(void)
124 struct lttng_viewer_cmd cmd
;
125 struct lttng_viewer_connect connect
;
128 cmd
.cmd
= htobe32(VIEWER_CONNECT
);
129 cmd
.data_size
= sizeof(connect
);
132 connect
.major
= htobe32(VERSION_MAJOR
);
133 connect
.minor
= htobe32(VERSION_MINOR
);
134 connect
.type
= htobe32(VIEWER_CLIENT_COMMAND
);
137 ret
= send(control_sock
, &cmd
, sizeof(cmd
), 0);
138 } while (ret
< 0 && errno
== EINTR
);
140 fprintf(stderr
, "Error sending cmd\n");
144 ret
= send(control_sock
, &connect
, sizeof(connect
), 0);
145 } while (ret
< 0 && errno
== EINTR
);
147 fprintf(stderr
, "Error sending version\n");
152 ret
= recv(control_sock
, &connect
, sizeof(connect
), 0);
153 } while (ret
< 0 && errno
== EINTR
);
155 fprintf(stderr
, "Error receiving version\n");
165 * Returns the number of sessions, should be 1 during the unit test.
167 int list_sessions(int *session_id
)
169 struct lttng_viewer_cmd cmd
;
170 struct lttng_viewer_list_sessions list
;
171 struct lttng_viewer_session lsession
;
173 int first_session
= 0;
175 cmd
.cmd
= htobe32(VIEWER_LIST_SESSIONS
);
180 ret
= send(control_sock
, &cmd
, sizeof(cmd
), 0);
181 } while (ret
< 0 && errno
== EINTR
);
183 fprintf(stderr
, "Error sending cmd\n");
188 ret
= recv(control_sock
, &list
, sizeof(list
), 0);
189 } while (ret
< 0 && errno
== EINTR
);
191 fprintf(stderr
, "Error receiving session list\n");
195 for (i
= 0; i
< be32toh(list
.sessions_count
); i
++) {
197 ret
= recv(control_sock
, &lsession
, sizeof(lsession
), 0);
198 } while (ret
< 0 && errno
== EINTR
);
200 fprintf(stderr
, "Error receiving session\n");
203 if (lsession
.streams
> 0 && first_session
<= 0) {
204 first_session
= be64toh(lsession
.id
);
205 *session_id
= first_session
;
209 ret
= be32toh(list
.sessions_count
);
215 int attach_session(int id
)
217 struct lttng_viewer_cmd cmd
;
218 struct lttng_viewer_attach_session_request rq
;
219 struct lttng_viewer_attach_session_response rp
;
220 struct lttng_viewer_stream stream
;
223 session
= zmalloc(sizeof(struct live_session
));
229 cmd
.cmd
= htobe32(VIEWER_ATTACH_SESSION
);
230 cmd
.data_size
= sizeof(rq
);
233 rq
.session_id
= htobe64(id
);
234 rq
.seek
= htobe32(VIEWER_SEEK_BEGINNING
);
237 ret
= send(control_sock
, &cmd
, sizeof(cmd
), 0);
238 } while (ret
< 0 && errno
== EINTR
);
240 fprintf(stderr
, "Error sending cmd\n");
244 ret
= send(control_sock
, &rq
, sizeof(rq
), 0);
245 } while (ret
< 0 && errno
== EINTR
);
247 fprintf(stderr
, "Error sending attach request\n");
252 ret
= recv(control_sock
, &rp
, sizeof(rp
), 0);
253 } while (ret
< 0 && errno
== EINTR
);
255 fprintf(stderr
, "Error receiving attach response\n");
258 if (be32toh(rp
.status
) != VIEWER_ATTACH_OK
) {
263 session
->stream_count
= be32toh(rp
.streams_count
);
264 session
->streams
= zmalloc(session
->stream_count
*
265 sizeof(struct viewer_stream
));
266 if (!session
->streams
) {
271 for (i
= 0; i
< be32toh(rp
.streams_count
); i
++) {
273 ret
= recv(control_sock
, &stream
, sizeof(stream
), 0);
274 } while (ret
< 0 && errno
== EINTR
);
276 fprintf(stderr
, "Error receiving stream\n");
279 session
->streams
[i
].id
= be64toh(stream
.id
);
281 session
->streams
[i
].ctf_trace_id
= be64toh(stream
.ctf_trace_id
);
282 session
->streams
[i
].first_read
= 1;
283 session
->streams
[i
].mmap_base
= mmap(NULL
, mmap_size
, PROT_READ
| PROT_WRITE
,
284 MAP_PRIVATE
| MAP_ANONYMOUS
, -1, 0);
285 if (session
->streams
[i
].mmap_base
== MAP_FAILED
) {
286 fprintf(stderr
, "mmap error\n");
291 if (be32toh(stream
.metadata_flag
)) {
292 session
->streams
[i
].metadata_flag
= 1;
295 ret
= session
->stream_count
;
302 int get_metadata(void)
304 struct lttng_viewer_cmd cmd
;
305 struct lttng_viewer_get_metadata rq
;
306 struct lttng_viewer_metadata_packet rp
;
311 int metadata_stream_id
= -1;
313 cmd
.cmd
= htobe32(VIEWER_GET_METADATA
);
314 cmd
.data_size
= sizeof(rq
);
317 for (i
= 0; i
< session
->stream_count
; i
++) {
318 if (session
->streams
[i
].metadata_flag
) {
319 metadata_stream_id
= i
;
324 if (metadata_stream_id
< 0) {
325 fprintf(stderr
, "No metadata stream found\n");
330 rq
.stream_id
= htobe64(session
->streams
[metadata_stream_id
].id
);
333 ret
= send(control_sock
, &cmd
, sizeof(cmd
), 0);
334 } while (ret
< 0 && errno
== EINTR
);
336 fprintf(stderr
, "Error sending cmd\n");
340 ret
= send(control_sock
, &rq
, sizeof(rq
), 0);
341 } while (ret
< 0 && errno
== EINTR
);
343 fprintf(stderr
, "Error sending get_metadata request\n");
347 ret
= recv(control_sock
, &rp
, sizeof(rp
), 0);
348 } while (ret
< 0 && errno
== EINTR
);
350 fprintf(stderr
, "Error receiving metadata response\n");
353 switch (be32toh(rp
.status
)) {
354 case VIEWER_METADATA_OK
:
356 case VIEWER_NO_NEW_METADATA
:
357 fprintf(stderr
, "NO NEW\n");
360 case VIEWER_METADATA_ERR
:
361 fprintf(stderr
, "ERR\n");
365 fprintf(stderr
, "UNKNOWN\n");
370 len
= be64toh(rp
.len
);
377 perror("relay data zmalloc");
381 ret
= recv(control_sock
, data
, len
, MSG_WAITALL
);
382 } while (ret
< 0 && errno
== EINTR
);
384 fprintf(stderr
, "Error receiving trace packet\n");
396 int get_next_index(void)
398 struct lttng_viewer_cmd cmd
;
399 struct lttng_viewer_get_next_index rq
;
400 struct lttng_viewer_index rp
;
404 cmd
.cmd
= htobe32(VIEWER_GET_NEXT_INDEX
);
405 cmd
.data_size
= sizeof(rq
);
408 for (id
= 0; id
< session
->stream_count
; id
++) {
409 if (session
->streams
[id
].metadata_flag
) {
412 rq
.stream_id
= htobe64(session
->streams
[id
].id
);
416 ret
= send(control_sock
, &cmd
, sizeof(cmd
), 0);
417 } while (ret
< 0 && errno
== EINTR
);
419 fprintf(stderr
, "Error sending cmd\n");
423 ret
= send(control_sock
, &rq
, sizeof(rq
), 0);
424 } while (ret
< 0 && errno
== EINTR
);
426 fprintf(stderr
, "Error sending get_next_index request\n");
430 ret
= recv(control_sock
, &rp
, sizeof(rp
), 0);
431 } while (ret
< 0 && errno
== EINTR
);
433 fprintf(stderr
, "Error receiving index response\n");
437 rp
.flags
= be32toh(rp
.flags
);
439 switch (be32toh(rp
.status
)) {
440 case VIEWER_INDEX_INACTIVE
:
441 fprintf(stderr
, "(INACTIVE)\n");
443 case VIEWER_INDEX_OK
:
445 case VIEWER_INDEX_RETRY
:
448 case VIEWER_INDEX_HUP
:
449 fprintf(stderr
, "(HUP)\n");
450 session
->streams
[id
].id
= -1ULL;
451 session
->streams
[id
].fd
= -1;
453 case VIEWER_INDEX_ERR
:
454 fprintf(stderr
, "(ERR)\n");
458 fprintf(stderr
, "SHOULD NOT HAPPEN\n");
462 if (!first_packet_stream_id
) {
463 first_packet_offset
= be64toh(rp
.offset
);
464 first_packet_len
= be64toh(rp
.packet_size
) / CHAR_BIT
;
465 first_packet_stream_id
= id
;
475 int get_data_packet(int id
, uint64_t offset
,
478 struct lttng_viewer_cmd cmd
;
479 struct lttng_viewer_get_packet rq
;
480 struct lttng_viewer_trace_packet rp
;
483 cmd
.cmd
= htobe32(VIEWER_GET_PACKET
);
484 cmd
.data_size
= sizeof(rq
);
487 rq
.stream_id
= htobe64(session
->streams
[id
].id
);
488 /* Already in big endian. */
490 rq
.len
= htobe32(len
);
493 ret
= send(control_sock
, &cmd
, sizeof(cmd
), 0);
494 } while (ret
< 0 && errno
== EINTR
);
496 fprintf(stderr
, "Error sending cmd\n");
500 ret
= send(control_sock
, &rq
, sizeof(rq
), 0);
501 } while (ret
< 0 && errno
== EINTR
);
503 fprintf(stderr
, "Error sending get_data_packet request\n");
507 ret
= recv(control_sock
, &rp
, sizeof(rp
), 0);
508 } while (ret
< 0 && errno
== EINTR
);
510 fprintf(stderr
, "Error receiving data response\n");
513 rp
.flags
= be32toh(rp
.flags
);
515 switch (be32toh(rp
.status
)) {
516 case VIEWER_GET_PACKET_OK
:
518 case VIEWER_GET_PACKET_RETRY
:
519 fprintf(stderr
, "RETRY\n");
522 case VIEWER_GET_PACKET_ERR
:
523 if (rp
.flags
& LTTNG_VIEWER_FLAG_NEW_METADATA
) {
524 fprintf(stderr
, "NEW_METADATA\n");
528 fprintf(stderr
, "ERR\n");
532 fprintf(stderr
, "UNKNOWN\n");
537 len
= be32toh(rp
.len
);
542 if (len
> mmap_size
) {
543 fprintf(stderr
, "mmap_size not big enough\n");
549 ret
= recv(control_sock
, session
->streams
[id
].mmap_base
, len
, MSG_WAITALL
);
550 } while (ret
< 0 && errno
== EINTR
);
552 fprintf(stderr
, "Error receiving trace packet\n");
562 int main(int argc
, char **argv
)
567 plan_tests(NUM_TESTS
);
569 diag("Live unit tests");
571 ret
= connect_viewer("localhost");
572 ok(ret
== 0, "Connect viewer to relayd");
574 ret
= establish_connection();
575 ok(ret
== 0, "Established connection and version check with %d.%d",
576 VERSION_MAJOR
, VERSION_MINOR
);
578 ret
= list_sessions(&session_id
);
579 ok(ret
> 0, "List sessions : %d session(s)", ret
);
581 ret
= attach_session(session_id
);
582 ok(ret
> 0, "Attach to session, %d streams received", ret
);
584 ret
= get_metadata();
585 ok(ret
> 0, "Get metadata, received %d bytes", ret
);
587 ret
= get_next_index();
588 ok(ret
== 0, "Get one index per stream");
590 ret
= get_data_packet(first_packet_stream_id
, first_packet_offset
,
592 ok(ret
== first_packet_len
,
593 "Get one data packet for stream %d, offset %d, len %d",
594 first_packet_stream_id
, first_packet_offset
,
597 return exit_status();