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