src/bin/lttng-sessiond/kernel-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 _LGPL_SOURCE
  19 #include <stdio.h>
  20 #include <stdlib.h>
  21 #include <sys/stat.h>
  22 #include <unistd.h>
  23 #include <inttypes.h>
  24
  25 #include <common/common.h>
  26 #include <common/defaults.h>
  27 #include <common/compat/string.h>
  28
  29 #include "consumer.h"
  30 #include "health-sessiond.h"
  31 #include "kernel-consumer.h"
  32 #include "notification-thread-commands.h"
  33 #include "session.h"
  34 #include "lttng-sessiond.h"
  35
  36 static char *create_channel_path(struct consumer_output *consumer,
  37                 uid_t uid, gid_t gid)
  38 {
  39         int ret;
  40         char tmp_path[PATH_MAX];
  41         char *pathname = NULL;
  42
  43         assert(consumer);
  44
  45         /* Get the right path name destination */
  46         if (consumer->type == CONSUMER_DST_LOCAL) {
  47                 /* Set application path to the destination path */
  48                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s%s",
  49                                 consumer->dst.session_root_path,
  50                                 consumer->chunk_path,
  51                                 consumer->domain_subdir);
  52                 if (ret < 0) {
  53                         PERROR("snprintf kernel channel path");
  54                         goto error;
  55                 } else if (ret >= sizeof(tmp_path)) {
  56                         ERR("Kernel channel path exceeds the maximal allowed length of of %zu bytes (%i bytes required) with path \"%s%s%s\"",
  57                                         sizeof(tmp_path), ret,
  58                                         consumer->dst.session_root_path,
  59                                         consumer->chunk_path,
  60                                         consumer->domain_subdir);
  61                         goto error;
  62                 }
  63                 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
  64                 if (!pathname) {
  65                         PERROR("lttng_strndup");
  66                         goto error;
  67                 }
  68
  69                 /* Create directory */
  70                 ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, uid, gid);
  71                 if (ret < 0) {
  72                         if (errno != EEXIST) {
  73                                 ERR("Trace directory creation error");
  74                                 goto error;
  75                         }
  76                 }
  77                 DBG3("Kernel local consumer tracefile path: %s", pathname);
  78         } else {
  79                 /* Network output. */
  80                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s",
  81                                 consumer->dst.net.base_dir,
  82                                 consumer->domain_subdir);
  83                 if (ret < 0) {
  84                         PERROR("snprintf kernel metadata path");
  85                         goto error;
  86                 } else if (ret >= sizeof(tmp_path)) {
  87                         ERR("Kernel channel path exceeds the maximal allowed length of of %zu bytes (%i bytes required) with path \"%s%s\"",
  88                                         sizeof(tmp_path), ret,
  89                                         consumer->dst.net.base_dir,
  90                                         consumer->domain_subdir);
  91                         goto error;
  92                 }
  93                 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
  94                 if (!pathname) {
  95                         PERROR("lttng_strndup");
  96                         goto error;
  97                 }
  98                 DBG3("Kernel network consumer subdir path: %s", pathname);
  99         }
 100
 101         return pathname;
 102
 103 error:
 104         free(pathname);
 105         return NULL;
 106 }
 107
 108 /*
 109  * Sending a single channel to the consumer with command ADD_CHANNEL.
 110  */
 111 static
 112 int kernel_consumer_add_channel(struct consumer_socket *sock,
 113                 struct ltt_kernel_channel *channel,
 114                 struct ltt_kernel_session *ksession,
 115                 unsigned int monitor)
 116 {
 117         int ret;
 118         char *pathname;
 119         struct lttcomm_consumer_msg lkm;
 120         struct consumer_output *consumer;
 121         enum lttng_error_code status;
 122         struct ltt_session *session = NULL;
 123         struct lttng_channel_extended *channel_attr_extended;
 124
 125         /* Safety net */
 126         assert(channel);
 127         assert(ksession);
 128         assert(ksession->consumer);
 129
 130         consumer = ksession->consumer;
 131         channel_attr_extended = (struct lttng_channel_extended *)
 132                         channel->channel->attr.extended.ptr;
 133
 134         DBG("Kernel consumer adding channel %s to kernel consumer",
 135                         channel->channel->name);
 136
 137         if (monitor) {
 138                 pathname = create_channel_path(consumer, ksession->uid,
 139                                 ksession->gid);
 140         } else {
 141                 /* Empty path. */
 142                 pathname = strdup("");
 143         }
 144         if (!pathname) {
 145                 ret = -1;
 146                 goto error;
 147         }
 148
 149         /* Prep channel message structure */
 150         consumer_init_add_channel_comm_msg(&lkm,
 151                         channel->key,
 152                         ksession->id,
 153                         pathname,
 154                         ksession->uid,
 155                         ksession->gid,
 156                         consumer->net_seq_index,
 157                         channel->channel->name,
 158                         channel->stream_count,
 159                         channel->channel->attr.output,
 160                         CONSUMER_CHANNEL_TYPE_DATA,
 161                         channel->channel->attr.tracefile_size,
 162                         channel->channel->attr.tracefile_count,
 163                         monitor,
 164                         channel->channel->attr.live_timer_interval,
 165                         channel_attr_extended->monitor_timer_interval);
 166
 167         health_code_update();
 168
 169         ret = consumer_send_channel(sock, &lkm);
 170         if (ret < 0) {
 171                 goto error;
 172         }
 173
 174         health_code_update();
 175         rcu_read_lock();
 176         session = session_find_by_id(ksession->id);
 177         assert(session);
 178         assert(pthread_mutex_trylock(&session->lock));
 179         assert(session_trylock_list());
 180
 181         status = notification_thread_command_add_channel(
 182                         notification_thread_handle, session->name,
 183                         ksession->uid, ksession->gid,
 184                         channel->channel->name, channel->key,
 185                         LTTNG_DOMAIN_KERNEL,
 186                         channel->channel->attr.subbuf_size * channel->channel->attr.num_subbuf);
 187         rcu_read_unlock();
 188         if (status != LTTNG_OK) {
 189                 ret = -1;
 190                 goto error;
 191         }
 192
 193         channel->published_to_notification_thread = true;
 194
 195 error:
 196         if (session) {
 197                 session_put(session);
 198         }
 199         free(pathname);
 200         return ret;
 201 }
 202
 203 /*
 204  * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 205  *
 206  * The consumer socket lock must be held by the caller.
 207  */
 208 int kernel_consumer_add_metadata(struct consumer_socket *sock,
 209                 struct ltt_kernel_session *ksession, unsigned int monitor)
 210 {
 211         int ret;
 212         char *pathname;
 213         struct lttcomm_consumer_msg lkm;
 214         struct consumer_output *consumer;
 215         struct ltt_session *session = NULL;
 216
 217         rcu_read_lock();
 218
 219         /* Safety net */
 220         assert(ksession);
 221         assert(ksession->consumer);
 222         assert(sock);
 223
 224         DBG("Sending metadata %d to kernel consumer",
 225                         ksession->metadata_stream_fd);
 226
 227         /* Get consumer output pointer */
 228         consumer = ksession->consumer;
 229
 230         if (monitor) {
 231                 pathname = create_channel_path(consumer,
 232                                 ksession->uid, ksession->gid);
 233         } else {
 234                 /* Empty path. */
 235                 pathname = strdup("");
 236         }
 237         if (!pathname) {
 238                 ret = -1;
 239                 goto error;
 240         }
 241
 242         session = session_find_by_id(ksession->id);
 243         assert(session);
 244         assert(pthread_mutex_trylock(&session->lock));
 245         assert(session_trylock_list());
 246
 247         /* Prep channel message structure */
 248         consumer_init_add_channel_comm_msg(&lkm,
 249                         ksession->metadata->key,
 250                         ksession->id,
 251                         pathname,
 252                         ksession->uid,
 253                         ksession->gid,
 254                         consumer->net_seq_index,
 255                         DEFAULT_METADATA_NAME,
 256                         1,
 257                         DEFAULT_KERNEL_CHANNEL_OUTPUT,
 258                         CONSUMER_CHANNEL_TYPE_METADATA,
 259                         0, 0,
 260                         monitor, 0, 0);
 261
 262         health_code_update();
 263
 264         ret = consumer_send_channel(sock, &lkm);
 265         if (ret < 0) {
 266                 goto error;
 267         }
 268
 269         health_code_update();
 270
 271         /* Prep stream message structure */
 272         consumer_init_add_stream_comm_msg(&lkm,
 273                         ksession->metadata->key,
 274                         ksession->metadata_stream_fd,
 275                         0 /* CPU: 0 for metadata. */,
 276                         session->current_archive_id);
 277
 278         health_code_update();
 279
 280         /* Send stream and file descriptor */
 281         ret = consumer_send_stream(sock, consumer, &lkm,
 282                         &ksession->metadata_stream_fd, 1);
 283         if (ret < 0) {
 284                 goto error;
 285         }
 286
 287         health_code_update();
 288
 289 error:
 290         rcu_read_unlock();
 291         free(pathname);
 292         if (session) {
 293                 session_put(session);
 294         }
 295         return ret;
 296 }
 297
 298 /*
 299  * Sending a single stream to the consumer with command ADD_STREAM.
 300  */
 301 static
 302 int kernel_consumer_add_stream(struct consumer_socket *sock,
 303                 struct ltt_kernel_channel *channel,
 304                 struct ltt_kernel_stream *stream,
 305                 struct ltt_kernel_session *session, unsigned int monitor,
 306                 uint64_t trace_archive_id)
 307 {
 308         int ret;
 309         struct lttcomm_consumer_msg lkm;
 310         struct consumer_output *consumer;
 311
 312         assert(channel);
 313         assert(stream);
 314         assert(session);
 315         assert(session->consumer);
 316         assert(sock);
 317
 318         DBG("Sending stream %d of channel %s to kernel consumer",
 319                         stream->fd, channel->channel->name);
 320
 321         /* Get consumer output pointer */
 322         consumer = session->consumer;
 323
 324         /* Prep stream consumer message */
 325         consumer_init_add_stream_comm_msg(&lkm,
 326                         channel->key,
 327                         stream->fd,
 328                         stream->cpu,
 329                         trace_archive_id);
 330
 331         health_code_update();
 332
 333         /* Send stream and file descriptor */
 334         ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
 335         if (ret < 0) {
 336                 goto error;
 337         }
 338
 339         health_code_update();
 340
 341 error:
 342         return ret;
 343 }
 344
 345 /*
 346  * Sending the notification that all streams were sent with STREAMS_SENT.
 347  */
 348 int kernel_consumer_streams_sent(struct consumer_socket *sock,
 349                 struct ltt_kernel_session *session, uint64_t channel_key)
 350 {
 351         int ret;
 352         struct lttcomm_consumer_msg lkm;
 353         struct consumer_output *consumer;
 354
 355         assert(sock);
 356         assert(session);
 357
 358         DBG("Sending streams_sent");
 359         /* Get consumer output pointer */
 360         consumer = session->consumer;
 361
 362         /* Prep stream consumer message */
 363         consumer_init_streams_sent_comm_msg(&lkm,
 364                         LTTNG_CONSUMER_STREAMS_SENT,
 365                         channel_key, consumer->net_seq_index);
 366
 367         health_code_update();
 368
 369         /* Send stream and file descriptor */
 370         ret = consumer_send_msg(sock, &lkm);
 371         if (ret < 0) {
 372                 goto error;
 373         }
 374
 375 error:
 376         return ret;
 377 }
 378
 379 /*
 380  * Send all stream fds of kernel channel to the consumer.
 381  *
 382  * The consumer socket lock must be held by the caller.
 383  */
 384 int kernel_consumer_send_channel_streams(struct consumer_socket *sock,
 385                 struct ltt_kernel_channel *channel, struct ltt_kernel_session *ksession,
 386                 unsigned int monitor)
 387 {
 388         int ret = LTTNG_OK;
 389         struct ltt_kernel_stream *stream;
 390         struct ltt_session *session = NULL;
 391
 392         /* Safety net */
 393         assert(channel);
 394         assert(ksession);
 395         assert(ksession->consumer);
 396         assert(sock);
 397
 398         rcu_read_lock();
 399
 400         session = session_find_by_id(ksession->id);
 401         assert(session);
 402         assert(pthread_mutex_trylock(&session->lock));
 403         assert(session_trylock_list());
 404
 405         /* Bail out if consumer is disabled */
 406         if (!ksession->consumer->enabled) {
 407                 ret = LTTNG_OK;
 408                 goto error;
 409         }
 410
 411         DBG("Sending streams of channel %s to kernel consumer",
 412                         channel->channel->name);
 413
 414         if (!channel->sent_to_consumer) {
 415                 ret = kernel_consumer_add_channel(sock, channel, ksession, monitor);
 416                 if (ret < 0) {
 417                         goto error;
 418                 }
 419                 channel->sent_to_consumer = true;
 420         }
 421
 422         /* Send streams */
 423         cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
 424                 if (!stream->fd || stream->sent_to_consumer) {
 425                         continue;
 426                 }
 427
 428                 /* Add stream on the kernel consumer side. */
 429                 ret = kernel_consumer_add_stream(sock, channel, stream,
 430                                 ksession, monitor, session->current_archive_id);
 431                 if (ret < 0) {
 432                         goto error;
 433                 }
 434                 stream->sent_to_consumer = true;
 435         }
 436
 437 error:
 438         rcu_read_unlock();
 439         if (session) {
 440                 session_put(session);
 441         }
 442         return ret;
 443 }
 444
 445 /*
 446  * Send all stream fds of the kernel session to the consumer.
 447  *
 448  * The consumer socket lock must be held by the caller.
 449  */
 450 int kernel_consumer_send_session(struct consumer_socket *sock,
 451                 struct ltt_kernel_session *session)
 452 {
 453         int ret, monitor = 0;
 454         struct ltt_kernel_channel *chan;
 455
 456         /* Safety net */
 457         assert(session);
 458         assert(session->consumer);
 459         assert(sock);
 460
 461         /* Bail out if consumer is disabled */
 462         if (!session->consumer->enabled) {
 463                 ret = LTTNG_OK;
 464                 goto error;
 465         }
 466
 467         /* Don't monitor the streams on the consumer if in flight recorder. */
 468         if (session->output_traces) {
 469                 monitor = 1;
 470         }
 471
 472         DBG("Sending session stream to kernel consumer");
 473
 474         if (session->metadata_stream_fd >= 0 && session->metadata) {
 475                 ret = kernel_consumer_add_metadata(sock, session, monitor);
 476                 if (ret < 0) {
 477                         goto error;
 478                 }
 479         }
 480
 481         /* Send channel and streams of it */
 482         cds_list_for_each_entry(chan, &session->channel_list.head, list) {
 483                 ret = kernel_consumer_send_channel_streams(sock, chan, session,
 484                                 monitor);
 485                 if (ret < 0) {
 486                         goto error;
 487                 }
 488                 if (monitor) {
 489                         /*
 490                          * Inform the relay that all the streams for the
 491                          * channel were sent.
 492                          */
 493                         ret = kernel_consumer_streams_sent(sock, session, chan->key);
 494                         if (ret < 0) {
 495                                 goto error;
 496                         }
 497                 }
 498         }
 499
 500         DBG("Kernel consumer FDs of metadata and channel streams sent");
 501
 502         session->consumer_fds_sent = 1;
 503         return 0;
 504
 505 error:
 506         return ret;
 507 }
 508
 509 int kernel_consumer_destroy_channel(struct consumer_socket *socket,
 510                 struct ltt_kernel_channel *channel)
 511 {
 512         int ret;
 513         struct lttcomm_consumer_msg msg;
 514
 515         assert(channel);
 516         assert(socket);
 517
 518         DBG("Sending kernel consumer destroy channel key %" PRIu64, channel->key);
 519
 520         memset(&msg, 0, sizeof(msg));
 521         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 522         msg.u.destroy_channel.key = channel->key;
 523
 524         pthread_mutex_lock(socket->lock);
 525         health_code_update();
 526
 527         ret = consumer_send_msg(socket, &msg);
 528         if (ret < 0) {
 529                 goto error;
 530         }
 531
 532 error:
 533         health_code_update();
 534         pthread_mutex_unlock(socket->lock);
 535         return ret;
 536 }
 537
 538 int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
 539                 struct ltt_kernel_metadata *metadata)
 540 {
 541         int ret;
 542         struct lttcomm_consumer_msg msg;
 543
 544         assert(metadata);
 545         assert(socket);
 546
 547         DBG("Sending kernel consumer destroy channel key %" PRIu64, metadata->key);
 548
 549         memset(&msg, 0, sizeof(msg));
 550         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 551         msg.u.destroy_channel.key = metadata->key;
 552
 553         pthread_mutex_lock(socket->lock);
 554         health_code_update();
 555
 556         ret = consumer_send_msg(socket, &msg);
 557         if (ret < 0) {
 558                 goto error;
 559         }
 560
 561 error:
 562         health_code_update();
 563         pthread_mutex_unlock(socket->lock);
 564         return ret;
 565 }