projects / babeltrace.git / blob
? search:


924e5fba24caf66fc0337b2d94bc251b04b9d4a8
[babeltrace.git] / src / plugins / ctf / lttng-live / viewer-connection.cpp
1 /*
2 * SPDX-License-Identifier: MIT
3 *
4 * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
5 * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
6 */
7
8 #include <glib.h>
9 #include <stdint.h>
10 #include <stdio.h>
11
12 #include <babeltrace2/babeltrace.h>
13
14 #include "common/common.h"
15 #include "compat/endian.h" /* IWYU pragma: keep */
16 #include "cpp-common/bt2s/make-unique.hpp"
17
18 #include "data-stream.hpp"
19 #include "lttng-live.hpp"
20 #include "lttng-viewer-abi.hpp"
21 #include "metadata.hpp"
22 #include "viewer-connection.hpp"
23
24 #define viewer_handle_send_recv_status(_status, _action, _msg_str) \
25 do { \
26 switch (_status) { \
27 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED: \
28 break; \
29 case LTTNG_LIVE_VIEWER_STATUS_ERROR: \
30 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, \
31 "Error " _action " " _msg_str); \
32 break; \
33 default: \
34 bt_common_abort(); \
35 } \
36 } while (0)
37
38 #define viewer_handle_send_status(_status, _msg_str) \
39 viewer_handle_send_recv_status(_status, "sending", _msg_str)
40
41 #define viewer_handle_recv_status(_status, _msg_str) \
42 viewer_handle_send_recv_status(_status, "receiving", _msg_str)
43
44 #define LTTNG_LIVE_CPPLOGE_APPEND_CAUSE_ERRNO(_msg, _fmt, ...) \
45 do { \
46 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, _msg ": {}" _fmt, \
47 bt_socket_errormsg(), ##__VA_ARGS__); \
48 } while (0)
49
50 static inline enum lttng_live_iterator_status
51 viewer_status_to_live_iterator_status(enum lttng_live_viewer_status viewer_status)
52 {
53 switch (viewer_status) {
54 case LTTNG_LIVE_VIEWER_STATUS_OK:
55 return LTTNG_LIVE_ITERATOR_STATUS_OK;
56 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
57 return LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
58 case LTTNG_LIVE_VIEWER_STATUS_ERROR:
59 return LTTNG_LIVE_ITERATOR_STATUS_ERROR;
60 }
61
62 bt_common_abort();
63 }
64
65 static inline enum ctf_msg_iter_medium_status
66 viewer_status_to_ctf_msg_iter_medium_status(enum lttng_live_viewer_status viewer_status)
67 {
68 switch (viewer_status) {
69 case LTTNG_LIVE_VIEWER_STATUS_OK:
70 return CTF_MSG_ITER_MEDIUM_STATUS_OK;
71 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
72 return CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
73 case LTTNG_LIVE_VIEWER_STATUS_ERROR:
74 return CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
75 }
76
77 bt_common_abort();
78 }
79
80 static inline void viewer_connection_close_socket(struct live_viewer_connection *viewer_connection)
81 {
82 int ret = bt_socket_close(viewer_connection->control_sock);
83 if (ret == -1) {
84 BT_CPPLOGW_ERRNO_SPEC(viewer_connection->logger,
85 "Error closing viewer connection socket: ", ".");
86 }
87
88 viewer_connection->control_sock = BT_INVALID_SOCKET;
89 }
90
91 /*
92 * This function receives a message from the Relay daemon.
93 * If it received the entire message, it returns _OK,
94 * If it's interrupted, it returns _INTERRUPTED,
95 * otherwise, it returns _ERROR.
96 */
97 static enum lttng_live_viewer_status
98 lttng_live_recv(struct live_viewer_connection *viewer_connection, void *buf, size_t len)
99 {
100 ssize_t received;
101 size_t total_received = 0, to_receive = len;
102 struct lttng_live_msg_iter *lttng_live_msg_iter = viewer_connection->lttng_live_msg_iter;
103 enum lttng_live_viewer_status status;
104 BT_SOCKET sock = viewer_connection->control_sock;
105
106 /*
107 * Receive a message from the Relay.
108 */
109 do {
110 received = bt_socket_recv(sock, (char *) buf + total_received, to_receive, 0);
111 if (received == BT_SOCKET_ERROR) {
112 if (bt_socket_interrupted()) {
113 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
114 /*
115 * This interruption was due to a
116 * SIGINT and the graph is being torn
117 * down.
118 */
119 status = LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED;
120 lttng_live_msg_iter->was_interrupted = true;
121 goto end;
122 } else {
123 /*
124 * A signal was received, but the graph
125 * is not being torn down. Carry on.
126 */
127 continue;
128 }
129 } else {
130 /*
131 * For any other types of socket error, close
132 * the socket and return an error.
133 */
134 LTTNG_LIVE_CPPLOGE_APPEND_CAUSE_ERRNO("Error receiving from Relay", ".");
135
136 viewer_connection_close_socket(viewer_connection);
137 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
138 goto end;
139 }
140 } else if (received == 0) {
141 /*
142 * The recv() call returned 0. This means the
143 * connection was orderly shutdown from the other peer.
144 * If that happens when we are trying to receive
145 * a message from it, it means something when wrong.
146 * Close the socket and return an error.
147 */
148 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
149 "Remote side has closed connection");
150 viewer_connection_close_socket(viewer_connection);
151 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
152 goto end;
153 }
154
155 BT_ASSERT(received <= to_receive);
156 total_received += received;
157 to_receive -= received;
158
159 } while (to_receive > 0);
160
161 BT_ASSERT(total_received == len);
162 status = LTTNG_LIVE_VIEWER_STATUS_OK;
163
164 end:
165 return status;
166 }
167
168 /*
169 * This function sends a message to the Relay daemon.
170 * If it send the message, it returns _OK,
171 * If it's interrupted, it returns _INTERRUPTED,
172 * otherwise, it returns _ERROR.
173 */
174 static enum lttng_live_viewer_status
175 lttng_live_send(struct live_viewer_connection *viewer_connection, const void *buf, size_t len)
176 {
177 enum lttng_live_viewer_status status;
178 struct lttng_live_msg_iter *lttng_live_msg_iter = viewer_connection->lttng_live_msg_iter;
179 BT_SOCKET sock = viewer_connection->control_sock;
180 size_t to_send = len;
181 ssize_t total_sent = 0;
182
183 do {
184 ssize_t sent = bt_socket_send_nosigpipe(sock, (char *) buf + total_sent, to_send);
185 if (sent == BT_SOCKET_ERROR) {
186 if (bt_socket_interrupted()) {
187 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
188 /*
189 * This interruption was a SIGINT and
190 * the graph is being teared down.
191 */
192 status = LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED;
193 lttng_live_msg_iter->was_interrupted = true;
194 goto end;
195 } else {
196 /*
197 * A signal was received, but the graph
198 * is not being teared down. Carry on.
199 */
200 continue;
201 }
202 } else {
203 /*
204 * For any other types of socket error, close
205 * the socket and return an error.
206 */
207 LTTNG_LIVE_CPPLOGE_APPEND_CAUSE_ERRNO("Error sending to Relay", ".");
208
209 viewer_connection_close_socket(viewer_connection);
210 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
211 goto end;
212 }
213 }
214
215 BT_ASSERT(sent <= to_send);
216 total_sent += sent;
217 to_send -= sent;
218
219 } while (to_send > 0);
220
221 BT_ASSERT(total_sent == len);
222 status = LTTNG_LIVE_VIEWER_STATUS_OK;
223
224 end:
225 return status;
226 }
227
228 static int parse_url(struct live_viewer_connection *viewer_connection)
229 {
230 char error_buf[256] = {0};
231 struct bt_common_lttng_live_url_parts lttng_live_url_parts = {};
232 bt_common_lttng_live_url_parts_deleter partsDeleter {lttng_live_url_parts};
233 int ret = -1;
234
235 if (viewer_connection->url.empty()) {
236 goto end;
237 }
238
239 lttng_live_url_parts = bt_common_parse_lttng_live_url(viewer_connection->url.c_str(), error_buf,
240 sizeof(error_buf));
241 if (!lttng_live_url_parts.proto) {
242 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Invalid LTTng live URL format: {}",
243 error_buf);
244 goto end;
245 }
246 viewer_connection->proto.reset(lttng_live_url_parts.proto);
247 lttng_live_url_parts.proto = NULL;
248
249 viewer_connection->relay_hostname.reset(lttng_live_url_parts.hostname);
250 lttng_live_url_parts.hostname = NULL;
251
252 if (lttng_live_url_parts.port >= 0) {
253 viewer_connection->port = lttng_live_url_parts.port;
254 } else {
255 viewer_connection->port = LTTNG_DEFAULT_NETWORK_VIEWER_PORT;
256 }
257
258 viewer_connection->target_hostname.reset(lttng_live_url_parts.target_hostname);
259 lttng_live_url_parts.target_hostname = NULL;
260
261 if (lttng_live_url_parts.session_name) {
262 viewer_connection->session_name.reset(lttng_live_url_parts.session_name);
263 lttng_live_url_parts.session_name = NULL;
264 }
265
266 ret = 0;
267
268 end:
269 return ret;
270 }
271
272 static enum lttng_live_viewer_status
273 lttng_live_handshake(struct live_viewer_connection *viewer_connection)
274 {
275 struct lttng_viewer_cmd cmd;
276 struct lttng_viewer_connect connect;
277 enum lttng_live_viewer_status status;
278 const size_t cmd_buf_len = sizeof(cmd) + sizeof(connect);
279 char cmd_buf[cmd_buf_len];
280
281 BT_CPPLOGD_SPEC(viewer_connection->logger,
282 "Handshaking with the relay daemon: cmd={}, major-version={}, minor-version={}",
283 LTTNG_VIEWER_CONNECT, LTTNG_LIVE_MAJOR, LTTNG_LIVE_MINOR);
284
285 cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
286 cmd.data_size = htobe64((uint64_t) sizeof(connect));
287 cmd.cmd_version = htobe32(0);
288
289 connect.viewer_session_id = -1ULL; /* will be set on recv */
290 connect.major = htobe32(LTTNG_LIVE_MAJOR);
291 connect.minor = htobe32(LTTNG_LIVE_MINOR);
292 connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
293
294 /*
295 * Merge the cmd and connection request to prevent a write-write
296 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
297 * second write to be performed quickly in presence of Nagle's algorithm
298 */
299 memcpy(cmd_buf, &cmd, sizeof(cmd));
300 memcpy(cmd_buf + sizeof(cmd), &connect, sizeof(connect));
301
302 status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
303 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
304 viewer_handle_send_status(status, "viewer connect command");
305 goto end;
306 }
307
308 status = lttng_live_recv(viewer_connection, &connect, sizeof(connect));
309 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
310 viewer_handle_recv_status(status, "viewer connect reply");
311 goto end;
312 }
313
314 BT_CPPLOGI_SPEC(viewer_connection->logger, "Received viewer session ID : {}",
315 (uint64_t) be64toh(connect.viewer_session_id));
316 BT_CPPLOGI_SPEC(viewer_connection->logger, "Relayd version : {}.{}", be32toh(connect.major),
317 be32toh(connect.minor));
318
319 if (LTTNG_LIVE_MAJOR != be32toh(connect.major)) {
320 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
321 "Incompatible lttng-relayd protocol");
322 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
323 goto end;
324 }
325 /* Use the smallest protocol version implemented. */
326 if (LTTNG_LIVE_MINOR > be32toh(connect.minor)) {
327 viewer_connection->minor = be32toh(connect.minor);
328 } else {
329 viewer_connection->minor = LTTNG_LIVE_MINOR;
330 }
331 viewer_connection->major = LTTNG_LIVE_MAJOR;
332
333 status = LTTNG_LIVE_VIEWER_STATUS_OK;
334
335 goto end;
336
337 end:
338 return status;
339 }
340
341 static enum lttng_live_viewer_status
342 lttng_live_connect_viewer(struct live_viewer_connection *viewer_connection)
343 {
344 struct hostent *host;
345 struct sockaddr_in server_addr;
346 enum lttng_live_viewer_status status;
347
348 if (parse_url(viewer_connection)) {
349 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Failed to parse URL");
350 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
351 goto error;
352 }
353
354 BT_CPPLOGD_SPEC(
355 viewer_connection->logger,
356 "Connecting to hostname : {}, port : {}, target hostname : {}, session name : {}, proto : {}",
357 viewer_connection->relay_hostname->str, viewer_connection->port,
358 !viewer_connection->target_hostname ? "<none>" : viewer_connection->target_hostname->str,
359 !viewer_connection->session_name ? "<none>" : viewer_connection->session_name->str,
360 viewer_connection->proto->str);
361
362 host = gethostbyname(viewer_connection->relay_hostname->str);
363 if (!host) {
364 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
365 "Cannot lookup hostname: hostname=\"{}\"",
366 viewer_connection->relay_hostname->str);
367 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
368 goto error;
369 }
370
371 if ((viewer_connection->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == BT_INVALID_SOCKET) {
372 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Socket creation failed: {}",
373 bt_socket_errormsg());
374 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
375 goto error;
376 }
377
378 server_addr.sin_family = AF_INET;
379 server_addr.sin_port = htons(viewer_connection->port);
380 server_addr.sin_addr = *((struct in_addr *) host->h_addr);
381 memset(&(server_addr.sin_zero), 0, 8);
382
383 if (connect(viewer_connection->control_sock, (struct sockaddr *) &server_addr,
384 sizeof(struct sockaddr)) == BT_SOCKET_ERROR) {
385 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Connection failed: {}",
386 bt_socket_errormsg());
387 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
388 goto error;
389 }
390
391 status = lttng_live_handshake(viewer_connection);
392
393 /*
394 * Only print error and append cause in case of error. not in case of
395 * interruption.
396 */
397 if (status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
398 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Viewer handshake failed");
399 goto error;
400 } else if (status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
401 goto end;
402 }
403
404 goto end;
405
406 error:
407 if (viewer_connection->control_sock != BT_INVALID_SOCKET) {
408 if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
409 BT_CPPLOGW_SPEC(viewer_connection->logger, "Error closing socket: {}.",
410 bt_socket_errormsg());
411 }
412 }
413 viewer_connection->control_sock = BT_INVALID_SOCKET;
414 end:
415 return status;
416 }
417
418 static void lttng_live_disconnect_viewer(struct live_viewer_connection *viewer_connection)
419 {
420 if (viewer_connection->control_sock == BT_INVALID_SOCKET) {
421 return;
422 }
423 if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
424 BT_CPPLOGW_SPEC(viewer_connection->logger, "Error closing socket: {}",
425 bt_socket_errormsg());
426 viewer_connection->control_sock = BT_INVALID_SOCKET;
427 }
428 }
429
430 static int list_update_session(bt_value *results, const struct lttng_viewer_session *session,
431 bool *_found, struct live_viewer_connection *viewer_connection)
432 {
433 int ret = 0;
434 uint64_t i, len;
435 bt_value *map = NULL;
436 bt_value *hostname = NULL;
437 bt_value *session_name = NULL;
438 bt_value *btval = NULL;
439 bool found = false;
440
441 len = bt_value_array_get_length(results);
442 for (i = 0; i < len; i++) {
443 const char *hostname_str = NULL;
444 const char *session_name_str = NULL;
445
446 map = bt_value_array_borrow_element_by_index(results, i);
447 hostname = bt_value_map_borrow_entry_value(map, "target-hostname");
448 if (!hostname) {
449 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
450 "Error borrowing \"target-hostname\" entry.");
451 ret = -1;
452 goto end;
453 }
454 session_name = bt_value_map_borrow_entry_value(map, "session-name");
455 if (!session_name) {
456 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
457 "Error borrowing \"session-name\" entry.");
458 ret = -1;
459 goto end;
460 }
461 hostname_str = bt_value_string_get(hostname);
462 session_name_str = bt_value_string_get(session_name);
463
464 if (strcmp(session->hostname, hostname_str) == 0 &&
465 strcmp(session->session_name, session_name_str) == 0) {
466 int64_t val;
467 uint32_t streams = be32toh(session->streams);
468 uint32_t clients = be32toh(session->clients);
469
470 found = true;
471
472 btval = bt_value_map_borrow_entry_value(map, "stream-count");
473 if (!btval) {
474 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
475 "Error borrowing \"stream-count\" entry.");
476 ret = -1;
477 goto end;
478 }
479 val = bt_value_integer_unsigned_get(btval);
480 /* sum */
481 val += streams;
482 bt_value_integer_unsigned_set(btval, val);
483
484 btval = bt_value_map_borrow_entry_value(map, "client-count");
485 if (!btval) {
486 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
487 "Error borrowing \"client-count\" entry.");
488 ret = -1;
489 goto end;
490 }
491 val = bt_value_integer_unsigned_get(btval);
492 /* max */
493 val = bt_max_t(int64_t, clients, val);
494 bt_value_integer_unsigned_set(btval, val);
495 }
496
497 if (found) {
498 break;
499 }
500 }
501 end:
502 *_found = found;
503 return ret;
504 }
505
506 static int list_append_session(bt_value *results, const std::string& base_url,
507 const struct lttng_viewer_session *session,
508 struct live_viewer_connection *viewer_connection)
509 {
510 int ret = 0;
511 bt_value_map_insert_entry_status insert_status;
512 bt_value_array_append_element_status append_status;
513 bt_value *map = NULL;
514 bool found = false;
515
516 /*
517 * If the session already exists, add the stream count to it,
518 * and do max of client counts.
519 */
520 ret = list_update_session(results, session, &found, viewer_connection);
521 if (ret || found) {
522 goto end;
523 }
524
525 map = bt_value_map_create();
526 if (!map) {
527 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error creating map value.");
528 ret = -1;
529 goto end;
530 }
531
532 if (base_url.empty()) {
533 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
534 "Error: base_url length smaller than 1.");
535 ret = -1;
536 goto end;
537 }
538
539 /*
540 * key = "url",
541 * value = <string>,
542 */
543 insert_status = bt_value_map_insert_string_entry(
544 map, "url",
545 fmt::format("{}/host/{}/{}", base_url, session->hostname, session->session_name).c_str());
546 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
547 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error inserting \"url\" entry.");
548 ret = -1;
549 goto end;
550 }
551
552 /*
553 * key = "target-hostname",
554 * value = <string>,
555 */
556 insert_status = bt_value_map_insert_string_entry(map, "target-hostname", session->hostname);
557 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
558 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
559 "Error inserting \"target-hostname\" entry.");
560 ret = -1;
561 goto end;
562 }
563
564 /*
565 * key = "session-name",
566 * value = <string>,
567 */
568 insert_status = bt_value_map_insert_string_entry(map, "session-name", session->session_name);
569 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
570 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
571 "Error inserting \"session-name\" entry.");
572 ret = -1;
573 goto end;
574 }
575
576 /*
577 * key = "timer-us",
578 * value = <integer>,
579 */
580 {
581 uint32_t live_timer = be32toh(session->live_timer);
582
583 insert_status = bt_value_map_insert_unsigned_integer_entry(map, "timer-us", live_timer);
584 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
585 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
586 "Error inserting \"timer-us\" entry.");
587 ret = -1;
588 goto end;
589 }
590 }
591
592 /*
593 * key = "stream-count",
594 * value = <integer>,
595 */
596 {
597 uint32_t streams = be32toh(session->streams);
598
599 insert_status = bt_value_map_insert_unsigned_integer_entry(map, "stream-count", streams);
600 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
601 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
602 "Error inserting \"stream-count\" entry.");
603 ret = -1;
604 goto end;
605 }
606 }
607
608 /*
609 * key = "client-count",
610 * value = <integer>,
611 */
612 {
613 uint32_t clients = be32toh(session->clients);
614
615 insert_status = bt_value_map_insert_unsigned_integer_entry(map, "client-count", clients);
616 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
617 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
618 "Error inserting \"client-count\" entry.");
619 ret = -1;
620 goto end;
621 }
622 }
623
624 append_status = bt_value_array_append_element(results, map);
625 if (append_status != BT_VALUE_ARRAY_APPEND_ELEMENT_STATUS_OK) {
626 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error appending map to results.");
627 ret = -1;
628 }
629
630 end:
631 BT_VALUE_PUT_REF_AND_RESET(map);
632 return ret;
633 }
634
635 /*
636 * Data structure returned:
637 *
638 * {
639 * <array> = {
640 * [n] = {
641 * <map> = {
642 * {
643 * key = "url",
644 * value = <string>,
645 * },
646 * {
647 * key = "target-hostname",
648 * value = <string>,
649 * },
650 * {
651 * key = "session-name",
652 * value = <string>,
653 * },
654 * {
655 * key = "timer-us",
656 * value = <integer>,
657 * },
658 * {
659 * key = "stream-count",
660 * value = <integer>,
661 * },
662 * {
663 * key = "client-count",
664 * value = <integer>,
665 * },
666 * },
667 * }
668 * }
669 */
670
671 bt_component_class_query_method_status
672 live_viewer_connection_list_sessions(struct live_viewer_connection *viewer_connection,
673 const bt_value **user_result)
674 {
675 bt_component_class_query_method_status status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
676 bt_value *result = NULL;
677 enum lttng_live_viewer_status viewer_status;
678 struct lttng_viewer_cmd cmd;
679 struct lttng_viewer_list_sessions list;
680 uint32_t i, sessions_count;
681
682 result = bt_value_array_create();
683 if (!result) {
684 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error creating array");
685 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
686 goto error;
687 }
688
689 BT_CPPLOGD_SPEC(viewer_connection->logger, "Requesting list of sessions: cmd={}",
690 LTTNG_VIEWER_LIST_SESSIONS);
691
692 cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
693 cmd.data_size = htobe64((uint64_t) 0);
694 cmd.cmd_version = htobe32(0);
695
696 viewer_status = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
697 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
698 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
699 "Error sending list sessions command");
700 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
701 goto error;
702 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
703 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
704 goto error;
705 }
706
707 viewer_status = lttng_live_recv(viewer_connection, &list, sizeof(list));
708 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
709 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error receiving session list");
710 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
711 goto error;
712 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
713 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
714 goto error;
715 }
716
717 sessions_count = be32toh(list.sessions_count);
718 for (i = 0; i < sessions_count; i++) {
719 struct lttng_viewer_session lsession;
720
721 viewer_status = lttng_live_recv(viewer_connection, &lsession, sizeof(lsession));
722 if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
723 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error receiving session:");
724 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
725 goto error;
726 } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
727 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
728 goto error;
729 }
730
731 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
732 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
733 if (list_append_session(result, viewer_connection->url, &lsession, viewer_connection)) {
734 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error appending session");
735 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
736 goto error;
737 }
738 }
739
740 *user_result = result;
741 goto end;
742 error:
743 BT_VALUE_PUT_REF_AND_RESET(result);
744 end:
745 return status;
746 }
747
748 static enum lttng_live_viewer_status
749 lttng_live_query_session_ids(struct lttng_live_msg_iter *lttng_live_msg_iter)
750 {
751 struct lttng_viewer_cmd cmd;
752 struct lttng_viewer_list_sessions list;
753 struct lttng_viewer_session lsession;
754 uint32_t i, sessions_count;
755 uint64_t session_id;
756 enum lttng_live_viewer_status status;
757 live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection.get();
758
759 BT_CPPLOGD_SPEC(viewer_connection->logger,
760 "Asking the relay daemon for the list of sessions: cmd={}",
761 LTTNG_VIEWER_LIST_SESSIONS);
762
763 cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
764 cmd.data_size = htobe64((uint64_t) 0);
765 cmd.cmd_version = htobe32(0);
766
767 status = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
768 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
769 viewer_handle_send_status(status, "list sessions command");
770 goto end;
771 }
772
773 status = lttng_live_recv(viewer_connection, &list, sizeof(list));
774 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
775 viewer_handle_recv_status(status, "session list reply");
776 goto end;
777 }
778
779 sessions_count = be32toh(list.sessions_count);
780 for (i = 0; i < sessions_count; i++) {
781 status = lttng_live_recv(viewer_connection, &lsession, sizeof(lsession));
782 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
783 viewer_handle_recv_status(status, "session reply");
784 goto end;
785 }
786 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
787 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
788 session_id = be64toh(lsession.id);
789
790 BT_CPPLOGI_SPEC(viewer_connection->logger,
791 "Adding session to internal list: "
792 "session-id={}, hostname=\"{}\", session-name=\"{}\"",
793 session_id, lsession.hostname, lsession.session_name);
794
795 if ((strncmp(lsession.session_name, viewer_connection->session_name->str,
796 LTTNG_VIEWER_NAME_MAX) == 0) &&
797 (strncmp(lsession.hostname, viewer_connection->target_hostname->str,
798 LTTNG_VIEWER_HOST_NAME_MAX) == 0)) {
799 if (lttng_live_add_session(lttng_live_msg_iter, session_id, lsession.hostname,
800 lsession.session_name)) {
801 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
802 "Failed to add live session");
803 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
804 goto end;
805 }
806 }
807 }
808
809 status = LTTNG_LIVE_VIEWER_STATUS_OK;
810
811 end:
812 return status;
813 }
814
815 enum lttng_live_viewer_status
816 lttng_live_create_viewer_session(struct lttng_live_msg_iter *lttng_live_msg_iter)
817 {
818 struct lttng_viewer_cmd cmd;
819 struct lttng_viewer_create_session_response resp;
820 enum lttng_live_viewer_status status;
821 live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection.get();
822
823 BT_CPPLOGD_SPEC(viewer_connection->logger, "Creating a viewer session: cmd={}",
824 LTTNG_VIEWER_CREATE_SESSION);
825
826 cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
827 cmd.data_size = htobe64((uint64_t) 0);
828 cmd.cmd_version = htobe32(0);
829
830 status = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
831 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
832 viewer_handle_send_status(status, "create session command");
833 goto end;
834 }
835
836 status = lttng_live_recv(viewer_connection, &resp, sizeof(resp));
837 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
838 viewer_handle_recv_status(status, "create session reply");
839 goto end;
840 }
841
842 if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
843 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error creating viewer session");
844 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
845 goto end;
846 }
847
848 status = lttng_live_query_session_ids(lttng_live_msg_iter);
849 if (status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
850 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
851 "Failed to query live viewer session ids");
852 goto end;
853 } else if (status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
854 goto end;
855 }
856
857 end:
858 return status;
859 }
860
861 static enum lttng_live_viewer_status receive_streams(struct lttng_live_session *session,
862 uint32_t stream_count,
863 bt_self_message_iterator *self_msg_iter)
864 {
865 uint32_t i;
866 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
867 enum lttng_live_viewer_status status;
868 live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection.get();
869
870 BT_CPPLOGI_SPEC(viewer_connection->logger, "Getting {} new streams", stream_count);
871 for (i = 0; i < stream_count; i++) {
872 struct lttng_viewer_stream stream;
873 struct lttng_live_stream_iterator *live_stream;
874 uint64_t stream_id;
875 uint64_t ctf_trace_id;
876
877 status = lttng_live_recv(viewer_connection, &stream, sizeof(stream));
878 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
879 viewer_handle_recv_status(status, "stream reply");
880 goto end;
881 }
882 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
883 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
884 stream_id = be64toh(stream.id);
885 ctf_trace_id = be64toh(stream.ctf_trace_id);
886
887 if (stream.metadata_flag) {
888 BT_CPPLOGI_SPEC(viewer_connection->logger, " metadata stream {} : {}/{}", stream_id,
889 stream.path_name, stream.channel_name);
890 if (lttng_live_metadata_create_stream(session, ctf_trace_id, stream_id)) {
891 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
892 "Error creating metadata stream");
893 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
894 goto end;
895 }
896 session->lazy_stream_msg_init = true;
897 } else {
898 BT_CPPLOGI_SPEC(viewer_connection->logger, " stream {} : {}/{}", stream_id,
899 stream.path_name, stream.channel_name);
900 live_stream =
901 lttng_live_stream_iterator_create(session, ctf_trace_id, stream_id, self_msg_iter);
902 if (!live_stream) {
903 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error creating stream");
904 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
905 goto end;
906 }
907 }
908 }
909 status = LTTNG_LIVE_VIEWER_STATUS_OK;
910
911 end:
912 return status;
913 }
914
915 enum lttng_live_viewer_status lttng_live_session_attach(struct lttng_live_session *session,
916 bt_self_message_iterator *self_msg_iter)
917 {
918 struct lttng_viewer_cmd cmd;
919 enum lttng_live_viewer_status status;
920 struct lttng_viewer_attach_session_request rq;
921 struct lttng_viewer_attach_session_response rp;
922 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
923 live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection.get();
924 uint64_t session_id = session->id;
925 uint32_t streams_count;
926 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
927 char cmd_buf[cmd_buf_len];
928
929 BT_CPPLOGD_SPEC(viewer_connection->logger,
930 "Attaching to session: cmd={}, session-id={}, seek={}",
931 LTTNG_VIEWER_ATTACH_SESSION, session_id, LTTNG_VIEWER_SEEK_LAST);
932
933 cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
934 cmd.data_size = htobe64((uint64_t) sizeof(rq));
935 cmd.cmd_version = htobe32(0);
936
937 memset(&rq, 0, sizeof(rq));
938 rq.session_id = htobe64(session_id);
939 // TODO: add cmd line parameter to select seek beginning
940 // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
941 rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
942
943 /*
944 * Merge the cmd and connection request to prevent a write-write
945 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
946 * second write to be performed quickly in presence of Nagle's algorithm.
947 */
948 memcpy(cmd_buf, &cmd, sizeof(cmd));
949 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
950 status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
951 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
952 viewer_handle_send_status(status, "attach session command");
953 goto end;
954 }
955
956 status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
957 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
958 viewer_handle_recv_status(status, "attach session reply");
959 goto end;
960 }
961
962 streams_count = be32toh(rp.streams_count);
963 switch (be32toh(rp.status)) {
964 case LTTNG_VIEWER_ATTACH_OK:
965 break;
966 case LTTNG_VIEWER_ATTACH_UNK:
967 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Session id {} is unknown",
968 session_id);
969 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
970 goto end;
971 case LTTNG_VIEWER_ATTACH_ALREADY:
972 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
973 "There is already a viewer attached to this session");
974 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
975 goto end;
976 case LTTNG_VIEWER_ATTACH_NOT_LIVE:
977 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Not a live session");
978 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
979 goto end;
980 case LTTNG_VIEWER_ATTACH_SEEK_ERR:
981 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Wrong seek parameter");
982 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
983 goto end;
984 default:
985 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Unknown attach return code {}",
986 be32toh(rp.status));
987 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
988 goto end;
989 }
990
991 /* We receive the initial list of streams. */
992 status = receive_streams(session, streams_count, self_msg_iter);
993 switch (status) {
994 case LTTNG_LIVE_VIEWER_STATUS_OK:
995 break;
996 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
997 goto end;
998 case LTTNG_LIVE_VIEWER_STATUS_ERROR:
999 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error receiving streams");
1000 goto end;
1001 default:
1002 bt_common_abort();
1003 }
1004
1005 session->attached = true;
1006 session->new_streams_needed = false;
1007
1008 end:
1009 return status;
1010 }
1011
1012 enum lttng_live_viewer_status lttng_live_session_detach(struct lttng_live_session *session)
1013 {
1014 struct lttng_viewer_cmd cmd;
1015 enum lttng_live_viewer_status status;
1016 struct lttng_viewer_detach_session_request rq;
1017 struct lttng_viewer_detach_session_response rp;
1018 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1019 live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection.get();
1020 uint64_t session_id = session->id;
1021 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1022 char cmd_buf[cmd_buf_len];
1023
1024 /*
1025 * The session might already be detached and the viewer socket might
1026 * already been closed. This happens when calling this function when
1027 * tearing down the graph after an error.
1028 */
1029 if (!session->attached || viewer_connection->control_sock == BT_INVALID_SOCKET) {
1030 return LTTNG_LIVE_VIEWER_STATUS_OK;
1031 }
1032
1033 BT_CPPLOGD_SPEC(viewer_connection->logger, "Detaching from session: cmd={}, session-id={}",
1034 LTTNG_VIEWER_DETACH_SESSION, session_id);
1035
1036 cmd.cmd = htobe32(LTTNG_VIEWER_DETACH_SESSION);
1037 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1038 cmd.cmd_version = htobe32(0);
1039
1040 memset(&rq, 0, sizeof(rq));
1041 rq.session_id = htobe64(session_id);
1042
1043 /*
1044 * Merge the cmd and connection request to prevent a write-write
1045 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1046 * second write to be performed quickly in presence of Nagle's algorithm.
1047 */
1048 memcpy(cmd_buf, &cmd, sizeof(cmd));
1049 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1050 status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1051 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1052 viewer_handle_send_status(status, "detach session command");
1053 goto end;
1054 }
1055
1056 status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1057 if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1058 viewer_handle_recv_status(status, "detach session reply");
1059 goto end;
1060 }
1061
1062 switch (be32toh(rp.status)) {
1063 case LTTNG_VIEWER_DETACH_SESSION_OK:
1064 break;
1065 case LTTNG_VIEWER_DETACH_SESSION_UNK:
1066 BT_CPPLOGW_SPEC(viewer_connection->logger, "Session id {} is unknown", session_id);
1067 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1068 goto end;
1069 case LTTNG_VIEWER_DETACH_SESSION_ERR:
1070 BT_CPPLOGW_SPEC(viewer_connection->logger, "Error detaching session id {}", session_id);
1071 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1072 goto end;
1073 default:
1074 BT_CPPLOGE_SPEC(viewer_connection->logger, "Unknown detach return code {}",
1075 be32toh(rp.status));
1076 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1077 goto end;
1078 }
1079
1080 session->attached = false;
1081
1082 status = LTTNG_LIVE_VIEWER_STATUS_OK;
1083
1084 end:
1085 return status;
1086 }
1087
1088 enum lttng_live_get_one_metadata_status
1089 lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, std::vector<char>& buf)
1090 {
1091 uint64_t len = 0;
1092 enum lttng_live_get_one_metadata_status status;
1093 enum lttng_live_viewer_status viewer_status;
1094 struct lttng_viewer_cmd cmd;
1095 struct lttng_viewer_get_metadata rq;
1096 struct lttng_viewer_metadata_packet rp;
1097 std::vector<char> data;
1098 struct lttng_live_session *session = trace->session;
1099 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1100 struct lttng_live_metadata *metadata = trace->metadata.get();
1101 live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection.get();
1102 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1103 char cmd_buf[cmd_buf_len];
1104
1105 BT_CPPLOGD_SPEC(viewer_connection->logger,
1106 "Requesting new metadata for trace:"
1107 "cmd={}, trace-id={}, metadata-stream-id={}",
1108 LTTNG_VIEWER_GET_METADATA, trace->id, metadata->stream_id);
1109
1110 rq.stream_id = htobe64(metadata->stream_id);
1111 cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
1112 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1113 cmd.cmd_version = htobe32(0);
1114
1115 /*
1116 * Merge the cmd and connection request to prevent a write-write
1117 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1118 * second write to be performed quickly in presence of Nagle's algorithm.
1119 */
1120 memcpy(cmd_buf, &cmd, sizeof(cmd));
1121 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1122 viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1123 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1124 viewer_handle_send_status(viewer_status, "get metadata command");
1125 status = (enum lttng_live_get_one_metadata_status) viewer_status;
1126 goto end;
1127 }
1128
1129 viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1130 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1131 viewer_handle_recv_status(viewer_status, "get metadata reply");
1132 status = (enum lttng_live_get_one_metadata_status) viewer_status;
1133 goto end;
1134 }
1135
1136 switch (be32toh(rp.status)) {
1137 case LTTNG_VIEWER_METADATA_OK:
1138 BT_CPPLOGD_SPEC(viewer_connection->logger, "Received get_metadata response: ok");
1139 break;
1140 case LTTNG_VIEWER_NO_NEW_METADATA:
1141 BT_CPPLOGD_SPEC(viewer_connection->logger, "Received get_metadata response: no new");
1142 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_END;
1143 goto end;
1144 case LTTNG_VIEWER_METADATA_ERR:
1145 /*
1146 * The Relayd cannot find this stream id. Maybe its
1147 * gone already. This can happen in short lived UST app
1148 * in a per-pid session.
1149 */
1150 BT_CPPLOGD_SPEC(viewer_connection->logger, "Received get_metadata response: error");
1151 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_CLOSED;
1152 goto end;
1153 default:
1154 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1155 "Received get_metadata response: unknown");
1156 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1157 goto end;
1158 }
1159
1160 len = be64toh(rp.len);
1161 if (len == 0) {
1162 /*
1163 * We received a `LTTNG_VIEWER_METADATA_OK` with a packet
1164 * length of 0. This means we must try again. This scenario
1165 * arises when a clear command is performed on an lttng session.
1166 */
1167 BT_CPPLOGD_SPEC(
1168 viewer_connection->logger,
1169 "Expecting a metadata packet of size 0. Retry to get a packet from the relay.");
1170 goto empty_metadata_packet_retry;
1171 }
1172
1173 BT_CPPLOGD_SPEC(viewer_connection->logger, "Writing {} bytes to metadata", len);
1174 if (len <= 0) {
1175 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Erroneous response length");
1176 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1177 goto end;
1178 }
1179
1180 data.resize(len);
1181
1182 viewer_status = lttng_live_recv(viewer_connection, data.data(), len);
1183 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1184 viewer_handle_recv_status(viewer_status, "get metadata packet");
1185 status = (enum lttng_live_get_one_metadata_status) viewer_status;
1186 goto end;
1187 }
1188
1189 /*
1190 * Write the metadata to the file handle.
1191 */
1192 buf.insert(buf.end(), data.begin(), data.end());
1193
1194 empty_metadata_packet_retry:
1195 status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_OK;
1196
1197 end:
1198 return status;
1199 }
1200
1201 /*
1202 * Assign the fields from a lttng_viewer_index to a packet_index.
1203 */
1204 static void lttng_index_to_packet_index(struct lttng_viewer_index *lindex,
1205 struct packet_index *pindex)
1206 {
1207 BT_ASSERT(lindex);
1208 BT_ASSERT(pindex);
1209
1210 pindex->offset = be64toh(lindex->offset);
1211 pindex->packet_size = be64toh(lindex->packet_size);
1212 pindex->content_size = be64toh(lindex->content_size);
1213 pindex->ts_cycles.timestamp_begin = be64toh(lindex->timestamp_begin);
1214 pindex->ts_cycles.timestamp_end = be64toh(lindex->timestamp_end);
1215 pindex->events_discarded = be64toh(lindex->events_discarded);
1216 }
1217
1218 static void lttng_live_need_new_streams(struct lttng_live_msg_iter *lttng_live_msg_iter)
1219 {
1220 for (const auto& session : lttng_live_msg_iter->sessions) {
1221 BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1222 "Marking session as needing new streams: "
1223 "session-id={}",
1224 session->id);
1225 session->new_streams_needed = true;
1226 }
1227 }
1228
1229 enum lttng_live_iterator_status
1230 lttng_live_get_next_index(struct lttng_live_msg_iter *lttng_live_msg_iter,
1231 struct lttng_live_stream_iterator *stream, struct packet_index *index)
1232 {
1233 struct lttng_viewer_cmd cmd;
1234 struct lttng_viewer_get_next_index rq;
1235 enum lttng_live_viewer_status viewer_status;
1236 struct lttng_viewer_index rp;
1237 enum lttng_live_iterator_status status;
1238 live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection.get();
1239 struct lttng_live_trace *trace = stream->trace;
1240 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1241 char cmd_buf[cmd_buf_len];
1242 uint32_t flags, rp_status;
1243
1244 BT_CPPLOGD_SPEC(viewer_connection->logger,
1245 "Requesting next index for stream: cmd={}, "
1246 "viewer-stream-id={}",
1247 LTTNG_VIEWER_GET_NEXT_INDEX, stream->viewer_stream_id);
1248 cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
1249 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1250 cmd.cmd_version = htobe32(0);
1251
1252 memset(&rq, 0, sizeof(rq));
1253 rq.stream_id = htobe64(stream->viewer_stream_id);
1254
1255 /*
1256 * Merge the cmd and connection request to prevent a write-write
1257 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1258 * second write to be performed quickly in presence of Nagle's algorithm.
1259 */
1260 memcpy(cmd_buf, &cmd, sizeof(cmd));
1261 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1262
1263 viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1264 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1265 viewer_handle_send_status(viewer_status, "get next index command");
1266 goto error;
1267 }
1268
1269 viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1270 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1271 viewer_handle_recv_status(viewer_status, "get next index reply");
1272 goto error;
1273 }
1274
1275 flags = be32toh(rp.flags);
1276 rp_status = be32toh(rp.status);
1277
1278 BT_CPPLOGD_SPEC(
1279 viewer_connection->logger, "Received response from relay daemon: cmd=%s, response={}",
1280 LTTNG_VIEWER_GET_NEXT_INDEX, static_cast<lttng_viewer_next_index_return_code>(rp_status));
1281
1282 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1283 BT_CPPLOGD_SPEC(viewer_connection->logger,
1284 "Marking all sessions as possibly needing new streams: "
1285 "response={}, response-flag=NEW_STREAM",
1286 static_cast<lttng_viewer_next_index_return_code>(rp_status));
1287 lttng_live_need_new_streams(lttng_live_msg_iter);
1288 }
1289
1290 switch (rp_status) {
1291 case LTTNG_VIEWER_INDEX_INACTIVE:
1292 {
1293 uint64_t ctf_stream_class_id;
1294
1295 memset(index, 0, sizeof(struct packet_index));
1296 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
1297 stream->current_inactivity_ts = index->ts_cycles.timestamp_end;
1298 ctf_stream_class_id = be64toh(rp.stream_id);
1299 if (stream->ctf_stream_class_id.is_set) {
1300 BT_ASSERT(stream->ctf_stream_class_id.value == ctf_stream_class_id);
1301 } else {
1302 stream->ctf_stream_class_id.value = ctf_stream_class_id;
1303 stream->ctf_stream_class_id.is_set = true;
1304 }
1305 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_QUIESCENT);
1306 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1307 break;
1308 }
1309 case LTTNG_VIEWER_INDEX_OK:
1310 {
1311 uint64_t ctf_stream_class_id;
1312
1313 lttng_index_to_packet_index(&rp, index);
1314 ctf_stream_class_id = be64toh(rp.stream_id);
1315 if (stream->ctf_stream_class_id.is_set) {
1316 BT_ASSERT(stream->ctf_stream_class_id.value == ctf_stream_class_id);
1317 } else {
1318 stream->ctf_stream_class_id.value = ctf_stream_class_id;
1319 stream->ctf_stream_class_id.is_set = true;
1320 }
1321 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_DATA);
1322
1323 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1324 BT_CPPLOGD_SPEC(viewer_connection->logger,
1325 "Marking trace as needing new metadata: "
1326 "response={}, response-flag=NEW_METADATA, trace-id={}",
1327 static_cast<lttng_viewer_next_index_return_code>(rp_status), trace->id);
1328 trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
1329 }
1330 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1331 break;
1332 }
1333 case LTTNG_VIEWER_INDEX_RETRY:
1334 memset(index, 0, sizeof(struct packet_index));
1335 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
1336 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1337 goto end;
1338 case LTTNG_VIEWER_INDEX_HUP:
1339 memset(index, 0, sizeof(struct packet_index));
1340 index->offset = EOF;
1341 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_EOF);
1342 stream->has_stream_hung_up = true;
1343 status = LTTNG_LIVE_ITERATOR_STATUS_END;
1344 break;
1345 case LTTNG_VIEWER_INDEX_ERR:
1346 memset(index, 0, sizeof(struct packet_index));
1347 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
1348 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1349 goto end;
1350 default:
1351 BT_CPPLOGD_SPEC(viewer_connection->logger,
1352 "Received get_next_index response: unknown value");
1353 memset(index, 0, sizeof(struct packet_index));
1354 lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
1355 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1356 goto end;
1357 }
1358
1359 goto end;
1360
1361 error:
1362 status = viewer_status_to_live_iterator_status(viewer_status);
1363 end:
1364 return status;
1365 }
1366
1367 enum ctf_msg_iter_medium_status
1368 lttng_live_get_stream_bytes(struct lttng_live_msg_iter *lttng_live_msg_iter,
1369 struct lttng_live_stream_iterator *stream, uint8_t *buf,
1370 uint64_t offset, uint64_t req_len, uint64_t *recv_len)
1371 {
1372 enum ctf_msg_iter_medium_status status;
1373 enum lttng_live_viewer_status viewer_status;
1374 struct lttng_viewer_trace_packet rp;
1375 struct lttng_viewer_cmd cmd;
1376 struct lttng_viewer_get_packet rq;
1377 live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection.get();
1378 struct lttng_live_trace *trace = stream->trace;
1379 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1380 char cmd_buf[cmd_buf_len];
1381 uint32_t flags, rp_status;
1382
1383 BT_CPPLOGD_SPEC(viewer_connection->logger,
1384 "Requesting data from stream: cmd={}, "
1385 "offset={}, request-len={}",
1386 LTTNG_VIEWER_GET_PACKET, offset, req_len);
1387
1388 cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
1389 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1390 cmd.cmd_version = htobe32(0);
1391
1392 memset(&rq, 0, sizeof(rq));
1393 rq.stream_id = htobe64(stream->viewer_stream_id);
1394 rq.offset = htobe64(offset);
1395 rq.len = htobe32(req_len);
1396
1397 /*
1398 * Merge the cmd and connection request to prevent a write-write
1399 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1400 * second write to be performed quickly in presence of Nagle's algorithm.
1401 */
1402 memcpy(cmd_buf, &cmd, sizeof(cmd));
1403 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1404
1405 viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1406 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1407 viewer_handle_send_status(viewer_status, "get data packet command");
1408 goto error_convert_status;
1409 }
1410
1411 viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1412 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1413 viewer_handle_recv_status(viewer_status, "get data packet reply");
1414 goto error_convert_status;
1415 }
1416
1417 flags = be32toh(rp.flags);
1418 rp_status = be32toh(rp.status);
1419
1420 BT_CPPLOGD_SPEC(
1421 viewer_connection->logger, "Received response from relay daemon: cmd={}, response={}",
1422 LTTNG_VIEWER_GET_PACKET, static_cast<lttng_viewer_get_packet_return_code>(rp_status));
1423 switch (rp_status) {
1424 case LTTNG_VIEWER_GET_PACKET_OK:
1425 req_len = be32toh(rp.len);
1426 BT_CPPLOGD_SPEC(viewer_connection->logger,
1427 "Got packet from relay daemon: response={}, packet-len={}",
1428 static_cast<lttng_viewer_get_packet_return_code>(rp_status), req_len);
1429 break;
1430 case LTTNG_VIEWER_GET_PACKET_RETRY:
1431 /* Unimplemented by relay daemon */
1432 status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
1433 goto end;
1434 case LTTNG_VIEWER_GET_PACKET_ERR:
1435 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1436 BT_CPPLOGD_SPEC(viewer_connection->logger,
1437 "Marking trace as needing new metadata: "
1438 "response={}, response-flag=NEW_METADATA, trace-id={}",
1439 static_cast<lttng_viewer_get_packet_return_code>(rp_status), trace->id);
1440 trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
1441 }
1442 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1443 BT_CPPLOGD_SPEC(viewer_connection->logger,
1444 "Marking all sessions as possibly needing new streams: "
1445 "response={}, response-flag=NEW_STREAM",
1446 static_cast<lttng_viewer_get_packet_return_code>(rp_status));
1447 lttng_live_need_new_streams(lttng_live_msg_iter);
1448 }
1449 if (flags & (LTTNG_VIEWER_FLAG_NEW_METADATA | LTTNG_VIEWER_FLAG_NEW_STREAM)) {
1450 status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
1451 BT_CPPLOGD_SPEC(viewer_connection->logger,
1452 "Reply with any one flags set means we should retry: response={}",
1453 static_cast<lttng_viewer_get_packet_return_code>(rp_status));
1454 goto end;
1455 }
1456 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1457 "Received get_data_packet response: error");
1458 status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
1459 goto end;
1460 case LTTNG_VIEWER_GET_PACKET_EOF:
1461 status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
1462 goto end;
1463 default:
1464 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1465 "Received get_data_packet response: unknown ({})", rp_status);
1466 status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
1467 goto end;
1468 }
1469
1470 if (req_len == 0) {
1471 status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
1472 goto end;
1473 }
1474
1475 viewer_status = lttng_live_recv(viewer_connection, buf, req_len);
1476 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1477 viewer_handle_recv_status(viewer_status, "get data packet");
1478 goto error_convert_status;
1479 }
1480 *recv_len = req_len;
1481
1482 status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
1483 goto end;
1484
1485 error_convert_status:
1486 status = viewer_status_to_ctf_msg_iter_medium_status(viewer_status);
1487 end:
1488 return status;
1489 }
1490
1491 /*
1492 * Request new streams for a session.
1493 */
1494 enum lttng_live_iterator_status
1495 lttng_live_session_get_new_streams(struct lttng_live_session *session,
1496 bt_self_message_iterator *self_msg_iter)
1497 {
1498 enum lttng_live_iterator_status status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1499 struct lttng_viewer_cmd cmd;
1500 struct lttng_viewer_new_streams_request rq;
1501 struct lttng_viewer_new_streams_response rp;
1502 struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1503 enum lttng_live_viewer_status viewer_status;
1504 live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection.get();
1505 uint32_t streams_count;
1506 const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1507 char cmd_buf[cmd_buf_len];
1508
1509 if (!session->new_streams_needed) {
1510 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1511 goto end;
1512 }
1513
1514 BT_CPPLOGD_SPEC(viewer_connection->logger,
1515 "Requesting new streams for session: cmd={}, session-id={}",
1516 LTTNG_VIEWER_GET_NEW_STREAMS, session->id);
1517
1518 cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1519 cmd.data_size = htobe64((uint64_t) sizeof(rq));
1520 cmd.cmd_version = htobe32(0);
1521
1522 memset(&rq, 0, sizeof(rq));
1523 rq.session_id = htobe64(session->id);
1524
1525 /*
1526 * Merge the cmd and connection request to prevent a write-write
1527 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1528 * second write to be performed quickly in presence of Nagle's algorithm.
1529 */
1530 memcpy(cmd_buf, &cmd, sizeof(cmd));
1531 memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1532
1533 viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1534 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1535 viewer_handle_send_status(viewer_status, "get new streams command");
1536 status = viewer_status_to_live_iterator_status(viewer_status);
1537 goto end;
1538 }
1539
1540 viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1541 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1542 viewer_handle_recv_status(viewer_status, "get new streams reply");
1543 status = viewer_status_to_live_iterator_status(viewer_status);
1544 goto end;
1545 }
1546
1547 streams_count = be32toh(rp.streams_count);
1548
1549 switch (be32toh(rp.status)) {
1550 case LTTNG_VIEWER_NEW_STREAMS_OK:
1551 session->new_streams_needed = false;
1552 break;
1553 case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1554 session->new_streams_needed = false;
1555 goto end;
1556 case LTTNG_VIEWER_NEW_STREAMS_HUP:
1557 session->new_streams_needed = false;
1558 session->closed = true;
1559 status = LTTNG_LIVE_ITERATOR_STATUS_END;
1560 goto end;
1561 case LTTNG_VIEWER_NEW_STREAMS_ERR:
1562 BT_CPPLOGD_SPEC(viewer_connection->logger, "Received get_new_streams response: error");
1563 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1564 goto end;
1565 default:
1566 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1567 "Received get_new_streams response: Unknown:"
1568 "return code {}",
1569 be32toh(rp.status));
1570 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1571 goto end;
1572 }
1573
1574 viewer_status = receive_streams(session, streams_count, self_msg_iter);
1575 if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1576 viewer_handle_recv_status(viewer_status, "new streams");
1577 status = viewer_status_to_live_iterator_status(viewer_status);
1578 goto end;
1579 }
1580
1581 status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1582 end:
1583 return status;
1584 }
1585
1586 enum lttng_live_viewer_status
1587 live_viewer_connection_create(const char *url, bool in_query,
1588 struct lttng_live_msg_iter *lttng_live_msg_iter,
1589 const bt2c::Logger& parentLogger, live_viewer_connection::UP& viewer)
1590 {
1591 auto viewer_connection = bt2s::make_unique<live_viewer_connection>(parentLogger);
1592
1593 if (bt_socket_init(viewer_connection->logger) != 0) {
1594 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Failed to init socket");
1595 return LTTNG_LIVE_VIEWER_STATUS_ERROR;
1596 }
1597
1598 viewer_connection->control_sock = BT_INVALID_SOCKET;
1599 viewer_connection->port = -1;
1600 viewer_connection->in_query = in_query;
1601 viewer_connection->lttng_live_msg_iter = lttng_live_msg_iter;
1602 viewer_connection->url = url;
1603
1604 BT_CPPLOGD_SPEC(viewer_connection->logger, "Establishing connection to url \"{}\"...", url);
1605 const auto status = lttng_live_connect_viewer(viewer_connection.get());
1606
1607 /*
1608 * Only print error and append cause in case of error. not in case of
1609 * interruption.
1610 */
1611 if (status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1612 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1613 "Failed to establish connection: "
1614 "url=\"{}\"",
1615 url);
1616 return status;
1617 } else if (status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1618 return status;
1619 }
1620 BT_CPPLOGD_SPEC(viewer_connection->logger, "Connection to url \"{}\" is established", url);
1621
1622 viewer = std::move(viewer_connection);
1623 return LTTNG_LIVE_VIEWER_STATUS_OK;
1624 }
1625
1626 live_viewer_connection::~live_viewer_connection()
1627 {
1628 BT_CPPLOGD_SPEC(this->logger, "Closing connection to relay: relay-url=\"{}\"", this->url);
1629
1630 lttng_live_disconnect_viewer(this);
1631
1632 bt_socket_fini();
1633 }
BabelTrace - Trace Format Babel Tower
This page took 0.062355 seconds and 3 git commands to generate.