Remove fcntl wrapper
[lttng-tools.git] / src / bin / lttng-sessiond / kernel-consumer.cpp
CommitLineData
f1e16794 1/*
ab5be9fa 2 * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
f1e16794 3 *
ab5be9fa 4 * SPDX-License-Identifier: GPL-2.0-only
f1e16794 5 *
f1e16794
DG
6 */
7
6c1c0768 8#define _LGPL_SOURCE
c9e313bc
SM
9#include "consumer.hpp"
10#include "health-sessiond.hpp"
11#include "kernel-consumer.hpp"
28ab034a 12#include "lttng-sessiond.hpp"
c9e313bc
SM
13#include "notification-thread-commands.hpp"
14#include "session.hpp"
f1e16794 15
28ab034a
JG
16#include <common/common.hpp>
17#include <common/compat/string.hpp>
18#include <common/defaults.hpp>
56047f5a 19#include <common/urcu.hpp>
28ab034a
JG
20
21#include <inttypes.h>
22#include <stdio.h>
23#include <stdlib.h>
24#include <sys/stat.h>
25#include <unistd.h>
26
27static char *create_channel_path(struct consumer_output *consumer, size_t *consumer_path_offset)
00e2e675
DG
28{
29 int ret;
ffe60014 30 char tmp_path[PATH_MAX];
cd9adb8b 31 char *pathname = nullptr;
00e2e675 32
a0377dfe 33 LTTNG_ASSERT(consumer);
00e2e675 34
ffe60014 35 /* Get the right path name destination */
a5ba6fdd 36 if (consumer->type == CONSUMER_DST_LOCAL ||
28ab034a
JG
37 (consumer->type == CONSUMER_DST_NET && consumer->relay_major_version == 2 &&
38 consumer->relay_minor_version >= 11)) {
d2956687 39 pathname = strdup(consumer->domain_subdir);
bb3c4e70 40 if (!pathname) {
d2956687 41 PERROR("Failed to copy domain subdirectory string %s",
28ab034a 42 consumer->domain_subdir);
bb3c4e70
MD
43 goto error;
44 }
5da88b0f 45 *consumer_path_offset = strlen(consumer->domain_subdir);
d2956687 46 DBG3("Kernel local consumer trace path relative to current trace chunk: \"%s\"",
28ab034a 47 pathname);
ffe60014 48 } else {
5da88b0f 49 /* Network output, relayd < 2.11. */
28ab034a
JG
50 ret = snprintf(tmp_path,
51 sizeof(tmp_path),
52 "%s%s",
53 consumer->dst.net.base_dir,
54 consumer->domain_subdir);
ffe60014 55 if (ret < 0) {
2bba9e53 56 PERROR("snprintf kernel metadata path");
ffe60014 57 goto error;
dba13f1d
JG
58 } else if (ret >= sizeof(tmp_path)) {
59 ERR("Kernel channel path exceeds the maximal allowed length of of %zu bytes (%i bytes required) with path \"%s%s\"",
28ab034a
JG
60 sizeof(tmp_path),
61 ret,
62 consumer->dst.net.base_dir,
63 consumer->domain_subdir);
dba13f1d 64 goto error;
ffe60014 65 }
f5436bfc 66 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
bb3c4e70 67 if (!pathname) {
f5436bfc 68 PERROR("lttng_strndup");
bb3c4e70
MD
69 goto error;
70 }
5da88b0f 71 *consumer_path_offset = 0;
ffe60014
DG
72 DBG3("Kernel network consumer subdir path: %s", pathname);
73 }
74
2bba9e53
DG
75 return pathname;
76
77error:
78 free(pathname);
cd9adb8b 79 return nullptr;
2bba9e53
DG
80}
81
82/*
83 * Sending a single channel to the consumer with command ADD_CHANNEL.
84 */
28ab034a
JG
85static int kernel_consumer_add_channel(struct consumer_socket *sock,
86 struct ltt_kernel_channel *channel,
87 struct ltt_kernel_session *ksession,
88 unsigned int monitor)
2bba9e53
DG
89{
90 int ret;
cd9adb8b 91 char *pathname = nullptr;
2bba9e53
DG
92 struct lttcomm_consumer_msg lkm;
93 struct consumer_output *consumer;
e9404c27 94 enum lttng_error_code status;
cd9adb8b 95 struct ltt_session *session = nullptr;
e9404c27 96 struct lttng_channel_extended *channel_attr_extended;
d2956687 97 bool is_local_trace;
5da88b0f 98 size_t consumer_path_offset = 0;
56047f5a 99 lttng::urcu::read_lock_guard read_lock;
2bba9e53
DG
100
101 /* Safety net */
a0377dfe
FD
102 LTTNG_ASSERT(channel);
103 LTTNG_ASSERT(ksession);
104 LTTNG_ASSERT(ksession->consumer);
2bba9e53 105
e9404c27 106 consumer = ksession->consumer;
28ab034a
JG
107 channel_attr_extended =
108 (struct lttng_channel_extended *) channel->channel->attr.extended.ptr;
2bba9e53 109
28ab034a 110 DBG("Kernel consumer adding channel %s to kernel consumer", channel->channel->name);
d2956687 111 is_local_trace = consumer->net_seq_index == -1ULL;
2bba9e53 112
5da88b0f 113 pathname = create_channel_path(consumer, &consumer_path_offset);
bb3c4e70
MD
114 if (!pathname) {
115 ret = -1;
116 goto error;
117 }
2bba9e53 118
d2956687
JG
119 if (is_local_trace && ksession->current_trace_chunk) {
120 enum lttng_trace_chunk_status chunk_status;
121 char *pathname_index;
122
28ab034a 123 ret = asprintf(&pathname_index, "%s/" DEFAULT_INDEX_DIR, pathname);
d2956687
JG
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 */
28ab034a
JG
134 chunk_status = lttng_trace_chunk_create_subdirectory(ksession->current_trace_chunk,
135 pathname_index);
d2956687
JG
136 free(pathname_index);
137 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
138 ret = -1;
139 goto error;
140 }
141 }
142
00e2e675 143 /* Prep channel message structure */
638e7b4e 144 consumer_init_add_channel_comm_msg(&lkm,
28ab034a
JG
145 channel->key,
146 ksession->id,
147 &pathname[consumer_path_offset],
148 consumer->net_seq_index,
149 channel->channel->name,
150 channel->stream_count,
151 channel->channel->attr.output,
152 CONSUMER_CHANNEL_TYPE_DATA,
153 channel->channel->attr.tracefile_size,
154 channel->channel->attr.tracefile_count,
155 monitor,
156 channel->channel->attr.live_timer_interval,
157 ksession->is_live_session,
158 channel_attr_extended->monitor_timer_interval,
159 ksession->current_trace_chunk);
00e2e675 160
840cb59c 161 health_code_update();
ca03de58 162
00e2e675
DG
163 ret = consumer_send_channel(sock, &lkm);
164 if (ret < 0) {
165 goto error;
166 }
167
840cb59c 168 health_code_update();
e9404c27 169 session = session_find_by_id(ksession->id);
a0377dfe 170 LTTNG_ASSERT(session);
3130a40c
JG
171 ASSERT_LOCKED(session->lock);
172 ASSERT_SESSION_LIST_LOCKED();
ca03de58 173
139a8d25 174 status = notification_thread_command_add_channel(the_notification_thread_handle,
28ab034a
JG
175 session->id,
176 channel->channel->name,
177 channel->key,
178 LTTNG_DOMAIN_KERNEL,
179 channel->channel->attr.subbuf_size *
180 channel->channel->attr.num_subbuf);
e9404c27
JG
181 if (status != LTTNG_OK) {
182 ret = -1;
183 goto error;
184 }
753873bf
JR
185
186 channel->published_to_notification_thread = true;
187
00e2e675 188error:
e32d7f27
JG
189 if (session) {
190 session_put(session);
191 }
53efb85a 192 free(pathname);
00e2e675
DG
193 return ret;
194}
195
196/*
197 * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
9a318688
JG
198 *
199 * The consumer socket lock must be held by the caller.
00e2e675 200 */
f50f23d9 201int kernel_consumer_add_metadata(struct consumer_socket *sock,
28ab034a
JG
202 struct ltt_kernel_session *ksession,
203 unsigned int monitor)
00e2e675
DG
204{
205 int ret;
00e2e675 206 struct lttcomm_consumer_msg lkm;
a7d9a3e7 207 struct consumer_output *consumer;
e098433c 208
56047f5a 209 lttng::urcu::read_lock_guard read_lock;
00e2e675
DG
210
211 /* Safety net */
a0377dfe
FD
212 LTTNG_ASSERT(ksession);
213 LTTNG_ASSERT(ksession->consumer);
214 LTTNG_ASSERT(sock);
00e2e675 215
28ab034a 216 DBG("Sending metadata %d to kernel consumer", ksession->metadata_stream_fd);
00e2e675
DG
217
218 /* Get consumer output pointer */
e098433c 219 consumer = ksession->consumer;
00e2e675 220
00e2e675 221 /* Prep channel message structure */
d42266a4 222 consumer_init_add_channel_comm_msg(&lkm,
28ab034a
JG
223 ksession->metadata->key,
224 ksession->id,
225 "",
226 consumer->net_seq_index,
227 ksession->metadata->conf->name,
228 1,
229 ksession->metadata->conf->attr.output,
230 CONSUMER_CHANNEL_TYPE_METADATA,
231 ksession->metadata->conf->attr.tracefile_size,
232 ksession->metadata->conf->attr.tracefile_count,
233 monitor,
234 ksession->metadata->conf->attr.live_timer_interval,
235 ksession->is_live_session,
236 0,
237 ksession->current_trace_chunk);
00e2e675 238
840cb59c 239 health_code_update();
ca03de58 240
00e2e675
DG
241 ret = consumer_send_channel(sock, &lkm);
242 if (ret < 0) {
243 goto error;
244 }
245
840cb59c 246 health_code_update();
ca03de58 247
00e2e675 248 /* Prep stream message structure */
e098433c 249 consumer_init_add_stream_comm_msg(&lkm,
28ab034a
JG
250 ksession->metadata->key,
251 ksession->metadata_stream_fd,
252 0 /* CPU: 0 for metadata. */);
00e2e675 253
840cb59c 254 health_code_update();
ca03de58 255
00e2e675 256 /* Send stream and file descriptor */
28ab034a 257 ret = consumer_send_stream(sock, consumer, &lkm, &ksession->metadata_stream_fd, 1);
00e2e675
DG
258 if (ret < 0) {
259 goto error;
260 }
261
840cb59c 262 health_code_update();
ca03de58 263
00e2e675
DG
264error:
265 return ret;
266}
267
268/*
269 * Sending a single stream to the consumer with command ADD_STREAM.
270 */
28ab034a
JG
271static int kernel_consumer_add_stream(struct consumer_socket *sock,
272 struct ltt_kernel_channel *channel,
273 struct ltt_kernel_stream *stream,
274 struct ltt_kernel_session *session)
00e2e675
DG
275{
276 int ret;
00e2e675 277 struct lttcomm_consumer_msg lkm;
a7d9a3e7 278 struct consumer_output *consumer;
00e2e675 279
a0377dfe
FD
280 LTTNG_ASSERT(channel);
281 LTTNG_ASSERT(stream);
282 LTTNG_ASSERT(session);
283 LTTNG_ASSERT(session->consumer);
284 LTTNG_ASSERT(sock);
00e2e675
DG
285
286 DBG("Sending stream %d of channel %s to kernel consumer",
28ab034a
JG
287 stream->fd,
288 channel->channel->name);
00e2e675
DG
289
290 /* Get consumer output pointer */
a7d9a3e7 291 consumer = session->consumer;
00e2e675 292
00e2e675 293 /* Prep stream consumer message */
28ab034a 294 consumer_init_add_stream_comm_msg(&lkm, channel->key, stream->fd, stream->cpu);
00e2e675 295
840cb59c 296 health_code_update();
ca03de58 297
00e2e675 298 /* Send stream and file descriptor */
a7d9a3e7 299 ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
00e2e675
DG
300 if (ret < 0) {
301 goto error;
302 }
303
840cb59c 304 health_code_update();
ca03de58 305
00e2e675
DG
306error:
307 return ret;
308}
309
a4baae1b
JD
310/*
311 * Sending the notification that all streams were sent with STREAMS_SENT.
312 */
313int kernel_consumer_streams_sent(struct consumer_socket *sock,
28ab034a
JG
314 struct ltt_kernel_session *session,
315 uint64_t channel_key)
a4baae1b
JD
316{
317 int ret;
318 struct lttcomm_consumer_msg lkm;
319 struct consumer_output *consumer;
320
a0377dfe
FD
321 LTTNG_ASSERT(sock);
322 LTTNG_ASSERT(session);
a4baae1b
JD
323
324 DBG("Sending streams_sent");
325 /* Get consumer output pointer */
326 consumer = session->consumer;
327
328 /* Prep stream consumer message */
28ab034a
JG
329 consumer_init_streams_sent_comm_msg(
330 &lkm, LTTNG_CONSUMER_STREAMS_SENT, channel_key, consumer->net_seq_index);
a4baae1b
JD
331
332 health_code_update();
333
334 /* Send stream and file descriptor */
335 ret = consumer_send_msg(sock, &lkm);
336 if (ret < 0) {
337 goto error;
338 }
339
340error:
341 return ret;
342}
343
f1e16794
DG
344/*
345 * Send all stream fds of kernel channel to the consumer.
9a318688
JG
346 *
347 * The consumer socket lock must be held by the caller.
f1e16794 348 */
1fc1b7c8 349int kernel_consumer_send_channel_streams(struct consumer_socket *sock,
28ab034a
JG
350 struct ltt_kernel_channel *channel,
351 struct ltt_kernel_session *ksession,
352 unsigned int monitor)
f1e16794 353{
e99f9447 354 int ret = LTTNG_OK;
f1e16794 355 struct ltt_kernel_stream *stream;
00e2e675
DG
356
357 /* Safety net */
a0377dfe
FD
358 LTTNG_ASSERT(channel);
359 LTTNG_ASSERT(ksession);
360 LTTNG_ASSERT(ksession->consumer);
361 LTTNG_ASSERT(sock);
00e2e675 362
56047f5a 363 lttng::urcu::read_lock_guard read_lock;
e098433c 364
00e2e675 365 /* Bail out if consumer is disabled */
e098433c 366 if (!ksession->consumer->enabled) {
f73fabfd 367 ret = LTTNG_OK;
00e2e675
DG
368 goto error;
369 }
f1e16794 370
28ab034a 371 DBG("Sending streams of channel %s to kernel consumer", channel->channel->name);
f1e16794 372
e99f9447 373 if (!channel->sent_to_consumer) {
e098433c 374 ret = kernel_consumer_add_channel(sock, channel, ksession, monitor);
e99f9447
MD
375 if (ret < 0) {
376 goto error;
377 }
378 channel->sent_to_consumer = true;
f1e16794
DG
379 }
380
381 /* Send streams */
28ab034a 382 cds_list_for_each_entry (stream, &channel->stream_list.head, list) {
6986ab9b 383 if (!stream->fd || stream->sent_to_consumer) {
f1e16794
DG
384 continue;
385 }
00e2e675
DG
386
387 /* Add stream on the kernel consumer side. */
28ab034a 388 ret = kernel_consumer_add_stream(sock, channel, stream, ksession);
f1e16794 389 if (ret < 0) {
f1e16794
DG
390 goto error;
391 }
6986ab9b 392 stream->sent_to_consumer = true;
f1e16794
DG
393 }
394
f1e16794
DG
395error:
396 return ret;
397}
398
399/*
400 * Send all stream fds of the kernel session to the consumer.
9a318688
JG
401 *
402 * The consumer socket lock must be held by the caller.
f1e16794 403 */
28ab034a 404int kernel_consumer_send_session(struct consumer_socket *sock, struct ltt_kernel_session *session)
f1e16794 405{
2bba9e53 406 int ret, monitor = 0;
f1e16794 407 struct ltt_kernel_channel *chan;
f1e16794 408
00e2e675 409 /* Safety net */
a0377dfe
FD
410 LTTNG_ASSERT(session);
411 LTTNG_ASSERT(session->consumer);
412 LTTNG_ASSERT(sock);
f1e16794 413
00e2e675
DG
414 /* Bail out if consumer is disabled */
415 if (!session->consumer->enabled) {
f73fabfd 416 ret = LTTNG_OK;
00e2e675 417 goto error;
f1e16794
DG
418 }
419
2bba9e53
DG
420 /* Don't monitor the streams on the consumer if in flight recorder. */
421 if (session->output_traces) {
422 monitor = 1;
423 }
424
00e2e675
DG
425 DBG("Sending session stream to kernel consumer");
426
609af759 427 if (session->metadata_stream_fd >= 0 && session->metadata) {
2bba9e53 428 ret = kernel_consumer_add_metadata(sock, session, monitor);
f1e16794 429 if (ret < 0) {
f1e16794
DG
430 goto error;
431 }
f1e16794
DG
432 }
433
00e2e675 434 /* Send channel and streams of it */
28ab034a
JG
435 cds_list_for_each_entry (chan, &session->channel_list.head, list) {
436 ret = kernel_consumer_send_channel_streams(sock, chan, session, monitor);
f1e16794
DG
437 if (ret < 0) {
438 goto error;
439 }
601262d6
JD
440 if (monitor) {
441 /*
442 * Inform the relay that all the streams for the
443 * channel were sent.
444 */
e1f3997a 445 ret = kernel_consumer_streams_sent(sock, session, chan->key);
601262d6
JD
446 if (ret < 0) {
447 goto error;
448 }
449 }
f1e16794
DG
450 }
451
00e2e675 452 DBG("Kernel consumer FDs of metadata and channel streams sent");
f1e16794 453
4ce9ff51 454 session->consumer_fds_sent = 1;
f1e16794
DG
455 return 0;
456
457error:
458 return ret;
459}
07b86b52
JD
460
461int kernel_consumer_destroy_channel(struct consumer_socket *socket,
28ab034a 462 struct ltt_kernel_channel *channel)
07b86b52
JD
463{
464 int ret;
465 struct lttcomm_consumer_msg msg;
466
a0377dfe
FD
467 LTTNG_ASSERT(channel);
468 LTTNG_ASSERT(socket);
07b86b52 469
e1f3997a 470 DBG("Sending kernel consumer destroy channel key %" PRIu64, channel->key);
07b86b52 471
53efb85a 472 memset(&msg, 0, sizeof(msg));
07b86b52 473 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
e1f3997a 474 msg.u.destroy_channel.key = channel->key;
07b86b52
JD
475
476 pthread_mutex_lock(socket->lock);
477 health_code_update();
478
479 ret = consumer_send_msg(socket, &msg);
480 if (ret < 0) {
481 goto error;
482 }
483
484error:
485 health_code_update();
486 pthread_mutex_unlock(socket->lock);
487 return ret;
488}
489
490int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
28ab034a 491 struct ltt_kernel_metadata *metadata)
07b86b52
JD
492{
493 int ret;
494 struct lttcomm_consumer_msg msg;
495
a0377dfe
FD
496 LTTNG_ASSERT(metadata);
497 LTTNG_ASSERT(socket);
07b86b52 498
d40f0359 499 DBG("Sending kernel consumer destroy channel key %" PRIu64, metadata->key);
07b86b52 500
53efb85a 501 memset(&msg, 0, sizeof(msg));
07b86b52 502 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
d40f0359 503 msg.u.destroy_channel.key = metadata->key;
07b86b52
JD
504
505 pthread_mutex_lock(socket->lock);
506 health_code_update();
507
508 ret = consumer_send_msg(socket, &msg);
509 if (ret < 0) {
510 goto error;
511 }
512
513error:
514 health_code_update();
515 pthread_mutex_unlock(socket->lock);
516 return ret;
517}
This page took 0.104036 seconds and 4 git commands to generate.