src/bin/lttng-sessiond/consumer.c

   1 /*
   2  * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
   3  *
   4  * This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License, version 2 only, as
   6  * published by the Free Software Foundation.
   7  *
   8  * This program is distributed in the hope that it will be useful, but WITHOUT
   9  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  11  * more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along with
  14  * this program; if not, write to the Free Software Foundation, Inc., 51
  15  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16  */
  17
  18 #define _GNU_SOURCE
  19 #include <assert.h>
  20 #include <stdio.h>
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <sys/stat.h>
  24 #include <sys/types.h>
  25 #include <unistd.h>
  26 #include <inttypes.h>
  27
  28 #include <common/common.h>
  29 #include <common/defaults.h>
  30 #include <common/uri.h>
  31
  32 #include "consumer.h"
  33 #include "health.h"
  34 #include "ust-app.h"
  35 #include "utils.h"
  36
  37 /*
  38  * Send a data payload using a given consumer socket of size len.
  39  *
  40  * The consumer socket lock MUST be acquired before calling this since this
  41  * function can change the fd value.
  42  *
  43  * Return 0 on success else a negative value on error.
  44  */
  45 int consumer_socket_send(struct consumer_socket *socket, void *msg, size_t len)
  46 {
  47         int fd;
  48         ssize_t size;
  49
  50         assert(socket);
  51         assert(socket->fd_ptr);
  52         assert(msg);
  53
  54         /* Consumer socket is invalid. Stopping. */
  55         fd = *socket->fd_ptr;
  56         if (fd < 0) {
  57                 goto error;
  58         }
  59
  60         size = lttcomm_send_unix_sock(fd, msg, len);
  61         if (size < 0) {
  62                 /* The above call will print a PERROR on error. */
  63                 DBG("Error when sending data to consumer on sock %d", fd);
  64                 /*
  65                  * At this point, the socket is not usable anymore thus closing it and
  66                  * setting the file descriptor to -1 so it is not reused.
  67                  */
  68
  69                 /* This call will PERROR on error. */
  70                 (void) lttcomm_close_unix_sock(fd);
  71                 *socket->fd_ptr = -1;
  72                 goto error;
  73         }
  74
  75         return 0;
  76
  77 error:
  78         return -1;
  79 }
  80
  81 /*
  82  * Receive a data payload using a given consumer socket of size len.
  83  *
  84  * The consumer socket lock MUST be acquired before calling this since this
  85  * function can change the fd value.
  86  *
  87  * Return 0 on success else a negative value on error.
  88  */
  89 int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len)
  90 {
  91         int fd;
  92         ssize_t size;
  93
  94         assert(socket);
  95         assert(socket->fd_ptr);
  96         assert(msg);
  97
  98         /* Consumer socket is invalid. Stopping. */
  99         fd = *socket->fd_ptr;
 100         if (fd < 0) {
 101                 goto error;
 102         }
 103
 104         size = lttcomm_recv_unix_sock(fd, msg, len);
 105         if (size <= 0) {
 106                 /* The above call will print a PERROR on error. */
 107                 DBG("Error when receiving data from the consumer socket %d", fd);
 108                 /*
 109                  * At this point, the socket is not usable anymore thus closing it and
 110                  * setting the file descriptor to -1 so it is not reused.
 111                  */
 112
 113                 /* This call will PERROR on error. */
 114                 (void) lttcomm_close_unix_sock(fd);
 115                 *socket->fd_ptr = -1;
 116                 goto error;
 117         }
 118
 119         return 0;
 120
 121 error:
 122         return -1;
 123 }
 124
 125 /*
 126  * Receive a reply command status message from the consumer. Consumer socket
 127  * lock MUST be acquired before calling this function.
 128  *
 129  * Return 0 on success, -1 on recv error or a negative lttng error code which
 130  * was possibly returned by the consumer.
 131  */
 132 int consumer_recv_status_reply(struct consumer_socket *sock)
 133 {
 134         int ret;
 135         struct lttcomm_consumer_status_msg reply;
 136
 137         assert(sock);
 138
 139         ret = consumer_socket_recv(sock, &reply, sizeof(reply));
 140         if (ret < 0) {
 141                 goto end;
 142         }
 143
 144         if (reply.ret_code == LTTCOMM_CONSUMERD_SUCCESS) {
 145                 /* All good. */
 146                 ret = 0;
 147         } else {
 148                 ret = -reply.ret_code;
 149                 DBG("Consumer ret code %d", ret);
 150         }
 151
 152 end:
 153         return ret;
 154 }
 155
 156 /*
 157  * Once the ASK_CHANNEL command is sent to the consumer, the channel
 158  * information are sent back. This call receives that data and populates key
 159  * and stream_count.
 160  *
 161  * On success return 0 and both key and stream_count are set. On error, a
 162  * negative value is sent back and both parameters are untouched.
 163  */
 164 int consumer_recv_status_channel(struct consumer_socket *sock,
 165                 uint64_t *key, unsigned int *stream_count)
 166 {
 167         int ret;
 168         struct lttcomm_consumer_status_channel reply;
 169
 170         assert(sock);
 171         assert(stream_count);
 172         assert(key);
 173
 174         ret = consumer_socket_recv(sock, &reply, sizeof(reply));
 175         if (ret < 0) {
 176                 goto end;
 177         }
 178
 179         /* An error is possible so don't touch the key and stream_count. */
 180         if (reply.ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
 181                 ret = -1;
 182                 goto end;
 183         }
 184
 185         *key = reply.key;
 186         *stream_count = reply.stream_count;
 187         ret = 0;
 188
 189 end:
 190         return ret;
 191 }
 192
 193 /*
 194  * Send destroy relayd command to consumer.
 195  *
 196  * On success return positive value. On error, negative value.
 197  */
 198 int consumer_send_destroy_relayd(struct consumer_socket *sock,
 199                 struct consumer_output *consumer)
 200 {
 201         int ret;
 202         struct lttcomm_consumer_msg msg;
 203
 204         assert(consumer);
 205         assert(sock);
 206
 207         DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr);
 208
 209         msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
 210         msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
 211
 212         pthread_mutex_lock(sock->lock);
 213         ret = consumer_socket_send(sock, &msg, sizeof(msg));
 214         if (ret < 0) {
 215                 goto error;
 216         }
 217
 218         /* Don't check the return value. The caller will do it. */
 219         ret = consumer_recv_status_reply(sock);
 220
 221         DBG2("Consumer send destroy relayd command done");
 222
 223 error:
 224         pthread_mutex_unlock(sock->lock);
 225         return ret;
 226 }
 227
 228 /*
 229  * For each consumer socket in the consumer output object, send a destroy
 230  * relayd command.
 231  */
 232 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
 233 {
 234         struct lttng_ht_iter iter;
 235         struct consumer_socket *socket;
 236
 237         assert(consumer);
 238
 239         /* Destroy any relayd connection */
 240         if (consumer->type == CONSUMER_DST_NET) {
 241                 rcu_read_lock();
 242                 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
 243                                 node.node) {
 244                         int ret;
 245
 246                         /* Send destroy relayd command */
 247                         ret = consumer_send_destroy_relayd(socket, consumer);
 248                         if (ret < 0) {
 249                                 DBG("Unable to send destroy relayd command to consumer");
 250                                 /* Continue since we MUST delete everything at this point. */
 251                         }
 252                 }
 253                 rcu_read_unlock();
 254         }
 255 }
 256
 257 /*
 258  * From a consumer_data structure, allocate and add a consumer socket to the
 259  * consumer output.
 260  *
 261  * Return 0 on success, else negative value on error
 262  */
 263 int consumer_create_socket(struct consumer_data *data,
 264                 struct consumer_output *output)
 265 {
 266         int ret = 0;
 267         struct consumer_socket *socket;
 268
 269         assert(data);
 270
 271         if (output == NULL || data->cmd_sock < 0) {
 272                 /*
 273                  * Not an error. Possible there is simply not spawned consumer or it's
 274                  * disabled for the tracing session asking the socket.
 275                  */
 276                 goto error;
 277         }
 278
 279         rcu_read_lock();
 280         socket = consumer_find_socket(data->cmd_sock, output);
 281         rcu_read_unlock();
 282         if (socket == NULL) {
 283                 socket = consumer_allocate_socket(&data->cmd_sock);
 284                 if (socket == NULL) {
 285                         ret = -1;
 286                         goto error;
 287                 }
 288
 289                 socket->registered = 0;
 290                 socket->lock = &data->lock;
 291                 rcu_read_lock();
 292                 consumer_add_socket(socket, output);
 293                 rcu_read_unlock();
 294         }
 295
 296         socket->type = data->type;
 297
 298         DBG3("Consumer socket created (fd: %d) and added to output",
 299                         data->cmd_sock);
 300
 301 error:
 302         return ret;
 303 }
 304
 305 /*
 306  * Return the consumer socket from the given consumer output with the right
 307  * bitness. On error, returns NULL.
 308  *
 309  * The caller MUST acquire a rcu read side lock and keep it until the socket
 310  * object reference is not needed anymore.
 311  */
 312 struct consumer_socket *consumer_find_socket_by_bitness(int bits,
 313                 struct consumer_output *consumer)
 314 {
 315         int consumer_fd;
 316         struct consumer_socket *socket = NULL;
 317
 318         switch (bits) {
 319         case 64:
 320                 consumer_fd = uatomic_read(&ust_consumerd64_fd);
 321                 break;
 322         case 32:
 323                 consumer_fd = uatomic_read(&ust_consumerd32_fd);
 324                 break;
 325         default:
 326                 assert(0);
 327                 goto end;
 328         }
 329
 330         socket = consumer_find_socket(consumer_fd, consumer);
 331         if (!socket) {
 332                 ERR("Consumer socket fd %d not found in consumer obj %p",
 333                                 consumer_fd, consumer);
 334         }
 335
 336 end:
 337         return socket;
 338 }
 339
 340 /*
 341  * Find a consumer_socket in a consumer_output hashtable. Read side lock must
 342  * be acquired before calling this function and across use of the
 343  * returned consumer_socket.
 344  */
 345 struct consumer_socket *consumer_find_socket(int key,
 346                 struct consumer_output *consumer)
 347 {
 348         struct lttng_ht_iter iter;
 349         struct lttng_ht_node_ulong *node;
 350         struct consumer_socket *socket = NULL;
 351
 352         /* Negative keys are lookup failures */
 353         if (key < 0 || consumer == NULL) {
 354                 return NULL;
 355         }
 356
 357         lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
 358                         &iter);
 359         node = lttng_ht_iter_get_node_ulong(&iter);
 360         if (node != NULL) {
 361                 socket = caa_container_of(node, struct consumer_socket, node);
 362         }
 363
 364         return socket;
 365 }
 366
 367 /*
 368  * Allocate a new consumer_socket and return the pointer.
 369  */
 370 struct consumer_socket *consumer_allocate_socket(int *fd)
 371 {
 372         struct consumer_socket *socket = NULL;
 373
 374         assert(fd);
 375
 376         socket = zmalloc(sizeof(struct consumer_socket));
 377         if (socket == NULL) {
 378                 PERROR("zmalloc consumer socket");
 379                 goto error;
 380         }
 381
 382         socket->fd_ptr = fd;
 383         lttng_ht_node_init_ulong(&socket->node, *fd);
 384
 385 error:
 386         return socket;
 387 }
 388
 389 /*
 390  * Add consumer socket to consumer output object. Read side lock must be
 391  * acquired before calling this function.
 392  */
 393 void consumer_add_socket(struct consumer_socket *sock,
 394                 struct consumer_output *consumer)
 395 {
 396         assert(sock);
 397         assert(consumer);
 398
 399         lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
 400 }
 401
 402 /*
 403  * Delte consumer socket to consumer output object. Read side lock must be
 404  * acquired before calling this function.
 405  */
 406 void consumer_del_socket(struct consumer_socket *sock,
 407                 struct consumer_output *consumer)
 408 {
 409         int ret;
 410         struct lttng_ht_iter iter;
 411
 412         assert(sock);
 413         assert(consumer);
 414
 415         iter.iter.node = &sock->node.node;
 416         ret = lttng_ht_del(consumer->socks, &iter);
 417         assert(!ret);
 418 }
 419
 420 /*
 421  * RCU destroy call function.
 422  */
 423 static void destroy_socket_rcu(struct rcu_head *head)
 424 {
 425         struct lttng_ht_node_ulong *node =
 426                 caa_container_of(head, struct lttng_ht_node_ulong, head);
 427         struct consumer_socket *socket =
 428                 caa_container_of(node, struct consumer_socket, node);
 429
 430         free(socket);
 431 }
 432
 433 /*
 434  * Destroy and free socket pointer in a call RCU. Read side lock must be
 435  * acquired before calling this function.
 436  */
 437 void consumer_destroy_socket(struct consumer_socket *sock)
 438 {
 439         assert(sock);
 440
 441         /*
 442          * We DO NOT close the file descriptor here since it is global to the
 443          * session daemon and is closed only if the consumer dies or a custom
 444          * consumer was registered,
 445          */
 446         if (sock->registered) {
 447                 DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
 448                 lttcomm_close_unix_sock(*sock->fd_ptr);
 449         }
 450
 451         call_rcu(&sock->node.head, destroy_socket_rcu);
 452 }
 453
 454 /*
 455  * Allocate and assign data to a consumer_output object.
 456  *
 457  * Return pointer to structure.
 458  */
 459 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
 460 {
 461         struct consumer_output *output = NULL;
 462
 463         output = zmalloc(sizeof(struct consumer_output));
 464         if (output == NULL) {
 465                 PERROR("zmalloc consumer_output");
 466                 goto error;
 467         }
 468
 469         /* By default, consumer output is enabled */
 470         output->enabled = 1;
 471         output->type = type;
 472         output->net_seq_index = (uint64_t) -1ULL;
 473
 474         output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 475
 476 error:
 477         return output;
 478 }
 479
 480 /*
 481  * Iterate over the consumer output socket hash table and destroy them. The
 482  * socket file descriptor are only closed if the consumer output was
 483  * registered meaning it's an external consumer.
 484  */
 485 void consumer_destroy_output_sockets(struct consumer_output *obj)
 486 {
 487         struct lttng_ht_iter iter;
 488         struct consumer_socket *socket;
 489
 490         if (!obj->socks) {
 491                 return;
 492         }
 493
 494         rcu_read_lock();
 495         cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
 496                 consumer_del_socket(socket, obj);
 497                 consumer_destroy_socket(socket);
 498         }
 499         rcu_read_unlock();
 500 }
 501
 502 /*
 503  * Delete the consumer_output object from the list and free the ptr.
 504  *
 505  * Should *NOT* be called with RCU read-side lock held.
 506  */
 507 void consumer_destroy_output(struct consumer_output *obj)
 508 {
 509         if (obj == NULL) {
 510                 return;
 511         }
 512
 513         consumer_destroy_output_sockets(obj);
 514
 515         if (obj->socks) {
 516                 /* Finally destroy HT */
 517                 ht_cleanup_push(obj->socks);
 518         }
 519
 520         free(obj);
 521 }
 522
 523 /*
 524  * Copy consumer output and returned the newly allocated copy.
 525  *
 526  * Should *NOT* be called with RCU read-side lock held.
 527  */
 528 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
 529 {
 530         int ret;
 531         struct lttng_ht *tmp_ht_ptr;
 532         struct consumer_output *output;
 533
 534         assert(obj);
 535
 536         output = consumer_create_output(obj->type);
 537         if (output == NULL) {
 538                 goto error;
 539         }
 540         /* Avoid losing the HT reference after the memcpy() */
 541         tmp_ht_ptr = output->socks;
 542
 543         memcpy(output, obj, sizeof(struct consumer_output));
 544
 545         /* Putting back the HT pointer and start copying socket(s). */
 546         output->socks = tmp_ht_ptr;
 547
 548         ret = consumer_copy_sockets(output, obj);
 549         if (ret < 0) {
 550                 goto malloc_error;
 551         }
 552
 553 error:
 554         return output;
 555
 556 malloc_error:
 557         consumer_destroy_output(output);
 558         return NULL;
 559 }
 560
 561 /*
 562  * Copy consumer sockets from src to dst.
 563  *
 564  * Return 0 on success or else a negative value.
 565  */
 566 int consumer_copy_sockets(struct consumer_output *dst,
 567                 struct consumer_output *src)
 568 {
 569         int ret = 0;
 570         struct lttng_ht_iter iter;
 571         struct consumer_socket *socket, *copy_sock;
 572
 573         assert(dst);
 574         assert(src);
 575
 576         rcu_read_lock();
 577         cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
 578                 /* Ignore socket that are already there. */
 579                 copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
 580                 if (copy_sock) {
 581                         continue;
 582                 }
 583
 584                 /* Create new socket object. */
 585                 copy_sock = consumer_allocate_socket(socket->fd_ptr);
 586                 if (copy_sock == NULL) {
 587                         rcu_read_unlock();
 588                         ret = -ENOMEM;
 589                         goto error;
 590                 }
 591
 592                 copy_sock->registered = socket->registered;
 593                 /*
 594                  * This is valid because this lock is shared accross all consumer
 595                  * object being the global lock of the consumer data structure of the
 596                  * session daemon.
 597                  */
 598                 copy_sock->lock = socket->lock;
 599                 consumer_add_socket(copy_sock, dst);
 600         }
 601         rcu_read_unlock();
 602
 603 error:
 604         return ret;
 605 }
 606
 607 /*
 608  * Set network URI to the consumer output object.
 609  *
 610  * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
 611  * error.
 612  */
 613 int consumer_set_network_uri(struct consumer_output *obj,
 614                 struct lttng_uri *uri)
 615 {
 616         int ret;
 617         char tmp_path[PATH_MAX];
 618         char hostname[HOST_NAME_MAX];
 619         struct lttng_uri *dst_uri = NULL;
 620
 621         /* Code flow error safety net. */
 622         assert(obj);
 623         assert(uri);
 624
 625         switch (uri->stype) {
 626         case LTTNG_STREAM_CONTROL:
 627                 dst_uri = &obj->dst.net.control;
 628                 obj->dst.net.control_isset = 1;
 629                 if (uri->port == 0) {
 630                         /* Assign default port. */
 631                         uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 632                 } else {
 633                         if (obj->dst.net.data_isset && uri->port ==
 634                                         obj->dst.net.data.port) {
 635                                 ret = -LTTNG_ERR_INVALID;
 636                                 goto error;
 637                         }
 638                 }
 639                 DBG3("Consumer control URI set with port %d", uri->port);
 640                 break;
 641         case LTTNG_STREAM_DATA:
 642                 dst_uri = &obj->dst.net.data;
 643                 obj->dst.net.data_isset = 1;
 644                 if (uri->port == 0) {
 645                         /* Assign default port. */
 646                         uri->port = DEFAULT_NETWORK_DATA_PORT;
 647                 } else {
 648                         if (obj->dst.net.control_isset && uri->port ==
 649                                         obj->dst.net.control.port) {
 650                                 ret = -LTTNG_ERR_INVALID;
 651                                 goto error;
 652                         }
 653                 }
 654                 DBG3("Consumer data URI set with port %d", uri->port);
 655                 break;
 656         default:
 657                 ERR("Set network uri type unknown %d", uri->stype);
 658                 ret = -LTTNG_ERR_INVALID;
 659                 goto error;
 660         }
 661
 662         ret = uri_compare(dst_uri, uri);
 663         if (!ret) {
 664                 /* Same URI, don't touch it and return success. */
 665                 DBG3("URI network compare are the same");
 666                 goto equal;
 667         }
 668
 669         /* URIs were not equal, replacing it. */
 670         memset(dst_uri, 0, sizeof(struct lttng_uri));
 671         memcpy(dst_uri, uri, sizeof(struct lttng_uri));
 672         obj->type = CONSUMER_DST_NET;
 673
 674         /* Handle subdir and add hostname in front. */
 675         if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
 676                 /* Get hostname to append it in the pathname */
 677                 ret = gethostname(hostname, sizeof(hostname));
 678                 if (ret < 0) {
 679                         PERROR("gethostname. Fallback on default localhost");
 680                         strncpy(hostname, "localhost", sizeof(hostname));
 681                 }
 682                 hostname[sizeof(hostname) - 1] = '\0';
 683
 684                 /* Setup consumer subdir if none present in the control URI */
 685                 if (strlen(dst_uri->subdir) == 0) {
 686                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 687                                         hostname, obj->subdir);
 688                 } else {
 689                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 690                                         hostname, dst_uri->subdir);
 691                 }
 692                 if (ret < 0) {
 693                         PERROR("snprintf set consumer uri subdir");
 694                         ret = -LTTNG_ERR_NOMEM;
 695                         goto error;
 696                 }
 697
 698                 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
 699                 DBG3("Consumer set network uri subdir path %s", tmp_path);
 700         }
 701
 702         return 0;
 703 equal:
 704         return 1;
 705 error:
 706         return ret;
 707 }
 708
 709 /*
 710  * Send file descriptor to consumer via sock.
 711  */
 712 int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
 713 {
 714         int ret;
 715
 716         assert(fds);
 717         assert(sock);
 718         assert(nb_fd > 0);
 719
 720         ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
 721         if (ret < 0) {
 722                 /* The above call will print a PERROR on error. */
 723                 DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
 724                 goto error;
 725         }
 726
 727         ret = consumer_recv_status_reply(sock);
 728
 729 error:
 730         return ret;
 731 }
 732
 733 /*
 734  * Consumer send communication message structure to consumer.
 735  */
 736 int consumer_send_msg(struct consumer_socket *sock,
 737                 struct lttcomm_consumer_msg *msg)
 738 {
 739         int ret;
 740
 741         assert(msg);
 742         assert(sock);
 743
 744         ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
 745         if (ret < 0) {
 746                 goto error;
 747         }
 748
 749         ret = consumer_recv_status_reply(sock);
 750
 751 error:
 752         return ret;
 753 }
 754
 755 /*
 756  * Consumer send channel communication message structure to consumer.
 757  */
 758 int consumer_send_channel(struct consumer_socket *sock,
 759                 struct lttcomm_consumer_msg *msg)
 760 {
 761         int ret;
 762
 763         assert(msg);
 764         assert(sock);
 765
 766         ret = consumer_send_msg(sock, msg);
 767         if (ret < 0) {
 768                 goto error;
 769         }
 770
 771 error:
 772         return ret;
 773 }
 774
 775 /*
 776  * Populate the given consumer msg structure with the ask_channel command
 777  * information.
 778  */
 779 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 780                 uint64_t subbuf_size,
 781                 uint64_t num_subbuf,
 782                 int overwrite,
 783                 unsigned int switch_timer_interval,
 784                 unsigned int read_timer_interval,
 785                 int output,
 786                 int type,
 787                 uint64_t session_id,
 788                 const char *pathname,
 789                 const char *name,
 790                 uid_t uid,
 791                 gid_t gid,
 792                 uint64_t relayd_id,
 793                 uint64_t key,
 794                 unsigned char *uuid,
 795                 uint32_t chan_id,
 796                 uint64_t tracefile_size,
 797                 uint64_t tracefile_count,
 798                 uint64_t session_id_per_pid,
 799                 unsigned int monitor,
 800                 uint32_t ust_app_uid)
 801 {
 802         assert(msg);
 803
 804         /* Zeroed structure */
 805         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 806
 807         msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
 808         msg->u.ask_channel.subbuf_size = subbuf_size;
 809         msg->u.ask_channel.num_subbuf = num_subbuf ;
 810         msg->u.ask_channel.overwrite = overwrite;
 811         msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
 812         msg->u.ask_channel.read_timer_interval = read_timer_interval;
 813         msg->u.ask_channel.output = output;
 814         msg->u.ask_channel.type = type;
 815         msg->u.ask_channel.session_id = session_id;
 816         msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
 817         msg->u.ask_channel.uid = uid;
 818         msg->u.ask_channel.gid = gid;
 819         msg->u.ask_channel.relayd_id = relayd_id;
 820         msg->u.ask_channel.key = key;
 821         msg->u.ask_channel.chan_id = chan_id;
 822         msg->u.ask_channel.tracefile_size = tracefile_size;
 823         msg->u.ask_channel.tracefile_count = tracefile_count;
 824         msg->u.ask_channel.monitor = monitor;
 825         msg->u.ask_channel.ust_app_uid = ust_app_uid;
 826
 827         memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
 828
 829         if (pathname) {
 830                 strncpy(msg->u.ask_channel.pathname, pathname,
 831                                 sizeof(msg->u.ask_channel.pathname));
 832                 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
 833         }
 834
 835         strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 836         msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
 837 }
 838
 839 /*
 840  * Init channel communication message structure.
 841  */
 842 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 843                 enum lttng_consumer_command cmd,
 844                 uint64_t channel_key,
 845                 uint64_t session_id,
 846                 const char *pathname,
 847                 uid_t uid,
 848                 gid_t gid,
 849                 uint64_t relayd_id,
 850                 const char *name,
 851                 unsigned int nb_init_streams,
 852                 enum lttng_event_output output,
 853                 int type,
 854                 uint64_t tracefile_size,
 855                 uint64_t tracefile_count,
 856                 unsigned int monitor)
 857 {
 858         assert(msg);
 859
 860         /* Zeroed structure */
 861         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 862
 863         /* Send channel */
 864         msg->cmd_type = cmd;
 865         msg->u.channel.channel_key = channel_key;
 866         msg->u.channel.session_id = session_id;
 867         msg->u.channel.uid = uid;
 868         msg->u.channel.gid = gid;
 869         msg->u.channel.relayd_id = relayd_id;
 870         msg->u.channel.nb_init_streams = nb_init_streams;
 871         msg->u.channel.output = output;
 872         msg->u.channel.type = type;
 873         msg->u.channel.tracefile_size = tracefile_size;
 874         msg->u.channel.tracefile_count = tracefile_count;
 875         msg->u.channel.monitor = monitor;
 876
 877         strncpy(msg->u.channel.pathname, pathname,
 878                         sizeof(msg->u.channel.pathname));
 879         msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
 880
 881         strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
 882         msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
 883 }
 884
 885 /*
 886  * Init stream communication message structure.
 887  */
 888 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
 889                 enum lttng_consumer_command cmd,
 890                 uint64_t channel_key,
 891                 uint64_t stream_key,
 892                 int cpu)
 893 {
 894         assert(msg);
 895
 896         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 897
 898         msg->cmd_type = cmd;
 899         msg->u.stream.channel_key = channel_key;
 900         msg->u.stream.stream_key = stream_key;
 901         msg->u.stream.cpu = cpu;
 902 }
 903
 904 /*
 905  * Send stream communication structure to the consumer.
 906  */
 907 int consumer_send_stream(struct consumer_socket *sock,
 908                 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
 909                 int *fds, size_t nb_fd)
 910 {
 911         int ret;
 912
 913         assert(msg);
 914         assert(dst);
 915         assert(sock);
 916         assert(fds);
 917
 918         ret = consumer_send_msg(sock, msg);
 919         if (ret < 0) {
 920                 goto error;
 921         }
 922
 923         ret = consumer_send_fds(sock, fds, nb_fd);
 924         if (ret < 0) {
 925                 goto error;
 926         }
 927
 928 error:
 929         return ret;
 930 }
 931
 932 /*
 933  * Send relayd socket to consumer associated with a session name.
 934  *
 935  * On success return positive value. On error, negative value.
 936  */
 937 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 938                 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
 939                 enum lttng_stream_type type, uint64_t session_id)
 940 {
 941         int ret;
 942         struct lttcomm_consumer_msg msg;
 943
 944         /* Code flow error. Safety net. */
 945         assert(rsock);
 946         assert(consumer);
 947         assert(consumer_sock);
 948
 949         /* Bail out if consumer is disabled */
 950         if (!consumer->enabled) {
 951                 ret = LTTNG_OK;
 952                 goto error;
 953         }
 954
 955         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
 956         /*
 957          * Assign network consumer output index using the temporary consumer since
 958          * this call should only be made from within a set_consumer_uri() function
 959          * call in the session daemon.
 960          */
 961         msg.u.relayd_sock.net_index = consumer->net_seq_index;
 962         msg.u.relayd_sock.type = type;
 963         msg.u.relayd_sock.session_id = session_id;
 964         memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
 965
 966         DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
 967         ret = consumer_send_msg(consumer_sock, &msg);
 968         if (ret < 0) {
 969                 goto error;
 970         }
 971
 972         DBG3("Sending relayd socket file descriptor to consumer");
 973         ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
 974         if (ret < 0) {
 975                 goto error;
 976         }
 977
 978         DBG2("Consumer relayd socket sent");
 979
 980 error:
 981         return ret;
 982 }
 983
 984 /*
 985  * Set consumer subdirectory using the session name and a generated datetime if
 986  * needed. This is appended to the current subdirectory.
 987  */
 988 int consumer_set_subdir(struct consumer_output *consumer,
 989                 const char *session_name)
 990 {
 991         int ret = 0;
 992         unsigned int have_default_name = 0;
 993         char datetime[16], tmp_path[PATH_MAX];
 994         time_t rawtime;
 995         struct tm *timeinfo;
 996
 997         assert(consumer);
 998         assert(session_name);
 999
1000         memset(tmp_path, 0, sizeof(tmp_path));
1001
1002         /* Flag if we have a default session. */
1003         if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
1004                                 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
1005                 have_default_name = 1;
1006         } else {
1007                 /* Get date and time for session path */
1008                 time(&rawtime);
1009                 timeinfo = localtime(&rawtime);
1010                 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
1011         }
1012
1013         if (have_default_name) {
1014                 ret = snprintf(tmp_path, sizeof(tmp_path),
1015                                 "%s/%s", consumer->subdir, session_name);
1016         } else {
1017                 ret = snprintf(tmp_path, sizeof(tmp_path),
1018                                 "%s/%s-%s/", consumer->subdir, session_name, datetime);
1019         }
1020         if (ret < 0) {
1021                 PERROR("snprintf session name date");
1022                 goto error;
1023         }
1024
1025         strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
1026         DBG2("Consumer subdir set to %s", consumer->subdir);
1027
1028 error:
1029         return ret;
1030 }
1031
1032 /*
1033  * Ask the consumer if the data is ready to read (NOT pending) for the specific
1034  * session id.
1035  *
1036  * This function has a different behavior with the consumer i.e. that it waits
1037  * for a reply from the consumer if yes or no the data is pending.
1038  */
1039 int consumer_is_data_pending(uint64_t session_id,
1040                 struct consumer_output *consumer)
1041 {
1042         int ret;
1043         int32_t ret_code = 0;  /* Default is that the data is NOT pending */
1044         struct consumer_socket *socket;
1045         struct lttng_ht_iter iter;
1046         struct lttcomm_consumer_msg msg;
1047
1048         assert(consumer);
1049
1050         msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
1051
1052         msg.u.data_pending.session_id = session_id;
1053
1054         DBG3("Consumer data pending for id %" PRIu64, session_id);
1055
1056         /* Send command for each consumer */
1057         rcu_read_lock();
1058         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1059                         node.node) {
1060                 pthread_mutex_lock(socket->lock);
1061                 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1062                 if (ret < 0) {
1063                         pthread_mutex_unlock(socket->lock);
1064                         goto error_unlock;
1065                 }
1066
1067                 /*
1068                  * No need for a recv reply status because the answer to the command is
1069                  * the reply status message.
1070                  */
1071
1072                 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1073                 if (ret < 0) {
1074                         pthread_mutex_unlock(socket->lock);
1075                         goto error_unlock;
1076                 }
1077                 pthread_mutex_unlock(socket->lock);
1078
1079                 if (ret_code == 1) {
1080                         break;
1081                 }
1082         }
1083         rcu_read_unlock();
1084
1085         DBG("Consumer data is %s pending for session id %" PRIu64,
1086                         ret_code == 1 ? "" : "NOT", session_id);
1087         return ret_code;
1088
1089 error_unlock:
1090         rcu_read_unlock();
1091         return -1;
1092 }
1093
1094 /*
1095  * Send a flush command to consumer using the given channel key.
1096  *
1097  * Return 0 on success else a negative value.
1098  */
1099 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1100 {
1101         int ret;
1102         struct lttcomm_consumer_msg msg;
1103
1104         assert(socket);
1105
1106         DBG2("Consumer flush channel key %" PRIu64, key);
1107
1108         msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1109         msg.u.flush_channel.key = key;
1110
1111         pthread_mutex_lock(socket->lock);
1112         health_code_update();
1113
1114         ret = consumer_send_msg(socket, &msg);
1115         if (ret < 0) {
1116                 goto end;
1117         }
1118
1119 end:
1120         health_code_update();
1121         pthread_mutex_unlock(socket->lock);
1122         return ret;
1123 }
1124
1125 /*
1126  * Send a close metdata command to consumer using the given channel key.
1127  *
1128  * Return 0 on success else a negative value.
1129  */
1130 int consumer_close_metadata(struct consumer_socket *socket,
1131                 uint64_t metadata_key)
1132 {
1133         int ret;
1134         struct lttcomm_consumer_msg msg;
1135
1136         assert(socket);
1137
1138         DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1139
1140         msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1141         msg.u.close_metadata.key = metadata_key;
1142
1143         pthread_mutex_lock(socket->lock);
1144         health_code_update();
1145
1146         ret = consumer_send_msg(socket, &msg);
1147         if (ret < 0) {
1148                 goto end;
1149         }
1150
1151 end:
1152         health_code_update();
1153         pthread_mutex_unlock(socket->lock);
1154         return ret;
1155 }
1156
1157 /*
1158  * Send a setup metdata command to consumer using the given channel key.
1159  *
1160  * Return 0 on success else a negative value.
1161  */
1162 int consumer_setup_metadata(struct consumer_socket *socket,
1163                 uint64_t metadata_key)
1164 {
1165         int ret;
1166         struct lttcomm_consumer_msg msg;
1167
1168         assert(socket);
1169
1170         DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1171
1172         msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1173         msg.u.setup_metadata.key = metadata_key;
1174
1175         pthread_mutex_lock(socket->lock);
1176         health_code_update();
1177
1178         ret = consumer_send_msg(socket, &msg);
1179         if (ret < 0) {
1180                 goto end;
1181         }
1182
1183 end:
1184         health_code_update();
1185         pthread_mutex_unlock(socket->lock);
1186         return ret;
1187 }
1188
1189 /*
1190  * Send metadata string to consumer. Socket lock MUST be acquired.
1191  *
1192  * Return 0 on success else a negative value.
1193  */
1194 int consumer_push_metadata(struct consumer_socket *socket,
1195                 uint64_t metadata_key, char *metadata_str, size_t len,
1196                 size_t target_offset)
1197 {
1198         int ret;
1199         struct lttcomm_consumer_msg msg;
1200
1201         assert(socket);
1202
1203         DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
1204
1205         msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1206         msg.u.push_metadata.key = metadata_key;
1207         msg.u.push_metadata.target_offset = target_offset;
1208         msg.u.push_metadata.len = len;
1209
1210         health_code_update();
1211         ret = consumer_send_msg(socket, &msg);
1212         if (ret < 0 || len == 0) {
1213                 goto end;
1214         }
1215
1216         DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1217                         len);
1218
1219         ret = consumer_socket_send(socket, metadata_str, len);
1220         if (ret < 0) {
1221                 goto end;
1222         }
1223
1224         health_code_update();
1225         ret = consumer_recv_status_reply(socket);
1226         if (ret < 0) {
1227                 goto end;
1228         }
1229
1230 end:
1231         health_code_update();
1232         return ret;
1233 }
1234
1235 /*
1236  * Ask the consumer to snapshot a specific channel using the key.
1237  *
1238  * Return 0 on success or else a negative error.
1239  */
1240 int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1241                 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
1242                 const char *session_path, int wait, int max_stream_size)
1243 {
1244         int ret;
1245         struct lttcomm_consumer_msg msg;
1246
1247         assert(socket);
1248         assert(output);
1249         assert(output->consumer);
1250
1251         DBG("Consumer snapshot channel key %" PRIu64, key);
1252
1253         memset(&msg, 0, sizeof(msg));
1254         msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1255         msg.u.snapshot_channel.key = key;
1256         msg.u.snapshot_channel.max_stream_size = max_stream_size;
1257         msg.u.snapshot_channel.metadata = metadata;
1258
1259         if (output->consumer->type == CONSUMER_DST_NET) {
1260                 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1261                 msg.u.snapshot_channel.use_relayd = 1;
1262                 ret = snprintf(msg.u.snapshot_channel.pathname,
1263                                 sizeof(msg.u.snapshot_channel.pathname),
1264                                 "%s/%s-%s-%" PRIu64 "%s", output->consumer->subdir,
1265                                 output->name, output->datetime, output->nb_snapshot,
1266                                 session_path);
1267                 if (ret < 0) {
1268                         ret = -LTTNG_ERR_NOMEM;
1269                         goto error;
1270                 }
1271         } else {
1272                 ret = snprintf(msg.u.snapshot_channel.pathname,
1273                                 sizeof(msg.u.snapshot_channel.pathname),
1274                                 "%s/%s-%s-%" PRIu64 "%s", output->consumer->dst.trace_path,
1275                                 output->name, output->datetime, output->nb_snapshot,
1276                                 session_path);
1277                 if (ret < 0) {
1278                         ret = -LTTNG_ERR_NOMEM;
1279                         goto error;
1280                 }
1281                 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1282
1283                 /* Create directory. Ignore if exist. */
1284                 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1285                                 S_IRWXU | S_IRWXG, uid, gid);
1286                 if (ret < 0) {
1287                         if (ret != -EEXIST) {
1288                                 ERR("Trace directory creation error");
1289                                 goto error;
1290                         }
1291                 }
1292         }
1293
1294         health_code_update();
1295         ret = consumer_send_msg(socket, &msg);
1296         if (ret < 0) {
1297                 goto error;
1298         }
1299
1300 error:
1301         health_code_update();
1302         return ret;
1303 }