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