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