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