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