src/bin/lttng-sessiond/kernel-consumer.c

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