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                         channel_attr_extended->monitor_timer_interval,
 166                         ksession->current_trace_chunk);
 167
 168         health_code_update();
 169
 170         ret = consumer_send_channel(sock, &lkm);
 171         if (ret < 0) {
 172                 goto error;
 173         }
 174
 175         health_code_update();
 176         rcu_read_lock();
 177         session = session_find_by_id(ksession->id);
 178         assert(session);
 179         assert(pthread_mutex_trylock(&session->lock));
 180         assert(session_trylock_list());
 181
 182         status = notification_thread_command_add_channel(
 183                         notification_thread_handle, session->name,
 184                         ksession->uid, ksession->gid,
 185                         channel->channel->name, channel->key,
 186                         LTTNG_DOMAIN_KERNEL,
 187                         channel->channel->attr.subbuf_size * channel->channel->attr.num_subbuf);
 188         rcu_read_unlock();
 189         if (status != LTTNG_OK) {
 190                 ret = -1;
 191                 goto error;
 192         }
 193
 194         channel->published_to_notification_thread = true;
 195
 196 error:
 197         if (session) {
 198                 session_put(session);
 199         }
 200         free(pathname);
 201         return ret;
 202 }
 203
 204 /*
 205  * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 206  *
 207  * The consumer socket lock must be held by the caller.
 208  */
 209 int kernel_consumer_add_metadata(struct consumer_socket *sock,
 210                 struct ltt_kernel_session *ksession, unsigned int monitor)
 211 {
 212         int ret;
 213         struct lttcomm_consumer_msg lkm;
 214         struct consumer_output *consumer;
 215
 216         rcu_read_lock();
 217
 218         /* Safety net */
 219         assert(ksession);
 220         assert(ksession->consumer);
 221         assert(sock);
 222
 223         DBG("Sending metadata %d to kernel consumer",
 224                         ksession->metadata_stream_fd);
 225
 226         /* Get consumer output pointer */
 227         consumer = ksession->consumer;
 228
 229         /* Prep channel message structure */
 230         consumer_init_add_channel_comm_msg(&lkm,
 231                         ksession->metadata->key,
 232                         ksession->id,
 233                         DEFAULT_KERNEL_TRACE_DIR,
 234                         ksession->uid,
 235                         ksession->gid,
 236                         consumer->net_seq_index,
 237                         DEFAULT_METADATA_NAME,
 238                         1,
 239                         DEFAULT_KERNEL_CHANNEL_OUTPUT,
 240                         CONSUMER_CHANNEL_TYPE_METADATA,
 241                         0, 0,
 242                         monitor, 0, 0, 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 }