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 "consumer.hpp"
  10 #include "health-sessiond.hpp"
  11 #include "kernel-consumer.hpp"
  12 #include "lttng-sessiond.hpp"
  13 #include "notification-thread-commands.hpp"
  14 #include "session.hpp"
  15
  16 #include <common/common.hpp>
  17 #include <common/compat/string.hpp>
  18 #include <common/defaults.hpp>
  19 #include <common/urcu.hpp>
  20
  21 #include <inttypes.h>
  22 #include <stdio.h>
  23 #include <stdlib.h>
  24 #include <sys/stat.h>
  25 #include <unistd.h>
  26
  27 static char *create_channel_path(struct consumer_output *consumer, size_t *consumer_path_offset)
  28 {
  29         int ret;
  30         char tmp_path[PATH_MAX];
  31         char *pathname = nullptr;
  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 && consumer->relay_major_version == 2 &&
  38              consumer->relay_minor_version >= 11)) {
  39                 pathname = strdup(consumer->domain_subdir);
  40                 if (!pathname) {
  41                         PERROR("Failed to copy domain subdirectory string %s",
  42                                consumer->domain_subdir);
  43                         goto error;
  44                 }
  45                 *consumer_path_offset = strlen(consumer->domain_subdir);
  46                 DBG3("Kernel local consumer trace path relative to current trace chunk: \"%s\"",
  47                      pathname);
  48         } else {
  49                 /* Network output, relayd < 2.11. */
  50                 ret = snprintf(tmp_path,
  51                                sizeof(tmp_path),
  52                                "%s%s",
  53                                consumer->dst.net.base_dir,
  54                                consumer->domain_subdir);
  55                 if (ret < 0) {
  56                         PERROR("snprintf kernel metadata path");
  57                         goto error;
  58                 } else if (ret >= sizeof(tmp_path)) {
  59                         ERR("Kernel channel path exceeds the maximal allowed length of of %zu bytes (%i bytes required) with path \"%s%s\"",
  60                             sizeof(tmp_path),
  61                             ret,
  62                             consumer->dst.net.base_dir,
  63                             consumer->domain_subdir);
  64                         goto error;
  65                 }
  66                 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
  67                 if (!pathname) {
  68                         PERROR("lttng_strndup");
  69                         goto error;
  70                 }
  71                 *consumer_path_offset = 0;
  72                 DBG3("Kernel network consumer subdir path: %s", pathname);
  73         }
  74
  75         return pathname;
  76
  77 error:
  78         free(pathname);
  79         return nullptr;
  80 }
  81
  82 /*
  83  * Sending a single channel to the consumer with command ADD_CHANNEL.
  84  */
  85 static int kernel_consumer_add_channel(struct consumer_socket *sock,
  86                                        struct ltt_kernel_channel *channel,
  87                                        struct ltt_kernel_session *ksession,
  88                                        unsigned int monitor)
  89 {
  90         int ret;
  91         char *pathname = nullptr;
  92         struct lttcomm_consumer_msg lkm;
  93         struct consumer_output *consumer;
  94         enum lttng_error_code status;
  95         struct lttng_channel_extended *channel_attr_extended;
  96         bool is_local_trace;
  97         size_t consumer_path_offset = 0;
  98         lttng::urcu::read_lock_guard read_lock;
  99         ltt_session::ref session;
 100
 101         /* Safety net */
 102         LTTNG_ASSERT(channel);
 103         LTTNG_ASSERT(ksession);
 104         LTTNG_ASSERT(ksession->consumer);
 105
 106         consumer = ksession->consumer;
 107         channel_attr_extended =
 108                 (struct lttng_channel_extended *) channel->channel->attr.extended.ptr;
 109
 110         DBG("Kernel consumer adding channel %s to kernel consumer", channel->channel->name);
 111         is_local_trace = consumer->net_seq_index == -1ULL;
 112
 113         pathname = create_channel_path(consumer, &consumer_path_offset);
 114         if (!pathname) {
 115                 ret = -1;
 116                 goto error;
 117         }
 118
 119         if (is_local_trace && ksession->current_trace_chunk) {
 120                 enum lttng_trace_chunk_status chunk_status;
 121                 char *pathname_index;
 122
 123                 ret = asprintf(&pathname_index, "%s/" DEFAULT_INDEX_DIR, 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(ksession->current_trace_chunk,
 135                                                                      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
 170         try {
 171                 session = ltt_session::find_session(ksession->id);
 172         } catch (const lttng::sessiond::exceptions::session_not_found_error& ex) {
 173                 ERR_FMT("Fatal error during the creation of a kernel channel: {}, location='{}'",
 174                         ex.what(),
 175                         ex.source_location);
 176                 abort();
 177         }
 178
 179         ASSERT_LOCKED(session->_lock);
 180         ASSERT_SESSION_LIST_LOCKED();
 181
 182         status = notification_thread_command_add_channel(the_notification_thread_handle,
 183                                                          session->id,
 184                                                          channel->channel->name,
 185                                                          channel->key,
 186                                                          LTTNG_DOMAIN_KERNEL,
 187                                                          channel->channel->attr.subbuf_size *
 188                                                                  channel->channel->attr.num_subbuf);
 189         if (status != LTTNG_OK) {
 190                 ret = -1;
 191                 goto error;
 192         }
 193
 194         channel->published_to_notification_thread = true;
 195
 196 error:
 197         free(pathname);
 198         return ret;
 199 }
 200
 201 /*
 202  * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 203  *
 204  * The consumer socket lock must be held by the caller.
 205  */
 206 int kernel_consumer_add_metadata(struct consumer_socket *sock,
 207                                  struct ltt_kernel_session *ksession,
 208                                  unsigned int monitor)
 209 {
 210         int ret;
 211         struct lttcomm_consumer_msg lkm;
 212         struct consumer_output *consumer;
 213
 214         lttng::urcu::read_lock_guard read_lock;
 215
 216         /* Safety net */
 217         LTTNG_ASSERT(ksession);
 218         LTTNG_ASSERT(ksession->consumer);
 219         LTTNG_ASSERT(sock);
 220
 221         DBG("Sending metadata %d to kernel consumer", ksession->metadata_stream_fd);
 222
 223         /* Get consumer output pointer */
 224         consumer = ksession->consumer;
 225
 226         /* Prep channel message structure */
 227         consumer_init_add_channel_comm_msg(&lkm,
 228                                            ksession->metadata->key,
 229                                            ksession->id,
 230                                            "",
 231                                            consumer->net_seq_index,
 232                                            ksession->metadata->conf->name,
 233                                            1,
 234                                            ksession->metadata->conf->attr.output,
 235                                            CONSUMER_CHANNEL_TYPE_METADATA,
 236                                            ksession->metadata->conf->attr.tracefile_size,
 237                                            ksession->metadata->conf->attr.tracefile_count,
 238                                            monitor,
 239                                            ksession->metadata->conf->attr.live_timer_interval,
 240                                            ksession->is_live_session,
 241                                            0,
 242                                            ksession->current_trace_chunk);
 243
 244         health_code_update();
 245
 246         ret = consumer_send_channel(sock, &lkm);
 247         if (ret < 0) {
 248                 goto error;
 249         }
 250
 251         health_code_update();
 252
 253         /* Prep stream message structure */
 254         consumer_init_add_stream_comm_msg(&lkm,
 255                                           ksession->metadata->key,
 256                                           ksession->metadata_stream_fd,
 257                                           0 /* CPU: 0 for metadata. */);
 258
 259         health_code_update();
 260
 261         /* Send stream and file descriptor */
 262         ret = consumer_send_stream(sock, consumer, &lkm, &ksession->metadata_stream_fd, 1);
 263         if (ret < 0) {
 264                 goto error;
 265         }
 266
 267         health_code_update();
 268
 269 error:
 270         return ret;
 271 }
 272
 273 /*
 274  * Sending a single stream to the consumer with command ADD_STREAM.
 275  */
 276 static int kernel_consumer_add_stream(struct consumer_socket *sock,
 277                                       struct ltt_kernel_channel *channel,
 278                                       struct ltt_kernel_stream *stream,
 279                                       struct ltt_kernel_session *session)
 280 {
 281         int ret;
 282         struct lttcomm_consumer_msg lkm;
 283         struct consumer_output *consumer;
 284
 285         LTTNG_ASSERT(channel);
 286         LTTNG_ASSERT(stream);
 287         LTTNG_ASSERT(session);
 288         LTTNG_ASSERT(session->consumer);
 289         LTTNG_ASSERT(sock);
 290
 291         DBG("Sending stream %d of channel %s to kernel consumer",
 292             stream->fd,
 293             channel->channel->name);
 294
 295         /* Get consumer output pointer */
 296         consumer = session->consumer;
 297
 298         /* Prep stream consumer message */
 299         consumer_init_add_stream_comm_msg(&lkm, channel->key, stream->fd, stream->cpu);
 300
 301         health_code_update();
 302
 303         /* Send stream and file descriptor */
 304         ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
 305         if (ret < 0) {
 306                 goto error;
 307         }
 308
 309         health_code_update();
 310
 311 error:
 312         return ret;
 313 }
 314
 315 /*
 316  * Sending the notification that all streams were sent with STREAMS_SENT.
 317  */
 318 int kernel_consumer_streams_sent(struct consumer_socket *sock,
 319                                  struct ltt_kernel_session *session,
 320                                  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(
 335                 &lkm, LTTNG_CONSUMER_STREAMS_SENT, channel_key, consumer->net_seq_index);
 336
 337         health_code_update();
 338
 339         /* Send stream and file descriptor */
 340         ret = consumer_send_msg(sock, &lkm);
 341         if (ret < 0) {
 342                 goto error;
 343         }
 344
 345 error:
 346         return ret;
 347 }
 348
 349 /*
 350  * Send all stream fds of kernel channel to the consumer.
 351  *
 352  * The consumer socket lock must be held by the caller.
 353  */
 354 int kernel_consumer_send_channel_streams(struct consumer_socket *sock,
 355                                          struct ltt_kernel_channel *channel,
 356                                          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         lttng::urcu::read_lock_guard 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", channel->channel->name);
 377
 378         if (!channel->sent_to_consumer) {
 379                 ret = kernel_consumer_add_channel(sock, channel, ksession, monitor);
 380                 if (ret < 0) {
 381                         goto error;
 382                 }
 383                 channel->sent_to_consumer = true;
 384         }
 385
 386         /* Send streams */
 387         cds_list_for_each_entry (stream, &channel->stream_list.head, list) {
 388                 if (!stream->fd || stream->sent_to_consumer) {
 389                         continue;
 390                 }
 391
 392                 /* Add stream on the kernel consumer side. */
 393                 ret = kernel_consumer_add_stream(sock, channel, stream, ksession);
 394                 if (ret < 0) {
 395                         goto error;
 396                 }
 397                 stream->sent_to_consumer = true;
 398         }
 399
 400 error:
 401         return ret;
 402 }
 403
 404 /*
 405  * Send all stream fds of the kernel session to the consumer.
 406  *
 407  * The consumer socket lock must be held by the caller.
 408  */
 409 int kernel_consumer_send_session(struct consumer_socket *sock, struct ltt_kernel_session *session)
 410 {
 411         int ret, monitor = 0;
 412         struct ltt_kernel_channel *chan;
 413
 414         /* Safety net */
 415         LTTNG_ASSERT(session);
 416         LTTNG_ASSERT(session->consumer);
 417         LTTNG_ASSERT(sock);
 418
 419         /* Bail out if consumer is disabled */
 420         if (!session->consumer->enabled) {
 421                 ret = LTTNG_OK;
 422                 goto error;
 423         }
 424
 425         /* Don't monitor the streams on the consumer if in flight recorder. */
 426         if (session->output_traces) {
 427                 monitor = 1;
 428         }
 429
 430         DBG("Sending session stream to kernel consumer");
 431
 432         if (session->metadata_stream_fd >= 0 && session->metadata) {
 433                 ret = kernel_consumer_add_metadata(sock, session, monitor);
 434                 if (ret < 0) {
 435                         goto error;
 436                 }
 437         }
 438
 439         /* Send channel and streams of it */
 440         cds_list_for_each_entry (chan, &session->channel_list.head, list) {
 441                 ret = kernel_consumer_send_channel_streams(sock, chan, session, monitor);
 442                 if (ret < 0) {
 443                         goto error;
 444                 }
 445                 if (monitor) {
 446                         /*
 447                          * Inform the relay that all the streams for the
 448                          * channel were sent.
 449                          */
 450                         ret = kernel_consumer_streams_sent(sock, session, chan->key);
 451                         if (ret < 0) {
 452                                 goto error;
 453                         }
 454                 }
 455         }
 456
 457         DBG("Kernel consumer FDs of metadata and channel streams sent");
 458
 459         session->consumer_fds_sent = 1;
 460         return 0;
 461
 462 error:
 463         return ret;
 464 }
 465
 466 int kernel_consumer_destroy_channel(struct consumer_socket *socket,
 467                                     struct ltt_kernel_channel *channel)
 468 {
 469         int ret;
 470         struct lttcomm_consumer_msg msg;
 471
 472         LTTNG_ASSERT(channel);
 473         LTTNG_ASSERT(socket);
 474
 475         DBG("Sending kernel consumer destroy channel key %" PRIu64, channel->key);
 476
 477         memset(&msg, 0, sizeof(msg));
 478         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 479         msg.u.destroy_channel.key = channel->key;
 480
 481         pthread_mutex_lock(socket->lock);
 482         health_code_update();
 483
 484         ret = consumer_send_msg(socket, &msg);
 485         if (ret < 0) {
 486                 goto error;
 487         }
 488
 489 error:
 490         health_code_update();
 491         pthread_mutex_unlock(socket->lock);
 492         return ret;
 493 }
 494
 495 int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
 496                                      struct ltt_kernel_metadata *metadata)
 497 {
 498         int ret;
 499         struct lttcomm_consumer_msg msg;
 500
 501         LTTNG_ASSERT(metadata);
 502         LTTNG_ASSERT(socket);
 503
 504         DBG("Sending kernel consumer destroy channel key %" PRIu64, metadata->key);
 505
 506         memset(&msg, 0, sizeof(msg));
 507         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 508         msg.u.destroy_channel.key = metadata->key;
 509
 510         pthread_mutex_lock(socket->lock);
 511         health_code_update();
 512
 513         ret = consumer_send_msg(socket, &msg);
 514         if (ret < 0) {
 515                 goto error;
 516         }
 517
 518 error:
 519         health_code_update();
 520         pthread_mutex_unlock(socket->lock);
 521         return ret;
 522 }