relayd: track the quit pipe with the fd-tracker
[lttng-tools.git] / src / bin / lttng-relayd / main.c
1 /*
2 * Copyright (C) 2012 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
4 * 2013 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 * 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
6 *
7 * This program is free software; you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License, version 2 only,
9 * as published by the Free Software Foundation.
10 *
11 * This program is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
14 * more details.
15 *
16 * You should have received a copy of the GNU General Public License along
17 * with this program; if not, write to the Free Software Foundation, Inc.,
18 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
19 */
20
21 #define _LGPL_SOURCE
22 #include <getopt.h>
23 #include <grp.h>
24 #include <limits.h>
25 #include <pthread.h>
26 #include <signal.h>
27 #include <stdio.h>
28 #include <stdlib.h>
29 #include <string.h>
30 #include <sys/mman.h>
31 #include <sys/mount.h>
32 #include <sys/resource.h>
33 #include <sys/socket.h>
34 #include <sys/stat.h>
35 #include <sys/types.h>
36 #include <sys/wait.h>
37 #include <sys/resource.h>
38 #include <inttypes.h>
39 #include <urcu/futex.h>
40 #include <urcu/uatomic.h>
41 #include <urcu/rculist.h>
42 #include <unistd.h>
43 #include <fcntl.h>
44 #include <strings.h>
45 #include <ctype.h>
46
47 #include <lttng/lttng.h>
48 #include <common/common.h>
49 #include <common/compat/poll.h>
50 #include <common/compat/socket.h>
51 #include <common/compat/endian.h>
52 #include <common/compat/getenv.h>
53 #include <common/defaults.h>
54 #include <common/daemonize.h>
55 #include <common/futex.h>
56 #include <common/sessiond-comm/sessiond-comm.h>
57 #include <common/sessiond-comm/inet.h>
58 #include <common/sessiond-comm/relayd.h>
59 #include <common/uri.h>
60 #include <common/utils.h>
61 #include <common/align.h>
62 #include <common/config/session-config.h>
63 #include <common/dynamic-buffer.h>
64 #include <common/buffer-view.h>
65 #include <common/string-utils/format.h>
66 #include <common/fd-tracker/fd-tracker.h>
67 #include <common/fd-tracker/utils.h>
68
69 #include "backward-compatibility-group-by.h"
70 #include "cmd.h"
71 #include "connection.h"
72 #include "ctf-trace.h"
73 #include "health-relayd.h"
74 #include "index.h"
75 #include "live.h"
76 #include "lttng-relayd.h"
77 #include "session.h"
78 #include "sessiond-trace-chunks.h"
79 #include "stream.h"
80 #include "tcp_keep_alive.h"
81 #include "testpoint.h"
82 #include "tracefile-array.h"
83 #include "utils.h"
84 #include "version.h"
85 #include "viewer-stream.h"
86
87 static const char *help_msg =
88 #ifdef LTTNG_EMBED_HELP
89 #include <lttng-relayd.8.h>
90 #else
91 NULL
92 #endif
93 ;
94
95 enum relay_connection_status {
96 RELAY_CONNECTION_STATUS_OK,
97 /* An error occurred while processing an event on the connection. */
98 RELAY_CONNECTION_STATUS_ERROR,
99 /* Connection closed/shutdown cleanly. */
100 RELAY_CONNECTION_STATUS_CLOSED,
101 };
102
103 /* command line options */
104 char *opt_output_path, *opt_working_directory;
105 static int opt_daemon, opt_background, opt_print_version, opt_allow_clear = 1;
106 enum relay_group_output_by opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_UNKNOWN;
107
108 /*
109 * We need to wait for listener and live listener threads, as well as
110 * health check thread, before being ready to signal readiness.
111 */
112 #define NR_LTTNG_RELAY_READY 3
113 static int lttng_relay_ready = NR_LTTNG_RELAY_READY;
114
115 /* Size of receive buffer. */
116 #define RECV_DATA_BUFFER_SIZE 65536
117
118 static int recv_child_signal; /* Set to 1 when a SIGUSR1 signal is received. */
119 static pid_t child_ppid; /* Internal parent PID use with daemonize. */
120
121 static struct lttng_uri *control_uri;
122 static struct lttng_uri *data_uri;
123 static struct lttng_uri *live_uri;
124
125 const char *progname;
126
127 const char *tracing_group_name = DEFAULT_TRACING_GROUP;
128 static int tracing_group_name_override;
129
130 const char * const config_section_name = "relayd";
131
132 /*
133 * Quit pipe for all threads. This permits a single cancellation point
134 * for all threads when receiving an event on the pipe.
135 */
136 int thread_quit_pipe[2] = { -1, -1 };
137
138 /*
139 * This pipe is used to inform the worker thread that a command is queued and
140 * ready to be processed.
141 */
142 static int relay_conn_pipe[2] = { -1, -1 };
143
144 /* Shared between threads */
145 static int dispatch_thread_exit;
146
147 static pthread_t listener_thread;
148 static pthread_t dispatcher_thread;
149 static pthread_t worker_thread;
150 static pthread_t health_thread;
151
152 /*
153 * last_relay_stream_id_lock protects last_relay_stream_id increment
154 * atomicity on 32-bit architectures.
155 */
156 static pthread_mutex_t last_relay_stream_id_lock = PTHREAD_MUTEX_INITIALIZER;
157 static uint64_t last_relay_stream_id;
158
159 /*
160 * Relay command queue.
161 *
162 * The relay_thread_listener and relay_thread_dispatcher communicate with this
163 * queue.
164 */
165 static struct relay_conn_queue relay_conn_queue;
166
167 /* Cap of file desriptors to be in simultaneous use by the relay daemon. */
168 static unsigned int lttng_opt_fd_cap;
169
170 /* Global relay stream hash table. */
171 struct lttng_ht *relay_streams_ht;
172
173 /* Global relay viewer stream hash table. */
174 struct lttng_ht *viewer_streams_ht;
175
176 /* Global relay sessions hash table. */
177 struct lttng_ht *sessions_ht;
178
179 /* Relayd health monitoring */
180 struct health_app *health_relayd;
181
182 struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry;
183
184 /* Global fd tracker. */
185 struct fd_tracker *the_fd_tracker;
186
187 static struct option long_options[] = {
188 { "control-port", 1, 0, 'C', },
189 { "data-port", 1, 0, 'D', },
190 { "live-port", 1, 0, 'L', },
191 { "daemonize", 0, 0, 'd', },
192 { "background", 0, 0, 'b', },
193 { "group", 1, 0, 'g', },
194 { "fd-cap", 1, 0, '\0', },
195 { "help", 0, 0, 'h', },
196 { "output", 1, 0, 'o', },
197 { "verbose", 0, 0, 'v', },
198 { "config", 1, 0, 'f' },
199 { "version", 0, 0, 'V' },
200 { "working-directory", 1, 0, 'w', },
201 { "group-output-by-session", 0, 0, 's', },
202 { "group-output-by-host", 0, 0, 'p', },
203 { "disallow-clear", 0, 0, 'x' },
204 { NULL, 0, 0, 0, },
205 };
206
207 static const char *config_ignore_options[] = { "help", "config", "version" };
208
209 static void print_version(void) {
210 fprintf(stdout, "%s\n", VERSION);
211 }
212
213 static void relayd_config_log(void)
214 {
215 DBG("LTTng-relayd " VERSION " - " VERSION_NAME "%s%s",
216 GIT_VERSION[0] == '\0' ? "" : " - " GIT_VERSION,
217 EXTRA_VERSION_NAME[0] == '\0' ? "" : " - " EXTRA_VERSION_NAME);
218 if (EXTRA_VERSION_DESCRIPTION[0] != '\0') {
219 DBG("LTTng-relayd extra version description:\n\t" EXTRA_VERSION_DESCRIPTION "\n");
220 }
221 if (EXTRA_VERSION_PATCHES[0] != '\0') {
222 DBG("LTTng-relayd extra patches:\n\t" EXTRA_VERSION_PATCHES "\n");
223 }
224 }
225
226 /*
227 * Take an option from the getopt output and set it in the right variable to be
228 * used later.
229 *
230 * Return 0 on success else a negative value.
231 */
232 static int set_option(int opt, const char *arg, const char *optname)
233 {
234 int ret;
235
236 switch (opt) {
237 case 0:
238 if (!strcmp(optname, "fd-cap")) {
239 unsigned long v;
240
241 errno = 0;
242 v = strtoul(arg, NULL, 0);
243 if (errno != 0 || !isdigit(arg[0])) {
244 ERR("Wrong value in --fd-cap parameter: %s",
245 arg);
246 ret = -1;
247 goto end;
248 }
249 if (v < DEFAULT_RELAYD_MINIMAL_FD_CAP) {
250 ERR("File descriptor cap must be set to at least %d",
251 DEFAULT_RELAYD_MINIMAL_FD_CAP);
252 }
253 if (v >= UINT_MAX) {
254 ERR("File descriptor cap overflow in --fd-cap parameter: %s",
255 arg);
256 ret = -1;
257 goto end;
258 }
259 lttng_opt_fd_cap = (unsigned int) v;
260 DBG3("File descriptor cap set to %u", lttng_opt_fd_cap);
261 } else {
262 fprintf(stderr, "unknown option %s", optname);
263 if (arg) {
264 fprintf(stderr, " with arg %s\n", arg);
265 }
266 }
267 break;
268 case 'C':
269 if (lttng_is_setuid_setgid()) {
270 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
271 "-C, --control-port");
272 } else {
273 ret = uri_parse(arg, &control_uri);
274 if (ret < 0) {
275 ERR("Invalid control URI specified");
276 goto end;
277 }
278 if (control_uri->port == 0) {
279 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
280 }
281 }
282 break;
283 case 'D':
284 if (lttng_is_setuid_setgid()) {
285 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
286 "-D, -data-port");
287 } else {
288 ret = uri_parse(arg, &data_uri);
289 if (ret < 0) {
290 ERR("Invalid data URI specified");
291 goto end;
292 }
293 if (data_uri->port == 0) {
294 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
295 }
296 }
297 break;
298 case 'L':
299 if (lttng_is_setuid_setgid()) {
300 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
301 "-L, -live-port");
302 } else {
303 ret = uri_parse(arg, &live_uri);
304 if (ret < 0) {
305 ERR("Invalid live URI specified");
306 goto end;
307 }
308 if (live_uri->port == 0) {
309 live_uri->port = DEFAULT_NETWORK_VIEWER_PORT;
310 }
311 }
312 break;
313 case 'd':
314 opt_daemon = 1;
315 break;
316 case 'b':
317 opt_background = 1;
318 break;
319 case 'g':
320 if (lttng_is_setuid_setgid()) {
321 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
322 "-g, --group");
323 } else {
324 tracing_group_name = strdup(arg);
325 if (tracing_group_name == NULL) {
326 ret = -errno;
327 PERROR("strdup");
328 goto end;
329 }
330 tracing_group_name_override = 1;
331 }
332 break;
333 case 'h':
334 ret = utils_show_help(8, "lttng-relayd", help_msg);
335 if (ret) {
336 ERR("Cannot show --help for `lttng-relayd`");
337 perror("exec");
338 }
339 exit(EXIT_FAILURE);
340 case 'V':
341 opt_print_version = 1;
342 break;
343 case 'o':
344 if (lttng_is_setuid_setgid()) {
345 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
346 "-o, --output");
347 } else {
348 ret = asprintf(&opt_output_path, "%s", arg);
349 if (ret < 0) {
350 ret = -errno;
351 PERROR("asprintf opt_output_path");
352 goto end;
353 }
354 }
355 break;
356 case 'w':
357 if (lttng_is_setuid_setgid()) {
358 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
359 "-w, --working-directory");
360 } else {
361 ret = asprintf(&opt_working_directory, "%s", arg);
362 if (ret < 0) {
363 ret = -errno;
364 PERROR("asprintf opt_working_directory");
365 goto end;
366 }
367 }
368 break;
369
370 case 'v':
371 /* Verbose level can increase using multiple -v */
372 if (arg) {
373 lttng_opt_verbose = config_parse_value(arg);
374 } else {
375 /* Only 3 level of verbosity (-vvv). */
376 if (lttng_opt_verbose < 3) {
377 lttng_opt_verbose += 1;
378 }
379 }
380 break;
381 case 's':
382 if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
383 ERR("Cannot set --group-output-by-session, another --group-output-by argument is present");
384 exit(EXIT_FAILURE);
385 }
386 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_SESSION;
387 break;
388 case 'p':
389 if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
390 ERR("Cannot set --group-output-by-host, another --group-output-by argument is present");
391 exit(EXIT_FAILURE);
392 }
393 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
394 break;
395 case 'x':
396 /* Disallow clear */
397 opt_allow_clear = 0;
398 break;
399 default:
400 /* Unknown option or other error.
401 * Error is printed by getopt, just return */
402 ret = -1;
403 goto end;
404 }
405
406 /* All good. */
407 ret = 0;
408
409 end:
410 return ret;
411 }
412
413 /*
414 * config_entry_handler_cb used to handle options read from a config file.
415 * See config_entry_handler_cb comment in common/config/session-config.h for the
416 * return value conventions.
417 */
418 static int config_entry_handler(const struct config_entry *entry, void *unused)
419 {
420 int ret = 0, i;
421
422 if (!entry || !entry->name || !entry->value) {
423 ret = -EINVAL;
424 goto end;
425 }
426
427 /* Check if the option is to be ignored */
428 for (i = 0; i < sizeof(config_ignore_options) / sizeof(char *); i++) {
429 if (!strcmp(entry->name, config_ignore_options[i])) {
430 goto end;
431 }
432 }
433
434 for (i = 0; i < (sizeof(long_options) / sizeof(struct option)) - 1; i++) {
435 /* Ignore if entry name is not fully matched. */
436 if (strcmp(entry->name, long_options[i].name)) {
437 continue;
438 }
439
440 /*
441 * If the option takes no argument on the command line,
442 * we have to check if the value is "true". We support
443 * non-zero numeric values, true, on and yes.
444 */
445 if (!long_options[i].has_arg) {
446 ret = config_parse_value(entry->value);
447 if (ret <= 0) {
448 if (ret) {
449 WARN("Invalid configuration value \"%s\" for option %s",
450 entry->value, entry->name);
451 }
452 /* False, skip boolean config option. */
453 goto end;
454 }
455 }
456
457 ret = set_option(long_options[i].val, entry->value, entry->name);
458 goto end;
459 }
460
461 WARN("Unrecognized option \"%s\" in daemon configuration file.",
462 entry->name);
463
464 end:
465 return ret;
466 }
467
468 static int parse_env_options(void)
469 {
470 int ret = 0;
471 char *value = NULL;
472
473 value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_WORKING_DIRECTORY_ENV);
474 if (value) {
475 opt_working_directory = strdup(value);
476 if (!opt_working_directory) {
477 ERR("Failed to allocate working directory string (\"%s\")",
478 value);
479 ret = -1;
480 }
481 }
482 return ret;
483 }
484
485 static int set_options(int argc, char **argv)
486 {
487 int c, ret = 0, option_index = 0, retval = 0;
488 int orig_optopt = optopt, orig_optind = optind;
489 char *default_address, *optstring;
490 const char *config_path = NULL;
491
492 optstring = utils_generate_optstring(long_options,
493 sizeof(long_options) / sizeof(struct option));
494 if (!optstring) {
495 retval = -ENOMEM;
496 goto exit;
497 }
498
499 /* Check for the --config option */
500
501 while ((c = getopt_long(argc, argv, optstring, long_options,
502 &option_index)) != -1) {
503 if (c == '?') {
504 retval = -EINVAL;
505 goto exit;
506 } else if (c != 'f') {
507 continue;
508 }
509
510 if (lttng_is_setuid_setgid()) {
511 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
512 "-f, --config");
513 } else {
514 config_path = utils_expand_path(optarg);
515 if (!config_path) {
516 ERR("Failed to resolve path: %s", optarg);
517 }
518 }
519 }
520
521 ret = config_get_section_entries(config_path, config_section_name,
522 config_entry_handler, NULL);
523 if (ret) {
524 if (ret > 0) {
525 ERR("Invalid configuration option at line %i", ret);
526 }
527 retval = -1;
528 goto exit;
529 }
530
531 /* Reset getopt's global state */
532 optopt = orig_optopt;
533 optind = orig_optind;
534 while (1) {
535 c = getopt_long(argc, argv, optstring, long_options, &option_index);
536 if (c == -1) {
537 break;
538 }
539
540 ret = set_option(c, optarg, long_options[option_index].name);
541 if (ret < 0) {
542 retval = -1;
543 goto exit;
544 }
545 }
546
547 /* assign default values */
548 if (control_uri == NULL) {
549 ret = asprintf(&default_address,
550 "tcp://" DEFAULT_NETWORK_CONTROL_BIND_ADDRESS ":%d",
551 DEFAULT_NETWORK_CONTROL_PORT);
552 if (ret < 0) {
553 PERROR("asprintf default data address");
554 retval = -1;
555 goto exit;
556 }
557
558 ret = uri_parse(default_address, &control_uri);
559 free(default_address);
560 if (ret < 0) {
561 ERR("Invalid control URI specified");
562 retval = -1;
563 goto exit;
564 }
565 }
566 if (data_uri == NULL) {
567 ret = asprintf(&default_address,
568 "tcp://" DEFAULT_NETWORK_DATA_BIND_ADDRESS ":%d",
569 DEFAULT_NETWORK_DATA_PORT);
570 if (ret < 0) {
571 PERROR("asprintf default data address");
572 retval = -1;
573 goto exit;
574 }
575
576 ret = uri_parse(default_address, &data_uri);
577 free(default_address);
578 if (ret < 0) {
579 ERR("Invalid data URI specified");
580 retval = -1;
581 goto exit;
582 }
583 }
584 if (live_uri == NULL) {
585 ret = asprintf(&default_address,
586 "tcp://" DEFAULT_NETWORK_VIEWER_BIND_ADDRESS ":%d",
587 DEFAULT_NETWORK_VIEWER_PORT);
588 if (ret < 0) {
589 PERROR("asprintf default viewer control address");
590 retval = -1;
591 goto exit;
592 }
593
594 ret = uri_parse(default_address, &live_uri);
595 free(default_address);
596 if (ret < 0) {
597 ERR("Invalid viewer control URI specified");
598 retval = -1;
599 goto exit;
600 }
601 }
602 if (lttng_opt_fd_cap == 0) {
603 int ret;
604 struct rlimit rlimit;
605
606 ret = getrlimit(RLIMIT_NOFILE, &rlimit);
607 if (ret) {
608 PERROR("Failed to get file descriptor limit");
609 retval = -1;
610 }
611
612 lttng_opt_fd_cap = rlimit.rlim_cur;
613 }
614
615 if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
616 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
617 }
618 if (opt_allow_clear) {
619 /* Check if env variable exists. */
620 const char *value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_DISALLOW_CLEAR_ENV);
621 if (value) {
622 ret = config_parse_value(value);
623 if (ret < 0) {
624 ERR("Invalid value for %s specified", DEFAULT_LTTNG_RELAYD_DISALLOW_CLEAR_ENV);
625 retval = -1;
626 goto exit;
627 }
628 opt_allow_clear = !ret;
629 }
630 }
631
632 exit:
633 free(optstring);
634 return retval;
635 }
636
637 static void print_global_objects(void)
638 {
639 print_viewer_streams();
640 print_relay_streams();
641 print_sessions();
642 }
643
644 /*
645 * Cleanup the daemon
646 */
647 static void relayd_cleanup(void)
648 {
649 print_global_objects();
650
651 DBG("Cleaning up");
652
653 if (viewer_streams_ht)
654 lttng_ht_destroy(viewer_streams_ht);
655 if (relay_streams_ht)
656 lttng_ht_destroy(relay_streams_ht);
657 if (sessions_ht)
658 lttng_ht_destroy(sessions_ht);
659
660 free(opt_output_path);
661 free(opt_working_directory);
662
663 if (health_relayd) {
664 health_app_destroy(health_relayd);
665 }
666 /* Close thread quit pipes */
667 utils_close_pipe(health_quit_pipe);
668 if (thread_quit_pipe[0] != -1) {
669 (void) fd_tracker_util_pipe_close(
670 the_fd_tracker, thread_quit_pipe);
671 }
672 if (sessiond_trace_chunk_registry) {
673 sessiond_trace_chunk_registry_destroy(
674 sessiond_trace_chunk_registry);
675 }
676 if (the_fd_tracker) {
677 fd_tracker_destroy(the_fd_tracker);
678 }
679
680 uri_free(control_uri);
681 uri_free(data_uri);
682 /* Live URI is freed in the live thread. */
683
684 if (tracing_group_name_override) {
685 free((void *) tracing_group_name);
686 }
687 fd_tracker_log(the_fd_tracker);
688 }
689
690 /*
691 * Write to writable pipe used to notify a thread.
692 */
693 static int notify_thread_pipe(int wpipe)
694 {
695 ssize_t ret;
696
697 ret = lttng_write(wpipe, "!", 1);
698 if (ret < 1) {
699 PERROR("write poll pipe");
700 goto end;
701 }
702 ret = 0;
703 end:
704 return ret;
705 }
706
707 static int notify_health_quit_pipe(int *pipe)
708 {
709 ssize_t ret;
710
711 ret = lttng_write(pipe[1], "4", 1);
712 if (ret < 1) {
713 PERROR("write relay health quit");
714 goto end;
715 }
716 ret = 0;
717 end:
718 return ret;
719 }
720
721 /*
722 * Stop all relayd and relayd-live threads.
723 */
724 int lttng_relay_stop_threads(void)
725 {
726 int retval = 0;
727
728 /* Stopping all threads */
729 DBG("Terminating all threads");
730 if (notify_thread_pipe(thread_quit_pipe[1])) {
731 ERR("write error on thread quit pipe");
732 retval = -1;
733 }
734
735 if (notify_health_quit_pipe(health_quit_pipe)) {
736 ERR("write error on health quit pipe");
737 }
738
739 /* Dispatch thread */
740 CMM_STORE_SHARED(dispatch_thread_exit, 1);
741 futex_nto1_wake(&relay_conn_queue.futex);
742
743 if (relayd_live_stop()) {
744 ERR("Error stopping live threads");
745 retval = -1;
746 }
747 return retval;
748 }
749
750 /*
751 * Signal handler for the daemon
752 *
753 * Simply stop all worker threads, leaving main() return gracefully after
754 * joining all threads and calling cleanup().
755 */
756 static void sighandler(int sig)
757 {
758 switch (sig) {
759 case SIGINT:
760 DBG("SIGINT caught");
761 if (lttng_relay_stop_threads()) {
762 ERR("Error stopping threads");
763 }
764 break;
765 case SIGTERM:
766 DBG("SIGTERM caught");
767 if (lttng_relay_stop_threads()) {
768 ERR("Error stopping threads");
769 }
770 break;
771 case SIGUSR1:
772 CMM_STORE_SHARED(recv_child_signal, 1);
773 break;
774 default:
775 break;
776 }
777 }
778
779 /*
780 * Setup signal handler for :
781 * SIGINT, SIGTERM, SIGPIPE
782 */
783 static int set_signal_handler(void)
784 {
785 int ret = 0;
786 struct sigaction sa;
787 sigset_t sigset;
788
789 if ((ret = sigemptyset(&sigset)) < 0) {
790 PERROR("sigemptyset");
791 return ret;
792 }
793
794 sa.sa_mask = sigset;
795 sa.sa_flags = 0;
796
797 sa.sa_handler = sighandler;
798 if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
799 PERROR("sigaction");
800 return ret;
801 }
802
803 if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
804 PERROR("sigaction");
805 return ret;
806 }
807
808 if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
809 PERROR("sigaction");
810 return ret;
811 }
812
813 sa.sa_handler = SIG_IGN;
814 if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
815 PERROR("sigaction");
816 return ret;
817 }
818
819 DBG("Signal handler set for SIGTERM, SIGUSR1, SIGPIPE and SIGINT");
820
821 return ret;
822 }
823
824 void lttng_relay_notify_ready(void)
825 {
826 /* Notify the parent of the fork() process that we are ready. */
827 if (opt_daemon || opt_background) {
828 if (uatomic_sub_return(&lttng_relay_ready, 1) == 0) {
829 kill(child_ppid, SIGUSR1);
830 }
831 }
832 }
833
834 /*
835 * Init thread quit pipe.
836 *
837 * Return -1 on error or 0 if all pipes are created.
838 */
839 static int init_thread_quit_pipe(void)
840 {
841 return fd_tracker_util_pipe_open_cloexec(
842 the_fd_tracker, "Quit pipe", thread_quit_pipe);
843 }
844
845 /*
846 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
847 */
848 static int create_thread_poll_set(struct lttng_poll_event *events, int size)
849 {
850 int ret;
851
852 if (events == NULL || size == 0) {
853 ret = -1;
854 goto error;
855 }
856
857 ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
858 if (ret < 0) {
859 goto error;
860 }
861
862 /* Add quit pipe */
863 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
864 if (ret < 0) {
865 goto error;
866 }
867
868 return 0;
869
870 error:
871 return ret;
872 }
873
874 /*
875 * Check if the thread quit pipe was triggered.
876 *
877 * Return 1 if it was triggered else 0;
878 */
879 static int check_thread_quit_pipe(int fd, uint32_t events)
880 {
881 if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
882 return 1;
883 }
884
885 return 0;
886 }
887
888 /*
889 * Create and init socket from uri.
890 */
891 static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri)
892 {
893 int ret;
894 struct lttcomm_sock *sock = NULL;
895
896 sock = lttcomm_alloc_sock_from_uri(uri);
897 if (sock == NULL) {
898 ERR("Allocating socket");
899 goto error;
900 }
901
902 ret = lttcomm_create_sock(sock);
903 if (ret < 0) {
904 goto error;
905 }
906 DBG("Listening on sock %d", sock->fd);
907
908 ret = sock->ops->bind(sock);
909 if (ret < 0) {
910 PERROR("Failed to bind socket");
911 goto error;
912 }
913
914 ret = sock->ops->listen(sock, -1);
915 if (ret < 0) {
916 goto error;
917
918 }
919
920 return sock;
921
922 error:
923 if (sock) {
924 lttcomm_destroy_sock(sock);
925 }
926 return NULL;
927 }
928
929 /*
930 * This thread manages the listening for new connections on the network
931 */
932 static void *relay_thread_listener(void *data)
933 {
934 int i, ret, pollfd, err = -1;
935 uint32_t revents, nb_fd;
936 struct lttng_poll_event events;
937 struct lttcomm_sock *control_sock, *data_sock;
938
939 DBG("[thread] Relay listener started");
940
941 health_register(health_relayd, HEALTH_RELAYD_TYPE_LISTENER);
942
943 health_code_update();
944
945 control_sock = relay_socket_create(control_uri);
946 if (!control_sock) {
947 goto error_sock_control;
948 }
949
950 data_sock = relay_socket_create(data_uri);
951 if (!data_sock) {
952 goto error_sock_relay;
953 }
954
955 /*
956 * Pass 3 as size here for the thread quit pipe, control and
957 * data socket.
958 */
959 ret = create_thread_poll_set(&events, 3);
960 if (ret < 0) {
961 goto error_create_poll;
962 }
963
964 /* Add the control socket */
965 ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
966 if (ret < 0) {
967 goto error_poll_add;
968 }
969
970 /* Add the data socket */
971 ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
972 if (ret < 0) {
973 goto error_poll_add;
974 }
975
976 lttng_relay_notify_ready();
977
978 if (testpoint(relayd_thread_listener)) {
979 goto error_testpoint;
980 }
981
982 while (1) {
983 health_code_update();
984
985 DBG("Listener accepting connections");
986
987 restart:
988 health_poll_entry();
989 ret = lttng_poll_wait(&events, -1);
990 health_poll_exit();
991 if (ret < 0) {
992 /*
993 * Restart interrupted system call.
994 */
995 if (errno == EINTR) {
996 goto restart;
997 }
998 goto error;
999 }
1000
1001 nb_fd = ret;
1002
1003 DBG("Relay new connection received");
1004 for (i = 0; i < nb_fd; i++) {
1005 health_code_update();
1006
1007 /* Fetch once the poll data */
1008 revents = LTTNG_POLL_GETEV(&events, i);
1009 pollfd = LTTNG_POLL_GETFD(&events, i);
1010
1011 /* Thread quit pipe has been closed. Killing thread. */
1012 ret = check_thread_quit_pipe(pollfd, revents);
1013 if (ret) {
1014 err = 0;
1015 goto exit;
1016 }
1017
1018 if (revents & LPOLLIN) {
1019 /*
1020 * A new connection is requested, therefore a
1021 * sessiond/consumerd connection is allocated in
1022 * this thread, enqueued to a global queue and
1023 * dequeued (and freed) in the worker thread.
1024 */
1025 int val = 1;
1026 struct relay_connection *new_conn;
1027 struct lttcomm_sock *newsock;
1028 enum connection_type type;
1029
1030 if (pollfd == data_sock->fd) {
1031 type = RELAY_DATA;
1032 newsock = data_sock->ops->accept(data_sock);
1033 DBG("Relay data connection accepted, socket %d",
1034 newsock->fd);
1035 } else {
1036 assert(pollfd == control_sock->fd);
1037 type = RELAY_CONTROL;
1038 newsock = control_sock->ops->accept(control_sock);
1039 DBG("Relay control connection accepted, socket %d",
1040 newsock->fd);
1041 }
1042 if (!newsock) {
1043 PERROR("accepting sock");
1044 goto error;
1045 }
1046
1047 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR, &val,
1048 sizeof(val));
1049 if (ret < 0) {
1050 PERROR("setsockopt inet");
1051 lttcomm_destroy_sock(newsock);
1052 goto error;
1053 }
1054
1055 ret = socket_apply_keep_alive_config(newsock->fd);
1056 if (ret < 0) {
1057 ERR("Failed to apply TCP keep-alive configuration on socket (%i)",
1058 newsock->fd);
1059 lttcomm_destroy_sock(newsock);
1060 goto error;
1061 }
1062
1063 new_conn = connection_create(newsock, type);
1064 if (!new_conn) {
1065 lttcomm_destroy_sock(newsock);
1066 goto error;
1067 }
1068
1069 /* Enqueue request for the dispatcher thread. */
1070 cds_wfcq_enqueue(&relay_conn_queue.head, &relay_conn_queue.tail,
1071 &new_conn->qnode);
1072
1073 /*
1074 * Wake the dispatch queue futex.
1075 * Implicit memory barrier with the
1076 * exchange in cds_wfcq_enqueue.
1077 */
1078 futex_nto1_wake(&relay_conn_queue.futex);
1079 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1080 ERR("socket poll error");
1081 goto error;
1082 } else {
1083 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
1084 goto error;
1085 }
1086 }
1087 }
1088
1089 exit:
1090 error:
1091 error_poll_add:
1092 error_testpoint:
1093 lttng_poll_clean(&events);
1094 error_create_poll:
1095 if (data_sock->fd >= 0) {
1096 ret = data_sock->ops->close(data_sock);
1097 if (ret) {
1098 PERROR("close");
1099 }
1100 }
1101 lttcomm_destroy_sock(data_sock);
1102 error_sock_relay:
1103 if (control_sock->fd >= 0) {
1104 ret = control_sock->ops->close(control_sock);
1105 if (ret) {
1106 PERROR("close");
1107 }
1108 }
1109 lttcomm_destroy_sock(control_sock);
1110 error_sock_control:
1111 if (err) {
1112 health_error();
1113 ERR("Health error occurred in %s", __func__);
1114 }
1115 health_unregister(health_relayd);
1116 DBG("Relay listener thread cleanup complete");
1117 lttng_relay_stop_threads();
1118 return NULL;
1119 }
1120
1121 /*
1122 * This thread manages the dispatching of the requests to worker threads
1123 */
1124 static void *relay_thread_dispatcher(void *data)
1125 {
1126 int err = -1;
1127 ssize_t ret;
1128 struct cds_wfcq_node *node;
1129 struct relay_connection *new_conn = NULL;
1130
1131 DBG("[thread] Relay dispatcher started");
1132
1133 health_register(health_relayd, HEALTH_RELAYD_TYPE_DISPATCHER);
1134
1135 if (testpoint(relayd_thread_dispatcher)) {
1136 goto error_testpoint;
1137 }
1138
1139 health_code_update();
1140
1141 for (;;) {
1142 health_code_update();
1143
1144 /* Atomically prepare the queue futex */
1145 futex_nto1_prepare(&relay_conn_queue.futex);
1146
1147 if (CMM_LOAD_SHARED(dispatch_thread_exit)) {
1148 break;
1149 }
1150
1151 do {
1152 health_code_update();
1153
1154 /* Dequeue commands */
1155 node = cds_wfcq_dequeue_blocking(&relay_conn_queue.head,
1156 &relay_conn_queue.tail);
1157 if (node == NULL) {
1158 DBG("Woken up but nothing in the relay command queue");
1159 /* Continue thread execution */
1160 break;
1161 }
1162 new_conn = caa_container_of(node, struct relay_connection, qnode);
1163
1164 DBG("Dispatching request waiting on sock %d", new_conn->sock->fd);
1165
1166 /*
1167 * Inform worker thread of the new request. This
1168 * call is blocking so we can be assured that
1169 * the data will be read at some point in time
1170 * or wait to the end of the world :)
1171 */
1172 ret = lttng_write(relay_conn_pipe[1], &new_conn, sizeof(new_conn));
1173 if (ret < 0) {
1174 PERROR("write connection pipe");
1175 connection_put(new_conn);
1176 goto error;
1177 }
1178 } while (node != NULL);
1179
1180 /* Futex wait on queue. Blocking call on futex() */
1181 health_poll_entry();
1182 futex_nto1_wait(&relay_conn_queue.futex);
1183 health_poll_exit();
1184 }
1185
1186 /* Normal exit, no error */
1187 err = 0;
1188
1189 error:
1190 error_testpoint:
1191 if (err) {
1192 health_error();
1193 ERR("Health error occurred in %s", __func__);
1194 }
1195 health_unregister(health_relayd);
1196 DBG("Dispatch thread dying");
1197 lttng_relay_stop_threads();
1198 return NULL;
1199 }
1200
1201 static bool session_streams_have_index(const struct relay_session *session)
1202 {
1203 return session->minor >= 4 && !session->snapshot;
1204 }
1205
1206 /*
1207 * Handle the RELAYD_CREATE_SESSION command.
1208 *
1209 * On success, send back the session id or else return a negative value.
1210 */
1211 static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr,
1212 struct relay_connection *conn,
1213 const struct lttng_buffer_view *payload)
1214 {
1215 int ret = 0;
1216 ssize_t send_ret;
1217 struct relay_session *session = NULL;
1218 struct lttcomm_relayd_create_session_reply_2_11 reply = {};
1219 char session_name[LTTNG_NAME_MAX] = {};
1220 char hostname[LTTNG_HOST_NAME_MAX] = {};
1221 uint32_t live_timer = 0;
1222 bool snapshot = false;
1223 bool session_name_contains_creation_timestamp = false;
1224 /* Left nil for peers < 2.11. */
1225 char base_path[LTTNG_PATH_MAX] = {};
1226 lttng_uuid sessiond_uuid = {};
1227 LTTNG_OPTIONAL(uint64_t) id_sessiond = {};
1228 LTTNG_OPTIONAL(uint64_t) current_chunk_id = {};
1229 LTTNG_OPTIONAL(time_t) creation_time = {};
1230 struct lttng_dynamic_buffer reply_payload;
1231
1232 lttng_dynamic_buffer_init(&reply_payload);
1233
1234 if (conn->minor < 4) {
1235 /* From 2.1 to 2.3 */
1236 ret = 0;
1237 } else if (conn->minor >= 4 && conn->minor < 11) {
1238 /* From 2.4 to 2.10 */
1239 ret = cmd_create_session_2_4(payload, session_name,
1240 hostname, &live_timer, &snapshot);
1241 } else {
1242 bool has_current_chunk;
1243 uint64_t current_chunk_id_value;
1244 time_t creation_time_value;
1245 uint64_t id_sessiond_value;
1246
1247 /* From 2.11 to ... */
1248 ret = cmd_create_session_2_11(payload, session_name, hostname,
1249 base_path, &live_timer, &snapshot, &id_sessiond_value,
1250 sessiond_uuid, &has_current_chunk,
1251 &current_chunk_id_value, &creation_time_value,
1252 &session_name_contains_creation_timestamp);
1253 if (lttng_uuid_is_nil(sessiond_uuid)) {
1254 /* The nil UUID is reserved for pre-2.11 clients. */
1255 ERR("Illegal nil UUID announced by peer in create session command");
1256 ret = -1;
1257 goto send_reply;
1258 }
1259 LTTNG_OPTIONAL_SET(&id_sessiond, id_sessiond_value);
1260 LTTNG_OPTIONAL_SET(&creation_time, creation_time_value);
1261 if (has_current_chunk) {
1262 LTTNG_OPTIONAL_SET(&current_chunk_id,
1263 current_chunk_id_value);
1264 }
1265 }
1266
1267 if (ret < 0) {
1268 goto send_reply;
1269 }
1270
1271 session = session_create(session_name, hostname, base_path, live_timer,
1272 snapshot, sessiond_uuid,
1273 id_sessiond.is_set ? &id_sessiond.value : NULL,
1274 current_chunk_id.is_set ? &current_chunk_id.value : NULL,
1275 creation_time.is_set ? &creation_time.value : NULL,
1276 conn->major, conn->minor,
1277 session_name_contains_creation_timestamp);
1278 if (!session) {
1279 ret = -1;
1280 goto send_reply;
1281 }
1282 assert(!conn->session);
1283 conn->session = session;
1284 DBG("Created session %" PRIu64, session->id);
1285
1286 reply.generic.session_id = htobe64(session->id);
1287
1288 send_reply:
1289 if (ret < 0) {
1290 reply.generic.ret_code = htobe32(LTTNG_ERR_FATAL);
1291 } else {
1292 reply.generic.ret_code = htobe32(LTTNG_OK);
1293 }
1294
1295 if (conn->minor < 11) {
1296 /* From 2.1 to 2.10 */
1297 ret = lttng_dynamic_buffer_append(&reply_payload,
1298 &reply.generic, sizeof(reply.generic));
1299 if (ret) {
1300 ERR("Failed to append \"create session\" command reply header to payload buffer");
1301 ret = -1;
1302 goto end;
1303 }
1304 } else {
1305 const uint32_t output_path_length =
1306 session ? strlen(session->output_path) + 1 : 0;
1307
1308 reply.output_path_length = htobe32(output_path_length);
1309 ret = lttng_dynamic_buffer_append(
1310 &reply_payload, &reply, sizeof(reply));
1311 if (ret) {
1312 ERR("Failed to append \"create session\" command reply header to payload buffer");
1313 goto end;
1314 }
1315
1316 if (output_path_length) {
1317 ret = lttng_dynamic_buffer_append(&reply_payload,
1318 session->output_path,
1319 output_path_length);
1320 if (ret) {
1321 ERR("Failed to append \"create session\" command reply path to payload buffer");
1322 goto end;
1323 }
1324 }
1325 }
1326
1327 send_ret = conn->sock->ops->sendmsg(conn->sock, reply_payload.data,
1328 reply_payload.size, 0);
1329 if (send_ret < (ssize_t) reply_payload.size) {
1330 ERR("Failed to send \"create session\" command reply of %zu bytes (ret = %zd)",
1331 reply_payload.size, send_ret);
1332 ret = -1;
1333 }
1334 end:
1335 if (ret < 0 && session) {
1336 session_put(session);
1337 }
1338 lttng_dynamic_buffer_reset(&reply_payload);
1339 return ret;
1340 }
1341
1342 /*
1343 * When we have received all the streams and the metadata for a channel,
1344 * we make them visible to the viewer threads.
1345 */
1346 static void publish_connection_local_streams(struct relay_connection *conn)
1347 {
1348 struct relay_stream *stream;
1349 struct relay_session *session = conn->session;
1350
1351 /*
1352 * We publish all streams belonging to a session atomically wrt
1353 * session lock.
1354 */
1355 pthread_mutex_lock(&session->lock);
1356 rcu_read_lock();
1357 cds_list_for_each_entry_rcu(stream, &session->recv_list,
1358 recv_node) {
1359 stream_publish(stream);
1360 }
1361 rcu_read_unlock();
1362
1363 /*
1364 * Inform the viewer that there are new streams in the session.
1365 */
1366 if (session->viewer_attached) {
1367 uatomic_set(&session->new_streams, 1);
1368 }
1369 pthread_mutex_unlock(&session->lock);
1370 }
1371
1372 static int conform_channel_path(char *channel_path)
1373 {
1374 int ret = 0;
1375
1376 if (strstr("../", channel_path)) {
1377 ERR("Refusing channel path as it walks up the path hierarchy: \"%s\"",
1378 channel_path);
1379 ret = -1;
1380 goto end;
1381 }
1382
1383 if (*channel_path == '/') {
1384 const size_t len = strlen(channel_path);
1385
1386 /*
1387 * Channel paths from peers prior to 2.11 are expressed as an
1388 * absolute path that is, in reality, relative to the relay
1389 * daemon's output directory. Remove the leading slash so it
1390 * is correctly interpreted as a relative path later on.
1391 *
1392 * len (and not len - 1) is used to copy the trailing NULL.
1393 */
1394 bcopy(channel_path + 1, channel_path, len);
1395 }
1396 end:
1397 return ret;
1398 }
1399
1400 /*
1401 * relay_add_stream: allocate a new stream for a session
1402 */
1403 static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr,
1404 struct relay_connection *conn,
1405 const struct lttng_buffer_view *payload)
1406 {
1407 int ret;
1408 ssize_t send_ret;
1409 struct relay_session *session = conn->session;
1410 struct relay_stream *stream = NULL;
1411 struct lttcomm_relayd_status_stream reply;
1412 struct ctf_trace *trace = NULL;
1413 uint64_t stream_handle = -1ULL;
1414 char *path_name = NULL, *channel_name = NULL;
1415 uint64_t tracefile_size = 0, tracefile_count = 0;
1416 LTTNG_OPTIONAL(uint64_t) stream_chunk_id = {};
1417
1418 if (!session || !conn->version_check_done) {
1419 ERR("Trying to add a stream before version check");
1420 ret = -1;
1421 goto end_no_session;
1422 }
1423
1424 if (session->minor == 1) {
1425 /* For 2.1 */
1426 ret = cmd_recv_stream_2_1(payload, &path_name,
1427 &channel_name);
1428 } else if (session->minor > 1 && session->minor < 11) {
1429 /* From 2.2 to 2.10 */
1430 ret = cmd_recv_stream_2_2(payload, &path_name,
1431 &channel_name, &tracefile_size, &tracefile_count);
1432 } else {
1433 /* From 2.11 to ... */
1434 ret = cmd_recv_stream_2_11(payload, &path_name,
1435 &channel_name, &tracefile_size, &tracefile_count,
1436 &stream_chunk_id.value);
1437 stream_chunk_id.is_set = true;
1438 }
1439
1440 if (ret < 0) {
1441 goto send_reply;
1442 }
1443
1444 if (conform_channel_path(path_name)) {
1445 goto send_reply;
1446 }
1447
1448 /*
1449 * Backward compatibility for --group-output-by-session.
1450 * Prior to lttng 2.11, the complete path is passed by the stream.
1451 * Starting at 2.11, lttng-relayd uses chunk. When dealing with producer
1452 * >=2.11 the chunk is responsible for the output path. When dealing
1453 * with producer < 2.11 the chunk output_path is the root output path
1454 * and the stream carries the complete path (path_name).
1455 * To support --group-output-by-session with older producer (<2.11), we
1456 * need to craft the path based on the stream path.
1457 */
1458 if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_SESSION) {
1459 if (conn->minor < 4) {
1460 /*
1461 * From 2.1 to 2.3, the session_name is not passed on
1462 * the RELAYD_CREATE_SESSION command. The session name
1463 * is necessary to detect the presence of a base_path
1464 * inside the stream path. Without it we cannot perform
1465 * a valid group-output-by-session transformation.
1466 */
1467 WARN("Unable to perform a --group-by-session transformation for session %" PRIu64
1468 " for stream with path \"%s\" as it is produced by a peer using a protocol older than v2.4",
1469 session->id, path_name);
1470 } else if (conn->minor >= 4 && conn->minor < 11) {
1471 char *group_by_session_path_name;
1472
1473 assert(session->session_name[0] != '\0');
1474
1475 group_by_session_path_name =
1476 backward_compat_group_by_session(
1477 path_name,
1478 session->session_name);
1479 if (!group_by_session_path_name) {
1480 ERR("Failed to apply group by session to stream of session %" PRIu64,
1481 session->id);
1482 goto send_reply;
1483 }
1484
1485 DBG("Transformed session path from \"%s\" to \"%s\" to honor per-session name grouping",
1486 path_name, group_by_session_path_name);
1487
1488 free(path_name);
1489 path_name = group_by_session_path_name;
1490 }
1491 }
1492
1493 trace = ctf_trace_get_by_path_or_create(session, path_name);
1494 if (!trace) {
1495 goto send_reply;
1496 }
1497
1498 /* This stream here has one reference on the trace. */
1499 pthread_mutex_lock(&last_relay_stream_id_lock);
1500 stream_handle = ++last_relay_stream_id;
1501 pthread_mutex_unlock(&last_relay_stream_id_lock);
1502
1503 /* We pass ownership of path_name and channel_name. */
1504 stream = stream_create(trace, stream_handle, path_name,
1505 channel_name, tracefile_size, tracefile_count);
1506 path_name = NULL;
1507 channel_name = NULL;
1508
1509 /*
1510 * Streams are the owners of their trace. Reference to trace is
1511 * kept within stream_create().
1512 */
1513 ctf_trace_put(trace);
1514
1515 send_reply:
1516 memset(&reply, 0, sizeof(reply));
1517 reply.handle = htobe64(stream_handle);
1518 if (!stream) {
1519 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1520 } else {
1521 reply.ret_code = htobe32(LTTNG_OK);
1522 }
1523
1524 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1525 sizeof(struct lttcomm_relayd_status_stream), 0);
1526 if (send_ret < (ssize_t) sizeof(reply)) {
1527 ERR("Failed to send \"add stream\" command reply (ret = %zd)",
1528 send_ret);
1529 ret = -1;
1530 }
1531
1532 end_no_session:
1533 free(path_name);
1534 free(channel_name);
1535 return ret;
1536 }
1537
1538 /*
1539 * relay_close_stream: close a specific stream
1540 */
1541 static int relay_close_stream(const struct lttcomm_relayd_hdr *recv_hdr,
1542 struct relay_connection *conn,
1543 const struct lttng_buffer_view *payload)
1544 {
1545 int ret;
1546 ssize_t send_ret;
1547 struct relay_session *session = conn->session;
1548 struct lttcomm_relayd_close_stream stream_info;
1549 struct lttcomm_relayd_generic_reply reply;
1550 struct relay_stream *stream;
1551
1552 DBG("Close stream received");
1553
1554 if (!session || !conn->version_check_done) {
1555 ERR("Trying to close a stream before version check");
1556 ret = -1;
1557 goto end_no_session;
1558 }
1559
1560 if (payload->size < sizeof(stream_info)) {
1561 ERR("Unexpected payload size in \"relay_close_stream\": expected >= %zu bytes, got %zu bytes",
1562 sizeof(stream_info), payload->size);
1563 ret = -1;
1564 goto end_no_session;
1565 }
1566 memcpy(&stream_info, payload->data, sizeof(stream_info));
1567 stream_info.stream_id = be64toh(stream_info.stream_id);
1568 stream_info.last_net_seq_num = be64toh(stream_info.last_net_seq_num);
1569
1570 stream = stream_get_by_id(stream_info.stream_id);
1571 if (!stream) {
1572 ret = -1;
1573 goto end;
1574 }
1575
1576 /*
1577 * Set last_net_seq_num before the close flag. Required by data
1578 * pending check.
1579 */
1580 pthread_mutex_lock(&stream->lock);
1581 stream->last_net_seq_num = stream_info.last_net_seq_num;
1582 pthread_mutex_unlock(&stream->lock);
1583
1584 /*
1585 * This is one of the conditions which may trigger a stream close
1586 * with the others being:
1587 * 1) A close command is received for a stream
1588 * 2) The control connection owning the stream is closed
1589 * 3) We have received all of the stream's data _after_ a close
1590 * request.
1591 */
1592 try_stream_close(stream);
1593 if (stream->is_metadata) {
1594 struct relay_viewer_stream *vstream;
1595
1596 vstream = viewer_stream_get_by_id(stream->stream_handle);
1597 if (vstream) {
1598 if (stream->no_new_metadata_notified) {
1599 /*
1600 * Since all the metadata has been sent to the
1601 * viewer and that we have a request to close
1602 * its stream, we can safely teardown the
1603 * corresponding metadata viewer stream.
1604 */
1605 viewer_stream_put(vstream);
1606 }
1607 /* Put local reference. */
1608 viewer_stream_put(vstream);
1609 }
1610 }
1611 stream_put(stream);
1612 ret = 0;
1613
1614 end:
1615 memset(&reply, 0, sizeof(reply));
1616 if (ret < 0) {
1617 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1618 } else {
1619 reply.ret_code = htobe32(LTTNG_OK);
1620 }
1621 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1622 sizeof(struct lttcomm_relayd_generic_reply), 0);
1623 if (send_ret < (ssize_t) sizeof(reply)) {
1624 ERR("Failed to send \"close stream\" command reply (ret = %zd)",
1625 send_ret);
1626 ret = -1;
1627 }
1628
1629 end_no_session:
1630 return ret;
1631 }
1632
1633 /*
1634 * relay_reset_metadata: reset a metadata stream
1635 */
1636 static
1637 int relay_reset_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
1638 struct relay_connection *conn,
1639 const struct lttng_buffer_view *payload)
1640 {
1641 int ret;
1642 ssize_t send_ret;
1643 struct relay_session *session = conn->session;
1644 struct lttcomm_relayd_reset_metadata stream_info;
1645 struct lttcomm_relayd_generic_reply reply;
1646 struct relay_stream *stream;
1647
1648 DBG("Reset metadata received");
1649
1650 if (!session || !conn->version_check_done) {
1651 ERR("Trying to reset a metadata stream before version check");
1652 ret = -1;
1653 goto end_no_session;
1654 }
1655
1656 if (payload->size < sizeof(stream_info)) {
1657 ERR("Unexpected payload size in \"relay_reset_metadata\": expected >= %zu bytes, got %zu bytes",
1658 sizeof(stream_info), payload->size);
1659 ret = -1;
1660 goto end_no_session;
1661 }
1662 memcpy(&stream_info, payload->data, sizeof(stream_info));
1663 stream_info.stream_id = be64toh(stream_info.stream_id);
1664 stream_info.version = be64toh(stream_info.version);
1665
1666 DBG("Update metadata to version %" PRIu64, stream_info.version);
1667
1668 /* Unsupported for live sessions for now. */
1669 if (session->live_timer != 0) {
1670 ret = -1;
1671 goto end;
1672 }
1673
1674 stream = stream_get_by_id(stream_info.stream_id);
1675 if (!stream) {
1676 ret = -1;
1677 goto end;
1678 }
1679 pthread_mutex_lock(&stream->lock);
1680 if (!stream->is_metadata) {
1681 ret = -1;
1682 goto end_unlock;
1683 }
1684
1685 ret = stream_reset_file(stream);
1686 if (ret < 0) {
1687 ERR("Failed to reset metadata stream %" PRIu64
1688 ": stream_path = %s, channel = %s",
1689 stream->stream_handle, stream->path_name,
1690 stream->channel_name);
1691 goto end_unlock;
1692 }
1693 end_unlock:
1694 pthread_mutex_unlock(&stream->lock);
1695 stream_put(stream);
1696
1697 end:
1698 memset(&reply, 0, sizeof(reply));
1699 if (ret < 0) {
1700 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1701 } else {
1702 reply.ret_code = htobe32(LTTNG_OK);
1703 }
1704 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1705 sizeof(struct lttcomm_relayd_generic_reply), 0);
1706 if (send_ret < (ssize_t) sizeof(reply)) {
1707 ERR("Failed to send \"reset metadata\" command reply (ret = %zd)",
1708 send_ret);
1709 ret = -1;
1710 }
1711
1712 end_no_session:
1713 return ret;
1714 }
1715
1716 /*
1717 * relay_unknown_command: send -1 if received unknown command
1718 */
1719 static void relay_unknown_command(struct relay_connection *conn)
1720 {
1721 struct lttcomm_relayd_generic_reply reply;
1722 ssize_t send_ret;
1723
1724 memset(&reply, 0, sizeof(reply));
1725 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1726 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1727 if (send_ret < sizeof(reply)) {
1728 ERR("Failed to send \"unknown command\" command reply (ret = %zd)", send_ret);
1729 }
1730 }
1731
1732 /*
1733 * relay_start: send an acknowledgment to the client to tell if we are
1734 * ready to receive data. We are ready if a session is established.
1735 */
1736 static int relay_start(const struct lttcomm_relayd_hdr *recv_hdr,
1737 struct relay_connection *conn,
1738 const struct lttng_buffer_view *payload)
1739 {
1740 int ret = 0;
1741 ssize_t send_ret;
1742 struct lttcomm_relayd_generic_reply reply;
1743 struct relay_session *session = conn->session;
1744
1745 if (!session) {
1746 DBG("Trying to start the streaming without a session established");
1747 ret = htobe32(LTTNG_ERR_UNK);
1748 }
1749
1750 memset(&reply, 0, sizeof(reply));
1751 reply.ret_code = htobe32(LTTNG_OK);
1752 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1753 sizeof(reply), 0);
1754 if (send_ret < (ssize_t) sizeof(reply)) {
1755 ERR("Failed to send \"relay_start\" command reply (ret = %zd)",
1756 send_ret);
1757 ret = -1;
1758 }
1759
1760 return ret;
1761 }
1762
1763 /*
1764 * relay_recv_metadata: receive the metadata for the session.
1765 */
1766 static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
1767 struct relay_connection *conn,
1768 const struct lttng_buffer_view *payload)
1769 {
1770 int ret = 0;
1771 struct relay_session *session = conn->session;
1772 struct lttcomm_relayd_metadata_payload metadata_payload_header;
1773 struct relay_stream *metadata_stream;
1774 uint64_t metadata_payload_size;
1775 struct lttng_buffer_view packet_view;
1776
1777 if (!session) {
1778 ERR("Metadata sent before version check");
1779 ret = -1;
1780 goto end;
1781 }
1782
1783 if (recv_hdr->data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1784 ERR("Incorrect data size");
1785 ret = -1;
1786 goto end;
1787 }
1788 metadata_payload_size = recv_hdr->data_size -
1789 sizeof(struct lttcomm_relayd_metadata_payload);
1790
1791 memcpy(&metadata_payload_header, payload->data,
1792 sizeof(metadata_payload_header));
1793 metadata_payload_header.stream_id = be64toh(
1794 metadata_payload_header.stream_id);
1795 metadata_payload_header.padding_size = be32toh(
1796 metadata_payload_header.padding_size);
1797
1798 metadata_stream = stream_get_by_id(metadata_payload_header.stream_id);
1799 if (!metadata_stream) {
1800 ret = -1;
1801 goto end;
1802 }
1803
1804 packet_view = lttng_buffer_view_from_view(payload,
1805 sizeof(metadata_payload_header), metadata_payload_size);
1806 if (!packet_view.data) {
1807 ERR("Invalid metadata packet length announced by header");
1808 ret = -1;
1809 goto end_put;
1810 }
1811
1812 pthread_mutex_lock(&metadata_stream->lock);
1813 ret = stream_write(metadata_stream, &packet_view,
1814 metadata_payload_header.padding_size);
1815 pthread_mutex_unlock(&metadata_stream->lock);
1816 if (ret){
1817 ret = -1;
1818 goto end_put;
1819 }
1820 end_put:
1821 stream_put(metadata_stream);
1822 end:
1823 return ret;
1824 }
1825
1826 /*
1827 * relay_send_version: send relayd version number
1828 */
1829 static int relay_send_version(const struct lttcomm_relayd_hdr *recv_hdr,
1830 struct relay_connection *conn,
1831 const struct lttng_buffer_view *payload)
1832 {
1833 int ret;
1834 ssize_t send_ret;
1835 struct lttcomm_relayd_version reply, msg;
1836 bool compatible = true;
1837
1838 conn->version_check_done = true;
1839
1840 /* Get version from the other side. */
1841 if (payload->size < sizeof(msg)) {
1842 ERR("Unexpected payload size in \"relay_send_version\": expected >= %zu bytes, got %zu bytes",
1843 sizeof(msg), payload->size);
1844 ret = -1;
1845 goto end;
1846 }
1847
1848 memcpy(&msg, payload->data, sizeof(msg));
1849 msg.major = be32toh(msg.major);
1850 msg.minor = be32toh(msg.minor);
1851
1852 memset(&reply, 0, sizeof(reply));
1853 reply.major = RELAYD_VERSION_COMM_MAJOR;
1854 reply.minor = RELAYD_VERSION_COMM_MINOR;
1855
1856 /* Major versions must be the same */
1857 if (reply.major != msg.major) {
1858 DBG("Incompatible major versions (%u vs %u), deleting session",
1859 reply.major, msg.major);
1860 compatible = false;
1861 }
1862
1863 conn->major = reply.major;
1864 /* We adapt to the lowest compatible version */
1865 if (reply.minor <= msg.minor) {
1866 conn->minor = reply.minor;
1867 } else {
1868 conn->minor = msg.minor;
1869 }
1870
1871 reply.major = htobe32(reply.major);
1872 reply.minor = htobe32(reply.minor);
1873 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1874 sizeof(reply), 0);
1875 if (send_ret < (ssize_t) sizeof(reply)) {
1876 ERR("Failed to send \"send version\" command reply (ret = %zd)",
1877 send_ret);
1878 ret = -1;
1879 goto end;
1880 } else {
1881 ret = 0;
1882 }
1883
1884 if (!compatible) {
1885 ret = -1;
1886 goto end;
1887 }
1888
1889 DBG("Version check done using protocol %u.%u", conn->major,
1890 conn->minor);
1891
1892 end:
1893 return ret;
1894 }
1895
1896 /*
1897 * Check for data pending for a given stream id from the session daemon.
1898 */
1899 static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
1900 struct relay_connection *conn,
1901 const struct lttng_buffer_view *payload)
1902 {
1903 struct relay_session *session = conn->session;
1904 struct lttcomm_relayd_data_pending msg;
1905 struct lttcomm_relayd_generic_reply reply;
1906 struct relay_stream *stream;
1907 ssize_t send_ret;
1908 int ret;
1909 uint64_t stream_seq;
1910
1911 DBG("Data pending command received");
1912
1913 if (!session || !conn->version_check_done) {
1914 ERR("Trying to check for data before version check");
1915 ret = -1;
1916 goto end_no_session;
1917 }
1918
1919 if (payload->size < sizeof(msg)) {
1920 ERR("Unexpected payload size in \"relay_data_pending\": expected >= %zu bytes, got %zu bytes",
1921 sizeof(msg), payload->size);
1922 ret = -1;
1923 goto end_no_session;
1924 }
1925 memcpy(&msg, payload->data, sizeof(msg));
1926 msg.stream_id = be64toh(msg.stream_id);
1927 msg.last_net_seq_num = be64toh(msg.last_net_seq_num);
1928
1929 stream = stream_get_by_id(msg.stream_id);
1930 if (stream == NULL) {
1931 ret = -1;
1932 goto end;
1933 }
1934
1935 pthread_mutex_lock(&stream->lock);
1936
1937 if (session_streams_have_index(session)) {
1938 /*
1939 * Ensure that both the index and stream data have been
1940 * flushed up to the requested point.
1941 */
1942 stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
1943 } else {
1944 stream_seq = stream->prev_data_seq;
1945 }
1946 DBG("Data pending for stream id %" PRIu64 ": prev_data_seq %" PRIu64
1947 ", prev_index_seq %" PRIu64
1948 ", and last_seq %" PRIu64, msg.stream_id,
1949 stream->prev_data_seq, stream->prev_index_seq,
1950 msg.last_net_seq_num);
1951
1952 /* Avoid wrapping issue */
1953 if (((int64_t) (stream_seq - msg.last_net_seq_num)) >= 0) {
1954 /* Data has in fact been written and is NOT pending */
1955 ret = 0;
1956 } else {
1957 /* Data still being streamed thus pending */
1958 ret = 1;
1959 }
1960
1961 stream->data_pending_check_done = true;
1962 pthread_mutex_unlock(&stream->lock);
1963
1964 stream_put(stream);
1965 end:
1966
1967 memset(&reply, 0, sizeof(reply));
1968 reply.ret_code = htobe32(ret);
1969 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1970 if (send_ret < (ssize_t) sizeof(reply)) {
1971 ERR("Failed to send \"data pending\" command reply (ret = %zd)",
1972 send_ret);
1973 ret = -1;
1974 }
1975
1976 end_no_session:
1977 return ret;
1978 }
1979
1980 /*
1981 * Wait for the control socket to reach a quiescent state.
1982 *
1983 * Note that for now, when receiving this command from the session
1984 * daemon, this means that every subsequent commands or data received on
1985 * the control socket has been handled. So, this is why we simply return
1986 * OK here.
1987 */
1988 static int relay_quiescent_control(const struct lttcomm_relayd_hdr *recv_hdr,
1989 struct relay_connection *conn,
1990 const struct lttng_buffer_view *payload)
1991 {
1992 int ret;
1993 ssize_t send_ret;
1994 struct relay_stream *stream;
1995 struct lttcomm_relayd_quiescent_control msg;
1996 struct lttcomm_relayd_generic_reply reply;
1997
1998 DBG("Checking quiescent state on control socket");
1999
2000 if (!conn->session || !conn->version_check_done) {
2001 ERR("Trying to check for data before version check");
2002 ret = -1;
2003 goto end_no_session;
2004 }
2005
2006 if (payload->size < sizeof(msg)) {
2007 ERR("Unexpected payload size in \"relay_quiescent_control\": expected >= %zu bytes, got %zu bytes",
2008 sizeof(msg), payload->size);
2009 ret = -1;
2010 goto end_no_session;
2011 }
2012 memcpy(&msg, payload->data, sizeof(msg));
2013 msg.stream_id = be64toh(msg.stream_id);
2014
2015 stream = stream_get_by_id(msg.stream_id);
2016 if (!stream) {
2017 goto reply;
2018 }
2019 pthread_mutex_lock(&stream->lock);
2020 stream->data_pending_check_done = true;
2021 pthread_mutex_unlock(&stream->lock);
2022
2023 DBG("Relay quiescent control pending flag set to %" PRIu64, msg.stream_id);
2024 stream_put(stream);
2025 reply:
2026 memset(&reply, 0, sizeof(reply));
2027 reply.ret_code = htobe32(LTTNG_OK);
2028 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2029 if (send_ret < (ssize_t) sizeof(reply)) {
2030 ERR("Failed to send \"quiescent control\" command reply (ret = %zd)",
2031 send_ret);
2032 ret = -1;
2033 } else {
2034 ret = 0;
2035 }
2036
2037 end_no_session:
2038 return ret;
2039 }
2040
2041 /*
2042 * Initialize a data pending command. This means that a consumer is about
2043 * to ask for data pending for each stream it holds. Simply iterate over
2044 * all streams of a session and set the data_pending_check_done flag.
2045 *
2046 * This command returns to the client a LTTNG_OK code.
2047 */
2048 static int relay_begin_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
2049 struct relay_connection *conn,
2050 const struct lttng_buffer_view *payload)
2051 {
2052 int ret;
2053 ssize_t send_ret;
2054 struct lttng_ht_iter iter;
2055 struct lttcomm_relayd_begin_data_pending msg;
2056 struct lttcomm_relayd_generic_reply reply;
2057 struct relay_stream *stream;
2058
2059 assert(recv_hdr);
2060 assert(conn);
2061
2062 DBG("Init streams for data pending");
2063
2064 if (!conn->session || !conn->version_check_done) {
2065 ERR("Trying to check for data before version check");
2066 ret = -1;
2067 goto end_no_session;
2068 }
2069
2070 if (payload->size < sizeof(msg)) {
2071 ERR("Unexpected payload size in \"relay_begin_data_pending\": expected >= %zu bytes, got %zu bytes",
2072 sizeof(msg), payload->size);
2073 ret = -1;
2074 goto end_no_session;
2075 }
2076 memcpy(&msg, payload->data, sizeof(msg));
2077 msg.session_id = be64toh(msg.session_id);
2078
2079 /*
2080 * Iterate over all streams to set the begin data pending flag.
2081 * For now, the streams are indexed by stream handle so we have
2082 * to iterate over all streams to find the one associated with
2083 * the right session_id.
2084 */
2085 rcu_read_lock();
2086 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2087 node.node) {
2088 if (!stream_get(stream)) {
2089 continue;
2090 }
2091 if (stream->trace->session->id == msg.session_id) {
2092 pthread_mutex_lock(&stream->lock);
2093 stream->data_pending_check_done = false;
2094 pthread_mutex_unlock(&stream->lock);
2095 DBG("Set begin data pending flag to stream %" PRIu64,
2096 stream->stream_handle);
2097 }
2098 stream_put(stream);
2099 }
2100 rcu_read_unlock();
2101
2102 memset(&reply, 0, sizeof(reply));
2103 /* All good, send back reply. */
2104 reply.ret_code = htobe32(LTTNG_OK);
2105
2106 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2107 if (send_ret < (ssize_t) sizeof(reply)) {
2108 ERR("Failed to send \"begin data pending\" command reply (ret = %zd)",
2109 send_ret);
2110 ret = -1;
2111 } else {
2112 ret = 0;
2113 }
2114
2115 end_no_session:
2116 return ret;
2117 }
2118
2119 /*
2120 * End data pending command. This will check, for a given session id, if
2121 * each stream associated with it has its data_pending_check_done flag
2122 * set. If not, this means that the client lost track of the stream but
2123 * the data is still being streamed on our side. In this case, we inform
2124 * the client that data is in flight.
2125 *
2126 * Return to the client if there is data in flight or not with a ret_code.
2127 */
2128 static int relay_end_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
2129 struct relay_connection *conn,
2130 const struct lttng_buffer_view *payload)
2131 {
2132 int ret;
2133 ssize_t send_ret;
2134 struct lttng_ht_iter iter;
2135 struct lttcomm_relayd_end_data_pending msg;
2136 struct lttcomm_relayd_generic_reply reply;
2137 struct relay_stream *stream;
2138 uint32_t is_data_inflight = 0;
2139
2140 DBG("End data pending command");
2141
2142 if (!conn->session || !conn->version_check_done) {
2143 ERR("Trying to check for data before version check");
2144 ret = -1;
2145 goto end_no_session;
2146 }
2147
2148 if (payload->size < sizeof(msg)) {
2149 ERR("Unexpected payload size in \"relay_end_data_pending\": expected >= %zu bytes, got %zu bytes",
2150 sizeof(msg), payload->size);
2151 ret = -1;
2152 goto end_no_session;
2153 }
2154 memcpy(&msg, payload->data, sizeof(msg));
2155 msg.session_id = be64toh(msg.session_id);
2156
2157 /*
2158 * Iterate over all streams to see if the begin data pending
2159 * flag is set.
2160 */
2161 rcu_read_lock();
2162 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2163 node.node) {
2164 if (!stream_get(stream)) {
2165 continue;
2166 }
2167 if (stream->trace->session->id != msg.session_id) {
2168 stream_put(stream);
2169 continue;
2170 }
2171 pthread_mutex_lock(&stream->lock);
2172 if (!stream->data_pending_check_done) {
2173 uint64_t stream_seq;
2174
2175 if (session_streams_have_index(conn->session)) {
2176 /*
2177 * Ensure that both the index and stream data have been
2178 * flushed up to the requested point.
2179 */
2180 stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
2181 } else {
2182 stream_seq = stream->prev_data_seq;
2183 }
2184 if (!stream->closed || !(((int64_t) (stream_seq - stream->last_net_seq_num)) >= 0)) {
2185 is_data_inflight = 1;
2186 DBG("Data is still in flight for stream %" PRIu64,
2187 stream->stream_handle);
2188 pthread_mutex_unlock(&stream->lock);
2189 stream_put(stream);
2190 break;
2191 }
2192 }
2193 pthread_mutex_unlock(&stream->lock);
2194 stream_put(stream);
2195 }
2196 rcu_read_unlock();
2197
2198 memset(&reply, 0, sizeof(reply));
2199 /* All good, send back reply. */
2200 reply.ret_code = htobe32(is_data_inflight);
2201
2202 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2203 if (send_ret < (ssize_t) sizeof(reply)) {
2204 ERR("Failed to send \"end data pending\" command reply (ret = %zd)",
2205 send_ret);
2206 ret = -1;
2207 } else {
2208 ret = 0;
2209 }
2210
2211 end_no_session:
2212 return ret;
2213 }
2214
2215 /*
2216 * Receive an index for a specific stream.
2217 *
2218 * Return 0 on success else a negative value.
2219 */
2220 static int relay_recv_index(const struct lttcomm_relayd_hdr *recv_hdr,
2221 struct relay_connection *conn,
2222 const struct lttng_buffer_view *payload)
2223 {
2224 int ret;
2225 ssize_t send_ret;
2226 struct relay_session *session = conn->session;
2227 struct lttcomm_relayd_index index_info;
2228 struct lttcomm_relayd_generic_reply reply;
2229 struct relay_stream *stream;
2230 size_t msg_len;
2231
2232 assert(conn);
2233
2234 DBG("Relay receiving index");
2235
2236 if (!session || !conn->version_check_done) {
2237 ERR("Trying to close a stream before version check");
2238 ret = -1;
2239 goto end_no_session;
2240 }
2241
2242 msg_len = lttcomm_relayd_index_len(
2243 lttng_to_index_major(conn->major, conn->minor),
2244 lttng_to_index_minor(conn->major, conn->minor));
2245 if (payload->size < msg_len) {
2246 ERR("Unexpected payload size in \"relay_recv_index\": expected >= %zu bytes, got %zu bytes",
2247 msg_len, payload->size);
2248 ret = -1;
2249 goto end_no_session;
2250 }
2251 memcpy(&index_info, payload->data, msg_len);
2252 index_info.relay_stream_id = be64toh(index_info.relay_stream_id);
2253 index_info.net_seq_num = be64toh(index_info.net_seq_num);
2254 index_info.packet_size = be64toh(index_info.packet_size);
2255 index_info.content_size = be64toh(index_info.content_size);
2256 index_info.timestamp_begin = be64toh(index_info.timestamp_begin);
2257 index_info.timestamp_end = be64toh(index_info.timestamp_end);
2258 index_info.events_discarded = be64toh(index_info.events_discarded);
2259 index_info.stream_id = be64toh(index_info.stream_id);
2260
2261 if (conn->minor >= 8) {
2262 index_info.stream_instance_id =
2263 be64toh(index_info.stream_instance_id);
2264 index_info.packet_seq_num = be64toh(index_info.packet_seq_num);
2265 } else {
2266 index_info.stream_instance_id = -1ULL;
2267 index_info.packet_seq_num = -1ULL;
2268 }
2269
2270 stream = stream_get_by_id(index_info.relay_stream_id);
2271 if (!stream) {
2272 ERR("stream_get_by_id not found");
2273 ret = -1;
2274 goto end;
2275 }
2276
2277 pthread_mutex_lock(&stream->lock);
2278 ret = stream_add_index(stream, &index_info);
2279 pthread_mutex_unlock(&stream->lock);
2280 if (ret) {
2281 goto end_stream_put;
2282 }
2283
2284 end_stream_put:
2285 stream_put(stream);
2286 end:
2287 memset(&reply, 0, sizeof(reply));
2288 if (ret < 0) {
2289 reply.ret_code = htobe32(LTTNG_ERR_UNK);
2290 } else {
2291 reply.ret_code = htobe32(LTTNG_OK);
2292 }
2293 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2294 if (send_ret < (ssize_t) sizeof(reply)) {
2295 ERR("Failed to send \"recv index\" command reply (ret = %zd)", send_ret);
2296 ret = -1;
2297 }
2298
2299 end_no_session:
2300 return ret;
2301 }
2302
2303 /*
2304 * Receive the streams_sent message.
2305 *
2306 * Return 0 on success else a negative value.
2307 */
2308 static int relay_streams_sent(const struct lttcomm_relayd_hdr *recv_hdr,
2309 struct relay_connection *conn,
2310 const struct lttng_buffer_view *payload)
2311 {
2312 int ret;
2313 ssize_t send_ret;
2314 struct lttcomm_relayd_generic_reply reply;
2315
2316 assert(conn);
2317
2318 DBG("Relay receiving streams_sent");
2319
2320 if (!conn->session || !conn->version_check_done) {
2321 ERR("Trying to close a stream before version check");
2322 ret = -1;
2323 goto end_no_session;
2324 }
2325
2326 /*
2327 * Publish every pending stream in the connection recv list which are
2328 * now ready to be used by the viewer.
2329 */
2330 publish_connection_local_streams(conn);
2331
2332 memset(&reply, 0, sizeof(reply));
2333 reply.ret_code = htobe32(LTTNG_OK);
2334 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2335 if (send_ret < (ssize_t) sizeof(reply)) {
2336 ERR("Failed to send \"streams sent\" command reply (ret = %zd)",
2337 send_ret);
2338 ret = -1;
2339 } else {
2340 /* Success. */
2341 ret = 0;
2342 }
2343
2344 end_no_session:
2345 return ret;
2346 }
2347
2348 /*
2349 * relay_rotate_session_stream: rotate a stream to a new tracefile for the
2350 * session rotation feature (not the tracefile rotation feature).
2351 */
2352 static int relay_rotate_session_streams(
2353 const struct lttcomm_relayd_hdr *recv_hdr,
2354 struct relay_connection *conn,
2355 const struct lttng_buffer_view *payload)
2356 {
2357 int ret = 0;
2358 uint32_t i;
2359 ssize_t send_ret;
2360 enum lttng_error_code reply_code = LTTNG_ERR_UNK;
2361 struct relay_session *session = conn->session;
2362 struct lttcomm_relayd_rotate_streams rotate_streams;
2363 struct lttcomm_relayd_generic_reply reply = {};
2364 struct relay_stream *stream = NULL;
2365 const size_t header_len = sizeof(struct lttcomm_relayd_rotate_streams);
2366 struct lttng_trace_chunk *next_trace_chunk = NULL;
2367 struct lttng_buffer_view stream_positions;
2368 char chunk_id_buf[MAX_INT_DEC_LEN(uint64_t)];
2369 const char *chunk_id_str = "none";
2370
2371 if (!session || !conn->version_check_done) {
2372 ERR("Trying to rotate a stream before version check");
2373 ret = -1;
2374 goto end_no_reply;
2375 }
2376
2377 if (session->major == 2 && session->minor < 11) {
2378 ERR("Unsupported feature before 2.11");
2379 ret = -1;
2380 goto end_no_reply;
2381 }
2382
2383 if (payload->size < header_len) {
2384 ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected >= %zu bytes, got %zu bytes",
2385 header_len, payload->size);
2386 ret = -1;
2387 goto end_no_reply;
2388 }
2389
2390 memcpy(&rotate_streams, payload->data, header_len);
2391
2392 /* Convert header to host endianness. */
2393 rotate_streams = (typeof(rotate_streams)) {
2394 .stream_count = be32toh(rotate_streams.stream_count),
2395 .new_chunk_id = (typeof(rotate_streams.new_chunk_id)) {
2396 .is_set = !!rotate_streams.new_chunk_id.is_set,
2397 .value = be64toh(rotate_streams.new_chunk_id.value),
2398 }
2399 };
2400
2401 if (rotate_streams.new_chunk_id.is_set) {
2402 /*
2403 * Retrieve the trace chunk the stream must transition to. As
2404 * per the protocol, this chunk should have been created
2405 * before this command is received.
2406 */
2407 next_trace_chunk = sessiond_trace_chunk_registry_get_chunk(
2408 sessiond_trace_chunk_registry,
2409 session->sessiond_uuid, session->id,
2410 rotate_streams.new_chunk_id.value);
2411 if (!next_trace_chunk) {
2412 char uuid_str[LTTNG_UUID_STR_LEN];
2413
2414 lttng_uuid_to_str(session->sessiond_uuid, uuid_str);
2415 ERR("Unknown next trace chunk in ROTATE_STREAMS command: sessiond_uuid = {%s}, session_id = %" PRIu64
2416 ", trace_chunk_id = %" PRIu64,
2417 uuid_str, session->id,
2418 rotate_streams.new_chunk_id.value);
2419 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2420 ret = -1;
2421 goto end;
2422 }
2423
2424 ret = snprintf(chunk_id_buf, sizeof(chunk_id_buf), "%" PRIu64,
2425 rotate_streams.new_chunk_id.value);
2426 if (ret < 0 || ret >= sizeof(chunk_id_buf)) {
2427 chunk_id_str = "formatting error";
2428 } else {
2429 chunk_id_str = chunk_id_buf;
2430 }
2431 }
2432
2433 DBG("Rotate %" PRIu32 " streams of session \"%s\" to chunk \"%s\"",
2434 rotate_streams.stream_count, session->session_name,
2435 chunk_id_str);
2436
2437 stream_positions = lttng_buffer_view_from_view(payload,
2438 sizeof(rotate_streams), -1);
2439 if (!stream_positions.data ||
2440 stream_positions.size <
2441 (rotate_streams.stream_count *
2442 sizeof(struct lttcomm_relayd_stream_rotation_position))) {
2443 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2444 ret = -1;
2445 goto end;
2446 }
2447
2448 for (i = 0; i < rotate_streams.stream_count; i++) {
2449 struct lttcomm_relayd_stream_rotation_position *position_comm =
2450 &((typeof(position_comm)) stream_positions.data)[i];
2451 const struct lttcomm_relayd_stream_rotation_position pos = {
2452 .stream_id = be64toh(position_comm->stream_id),
2453 .rotate_at_seq_num = be64toh(
2454 position_comm->rotate_at_seq_num),
2455 };
2456
2457 stream = stream_get_by_id(pos.stream_id);
2458 if (!stream) {
2459 reply_code = LTTNG_ERR_INVALID;
2460 ret = -1;
2461 goto end;
2462 }
2463
2464 pthread_mutex_lock(&stream->lock);
2465 ret = stream_set_pending_rotation(stream, next_trace_chunk,
2466 pos.rotate_at_seq_num);
2467 pthread_mutex_unlock(&stream->lock);
2468 if (ret) {
2469 reply_code = LTTNG_ERR_FILE_CREATION_ERROR;
2470 goto end;
2471 }
2472
2473 stream_put(stream);
2474 stream = NULL;
2475 }
2476
2477 reply_code = LTTNG_OK;
2478 ret = 0;
2479 end:
2480 if (stream) {
2481 stream_put(stream);
2482 }
2483
2484 reply.ret_code = htobe32((uint32_t) reply_code);
2485 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
2486 sizeof(struct lttcomm_relayd_generic_reply), 0);
2487 if (send_ret < (ssize_t) sizeof(reply)) {
2488 ERR("Failed to send \"rotate session stream\" command reply (ret = %zd)",
2489 send_ret);
2490 ret = -1;
2491 }
2492 end_no_reply:
2493 lttng_trace_chunk_put(next_trace_chunk);
2494 return ret;
2495 }
2496
2497
2498
2499 /*
2500 * relay_create_trace_chunk: create a new trace chunk
2501 */
2502 static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr,
2503 struct relay_connection *conn,
2504 const struct lttng_buffer_view *payload)
2505 {
2506 int ret = 0;
2507 ssize_t send_ret;
2508 struct relay_session *session = conn->session;
2509 struct lttcomm_relayd_create_trace_chunk *msg;
2510 struct lttcomm_relayd_generic_reply reply = {};
2511 struct lttng_buffer_view header_view;
2512 struct lttng_buffer_view chunk_name_view;
2513 struct lttng_trace_chunk *chunk = NULL, *published_chunk = NULL;
2514 enum lttng_error_code reply_code = LTTNG_OK;
2515 enum lttng_trace_chunk_status chunk_status;
2516 struct lttng_directory_handle *session_output = NULL;
2517 const char *new_path;
2518
2519 if (!session || !conn->version_check_done) {
2520 ERR("Trying to create a trace chunk before version check");
2521 ret = -1;
2522 goto end_no_reply;
2523 }
2524
2525 if (session->major == 2 && session->minor < 11) {
2526 ERR("Chunk creation command is unsupported before 2.11");
2527 ret = -1;
2528 goto end_no_reply;
2529 }
2530
2531 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2532 if (!header_view.data) {
2533 ERR("Failed to receive payload of chunk creation command");
2534 ret = -1;
2535 goto end_no_reply;
2536 }
2537
2538 /* Convert to host endianness. */
2539 msg = (typeof(msg)) header_view.data;
2540 msg->chunk_id = be64toh(msg->chunk_id);
2541 msg->creation_timestamp = be64toh(msg->creation_timestamp);
2542 msg->override_name_length = be32toh(msg->override_name_length);
2543
2544 if (session->current_trace_chunk &&
2545 !lttng_trace_chunk_get_name_overridden(session->current_trace_chunk)) {
2546 chunk_status = lttng_trace_chunk_rename_path(session->current_trace_chunk,
2547 DEFAULT_CHUNK_TMP_OLD_DIRECTORY);
2548 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2549 ERR("Failed to rename old chunk");
2550 ret = -1;
2551 reply_code = LTTNG_ERR_UNK;
2552 goto end;
2553 }
2554 }
2555 session->ongoing_rotation = true;
2556 if (!session->current_trace_chunk) {
2557 if (!session->has_rotated) {
2558 new_path = "";
2559 } else {
2560 new_path = NULL;
2561 }
2562 } else {
2563 new_path = DEFAULT_CHUNK_TMP_NEW_DIRECTORY;
2564 }
2565 chunk = lttng_trace_chunk_create(
2566 msg->chunk_id, msg->creation_timestamp, new_path);
2567 if (!chunk) {
2568 ERR("Failed to create trace chunk in trace chunk creation command");
2569 ret = -1;
2570 reply_code = LTTNG_ERR_NOMEM;
2571 goto end;
2572 }
2573
2574 if (msg->override_name_length) {
2575 const char *name;
2576
2577 chunk_name_view = lttng_buffer_view_from_view(payload,
2578 sizeof(*msg),
2579 msg->override_name_length);
2580 name = chunk_name_view.data;
2581 if (!name || name[msg->override_name_length - 1]) {
2582 ERR("Failed to receive payload of chunk creation command");
2583 ret = -1;
2584 reply_code = LTTNG_ERR_INVALID;
2585 goto end;
2586 }
2587
2588 chunk_status = lttng_trace_chunk_override_name(
2589 chunk, chunk_name_view.data);
2590 switch (chunk_status) {
2591 case LTTNG_TRACE_CHUNK_STATUS_OK:
2592 break;
2593 case LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT:
2594 ERR("Failed to set the name of new trace chunk in trace chunk creation command (invalid name)");
2595 reply_code = LTTNG_ERR_INVALID;
2596 ret = -1;
2597 goto end;
2598 default:
2599 ERR("Failed to set the name of new trace chunk in trace chunk creation command (unknown error)");
2600 reply_code = LTTNG_ERR_UNK;
2601 ret = -1;
2602 goto end;
2603 }
2604 }
2605
2606 chunk_status = lttng_trace_chunk_set_credentials_current_user(chunk);
2607 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2608 reply_code = LTTNG_ERR_UNK;
2609 ret = -1;
2610 goto end;
2611 }
2612
2613 session_output = session_create_output_directory_handle(
2614 conn->session);
2615 if (!session_output) {
2616 reply_code = LTTNG_ERR_CREATE_DIR_FAIL;
2617 goto end;
2618 }
2619 chunk_status = lttng_trace_chunk_set_as_owner(chunk, session_output);
2620 lttng_directory_handle_put(session_output);
2621 session_output = NULL;
2622 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2623 reply_code = LTTNG_ERR_UNK;
2624 ret = -1;
2625 goto end;
2626 }
2627
2628 published_chunk = sessiond_trace_chunk_registry_publish_chunk(
2629 sessiond_trace_chunk_registry,
2630 conn->session->sessiond_uuid,
2631 conn->session->id,
2632 chunk);
2633 if (!published_chunk) {
2634 char uuid_str[LTTNG_UUID_STR_LEN];
2635
2636 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
2637 ERR("Failed to publish chunk: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
2638 uuid_str,
2639 conn->session->id,
2640 msg->chunk_id);
2641 ret = -1;
2642 reply_code = LTTNG_ERR_NOMEM;
2643 goto end;
2644 }
2645
2646 pthread_mutex_lock(&conn->session->lock);
2647 if (conn->session->pending_closure_trace_chunk) {
2648 /*
2649 * Invalid; this means a second create_trace_chunk command was
2650 * received before a close_trace_chunk.
2651 */
2652 ERR("Invalid trace chunk close command received; a trace chunk is already waiting for a trace chunk close command");
2653 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2654 ret = -1;
2655 goto end_unlock_session;
2656 }
2657 conn->session->pending_closure_trace_chunk =
2658 conn->session->current_trace_chunk;
2659 conn->session->current_trace_chunk = published_chunk;
2660 published_chunk = NULL;
2661 if (!conn->session->pending_closure_trace_chunk) {
2662 session->ongoing_rotation = false;
2663 }
2664 end_unlock_session:
2665 pthread_mutex_unlock(&conn->session->lock);
2666 end:
2667 reply.ret_code = htobe32((uint32_t) reply_code);
2668 send_ret = conn->sock->ops->sendmsg(conn->sock,
2669 &reply,
2670 sizeof(struct lttcomm_relayd_generic_reply),
2671 0);
2672 if (send_ret < (ssize_t) sizeof(reply)) {
2673 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
2674 send_ret);
2675 ret = -1;
2676 }
2677 end_no_reply:
2678 lttng_trace_chunk_put(chunk);
2679 lttng_trace_chunk_put(published_chunk);
2680 lttng_directory_handle_put(session_output);
2681 return ret;
2682 }
2683
2684 /*
2685 * relay_close_trace_chunk: close a trace chunk
2686 */
2687 static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr,
2688 struct relay_connection *conn,
2689 const struct lttng_buffer_view *payload)
2690 {
2691 int ret = 0, buf_ret;
2692 ssize_t send_ret;
2693 struct relay_session *session = conn->session;
2694 struct lttcomm_relayd_close_trace_chunk *msg;
2695 struct lttcomm_relayd_close_trace_chunk_reply reply = {};
2696 struct lttng_buffer_view header_view;
2697 struct lttng_trace_chunk *chunk = NULL;
2698 enum lttng_error_code reply_code = LTTNG_OK;
2699 enum lttng_trace_chunk_status chunk_status;
2700 uint64_t chunk_id;
2701 LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command = {};
2702 time_t close_timestamp;
2703 char closed_trace_chunk_path[LTTNG_PATH_MAX];
2704 size_t path_length = 0;
2705 const char *chunk_name = NULL;
2706 struct lttng_dynamic_buffer reply_payload;
2707 const char *new_path;
2708
2709 lttng_dynamic_buffer_init(&reply_payload);
2710
2711 if (!session || !conn->version_check_done) {
2712 ERR("Trying to close a trace chunk before version check");
2713 ret = -1;
2714 goto end_no_reply;
2715 }
2716
2717 if (session->major == 2 && session->minor < 11) {
2718 ERR("Chunk close command is unsupported before 2.11");
2719 ret = -1;
2720 goto end_no_reply;
2721 }
2722
2723 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2724 if (!header_view.data) {
2725 ERR("Failed to receive payload of chunk close command");
2726 ret = -1;
2727 goto end_no_reply;
2728 }
2729
2730 /* Convert to host endianness. */
2731 msg = (typeof(msg)) header_view.data;
2732 chunk_id = be64toh(msg->chunk_id);
2733 close_timestamp = (time_t) be64toh(msg->close_timestamp);
2734 close_command = (typeof(close_command)){
2735 .value = be32toh(msg->close_command.value),
2736 .is_set = msg->close_command.is_set,
2737 };
2738
2739 chunk = sessiond_trace_chunk_registry_get_chunk(
2740 sessiond_trace_chunk_registry,
2741 conn->session->sessiond_uuid,
2742 conn->session->id,
2743 chunk_id);
2744 if (!chunk) {
2745 char uuid_str[LTTNG_UUID_STR_LEN];
2746
2747 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
2748 ERR("Failed to find chunk to close: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
2749 uuid_str,
2750 conn->session->id,
2751 msg->chunk_id);
2752 ret = -1;
2753 reply_code = LTTNG_ERR_NOMEM;
2754 goto end;
2755 }
2756
2757 pthread_mutex_lock(&session->lock);
2758 if (close_command.is_set &&
2759 close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_DELETE) {
2760 /*
2761 * Clear command. It is a protocol error to ask for a
2762 * clear on a relay which does not allow it. Querying
2763 * the configuration allows figuring out whether
2764 * clearing is allowed before doing the clear.
2765 */
2766 if (!opt_allow_clear) {
2767 ret = -1;
2768 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2769 goto end_unlock_session;
2770 }
2771 }
2772 if (session->pending_closure_trace_chunk &&
2773 session->pending_closure_trace_chunk != chunk) {
2774 ERR("Trace chunk close command for session \"%s\" does not target the trace chunk pending closure",
2775 session->session_name);
2776 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2777 ret = -1;
2778 goto end_unlock_session;
2779 }
2780
2781 if (session->current_trace_chunk && session->current_trace_chunk != chunk &&
2782 !lttng_trace_chunk_get_name_overridden(session->current_trace_chunk)) {
2783 if (close_command.is_set &&
2784 close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_DELETE &&
2785 !session->has_rotated) {
2786 /* New chunk stays in session output directory. */
2787 new_path = "";
2788 } else {
2789 /* Use chunk name for new chunk. */
2790 new_path = NULL;
2791 }
2792 /* Rename new chunk path. */
2793 chunk_status = lttng_trace_chunk_rename_path(session->current_trace_chunk,
2794 new_path);
2795 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2796 ret = -1;
2797 goto end;
2798 }
2799 session->ongoing_rotation = false;
2800 }
2801 if ((!close_command.is_set ||
2802 close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_NO_OPERATION) &&
2803 !lttng_trace_chunk_get_name_overridden(chunk)) {
2804 const char *old_path;
2805
2806 if (!session->has_rotated) {
2807 old_path = "";
2808 } else {
2809 old_path = NULL;
2810 }
2811 /* We need to move back the .tmp_old_chunk to its rightful place. */
2812 chunk_status = lttng_trace_chunk_rename_path(chunk, old_path);
2813 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2814 ret = -1;
2815 goto end;
2816 }
2817 }
2818 chunk_status = lttng_trace_chunk_set_close_timestamp(
2819 chunk, close_timestamp);
2820 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2821 ERR("Failed to set trace chunk close timestamp");
2822 ret = -1;
2823 reply_code = LTTNG_ERR_UNK;
2824 goto end_unlock_session;
2825 }
2826
2827 if (close_command.is_set) {
2828 chunk_status = lttng_trace_chunk_set_close_command(
2829 chunk, close_command.value);
2830 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2831 ret = -1;
2832 reply_code = LTTNG_ERR_INVALID;
2833 goto end_unlock_session;
2834 }
2835 }
2836 chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name, NULL);
2837 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2838 ERR("Failed to get chunk name");
2839 ret = -1;
2840 reply_code = LTTNG_ERR_UNK;
2841 goto end_unlock_session;
2842 }
2843 if (!session->has_rotated && !session->snapshot) {
2844 ret = lttng_strncpy(closed_trace_chunk_path,
2845 session->output_path,
2846 sizeof(closed_trace_chunk_path));
2847 if (ret) {
2848 ERR("Failed to send trace chunk path: path length of %zu bytes exceeds the maximal allowed length of %zu bytes",
2849 strlen(session->output_path),
2850 sizeof(closed_trace_chunk_path));
2851 reply_code = LTTNG_ERR_NOMEM;
2852 ret = -1;
2853 goto end_unlock_session;
2854 }
2855 } else {
2856 if (session->snapshot) {
2857 ret = snprintf(closed_trace_chunk_path,
2858 sizeof(closed_trace_chunk_path),
2859 "%s/%s", session->output_path,
2860 chunk_name);
2861 } else {
2862 ret = snprintf(closed_trace_chunk_path,
2863 sizeof(closed_trace_chunk_path),
2864 "%s/" DEFAULT_ARCHIVED_TRACE_CHUNKS_DIRECTORY
2865 "/%s",
2866 session->output_path, chunk_name);
2867 }
2868 if (ret < 0 || ret == sizeof(closed_trace_chunk_path)) {
2869 ERR("Failed to format closed trace chunk resulting path");
2870 reply_code = ret < 0 ? LTTNG_ERR_UNK : LTTNG_ERR_NOMEM;
2871 ret = -1;
2872 goto end_unlock_session;
2873 }
2874 }
2875 if (close_command.is_set &&
2876 close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED) {
2877 session->has_rotated = true;
2878 }
2879 DBG("Reply chunk path on close: %s", closed_trace_chunk_path);
2880 path_length = strlen(closed_trace_chunk_path) + 1;
2881 if (path_length > UINT32_MAX) {
2882 ERR("Closed trace chunk path exceeds the maximal length allowed by the protocol");
2883 ret = -1;
2884 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2885 goto end_unlock_session;
2886 }
2887
2888 if (session->current_trace_chunk == chunk) {
2889 /*
2890 * After a trace chunk close command, no new streams
2891 * referencing the chunk may be created. Hence, on the
2892 * event that no new trace chunk have been created for
2893 * the session, the reference to the current trace chunk
2894 * is released in order to allow it to be reclaimed when
2895 * the last stream releases its reference to it.
2896 */
2897 lttng_trace_chunk_put(session->current_trace_chunk);
2898 session->current_trace_chunk = NULL;
2899 }
2900 lttng_trace_chunk_put(session->pending_closure_trace_chunk);
2901 session->pending_closure_trace_chunk = NULL;
2902 end_unlock_session:
2903 pthread_mutex_unlock(&session->lock);
2904
2905 end:
2906 reply.generic.ret_code = htobe32((uint32_t) reply_code);
2907 reply.path_length = htobe32((uint32_t) path_length);
2908 buf_ret = lttng_dynamic_buffer_append(
2909 &reply_payload, &reply, sizeof(reply));
2910 if (buf_ret) {
2911 ERR("Failed to append \"close trace chunk\" command reply header to payload buffer");
2912 goto end_no_reply;
2913 }
2914
2915 if (reply_code == LTTNG_OK) {
2916 buf_ret = lttng_dynamic_buffer_append(&reply_payload,
2917 closed_trace_chunk_path, path_length);
2918 if (buf_ret) {
2919 ERR("Failed to append \"close trace chunk\" command reply path to payload buffer");
2920 goto end_no_reply;
2921 }
2922 }
2923
2924 send_ret = conn->sock->ops->sendmsg(conn->sock,
2925 reply_payload.data,
2926 reply_payload.size,
2927 0);
2928 if (send_ret < reply_payload.size) {
2929 ERR("Failed to send \"close trace chunk\" command reply of %zu bytes (ret = %zd)",
2930 reply_payload.size, send_ret);
2931 ret = -1;
2932 goto end_no_reply;
2933 }
2934 end_no_reply:
2935 lttng_trace_chunk_put(chunk);
2936 lttng_dynamic_buffer_reset(&reply_payload);
2937 return ret;
2938 }
2939
2940 /*
2941 * relay_trace_chunk_exists: check if a trace chunk exists
2942 */
2943 static int relay_trace_chunk_exists(const struct lttcomm_relayd_hdr *recv_hdr,
2944 struct relay_connection *conn,
2945 const struct lttng_buffer_view *payload)
2946 {
2947 int ret = 0;
2948 ssize_t send_ret;
2949 struct relay_session *session = conn->session;
2950 struct lttcomm_relayd_trace_chunk_exists *msg;
2951 struct lttcomm_relayd_trace_chunk_exists_reply reply = {};
2952 struct lttng_buffer_view header_view;
2953 uint64_t chunk_id;
2954 bool chunk_exists;
2955
2956 if (!session || !conn->version_check_done) {
2957 ERR("Trying to close a trace chunk before version check");
2958 ret = -1;
2959 goto end_no_reply;
2960 }
2961
2962 if (session->major == 2 && session->minor < 11) {
2963 ERR("Chunk close command is unsupported before 2.11");
2964 ret = -1;
2965 goto end_no_reply;
2966 }
2967
2968 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2969 if (!header_view.data) {
2970 ERR("Failed to receive payload of chunk close command");
2971 ret = -1;
2972 goto end_no_reply;
2973 }
2974
2975 /* Convert to host endianness. */
2976 msg = (typeof(msg)) header_view.data;
2977 chunk_id = be64toh(msg->chunk_id);
2978
2979 ret = sessiond_trace_chunk_registry_chunk_exists(
2980 sessiond_trace_chunk_registry,
2981 conn->session->sessiond_uuid,
2982 conn->session->id,
2983 chunk_id, &chunk_exists);
2984 /*
2985 * If ret is not 0, send the reply and report the error to the caller.
2986 * It is a protocol (or internal) error and the session/connection
2987 * should be torn down.
2988 */
2989 reply = (typeof(reply)){
2990 .generic.ret_code = htobe32((uint32_t)
2991 (ret == 0 ? LTTNG_OK : LTTNG_ERR_INVALID_PROTOCOL)),
2992 .trace_chunk_exists = ret == 0 ? chunk_exists : 0,
2993 };
2994 send_ret = conn->sock->ops->sendmsg(
2995 conn->sock, &reply, sizeof(reply), 0);
2996 if (send_ret < (ssize_t) sizeof(reply)) {
2997 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
2998 send_ret);
2999 ret = -1;
3000 }
3001 end_no_reply:
3002 return ret;
3003 }
3004
3005 /*
3006 * relay_get_configuration: query whether feature is available
3007 */
3008 static int relay_get_configuration(const struct lttcomm_relayd_hdr *recv_hdr,
3009 struct relay_connection *conn,
3010 const struct lttng_buffer_view *payload)
3011 {
3012 int ret = 0;
3013 ssize_t send_ret;
3014 struct lttcomm_relayd_get_configuration *msg;
3015 struct lttcomm_relayd_get_configuration_reply reply = {};
3016 struct lttng_buffer_view header_view;
3017 uint64_t query_flags = 0;
3018 uint64_t result_flags = 0;
3019
3020 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
3021 if (!header_view.data) {
3022 ERR("Failed to receive payload of chunk close command");
3023 ret = -1;
3024 goto end_no_reply;
3025 }
3026
3027 /* Convert to host endianness. */
3028 msg = (typeof(msg)) header_view.data;
3029 query_flags = be64toh(msg->query_flags);
3030
3031 if (query_flags) {
3032 ret = LTTNG_ERR_INVALID_PROTOCOL;
3033 goto reply;
3034 }
3035 if (opt_allow_clear) {
3036 result_flags |= LTTCOMM_RELAYD_CONFIGURATION_FLAG_CLEAR_ALLOWED;
3037 }
3038 ret = 0;
3039 reply:
3040 reply = (typeof(reply)){
3041 .generic.ret_code = htobe32((uint32_t)
3042 (ret == 0 ? LTTNG_OK : LTTNG_ERR_INVALID_PROTOCOL)),
3043 .relayd_configuration_flags = htobe64(result_flags),
3044 };
3045 send_ret = conn->sock->ops->sendmsg(
3046 conn->sock, &reply, sizeof(reply), 0);
3047 if (send_ret < (ssize_t) sizeof(reply)) {
3048 ERR("Failed to send \"get configuration\" command reply (ret = %zd)",
3049 send_ret);
3050 ret = -1;
3051 }
3052 end_no_reply:
3053 return ret;
3054 }
3055
3056 #define DBG_CMD(cmd_name, conn) \
3057 DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd);
3058
3059 static int relay_process_control_command(struct relay_connection *conn,
3060 const struct lttcomm_relayd_hdr *header,
3061 const struct lttng_buffer_view *payload)
3062 {
3063 int ret = 0;
3064
3065 switch (header->cmd) {
3066 case RELAYD_CREATE_SESSION:
3067 DBG_CMD("RELAYD_CREATE_SESSION", conn);
3068 ret = relay_create_session(header, conn, payload);
3069 break;
3070 case RELAYD_ADD_STREAM:
3071 DBG_CMD("RELAYD_ADD_STREAM", conn);
3072 ret = relay_add_stream(header, conn, payload);
3073 break;
3074 case RELAYD_START_DATA:
3075 DBG_CMD("RELAYD_START_DATA", conn);
3076 ret = relay_start(header, conn, payload);
3077 break;
3078 case RELAYD_SEND_METADATA:
3079 DBG_CMD("RELAYD_SEND_METADATA", conn);
3080 ret = relay_recv_metadata(header, conn, payload);
3081 break;
3082 case RELAYD_VERSION:
3083 DBG_CMD("RELAYD_VERSION", conn);
3084 ret = relay_send_version(header, conn, payload);
3085 break;
3086 case RELAYD_CLOSE_STREAM:
3087 DBG_CMD("RELAYD_CLOSE_STREAM", conn);
3088 ret = relay_close_stream(header, conn, payload);
3089 break;
3090 case RELAYD_DATA_PENDING:
3091 DBG_CMD("RELAYD_DATA_PENDING", conn);
3092 ret = relay_data_pending(header, conn, payload);
3093 break;
3094 case RELAYD_QUIESCENT_CONTROL:
3095 DBG_CMD("RELAYD_QUIESCENT_CONTROL", conn);
3096 ret = relay_quiescent_control(header, conn, payload);
3097 break;
3098 case RELAYD_BEGIN_DATA_PENDING:
3099 DBG_CMD("RELAYD_BEGIN_DATA_PENDING", conn);
3100 ret = relay_begin_data_pending(header, conn, payload);
3101 break;
3102 case RELAYD_END_DATA_PENDING:
3103 DBG_CMD("RELAYD_END_DATA_PENDING", conn);
3104 ret = relay_end_data_pending(header, conn, payload);
3105 break;
3106 case RELAYD_SEND_INDEX:
3107 DBG_CMD("RELAYD_SEND_INDEX", conn);
3108 ret = relay_recv_index(header, conn, payload);
3109 break;
3110 case RELAYD_STREAMS_SENT:
3111 DBG_CMD("RELAYD_STREAMS_SENT", conn);
3112 ret = relay_streams_sent(header, conn, payload);
3113 break;
3114 case RELAYD_RESET_METADATA:
3115 DBG_CMD("RELAYD_RESET_METADATA", conn);
3116 ret = relay_reset_metadata(header, conn, payload);
3117 break;
3118 case RELAYD_ROTATE_STREAMS:
3119 DBG_CMD("RELAYD_ROTATE_STREAMS", conn);
3120 ret = relay_rotate_session_streams(header, conn, payload);
3121 break;
3122 case RELAYD_CREATE_TRACE_CHUNK:
3123 DBG_CMD("RELAYD_CREATE_TRACE_CHUNK", conn);
3124 ret = relay_create_trace_chunk(header, conn, payload);
3125 break;
3126 case RELAYD_CLOSE_TRACE_CHUNK:
3127 DBG_CMD("RELAYD_CLOSE_TRACE_CHUNK", conn);
3128 ret = relay_close_trace_chunk(header, conn, payload);
3129 break;
3130 case RELAYD_TRACE_CHUNK_EXISTS:
3131 DBG_CMD("RELAYD_TRACE_CHUNK_EXISTS", conn);
3132 ret = relay_trace_chunk_exists(header, conn, payload);
3133 break;
3134 case RELAYD_GET_CONFIGURATION:
3135 DBG_CMD("RELAYD_GET_CONFIGURATION", conn);
3136 ret = relay_get_configuration(header, conn, payload);
3137 break;
3138 case RELAYD_UPDATE_SYNC_INFO:
3139 default:
3140 ERR("Received unknown command (%u)", header->cmd);
3141 relay_unknown_command(conn);
3142 ret = -1;
3143 goto end;
3144 }
3145
3146 end:
3147 return ret;
3148 }
3149
3150 static enum relay_connection_status relay_process_control_receive_payload(
3151 struct relay_connection *conn)
3152 {
3153 int ret = 0;
3154 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3155 struct lttng_dynamic_buffer *reception_buffer =
3156 &conn->protocol.ctrl.reception_buffer;
3157 struct ctrl_connection_state_receive_payload *state =
3158 &conn->protocol.ctrl.state.receive_payload;
3159 struct lttng_buffer_view payload_view;
3160
3161 if (state->left_to_receive == 0<