Fix: consumer should await for initial streams
[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
27 #include <common/common.h>
28 #include <common/defaults.h>
29 #include <common/uri.h>
30
31 #include "consumer.h"
32
33 /*
34 * Send destroy relayd command to consumer.
35 *
36 * On success return positive value. On error, negative value.
37 */
38 int consumer_send_destroy_relayd(struct consumer_socket *sock,
39 struct consumer_output *consumer)
40 {
41 int ret;
42 struct lttcomm_consumer_msg msg;
43
44 assert(consumer);
45 assert(sock);
46
47 DBG2("Sending destroy relayd command to consumer...");
48
49 /* Bail out if consumer is disabled */
50 if (!consumer->enabled) {
51 ret = LTTNG_OK;
52 DBG3("Consumer is disabled");
53 goto error;
54 }
55
56 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
57 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
58
59 pthread_mutex_lock(sock->lock);
60 ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg));
61 pthread_mutex_unlock(sock->lock);
62 if (ret < 0) {
63 PERROR("send consumer destroy relayd command");
64 goto error;
65 }
66
67 DBG2("Consumer send destroy relayd command done");
68
69 error:
70 return ret;
71 }
72
73 /*
74 * For each consumer socket in the consumer output object, send a destroy
75 * relayd command.
76 */
77 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
78 {
79 int ret;
80 struct lttng_ht_iter iter;
81 struct consumer_socket *socket;
82
83 assert(consumer);
84
85 /* Destroy any relayd connection */
86 if (consumer && consumer->type == CONSUMER_DST_NET) {
87 rcu_read_lock();
88 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
89 node.node) {
90 /* Send destroy relayd command */
91 ret = consumer_send_destroy_relayd(socket, consumer);
92 if (ret < 0) {
93 ERR("Unable to send destroy relayd command to consumer");
94 /* Continue since we MUST delete everything at this point. */
95 }
96 }
97 rcu_read_unlock();
98 }
99 }
100
101 /*
102 * From a consumer_data structure, allocate and add a consumer socket to the
103 * consumer output.
104 *
105 * Return 0 on success, else negative value on error
106 */
107 int consumer_create_socket(struct consumer_data *data,
108 struct consumer_output *output)
109 {
110 int ret = 0;
111 struct consumer_socket *socket;
112
113 assert(data);
114
115 if (output == NULL || data->cmd_sock < 0) {
116 /*
117 * Not an error. Possible there is simply not spawned consumer or it's
118 * disabled for the tracing session asking the socket.
119 */
120 goto error;
121 }
122
123 rcu_read_lock();
124 socket = consumer_find_socket(data->cmd_sock, output);
125 rcu_read_unlock();
126 if (socket == NULL) {
127 socket = consumer_allocate_socket(data->cmd_sock);
128 if (socket == NULL) {
129 ret = -1;
130 goto error;
131 }
132
133 socket->registered = 0;
134 socket->lock = &data->lock;
135 rcu_read_lock();
136 consumer_add_socket(socket, output);
137 rcu_read_unlock();
138 }
139
140 DBG3("Consumer socket created (fd: %d) and added to output",
141 data->cmd_sock);
142
143 error:
144 return ret;
145 }
146
147 /*
148 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
149 * be acquired before calling this function and across use of the
150 * returned consumer_socket.
151 */
152 struct consumer_socket *consumer_find_socket(int key,
153 struct consumer_output *consumer)
154 {
155 struct lttng_ht_iter iter;
156 struct lttng_ht_node_ulong *node;
157 struct consumer_socket *socket = NULL;
158
159 /* Negative keys are lookup failures */
160 if (key < 0 || consumer == NULL) {
161 return NULL;
162 }
163
164 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
165 &iter);
166 node = lttng_ht_iter_get_node_ulong(&iter);
167 if (node != NULL) {
168 socket = caa_container_of(node, struct consumer_socket, node);
169 }
170
171 return socket;
172 }
173
174 /*
175 * Allocate a new consumer_socket and return the pointer.
176 */
177 struct consumer_socket *consumer_allocate_socket(int fd)
178 {
179 struct consumer_socket *socket = NULL;
180
181 socket = zmalloc(sizeof(struct consumer_socket));
182 if (socket == NULL) {
183 PERROR("zmalloc consumer socket");
184 goto error;
185 }
186
187 socket->fd = fd;
188 lttng_ht_node_init_ulong(&socket->node, fd);
189
190 error:
191 return socket;
192 }
193
194 /*
195 * Add consumer socket to consumer output object. Read side lock must be
196 * acquired before calling this function.
197 */
198 void consumer_add_socket(struct consumer_socket *sock,
199 struct consumer_output *consumer)
200 {
201 assert(sock);
202 assert(consumer);
203
204 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
205 }
206
207 /*
208 * Delte consumer socket to consumer output object. Read side lock must be
209 * acquired before calling this function.
210 */
211 void consumer_del_socket(struct consumer_socket *sock,
212 struct consumer_output *consumer)
213 {
214 int ret;
215 struct lttng_ht_iter iter;
216
217 assert(sock);
218 assert(consumer);
219
220 iter.iter.node = &sock->node.node;
221 ret = lttng_ht_del(consumer->socks, &iter);
222 assert(!ret);
223 }
224
225 /*
226 * RCU destroy call function.
227 */
228 static void destroy_socket_rcu(struct rcu_head *head)
229 {
230 struct lttng_ht_node_ulong *node =
231 caa_container_of(head, struct lttng_ht_node_ulong, head);
232 struct consumer_socket *socket =
233 caa_container_of(node, struct consumer_socket, node);
234
235 free(socket);
236 }
237
238 /*
239 * Destroy and free socket pointer in a call RCU. Read side lock must be
240 * acquired before calling this function.
241 */
242 void consumer_destroy_socket(struct consumer_socket *sock)
243 {
244 assert(sock);
245
246 /*
247 * We DO NOT close the file descriptor here since it is global to the
248 * session daemon and is closed only if the consumer dies or a custom
249 * consumer was registered,
250 */
251 if (sock->registered) {
252 DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
253 lttcomm_close_unix_sock(sock->fd);
254 }
255
256 call_rcu(&sock->node.head, destroy_socket_rcu);
257 }
258
259 /*
260 * Allocate and assign data to a consumer_output object.
261 *
262 * Return pointer to structure.
263 */
264 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
265 {
266 struct consumer_output *output = NULL;
267
268 output = zmalloc(sizeof(struct consumer_output));
269 if (output == NULL) {
270 PERROR("zmalloc consumer_output");
271 goto error;
272 }
273
274 /* By default, consumer output is enabled */
275 output->enabled = 1;
276 output->type = type;
277 output->net_seq_index = -1;
278
279 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
280
281 error:
282 return output;
283 }
284
285 /*
286 * Delete the consumer_output object from the list and free the ptr.
287 */
288 void consumer_destroy_output(struct consumer_output *obj)
289 {
290 if (obj == NULL) {
291 return;
292 }
293
294 if (obj->socks) {
295 struct lttng_ht_iter iter;
296 struct consumer_socket *socket;
297
298 rcu_read_lock();
299 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
300 consumer_del_socket(socket, obj);
301 consumer_destroy_socket(socket);
302 }
303 rcu_read_unlock();
304
305 /* Finally destroy HT */
306 lttng_ht_destroy(obj->socks);
307 }
308
309 free(obj);
310 }
311
312 /*
313 * Copy consumer output and returned the newly allocated copy.
314 */
315 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
316 {
317 struct lttng_ht_iter iter;
318 struct consumer_socket *socket, *copy_sock;
319 struct consumer_output *output;
320
321 assert(obj);
322
323 output = consumer_create_output(obj->type);
324 if (output == NULL) {
325 goto error;
326 }
327
328 memcpy(output, obj, sizeof(struct consumer_output));
329
330 /* Copy sockets */
331 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
332
333 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
334 /* Create new socket object. */
335 copy_sock = consumer_allocate_socket(socket->fd);
336 if (copy_sock == NULL) {
337 goto malloc_error;
338 }
339
340 copy_sock->lock = socket->lock;
341 consumer_add_socket(copy_sock, output);
342 }
343
344 error:
345 return output;
346
347 malloc_error:
348 consumer_destroy_output(output);
349 return NULL;
350 }
351
352 /*
353 * Set network URI to the consumer output object.
354 *
355 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
356 * error.
357 */
358 int consumer_set_network_uri(struct consumer_output *obj,
359 struct lttng_uri *uri)
360 {
361 int ret;
362 char tmp_path[PATH_MAX];
363 char hostname[HOST_NAME_MAX];
364 struct lttng_uri *dst_uri = NULL;
365
366 /* Code flow error safety net. */
367 assert(obj);
368 assert(uri);
369
370 switch (uri->stype) {
371 case LTTNG_STREAM_CONTROL:
372 dst_uri = &obj->dst.net.control;
373 obj->dst.net.control_isset = 1;
374 if (uri->port == 0) {
375 /* Assign default port. */
376 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
377 }
378 DBG3("Consumer control URI set with port %d", uri->port);
379 break;
380 case LTTNG_STREAM_DATA:
381 dst_uri = &obj->dst.net.data;
382 obj->dst.net.data_isset = 1;
383 if (uri->port == 0) {
384 /* Assign default port. */
385 uri->port = DEFAULT_NETWORK_DATA_PORT;
386 }
387 DBG3("Consumer data URI set with port %d", uri->port);
388 break;
389 default:
390 ERR("Set network uri type unknown %d", uri->stype);
391 goto error;
392 }
393
394 ret = uri_compare(dst_uri, uri);
395 if (!ret) {
396 /* Same URI, don't touch it and return success. */
397 DBG3("URI network compare are the same");
398 goto equal;
399 }
400
401 /* URIs were not equal, replacing it. */
402 memset(dst_uri, 0, sizeof(struct lttng_uri));
403 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
404 obj->type = CONSUMER_DST_NET;
405
406 /* Handle subdir and add hostname in front. */
407 if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
408 /* Get hostname to append it in the pathname */
409 ret = gethostname(hostname, sizeof(hostname));
410 if (ret < 0) {
411 PERROR("gethostname. Fallback on default localhost");
412 strncpy(hostname, "localhost", sizeof(hostname));
413 }
414 hostname[sizeof(hostname) - 1] = '\0';
415
416 /* Setup consumer subdir if none present in the control URI */
417 if (strlen(dst_uri->subdir) == 0) {
418 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
419 hostname, obj->subdir);
420 } else {
421 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
422 hostname, dst_uri->subdir);
423 }
424 if (ret < 0) {
425 PERROR("snprintf set consumer uri subdir");
426 goto error;
427 }
428
429 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
430 DBG3("Consumer set network uri subdir path %s", tmp_path);
431 }
432
433 return 0;
434 equal:
435 return 1;
436 error:
437 return -1;
438 }
439
440 /*
441 * Send file descriptor to consumer via sock.
442 */
443 int consumer_send_fds(int sock, int *fds, size_t nb_fd)
444 {
445 int ret;
446
447 assert(fds);
448 assert(nb_fd > 0);
449
450 ret = lttcomm_send_fds_unix_sock(sock, fds, nb_fd);
451 if (ret < 0) {
452 PERROR("send consumer fds");
453 goto error;
454 }
455
456 error:
457 return ret;
458 }
459
460 /*
461 * Consumer send channel communication message structure to consumer.
462 */
463 int consumer_send_channel(int sock, struct lttcomm_consumer_msg *msg)
464 {
465 int ret;
466
467 assert(msg);
468 assert(sock >= 0);
469
470 ret = lttcomm_send_unix_sock(sock, msg,
471 sizeof(struct lttcomm_consumer_msg));
472 if (ret < 0) {
473 PERROR("send consumer channel");
474 goto error;
475 }
476
477 error:
478 return ret;
479 }
480
481 /*
482 * Init channel communication message structure.
483 */
484 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
485 enum lttng_consumer_command cmd,
486 int channel_key,
487 uint64_t max_sb_size,
488 uint64_t mmap_len,
489 const char *name,
490 unsigned int nb_init_streams)
491 {
492 assert(msg);
493
494 /* TODO: Args validation */
495
496 /* Zeroed structure */
497 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
498
499 /* Send channel */
500 msg->cmd_type = cmd;
501 msg->u.channel.channel_key = channel_key;
502 msg->u.channel.max_sb_size = max_sb_size;
503 msg->u.channel.mmap_len = mmap_len;
504 msg->u.channel.nb_init_streams = nb_init_streams;
505 }
506
507 /*
508 * Init stream communication message structure.
509 */
510 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
511 enum lttng_consumer_command cmd,
512 int channel_key,
513 int stream_key,
514 uint32_t state,
515 enum lttng_event_output output,
516 uint64_t mmap_len,
517 uid_t uid,
518 gid_t gid,
519 int net_index,
520 unsigned int metadata_flag,
521 const char *name,
522 const char *pathname)
523 {
524 assert(msg);
525
526 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
527
528 /* TODO: Args validation */
529
530 msg->cmd_type = cmd;
531 msg->u.stream.channel_key = channel_key;
532 msg->u.stream.stream_key = stream_key;
533 msg->u.stream.state = state;
534 msg->u.stream.output = output;
535 msg->u.stream.mmap_len = mmap_len;
536 msg->u.stream.uid = uid;
537 msg->u.stream.gid = gid;
538 msg->u.stream.net_index = net_index;
539 msg->u.stream.metadata_flag = metadata_flag;
540 strncpy(msg->u.stream.name, name, sizeof(msg->u.stream.name));
541 msg->u.stream.name[sizeof(msg->u.stream.name) - 1] = '\0';
542 strncpy(msg->u.stream.path_name, pathname,
543 sizeof(msg->u.stream.path_name));
544 msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
545 }
546
547 /*
548 * Send stream communication structure to the consumer.
549 */
550 int consumer_send_stream(int sock, struct consumer_output *dst,
551 struct lttcomm_consumer_msg *msg, int *fds, size_t nb_fd)
552 {
553 int ret;
554
555 assert(msg);
556 assert(dst);
557
558 switch (dst->type) {
559 case CONSUMER_DST_NET:
560 /* Consumer should send the stream on the network. */
561 msg->u.stream.net_index = dst->net_seq_index;
562 break;
563 case CONSUMER_DST_LOCAL:
564 /* Add stream file name to stream path */
565 strncat(msg->u.stream.path_name, "/",
566 sizeof(msg->u.stream.path_name) -
567 strlen(msg->u.stream.path_name) - 1);
568 strncat(msg->u.stream.path_name, msg->u.stream.name,
569 sizeof(msg->u.stream.path_name) -
570 strlen(msg->u.stream.path_name) - 1);
571 msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
572 /* Indicate that the stream is NOT network */
573 msg->u.stream.net_index = -1;
574 break;
575 default:
576 ERR("Consumer unknown output type (%d)", dst->type);
577 ret = -1;
578 goto error;
579 }
580
581 /* Send on socket */
582 ret = lttcomm_send_unix_sock(sock, msg,
583 sizeof(struct lttcomm_consumer_msg));
584 if (ret < 0) {
585 PERROR("send consumer stream");
586 goto error;
587 }
588
589 ret = consumer_send_fds(sock, fds, nb_fd);
590 if (ret < 0) {
591 goto error;
592 }
593
594 error:
595 return ret;
596 }
597
598 /*
599 * Send relayd socket to consumer associated with a session name.
600 *
601 * On success return positive value. On error, negative value.
602 */
603 int consumer_send_relayd_socket(int consumer_sock,
604 struct lttcomm_sock *sock, struct consumer_output *consumer,
605 enum lttng_stream_type type)
606 {
607 int ret;
608 struct lttcomm_consumer_msg msg;
609
610 /* Code flow error. Safety net. */
611 assert(sock);
612 assert(consumer);
613
614 /* Bail out if consumer is disabled */
615 if (!consumer->enabled) {
616 ret = LTTNG_OK;
617 goto error;
618 }
619
620 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
621 /*
622 * Assign network consumer output index using the temporary consumer since
623 * this call should only be made from within a set_consumer_uri() function
624 * call in the session daemon.
625 */
626 msg.u.relayd_sock.net_index = consumer->net_seq_index;
627 msg.u.relayd_sock.type = type;
628 memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock));
629
630 DBG3("Sending relayd sock info to consumer on %d", consumer_sock);
631 ret = lttcomm_send_unix_sock(consumer_sock, &msg, sizeof(msg));
632 if (ret < 0) {
633 PERROR("send consumer relayd socket info");
634 goto error;
635 }
636
637 DBG3("Sending relayd socket file descriptor to consumer");
638 ret = consumer_send_fds(consumer_sock, &sock->fd, 1);
639 if (ret < 0) {
640 goto error;
641 }
642
643 DBG2("Consumer relayd socket sent");
644
645 error:
646 return ret;
647 }
648
649 /*
650 * Set consumer subdirectory using the session name and a generated datetime if
651 * needed. This is appended to the current subdirectory.
652 */
653 int consumer_set_subdir(struct consumer_output *consumer,
654 const char *session_name)
655 {
656 int ret = 0;
657 unsigned int have_default_name = 0;
658 char datetime[16], tmp_path[PATH_MAX];
659 time_t rawtime;
660 struct tm *timeinfo;
661
662 assert(consumer);
663 assert(session_name);
664
665 memset(tmp_path, 0, sizeof(tmp_path));
666
667 /* Flag if we have a default session. */
668 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
669 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
670 have_default_name = 1;
671 } else {
672 /* Get date and time for session path */
673 time(&rawtime);
674 timeinfo = localtime(&rawtime);
675 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
676 }
677
678 if (have_default_name) {
679 ret = snprintf(tmp_path, sizeof(tmp_path),
680 "%s/%s", consumer->subdir, session_name);
681 } else {
682 ret = snprintf(tmp_path, sizeof(tmp_path),
683 "%s/%s-%s/", consumer->subdir, session_name, datetime);
684 }
685 if (ret < 0) {
686 PERROR("snprintf session name date");
687 goto error;
688 }
689
690 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
691 DBG2("Consumer subdir set to %s", consumer->subdir);
692
693 error:
694 return ret;
695 }
This page took 0.06904 seconds and 4 git commands to generate.