src/bin/lttng-sessiond/ust-consumer.cpp

   1 /*
   2  * Copyright (C) 2011 EfficiOS Inc.
   3  *
   4  * SPDX-License-Identifier: GPL-2.0-only
   5  *
   6  */
   7
   8 #define _LGPL_SOURCE
   9 #include "buffer-registry.hpp"
  10 #include "consumer.hpp"
  11 #include "health-sessiond.hpp"
  12 #include "lttng-sessiond.hpp"
  13 #include "lttng-ust-error.hpp"
  14 #include "session.hpp"
  15 #include "ust-consumer.hpp"
  16
  17 #include <common/common.hpp>
  18 #include <common/compat/errno.hpp>
  19 #include <common/consumer/consumer.hpp>
  20 #include <common/defaults.hpp>
  21
  22 #include <inttypes.h>
  23 #include <stdio.h>
  24 #include <stdlib.h>
  25 #include <string.h>
  26 #include <unistd.h>
  27
  28 namespace lsu = lttng::sessiond::ust;
  29
  30 /*
  31  * Send a single channel to the consumer using command ASK_CHANNEL_CREATION.
  32  *
  33  * Consumer socket lock MUST be acquired before calling this.
  34  */
  35 static int ask_channel_creation(struct ust_app_session *ua_sess,
  36                                 struct ust_app_channel *ua_chan,
  37                                 struct consumer_output *consumer,
  38                                 struct consumer_socket *socket,
  39                                 lsu::registry_session *registry,
  40                                 struct lttng_trace_chunk *trace_chunk)
  41 {
  42         int ret, output;
  43         uint32_t chan_id;
  44         uint64_t key, chan_reg_key;
  45         char *pathname = nullptr;
  46         struct lttcomm_consumer_msg msg;
  47         char shm_path[PATH_MAX] = "";
  48         char root_shm_path[PATH_MAX] = "";
  49         bool is_local_trace;
  50         size_t consumer_path_offset = 0;
  51
  52         LTTNG_ASSERT(ua_sess);
  53         LTTNG_ASSERT(ua_chan);
  54         LTTNG_ASSERT(socket);
  55         LTTNG_ASSERT(consumer);
  56         LTTNG_ASSERT(registry);
  57
  58         DBG2("Asking UST consumer for channel");
  59
  60         is_local_trace = consumer->net_seq_index == -1ULL;
  61         /* Format the channel's path (relative to the current trace chunk). */
  62         pathname = setup_channel_trace_path(consumer, ua_sess->path, &consumer_path_offset);
  63         if (!pathname) {
  64                 ret = -1;
  65                 goto error;
  66         }
  67
  68         if (is_local_trace && trace_chunk) {
  69                 enum lttng_trace_chunk_status chunk_status;
  70                 char *pathname_index;
  71
  72                 ret = asprintf(&pathname_index, "%s/" DEFAULT_INDEX_DIR, pathname);
  73                 if (ret < 0) {
  74                         ERR("Failed to format channel index directory");
  75                         ret = -1;
  76                         goto error;
  77                 }
  78
  79                 /*
  80                  * Create the index subdirectory which will take care
  81                  * of implicitly creating the channel's path.
  82                  */
  83                 chunk_status = lttng_trace_chunk_create_subdirectory(trace_chunk, pathname_index);
  84                 free(pathname_index);
  85                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
  86                         ret = -1;
  87                         goto error;
  88                 }
  89         }
  90
  91         /* Depending on the buffer type, a different channel key is used. */
  92         if (ua_sess->buffer_type == LTTNG_BUFFER_PER_UID) {
  93                 chan_reg_key = ua_chan->tracing_channel_id;
  94         } else {
  95                 chan_reg_key = ua_chan->key;
  96         }
  97
  98         if (ua_chan->attr.type == LTTNG_UST_ABI_CHAN_METADATA) {
  99                 chan_id = -1U;
 100                 /*
 101                  * Metadata channels shm_path (buffers) are handled within
 102                  * session daemon. Consumer daemon should not try to create
 103                  * those buffer files.
 104                  */
 105         } else {
 106                 {
 107                         auto locked_registry = registry->lock();
 108                         auto& ust_reg_chan = registry->channel(chan_reg_key);
 109
 110                         chan_id = ust_reg_chan.id;
 111                 }
 112
 113                 if (ua_sess->shm_path[0]) {
 114                         strncpy(shm_path, ua_sess->shm_path, sizeof(shm_path));
 115                         shm_path[sizeof(shm_path) - 1] = '\0';
 116                         strncat(shm_path, "/", sizeof(shm_path) - strlen(shm_path) - 1);
 117                         strncat(shm_path, ua_chan->name, sizeof(shm_path) - strlen(shm_path) - 1);
 118                         strncat(shm_path, "_", sizeof(shm_path) - strlen(shm_path) - 1);
 119                 }
 120                 strncpy(root_shm_path, ua_sess->root_shm_path, sizeof(root_shm_path));
 121                 root_shm_path[sizeof(root_shm_path) - 1] = '\0';
 122         }
 123
 124         switch (ua_chan->attr.output) {
 125         case LTTNG_UST_ABI_MMAP:
 126         default:
 127                 output = LTTNG_EVENT_MMAP;
 128                 break;
 129         }
 130
 131         consumer_init_ask_channel_comm_msg(&msg,
 132                                            ua_chan->attr.subbuf_size,
 133                                            ua_chan->attr.num_subbuf,
 134                                            ua_chan->attr.overwrite,
 135                                            ua_chan->attr.switch_timer_interval,
 136                                            ua_chan->attr.read_timer_interval,
 137                                            ua_sess->live_timer_interval,
 138                                            ua_sess->live_timer_interval != 0,
 139                                            ua_chan->monitor_timer_interval,
 140                                            output,
 141                                            (int) ua_chan->attr.type,
 142                                            ua_sess->tracing_id,
 143                                            &pathname[consumer_path_offset],
 144                                            ua_chan->name,
 145                                            consumer->net_seq_index,
 146                                            ua_chan->key,
 147                                            registry->uuid,
 148                                            chan_id,
 149                                            ua_chan->tracefile_size,
 150                                            ua_chan->tracefile_count,
 151                                            ua_sess->id,
 152                                            ua_sess->output_traces,
 153                                            lttng_credentials_get_uid(&ua_sess->real_credentials),
 154                                            ua_chan->attr.blocking_timeout,
 155                                            root_shm_path,
 156                                            shm_path,
 157                                            trace_chunk,
 158                                            &ua_sess->effective_credentials);
 159
 160         health_code_update();
 161
 162         ret = consumer_socket_send(socket, &msg, sizeof(msg));
 163         if (ret < 0) {
 164                 goto error;
 165         }
 166
 167         ret = consumer_recv_status_channel(socket, &key, &ua_chan->expected_stream_count);
 168         if (ret < 0) {
 169                 goto error;
 170         }
 171         /* Communication protocol error. */
 172         LTTNG_ASSERT(key == ua_chan->key);
 173         /* We need at least one where 1 stream for 1 cpu. */
 174         if (ua_sess->output_traces) {
 175                 LTTNG_ASSERT(ua_chan->expected_stream_count > 0);
 176         }
 177
 178         DBG2("UST ask channel %" PRIu64 " successfully done with %u stream(s)",
 179              key,
 180              ua_chan->expected_stream_count);
 181
 182 error:
 183         free(pathname);
 184         health_code_update();
 185         return ret;
 186 }
 187
 188 /*
 189  * Ask consumer to create a channel for a given session.
 190  *
 191  * Session list and rcu read side locks must be held by the caller.
 192  *
 193  * Returns 0 on success else a negative value.
 194  */
 195 int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
 196                              struct ust_app_channel *ua_chan,
 197                              struct consumer_output *consumer,
 198                              struct consumer_socket *socket,
 199                              lsu::registry_session *registry,
 200                              struct lttng_trace_chunk *trace_chunk)
 201 {
 202         int ret;
 203
 204         LTTNG_ASSERT(ua_sess);
 205         LTTNG_ASSERT(ua_chan);
 206         LTTNG_ASSERT(consumer);
 207         LTTNG_ASSERT(socket);
 208         LTTNG_ASSERT(registry);
 209
 210         if (!consumer->enabled) {
 211                 ret = -LTTNG_ERR_NO_CONSUMER;
 212                 DBG3("Consumer is disabled");
 213                 goto error;
 214         }
 215
 216         pthread_mutex_lock(socket->lock);
 217         ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket, registry, trace_chunk);
 218         pthread_mutex_unlock(socket->lock);
 219         if (ret < 0) {
 220                 ERR("ask_channel_creation consumer command failed");
 221                 goto error;
 222         }
 223
 224 error:
 225         return ret;
 226 }
 227
 228 /*
 229  * Send a get channel command to consumer using the given channel key.  The
 230  * channel object is populated and the stream list.
 231  *
 232  * Return 0 on success else a negative value.
 233  */
 234 int ust_consumer_get_channel(struct consumer_socket *socket, struct ust_app_channel *ua_chan)
 235 {
 236         int ret;
 237         struct lttcomm_consumer_msg msg;
 238
 239         LTTNG_ASSERT(ua_chan);
 240         LTTNG_ASSERT(socket);
 241
 242         memset(&msg, 0, sizeof(msg));
 243         msg.cmd_type = LTTNG_CONSUMER_GET_CHANNEL;
 244         msg.u.get_channel.key = ua_chan->key;
 245
 246         pthread_mutex_lock(socket->lock);
 247         health_code_update();
 248
 249         /* Send command and wait for OK reply. */
 250         ret = consumer_send_msg(socket, &msg);
 251         if (ret < 0) {
 252                 goto error;
 253         }
 254
 255         /* First, get the channel from consumer. */
 256         ret = lttng_ust_ctl_recv_channel_from_consumer(*socket->fd_ptr, &ua_chan->obj);
 257         if (ret < 0) {
 258                 if (ret != -EPIPE) {
 259                         ERR("Error recv channel from consumer %d with ret %d",
 260                             *socket->fd_ptr,
 261                             ret);
 262                 } else {
 263                         DBG3("UST app recv channel from consumer. Consumer is dead.");
 264                 }
 265                 goto error;
 266         }
 267
 268         /* Next, get all streams. */
 269         while (true) {
 270                 struct ust_app_stream *stream;
 271
 272                 /* Create UST stream */
 273                 stream = ust_app_alloc_stream();
 274                 if (stream == nullptr) {
 275                         ret = -ENOMEM;
 276                         goto error;
 277                 }
 278
 279                 /* Stream object is populated by this call if successful. */
 280                 ret = lttng_ust_ctl_recv_stream_from_consumer(*socket->fd_ptr, &stream->obj);
 281                 if (ret < 0) {
 282                         free(stream);
 283                         if (ret == -LTTNG_UST_ERR_NOENT) {
 284                                 DBG3("UST app consumer has no more stream available");
 285                                 break;
 286                         }
 287                         if (ret != -EPIPE) {
 288                                 ERR("Recv stream from consumer %d with ret %d",
 289                                     *socket->fd_ptr,
 290                                     ret);
 291                         } else {
 292                                 DBG3("UST app recv stream from consumer. Consumer is dead.");
 293                         }
 294                         goto error;
 295                 }
 296
 297                 /* Order is important this is why a list is used. */
 298                 cds_list_add_tail(&stream->list, &ua_chan->streams.head);
 299                 ua_chan->streams.count++;
 300
 301                 DBG2("UST app stream %d received successfully", ua_chan->streams.count);
 302         }
 303
 304         /* This MUST match or else we have a synchronization problem. */
 305         LTTNG_ASSERT(ua_chan->expected_stream_count == ua_chan->streams.count);
 306
 307         /* Wait for confirmation that we can proceed with the streams. */
 308         ret = consumer_recv_status_reply(socket);
 309         if (ret < 0) {
 310                 goto error;
 311         }
 312
 313 error:
 314         health_code_update();
 315         pthread_mutex_unlock(socket->lock);
 316         return ret;
 317 }
 318
 319 /*
 320  * Send a destroy channel command to consumer using the given channel key.
 321  *
 322  * Note that this command MUST be used prior to a successful
 323  * LTTNG_CONSUMER_GET_CHANNEL because once this command is done successfully,
 324  * the streams are dispatched to the consumer threads and MUST be teardown
 325  * through the hang up process.
 326  *
 327  * Return 0 on success else a negative value.
 328  */
 329 int ust_consumer_destroy_channel(struct consumer_socket *socket, struct ust_app_channel *ua_chan)
 330 {
 331         int ret;
 332         struct lttcomm_consumer_msg msg;
 333
 334         LTTNG_ASSERT(ua_chan);
 335         LTTNG_ASSERT(socket);
 336
 337         memset(&msg, 0, sizeof(msg));
 338         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 339         msg.u.destroy_channel.key = ua_chan->key;
 340
 341         pthread_mutex_lock(socket->lock);
 342         health_code_update();
 343
 344         ret = consumer_send_msg(socket, &msg);
 345         if (ret < 0) {
 346                 goto error;
 347         }
 348
 349 error:
 350         health_code_update();
 351         pthread_mutex_unlock(socket->lock);
 352         return ret;
 353 }
 354
 355 /*
 356  * Send a given stream to UST tracer.
 357  *
 358  * On success return 0 else a negative value.
 359  */
 360 int ust_consumer_send_stream_to_ust(struct ust_app *app,
 361                                     struct ust_app_channel *channel,
 362                                     struct ust_app_stream *stream)
 363 {
 364         int ret;
 365
 366         LTTNG_ASSERT(app);
 367         LTTNG_ASSERT(stream);
 368         LTTNG_ASSERT(channel);
 369
 370         DBG2("UST consumer send stream to app %d", app->sock);
 371
 372         /* Relay stream to application. */
 373         pthread_mutex_lock(&app->sock_lock);
 374         ret = lttng_ust_ctl_send_stream_to_ust(app->sock, channel->obj, stream->obj);
 375         pthread_mutex_unlock(&app->sock_lock);
 376         if (ret < 0) {
 377                 if (ret == -EPIPE || ret == -LTTNG_UST_ERR_EXITING) {
 378                         DBG3("UST app send stream to ust failed. Application is dead. (pid: %d, sock: %d).",
 379                              app->pid,
 380                              app->sock);
 381                 } else if (ret == -EAGAIN) {
 382                         WARN("UST app send stream to ust failed. Communication time out (pid: %d, sock: %d).",
 383                              app->pid,
 384                              app->sock);
 385                 } else {
 386                         ERR("UST app send stream, handle %d, to ust failed with ret %d (pid: %d, sock: %d).",
 387                             stream->obj->handle,
 388                             ret,
 389                             app->pid,
 390                             app->sock);
 391                 }
 392                 goto error;
 393         }
 394         channel->handle = channel->obj->handle;
 395
 396 error:
 397         return ret;
 398 }
 399
 400 /*
 401  * Send channel previously received from the consumer to the UST tracer.
 402  *
 403  * On success return 0 else a negative value.
 404  */
 405 int ust_consumer_send_channel_to_ust(struct ust_app *app,
 406                                      struct ust_app_session *ua_sess,
 407                                      struct ust_app_channel *channel)
 408 {
 409         int ret;
 410
 411         LTTNG_ASSERT(app);
 412         LTTNG_ASSERT(ua_sess);
 413         LTTNG_ASSERT(channel);
 414         LTTNG_ASSERT(channel->obj);
 415
 416         DBG2("UST app send channel to sock %d pid %d (name: %s, key: %" PRIu64 ")",
 417              app->sock,
 418              app->pid,
 419              channel->name,
 420              channel->tracing_channel_id);
 421
 422         /* Send stream to application. */
 423         pthread_mutex_lock(&app->sock_lock);
 424         ret = lttng_ust_ctl_send_channel_to_ust(app->sock, ua_sess->handle, channel->obj);
 425         pthread_mutex_unlock(&app->sock_lock);
 426         if (ret < 0) {
 427                 if (ret == -EPIPE || ret == -LTTNG_UST_ERR_EXITING) {
 428                         DBG3("UST app send channel to ust failed. Application is dead (pid: %d, sock: %d).",
 429                              app->pid,
 430                              app->sock);
 431                 } else if (ret == -EAGAIN) {
 432                         WARN("UST app send channel to ust failed. Communication timeout (pid: %d, sock: %d).",
 433                              app->pid,
 434                              app->sock);
 435                 } else {
 436                         ERR("UST app send channel %s, to ust failed with ret %d (pid: %d, sock: %d).",
 437                             channel->name,
 438                             ret,
 439                             app->pid,
 440                             app->sock);
 441                 }
 442                 goto error;
 443         }
 444
 445 error:
 446         return ret;
 447 }
 448
 449 /*
 450  * Handle the metadata requests from the UST consumer
 451  *
 452  * Return 0 on success else a negative value.
 453  */
 454 int ust_consumer_metadata_request(struct consumer_socket *socket)
 455 {
 456         int ret;
 457         ssize_t ret_push;
 458         struct lttcomm_metadata_request_msg request;
 459         struct buffer_reg_uid *reg_uid;
 460         lsu::registry_session *ust_reg;
 461         struct lttcomm_consumer_msg msg;
 462
 463         LTTNG_ASSERT(socket);
 464
 465         rcu_read_lock();
 466         health_code_update();
 467
 468         /* Wait for a metadata request */
 469         pthread_mutex_lock(socket->lock);
 470         ret = consumer_socket_recv(socket, &request, sizeof(request));
 471         pthread_mutex_unlock(socket->lock);
 472         if (ret < 0) {
 473                 goto end;
 474         }
 475
 476         DBG("Metadata request received for session %" PRIu64 ", key %" PRIu64,
 477             request.session_id,
 478             request.key);
 479
 480         reg_uid = buffer_reg_uid_find(request.session_id, request.bits_per_long, request.uid);
 481         if (reg_uid) {
 482                 ust_reg = reg_uid->registry->reg.ust;
 483         } else {
 484                 struct buffer_reg_pid *reg_pid = buffer_reg_pid_find(request.session_id_per_pid);
 485                 if (!reg_pid) {
 486                         DBG("PID registry not found for session id %" PRIu64,
 487                             request.session_id_per_pid);
 488
 489                         memset(&msg, 0, sizeof(msg));
 490                         msg.cmd_type = LTTNG_ERR_UND;
 491                         pthread_mutex_lock(socket->lock);
 492                         (void) consumer_send_msg(socket, &msg);
 493                         pthread_mutex_unlock(socket->lock);
 494                         /*
 495                          * This is possible since the session might have been destroyed
 496                          * during a consumer metadata request. So here, return gracefully
 497                          * because the destroy session will push the remaining metadata to
 498                          * the consumer.
 499                          */
 500                         ret = 0;
 501                         goto end;
 502                 }
 503                 ust_reg = reg_pid->registry->reg.ust;
 504         }
 505         LTTNG_ASSERT(ust_reg);
 506
 507         {
 508                 auto locked_ust_reg = ust_reg->lock();
 509                 ret_push = ust_app_push_metadata(locked_ust_reg, socket, 1);
 510         }
 511         if (ret_push == -EPIPE) {
 512                 DBG("Application or relay closed while pushing metadata");
 513         } else if (ret_push < 0) {
 514                 ERR("Pushing metadata");
 515                 ret = -1;
 516                 goto end;
 517         } else {
 518                 DBG("UST Consumer metadata pushed successfully");
 519         }
 520         ret = 0;
 521
 522 end:
 523         rcu_read_unlock();
 524         return ret;
 525 }