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