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