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

   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.hpp>
  16 #include <common/defaults.hpp>
  17 #include <common/compat/string.hpp>
  18
  19 #include "consumer.hpp"
  20 #include "health-sessiond.hpp"
  21 #include "kernel-consumer.hpp"
  22 #include "notification-thread-commands.hpp"
  23 #include "session.hpp"
  24 #include "lttng-sessiond.hpp"
  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         LTTNG_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         LTTNG_ASSERT(channel);
 101         LTTNG_ASSERT(ksession);
 102         LTTNG_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                         consumer->net_seq_index,
 149                         channel->channel->name,
 150                         channel->stream_count,
 151                         channel->channel->attr.output,
 152                         CONSUMER_CHANNEL_TYPE_DATA,
 153                         channel->channel->attr.tracefile_size,
 154                         channel->channel->attr.tracefile_count,
 155                         monitor,
 156                         channel->channel->attr.live_timer_interval,
 157                         ksession->is_live_session,
 158                         channel_attr_extended->monitor_timer_interval,
 159                         ksession->current_trace_chunk);
 160
 161         health_code_update();
 162
 163         ret = consumer_send_channel(sock, &lkm);
 164         if (ret < 0) {
 165                 goto error;
 166         }
 167
 168         health_code_update();
 169         rcu_read_lock();
 170         session = session_find_by_id(ksession->id);
 171         LTTNG_ASSERT(session);
 172         LTTNG_ASSERT(pthread_mutex_trylock(&session->lock));
 173         LTTNG_ASSERT(session_trylock_list());
 174
 175         status = notification_thread_command_add_channel(
 176                         the_notification_thread_handle, session->name,
 177                         ksession->uid, ksession->gid, channel->channel->name,
 178                         channel->key, LTTNG_DOMAIN_KERNEL,
 179                         channel->channel->attr.subbuf_size *
 180                                         channel->channel->attr.num_subbuf);
 181         rcu_read_unlock();
 182         if (status != LTTNG_OK) {
 183                 ret = -1;
 184                 goto error;
 185         }
 186
 187         channel->published_to_notification_thread = true;
 188
 189 error:
 190         if (session) {
 191                 session_put(session);
 192         }
 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 *ksession, unsigned int monitor)
 204 {
 205         int ret;
 206         struct lttcomm_consumer_msg lkm;
 207         struct consumer_output *consumer;
 208
 209         rcu_read_lock();
 210
 211         /* Safety net */
 212         LTTNG_ASSERT(ksession);
 213         LTTNG_ASSERT(ksession->consumer);
 214         LTTNG_ASSERT(sock);
 215
 216         DBG("Sending metadata %d to kernel consumer",
 217                         ksession->metadata_stream_fd);
 218
 219         /* Get consumer output pointer */
 220         consumer = ksession->consumer;
 221
 222         /* Prep channel message structure */
 223         consumer_init_add_channel_comm_msg(&lkm,
 224                         ksession->metadata->key,
 225                         ksession->id,
 226                         "",
 227                         consumer->net_seq_index,
 228                         ksession->metadata->conf->name,
 229                         1,
 230                         ksession->metadata->conf->attr.output,
 231                         CONSUMER_CHANNEL_TYPE_METADATA,
 232                         ksession->metadata->conf->attr.tracefile_size,
 233                         ksession->metadata->conf->attr.tracefile_count,
 234                         monitor,
 235                         ksession->metadata->conf->attr.live_timer_interval,
 236                         ksession->is_live_session,
 237                         0,
 238                         ksession->current_trace_chunk);
 239
 240         health_code_update();
 241
 242         ret = consumer_send_channel(sock, &lkm);
 243         if (ret < 0) {
 244                 goto error;
 245         }
 246
 247         health_code_update();
 248
 249         /* Prep stream message structure */
 250         consumer_init_add_stream_comm_msg(&lkm,
 251                         ksession->metadata->key,
 252                         ksession->metadata_stream_fd,
 253                         0 /* CPU: 0 for metadata. */);
 254
 255         health_code_update();
 256
 257         /* Send stream and file descriptor */
 258         ret = consumer_send_stream(sock, consumer, &lkm,
 259                         &ksession->metadata_stream_fd, 1);
 260         if (ret < 0) {
 261                 goto error;
 262         }
 263
 264         health_code_update();
 265
 266 error:
 267         rcu_read_unlock();
 268         return ret;
 269 }
 270
 271 /*
 272  * Sending a single stream to the consumer with command ADD_STREAM.
 273  */
 274 static
 275 int kernel_consumer_add_stream(struct consumer_socket *sock,
 276                 struct ltt_kernel_channel *channel,
 277                 struct ltt_kernel_stream *stream,
 278                 struct ltt_kernel_session *session)
 279 {
 280         int ret;
 281         struct lttcomm_consumer_msg lkm;
 282         struct consumer_output *consumer;
 283
 284         LTTNG_ASSERT(channel);
 285         LTTNG_ASSERT(stream);
 286         LTTNG_ASSERT(session);
 287         LTTNG_ASSERT(session->consumer);
 288         LTTNG_ASSERT(sock);
 289
 290         DBG("Sending stream %d of channel %s to kernel consumer",
 291                         stream->fd, channel->channel->name);
 292
 293         /* Get consumer output pointer */
 294         consumer = session->consumer;
 295
 296         /* Prep stream consumer message */
 297         consumer_init_add_stream_comm_msg(&lkm,
 298                         channel->key,
 299                         stream->fd,
 300                         stream->cpu);
 301
 302         health_code_update();
 303
 304         /* Send stream and file descriptor */
 305         ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
 306         if (ret < 0) {
 307                 goto error;
 308         }
 309
 310         health_code_update();
 311
 312 error:
 313         return ret;
 314 }
 315
 316 /*
 317  * Sending the notification that all streams were sent with STREAMS_SENT.
 318  */
 319 int kernel_consumer_streams_sent(struct consumer_socket *sock,
 320                 struct ltt_kernel_session *session, uint64_t channel_key)
 321 {
 322         int ret;
 323         struct lttcomm_consumer_msg lkm;
 324         struct consumer_output *consumer;
 325
 326         LTTNG_ASSERT(sock);
 327         LTTNG_ASSERT(session);
 328
 329         DBG("Sending streams_sent");
 330         /* Get consumer output pointer */
 331         consumer = session->consumer;
 332
 333         /* Prep stream consumer message */
 334         consumer_init_streams_sent_comm_msg(&lkm,
 335                         LTTNG_CONSUMER_STREAMS_SENT,
 336                         channel_key, consumer->net_seq_index);
 337
 338         health_code_update();
 339
 340         /* Send stream and file descriptor */
 341         ret = consumer_send_msg(sock, &lkm);
 342         if (ret < 0) {
 343                 goto error;
 344         }
 345
 346 error:
 347         return ret;
 348 }
 349
 350 /*
 351  * Send all stream fds of kernel channel to the consumer.
 352  *
 353  * The consumer socket lock must be held by the caller.
 354  */
 355 int kernel_consumer_send_channel_streams(struct consumer_socket *sock,
 356                 struct ltt_kernel_channel *channel, struct ltt_kernel_session *ksession,
 357                 unsigned int monitor)
 358 {
 359         int ret = LTTNG_OK;
 360         struct ltt_kernel_stream *stream;
 361
 362         /* Safety net */
 363         LTTNG_ASSERT(channel);
 364         LTTNG_ASSERT(ksession);
 365         LTTNG_ASSERT(ksession->consumer);
 366         LTTNG_ASSERT(sock);
 367
 368         rcu_read_lock();
 369
 370         /* Bail out if consumer is disabled */
 371         if (!ksession->consumer->enabled) {
 372                 ret = LTTNG_OK;
 373                 goto error;
 374         }
 375
 376         DBG("Sending streams of channel %s to kernel consumer",
 377                         channel->channel->name);
 378
 379         if (!channel->sent_to_consumer) {
 380                 ret = kernel_consumer_add_channel(sock, channel, ksession, monitor);
 381                 if (ret < 0) {
 382                         goto error;
 383                 }
 384                 channel->sent_to_consumer = true;
 385         }
 386
 387         /* Send streams */
 388         cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
 389                 if (!stream->fd || stream->sent_to_consumer) {
 390                         continue;
 391                 }
 392
 393                 /* Add stream on the kernel consumer side. */
 394                 ret = kernel_consumer_add_stream(sock, channel, stream,
 395                                 ksession);
 396                 if (ret < 0) {
 397                         goto error;
 398                 }
 399                 stream->sent_to_consumer = true;
 400         }
 401
 402 error:
 403         rcu_read_unlock();
 404         return ret;
 405 }
 406
 407 /*
 408  * Send all stream fds of the kernel session to the consumer.
 409  *
 410  * The consumer socket lock must be held by the caller.
 411  */
 412 int kernel_consumer_send_session(struct consumer_socket *sock,
 413                 struct ltt_kernel_session *session)
 414 {
 415         int ret, monitor = 0;
 416         struct ltt_kernel_channel *chan;
 417
 418         /* Safety net */
 419         LTTNG_ASSERT(session);
 420         LTTNG_ASSERT(session->consumer);
 421         LTTNG_ASSERT(sock);
 422
 423         /* Bail out if consumer is disabled */
 424         if (!session->consumer->enabled) {
 425                 ret = LTTNG_OK;
 426                 goto error;
 427         }
 428
 429         /* Don't monitor the streams on the consumer if in flight recorder. */
 430         if (session->output_traces) {
 431                 monitor = 1;
 432         }
 433
 434         DBG("Sending session stream to kernel consumer");
 435
 436         if (session->metadata_stream_fd >= 0 && session->metadata) {
 437                 ret = kernel_consumer_add_metadata(sock, session, monitor);
 438                 if (ret < 0) {
 439                         goto error;
 440                 }
 441         }
 442
 443         /* Send channel and streams of it */
 444         cds_list_for_each_entry(chan, &session->channel_list.head, list) {
 445                 ret = kernel_consumer_send_channel_streams(sock, chan, session,
 446                                 monitor);
 447                 if (ret < 0) {
 448                         goto error;
 449                 }
 450                 if (monitor) {
 451                         /*
 452                          * Inform the relay that all the streams for the
 453                          * channel were sent.
 454                          */
 455                         ret = kernel_consumer_streams_sent(sock, session, chan->key);
 456                         if (ret < 0) {
 457                                 goto error;
 458                         }
 459                 }
 460         }
 461
 462         DBG("Kernel consumer FDs of metadata and channel streams sent");
 463
 464         session->consumer_fds_sent = 1;
 465         return 0;
 466
 467 error:
 468         return ret;
 469 }
 470
 471 int kernel_consumer_destroy_channel(struct consumer_socket *socket,
 472                 struct ltt_kernel_channel *channel)
 473 {
 474         int ret;
 475         struct lttcomm_consumer_msg msg;
 476
 477         LTTNG_ASSERT(channel);
 478         LTTNG_ASSERT(socket);
 479
 480         DBG("Sending kernel consumer destroy channel key %" PRIu64, channel->key);
 481
 482         memset(&msg, 0, sizeof(msg));
 483         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 484         msg.u.destroy_channel.key = channel->key;
 485
 486         pthread_mutex_lock(socket->lock);
 487         health_code_update();
 488
 489         ret = consumer_send_msg(socket, &msg);
 490         if (ret < 0) {
 491                 goto error;
 492         }
 493
 494 error:
 495         health_code_update();
 496         pthread_mutex_unlock(socket->lock);
 497         return ret;
 498 }
 499
 500 int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
 501                 struct ltt_kernel_metadata *metadata)
 502 {
 503         int ret;
 504         struct lttcomm_consumer_msg msg;
 505
 506         LTTNG_ASSERT(metadata);
 507         LTTNG_ASSERT(socket);
 508
 509         DBG("Sending kernel consumer destroy channel key %" PRIu64, metadata->key);
 510
 511         memset(&msg, 0, sizeof(msg));
 512         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 513         msg.u.destroy_channel.key = metadata->key;
 514
 515         pthread_mutex_lock(socket->lock);
 516         health_code_update();
 517
 518         ret = consumer_send_msg(socket, &msg);
 519         if (ret < 0) {
 520                 goto error;
 521         }
 522
 523 error:
 524         health_code_update();
 525         pthread_mutex_unlock(socket->lock);
 526         return ret;
 527 }