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