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