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