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