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