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                         ksession->is_live_session,
 160                         channel_attr_extended->monitor_timer_interval,
 161                         ksession->current_trace_chunk);
 162
 163         health_code_update();
 164
 165         ret = consumer_send_channel(sock, &lkm);
 166         if (ret < 0) {
 167                 goto error;
 168         }
 169
 170         health_code_update();
 171         rcu_read_lock();
 172         session = session_find_by_id(ksession->id);
 173         assert(session);
 174         assert(pthread_mutex_trylock(&session->lock));
 175         assert(session_trylock_list());
 176
 177         status = notification_thread_command_add_channel(
 178                         notification_thread_handle, session->name,
 179                         ksession->uid, ksession->gid,
 180                         channel->channel->name, channel->key,
 181                         LTTNG_DOMAIN_KERNEL,
 182                         channel->channel->attr.subbuf_size * channel->channel->attr.num_subbuf);
 183         rcu_read_unlock();
 184         if (status != LTTNG_OK) {
 185                 ret = -1;
 186                 goto error;
 187         }
 188
 189         channel->published_to_notification_thread = true;
 190
 191 error:
 192         if (session) {
 193                 session_put(session);
 194         }
 195         free(pathname);
 196         return ret;
 197 }
 198
 199 /*
 200  * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 201  *
 202  * The consumer socket lock must be held by the caller.
 203  */
 204 int kernel_consumer_add_metadata(struct consumer_socket *sock,
 205                 struct ltt_kernel_session *ksession, unsigned int monitor)
 206 {
 207         int ret;
 208         struct lttcomm_consumer_msg lkm;
 209         struct consumer_output *consumer;
 210
 211         rcu_read_lock();
 212
 213         /* Safety net */
 214         assert(ksession);
 215         assert(ksession->consumer);
 216         assert(sock);
 217
 218         DBG("Sending metadata %d to kernel consumer",
 219                         ksession->metadata_stream_fd);
 220
 221         /* Get consumer output pointer */
 222         consumer = ksession->consumer;
 223
 224         /* Prep channel message structure */
 225         consumer_init_add_channel_comm_msg(&lkm,
 226                         ksession->metadata->key,
 227                         ksession->id,
 228                         "",
 229                         ksession->uid,
 230                         ksession->gid,
 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,
 263                         &ksession->metadata_stream_fd, 1);
 264         if (ret < 0) {
 265                 goto error;
 266         }
 267
 268         health_code_update();
 269
 270 error:
 271         rcu_read_unlock();
 272         return ret;
 273 }
 274
 275 /*
 276  * Sending a single stream to the consumer with command ADD_STREAM.
 277  */
 278 static
 279 int kernel_consumer_add_stream(struct consumer_socket *sock,
 280                 struct ltt_kernel_channel *channel,
 281                 struct ltt_kernel_stream *stream,
 282                 struct ltt_kernel_session *session, unsigned int monitor)
 283 {
 284         int ret;
 285         struct lttcomm_consumer_msg lkm;
 286         struct consumer_output *consumer;
 287
 288         assert(channel);
 289         assert(stream);
 290         assert(session);
 291         assert(session->consumer);
 292         assert(sock);
 293
 294         DBG("Sending stream %d of channel %s to kernel consumer",
 295                         stream->fd, channel->channel->name);
 296
 297         /* Get consumer output pointer */
 298         consumer = session->consumer;
 299
 300         /* Prep stream consumer message */
 301         consumer_init_add_stream_comm_msg(&lkm,
 302                         channel->key,
 303                         stream->fd,
 304                         stream->cpu);
 305
 306         health_code_update();
 307
 308         /* Send stream and file descriptor */
 309         ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
 310         if (ret < 0) {
 311                 goto error;
 312         }
 313
 314         health_code_update();
 315
 316 error:
 317         return ret;
 318 }
 319
 320 /*
 321  * Sending the notification that all streams were sent with STREAMS_SENT.
 322  */
 323 int kernel_consumer_streams_sent(struct consumer_socket *sock,
 324                 struct ltt_kernel_session *session, uint64_t channel_key)
 325 {
 326         int ret;
 327         struct lttcomm_consumer_msg lkm;
 328         struct consumer_output *consumer;
 329
 330         assert(sock);
 331         assert(session);
 332
 333         DBG("Sending streams_sent");
 334         /* Get consumer output pointer */
 335         consumer = session->consumer;
 336
 337         /* Prep stream consumer message */
 338         consumer_init_streams_sent_comm_msg(&lkm,
 339                         LTTNG_CONSUMER_STREAMS_SENT,
 340                         channel_key, consumer->net_seq_index);
 341
 342         health_code_update();
 343
 344         /* Send stream and file descriptor */
 345         ret = consumer_send_msg(sock, &lkm);
 346         if (ret < 0) {
 347                 goto error;
 348         }
 349
 350 error:
 351         return ret;
 352 }
 353
 354 /*
 355  * Send all stream fds of kernel channel to the consumer.
 356  *
 357  * The consumer socket lock must be held by the caller.
 358  */
 359 int kernel_consumer_send_channel_streams(struct consumer_socket *sock,
 360                 struct ltt_kernel_channel *channel, struct ltt_kernel_session *ksession,
 361                 unsigned int monitor)
 362 {
 363         int ret = LTTNG_OK;
 364         struct ltt_kernel_stream *stream;
 365
 366         /* Safety net */
 367         assert(channel);
 368         assert(ksession);
 369         assert(ksession->consumer);
 370         assert(sock);
 371
 372         rcu_read_lock();
 373
 374         /* Bail out if consumer is disabled */
 375         if (!ksession->consumer->enabled) {
 376                 ret = LTTNG_OK;
 377                 goto error;
 378         }
 379
 380         DBG("Sending streams of channel %s to kernel consumer",
 381                         channel->channel->name);
 382
 383         if (!channel->sent_to_consumer) {
 384                 ret = kernel_consumer_add_channel(sock, channel, ksession, monitor);
 385                 if (ret < 0) {
 386                         goto error;
 387                 }
 388                 channel->sent_to_consumer = true;
 389         }
 390
 391         /* Send streams */
 392         cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
 393                 if (!stream->fd || stream->sent_to_consumer) {
 394                         continue;
 395                 }
 396
 397                 /* Add stream on the kernel consumer side. */
 398                 ret = kernel_consumer_add_stream(sock, channel, stream,
 399                                 ksession, monitor);
 400                 if (ret < 0) {
 401                         goto error;
 402                 }
 403                 stream->sent_to_consumer = true;
 404         }
 405
 406 error:
 407         rcu_read_unlock();
 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_streams(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 }