Tests: add kernel snapshot streaming to root regression
[lttng-tools.git] / src / bin / lttng-sessiond / consumer.c
CommitLineData
00e2e675
DG
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#include <assert.h>
20#include <stdio.h>
21#include <stdlib.h>
22#include <string.h>
23#include <sys/stat.h>
24#include <sys/types.h>
25#include <unistd.h>
d88aee68 26#include <inttypes.h>
00e2e675
DG
27
28#include <common/common.h>
29#include <common/defaults.h>
30#include <common/uri.h>
ee91bab2 31#include <common/utils.h>
00e2e675
DG
32
33#include "consumer.h"
7972aab2
DG
34#include "health.h"
35#include "ust-app.h"
0b2dc8df 36#include "utils.h"
00e2e675 37
f50f23d9
DG
38/*
39 * Receive a reply command status message from the consumer. Consumer socket
40 * lock MUST be acquired before calling this function.
41 *
42 * Return 0 on success, -1 on recv error or a negative lttng error code which
43 * was possibly returned by the consumer.
44 */
45int consumer_recv_status_reply(struct consumer_socket *sock)
46{
47 int ret;
48 struct lttcomm_consumer_status_msg reply;
49
50 assert(sock);
51
52 ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
a6cd2b97
DG
53 if (ret <= 0) {
54 if (ret == 0) {
55 /* Orderly shutdown. Don't return 0 which means success. */
56 ret = -1;
57 }
3448e266
DG
58 /* The above call will print a PERROR on error. */
59 DBG("Fail to receive status reply on sock %d", sock->fd);
f50f23d9
DG
60 goto end;
61 }
62
63 if (reply.ret_code == LTTNG_OK) {
64 /* All good. */
65 ret = 0;
66 } else {
67 ret = -reply.ret_code;
ffe60014 68 DBG("Consumer ret code %d", ret);
f50f23d9
DG
69 }
70
71end:
72 return ret;
73}
74
ffe60014
DG
75/*
76 * Once the ASK_CHANNEL command is sent to the consumer, the channel
77 * information are sent back. This call receives that data and populates key
78 * and stream_count.
79 *
80 * On success return 0 and both key and stream_count are set. On error, a
81 * negative value is sent back and both parameters are untouched.
82 */
83int consumer_recv_status_channel(struct consumer_socket *sock,
d88aee68 84 uint64_t *key, unsigned int *stream_count)
ffe60014
DG
85{
86 int ret;
87 struct lttcomm_consumer_status_channel reply;
88
89 assert(sock);
90 assert(stream_count);
91 assert(key);
92
93 ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
94 if (ret <= 0) {
95 if (ret == 0) {
96 /* Orderly shutdown. Don't return 0 which means success. */
97 ret = -1;
98 }
99 /* The above call will print a PERROR on error. */
100 DBG("Fail to receive status reply on sock %d", sock->fd);
101 goto end;
102 }
103
104 /* An error is possible so don't touch the key and stream_count. */
105 if (reply.ret_code != LTTNG_OK) {
106 ret = -1;
107 goto end;
108 }
109
110 *key = reply.key;
111 *stream_count = reply.stream_count;
112
113end:
114 return ret;
115}
116
2f77fc4b
DG
117/*
118 * Send destroy relayd command to consumer.
119 *
120 * On success return positive value. On error, negative value.
121 */
122int consumer_send_destroy_relayd(struct consumer_socket *sock,
123 struct consumer_output *consumer)
124{
125 int ret;
126 struct lttcomm_consumer_msg msg;
127
128 assert(consumer);
129 assert(sock);
130
ffe60014 131 DBG2("Sending destroy relayd command to consumer sock %d", sock->fd);
2f77fc4b
DG
132
133 /* Bail out if consumer is disabled */
134 if (!consumer->enabled) {
f73fabfd 135 ret = LTTNG_OK;
2f77fc4b
DG
136 DBG3("Consumer is disabled");
137 goto error;
138 }
139
140 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
141 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
142
143 pthread_mutex_lock(sock->lock);
144 ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg));
2f77fc4b 145 if (ret < 0) {
c5c45efa
DG
146 /* Indicate that the consumer is probably closing at this point. */
147 DBG("send consumer destroy relayd command");
f50f23d9 148 goto error_send;
2f77fc4b
DG
149 }
150
f50f23d9
DG
151 /* Don't check the return value. The caller will do it. */
152 ret = consumer_recv_status_reply(sock);
153
2f77fc4b
DG
154 DBG2("Consumer send destroy relayd command done");
155
f50f23d9
DG
156error_send:
157 pthread_mutex_unlock(sock->lock);
2f77fc4b
DG
158error:
159 return ret;
160}
161
162/*
163 * For each consumer socket in the consumer output object, send a destroy
164 * relayd command.
165 */
166void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
167{
2f77fc4b
DG
168 struct lttng_ht_iter iter;
169 struct consumer_socket *socket;
170
171 assert(consumer);
172
173 /* Destroy any relayd connection */
6dc3064a 174 if (consumer->type == CONSUMER_DST_NET) {
2f77fc4b
DG
175 rcu_read_lock();
176 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
177 node.node) {
c617c0c6
MD
178 int ret;
179
2f77fc4b
DG
180 /* Send destroy relayd command */
181 ret = consumer_send_destroy_relayd(socket, consumer);
182 if (ret < 0) {
c5c45efa 183 DBG("Unable to send destroy relayd command to consumer");
2f77fc4b
DG
184 /* Continue since we MUST delete everything at this point. */
185 }
186 }
187 rcu_read_unlock();
188 }
189}
190
a4b92340
DG
191/*
192 * From a consumer_data structure, allocate and add a consumer socket to the
193 * consumer output.
194 *
195 * Return 0 on success, else negative value on error
196 */
197int consumer_create_socket(struct consumer_data *data,
198 struct consumer_output *output)
199{
200 int ret = 0;
201 struct consumer_socket *socket;
202
203 assert(data);
204
205 if (output == NULL || data->cmd_sock < 0) {
206 /*
207 * Not an error. Possible there is simply not spawned consumer or it's
208 * disabled for the tracing session asking the socket.
209 */
210 goto error;
211 }
212
213 rcu_read_lock();
214 socket = consumer_find_socket(data->cmd_sock, output);
215 rcu_read_unlock();
216 if (socket == NULL) {
217 socket = consumer_allocate_socket(data->cmd_sock);
218 if (socket == NULL) {
219 ret = -1;
220 goto error;
221 }
222
2f77fc4b 223 socket->registered = 0;
a4b92340
DG
224 socket->lock = &data->lock;
225 rcu_read_lock();
226 consumer_add_socket(socket, output);
227 rcu_read_unlock();
228 }
229
6dc3064a
DG
230 socket->type = data->type;
231
a4b92340
DG
232 DBG3("Consumer socket created (fd: %d) and added to output",
233 data->cmd_sock);
234
235error:
236 return ret;
237}
238
7972aab2
DG
239/*
240 * Return the consumer socket from the given consumer output with the right
241 * bitness. On error, returns NULL.
242 *
243 * The caller MUST acquire a rcu read side lock and keep it until the socket
244 * object reference is not needed anymore.
245 */
246struct consumer_socket *consumer_find_socket_by_bitness(int bits,
247 struct consumer_output *consumer)
248{
249 int consumer_fd;
250 struct consumer_socket *socket = NULL;
251
252 switch (bits) {
253 case 64:
254 consumer_fd = uatomic_read(&ust_consumerd64_fd);
255 break;
256 case 32:
257 consumer_fd = uatomic_read(&ust_consumerd32_fd);
258 break;
259 default:
260 assert(0);
261 goto end;
262 }
263
264 socket = consumer_find_socket(consumer_fd, consumer);
265 if (!socket) {
266 ERR("Consumer socket fd %d not found in consumer obj %p",
267 consumer_fd, consumer);
268 }
269
270end:
271 return socket;
272}
273
173af62f
DG
274/*
275 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
276 * be acquired before calling this function and across use of the
277 * returned consumer_socket.
278 */
279struct consumer_socket *consumer_find_socket(int key,
280 struct consumer_output *consumer)
281{
282 struct lttng_ht_iter iter;
283 struct lttng_ht_node_ulong *node;
284 struct consumer_socket *socket = NULL;
285
286 /* Negative keys are lookup failures */
a4b92340 287 if (key < 0 || consumer == NULL) {
173af62f
DG
288 return NULL;
289 }
290
291 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
292 &iter);
293 node = lttng_ht_iter_get_node_ulong(&iter);
294 if (node != NULL) {
295 socket = caa_container_of(node, struct consumer_socket, node);
296 }
297
298 return socket;
299}
300
301/*
302 * Allocate a new consumer_socket and return the pointer.
303 */
304struct consumer_socket *consumer_allocate_socket(int fd)
305{
306 struct consumer_socket *socket = NULL;
307
308 socket = zmalloc(sizeof(struct consumer_socket));
309 if (socket == NULL) {
310 PERROR("zmalloc consumer socket");
311 goto error;
312 }
313
314 socket->fd = fd;
315 lttng_ht_node_init_ulong(&socket->node, fd);
316
317error:
318 return socket;
319}
320
321/*
322 * Add consumer socket to consumer output object. Read side lock must be
323 * acquired before calling this function.
324 */
325void consumer_add_socket(struct consumer_socket *sock,
326 struct consumer_output *consumer)
327{
328 assert(sock);
329 assert(consumer);
330
331 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
332}
333
334/*
335 * Delte consumer socket to consumer output object. Read side lock must be
336 * acquired before calling this function.
337 */
338void consumer_del_socket(struct consumer_socket *sock,
339 struct consumer_output *consumer)
340{
341 int ret;
342 struct lttng_ht_iter iter;
343
344 assert(sock);
345 assert(consumer);
346
347 iter.iter.node = &sock->node.node;
348 ret = lttng_ht_del(consumer->socks, &iter);
349 assert(!ret);
350}
351
352/*
353 * RCU destroy call function.
354 */
355static void destroy_socket_rcu(struct rcu_head *head)
356{
357 struct lttng_ht_node_ulong *node =
358 caa_container_of(head, struct lttng_ht_node_ulong, head);
359 struct consumer_socket *socket =
360 caa_container_of(node, struct consumer_socket, node);
361
362 free(socket);
363}
364
365/*
366 * Destroy and free socket pointer in a call RCU. Read side lock must be
367 * acquired before calling this function.
368 */
369void consumer_destroy_socket(struct consumer_socket *sock)
370{
371 assert(sock);
372
373 /*
374 * We DO NOT close the file descriptor here since it is global to the
2f77fc4b
DG
375 * session daemon and is closed only if the consumer dies or a custom
376 * consumer was registered,
173af62f 377 */
2f77fc4b
DG
378 if (sock->registered) {
379 DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
380 lttcomm_close_unix_sock(sock->fd);
381 }
173af62f
DG
382
383 call_rcu(&sock->node.head, destroy_socket_rcu);
384}
385
00e2e675
DG
386/*
387 * Allocate and assign data to a consumer_output object.
388 *
389 * Return pointer to structure.
390 */
391struct consumer_output *consumer_create_output(enum consumer_dst_type type)
392{
393 struct consumer_output *output = NULL;
394
395 output = zmalloc(sizeof(struct consumer_output));
396 if (output == NULL) {
397 PERROR("zmalloc consumer_output");
398 goto error;
399 }
400
401 /* By default, consumer output is enabled */
402 output->enabled = 1;
403 output->type = type;
d88aee68 404 output->net_seq_index = (uint64_t) -1ULL;
173af62f
DG
405
406 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
00e2e675
DG
407
408error:
409 return output;
410}
411
412/*
413 * Delete the consumer_output object from the list and free the ptr.
36b588ed
MD
414 *
415 * Should *NOT* be called with RCU read-side lock held.
00e2e675
DG
416 */
417void consumer_destroy_output(struct consumer_output *obj)
418{
419 if (obj == NULL) {
420 return;
421 }
422
173af62f
DG
423 if (obj->socks) {
424 struct lttng_ht_iter iter;
425 struct consumer_socket *socket;
426
2f77fc4b 427 rcu_read_lock();
173af62f 428 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
2f77fc4b 429 consumer_del_socket(socket, obj);
173af62f
DG
430 consumer_destroy_socket(socket);
431 }
2f77fc4b
DG
432 rcu_read_unlock();
433
434 /* Finally destroy HT */
0b2dc8df 435 ht_cleanup_push(obj->socks);
00e2e675 436 }
173af62f 437
00e2e675
DG
438 free(obj);
439}
440
441/*
442 * Copy consumer output and returned the newly allocated copy.
36b588ed
MD
443 *
444 * Should *NOT* be called with RCU read-side lock held.
00e2e675
DG
445 */
446struct consumer_output *consumer_copy_output(struct consumer_output *obj)
447{
6dc3064a 448 int ret;
09a90bcd 449 struct lttng_ht *tmp_ht_ptr;
00e2e675
DG
450 struct consumer_output *output;
451
452 assert(obj);
453
454 output = consumer_create_output(obj->type);
455 if (output == NULL) {
456 goto error;
457 }
09a90bcd
DG
458 /* Avoid losing the HT reference after the memcpy() */
459 tmp_ht_ptr = output->socks;
00e2e675
DG
460
461 memcpy(output, obj, sizeof(struct consumer_output));
462
09a90bcd
DG
463 /* Putting back the HT pointer and start copying socket(s). */
464 output->socks = tmp_ht_ptr;
173af62f 465
6dc3064a
DG
466 ret = consumer_copy_sockets(output, obj);
467 if (ret < 0) {
468 goto malloc_error;
469 }
470
471error:
472 return output;
473
474malloc_error:
475 consumer_destroy_output(output);
476 return NULL;
477}
478
479/*
480 * Copy consumer sockets from src to dst.
481 *
482 * Return 0 on success or else a negative value.
483 */
484int consumer_copy_sockets(struct consumer_output *dst,
485 struct consumer_output *src)
486{
487 int ret = 0;
488 struct lttng_ht_iter iter;
489 struct consumer_socket *socket, *copy_sock;
490
491 assert(dst);
492 assert(src);
493
b82c5c4d 494 rcu_read_lock();
6dc3064a
DG
495 cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
496 /* Ignore socket that are already there. */
497 copy_sock = consumer_find_socket(socket->fd, dst);
498 if (copy_sock) {
499 continue;
500 }
501
173af62f
DG
502 /* Create new socket object. */
503 copy_sock = consumer_allocate_socket(socket->fd);
504 if (copy_sock == NULL) {
b82c5c4d 505 rcu_read_unlock();
6dc3064a
DG
506 ret = -ENOMEM;
507 goto error;
173af62f
DG
508 }
509
09a90bcd 510 copy_sock->registered = socket->registered;
6dc3064a
DG
511 /*
512 * This is valid because this lock is shared accross all consumer
513 * object being the global lock of the consumer data structure of the
514 * session daemon.
515 */
173af62f 516 copy_sock->lock = socket->lock;
6dc3064a 517 consumer_add_socket(copy_sock, dst);
173af62f 518 }
b82c5c4d 519 rcu_read_unlock();
173af62f 520
00e2e675 521error:
6dc3064a 522 return ret;
00e2e675
DG
523}
524
525/*
526 * Set network URI to the consumer output object.
527 *
ad20f474
DG
528 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
529 * error.
00e2e675
DG
530 */
531int consumer_set_network_uri(struct consumer_output *obj,
532 struct lttng_uri *uri)
533{
534 int ret;
535 char tmp_path[PATH_MAX];
536 char hostname[HOST_NAME_MAX];
537 struct lttng_uri *dst_uri = NULL;
538
539 /* Code flow error safety net. */
540 assert(obj);
541 assert(uri);
542
543 switch (uri->stype) {
544 case LTTNG_STREAM_CONTROL:
545 dst_uri = &obj->dst.net.control;
546 obj->dst.net.control_isset = 1;
547 if (uri->port == 0) {
548 /* Assign default port. */
549 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
a74934ba
DG
550 } else {
551 if (obj->dst.net.data_isset && uri->port ==
552 obj->dst.net.data.port) {
553 ret = -LTTNG_ERR_INVALID;
554 goto error;
555 }
00e2e675 556 }
ad20f474 557 DBG3("Consumer control URI set with port %d", uri->port);
00e2e675
DG
558 break;
559 case LTTNG_STREAM_DATA:
560 dst_uri = &obj->dst.net.data;
561 obj->dst.net.data_isset = 1;
562 if (uri->port == 0) {
563 /* Assign default port. */
564 uri->port = DEFAULT_NETWORK_DATA_PORT;
a74934ba
DG
565 } else {
566 if (obj->dst.net.control_isset && uri->port ==
567 obj->dst.net.control.port) {
568 ret = -LTTNG_ERR_INVALID;
569 goto error;
570 }
00e2e675 571 }
ad20f474 572 DBG3("Consumer data URI set with port %d", uri->port);
00e2e675
DG
573 break;
574 default:
575 ERR("Set network uri type unknown %d", uri->stype);
a74934ba 576 ret = -LTTNG_ERR_INVALID;
00e2e675
DG
577 goto error;
578 }
579
580 ret = uri_compare(dst_uri, uri);
581 if (!ret) {
582 /* Same URI, don't touch it and return success. */
583 DBG3("URI network compare are the same");
ad20f474 584 goto equal;
00e2e675
DG
585 }
586
587 /* URIs were not equal, replacing it. */
588 memset(dst_uri, 0, sizeof(struct lttng_uri));
589 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
590 obj->type = CONSUMER_DST_NET;
591
592 /* Handle subdir and add hostname in front. */
593 if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
594 /* Get hostname to append it in the pathname */
595 ret = gethostname(hostname, sizeof(hostname));
596 if (ret < 0) {
597 PERROR("gethostname. Fallback on default localhost");
598 strncpy(hostname, "localhost", sizeof(hostname));
599 }
600 hostname[sizeof(hostname) - 1] = '\0';
601
602 /* Setup consumer subdir if none present in the control URI */
603 if (strlen(dst_uri->subdir) == 0) {
604 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
605 hostname, obj->subdir);
606 } else {
607 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
608 hostname, dst_uri->subdir);
609 }
610 if (ret < 0) {
611 PERROR("snprintf set consumer uri subdir");
a74934ba 612 ret = -LTTNG_ERR_NOMEM;
00e2e675
DG
613 goto error;
614 }
615
616 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
617 DBG3("Consumer set network uri subdir path %s", tmp_path);
618 }
619
00e2e675 620 return 0;
ad20f474
DG
621equal:
622 return 1;
00e2e675 623error:
a74934ba 624 return ret;
00e2e675
DG
625}
626
627/*
628 * Send file descriptor to consumer via sock.
629 */
f50f23d9 630int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
00e2e675
DG
631{
632 int ret;
633
634 assert(fds);
f50f23d9 635 assert(sock);
00e2e675
DG
636 assert(nb_fd > 0);
637
f50f23d9 638 ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd);
00e2e675 639 if (ret < 0) {
3448e266
DG
640 /* The above call will print a PERROR on error. */
641 DBG("Error when sending consumer fds on sock %d", sock->fd);
00e2e675
DG
642 goto error;
643 }
644
f50f23d9
DG
645 ret = consumer_recv_status_reply(sock);
646
00e2e675
DG
647error:
648 return ret;
649}
650
ffe60014
DG
651/*
652 * Consumer send communication message structure to consumer.
653 */
654int consumer_send_msg(struct consumer_socket *sock,
655 struct lttcomm_consumer_msg *msg)
656{
657 int ret;
658
659 assert(msg);
660 assert(sock);
661 assert(sock->fd >= 0);
662
663 ret = lttcomm_send_unix_sock(sock->fd, msg,
664 sizeof(struct lttcomm_consumer_msg));
665 if (ret < 0) {
666 /* The above call will print a PERROR on error. */
667 DBG("Error when sending consumer channel on sock %d", sock->fd);
668 goto error;
669 }
670
671 ret = consumer_recv_status_reply(sock);
672
673error:
674 return ret;
675}
676
00e2e675
DG
677/*
678 * Consumer send channel communication message structure to consumer.
679 */
f50f23d9
DG
680int consumer_send_channel(struct consumer_socket *sock,
681 struct lttcomm_consumer_msg *msg)
00e2e675
DG
682{
683 int ret;
684
685 assert(msg);
f50f23d9
DG
686 assert(sock);
687 assert(sock->fd >= 0);
00e2e675 688
f50f23d9 689 ret = lttcomm_send_unix_sock(sock->fd, msg,
00e2e675
DG
690 sizeof(struct lttcomm_consumer_msg));
691 if (ret < 0) {
3448e266
DG
692 /* The above call will print a PERROR on error. */
693 DBG("Error when sending consumer channel on sock %d", sock->fd);
00e2e675
DG
694 goto error;
695 }
696
f50f23d9
DG
697 ret = consumer_recv_status_reply(sock);
698
00e2e675
DG
699error:
700 return ret;
701}
702
ffe60014
DG
703/*
704 * Populate the given consumer msg structure with the ask_channel command
705 * information.
706 */
707void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
708 uint64_t subbuf_size,
709 uint64_t num_subbuf,
710 int overwrite,
711 unsigned int switch_timer_interval,
712 unsigned int read_timer_interval,
713 int output,
714 int type,
715 uint64_t session_id,
716 const char *pathname,
717 const char *name,
718 uid_t uid,
719 gid_t gid,
d88aee68
DG
720 uint64_t relayd_id,
721 uint64_t key,
7972aab2 722 unsigned char *uuid,
1624d5b7
JD
723 uint32_t chan_id,
724 uint64_t tracefile_size,
2bba9e53 725 uint64_t tracefile_count,
1950109e 726 uint64_t session_id_per_pid,
2bba9e53 727 unsigned int monitor)
ffe60014
DG
728{
729 assert(msg);
730
731 /* Zeroed structure */
732 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
733
734 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
735 msg->u.ask_channel.subbuf_size = subbuf_size;
736 msg->u.ask_channel.num_subbuf = num_subbuf ;
737 msg->u.ask_channel.overwrite = overwrite;
738 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
739 msg->u.ask_channel.read_timer_interval = read_timer_interval;
740 msg->u.ask_channel.output = output;
741 msg->u.ask_channel.type = type;
742 msg->u.ask_channel.session_id = session_id;
1950109e 743 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
ffe60014
DG
744 msg->u.ask_channel.uid = uid;
745 msg->u.ask_channel.gid = gid;
746 msg->u.ask_channel.relayd_id = relayd_id;
747 msg->u.ask_channel.key = key;
7972aab2 748 msg->u.ask_channel.chan_id = chan_id;
1624d5b7
JD
749 msg->u.ask_channel.tracefile_size = tracefile_size;
750 msg->u.ask_channel.tracefile_count = tracefile_count;
2bba9e53 751 msg->u.ask_channel.monitor = monitor;
ffe60014
DG
752
753 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
754
755 strncpy(msg->u.ask_channel.pathname, pathname,
756 sizeof(msg->u.ask_channel.pathname));
757 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
758
759 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
760 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
761}
762
00e2e675
DG
763/*
764 * Init channel communication message structure.
765 */
766void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
767 enum lttng_consumer_command cmd,
d88aee68 768 uint64_t channel_key,
ffe60014
DG
769 uint64_t session_id,
770 const char *pathname,
771 uid_t uid,
772 gid_t gid,
d88aee68 773 uint64_t relayd_id,
c30aaa51 774 const char *name,
ffe60014
DG
775 unsigned int nb_init_streams,
776 enum lttng_event_output output,
1624d5b7
JD
777 int type,
778 uint64_t tracefile_size,
2bba9e53
DG
779 uint64_t tracefile_count,
780 unsigned int monitor)
00e2e675
DG
781{
782 assert(msg);
783
00e2e675
DG
784 /* Zeroed structure */
785 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
786
787 /* Send channel */
788 msg->cmd_type = cmd;
789 msg->u.channel.channel_key = channel_key;
ffe60014
DG
790 msg->u.channel.session_id = session_id;
791 msg->u.channel.uid = uid;
792 msg->u.channel.gid = gid;
793 msg->u.channel.relayd_id = relayd_id;
c30aaa51 794 msg->u.channel.nb_init_streams = nb_init_streams;
ffe60014
DG
795 msg->u.channel.output = output;
796 msg->u.channel.type = type;
1624d5b7
JD
797 msg->u.channel.tracefile_size = tracefile_size;
798 msg->u.channel.tracefile_count = tracefile_count;
2bba9e53 799 msg->u.channel.monitor = monitor;
ffe60014
DG
800
801 strncpy(msg->u.channel.pathname, pathname,
802 sizeof(msg->u.channel.pathname));
803 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
804
805 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
806 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
00e2e675
DG
807}
808
809/*
810 * Init stream communication message structure.
811 */
812void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
813 enum lttng_consumer_command cmd,
d88aee68
DG
814 uint64_t channel_key,
815 uint64_t stream_key,
ffe60014 816 int cpu)
00e2e675
DG
817{
818 assert(msg);
819
820 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
821
00e2e675
DG
822 msg->cmd_type = cmd;
823 msg->u.stream.channel_key = channel_key;
824 msg->u.stream.stream_key = stream_key;
ffe60014 825 msg->u.stream.cpu = cpu;
00e2e675
DG
826}
827
828/*
829 * Send stream communication structure to the consumer.
830 */
f50f23d9
DG
831int consumer_send_stream(struct consumer_socket *sock,
832 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
833 int *fds, size_t nb_fd)
00e2e675
DG
834{
835 int ret;
836
837 assert(msg);
838 assert(dst);
f50f23d9 839 assert(sock);
ffe60014 840 assert(fds);
00e2e675
DG
841
842 /* Send on socket */
f50f23d9 843 ret = lttcomm_send_unix_sock(sock->fd, msg,
00e2e675
DG
844 sizeof(struct lttcomm_consumer_msg));
845 if (ret < 0) {
3448e266
DG
846 /* The above call will print a PERROR on error. */
847 DBG("Error when sending consumer stream on sock %d", sock->fd);
00e2e675
DG
848 goto error;
849 }
850
f50f23d9
DG
851 ret = consumer_recv_status_reply(sock);
852 if (ret < 0) {
853 goto error;
854 }
855
00e2e675
DG
856 ret = consumer_send_fds(sock, fds, nb_fd);
857 if (ret < 0) {
858 goto error;
859 }
860
861error:
862 return ret;
863}
37278a1e
DG
864
865/*
866 * Send relayd socket to consumer associated with a session name.
867 *
868 * On success return positive value. On error, negative value.
869 */
f50f23d9 870int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
6151a90f 871 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
d88aee68 872 enum lttng_stream_type type, uint64_t session_id)
37278a1e
DG
873{
874 int ret;
875 struct lttcomm_consumer_msg msg;
876
877 /* Code flow error. Safety net. */
6151a90f 878 assert(rsock);
37278a1e 879 assert(consumer);
f50f23d9 880 assert(consumer_sock);
37278a1e
DG
881
882 /* Bail out if consumer is disabled */
883 if (!consumer->enabled) {
f73fabfd 884 ret = LTTNG_OK;
37278a1e
DG
885 goto error;
886 }
887
888 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
889 /*
890 * Assign network consumer output index using the temporary consumer since
891 * this call should only be made from within a set_consumer_uri() function
892 * call in the session daemon.
893 */
894 msg.u.relayd_sock.net_index = consumer->net_seq_index;
895 msg.u.relayd_sock.type = type;
46e6455f 896 msg.u.relayd_sock.session_id = session_id;
6151a90f 897 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
37278a1e 898
f50f23d9
DG
899 DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
900 ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
37278a1e 901 if (ret < 0) {
3448e266 902 /* The above call will print a PERROR on error. */
6151a90f 903 DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
37278a1e
DG
904 goto error;
905 }
906
f50f23d9
DG
907 ret = consumer_recv_status_reply(consumer_sock);
908 if (ret < 0) {
909 goto error;
910 }
911
37278a1e 912 DBG3("Sending relayd socket file descriptor to consumer");
6151a90f 913 ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
37278a1e
DG
914 if (ret < 0) {
915 goto error;
916 }
917
918 DBG2("Consumer relayd socket sent");
919
920error:
921 return ret;
922}
173af62f
DG
923
924/*
2f77fc4b
DG
925 * Set consumer subdirectory using the session name and a generated datetime if
926 * needed. This is appended to the current subdirectory.
173af62f 927 */
2f77fc4b
DG
928int consumer_set_subdir(struct consumer_output *consumer,
929 const char *session_name)
173af62f 930{
2f77fc4b
DG
931 int ret = 0;
932 unsigned int have_default_name = 0;
933 char datetime[16], tmp_path[PATH_MAX];
934 time_t rawtime;
935 struct tm *timeinfo;
173af62f
DG
936
937 assert(consumer);
2f77fc4b
DG
938 assert(session_name);
939
940 memset(tmp_path, 0, sizeof(tmp_path));
941
942 /* Flag if we have a default session. */
943 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
944 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
945 have_default_name = 1;
946 } else {
947 /* Get date and time for session path */
948 time(&rawtime);
949 timeinfo = localtime(&rawtime);
950 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
173af62f
DG
951 }
952
2f77fc4b
DG
953 if (have_default_name) {
954 ret = snprintf(tmp_path, sizeof(tmp_path),
955 "%s/%s", consumer->subdir, session_name);
956 } else {
957 ret = snprintf(tmp_path, sizeof(tmp_path),
958 "%s/%s-%s/", consumer->subdir, session_name, datetime);
959 }
173af62f 960 if (ret < 0) {
2f77fc4b 961 PERROR("snprintf session name date");
173af62f
DG
962 goto error;
963 }
964
2f77fc4b
DG
965 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
966 DBG2("Consumer subdir set to %s", consumer->subdir);
173af62f
DG
967
968error:
969 return ret;
970}
806e2684
DG
971
972/*
6d805429 973 * Ask the consumer if the data is ready to read (NOT pending) for the specific
806e2684
DG
974 * session id.
975 *
976 * This function has a different behavior with the consumer i.e. that it waits
6d805429 977 * for a reply from the consumer if yes or no the data is pending.
806e2684 978 */
d88aee68 979int consumer_is_data_pending(uint64_t session_id,
806e2684
DG
980 struct consumer_output *consumer)
981{
982 int ret;
6d805429 983 int32_t ret_code = 0; /* Default is that the data is NOT pending */
806e2684
DG
984 struct consumer_socket *socket;
985 struct lttng_ht_iter iter;
986 struct lttcomm_consumer_msg msg;
987
988 assert(consumer);
989
6d805429 990 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
806e2684 991
d88aee68 992 msg.u.data_pending.session_id = session_id;
806e2684 993
d88aee68 994 DBG3("Consumer data pending for id %" PRIu64, session_id);
806e2684 995
c8f59ee5 996 /* Send command for each consumer */
b82c5c4d 997 rcu_read_lock();
806e2684
DG
998 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
999 node.node) {
1000 /* Code flow error */
1001 assert(socket->fd >= 0);
1002
1003 pthread_mutex_lock(socket->lock);
1004
1005 ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
1006 if (ret < 0) {
3448e266
DG
1007 /* The above call will print a PERROR on error. */
1008 DBG("Error on consumer is data pending on sock %d", socket->fd);
806e2684 1009 pthread_mutex_unlock(socket->lock);
b82c5c4d 1010 goto error_unlock;
806e2684
DG
1011 }
1012
f50f23d9
DG
1013 /*
1014 * No need for a recv reply status because the answer to the command is
1015 * the reply status message.
1016 */
1017
806e2684 1018 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
a6cd2b97
DG
1019 if (ret <= 0) {
1020 if (ret == 0) {
1021 /* Orderly shutdown. Don't return 0 which means success. */
1022 ret = -1;
1023 }
3448e266
DG
1024 /* The above call will print a PERROR on error. */
1025 DBG("Error on recv consumer is data pending on sock %d", socket->fd);
806e2684 1026 pthread_mutex_unlock(socket->lock);
b82c5c4d 1027 goto error_unlock;
806e2684
DG
1028 }
1029
1030 pthread_mutex_unlock(socket->lock);
1031
6d805429 1032 if (ret_code == 1) {
806e2684
DG
1033 break;
1034 }
1035 }
b82c5c4d 1036 rcu_read_unlock();
806e2684 1037
d88aee68
DG
1038 DBG("Consumer data is %s pending for session id %" PRIu64,
1039 ret_code == 1 ? "" : "NOT", session_id);
806e2684
DG
1040 return ret_code;
1041
b82c5c4d
DG
1042error_unlock:
1043 rcu_read_unlock();
806e2684
DG
1044 return -1;
1045}
7972aab2
DG
1046
1047/*
1048 * Send a flush command to consumer using the given channel key.
1049 *
1050 * Return 0 on success else a negative value.
1051 */
1052int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1053{
1054 int ret;
1055 struct lttcomm_consumer_msg msg;
1056
1057 assert(socket);
1058 assert(socket->fd >= 0);
1059
1060 DBG2("Consumer flush channel key %" PRIu64, key);
1061
1062 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1063 msg.u.flush_channel.key = key;
1064
1065 pthread_mutex_lock(socket->lock);
1066 health_code_update();
1067
1068 ret = consumer_send_msg(socket, &msg);
1069 if (ret < 0) {
1070 goto end;
1071 }
1072
1073end:
1074 health_code_update();
1075 pthread_mutex_unlock(socket->lock);
1076 return ret;
1077}
1078
1079/*
1080 * Send a close metdata command to consumer using the given channel key.
1081 *
1082 * Return 0 on success else a negative value.
1083 */
1084int consumer_close_metadata(struct consumer_socket *socket,
1085 uint64_t metadata_key)
1086{
1087 int ret;
1088 struct lttcomm_consumer_msg msg;
1089
1090 assert(socket);
1091 assert(socket->fd >= 0);
1092
1093 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1094
1095 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1096 msg.u.close_metadata.key = metadata_key;
1097
1098 pthread_mutex_lock(socket->lock);
1099 health_code_update();
1100
1101 ret = consumer_send_msg(socket, &msg);
1102 if (ret < 0) {
1103 goto end;
1104 }
1105
1106end:
1107 health_code_update();
1108 pthread_mutex_unlock(socket->lock);
1109 return ret;
1110}
1111
1112/*
1113 * Send a setup metdata command to consumer using the given channel key.
1114 *
1115 * Return 0 on success else a negative value.
1116 */
1117int consumer_setup_metadata(struct consumer_socket *socket,
1118 uint64_t metadata_key)
1119{
1120 int ret;
1121 struct lttcomm_consumer_msg msg;
1122
1123 assert(socket);
1124 assert(socket->fd >= 0);
1125
1126 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1127
1128 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1129 msg.u.setup_metadata.key = metadata_key;
1130
1131 pthread_mutex_lock(socket->lock);
1132 health_code_update();
1133
1134 ret = consumer_send_msg(socket, &msg);
1135 if (ret < 0) {
1136 goto end;
1137 }
1138
1139end:
1140 health_code_update();
1141 pthread_mutex_unlock(socket->lock);
1142 return ret;
1143}
1144
1145/*
331744e3 1146 * Send metadata string to consumer. Socket lock MUST be acquired.
7972aab2
DG
1147 *
1148 * Return 0 on success else a negative value.
1149 */
1150int consumer_push_metadata(struct consumer_socket *socket,
1151 uint64_t metadata_key, char *metadata_str, size_t len,
1152 size_t target_offset)
1153{
1154 int ret;
1155 struct lttcomm_consumer_msg msg;
1156
1157 assert(socket);
1158 assert(socket->fd >= 0);
1159
1160 DBG2("Consumer push metadata to consumer socket %d", socket->fd);
1161
1162 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1163 msg.u.push_metadata.key = metadata_key;
1164 msg.u.push_metadata.target_offset = target_offset;
1165 msg.u.push_metadata.len = len;
1166
7972aab2
DG
1167 health_code_update();
1168 ret = consumer_send_msg(socket, &msg);
331744e3 1169 if (ret < 0 || len == 0) {
7972aab2
DG
1170 goto end;
1171 }
1172
8fd623e0 1173 DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len);
7972aab2
DG
1174
1175 ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
1176 if (ret < 0) {
1177 goto end;
1178 }
1179
1180 health_code_update();
1181 ret = consumer_recv_status_reply(socket);
1182 if (ret < 0) {
1183 goto end;
1184 }
1185
1186end:
1187 health_code_update();
7972aab2
DG
1188 return ret;
1189}
6dc3064a
DG
1190
1191/*
1192 * Ask the consumer to snapshot a specific channel using the key.
1193 *
1194 * Return 0 on success or else a negative error.
1195 */
1196int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1197 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
ee91bab2 1198 const char *session_path, int wait)
6dc3064a
DG
1199{
1200 int ret;
ee91bab2 1201 char datetime[16];
6dc3064a
DG
1202 struct lttcomm_consumer_msg msg;
1203
1204 assert(socket);
1205 assert(socket->fd >= 0);
1206 assert(output);
1207 assert(output->consumer);
1208
1209 DBG("Consumer snapshot channel key %" PRIu64, key);
1210
ee91bab2
DG
1211 ret = utils_get_current_time_str("%Y%m%d-%H%M%S", datetime,
1212 sizeof(datetime));
1213 if (!ret) {
1214 ret = -EINVAL;
1215 goto error;
1216 }
6dc3064a 1217
ee91bab2 1218 memset(&msg, 0, sizeof(msg));
6dc3064a
DG
1219 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1220 msg.u.snapshot_channel.key = key;
1221 msg.u.snapshot_channel.max_size = output->max_size;
1222 msg.u.snapshot_channel.metadata = metadata;
1223
1224 if (output->consumer->type == CONSUMER_DST_NET) {
1225 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1226 msg.u.snapshot_channel.use_relayd = 1;
1227 ret = snprintf(msg.u.snapshot_channel.pathname,
ee91bab2
DG
1228 sizeof(msg.u.snapshot_channel.pathname), "%s/%s-%s%s",
1229 output->consumer->subdir, output->name, datetime,
1230 session_path);
6dc3064a
DG
1231 if (ret < 0) {
1232 ret = -LTTNG_ERR_NOMEM;
1233 goto error;
1234 }
1235 } else {
1236 ret = snprintf(msg.u.snapshot_channel.pathname,
ee91bab2
DG
1237 sizeof(msg.u.snapshot_channel.pathname), "%s/%s-%s%s",
1238 output->consumer->dst.trace_path, output->name, datetime,
1239 session_path);
6dc3064a
DG
1240 if (ret < 0) {
1241 ret = -LTTNG_ERR_NOMEM;
1242 goto error;
1243 }
07b86b52 1244 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
6dc3064a
DG
1245
1246 /* Create directory. Ignore if exist. */
1247 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1248 S_IRWXU | S_IRWXG, uid, gid);
1249 if (ret < 0) {
1250 if (ret != -EEXIST) {
1251 ERR("Trace directory creation error");
1252 goto error;
1253 }
1254 }
1255 }
1256
1257 health_code_update();
1258 ret = consumer_send_msg(socket, &msg);
1259 if (ret < 0) {
1260 goto error;
1261 }
1262
1263error:
1264 health_code_update();
1265 return ret;
1266}
This page took 0.084524 seconds and 4 git commands to generate.