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