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