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