src/common/relayd/relayd.cpp

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