2 * SPDX-License-Identifier: MIT
4 * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
5 * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
8 #define BT_COMP_LOG_SELF_COMP (viewer_connection->self_comp)
9 #define BT_LOG_OUTPUT_LEVEL (viewer_connection->log_level)
10 #define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE/VIEWER"
11 #include "logging/comp-logging.h"
18 #include <sys/types.h>
23 #include "compat/socket.h"
24 #include "compat/endian.h"
25 #include "compat/compiler.h"
26 #include "common/common.h"
27 #include <babeltrace2/babeltrace.h>
29 #include "lttng-live.h"
30 #include "viewer-connection.h"
31 #include "lttng-viewer-abi.h"
32 #include "data-stream.h"
35 #define viewer_handle_send_recv_status(_self_comp, _self_comp_class, \
36 _status, _action, _msg_str) \
39 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED: \
41 case LTTNG_LIVE_VIEWER_STATUS_ERROR: \
42 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(_self_comp, \
43 _self_comp_class, "Error " _action " " _msg_str); \
50 #define viewer_handle_send_status(_self_comp, _self_comp_class, _status, _msg_str) \
51 viewer_handle_send_recv_status(_self_comp, _self_comp_class, _status, \
54 #define viewer_handle_recv_status(_self_comp, _self_comp_class, _status, _msg_str) \
55 viewer_handle_send_recv_status(_self_comp, _self_comp_class, _status, \
56 "receiving", _msg_str)
58 #define LTTNG_LIVE_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE_ERRNO(_self_comp, \
59 _self_comp_class, _msg, _fmt, ...) \
61 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(_self_comp, _self_comp_class, \
62 _msg ": %s" _fmt, bt_socket_errormsg(), ##__VA_ARGS__); \
66 enum lttng_live_iterator_status
viewer_status_to_live_iterator_status(
67 enum lttng_live_viewer_status viewer_status
)
69 switch (viewer_status
) {
70 case LTTNG_LIVE_VIEWER_STATUS_OK
:
71 return LTTNG_LIVE_ITERATOR_STATUS_OK
;
72 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
:
73 return LTTNG_LIVE_ITERATOR_STATUS_AGAIN
;
74 case LTTNG_LIVE_VIEWER_STATUS_ERROR
:
75 return LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
82 enum ctf_msg_iter_medium_status
viewer_status_to_ctf_msg_iter_medium_status(
83 enum lttng_live_viewer_status viewer_status
)
85 switch (viewer_status
) {
86 case LTTNG_LIVE_VIEWER_STATUS_OK
:
87 return CTF_MSG_ITER_MEDIUM_STATUS_OK
;
88 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
:
89 return CTF_MSG_ITER_MEDIUM_STATUS_AGAIN
;
90 case LTTNG_LIVE_VIEWER_STATUS_ERROR
:
91 return CTF_MSG_ITER_MEDIUM_STATUS_ERROR
;
98 void viewer_connection_close_socket(
99 struct live_viewer_connection
*viewer_connection
)
101 bt_self_component_class
*self_comp_class
=
102 viewer_connection
->self_comp_class
;
103 bt_self_component
*self_comp
=
104 viewer_connection
->self_comp
;
105 int ret
= bt_socket_close(viewer_connection
->control_sock
);
107 BT_COMP_OR_COMP_CLASS_LOGW_ERRNO(
108 self_comp
, self_comp_class
,
109 "Error closing viewer connection socket: ", ".");
112 viewer_connection
->control_sock
= BT_INVALID_SOCKET
;
116 * This function receives a message from the Relay daemon.
117 * If it received the entire message, it returns _OK,
118 * If it's interrupted, it returns _INTERRUPTED,
119 * otherwise, it returns _ERROR.
122 enum lttng_live_viewer_status
lttng_live_recv(
123 struct live_viewer_connection
*viewer_connection
,
124 void *buf
, size_t len
)
127 bt_self_component_class
*self_comp_class
=
128 viewer_connection
->self_comp_class
;
129 bt_self_component
*self_comp
=
130 viewer_connection
->self_comp
;
131 size_t total_received
= 0, to_receive
= len
;
132 struct lttng_live_msg_iter
*lttng_live_msg_iter
=
133 viewer_connection
->lttng_live_msg_iter
;
134 enum lttng_live_viewer_status status
;
135 BT_SOCKET sock
= viewer_connection
->control_sock
;
138 * Receive a message from the Relay.
141 received
= bt_socket_recv(sock
, buf
+ total_received
, to_receive
, 0);
142 if (received
== BT_SOCKET_ERROR
) {
143 if (bt_socket_interrupted()) {
144 if (lttng_live_graph_is_canceled(lttng_live_msg_iter
)) {
146 * This interruption was due to a
147 * SIGINT and the graph is being torn
150 status
= LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
;
151 lttng_live_msg_iter
->was_interrupted
= true;
155 * A signal was received, but the graph
156 * is not being torn down. Carry on.
162 * For any other types of socket error, close
163 * the socket and return an error.
165 LTTNG_LIVE_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE_ERRNO(
166 self_comp
, self_comp_class
,
167 "Error receiving from Relay", ".");
169 viewer_connection_close_socket(viewer_connection
);
170 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
173 } else if (received
== 0) {
175 * The recv() call returned 0. This means the
176 * connection was orderly shutdown from the other peer.
177 * If that happens when we are trying to receive
178 * a message from it, it means something when wrong.
179 * Close the socket and return an error.
181 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
182 self_comp_class
, "Remote side has closed connection");
183 viewer_connection_close_socket(viewer_connection
);
184 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
188 BT_ASSERT(received
<= to_receive
);
189 total_received
+= received
;
190 to_receive
-= received
;
192 } while (to_receive
> 0);
194 BT_ASSERT(total_received
== len
);
195 status
= LTTNG_LIVE_VIEWER_STATUS_OK
;
202 * This function sends a message to the Relay daemon.
203 * If it send the message, it returns _OK,
204 * If it's interrupted, it returns _INTERRUPTED,
205 * otherwise, it returns _ERROR.
208 enum lttng_live_viewer_status
lttng_live_send(
209 struct live_viewer_connection
*viewer_connection
,
210 const void *buf
, size_t len
)
212 enum lttng_live_viewer_status status
;
213 bt_self_component_class
*self_comp_class
=
214 viewer_connection
->self_comp_class
;
215 bt_self_component
*self_comp
=
216 viewer_connection
->self_comp
;
217 struct lttng_live_msg_iter
*lttng_live_msg_iter
=
218 viewer_connection
->lttng_live_msg_iter
;
219 BT_SOCKET sock
= viewer_connection
->control_sock
;
220 size_t to_send
= len
;
221 ssize_t total_sent
= 0;
224 ssize_t sent
= bt_socket_send_nosigpipe(sock
, buf
+ total_sent
,
226 if (sent
== BT_SOCKET_ERROR
) {
227 if (bt_socket_interrupted()) {
228 if (lttng_live_graph_is_canceled(lttng_live_msg_iter
)) {
230 * This interruption was a SIGINT and
231 * the graph is being teared down.
233 status
= LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
;
234 lttng_live_msg_iter
->was_interrupted
= true;
238 * A signal was received, but the graph
239 * is not being teared down. Carry on.
245 * For any other types of socket error, close
246 * the socket and return an error.
248 LTTNG_LIVE_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE_ERRNO(
249 self_comp
, self_comp_class
,
250 "Error sending to Relay", ".");
252 viewer_connection_close_socket(viewer_connection
);
253 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
258 BT_ASSERT(sent
<= to_send
);
262 } while (to_send
> 0);
264 BT_ASSERT(total_sent
== len
);
265 status
= LTTNG_LIVE_VIEWER_STATUS_OK
;
272 int parse_url(struct live_viewer_connection
*viewer_connection
)
274 char error_buf
[256] = { 0 };
275 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
276 bt_self_component_class
*self_comp_class
= viewer_connection
->self_comp_class
;
277 struct bt_common_lttng_live_url_parts lttng_live_url_parts
= { 0 };
279 const char *path
= viewer_connection
->url
->str
;
285 lttng_live_url_parts
= bt_common_parse_lttng_live_url(path
, error_buf
,
287 if (!lttng_live_url_parts
.proto
) {
288 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
289 self_comp_class
,"Invalid LTTng live URL format: %s",
293 viewer_connection
->proto
= lttng_live_url_parts
.proto
;
294 lttng_live_url_parts
.proto
= NULL
;
296 viewer_connection
->relay_hostname
= lttng_live_url_parts
.hostname
;
297 lttng_live_url_parts
.hostname
= NULL
;
299 if (lttng_live_url_parts
.port
>= 0) {
300 viewer_connection
->port
= lttng_live_url_parts
.port
;
302 viewer_connection
->port
= LTTNG_DEFAULT_NETWORK_VIEWER_PORT
;
305 viewer_connection
->target_hostname
= lttng_live_url_parts
.target_hostname
;
306 lttng_live_url_parts
.target_hostname
= NULL
;
308 if (lttng_live_url_parts
.session_name
) {
309 viewer_connection
->session_name
= lttng_live_url_parts
.session_name
;
310 lttng_live_url_parts
.session_name
= NULL
;
316 bt_common_destroy_lttng_live_url_parts(<tng_live_url_parts
);
321 enum lttng_live_viewer_status
lttng_live_handshake(
322 struct live_viewer_connection
*viewer_connection
)
324 struct lttng_viewer_cmd cmd
;
325 struct lttng_viewer_connect connect
;
326 enum lttng_live_viewer_status status
;
327 bt_self_component_class
*self_comp_class
= viewer_connection
->self_comp_class
;
328 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
329 const size_t cmd_buf_len
= sizeof(cmd
) + sizeof(connect
);
330 char cmd_buf
[cmd_buf_len
];
332 BT_COMP_OR_COMP_CLASS_LOGD(self_comp
, self_comp_class
,
333 "Handshaking with the Relay: "
334 "major-version=%u, minor-version=%u",
335 LTTNG_LIVE_MAJOR
, LTTNG_LIVE_MINOR
);
337 cmd
.cmd
= htobe32(LTTNG_VIEWER_CONNECT
);
338 cmd
.data_size
= htobe64((uint64_t) sizeof(connect
));
339 cmd
.cmd_version
= htobe32(0);
341 connect
.viewer_session_id
= -1ULL; /* will be set on recv */
342 connect
.major
= htobe32(LTTNG_LIVE_MAJOR
);
343 connect
.minor
= htobe32(LTTNG_LIVE_MINOR
);
344 connect
.type
= htobe32(LTTNG_VIEWER_CLIENT_COMMAND
);
347 * Merge the cmd and connection request to prevent a write-write
348 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
349 * second write to be performed quickly in presence of Nagle's algorithm
351 memcpy(cmd_buf
, &cmd
, sizeof(cmd
));
352 memcpy(cmd_buf
+ sizeof(cmd
), &connect
, sizeof(connect
));
354 status
= lttng_live_send(viewer_connection
, &cmd_buf
, cmd_buf_len
);
355 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
356 viewer_handle_send_status(self_comp
, self_comp_class
,
357 status
, "viewer connect command");
361 status
= lttng_live_recv(viewer_connection
, &connect
, sizeof(connect
));
362 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
363 viewer_handle_recv_status(self_comp
, self_comp_class
,
364 status
, "viewer connect reply");
368 BT_COMP_OR_COMP_CLASS_LOGI(self_comp
, self_comp_class
,
369 "Received viewer session ID : %" PRIu64
,
370 (uint64_t) be64toh(connect
.viewer_session_id
));
371 BT_COMP_OR_COMP_CLASS_LOGI(self_comp
, self_comp_class
,
372 "Relayd version : %u.%u", be32toh(connect
.major
),
373 be32toh(connect
.minor
));
375 if (LTTNG_LIVE_MAJOR
!= be32toh(connect
.major
)) {
376 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
377 self_comp_class
, "Incompatible lttng-relayd protocol");
378 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
381 /* Use the smallest protocol version implemented. */
382 if (LTTNG_LIVE_MINOR
> be32toh(connect
.minor
)) {
383 viewer_connection
->minor
= be32toh(connect
.minor
);
385 viewer_connection
->minor
= LTTNG_LIVE_MINOR
;
387 viewer_connection
->major
= LTTNG_LIVE_MAJOR
;
389 status
= LTTNG_LIVE_VIEWER_STATUS_OK
;
398 enum lttng_live_viewer_status
lttng_live_connect_viewer(
399 struct live_viewer_connection
*viewer_connection
)
401 struct hostent
*host
;
402 struct sockaddr_in server_addr
;
403 enum lttng_live_viewer_status status
;
404 bt_self_component_class
*self_comp_class
= viewer_connection
->self_comp_class
;
405 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
407 if (parse_url(viewer_connection
)) {
408 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
409 self_comp_class
, "Failed to parse URL");
410 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
414 BT_COMP_OR_COMP_CLASS_LOGD(self_comp
, self_comp_class
,
415 "Connecting to hostname : %s, port : %d, "
416 "target hostname : %s, session name : %s, proto : %s",
417 viewer_connection
->relay_hostname
->str
,
418 viewer_connection
->port
,
419 !viewer_connection
->target_hostname
?
420 "<none>" : viewer_connection
->target_hostname
->str
,
421 !viewer_connection
->session_name
?
422 "<none>" : viewer_connection
->session_name
->str
,
423 viewer_connection
->proto
->str
);
425 host
= gethostbyname(viewer_connection
->relay_hostname
->str
);
427 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
428 self_comp_class
, "Cannot lookup hostname: hostname=\"%s\"",
429 viewer_connection
->relay_hostname
->str
);
430 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
434 if ((viewer_connection
->control_sock
= socket(AF_INET
, SOCK_STREAM
, 0)) == BT_INVALID_SOCKET
) {
435 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
436 self_comp_class
, "Socket creation failed: %s", bt_socket_errormsg());
437 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
441 server_addr
.sin_family
= AF_INET
;
442 server_addr
.sin_port
= htons(viewer_connection
->port
);
443 server_addr
.sin_addr
= *((struct in_addr
*) host
->h_addr
);
444 memset(&(server_addr
.sin_zero
), 0, 8);
446 if (connect(viewer_connection
->control_sock
, (struct sockaddr
*) &server_addr
,
447 sizeof(struct sockaddr
)) == BT_SOCKET_ERROR
) {
448 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
449 self_comp_class
, "Connection failed: %s",
450 bt_socket_errormsg());
451 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
455 status
= lttng_live_handshake(viewer_connection
);
458 * Only print error and append cause in case of error. not in case of
461 if (status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
462 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
463 self_comp_class
, "Viewer handshake failed");
465 } else if (status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
472 if (viewer_connection
->control_sock
!= BT_INVALID_SOCKET
) {
473 if (bt_socket_close(viewer_connection
->control_sock
) == BT_SOCKET_ERROR
) {
474 BT_COMP_OR_COMP_CLASS_LOGW(self_comp
, self_comp_class
,
475 "Error closing socket: %s.", bt_socket_errormsg());
478 viewer_connection
->control_sock
= BT_INVALID_SOCKET
;
484 void lttng_live_disconnect_viewer(
485 struct live_viewer_connection
*viewer_connection
)
487 bt_self_component_class
*self_comp_class
= viewer_connection
->self_comp_class
;
488 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
490 if (viewer_connection
->control_sock
== BT_INVALID_SOCKET
) {
493 if (bt_socket_close(viewer_connection
->control_sock
) == BT_SOCKET_ERROR
) {
494 BT_COMP_OR_COMP_CLASS_LOGW(self_comp
, self_comp_class
,
495 "Error closing socket: %s", bt_socket_errormsg());
496 viewer_connection
->control_sock
= BT_INVALID_SOCKET
;
501 int list_update_session(bt_value
*results
,
502 const struct lttng_viewer_session
*session
,
503 bool *_found
, struct live_viewer_connection
*viewer_connection
)
505 bt_self_component_class
*self_comp_class
= viewer_connection
->self_comp_class
;
506 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
509 bt_value
*map
= NULL
;
510 bt_value
*hostname
= NULL
;
511 bt_value
*session_name
= NULL
;
512 bt_value
*btval
= NULL
;
515 len
= bt_value_array_get_length(results
);
516 for (i
= 0; i
< len
; i
++) {
517 const char *hostname_str
= NULL
;
518 const char *session_name_str
= NULL
;
520 map
= bt_value_array_borrow_element_by_index(results
, i
);
521 hostname
= bt_value_map_borrow_entry_value(map
, "target-hostname");
523 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
525 "Error borrowing \"target-hostname\" entry.");
529 session_name
= bt_value_map_borrow_entry_value(map
, "session-name");
531 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
533 "Error borrowing \"session-name\" entry.");
537 hostname_str
= bt_value_string_get(hostname
);
538 session_name_str
= bt_value_string_get(session_name
);
540 if (strcmp(session
->hostname
, hostname_str
) == 0
541 && strcmp(session
->session_name
, session_name_str
) == 0) {
543 uint32_t streams
= be32toh(session
->streams
);
544 uint32_t clients
= be32toh(session
->clients
);
548 btval
= bt_value_map_borrow_entry_value(map
, "stream-count");
550 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(
551 self_comp
, self_comp_class
,
552 "Error borrowing \"stream-count\" entry.");
556 val
= bt_value_integer_unsigned_get(btval
);
559 bt_value_integer_unsigned_set(btval
, val
);
561 btval
= bt_value_map_borrow_entry_value(map
, "client-count");
563 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(
564 self_comp
, self_comp_class
,
565 "Error borrowing \"client-count\" entry.");
569 val
= bt_value_integer_unsigned_get(btval
);
571 val
= bt_max_t(int64_t, clients
, val
);
572 bt_value_integer_unsigned_set(btval
, val
);
585 int list_append_session(bt_value
*results
,
587 const struct lttng_viewer_session
*session
,
588 struct live_viewer_connection
*viewer_connection
)
591 bt_self_component_class
*self_comp_class
= viewer_connection
->self_comp_class
;
592 bt_value_map_insert_entry_status insert_status
;
593 bt_value_array_append_element_status append_status
;
594 bt_value
*map
= NULL
;
599 * If the session already exists, add the stream count to it,
600 * and do max of client counts.
602 ret
= list_update_session(results
, session
, &found
, viewer_connection
);
607 map
= bt_value_map_create();
609 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
610 "Error creating map value.");
615 if (base_url
->len
< 1) {
616 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
617 "Error: base_url length smaller than 1.");
625 url
= g_string_new(base_url
->str
);
626 g_string_append(url
, "/host/");
627 g_string_append(url
, session
->hostname
);
628 g_string_append_c(url
, '/');
629 g_string_append(url
, session
->session_name
);
631 insert_status
= bt_value_map_insert_string_entry(map
, "url", url
->str
);
632 if (insert_status
!= BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK
) {
633 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
634 "Error inserting \"url\" entry.");
640 * key = "target-hostname",
643 insert_status
= bt_value_map_insert_string_entry(map
, "target-hostname",
645 if (insert_status
!= BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK
) {
646 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
647 "Error inserting \"target-hostname\" entry.");
653 * key = "session-name",
656 insert_status
= bt_value_map_insert_string_entry(map
, "session-name",
657 session
->session_name
);
658 if (insert_status
!= BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK
) {
659 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
660 "Error inserting \"session-name\" entry.");
670 uint32_t live_timer
= be32toh(session
->live_timer
);
672 insert_status
= bt_value_map_insert_unsigned_integer_entry(
673 map
, "timer-us", live_timer
);
674 if (insert_status
!= BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK
) {
675 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
676 "Error inserting \"timer-us\" entry.");
683 * key = "stream-count",
687 uint32_t streams
= be32toh(session
->streams
);
689 insert_status
= bt_value_map_insert_unsigned_integer_entry(map
,
690 "stream-count", streams
);
691 if (insert_status
!= BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK
) {
692 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
693 "Error inserting \"stream-count\" entry.");
700 * key = "client-count",
704 uint32_t clients
= be32toh(session
->clients
);
706 insert_status
= bt_value_map_insert_unsigned_integer_entry(map
,
707 "client-count", clients
);
708 if (insert_status
!= BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK
) {
709 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
710 "Error inserting \"client-count\" entry.");
716 append_status
= bt_value_array_append_element(results
, map
);
717 if (append_status
!= BT_VALUE_ARRAY_APPEND_ELEMENT_STATUS_OK
) {
718 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
719 "Error appending map to results.");
725 g_string_free(url
, true);
727 BT_VALUE_PUT_REF_AND_RESET(map
);
732 * Data structure returned:
743 * key = "target-hostname",
747 * key = "session-name",
755 * key = "stream-count",
759 * key = "client-count",
768 bt_component_class_query_method_status
live_viewer_connection_list_sessions(
769 struct live_viewer_connection
*viewer_connection
,
770 const bt_value
**user_result
)
772 bt_self_component_class
*self_comp_class
= viewer_connection
->self_comp_class
;
773 bt_component_class_query_method_status status
=
774 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK
;
775 bt_value
*result
= NULL
;
776 enum lttng_live_viewer_status viewer_status
;
777 struct lttng_viewer_cmd cmd
;
778 struct lttng_viewer_list_sessions list
;
779 uint32_t i
, sessions_count
;
781 result
= bt_value_array_create();
783 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
784 "Error creating array");
785 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR
;
789 cmd
.cmd
= htobe32(LTTNG_VIEWER_LIST_SESSIONS
);
790 cmd
.data_size
= htobe64((uint64_t) 0);
791 cmd
.cmd_version
= htobe32(0);
793 viewer_status
= lttng_live_send(viewer_connection
, &cmd
, sizeof(cmd
));
794 if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
795 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
796 "Error sending list sessions command");
797 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR
;
799 } else if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
800 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN
;
804 viewer_status
= lttng_live_recv(viewer_connection
, &list
, sizeof(list
));
805 if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
806 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
807 "Error receiving session list");
808 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR
;
810 } else if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
811 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN
;
815 sessions_count
= be32toh(list
.sessions_count
);
816 for (i
= 0; i
< sessions_count
; i
++) {
817 struct lttng_viewer_session lsession
;
819 viewer_status
= lttng_live_recv(viewer_connection
, &lsession
,
821 if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
822 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
823 "Error receiving session:");
824 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR
;
826 } else if (viewer_status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
827 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN
;
831 lsession
.hostname
[LTTNG_VIEWER_HOST_NAME_MAX
- 1] = '\0';
832 lsession
.session_name
[LTTNG_VIEWER_NAME_MAX
- 1] = '\0';
833 if (list_append_session(result
, viewer_connection
->url
,
834 &lsession
, viewer_connection
)) {
835 BT_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp_class
,
836 "Error appending session");
837 status
= BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR
;
842 *user_result
= result
;
845 BT_VALUE_PUT_REF_AND_RESET(result
);
851 enum lttng_live_viewer_status
lttng_live_query_session_ids(
852 struct lttng_live_msg_iter
*lttng_live_msg_iter
)
854 struct lttng_viewer_cmd cmd
;
855 struct lttng_viewer_list_sessions list
;
856 struct lttng_viewer_session lsession
;
857 uint32_t i
, sessions_count
;
859 enum lttng_live_viewer_status status
;
860 struct live_viewer_connection
*viewer_connection
=
861 lttng_live_msg_iter
->viewer_connection
;
862 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
863 bt_self_component_class
*self_comp_class
=
864 viewer_connection
->self_comp_class
;
866 BT_COMP_LOGD("Asking the Relay for the list of sessions");
868 cmd
.cmd
= htobe32(LTTNG_VIEWER_LIST_SESSIONS
);
869 cmd
.data_size
= htobe64((uint64_t) 0);
870 cmd
.cmd_version
= htobe32(0);
872 status
= lttng_live_send(viewer_connection
, &cmd
, sizeof(cmd
));
873 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
874 viewer_handle_send_status(self_comp
, self_comp_class
,
875 status
, "list sessions command");
879 status
= lttng_live_recv(viewer_connection
, &list
, sizeof(list
));
880 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
881 viewer_handle_recv_status(self_comp
, self_comp_class
,
882 status
, "session list reply");
886 sessions_count
= be32toh(list
.sessions_count
);
887 for (i
= 0; i
< sessions_count
; i
++) {
888 status
= lttng_live_recv(viewer_connection
, &lsession
,
890 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
891 viewer_handle_recv_status(self_comp
, self_comp_class
,
892 status
, "session reply");
895 lsession
.hostname
[LTTNG_VIEWER_HOST_NAME_MAX
- 1] = '\0';
896 lsession
.session_name
[LTTNG_VIEWER_NAME_MAX
- 1] = '\0';
897 session_id
= be64toh(lsession
.id
);
899 BT_COMP_LOGI("Adding session to internal list: "
900 "session-id=%" PRIu64
", hostname=\"%s\", session-name=\"%s\"",
901 session_id
, lsession
.hostname
, lsession
.session_name
);
903 if ((strncmp(lsession
.session_name
,
904 viewer_connection
->session_name
->str
,
905 LTTNG_VIEWER_NAME_MAX
) == 0) && (strncmp(lsession
.hostname
,
906 viewer_connection
->target_hostname
->str
,
907 LTTNG_VIEWER_HOST_NAME_MAX
) == 0)) {
909 if (lttng_live_add_session(lttng_live_msg_iter
, session_id
,
911 lsession
.session_name
)) {
912 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
913 "Failed to add live session");
914 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
920 status
= LTTNG_LIVE_VIEWER_STATUS_OK
;
927 enum lttng_live_viewer_status
lttng_live_create_viewer_session(
928 struct lttng_live_msg_iter
*lttng_live_msg_iter
)
930 struct lttng_viewer_cmd cmd
;
931 struct lttng_viewer_create_session_response resp
;
932 enum lttng_live_viewer_status status
;
933 struct live_viewer_connection
*viewer_connection
=
934 lttng_live_msg_iter
->viewer_connection
;
935 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
936 bt_self_component_class
*self_comp_class
=
937 viewer_connection
->self_comp_class
;
939 BT_COMP_OR_COMP_CLASS_LOGD(self_comp
, self_comp_class
,
940 "Creating a viewer session");
942 cmd
.cmd
= htobe32(LTTNG_VIEWER_CREATE_SESSION
);
943 cmd
.data_size
= htobe64((uint64_t) 0);
944 cmd
.cmd_version
= htobe32(0);
946 status
= lttng_live_send(viewer_connection
, &cmd
, sizeof(cmd
));
947 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
948 viewer_handle_send_status(self_comp
, self_comp_class
,
949 status
, "create session command");
953 status
= lttng_live_recv(viewer_connection
, &resp
, sizeof(resp
));
954 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
955 viewer_handle_recv_status(self_comp
, self_comp_class
,
956 status
, "create session reply");
960 if (be32toh(resp
.status
) != LTTNG_VIEWER_CREATE_SESSION_OK
) {
961 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
962 "Error creating viewer session");
963 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
967 status
= lttng_live_query_session_ids(lttng_live_msg_iter
);
968 if (status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
969 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
970 "Failed to query live viewer session ids");
972 } else if (status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
981 enum lttng_live_viewer_status
receive_streams(struct lttng_live_session
*session
,
982 uint32_t stream_count
,
983 bt_self_message_iterator
*self_msg_iter
)
986 struct lttng_live_msg_iter
*lttng_live_msg_iter
=
987 session
->lttng_live_msg_iter
;
988 enum lttng_live_viewer_status status
;
989 struct live_viewer_connection
*viewer_connection
=
990 lttng_live_msg_iter
->viewer_connection
;
991 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
993 BT_COMP_LOGI("Getting %" PRIu32
" new streams:", stream_count
);
994 for (i
= 0; i
< stream_count
; i
++) {
995 struct lttng_viewer_stream stream
;
996 struct lttng_live_stream_iterator
*live_stream
;
998 uint64_t ctf_trace_id
;
1000 status
= lttng_live_recv(viewer_connection
, &stream
,
1002 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1003 viewer_handle_recv_status(self_comp
, NULL
,
1004 status
, "stream reply");
1007 stream
.path_name
[LTTNG_VIEWER_PATH_MAX
- 1] = '\0';
1008 stream
.channel_name
[LTTNG_VIEWER_NAME_MAX
- 1] = '\0';
1009 stream_id
= be64toh(stream
.id
);
1010 ctf_trace_id
= be64toh(stream
.ctf_trace_id
);
1012 if (stream
.metadata_flag
) {
1013 BT_COMP_LOGI(" metadata stream %" PRIu64
" : %s/%s",
1014 stream_id
, stream
.path_name
, stream
.channel_name
);
1015 if (lttng_live_metadata_create_stream(session
,
1016 ctf_trace_id
, stream_id
,
1017 stream
.path_name
)) {
1018 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1019 "Error creating metadata stream");
1020 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1023 session
->lazy_stream_msg_init
= true;
1025 BT_COMP_LOGI(" stream %" PRIu64
" : %s/%s",
1026 stream_id
, stream
.path_name
, stream
.channel_name
);
1027 live_stream
= lttng_live_stream_iterator_create(session
,
1028 ctf_trace_id
, stream_id
, self_msg_iter
);
1030 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1031 "Error creating stream");
1032 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1037 status
= LTTNG_LIVE_VIEWER_STATUS_OK
;
1044 enum lttng_live_viewer_status
lttng_live_session_attach(
1045 struct lttng_live_session
*session
,
1046 bt_self_message_iterator
*self_msg_iter
)
1048 struct lttng_viewer_cmd cmd
;
1049 enum lttng_live_viewer_status status
;
1050 struct lttng_viewer_attach_session_request rq
;
1051 struct lttng_viewer_attach_session_response rp
;
1052 struct lttng_live_msg_iter
*lttng_live_msg_iter
=
1053 session
->lttng_live_msg_iter
;
1054 struct live_viewer_connection
*viewer_connection
=
1055 lttng_live_msg_iter
->viewer_connection
;
1056 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
1057 uint64_t session_id
= session
->id
;
1058 uint32_t streams_count
;
1059 const size_t cmd_buf_len
= sizeof(cmd
) + sizeof(rq
);
1060 char cmd_buf
[cmd_buf_len
];
1062 BT_COMP_LOGD("Attaching to session: session-id=%"PRIu64
, session_id
);
1064 cmd
.cmd
= htobe32(LTTNG_VIEWER_ATTACH_SESSION
);
1065 cmd
.data_size
= htobe64((uint64_t) sizeof(rq
));
1066 cmd
.cmd_version
= htobe32(0);
1068 memset(&rq
, 0, sizeof(rq
));
1069 rq
.session_id
= htobe64(session_id
);
1070 // TODO: add cmd line parameter to select seek beginning
1071 // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
1072 rq
.seek
= htobe32(LTTNG_VIEWER_SEEK_LAST
);
1075 * Merge the cmd and connection request to prevent a write-write
1076 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1077 * second write to be performed quickly in presence of Nagle's algorithm.
1079 memcpy(cmd_buf
, &cmd
, sizeof(cmd
));
1080 memcpy(cmd_buf
+ sizeof(cmd
), &rq
, sizeof(rq
));
1081 status
= lttng_live_send(viewer_connection
, &cmd_buf
, cmd_buf_len
);
1082 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1083 viewer_handle_send_status(self_comp
, NULL
,
1084 status
, "attach session command");
1088 status
= lttng_live_recv(viewer_connection
, &rp
, sizeof(rp
));
1089 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1090 viewer_handle_recv_status(self_comp
, NULL
,
1091 status
, "attach session reply");
1095 streams_count
= be32toh(rp
.streams_count
);
1096 switch(be32toh(rp
.status
)) {
1097 case LTTNG_VIEWER_ATTACH_OK
:
1099 case LTTNG_VIEWER_ATTACH_UNK
:
1100 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1101 "Session id %" PRIu64
" is unknown", session_id
);
1102 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1104 case LTTNG_VIEWER_ATTACH_ALREADY
:
1105 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1106 "There is already a viewer attached to this session");
1107 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1109 case LTTNG_VIEWER_ATTACH_NOT_LIVE
:
1110 BT_COMP_LOGE_APPEND_CAUSE(self_comp
, "Not a live session");
1111 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1113 case LTTNG_VIEWER_ATTACH_SEEK_ERR
:
1114 BT_COMP_LOGE_APPEND_CAUSE(self_comp
, "Wrong seek parameter");
1115 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1118 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1119 "Unknown attach return code %u", be32toh(rp
.status
));
1120 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1124 /* We receive the initial list of streams. */
1125 status
= receive_streams(session
, streams_count
, self_msg_iter
);
1127 case LTTNG_LIVE_VIEWER_STATUS_OK
:
1129 case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
:
1131 case LTTNG_LIVE_VIEWER_STATUS_ERROR
:
1132 BT_COMP_LOGE_APPEND_CAUSE(self_comp
, "Error receiving streams");
1138 session
->attached
= true;
1139 session
->new_streams_needed
= false;
1146 enum lttng_live_viewer_status
lttng_live_session_detach(
1147 struct lttng_live_session
*session
)
1149 struct lttng_viewer_cmd cmd
;
1150 enum lttng_live_viewer_status status
;
1151 struct lttng_viewer_detach_session_request rq
;
1152 struct lttng_viewer_detach_session_response rp
;
1153 struct lttng_live_msg_iter
*lttng_live_msg_iter
=
1154 session
->lttng_live_msg_iter
;
1155 bt_self_component
*self_comp
= session
->self_comp
;
1156 struct live_viewer_connection
*viewer_connection
=
1157 lttng_live_msg_iter
->viewer_connection
;
1158 uint64_t session_id
= session
->id
;
1159 const size_t cmd_buf_len
= sizeof(cmd
) + sizeof(rq
);
1160 char cmd_buf
[cmd_buf_len
];
1163 * The session might already be detached and the viewer socket might
1164 * already been closed. This happens when calling this function when
1165 * tearing down the graph after an error.
1167 if (!session
->attached
|| viewer_connection
->control_sock
== BT_INVALID_SOCKET
) {
1171 cmd
.cmd
= htobe32(LTTNG_VIEWER_DETACH_SESSION
);
1172 cmd
.data_size
= htobe64((uint64_t) sizeof(rq
));
1173 cmd
.cmd_version
= htobe32(0);
1175 memset(&rq
, 0, sizeof(rq
));
1176 rq
.session_id
= htobe64(session_id
);
1179 * Merge the cmd and connection request to prevent a write-write
1180 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1181 * second write to be performed quickly in presence of Nagle's algorithm.
1183 memcpy(cmd_buf
, &cmd
, sizeof(cmd
));
1184 memcpy(cmd_buf
+ sizeof(cmd
), &rq
, sizeof(rq
));
1185 status
= lttng_live_send(viewer_connection
, &cmd_buf
, cmd_buf_len
);
1186 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1187 viewer_handle_send_status(self_comp
, NULL
,
1188 status
, "detach session command");
1192 status
= lttng_live_recv(viewer_connection
, &rp
, sizeof(rp
));
1193 if (status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1194 viewer_handle_recv_status(self_comp
, NULL
,
1195 status
, "detach session reply");
1199 switch(be32toh(rp
.status
)) {
1200 case LTTNG_VIEWER_DETACH_SESSION_OK
:
1202 case LTTNG_VIEWER_DETACH_SESSION_UNK
:
1203 BT_COMP_LOGW("Session id %" PRIu64
" is unknown", session_id
);
1204 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1206 case LTTNG_VIEWER_DETACH_SESSION_ERR
:
1207 BT_COMP_LOGW("Error detaching session id %" PRIu64
"", session_id
);
1208 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1211 BT_COMP_LOGE("Unknown detach return code %u", be32toh(rp
.status
));
1212 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1216 session
->attached
= false;
1218 status
= LTTNG_LIVE_VIEWER_STATUS_OK
;
1225 enum lttng_live_get_one_metadata_status
lttng_live_get_one_metadata_packet(
1226 struct lttng_live_trace
*trace
, FILE *fp
, size_t *reply_len
)
1229 enum lttng_live_get_one_metadata_status status
;
1230 enum lttng_live_viewer_status viewer_status
;
1231 struct lttng_viewer_cmd cmd
;
1232 struct lttng_viewer_get_metadata rq
;
1233 struct lttng_viewer_metadata_packet rp
;
1236 struct lttng_live_session
*session
= trace
->session
;
1237 struct lttng_live_msg_iter
*lttng_live_msg_iter
=
1238 session
->lttng_live_msg_iter
;
1239 struct lttng_live_metadata
*metadata
= trace
->metadata
;
1240 struct live_viewer_connection
*viewer_connection
=
1241 lttng_live_msg_iter
->viewer_connection
;
1242 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
1243 const size_t cmd_buf_len
= sizeof(cmd
) + sizeof(rq
);
1244 char cmd_buf
[cmd_buf_len
];
1246 BT_COMP_LOGD("Requesting new metadata for trace: "
1247 "trace-id=%"PRIu64
", metadata-stream-id=%"PRIu64
,
1248 trace
->id
, metadata
->stream_id
);
1250 rq
.stream_id
= htobe64(metadata
->stream_id
);
1251 cmd
.cmd
= htobe32(LTTNG_VIEWER_GET_METADATA
);
1252 cmd
.data_size
= htobe64((uint64_t) sizeof(rq
));
1253 cmd
.cmd_version
= htobe32(0);
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.
1260 memcpy(cmd_buf
, &cmd
, sizeof(cmd
));
1261 memcpy(cmd_buf
+ sizeof(cmd
), &rq
, sizeof(rq
));
1262 viewer_status
= lttng_live_send(viewer_connection
, &cmd_buf
, cmd_buf_len
);
1263 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1264 viewer_handle_send_status(self_comp
, NULL
,
1265 viewer_status
, "get metadata command");
1266 status
= (enum lttng_live_get_one_metadata_status
) viewer_status
;
1270 viewer_status
= lttng_live_recv(viewer_connection
, &rp
, sizeof(rp
));
1271 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1272 viewer_handle_recv_status(self_comp
, NULL
,
1273 viewer_status
, "get metadata reply");
1274 status
= (enum lttng_live_get_one_metadata_status
) viewer_status
;
1278 switch (be32toh(rp
.status
)) {
1279 case LTTNG_VIEWER_METADATA_OK
:
1280 BT_COMP_LOGD("Received get_metadata response: ok");
1282 case LTTNG_VIEWER_NO_NEW_METADATA
:
1283 BT_COMP_LOGD("Received get_metadata response: no new");
1284 status
= LTTNG_LIVE_GET_ONE_METADATA_STATUS_END
;
1286 case LTTNG_VIEWER_METADATA_ERR
:
1288 * The Relayd cannot find this stream id. Maybe its
1289 * gone already. This can happen in short lived UST app
1290 * in a per-pid session.
1292 BT_COMP_LOGD("Received get_metadata response: error");
1293 status
= LTTNG_LIVE_GET_ONE_METADATA_STATUS_CLOSED
;
1296 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1297 "Received get_metadata response: unknown");
1298 status
= LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR
;
1302 len
= be64toh(rp
.len
);
1303 BT_COMP_LOGD("Writing %" PRIu64
" bytes to metadata", len
);
1305 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1306 "Erroneous response length");
1307 status
= LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR
;
1311 data
= g_new0(gchar
, len
);
1313 BT_COMP_LOGE_APPEND_CAUSE_ERRNO(self_comp
,
1314 "Failed to allocate data buffer", ".");
1315 status
= LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR
;
1319 viewer_status
= lttng_live_recv(viewer_connection
, data
, len
);
1320 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1321 viewer_handle_recv_status(self_comp
, NULL
,
1322 viewer_status
, "get metadata packet");
1323 status
= (enum lttng_live_get_one_metadata_status
) viewer_status
;
1328 * Write the metadata to the file handle.
1330 writelen
= fwrite(data
, sizeof(uint8_t), len
, fp
);
1331 if (writelen
!= len
) {
1332 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1333 "Writing in the metadata file stream");
1334 status
= LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR
;
1339 status
= LTTNG_LIVE_GET_ONE_METADATA_STATUS_OK
;
1347 * Assign the fields from a lttng_viewer_index to a packet_index.
1350 void lttng_index_to_packet_index(struct lttng_viewer_index
*lindex
,
1351 struct packet_index
*pindex
)
1356 pindex
->offset
= be64toh(lindex
->offset
);
1357 pindex
->packet_size
= be64toh(lindex
->packet_size
);
1358 pindex
->content_size
= be64toh(lindex
->content_size
);
1359 pindex
->ts_cycles
.timestamp_begin
= be64toh(lindex
->timestamp_begin
);
1360 pindex
->ts_cycles
.timestamp_end
= be64toh(lindex
->timestamp_end
);
1361 pindex
->events_discarded
= be64toh(lindex
->events_discarded
);
1365 void lttng_live_need_new_streams(struct lttng_live_msg_iter
*lttng_live_msg_iter
)
1367 uint64_t session_idx
;
1369 for (session_idx
= 0; session_idx
< lttng_live_msg_iter
->sessions
->len
;
1371 struct lttng_live_session
*session
=
1372 g_ptr_array_index(lttng_live_msg_iter
->sessions
, session_idx
);
1373 session
->new_streams_needed
= true;
1378 enum lttng_live_iterator_status
lttng_live_get_next_index(
1379 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
1380 struct lttng_live_stream_iterator
*stream
,
1381 struct packet_index
*index
)
1383 struct lttng_viewer_cmd cmd
;
1384 struct lttng_viewer_get_next_index rq
;
1385 enum lttng_live_viewer_status viewer_status
;
1386 struct lttng_viewer_index rp
;
1387 enum lttng_live_iterator_status status
;
1388 struct live_viewer_connection
*viewer_connection
=
1389 lttng_live_msg_iter
->viewer_connection
;
1390 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
1391 struct lttng_live_trace
*trace
= stream
->trace
;
1392 const size_t cmd_buf_len
= sizeof(cmd
) + sizeof(rq
);
1393 char cmd_buf
[cmd_buf_len
];
1394 uint32_t flags
, rp_status
;
1396 BT_COMP_LOGD("Requesting next index for stream: "
1397 "stream-id=%"PRIu64
, stream
->viewer_stream_id
);
1399 cmd
.cmd
= htobe32(LTTNG_VIEWER_GET_NEXT_INDEX
);
1400 cmd
.data_size
= htobe64((uint64_t) sizeof(rq
));
1401 cmd
.cmd_version
= htobe32(0);
1404 memset(&rq
, 0, sizeof(rq
));
1405 rq
.stream_id
= htobe64(stream
->viewer_stream_id
);
1408 * Merge the cmd and connection request to prevent a write-write
1409 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1410 * second write to be performed quickly in presence of Nagle's algorithm.
1412 memcpy(cmd_buf
, &cmd
, sizeof(cmd
));
1413 memcpy(cmd_buf
+ sizeof(cmd
), &rq
, sizeof(rq
));
1414 viewer_status
= lttng_live_send(viewer_connection
, &cmd_buf
, cmd_buf_len
);
1415 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1416 viewer_handle_send_status(self_comp
, NULL
,
1417 viewer_status
, "get next index command");
1421 viewer_status
= lttng_live_recv(viewer_connection
, &rp
, sizeof(rp
));
1422 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1423 viewer_handle_recv_status(self_comp
, NULL
,
1424 viewer_status
, "get next index reply");
1428 flags
= be32toh(rp
.flags
);
1429 rp_status
= be32toh(rp
.status
);
1431 switch (rp_status
) {
1432 case LTTNG_VIEWER_INDEX_INACTIVE
:
1434 uint64_t ctf_stream_class_id
;
1436 BT_COMP_LOGD("Received get_next_index response: inactive");
1437 memset(index
, 0, sizeof(struct packet_index
));
1438 index
->ts_cycles
.timestamp_end
= be64toh(rp
.timestamp_end
);
1439 stream
->current_inactivity_ts
= index
->ts_cycles
.timestamp_end
;
1440 ctf_stream_class_id
= be64toh(rp
.stream_id
);
1441 if (stream
->ctf_stream_class_id
!= -1ULL) {
1442 BT_ASSERT(stream
->ctf_stream_class_id
==
1443 ctf_stream_class_id
);
1445 stream
->ctf_stream_class_id
= ctf_stream_class_id
;
1447 stream
->state
= LTTNG_LIVE_STREAM_QUIESCENT
;
1448 status
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
1451 case LTTNG_VIEWER_INDEX_OK
:
1453 uint64_t ctf_stream_class_id
;
1455 BT_COMP_LOGD("Received get_next_index response: OK");
1456 lttng_index_to_packet_index(&rp
, index
);
1457 ctf_stream_class_id
= be64toh(rp
.stream_id
);
1458 if (stream
->ctf_stream_class_id
!= -1ULL) {
1459 BT_ASSERT(stream
->ctf_stream_class_id
==
1460 ctf_stream_class_id
);
1462 stream
->ctf_stream_class_id
= ctf_stream_class_id
;
1465 stream
->state
= LTTNG_LIVE_STREAM_ACTIVE_DATA
;
1467 if (flags
& LTTNG_VIEWER_FLAG_NEW_METADATA
) {
1468 BT_COMP_LOGD("Received get_next_index response: new metadata needed");
1469 trace
->metadata_stream_state
= LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED
;
1471 if (flags
& LTTNG_VIEWER_FLAG_NEW_STREAM
) {
1472 BT_COMP_LOGD("Received get_next_index response: new streams needed");
1473 lttng_live_need_new_streams(lttng_live_msg_iter
);
1475 status
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
1478 case LTTNG_VIEWER_INDEX_RETRY
:
1479 BT_COMP_LOGD("Received get_next_index response: retry");
1480 memset(index
, 0, sizeof(struct packet_index
));
1481 stream
->state
= LTTNG_LIVE_STREAM_ACTIVE_NO_DATA
;
1482 status
= LTTNG_LIVE_ITERATOR_STATUS_AGAIN
;
1484 case LTTNG_VIEWER_INDEX_HUP
:
1485 BT_COMP_LOGD("Received get_next_index response: stream hung up");
1486 memset(index
, 0, sizeof(struct packet_index
));
1487 index
->offset
= EOF
;
1488 stream
->state
= LTTNG_LIVE_STREAM_EOF
;
1489 stream
->has_stream_hung_up
= true;
1490 status
= LTTNG_LIVE_ITERATOR_STATUS_END
;
1492 case LTTNG_VIEWER_INDEX_ERR
:
1493 BT_COMP_LOGD("Received get_next_index response: error");
1494 memset(index
, 0, sizeof(struct packet_index
));
1495 stream
->state
= LTTNG_LIVE_STREAM_ACTIVE_NO_DATA
;
1496 status
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
1499 BT_COMP_LOGD("Received get_next_index response: unknown value");
1500 memset(index
, 0, sizeof(struct packet_index
));
1501 stream
->state
= LTTNG_LIVE_STREAM_ACTIVE_NO_DATA
;
1502 status
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
1508 status
= viewer_status_to_live_iterator_status(viewer_status
);
1514 enum ctf_msg_iter_medium_status
lttng_live_get_stream_bytes(
1515 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
1516 struct lttng_live_stream_iterator
*stream
, uint8_t *buf
,
1517 uint64_t offset
, uint64_t req_len
, uint64_t *recv_len
)
1519 enum ctf_msg_iter_medium_status status
;
1520 enum lttng_live_viewer_status viewer_status
;
1521 struct lttng_viewer_trace_packet rp
;
1522 struct lttng_viewer_cmd cmd
;
1523 struct lttng_viewer_get_packet rq
;
1524 struct live_viewer_connection
*viewer_connection
=
1525 lttng_live_msg_iter
->viewer_connection
;
1526 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
1527 struct lttng_live_trace
*trace
= stream
->trace
;
1528 const size_t cmd_buf_len
= sizeof(cmd
) + sizeof(rq
);
1529 char cmd_buf
[cmd_buf_len
];
1530 uint32_t flags
, rp_status
;
1532 BT_COMP_LOGD("lttng_live_get_stream_bytes: offset=%" PRIu64
", req_len=%" PRIu64
,
1534 cmd
.cmd
= htobe32(LTTNG_VIEWER_GET_PACKET
);
1535 cmd
.data_size
= htobe64((uint64_t) sizeof(rq
));
1536 cmd
.cmd_version
= htobe32(0);
1538 memset(&rq
, 0, sizeof(rq
));
1539 rq
.stream_id
= htobe64(stream
->viewer_stream_id
);
1540 rq
.offset
= htobe64(offset
);
1541 rq
.len
= htobe32(req_len
);
1544 * Merge the cmd and connection request to prevent a write-write
1545 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1546 * second write to be performed quickly in presence of Nagle's algorithm.
1548 memcpy(cmd_buf
, &cmd
, sizeof(cmd
));
1549 memcpy(cmd_buf
+ sizeof(cmd
), &rq
, sizeof(rq
));
1550 viewer_status
= lttng_live_send(viewer_connection
, &cmd_buf
, cmd_buf_len
);
1551 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1552 viewer_handle_send_status(self_comp
, NULL
,
1553 viewer_status
, "get data packet command");
1554 goto error_convert_status
;
1557 viewer_status
= lttng_live_recv(viewer_connection
, &rp
, sizeof(rp
));
1558 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1559 viewer_handle_recv_status(self_comp
, NULL
,
1560 viewer_status
, "get data packet reply");
1561 goto error_convert_status
;
1564 flags
= be32toh(rp
.flags
);
1565 rp_status
= be32toh(rp
.status
);
1567 switch (rp_status
) {
1568 case LTTNG_VIEWER_GET_PACKET_OK
:
1569 req_len
= be32toh(rp
.len
);
1570 BT_COMP_LOGD("Received get_data_packet response: Ok, "
1571 "packet size : %" PRIu64
"", req_len
);
1572 status
= CTF_MSG_ITER_MEDIUM_STATUS_OK
;
1574 case LTTNG_VIEWER_GET_PACKET_RETRY
:
1575 /* Unimplemented by relay daemon */
1576 BT_COMP_LOGD("Received get_data_packet response: retry");
1577 status
= CTF_MSG_ITER_MEDIUM_STATUS_AGAIN
;
1579 case LTTNG_VIEWER_GET_PACKET_ERR
:
1580 if (flags
& LTTNG_VIEWER_FLAG_NEW_METADATA
) {
1581 BT_COMP_LOGD("get_data_packet: new metadata needed, try again later");
1582 trace
->metadata_stream_state
= LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED
;
1584 if (flags
& LTTNG_VIEWER_FLAG_NEW_STREAM
) {
1585 BT_COMP_LOGD("get_data_packet: new streams needed, try again later");
1586 lttng_live_need_new_streams(lttng_live_msg_iter
);
1588 if (flags
& (LTTNG_VIEWER_FLAG_NEW_METADATA
1589 | LTTNG_VIEWER_FLAG_NEW_STREAM
)) {
1590 status
= CTF_MSG_ITER_MEDIUM_STATUS_AGAIN
;
1593 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1594 "Received get_data_packet response: error");
1595 status
= CTF_MSG_ITER_MEDIUM_STATUS_ERROR
;
1597 case LTTNG_VIEWER_GET_PACKET_EOF
:
1598 status
= CTF_MSG_ITER_MEDIUM_STATUS_EOF
;
1601 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1602 "Received get_data_packet response: unknown (%d)", rp_status
);
1603 status
= CTF_MSG_ITER_MEDIUM_STATUS_ERROR
;
1608 status
= CTF_MSG_ITER_MEDIUM_STATUS_ERROR
;
1612 viewer_status
= lttng_live_recv(viewer_connection
, buf
, req_len
);
1613 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1614 viewer_handle_recv_status(self_comp
, NULL
,
1615 viewer_status
, "get data packet");
1616 goto error_convert_status
;
1618 *recv_len
= req_len
;
1620 status
= CTF_MSG_ITER_MEDIUM_STATUS_OK
;
1623 error_convert_status
:
1624 status
= viewer_status_to_ctf_msg_iter_medium_status(viewer_status
);
1630 * Request new streams for a session.
1633 enum lttng_live_iterator_status
lttng_live_session_get_new_streams(
1634 struct lttng_live_session
*session
,
1635 bt_self_message_iterator
*self_msg_iter
)
1637 enum lttng_live_iterator_status status
=
1638 LTTNG_LIVE_ITERATOR_STATUS_OK
;
1639 struct lttng_viewer_cmd cmd
;
1640 struct lttng_viewer_new_streams_request rq
;
1641 struct lttng_viewer_new_streams_response rp
;
1642 struct lttng_live_msg_iter
*lttng_live_msg_iter
=
1643 session
->lttng_live_msg_iter
;
1644 enum lttng_live_viewer_status viewer_status
;
1645 struct live_viewer_connection
*viewer_connection
=
1646 lttng_live_msg_iter
->viewer_connection
;
1647 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
1648 uint32_t streams_count
;
1649 const size_t cmd_buf_len
= sizeof(cmd
) + sizeof(rq
);
1650 char cmd_buf
[cmd_buf_len
];
1652 if (!session
->new_streams_needed
) {
1653 status
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
1657 BT_COMP_LOGD("Requesting new streams for session: "
1658 "session-id=%"PRIu64
, session
->id
);
1660 cmd
.cmd
= htobe32(LTTNG_VIEWER_GET_NEW_STREAMS
);
1661 cmd
.data_size
= htobe64((uint64_t) sizeof(rq
));
1662 cmd
.cmd_version
= htobe32(0);
1664 memset(&rq
, 0, sizeof(rq
));
1665 rq
.session_id
= htobe64(session
->id
);
1668 * Merge the cmd and connection request to prevent a write-write
1669 * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1670 * second write to be performed quickly in presence of Nagle's algorithm.
1672 memcpy(cmd_buf
, &cmd
, sizeof(cmd
));
1673 memcpy(cmd_buf
+ sizeof(cmd
), &rq
, sizeof(rq
));
1674 viewer_status
= lttng_live_send(viewer_connection
, &cmd_buf
, cmd_buf_len
);
1675 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1676 viewer_handle_send_status(self_comp
, NULL
,
1677 viewer_status
, "get new streams command");
1678 status
= viewer_status_to_live_iterator_status(viewer_status
);
1682 viewer_status
= lttng_live_recv(viewer_connection
, &rp
, sizeof(rp
));
1683 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1684 viewer_handle_recv_status(self_comp
, NULL
,
1685 viewer_status
, "get new streams reply");
1686 status
= viewer_status_to_live_iterator_status(viewer_status
);
1690 streams_count
= be32toh(rp
.streams_count
);
1692 switch(be32toh(rp
.status
)) {
1693 case LTTNG_VIEWER_NEW_STREAMS_OK
:
1694 session
->new_streams_needed
= false;
1696 case LTTNG_VIEWER_NEW_STREAMS_NO_NEW
:
1697 session
->new_streams_needed
= false;
1699 case LTTNG_VIEWER_NEW_STREAMS_HUP
:
1700 session
->new_streams_needed
= false;
1701 session
->closed
= true;
1702 status
= LTTNG_LIVE_ITERATOR_STATUS_END
;
1704 case LTTNG_VIEWER_NEW_STREAMS_ERR
:
1705 BT_COMP_LOGD("Received get_new_streams response: error");
1706 status
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
1709 BT_COMP_LOGE_APPEND_CAUSE(self_comp
,
1710 "Received get_new_streams response: Unknown:"
1711 "return code %u", be32toh(rp
.status
));
1712 status
= LTTNG_LIVE_ITERATOR_STATUS_ERROR
;
1716 viewer_status
= receive_streams(session
, streams_count
, self_msg_iter
);
1717 if (viewer_status
!= LTTNG_LIVE_VIEWER_STATUS_OK
) {
1718 viewer_handle_recv_status(self_comp
, NULL
,
1719 viewer_status
, "new streams");
1720 status
= viewer_status_to_live_iterator_status(viewer_status
);
1724 status
= LTTNG_LIVE_ITERATOR_STATUS_OK
;
1730 enum lttng_live_viewer_status
live_viewer_connection_create(
1731 bt_self_component
*self_comp
,
1732 bt_self_component_class
*self_comp_class
,
1733 bt_logging_level log_level
,
1734 const char *url
, bool in_query
,
1735 struct lttng_live_msg_iter
*lttng_live_msg_iter
,
1736 struct live_viewer_connection
**viewer
)
1738 struct live_viewer_connection
*viewer_connection
;
1739 enum lttng_live_viewer_status status
;
1741 viewer_connection
= g_new0(struct live_viewer_connection
, 1);
1743 if (bt_socket_init(log_level
) != 0) {
1744 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
1745 self_comp_class
, "Failed to init socket");
1746 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1750 viewer_connection
->log_level
= log_level
;
1752 viewer_connection
->self_comp
= self_comp
;
1753 viewer_connection
->self_comp_class
= self_comp_class
;
1755 viewer_connection
->control_sock
= BT_INVALID_SOCKET
;
1756 viewer_connection
->port
= -1;
1757 viewer_connection
->in_query
= in_query
;
1758 viewer_connection
->lttng_live_msg_iter
= lttng_live_msg_iter
;
1759 viewer_connection
->url
= g_string_new(url
);
1760 if (!viewer_connection
->url
) {
1761 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
1762 self_comp_class
, "Failed to allocate URL buffer");
1763 status
= LTTNG_LIVE_VIEWER_STATUS_ERROR
;
1767 BT_COMP_OR_COMP_CLASS_LOGD(self_comp
, self_comp_class
,
1768 "Establishing connection to url \"%s\"...", url
);
1769 status
= lttng_live_connect_viewer(viewer_connection
);
1771 * Only print error and append cause in case of error. not in case of
1774 if (status
== LTTNG_LIVE_VIEWER_STATUS_ERROR
) {
1775 BT_COMP_OR_COMP_CLASS_LOGE_APPEND_CAUSE(self_comp
,
1776 self_comp_class
, "Failed to establish connection: "
1779 } else if (status
== LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED
) {
1782 BT_COMP_OR_COMP_CLASS_LOGD(self_comp
, self_comp_class
,
1783 "Connection to url \"%s\" is established", url
);
1785 *viewer
= viewer_connection
;
1786 status
= LTTNG_LIVE_VIEWER_STATUS_OK
;
1790 if (viewer_connection
) {
1791 live_viewer_connection_destroy(viewer_connection
);
1798 void live_viewer_connection_destroy(
1799 struct live_viewer_connection
*viewer_connection
)
1801 bt_self_component
*self_comp
= viewer_connection
->self_comp
;
1802 bt_self_component_class
*self_comp_class
= viewer_connection
->self_comp_class
;
1804 if (!viewer_connection
) {
1808 BT_COMP_OR_COMP_CLASS_LOGD(self_comp
, self_comp_class
,
1809 "Closing connection to relay:"
1810 "relay-url=\"%s\"", viewer_connection
->url
->str
);
1812 lttng_live_disconnect_viewer(viewer_connection
);
1814 if (viewer_connection
->url
) {
1815 g_string_free(viewer_connection
->url
, true);
1818 if (viewer_connection
->relay_hostname
) {
1819 g_string_free(viewer_connection
->relay_hostname
, true);
1822 if (viewer_connection
->target_hostname
) {
1823 g_string_free(viewer_connection
->target_hostname
, true);
1826 if (viewer_connection
->session_name
) {
1827 g_string_free(viewer_connection
->session_name
, true);
1830 if (viewer_connection
->proto
) {
1831 g_string_free(viewer_connection
->proto
, true);
1834 g_free(viewer_connection
);