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