src/common/relayd/relayd.c

   1 /*
   2  * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
   3  *
   4  * This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License, version 2 only, as
   6  * published by the Free Software Foundation.
   7  *
   8  * This program is distributed in the hope that it will be useful, but WITHOUT
   9  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  11  * more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along with
  14  * this program; if not, write to the Free Software Foundation, Inc., 51
  15  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16  */
  17
  18 #define _LGPL_SOURCE
  19 #include <assert.h>
  20 #include <stdio.h>
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <sys/stat.h>
  24 #include <inttypes.h>
  25
  26 #include <common/common.h>
  27 #include <common/defaults.h>
  28 #include <common/compat/endian.h>
  29 #include <common/compat/string.h>
  30 #include <common/sessiond-comm/relayd.h>
  31 #include <common/index/ctf-index.h>
  32 #include <common/trace-chunk.h>
  33
  34 #include "relayd.h"
  35
  36 /*
  37  * Send command. Fill up the header and append the data.
  38  */
  39 static int send_command(struct lttcomm_relayd_sock *rsock,
  40                 enum lttcomm_relayd_command cmd, const void *data, size_t size,
  41                 int flags)
  42 {
  43         int ret;
  44         struct lttcomm_relayd_hdr header;
  45         char *buf;
  46         uint64_t buf_size = sizeof(header);
  47
  48         if (rsock->sock.fd < 0) {
  49                 return -ECONNRESET;
  50         }
  51
  52         if (data) {
  53                 buf_size += size;
  54         }
  55
  56         buf = zmalloc(buf_size);
  57         if (buf == NULL) {
  58                 PERROR("zmalloc relayd send command buf");
  59                 ret = -1;
  60                 goto alloc_error;
  61         }
  62
  63         memset(&header, 0, sizeof(header));
  64         header.cmd = htobe32(cmd);
  65         header.data_size = htobe64(size);
  66
  67         /* Zeroed for now since not used. */
  68         header.cmd_version = 0;
  69         header.circuit_id = 0;
  70
  71         /* Prepare buffer to send. */
  72         memcpy(buf, &header, sizeof(header));
  73         if (data) {
  74                 memcpy(buf + sizeof(header), data, size);
  75         }
  76
  77         DBG3("Relayd sending command %d of size %" PRIu64, (int) cmd, buf_size);
  78         ret = rsock->sock.ops->sendmsg(&rsock->sock, buf, buf_size, flags);
  79         if (ret < 0) {
  80                 PERROR("Failed to send command %d of size %" PRIu64,
  81                                 (int) cmd, buf_size);
  82                 ret = -errno;
  83                 goto error;
  84         }
  85 error:
  86         free(buf);
  87 alloc_error:
  88         return ret;
  89 }
  90
  91 /*
  92  * Receive reply data on socket. This MUST be call after send_command or else
  93  * could result in unexpected behavior(s).
  94  */
  95 static int recv_reply(struct lttcomm_relayd_sock *rsock, void *data, size_t size)
  96 {
  97         int ret;
  98
  99         if (rsock->sock.fd < 0) {
 100                 return -ECONNRESET;
 101         }
 102
 103         DBG3("Relayd waiting for reply of size %zu", size);
 104
 105         ret = rsock->sock.ops->recvmsg(&rsock->sock, data, size, 0);
 106         if (ret <= 0 || ret != size) {
 107                 if (ret == 0) {
 108                         /* Orderly shutdown. */
 109                         DBG("Socket %d has performed an orderly shutdown", rsock->sock.fd);
 110                 } else {
 111                         DBG("Receiving reply failed on sock %d for size %zu with ret %d",
 112                                         rsock->sock.fd, size, ret);
 113                 }
 114                 /* Always return -1 here and the caller can use errno. */
 115                 ret = -1;
 116                 goto error;
 117         }
 118
 119 error:
 120         return ret;
 121 }
 122
 123 /*
 124  * Starting from 2.11, RELAYD_CREATE_SESSION payload (session_name & hostname)
 125  * have no length restriction on the sender side.
 126  * Length for both payloads is stored in the msg struct. A new dynamic size
 127  * payload size is introduced.
 128  */
 129 static int relayd_create_session_2_11(struct lttcomm_relayd_sock *rsock,
 130                 const char *session_name, const char *hostname,
 131                 int session_live_timer, unsigned int snapshot,
 132                 uint64_t sessiond_session_id, const lttng_uuid sessiond_uuid,
 133                 const uint64_t *current_chunk_id)
 134 {
 135         int ret;
 136         struct lttcomm_relayd_create_session_2_11 *msg = NULL;
 137         size_t session_name_len;
 138         size_t hostname_len;
 139         size_t msg_length;
 140
 141         /* The two names are sent with a '\0' delimiter between them. */
 142         session_name_len = strlen(session_name) + 1;
 143         hostname_len = strlen(hostname) + 1;
 144
 145         msg_length = sizeof(*msg) + session_name_len + hostname_len;
 146         msg = zmalloc(msg_length);
 147         if (!msg) {
 148                 PERROR("zmalloc create_session_2_11 command message");
 149                 ret = -1;
 150                 goto error;
 151         }
 152
 153         assert(session_name_len <= UINT32_MAX);
 154         msg->session_name_len = htobe32(session_name_len);
 155
 156         assert(hostname_len <= UINT32_MAX);
 157         msg->hostname_len = htobe32(hostname_len);
 158
 159         if (lttng_strncpy(msg->names, session_name, session_name_len)) {
 160                 ret = -1;
 161                 goto error;
 162         }
 163         if (lttng_strncpy(msg->names + session_name_len, hostname, hostname_len)) {
 164                 ret = -1;
 165                 goto error;
 166         }
 167
 168         msg->live_timer = htobe32(session_live_timer);
 169         msg->snapshot = !!snapshot;
 170
 171         lttng_uuid_copy(msg->sessiond_uuid, sessiond_uuid);
 172         msg->session_id = htobe64(sessiond_session_id);
 173
 174         if (current_chunk_id) {
 175                 LTTNG_OPTIONAL_SET(&msg->current_chunk_id,
 176                                 htobe64(*current_chunk_id));
 177         }
 178
 179         /* Send command */
 180         ret = send_command(rsock, RELAYD_CREATE_SESSION, msg, msg_length, 0);
 181         if (ret < 0) {
 182                 goto error;
 183         }
 184 error:
 185         free(msg);
 186         return ret;
 187 }
 188 /*
 189  * From 2.4 to 2.10, RELAYD_CREATE_SESSION takes additional parameters to
 190  * support the live reading capability.
 191  */
 192 static int relayd_create_session_2_4(struct lttcomm_relayd_sock *rsock,
 193                 const char *session_name, const char *hostname,
 194                 int session_live_timer, unsigned int snapshot)
 195 {
 196         int ret;
 197         struct lttcomm_relayd_create_session_2_4 msg;
 198
 199         if (lttng_strncpy(msg.session_name, session_name,
 200                         sizeof(msg.session_name))) {
 201                 ret = -1;
 202                 goto error;
 203         }
 204         if (lttng_strncpy(msg.hostname, hostname, sizeof(msg.hostname))) {
 205                 ret = -1;
 206                 goto error;
 207         }
 208         msg.live_timer = htobe32(session_live_timer);
 209         msg.snapshot = htobe32(snapshot);
 210
 211         /* Send command */
 212         ret = send_command(rsock, RELAYD_CREATE_SESSION, &msg, sizeof(msg), 0);
 213         if (ret < 0) {
 214                 goto error;
 215         }
 216
 217 error:
 218         return ret;
 219 }
 220
 221 /*
 222  * RELAYD_CREATE_SESSION from 2.1 to 2.3.
 223  */
 224 static int relayd_create_session_2_1(struct lttcomm_relayd_sock *rsock)
 225 {
 226         int ret;
 227
 228         /* Send command */
 229         ret = send_command(rsock, RELAYD_CREATE_SESSION, NULL, 0, 0);
 230         if (ret < 0) {
 231                 goto error;
 232         }
 233
 234 error:
 235         return ret;
 236 }
 237
 238 /*
 239  * Send a RELAYD_CREATE_SESSION command to the relayd with the given socket and
 240  * set session_id of the relayd if we have a successful reply from the relayd.
 241  *
 242  * On success, return 0 else a negative value which is either an errno error or
 243  * a lttng error code from the relayd.
 244  */
 245 int relayd_create_session(struct lttcomm_relayd_sock *rsock,
 246                 uint64_t *relayd_session_id,
 247                 const char *session_name, const char *hostname,
 248                 int session_live_timer,
 249                 unsigned int snapshot, uint64_t sessiond_session_id,
 250                 const lttng_uuid sessiond_uuid,
 251                 const uint64_t *current_chunk_id)
 252 {
 253         int ret;
 254         struct lttcomm_relayd_status_session reply;
 255
 256         assert(rsock);
 257         assert(relayd_session_id);
 258
 259         DBG("Relayd create session");
 260
 261         if (rsock->minor < 4) {
 262                 /* From 2.1 to 2.3 */
 263                 ret = relayd_create_session_2_1(rsock);
 264         } else if (rsock->minor >= 4 && rsock->minor < 11) {
 265                 /* From 2.4 to 2.10 */
 266                 ret = relayd_create_session_2_4(rsock, session_name,
 267                                 hostname, session_live_timer, snapshot);
 268         } else {
 269                 /* From 2.11 to ... */
 270                 ret = relayd_create_session_2_11(rsock, session_name,
 271                                 hostname, session_live_timer, snapshot,
 272                                 sessiond_session_id, sessiond_uuid,
 273                                 current_chunk_id);
 274         }
 275
 276         if (ret < 0) {
 277                 goto error;
 278         }
 279
 280         /* Receive response */
 281         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 282         if (ret < 0) {
 283                 goto error;
 284         }
 285
 286         reply.session_id = be64toh(reply.session_id);
 287         reply.ret_code = be32toh(reply.ret_code);
 288
 289         /* Return session id or negative ret code. */
 290         if (reply.ret_code != LTTNG_OK) {
 291                 ret = -1;
 292                 ERR("Relayd create session replied error %d", reply.ret_code);
 293                 goto error;
 294         } else {
 295                 ret = 0;
 296                 *relayd_session_id = reply.session_id;
 297         }
 298
 299         DBG("Relayd session created with id %" PRIu64, reply.session_id);
 300
 301 error:
 302         return ret;
 303 }
 304
 305 static int relayd_add_stream_2_1(struct lttcomm_relayd_sock *rsock,
 306                 const char *channel_name, const char *pathname)
 307 {
 308         int ret;
 309         struct lttcomm_relayd_add_stream msg;
 310
 311         memset(&msg, 0, sizeof(msg));
 312         if (lttng_strncpy(msg.channel_name, channel_name,
 313                                 sizeof(msg.channel_name))) {
 314                 ret = -1;
 315                 goto error;
 316         }
 317
 318         if (lttng_strncpy(msg.pathname, pathname,
 319                                 sizeof(msg.pathname))) {
 320                 ret = -1;
 321                 goto error;
 322         }
 323
 324         /* Send command */
 325         ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
 326         if (ret < 0) {
 327                 ret = -1;
 328                 goto error;
 329         }
 330         ret = 0;
 331 error:
 332         return ret;
 333 }
 334
 335 static int relayd_add_stream_2_2(struct lttcomm_relayd_sock *rsock,
 336                 const char *channel_name, const char *pathname,
 337                 uint64_t tracefile_size, uint64_t tracefile_count)
 338 {
 339         int ret;
 340         struct lttcomm_relayd_add_stream_2_2 msg;
 341
 342         memset(&msg, 0, sizeof(msg));
 343         /* Compat with relayd 2.2 to 2.10 */
 344         if (lttng_strncpy(msg.channel_name, channel_name,
 345                                 sizeof(msg.channel_name))) {
 346                 ret = -1;
 347                 goto error;
 348         }
 349         if (lttng_strncpy(msg.pathname, pathname,
 350                                 sizeof(msg.pathname))) {
 351                 ret = -1;
 352                 goto error;
 353         }
 354         msg.tracefile_size = htobe64(tracefile_size);
 355         msg.tracefile_count = htobe64(tracefile_count);
 356
 357         /* Send command */
 358         ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
 359         if (ret < 0) {
 360                 goto error;
 361         }
 362         ret = 0;
 363 error:
 364         return ret;
 365 }
 366
 367 static int relayd_add_stream_2_11(struct lttcomm_relayd_sock *rsock,
 368                 const char *channel_name, const char *pathname,
 369                 uint64_t tracefile_size, uint64_t tracefile_count,
 370                 uint64_t trace_archive_id)
 371 {
 372         int ret;
 373         struct lttcomm_relayd_add_stream_2_11 *msg = NULL;
 374         size_t channel_name_len;
 375         size_t pathname_len;
 376         size_t msg_length;
 377
 378         /* The two names are sent with a '\0' delimiter between them. */
 379         channel_name_len = strlen(channel_name) + 1;
 380         pathname_len = strlen(pathname) + 1;
 381
 382         msg_length = sizeof(*msg) + channel_name_len + pathname_len;
 383         msg = zmalloc(msg_length);
 384         if (!msg) {
 385                 PERROR("zmalloc add_stream_2_11 command message");
 386                 ret = -1;
 387                 goto error;
 388         }
 389
 390         assert(channel_name_len <= UINT32_MAX);
 391         msg->channel_name_len = htobe32(channel_name_len);
 392
 393         assert(pathname_len <= UINT32_MAX);
 394         msg->pathname_len = htobe32(pathname_len);
 395
 396         if (lttng_strncpy(msg->names, channel_name, channel_name_len)) {
 397                 ret = -1;
 398                 goto error;
 399         }
 400         if (lttng_strncpy(msg->names + channel_name_len, pathname, pathname_len)) {
 401                 ret = -1;
 402                 goto error;
 403         }
 404
 405         msg->tracefile_size = htobe64(tracefile_size);
 406         msg->tracefile_count = htobe64(tracefile_count);
 407         msg->trace_archive_id = htobe64(trace_archive_id);
 408
 409         /* Send command */
 410         ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) msg, msg_length, 0);
 411         if (ret < 0) {
 412                 goto error;
 413         }
 414         ret = 0;
 415 error:
 416         free(msg);
 417         return ret;
 418 }
 419
 420 /*
 421  * Add stream on the relayd and assign stream handle to the stream_id argument.
 422  *
 423  * On success return 0 else return ret_code negative value.
 424  */
 425 int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_name,
 426                 const char *pathname, uint64_t *stream_id,
 427                 uint64_t tracefile_size, uint64_t tracefile_count,
 428                 struct lttng_trace_chunk *trace_chunk)
 429 {
 430         int ret;
 431         struct lttcomm_relayd_status_stream reply;
 432
 433         /* Code flow error. Safety net. */
 434         assert(rsock);
 435         assert(channel_name);
 436         assert(pathname);
 437
 438         DBG("Relayd adding stream for channel name %s", channel_name);
 439
 440         /* Compat with relayd 2.1 */
 441         if (rsock->minor == 1) {
 442                 /* For 2.1 */
 443                 assert(!trace_chunk);
 444                 ret = relayd_add_stream_2_1(rsock, channel_name, pathname);
 445
 446         } else if (rsock->minor > 1 && rsock->minor < 11) {
 447                 /* From 2.2 to 2.10 */
 448                 assert(!trace_chunk);
 449                 ret = relayd_add_stream_2_2(rsock, channel_name, pathname,
 450                                 tracefile_size, tracefile_count);
 451         } else {
 452                 enum lttng_trace_chunk_status chunk_status;
 453                 uint64_t chunk_id;
 454
 455                 assert(trace_chunk);
 456                 chunk_status = lttng_trace_chunk_get_id(trace_chunk,
 457                                 &chunk_id);
 458                 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
 459
 460                 /* From 2.11 to ...*/
 461                 ret = relayd_add_stream_2_11(rsock, channel_name, pathname,
 462                                 tracefile_size, tracefile_count,
 463                                 chunk_id);
 464         }
 465
 466         if (ret) {
 467                 ret = -1;
 468                 goto error;
 469         }
 470
 471         /* Waiting for reply */
 472         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 473         if (ret < 0) {
 474                 goto error;
 475         }
 476
 477         /* Back to host bytes order. */
 478         reply.handle = be64toh(reply.handle);
 479         reply.ret_code = be32toh(reply.ret_code);
 480
 481         /* Return session id or negative ret code. */
 482         if (reply.ret_code != LTTNG_OK) {
 483                 ret = -1;
 484                 ERR("Relayd add stream replied error %d", reply.ret_code);
 485         } else {
 486                 /* Success */
 487                 ret = 0;
 488                 *stream_id = reply.handle;
 489         }
 490
 491         DBG("Relayd stream added successfully with handle %" PRIu64,
 492                         reply.handle);
 493
 494 error:
 495         return ret;
 496 }
 497
 498 /*
 499  * Inform the relay that all the streams for the current channel has been sent.
 500  *
 501  * On success return 0 else return ret_code negative value.
 502  */
 503 int relayd_streams_sent(struct lttcomm_relayd_sock *rsock)
 504 {
 505         int ret;
 506         struct lttcomm_relayd_generic_reply reply;
 507
 508         /* Code flow error. Safety net. */
 509         assert(rsock);
 510
 511         DBG("Relayd sending streams sent.");
 512
 513         /* This feature was introduced in 2.4, ignore it for earlier versions. */
 514         if (rsock->minor < 4) {
 515                 ret = 0;
 516                 goto end;
 517         }
 518
 519         /* Send command */
 520         ret = send_command(rsock, RELAYD_STREAMS_SENT, NULL, 0, 0);
 521         if (ret < 0) {
 522                 goto error;
 523         }
 524
 525         /* Waiting for reply */
 526         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 527         if (ret < 0) {
 528                 goto error;
 529         }
 530
 531         /* Back to host bytes order. */
 532         reply.ret_code = be32toh(reply.ret_code);
 533
 534         /* Return session id or negative ret code. */
 535         if (reply.ret_code != LTTNG_OK) {
 536                 ret = -1;
 537                 ERR("Relayd streams sent replied error %d", reply.ret_code);
 538                 goto error;
 539         } else {
 540                 /* Success */
 541                 ret = 0;
 542         }
 543
 544         DBG("Relayd streams sent success");
 545
 546 error:
 547 end:
 548         return ret;
 549 }
 550
 551 /*
 552  * Check version numbers on the relayd.
 553  * If major versions are compatible, we assign minor_to_use to the
 554  * minor version of the procotol we are going to use for this session.
 555  *
 556  * Return 0 if the two daemons are compatible, LTTNG_ERR_RELAYD_VERSION_FAIL
 557  * otherwise, or a negative value on network errors.
 558  */
 559 int relayd_version_check(struct lttcomm_relayd_sock *rsock)
 560 {
 561         int ret;
 562         struct lttcomm_relayd_version msg;
 563
 564         /* Code flow error. Safety net. */
 565         assert(rsock);
 566
 567         DBG("Relayd version check for major.minor %u.%u", rsock->major,
 568                         rsock->minor);
 569
 570         memset(&msg, 0, sizeof(msg));
 571         /* Prepare network byte order before transmission. */
 572         msg.major = htobe32(rsock->major);
 573         msg.minor = htobe32(rsock->minor);
 574
 575         /* Send command */
 576         ret = send_command(rsock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0);
 577         if (ret < 0) {
 578                 goto error;
 579         }
 580
 581         /* Receive response */
 582         ret = recv_reply(rsock, (void *) &msg, sizeof(msg));
 583         if (ret < 0) {
 584                 goto error;
 585         }
 586
 587         /* Set back to host bytes order */
 588         msg.major = be32toh(msg.major);
 589         msg.minor = be32toh(msg.minor);
 590
 591         /*
 592          * Only validate the major version. If the other side is higher,
 593          * communication is not possible. Only major version equal can talk to each
 594          * other. If the minor version differs, the lowest version is used by both
 595          * sides.
 596          */
 597         if (msg.major != rsock->major) {
 598                 /* Not compatible */
 599                 ret = LTTNG_ERR_RELAYD_VERSION_FAIL;
 600                 DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)",
 601                                 msg.major, rsock->major);
 602                 goto error;
 603         }
 604
 605         /*
 606          * If the relayd's minor version is higher, it will adapt to our version so
 607          * we can continue to use the latest relayd communication data structure.
 608          * If the received minor version is higher, the relayd should adapt to us.
 609          */
 610         if (rsock->minor > msg.minor) {
 611                 rsock->minor = msg.minor;
 612         }
 613
 614         /* Version number compatible */
 615         DBG2("Relayd version is compatible, using protocol version %u.%u",
 616                         rsock->major, rsock->minor);
 617         ret = 0;
 618
 619 error:
 620         return ret;
 621 }
 622
 623 /*
 624  * Add stream on the relayd and assign stream handle to the stream_id argument.
 625  *
 626  * On success return 0 else return ret_code negative value.
 627  */
 628 int relayd_send_metadata(struct lttcomm_relayd_sock *rsock, size_t len)
 629 {
 630         int ret;
 631
 632         /* Code flow error. Safety net. */
 633         assert(rsock);
 634
 635         DBG("Relayd sending metadata of size %zu", len);
 636
 637         /* Send command */
 638         ret = send_command(rsock, RELAYD_SEND_METADATA, NULL, len, 0);
 639         if (ret < 0) {
 640                 goto error;
 641         }
 642
 643         DBG2("Relayd metadata added successfully");
 644
 645         /*
 646          * After that call, the metadata data MUST be sent to the relayd so the
 647          * receive size on the other end matches the len of the metadata packet
 648          * header. This is why we don't wait for a reply here.
 649          */
 650
 651 error:
 652         return ret;
 653 }
 654
 655 /*
 656  * Connect to relay daemon with an allocated lttcomm_relayd_sock.
 657  */
 658 int relayd_connect(struct lttcomm_relayd_sock *rsock)
 659 {
 660         /* Code flow error. Safety net. */
 661         assert(rsock);
 662
 663         if (!rsock->sock.ops) {
 664                 /*
 665                  * Attempting a connect on a non-initialized socket.
 666                  */
 667                 return -ECONNRESET;
 668         }
 669
 670         DBG3("Relayd connect ...");
 671
 672         return rsock->sock.ops->connect(&rsock->sock);
 673 }
 674
 675 /*
 676  * Close relayd socket with an allocated lttcomm_relayd_sock.
 677  *
 678  * If no socket operations are found, simply return 0 meaning that everything
 679  * is fine. Without operations, the socket can not possibly be opened or used.
 680  * This is possible if the socket was allocated but not created. However, the
 681  * caller could simply use it to store a valid file descriptor for instance
 682  * passed over a Unix socket and call this to cleanup but still without a valid
 683  * ops pointer.
 684  *
 685  * Return the close returned value. On error, a negative value is usually
 686  * returned back from close(2).
 687  */
 688 int relayd_close(struct lttcomm_relayd_sock *rsock)
 689 {
 690         int ret;
 691
 692         /* Code flow error. Safety net. */
 693         assert(rsock);
 694
 695         /* An invalid fd is fine, return success. */
 696         if (rsock->sock.fd < 0) {
 697                 ret = 0;
 698                 goto end;
 699         }
 700
 701         DBG3("Relayd closing socket %d", rsock->sock.fd);
 702
 703         if (rsock->sock.ops) {
 704                 ret = rsock->sock.ops->close(&rsock->sock);
 705         } else {
 706                 /* Default call if no specific ops found. */
 707                 ret = close(rsock->sock.fd);
 708                 if (ret < 0) {
 709                         PERROR("relayd_close default close");
 710                 }
 711         }
 712         rsock->sock.fd = -1;
 713
 714 end:
 715         return ret;
 716 }
 717
 718 /*
 719  * Send data header structure to the relayd.
 720  */
 721 int relayd_send_data_hdr(struct lttcomm_relayd_sock *rsock,
 722                 struct lttcomm_relayd_data_hdr *hdr, size_t size)
 723 {
 724         int ret;
 725
 726         /* Code flow error. Safety net. */
 727         assert(rsock);
 728         assert(hdr);
 729
 730         if (rsock->sock.fd < 0) {
 731                 return -ECONNRESET;
 732         }
 733
 734         DBG3("Relayd sending data header of size %zu", size);
 735
 736         /* Again, safety net */
 737         if (size == 0) {
 738                 size = sizeof(struct lttcomm_relayd_data_hdr);
 739         }
 740
 741         /* Only send data header. */
 742         ret = rsock->sock.ops->sendmsg(&rsock->sock, hdr, size, 0);
 743         if (ret < 0) {
 744                 ret = -errno;
 745                 goto error;
 746         }
 747
 748         /*
 749          * The data MUST be sent right after that command for the receive on the
 750          * other end to match the size in the header.
 751          */
 752
 753 error:
 754         return ret;
 755 }
 756
 757 /*
 758  * Send close stream command to the relayd.
 759  */
 760 int relayd_send_close_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
 761                 uint64_t last_net_seq_num)
 762 {
 763         int ret;
 764         struct lttcomm_relayd_close_stream msg;
 765         struct lttcomm_relayd_generic_reply reply;
 766
 767         /* Code flow error. Safety net. */
 768         assert(rsock);
 769
 770         DBG("Relayd closing stream id %" PRIu64, stream_id);
 771
 772         memset(&msg, 0, sizeof(msg));
 773         msg.stream_id = htobe64(stream_id);
 774         msg.last_net_seq_num = htobe64(last_net_seq_num);
 775
 776         /* Send command */
 777         ret = send_command(rsock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0);
 778         if (ret < 0) {
 779                 goto error;
 780         }
 781
 782         /* Receive response */
 783         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 784         if (ret < 0) {
 785                 goto error;
 786         }
 787
 788         reply.ret_code = be32toh(reply.ret_code);
 789
 790         /* Return session id or negative ret code. */
 791         if (reply.ret_code != LTTNG_OK) {
 792                 ret = -1;
 793                 ERR("Relayd close stream replied error %d", reply.ret_code);
 794         } else {
 795                 /* Success */
 796                 ret = 0;
 797         }
 798
 799         DBG("Relayd close stream id %" PRIu64 " successfully", stream_id);
 800
 801 error:
 802         return ret;
 803 }
 804
 805 /*
 806  * Check for data availability for a given stream id.
 807  *
 808  * Return 0 if NOT pending, 1 if so and a negative value on error.
 809  */
 810 int relayd_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
 811                 uint64_t last_net_seq_num)
 812 {
 813         int ret;
 814         struct lttcomm_relayd_data_pending msg;
 815         struct lttcomm_relayd_generic_reply reply;
 816
 817         /* Code flow error. Safety net. */
 818         assert(rsock);
 819
 820         DBG("Relayd data pending for stream id %" PRIu64, stream_id);
 821
 822         memset(&msg, 0, sizeof(msg));
 823         msg.stream_id = htobe64(stream_id);
 824         msg.last_net_seq_num = htobe64(last_net_seq_num);
 825
 826         /* Send command */
 827         ret = send_command(rsock, RELAYD_DATA_PENDING, (void *) &msg,
 828                         sizeof(msg), 0);
 829         if (ret < 0) {
 830                 goto error;
 831         }
 832
 833         /* Receive response */
 834         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 835         if (ret < 0) {
 836                 goto error;
 837         }
 838
 839         reply.ret_code = be32toh(reply.ret_code);
 840
 841         /* Return session id or negative ret code. */
 842         if (reply.ret_code >= LTTNG_OK) {
 843                 ERR("Relayd data pending replied error %d", reply.ret_code);
 844         }
 845
 846         /* At this point, the ret code is either 1 or 0 */
 847         ret = reply.ret_code;
 848
 849         DBG("Relayd data is %s pending for stream id %" PRIu64,
 850                         ret == 1 ? "" : "NOT", stream_id);
 851
 852 error:
 853         return ret;
 854 }
 855
 856 /*
 857  * Check on the relayd side for a quiescent state on the control socket.
 858  */
 859 int relayd_quiescent_control(struct lttcomm_relayd_sock *rsock,
 860                 uint64_t metadata_stream_id)
 861 {
 862         int ret;
 863         struct lttcomm_relayd_quiescent_control msg;
 864         struct lttcomm_relayd_generic_reply reply;
 865
 866         /* Code flow error. Safety net. */
 867         assert(rsock);
 868
 869         DBG("Relayd checking quiescent control state");
 870
 871         memset(&msg, 0, sizeof(msg));
 872         msg.stream_id = htobe64(metadata_stream_id);
 873
 874         /* Send command */
 875         ret = send_command(rsock, RELAYD_QUIESCENT_CONTROL, &msg, sizeof(msg), 0);
 876         if (ret < 0) {
 877                 goto error;
 878         }
 879
 880         /* Receive response */
 881         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 882         if (ret < 0) {
 883                 goto error;
 884         }
 885
 886         reply.ret_code = be32toh(reply.ret_code);
 887
 888         /* Return session id or negative ret code. */
 889         if (reply.ret_code != LTTNG_OK) {
 890                 ret = -1;
 891                 ERR("Relayd quiescent control replied error %d", reply.ret_code);
 892                 goto error;
 893         }
 894
 895         /* Control socket is quiescent */
 896         return 0;
 897
 898 error:
 899         return ret;
 900 }
 901
 902 /*
 903  * Begin a data pending command for a specific session id.
 904  */
 905 int relayd_begin_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id)
 906 {
 907         int ret;
 908         struct lttcomm_relayd_begin_data_pending msg;
 909         struct lttcomm_relayd_generic_reply reply;
 910
 911         /* Code flow error. Safety net. */
 912         assert(rsock);
 913
 914         DBG("Relayd begin data pending");
 915
 916         memset(&msg, 0, sizeof(msg));
 917         msg.session_id = htobe64(id);
 918
 919         /* Send command */
 920         ret = send_command(rsock, RELAYD_BEGIN_DATA_PENDING, &msg, sizeof(msg), 0);
 921         if (ret < 0) {
 922                 goto error;
 923         }
 924
 925         /* Receive response */
 926         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 927         if (ret < 0) {
 928                 goto error;
 929         }
 930
 931         reply.ret_code = be32toh(reply.ret_code);
 932
 933         /* Return session id or negative ret code. */
 934         if (reply.ret_code != LTTNG_OK) {
 935                 ret = -1;
 936                 ERR("Relayd begin data pending replied error %d", reply.ret_code);
 937                 goto error;
 938         }
 939
 940         return 0;
 941
 942 error:
 943         return ret;
 944 }
 945
 946 /*
 947  * End a data pending command for a specific session id.
 948  *
 949  * Return 0 on success and set is_data_inflight to 0 if no data is being
 950  * streamed or 1 if it is the case.
 951  */
 952 int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id,
 953                 unsigned int *is_data_inflight)
 954 {
 955         int ret, recv_ret;
 956         struct lttcomm_relayd_end_data_pending msg;
 957         struct lttcomm_relayd_generic_reply reply;
 958
 959         /* Code flow error. Safety net. */
 960         assert(rsock);
 961
 962         DBG("Relayd end data pending");
 963
 964         memset(&msg, 0, sizeof(msg));
 965         msg.session_id = htobe64(id);
 966
 967         /* Send command */
 968         ret = send_command(rsock, RELAYD_END_DATA_PENDING, &msg, sizeof(msg), 0);
 969         if (ret < 0) {
 970                 goto error;
 971         }
 972
 973         /* Receive response */
 974         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 975         if (ret < 0) {
 976                 goto error;
 977         }
 978
 979         recv_ret = be32toh(reply.ret_code);
 980         if (recv_ret < 0) {
 981                 ret = recv_ret;
 982                 goto error;
 983         }
 984
 985         *is_data_inflight = recv_ret;
 986
 987         DBG("Relayd end data pending is data inflight: %d", recv_ret);
 988
 989         return 0;
 990
 991 error:
 992         return ret;
 993 }
 994
 995 /*
 996  * Send index to the relayd.
 997  */
 998 int relayd_send_index(struct lttcomm_relayd_sock *rsock,
 999                 struct ctf_packet_index *index, uint64_t relay_stream_id,
1000                 uint64_t net_seq_num)
1001 {
1002         int ret;
1003         struct lttcomm_relayd_index msg;
1004         struct lttcomm_relayd_generic_reply reply;
1005
1006         /* Code flow error. Safety net. */
1007         assert(rsock);
1008
1009         if (rsock->minor < 4) {
1010                 DBG("Not sending indexes before protocol 2.4");
1011                 ret = 0;
1012                 goto error;
1013         }
1014
1015         DBG("Relayd sending index for stream ID %" PRIu64, relay_stream_id);
1016
1017         memset(&msg, 0, sizeof(msg));
1018         msg.relay_stream_id = htobe64(relay_stream_id);
1019         msg.net_seq_num = htobe64(net_seq_num);
1020
1021         /* The index is already in big endian. */
1022         msg.packet_size = index->packet_size;
1023         msg.content_size = index->content_size;
1024         msg.timestamp_begin = index->timestamp_begin;
1025         msg.timestamp_end = index->timestamp_end;
1026         msg.events_discarded = index->events_discarded;
1027         msg.stream_id = index->stream_id;
1028
1029         if (rsock->minor >= 8) {
1030                 msg.stream_instance_id = index->stream_instance_id;
1031                 msg.packet_seq_num = index->packet_seq_num;
1032         }
1033
1034         /* Send command */
1035         ret = send_command(rsock, RELAYD_SEND_INDEX, &msg,
1036                 lttcomm_relayd_index_len(lttng_to_index_major(rsock->major,
1037                                                                 rsock->minor),
1038                                 lttng_to_index_minor(rsock->major, rsock->minor)),
1039                                 0);
1040         if (ret < 0) {
1041                 goto error;
1042         }
1043
1044         /* Receive response */
1045         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
1046         if (ret < 0) {
1047                 goto error;
1048         }
1049
1050         reply.ret_code = be32toh(reply.ret_code);
1051
1052         /* Return session id or negative ret code. */
1053         if (reply.ret_code != LTTNG_OK) {
1054                 ret = -1;
1055                 ERR("Relayd send index replied error %d", reply.ret_code);
1056         } else {
1057                 /* Success */
1058                 ret = 0;
1059         }
1060
1061 error:
1062         return ret;
1063 }
1064
1065 /*
1066  * Ask the relay to reset the metadata trace file (regeneration).
1067  */
1068 int relayd_reset_metadata(struct lttcomm_relayd_sock *rsock,
1069                 uint64_t stream_id, uint64_t version)
1070 {
1071         int ret;
1072         struct lttcomm_relayd_reset_metadata msg;
1073         struct lttcomm_relayd_generic_reply reply;
1074
1075         /* Code flow error. Safety net. */
1076         assert(rsock);
1077
1078         /* Should have been prevented by the sessiond. */
1079         if (rsock->minor < 8) {
1080                 ERR("Metadata regeneration unsupported before 2.8");
1081                 ret = -1;
1082                 goto error;
1083         }
1084
1085         DBG("Relayd reset metadata stream id %" PRIu64, stream_id);
1086
1087         memset(&msg, 0, sizeof(msg));
1088         msg.stream_id = htobe64(stream_id);
1089         msg.version = htobe64(version);
1090
1091         /* Send command */
1092         ret = send_command(rsock, RELAYD_RESET_METADATA, (void *) &msg, sizeof(msg), 0);
1093         if (ret < 0) {
1094                 goto error;
1095         }
1096
1097         /* Receive response */
1098         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
1099         if (ret < 0) {
1100                 goto error;
1101         }
1102
1103         reply.ret_code = be32toh(reply.ret_code);
1104
1105         /* Return session id or negative ret code. */
1106         if (reply.ret_code != LTTNG_OK) {
1107                 ret = -1;
1108                 ERR("Relayd reset metadata replied error %d", reply.ret_code);
1109         } else {
1110                 /* Success */
1111                 ret = 0;
1112         }
1113
1114         DBG("Relayd reset metadata stream id %" PRIu64 " successfully", stream_id);
1115
1116 error:
1117         return ret;
1118 }
1119
1120 int relayd_rotate_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
1121                 uint64_t new_chunk_id, uint64_t seq_num)
1122 {
1123         int ret;
1124         struct lttcomm_relayd_rotate_stream *msg = NULL;
1125         struct lttcomm_relayd_generic_reply reply;
1126         size_t len;
1127         int msg_len;
1128         /* FIXME */
1129         char *new_pathname = NULL;
1130
1131         /* Code flow error. Safety net. */
1132         assert(rsock);
1133
1134         DBG("Sending rotate stream id %" PRIu64 " command to relayd", stream_id);
1135
1136         /* Account for the trailing NULL. */
1137         len = lttng_strnlen(new_pathname, LTTNG_PATH_MAX) + 1;
1138         if (len > LTTNG_PATH_MAX) {
1139                 ERR("Path used in relayd rotate stream command exceeds the maximal allowed length");
1140                 ret = -1;
1141                 goto error;
1142         }
1143
1144         msg_len = offsetof(struct lttcomm_relayd_rotate_stream, new_pathname) + len;
1145         msg = zmalloc(msg_len);
1146         if (!msg) {
1147                 PERROR("Failed to allocate relayd rotate stream command of %d bytes",
1148                                 msg_len);
1149                 ret = -1;
1150                 goto error;
1151         }
1152
1153         if (lttng_strncpy(msg->new_pathname, new_pathname, len)) {
1154                 ret = -1;
1155                 ERR("Failed to copy relayd rotate stream command's new path name");
1156                 goto error;
1157         }
1158
1159         msg->pathname_length = htobe32(len);
1160         msg->stream_id = htobe64(stream_id);
1161         msg->new_chunk_id = htobe64(new_chunk_id);
1162         /*
1163          * The seq_num is invalid for metadata streams, but it is ignored on
1164          * the relay.
1165          */
1166         msg->rotate_at_seq_num = htobe64(seq_num);
1167
1168         /* Send command. */
1169         ret = send_command(rsock, RELAYD_ROTATE_STREAM, (void *) msg, msg_len, 0);
1170         if (ret < 0) {
1171                 ERR("Send rotate command");
1172                 goto error;
1173         }
1174
1175         /* Receive response. */
1176         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
1177         if (ret < 0) {
1178                 ERR("Receive rotate reply");
1179                 goto error;
1180         }
1181
1182         reply.ret_code = be32toh(reply.ret_code);
1183
1184         /* Return session id or negative ret code. */
1185         if (reply.ret_code != LTTNG_OK) {
1186                 ret = -1;
1187                 ERR("Relayd rotate stream replied error %d", reply.ret_code);
1188         } else {
1189                 /* Success. */
1190                 ret = 0;
1191                 DBG("Relayd rotated stream id %" PRIu64 " successfully", stream_id);
1192         }
1193
1194 error:
1195         free(msg);
1196         return ret;
1197 }
1198
1199 int relayd_create_trace_chunk(struct lttcomm_relayd_sock *sock,
1200                 struct lttng_trace_chunk *chunk)
1201 {
1202         int ret = 0;
1203         enum lttng_trace_chunk_status status;
1204         struct lttcomm_relayd_create_trace_chunk msg = {};
1205         struct lttcomm_relayd_generic_reply reply = {};
1206         struct lttng_dynamic_buffer payload;
1207         uint64_t chunk_id;
1208         time_t creation_timestamp;
1209         const char *chunk_name;
1210         size_t chunk_name_length;
1211         bool overriden_name;
1212
1213         lttng_dynamic_buffer_init(&payload);
1214
1215         status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1216         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1217                 ret = -1;
1218                 goto end;
1219         }
1220
1221         status = lttng_trace_chunk_get_creation_timestamp(
1222                         chunk, &creation_timestamp);
1223         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1224                 ret = -1;
1225                 goto end;
1226         }
1227
1228         status = lttng_trace_chunk_get_name(
1229                         chunk, &chunk_name, &overriden_name);
1230         if (status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1231                         status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1232                 ret = -1;
1233                 goto end;
1234         }
1235
1236         chunk_name_length = overriden_name ? (strlen(chunk_name) + 1) : 0;
1237         msg = (typeof(msg)){
1238                 .chunk_id = htobe64(chunk_id),
1239                 .creation_timestamp = htobe64((uint64_t) creation_timestamp),
1240                 .override_name_length = htobe32((uint32_t) chunk_name_length),
1241         };
1242
1243         ret = lttng_dynamic_buffer_append(&payload, &msg, sizeof(msg));
1244         if (ret) {
1245                 goto end;
1246         }
1247         if (chunk_name_length) {
1248                 ret = lttng_dynamic_buffer_append(
1249                                 &payload, chunk_name, chunk_name_length);
1250                 if (ret) {
1251                         goto end;
1252                 }
1253         }
1254
1255         ret = send_command(sock, RELAYD_CREATE_TRACE_CHUNK, payload.data,
1256                         payload.size, 0);
1257         if (ret < 0) {
1258                 ERR("Failed to send trace chunk creation command to relay daemon");
1259                 goto end;
1260         }
1261
1262         ret = recv_reply(sock, &reply, sizeof(reply));
1263         if (ret < 0) {
1264                 ERR("Failed to receive relay daemon trace chunk creation command reply");
1265                 goto end;
1266         }
1267
1268         reply.ret_code = be32toh(reply.ret_code);
1269         if (reply.ret_code != LTTNG_OK) {
1270                 ret = -1;
1271                 ERR("Relayd trace chunk create replied error %d",
1272                                 reply.ret_code);
1273         } else {
1274                 ret = 0;
1275                 DBG("Relayd successfully created trace chunk: chunk_id = %" PRIu64,
1276                                 chunk_id);
1277         }
1278
1279 end:
1280         lttng_dynamic_buffer_reset(&payload);
1281         return ret;
1282 }
1283
1284 int relayd_close_trace_chunk(struct lttcomm_relayd_sock *sock,
1285                 struct lttng_trace_chunk *chunk)
1286 {
1287         int ret = 0;
1288         enum lttng_trace_chunk_status status;
1289         struct lttcomm_relayd_close_trace_chunk msg = {};
1290         struct lttcomm_relayd_generic_reply reply = {};
1291         uint64_t chunk_id;
1292         time_t close_timestamp;
1293         LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command = {};
1294
1295         status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1296         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1297                 ERR("Failed to get trace chunk id");
1298                 ret = -1;
1299                 goto end;
1300         }
1301
1302         status = lttng_trace_chunk_get_close_timestamp(chunk, &close_timestamp);
1303         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1304                 ERR("Failed to get trace chunk close timestamp");
1305                 ret = -1;
1306                 goto end;
1307         }
1308
1309         status = lttng_trace_chunk_get_close_command(chunk,
1310                         &close_command.value);
1311         switch (status) {
1312         case LTTNG_TRACE_CHUNK_STATUS_OK:
1313                 close_command.is_set = 1;
1314                 break;
1315         case LTTNG_TRACE_CHUNK_STATUS_NONE:
1316                 break;
1317         default:
1318                 ERR("Failed to get trace chunk close command");
1319                 ret = -1;
1320                 goto end;
1321         }
1322
1323         msg = (typeof(msg)){
1324                 .chunk_id = htobe64(chunk_id),
1325                 .close_timestamp = htobe64((uint64_t) close_timestamp),
1326                 .close_command = {
1327                         .value = htobe32((uint32_t) close_command.value),
1328                         .is_set = close_command.is_set,
1329                 },
1330         };
1331
1332         ret = send_command(sock, RELAYD_CLOSE_TRACE_CHUNK, &msg, sizeof(msg),
1333                         0);
1334         if (ret < 0) {
1335                 ERR("Failed to send trace chunk close command to relay daemon");
1336                 goto end;
1337         }
1338
1339         ret = recv_reply(sock, &reply, sizeof(reply));
1340         if (ret < 0) {
1341                 ERR("Failed to receive relay daemon trace chunk close command reply");
1342                 goto end;
1343         }
1344
1345         reply.ret_code = be32toh(reply.ret_code);
1346         if (reply.ret_code != LTTNG_OK) {
1347                 ret = -1;
1348                 ERR("Relayd trace chunk close replied error %d",
1349                                 reply.ret_code);
1350         } else {
1351                 ret = 0;
1352                 DBG("Relayd successfully closed trace chunk: chunk_id = %" PRIu64,
1353                                 chunk_id);
1354         }
1355 end:
1356         return ret;
1357 }