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