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
  27 #include <common/common.h>
  28 #include <common/defaults.h>
  29 #include <common/uri.h>
  30
  31 #include "consumer.h"
  32
  33 /*
  34  * Send destroy relayd command to consumer.
  35  *
  36  * On success return positive value. On error, negative value.
  37  */
  38 int consumer_send_destroy_relayd(struct consumer_socket *sock,
  39                 struct consumer_output *consumer)
  40 {
  41         int ret;
  42         struct lttcomm_consumer_msg msg;
  43
  44         assert(consumer);
  45         assert(sock);
  46
  47         DBG2("Sending destroy relayd command to consumer...");
  48
  49         /* Bail out if consumer is disabled */
  50         if (!consumer->enabled) {
  51                 ret = LTTNG_OK;
  52                 DBG3("Consumer is disabled");
  53                 goto error;
  54         }
  55
  56         msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
  57         msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
  58
  59         pthread_mutex_lock(sock->lock);
  60         ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg));
  61         pthread_mutex_unlock(sock->lock);
  62         if (ret < 0) {
  63                 PERROR("send consumer destroy relayd command");
  64                 goto error;
  65         }
  66
  67         DBG2("Consumer send destroy relayd command done");
  68
  69 error:
  70         return ret;
  71 }
  72
  73 /*
  74  * For each consumer socket in the consumer output object, send a destroy
  75  * relayd command.
  76  */
  77 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
  78 {
  79         int ret;
  80         struct lttng_ht_iter iter;
  81         struct consumer_socket *socket;
  82
  83         assert(consumer);
  84
  85         /* Destroy any relayd connection */
  86         if (consumer && consumer->type == CONSUMER_DST_NET) {
  87                 rcu_read_lock();
  88                 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
  89                                 node.node) {
  90                         /* Send destroy relayd command */
  91                         ret = consumer_send_destroy_relayd(socket, consumer);
  92                         if (ret < 0) {
  93                                 ERR("Unable to send destroy relayd command to consumer");
  94                                 /* Continue since we MUST delete everything at this point. */
  95                         }
  96                 }
  97                 rcu_read_unlock();
  98         }
  99 }
 100
 101 /*
 102  * From a consumer_data structure, allocate and add a consumer socket to the
 103  * consumer output.
 104  *
 105  * Return 0 on success, else negative value on error
 106  */
 107 int consumer_create_socket(struct consumer_data *data,
 108                 struct consumer_output *output)
 109 {
 110         int ret = 0;
 111         struct consumer_socket *socket;
 112
 113         assert(data);
 114
 115         if (output == NULL || data->cmd_sock < 0) {
 116                 /*
 117                  * Not an error. Possible there is simply not spawned consumer or it's
 118                  * disabled for the tracing session asking the socket.
 119                  */
 120                 goto error;
 121         }
 122
 123         rcu_read_lock();
 124         socket = consumer_find_socket(data->cmd_sock, output);
 125         rcu_read_unlock();
 126         if (socket == NULL) {
 127                 socket = consumer_allocate_socket(data->cmd_sock);
 128                 if (socket == NULL) {
 129                         ret = -1;
 130                         goto error;
 131                 }
 132
 133                 socket->registered = 0;
 134                 socket->lock = &data->lock;
 135                 rcu_read_lock();
 136                 consumer_add_socket(socket, output);
 137                 rcu_read_unlock();
 138         }
 139
 140         DBG3("Consumer socket created (fd: %d) and added to output",
 141                         data->cmd_sock);
 142
 143 error:
 144         return ret;
 145 }
 146
 147 /*
 148  * Find a consumer_socket in a consumer_output hashtable. Read side lock must
 149  * be acquired before calling this function and across use of the
 150  * returned consumer_socket.
 151  */
 152 struct consumer_socket *consumer_find_socket(int key,
 153                 struct consumer_output *consumer)
 154 {
 155         struct lttng_ht_iter iter;
 156         struct lttng_ht_node_ulong *node;
 157         struct consumer_socket *socket = NULL;
 158
 159         /* Negative keys are lookup failures */
 160         if (key < 0 || consumer == NULL) {
 161                 return NULL;
 162         }
 163
 164         lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
 165                         &iter);
 166         node = lttng_ht_iter_get_node_ulong(&iter);
 167         if (node != NULL) {
 168                 socket = caa_container_of(node, struct consumer_socket, node);
 169         }
 170
 171         return socket;
 172 }
 173
 174 /*
 175  * Allocate a new consumer_socket and return the pointer.
 176  */
 177 struct consumer_socket *consumer_allocate_socket(int fd)
 178 {
 179         struct consumer_socket *socket = NULL;
 180
 181         socket = zmalloc(sizeof(struct consumer_socket));
 182         if (socket == NULL) {
 183                 PERROR("zmalloc consumer socket");
 184                 goto error;
 185         }
 186
 187         socket->fd = fd;
 188         lttng_ht_node_init_ulong(&socket->node, fd);
 189
 190 error:
 191         return socket;
 192 }
 193
 194 /*
 195  * Add consumer socket to consumer output object. Read side lock must be
 196  * acquired before calling this function.
 197  */
 198 void consumer_add_socket(struct consumer_socket *sock,
 199                 struct consumer_output *consumer)
 200 {
 201         assert(sock);
 202         assert(consumer);
 203
 204         lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
 205 }
 206
 207 /*
 208  * Delte consumer socket to consumer output object. Read side lock must be
 209  * acquired before calling this function.
 210  */
 211 void consumer_del_socket(struct consumer_socket *sock,
 212                 struct consumer_output *consumer)
 213 {
 214         int ret;
 215         struct lttng_ht_iter iter;
 216
 217         assert(sock);
 218         assert(consumer);
 219
 220         iter.iter.node = &sock->node.node;
 221         ret = lttng_ht_del(consumer->socks, &iter);
 222         assert(!ret);
 223 }
 224
 225 /*
 226  * RCU destroy call function.
 227  */
 228 static void destroy_socket_rcu(struct rcu_head *head)
 229 {
 230         struct lttng_ht_node_ulong *node =
 231                 caa_container_of(head, struct lttng_ht_node_ulong, head);
 232         struct consumer_socket *socket =
 233                 caa_container_of(node, struct consumer_socket, node);
 234
 235         free(socket);
 236 }
 237
 238 /*
 239  * Destroy and free socket pointer in a call RCU. Read side lock must be
 240  * acquired before calling this function.
 241  */
 242 void consumer_destroy_socket(struct consumer_socket *sock)
 243 {
 244         assert(sock);
 245
 246         /*
 247          * We DO NOT close the file descriptor here since it is global to the
 248          * session daemon and is closed only if the consumer dies or a custom
 249          * consumer was registered,
 250          */
 251         if (sock->registered) {
 252                 DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
 253                 lttcomm_close_unix_sock(sock->fd);
 254         }
 255
 256         call_rcu(&sock->node.head, destroy_socket_rcu);
 257 }
 258
 259 /*
 260  * Allocate and assign data to a consumer_output object.
 261  *
 262  * Return pointer to structure.
 263  */
 264 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
 265 {
 266         struct consumer_output *output = NULL;
 267
 268         output = zmalloc(sizeof(struct consumer_output));
 269         if (output == NULL) {
 270                 PERROR("zmalloc consumer_output");
 271                 goto error;
 272         }
 273
 274         /* By default, consumer output is enabled */
 275         output->enabled = 1;
 276         output->type = type;
 277         output->net_seq_index = -1;
 278
 279         output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 280
 281 error:
 282         return output;
 283 }
 284
 285 /*
 286  * Delete the consumer_output object from the list and free the ptr.
 287  */
 288 void consumer_destroy_output(struct consumer_output *obj)
 289 {
 290         if (obj == NULL) {
 291                 return;
 292         }
 293
 294         if (obj->socks) {
 295                 struct lttng_ht_iter iter;
 296                 struct consumer_socket *socket;
 297
 298                 rcu_read_lock();
 299                 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
 300                         consumer_del_socket(socket, obj);
 301                         consumer_destroy_socket(socket);
 302                 }
 303                 rcu_read_unlock();
 304
 305                 /* Finally destroy HT */
 306                 lttng_ht_destroy(obj->socks);
 307         }
 308
 309         free(obj);
 310 }
 311
 312 /*
 313  * Copy consumer output and returned the newly allocated copy.
 314  */
 315 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
 316 {
 317         struct lttng_ht_iter iter;
 318         struct consumer_socket *socket, *copy_sock;
 319         struct consumer_output *output;
 320
 321         assert(obj);
 322
 323         output = consumer_create_output(obj->type);
 324         if (output == NULL) {
 325                 goto error;
 326         }
 327
 328         memcpy(output, obj, sizeof(struct consumer_output));
 329
 330         /* Copy sockets */
 331         output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 332
 333         cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
 334                 /* Create new socket object. */
 335                 copy_sock = consumer_allocate_socket(socket->fd);
 336                 if (copy_sock == NULL) {
 337                         goto malloc_error;
 338                 }
 339
 340                 copy_sock->lock = socket->lock;
 341                 consumer_add_socket(copy_sock, output);
 342         }
 343
 344 error:
 345         return output;
 346
 347 malloc_error:
 348         consumer_destroy_output(output);
 349         return NULL;
 350 }
 351
 352 /*
 353  * Set network URI to the consumer output object.
 354  *
 355  * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
 356  * error.
 357  */
 358 int consumer_set_network_uri(struct consumer_output *obj,
 359                 struct lttng_uri *uri)
 360 {
 361         int ret;
 362         char tmp_path[PATH_MAX];
 363         char hostname[HOST_NAME_MAX];
 364         struct lttng_uri *dst_uri = NULL;
 365
 366         /* Code flow error safety net. */
 367         assert(obj);
 368         assert(uri);
 369
 370         switch (uri->stype) {
 371         case LTTNG_STREAM_CONTROL:
 372                 dst_uri = &obj->dst.net.control;
 373                 obj->dst.net.control_isset = 1;
 374                 if (uri->port == 0) {
 375                         /* Assign default port. */
 376                         uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 377                 }
 378                 DBG3("Consumer control URI set with port %d", uri->port);
 379                 break;
 380         case LTTNG_STREAM_DATA:
 381                 dst_uri = &obj->dst.net.data;
 382                 obj->dst.net.data_isset = 1;
 383                 if (uri->port == 0) {
 384                         /* Assign default port. */
 385                         uri->port = DEFAULT_NETWORK_DATA_PORT;
 386                 }
 387                 DBG3("Consumer data URI set with port %d", uri->port);
 388                 break;
 389         default:
 390                 ERR("Set network uri type unknown %d", uri->stype);
 391                 goto error;
 392         }
 393
 394         ret = uri_compare(dst_uri, uri);
 395         if (!ret) {
 396                 /* Same URI, don't touch it and return success. */
 397                 DBG3("URI network compare are the same");
 398                 goto equal;
 399         }
 400
 401         /* URIs were not equal, replacing it. */
 402         memset(dst_uri, 0, sizeof(struct lttng_uri));
 403         memcpy(dst_uri, uri, sizeof(struct lttng_uri));
 404         obj->type = CONSUMER_DST_NET;
 405
 406         /* Handle subdir and add hostname in front. */
 407         if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
 408                 /* Get hostname to append it in the pathname */
 409                 ret = gethostname(hostname, sizeof(hostname));
 410                 if (ret < 0) {
 411                         PERROR("gethostname. Fallback on default localhost");
 412                         strncpy(hostname, "localhost", sizeof(hostname));
 413                 }
 414                 hostname[sizeof(hostname) - 1] = '\0';
 415
 416                 /* Setup consumer subdir if none present in the control URI */
 417                 if (strlen(dst_uri->subdir) == 0) {
 418                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 419                                         hostname, obj->subdir);
 420                 } else {
 421                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 422                                         hostname, dst_uri->subdir);
 423                 }
 424                 if (ret < 0) {
 425                         PERROR("snprintf set consumer uri subdir");
 426                         goto error;
 427                 }
 428
 429                 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
 430                 DBG3("Consumer set network uri subdir path %s", tmp_path);
 431         }
 432
 433         return 0;
 434 equal:
 435         return 1;
 436 error:
 437         return -1;
 438 }
 439
 440 /*
 441  * Send file descriptor to consumer via sock.
 442  */
 443 int consumer_send_fds(int sock, int *fds, size_t nb_fd)
 444 {
 445         int ret;
 446
 447         assert(fds);
 448         assert(nb_fd > 0);
 449
 450         ret = lttcomm_send_fds_unix_sock(sock, fds, nb_fd);
 451         if (ret < 0) {
 452                 PERROR("send consumer fds");
 453                 goto error;
 454         }
 455
 456 error:
 457         return ret;
 458 }
 459
 460 /*
 461  * Consumer send channel communication message structure to consumer.
 462  */
 463 int consumer_send_channel(int sock, struct lttcomm_consumer_msg *msg)
 464 {
 465         int ret;
 466
 467         assert(msg);
 468         assert(sock >= 0);
 469
 470         ret = lttcomm_send_unix_sock(sock, msg,
 471                         sizeof(struct lttcomm_consumer_msg));
 472         if (ret < 0) {
 473                 PERROR("send consumer channel");
 474                 goto error;
 475         }
 476
 477 error:
 478         return ret;
 479 }
 480
 481 /*
 482  * Init channel communication message structure.
 483  */
 484 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 485                 enum lttng_consumer_command cmd,
 486                 int channel_key,
 487                 uint64_t max_sb_size,
 488                 uint64_t mmap_len,
 489                 const char *name,
 490                 unsigned int nb_init_streams)
 491 {
 492         assert(msg);
 493
 494         /* TODO: Args validation */
 495
 496         /* Zeroed structure */
 497         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 498
 499         /* Send channel */
 500         msg->cmd_type = cmd;
 501         msg->u.channel.channel_key = channel_key;
 502         msg->u.channel.max_sb_size = max_sb_size;
 503         msg->u.channel.mmap_len = mmap_len;
 504         msg->u.channel.nb_init_streams = nb_init_streams;
 505 }
 506
 507 /*
 508  * Init stream communication message structure.
 509  */
 510 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
 511                 enum lttng_consumer_command cmd,
 512                 int channel_key,
 513                 int stream_key,
 514                 uint32_t state,
 515                 enum lttng_event_output output,
 516                 uint64_t mmap_len,
 517                 uid_t uid,
 518                 gid_t gid,
 519                 int net_index,
 520                 unsigned int metadata_flag,
 521                 const char *name,
 522                 const char *pathname)
 523 {
 524         assert(msg);
 525
 526         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 527
 528         /* TODO: Args validation */
 529
 530         msg->cmd_type = cmd;
 531         msg->u.stream.channel_key = channel_key;
 532         msg->u.stream.stream_key = stream_key;
 533         msg->u.stream.state = state;
 534         msg->u.stream.output = output;
 535         msg->u.stream.mmap_len = mmap_len;
 536         msg->u.stream.uid = uid;
 537         msg->u.stream.gid = gid;
 538         msg->u.stream.net_index = net_index;
 539         msg->u.stream.metadata_flag = metadata_flag;
 540         strncpy(msg->u.stream.name, name, sizeof(msg->u.stream.name));
 541         msg->u.stream.name[sizeof(msg->u.stream.name) - 1] = '\0';
 542         strncpy(msg->u.stream.path_name, pathname,
 543                         sizeof(msg->u.stream.path_name));
 544         msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
 545 }
 546
 547 /*
 548  * Send stream communication structure to the consumer.
 549  */
 550 int consumer_send_stream(int sock, struct consumer_output *dst,
 551         struct lttcomm_consumer_msg *msg, int *fds, size_t nb_fd)
 552 {
 553         int ret;
 554
 555         assert(msg);
 556         assert(dst);
 557
 558         switch (dst->type) {
 559         case CONSUMER_DST_NET:
 560                 /* Consumer should send the stream on the network. */
 561                 msg->u.stream.net_index = dst->net_seq_index;
 562                 break;
 563         case CONSUMER_DST_LOCAL:
 564                 /* Add stream file name to stream path */
 565                 strncat(msg->u.stream.path_name, "/",
 566                                 sizeof(msg->u.stream.path_name) -
 567                                 strlen(msg->u.stream.path_name) - 1);
 568                 strncat(msg->u.stream.path_name, msg->u.stream.name,
 569                                 sizeof(msg->u.stream.path_name) -
 570                                 strlen(msg->u.stream.path_name) - 1);
 571                 msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
 572                 /* Indicate that the stream is NOT network */
 573                 msg->u.stream.net_index = -1;
 574                 break;
 575         default:
 576                 ERR("Consumer unknown output type (%d)", dst->type);
 577                 ret = -1;
 578                 goto error;
 579         }
 580
 581         /* Send on socket */
 582         ret = lttcomm_send_unix_sock(sock, msg,
 583                         sizeof(struct lttcomm_consumer_msg));
 584         if (ret < 0) {
 585                 PERROR("send consumer stream");
 586                 goto error;
 587         }
 588
 589         ret = consumer_send_fds(sock, fds, nb_fd);
 590         if (ret < 0) {
 591                 goto error;
 592         }
 593
 594 error:
 595         return ret;
 596 }
 597
 598 /*
 599  * Send relayd socket to consumer associated with a session name.
 600  *
 601  * On success return positive value. On error, negative value.
 602  */
 603 int consumer_send_relayd_socket(int consumer_sock,
 604                 struct lttcomm_sock *sock, struct consumer_output *consumer,
 605                 enum lttng_stream_type type)
 606 {
 607         int ret;
 608         struct lttcomm_consumer_msg msg;
 609
 610         /* Code flow error. Safety net. */
 611         assert(sock);
 612         assert(consumer);
 613
 614         /* Bail out if consumer is disabled */
 615         if (!consumer->enabled) {
 616                 ret = LTTNG_OK;
 617                 goto error;
 618         }
 619
 620         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
 621         /*
 622          * Assign network consumer output index using the temporary consumer since
 623          * this call should only be made from within a set_consumer_uri() function
 624          * call in the session daemon.
 625          */
 626         msg.u.relayd_sock.net_index = consumer->net_seq_index;
 627         msg.u.relayd_sock.type = type;
 628         memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock));
 629
 630         DBG3("Sending relayd sock info to consumer on %d", consumer_sock);
 631         ret = lttcomm_send_unix_sock(consumer_sock, &msg, sizeof(msg));
 632         if (ret < 0) {
 633                 PERROR("send consumer relayd socket info");
 634                 goto error;
 635         }
 636
 637         DBG3("Sending relayd socket file descriptor to consumer");
 638         ret = consumer_send_fds(consumer_sock, &sock->fd, 1);
 639         if (ret < 0) {
 640                 goto error;
 641         }
 642
 643         DBG2("Consumer relayd socket sent");
 644
 645 error:
 646         return ret;
 647 }
 648
 649 /*
 650  * Set consumer subdirectory using the session name and a generated datetime if
 651  * needed. This is appended to the current subdirectory.
 652  */
 653 int consumer_set_subdir(struct consumer_output *consumer,
 654                 const char *session_name)
 655 {
 656         int ret = 0;
 657         unsigned int have_default_name = 0;
 658         char datetime[16], tmp_path[PATH_MAX];
 659         time_t rawtime;
 660         struct tm *timeinfo;
 661
 662         assert(consumer);
 663         assert(session_name);
 664
 665         memset(tmp_path, 0, sizeof(tmp_path));
 666
 667         /* Flag if we have a default session. */
 668         if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
 669                                 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
 670                 have_default_name = 1;
 671         } else {
 672                 /* Get date and time for session path */
 673                 time(&rawtime);
 674                 timeinfo = localtime(&rawtime);
 675                 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
 676         }
 677
 678         if (have_default_name) {
 679                 ret = snprintf(tmp_path, sizeof(tmp_path),
 680                                 "%s/%s", consumer->subdir, session_name);
 681         } else {
 682                 ret = snprintf(tmp_path, sizeof(tmp_path),
 683                                 "%s/%s-%s/", consumer->subdir, session_name, datetime);
 684         }
 685         if (ret < 0) {
 686                 PERROR("snprintf session name date");
 687                 goto error;
 688         }
 689
 690         strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
 691         DBG2("Consumer subdir set to %s", consumer->subdir);
 692
 693 error:
 694         return ret;
 695 }