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