X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=7e5733ec8c3253432bd61dbac17b0968fefef21a;hp=a6b408a90740943138e43ad69160d34a5395baa6;hb=a4baae1b0463bc4ce65c2a458c4a941e7fabc594;hpb=157df58664846e22bdeea84dfcf717cb43360b3f

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index a6b408a90..7e5733ec8 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -1,6 +1,7 @@
 /*
  * Copyright (C) 2012 - Julien Desfossez <jdesfossez@efficios.com>
  *                      David Goulet <dgoulet@efficios.com>
+ *               2013 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2 only,
@@ -50,6 +51,7 @@
 #include <common/sessiond-comm/relayd.h>
 #include <common/uri.h>
 #include <common/utils.h>
+#include <common/config/config.h>
 
 #include "cmd.h"
 #include "ctf-trace.h"
@@ -69,6 +71,9 @@ static struct lttng_uri *live_uri;
 const char *progname;
 
 const char *tracing_group_name = DEFAULT_TRACING_GROUP;
+static int tracing_group_name_override;
+
+const char * const config_section_name = "relayd";
 
 /*
  * Quit pipe for all threads. This permits a single cancellation point
@@ -121,6 +126,21 @@ struct lttng_ht *indexes_ht;
 /* Relayd health monitoring */
 struct health_app *health_relayd;
 
+static struct option long_options[] = {
+	{ "control-port", 1, 0, 'C', },
+	{ "data-port", 1, 0, 'D', },
+	{ "live-port", 1, 0, 'L', },
+	{ "daemonize", 0, 0, 'd', },
+	{ "group", 1, 0, 'g', },
+	{ "help", 0, 0, 'h', },
+	{ "output", 1, 0, 'o', },
+	{ "verbose", 0, 0, 'v', },
+	{ "config", 1, 0, 'f' },
+	{ NULL, 0, 0, 0, },
+};
+
+static const char *config_ignore_options[] = { "help", "config" };
+
 /*
  * usage function on stderr
  */
@@ -132,88 +152,210 @@ void usage(void)
 	fprintf(stderr, "  -d, --daemonize           Start as a daemon.\n");
 	fprintf(stderr, "  -C, --control-port URL    Control port listening.\n");
 	fprintf(stderr, "  -D, --data-port URL       Data port listening.\n");
+	fprintf(stderr, "  -L, --live-port URL       Live view port listening.\n");
 	fprintf(stderr, "  -o, --output PATH         Output path for traces. Must use an absolute path.\n");
 	fprintf(stderr, "  -v, --verbose             Verbose mode. Activate DBG() macro.\n");
 	fprintf(stderr, "  -g, --group NAME          Specify the tracing group name. (default: tracing)\n");
+	fprintf(stderr, "  -f  --config              Load daemon configuration file\n");
 }
 
+/*
+ * Take an option from the getopt output and set it in the right variable to be
+ * used later.
+ *
+ * Return 0 on success else a negative value.
+ */
 static
-int parse_args(int argc, char **argv)
+int set_option(int opt, const char *arg, const char *optname)
 {
-	int c;
-	int ret = 0;
-	char *default_address;
-
-	static struct option long_options[] = {
-		{ "control-port", 1, 0, 'C', },
-		{ "data-port", 1, 0, 'D', },
-		{ "daemonize", 0, 0, 'd', },
-		{ "group", 1, 0, 'g', },
-		{ "help", 0, 0, 'h', },
-		{ "output", 1, 0, 'o', },
-		{ "verbose", 0, 0, 'v', },
-		{ NULL, 0, 0, 0, },
-	};
+	int ret;
+
+	switch (opt) {
+	case 0:
+		fprintf(stderr, "option %s", optname);
+		if (arg) {
+			fprintf(stderr, " with arg %s\n", arg);
+		}
+		break;
+	case 'C':
+		ret = uri_parse(arg, &control_uri);
+		if (ret < 0) {
+			ERR("Invalid control URI specified");
+			goto end;
+		}
+		if (control_uri->port == 0) {
+			control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
+		}
+		break;
+	case 'D':
+		ret = uri_parse(arg, &data_uri);
+		if (ret < 0) {
+			ERR("Invalid data URI specified");
+			goto end;
+		}
+		if (data_uri->port == 0) {
+			data_uri->port = DEFAULT_NETWORK_DATA_PORT;
+		}
+		break;
+	case 'L':
+		ret = uri_parse(arg, &live_uri);
+		if (ret < 0) {
+			ERR("Invalid live URI specified");
+			goto end;
+		}
+		if (live_uri->port == 0) {
+			live_uri->port = DEFAULT_NETWORK_VIEWER_PORT;
+		}
+		break;
+	case 'd':
+		opt_daemon = 1;
+		break;
+	case 'g':
+		tracing_group_name = strdup(arg);
+		tracing_group_name_override = 1;
+		break;
+	case 'h':
+		usage();
+		exit(EXIT_FAILURE);
+	case 'o':
+		ret = asprintf(&opt_output_path, "%s", arg);
+		if (ret < 0) {
+			ret = -errno;
+			PERROR("asprintf opt_output_path");
+			goto end;
+		}
+		break;
+	case 'v':
+		/* Verbose level can increase using multiple -v */
+		if (arg) {
+			lttng_opt_verbose = config_parse_value(arg);
+		} else {
+			lttng_opt_verbose += 1;
+		}
+		break;
+	default:
+		/* Unknown option or other error.
+		 * Error is printed by getopt, just return */
+		ret = -1;
+		goto end;
+	}
+
+	/* All good. */
+	ret = 0;
+
+end:
+	return ret;
+}
+
+/*
+ * config_entry_handler_cb used to handle options read from a config file.
+ * See config_entry_handler_cb comment in common/config/config.h for the
+ * return value conventions.
+ */
+static
+int config_entry_handler(const struct config_entry *entry, void *unused)
+{
+	int ret = 0, i;
+
+	if (!entry || !entry->name || !entry->value) {
+		ret = -EINVAL;
+		goto end;
+	}
 
+	/* Check if the option is to be ignored */
+	for (i = 0; i < sizeof(config_ignore_options) / sizeof(char *); i++) {
+		if (!strcmp(entry->name, config_ignore_options[i])) {
+			goto end;
+		}
+	}
+
+	for (i = 0; i < (sizeof(long_options) / sizeof(struct option)) - 1; i++) {
+		/* Ignore if entry name is not fully matched. */
+		if (strcmp(entry->name, long_options[i].name)) {
+			continue;
+		}
+
+		/*
+		 * If the option takes no argument on the command line, we have to
+		 * check if the value is "true". We support non-zero numeric values,
+		 * true, on and yes.
+		 */
+		if (!long_options[i].has_arg) {
+			ret = config_parse_value(entry->value);
+			if (ret <= 0) {
+				if (ret) {
+					WARN("Invalid configuration value \"%s\" for option %s",
+							entry->value, entry->name);
+				}
+				/* False, skip boolean config option. */
+				goto end;
+			}
+		}
+
+		ret = set_option(long_options[i].val, entry->value, entry->name);
+		goto end;
+	}
+
+	WARN("Unrecognized option \"%s\" in daemon configuration file.",
+			entry->name);
+
+end:
+	return ret;
+}
+
+static
+int set_options(int argc, char **argv)
+{
+	int c, ret = 0, option_index = 0;
+	int orig_optopt = optopt, orig_optind = optind;
+	char *default_address, *optstring;
+	const char *config_path = NULL;
+
+	optstring = utils_generate_optstring(long_options,
+			sizeof(long_options) / sizeof(struct option));
+	if (!optstring) {
+		ret = -ENOMEM;
+		goto exit;
+	}
+
+	/* Check for the --config option */
+
+	while ((c = getopt_long(argc, argv, optstring, long_options,
+					&option_index)) != -1) {
+		if (c == '?') {
+			ret = -EINVAL;
+			goto exit;
+		} else if (c != 'f') {
+			continue;
+		}
+
+		config_path = utils_expand_path(optarg);
+		if (!config_path) {
+			ERR("Failed to resolve path: %s", optarg);
+		}
+	}
+
+	ret = config_get_section_entries(config_path, config_section_name,
+			config_entry_handler, NULL);
+	if (ret) {
+		if (ret > 0) {
+			ERR("Invalid configuration option at line %i", ret);
+			ret = -1;
+		}
+		goto exit;
+	}
+
+	/* Reset getopt's global state */
+	optopt = orig_optopt;
+	optind = orig_optind;
 	while (1) {
-		int option_index = 0;
-		c = getopt_long(argc, argv, "dhv" "C:D:o:g:",
-				long_options, &option_index);
+		c = getopt_long(argc, argv, optstring, long_options, &option_index);
 		if (c == -1) {
 			break;
 		}
 
-		switch (c) {
-		case 0:
-			fprintf(stderr, "option %s", long_options[option_index].name);
-			if (optarg) {
-				fprintf(stderr, " with arg %s\n", optarg);
-			}
-			break;
-		case 'C':
-			ret = uri_parse(optarg, &control_uri);
-			if (ret < 0) {
-				ERR("Invalid control URI specified");
-				goto exit;
-			}
-			if (control_uri->port == 0) {
-				control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
-			}
-			break;
-		case 'D':
-			ret = uri_parse(optarg, &data_uri);
-			if (ret < 0) {
-				ERR("Invalid data URI specified");
-				goto exit;
-			}
-			if (data_uri->port == 0) {
-				data_uri->port = DEFAULT_NETWORK_DATA_PORT;
-			}
-			break;
-		case 'd':
-			opt_daemon = 1;
-			break;
-		case 'g':
-			tracing_group_name = optarg;
-			break;
-		case 'h':
-			usage();
-			exit(EXIT_FAILURE);
-		case 'o':
-			ret = asprintf(&opt_output_path, "%s", optarg);
-			if (ret < 0) {
-				PERROR("asprintf opt_output_path");
-				goto exit;
-			}
-			break;
-		case 'v':
-			/* Verbose level can increase using multiple -v */
-			lttng_opt_verbose += 1;
-			break;
-		default:
-			/* Unknown option or other error.
-			 * Error is printed by getopt, just return */
-			ret = -1;
+		ret = set_option(c, optarg, long_options[option_index].name);
+		if (ret < 0) {
 			goto exit;
 		}
 	}
@@ -266,6 +408,7 @@ int parse_args(int argc, char **argv)
 	}
 
 exit:
+	free(optstring);
 	return ret;
 }
 
@@ -285,6 +428,11 @@ void cleanup(void)
 
 	uri_free(control_uri);
 	uri_free(data_uri);
+	/* Live URI is freed in the live thread. */
+
+	if (tracing_group_name_override) {
+		free((void *) tracing_group_name);
+	}
 }
 
 /*
@@ -999,6 +1147,70 @@ error:
 	return ret;
 }
 
+/*
+ * When we have received all the streams and the metadata for a channel,
+ * we make them visible to the viewer threads.
+ */
+static
+void set_viewer_ready_flag(struct relay_command *cmd)
+{
+	struct relay_stream_recv_handle *node, *tmp_node;
+
+	cds_list_for_each_entry_safe(node, tmp_node, &cmd->recv_head, node) {
+		struct relay_stream *stream;
+
+		rcu_read_lock();
+		stream = relay_stream_find_by_id(node->id);
+		if (!stream) {
+			/*
+			 * Stream is most probably being cleaned up by the data thread thus
+			 * simply continue to the next one.
+			 */
+			continue;
+		}
+
+		/*
+		 * If any of the streams in the list doesn't have a ctf_trace assigned,
+		 * it means that we never received the metadata stream, so we have to
+		 * wait until it arrives to make the streams available to the viewer.
+		 */
+		if (!stream->ctf_trace) {
+			goto end;
+		}
+
+		stream->viewer_ready = 1;
+		rcu_read_unlock();
+
+		/* Clean stream handle node. */
+		cds_list_del(&node->node);
+		free(node);
+	}
+
+end:
+	return;
+}
+
+/*
+ * Add a recv handle node to the connection recv list with the given stream
+ * handle. A new node is allocated thus must be freed when the node is deleted
+ * from the list.
+ */
+static void queue_stream_handle(uint64_t handle, struct relay_command *cmd)
+{
+	struct relay_stream_recv_handle *node;
+
+	assert(cmd);
+
+	node = zmalloc(sizeof(*node));
+	if (!node) {
+		PERROR("zmalloc queue stream handle");
+		return;
+	}
+
+	node->id = handle;
+	cds_list_add(&node->node, &cmd->recv_head);
+}
+
 /*
  * relay_add_stream: allocate a new stream for a session
  */
@@ -1091,6 +1303,13 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
 	ctf_trace_assign(cmd->ctf_traces_ht, stream);
 	stream->ctf_traces_ht = cmd->ctf_traces_ht;
 
+	/*
+	 * Add the stream handle in the recv list of the connection. Once the end
+	 * stream message is received, this list is emptied and streams are set
+	 * with the viewer ready flag.
+	 */
+	queue_stream_handle(stream->stream_handle, cmd);
+
 	lttng_ht_node_init_ulong(&stream->stream_n,
 			(unsigned long) stream->stream_handle);
 	lttng_ht_add_unique_ulong(relay_streams_ht,
@@ -1839,6 +2058,48 @@ end_no_session:
 	return ret;
 }
 
+/*
+ * Receive the streams_sent message.
+ *
+ * Return 0 on success else a negative value.
+ */
+static
+int relay_streams_sent(struct lttcomm_relayd_hdr *recv_hdr,
+		struct relay_command *cmd)
+{
+	int ret, send_ret;
+	struct lttcomm_relayd_generic_reply reply;
+
+	assert(cmd);
+
+	DBG("Relay receiving streams_sent");
+
+	if (!cmd->session || cmd->version_check_done == 0) {
+		ERR("Trying to close a stream before version check");
+		ret = -1;
+		goto end_no_session;
+	}
+
+	/*
+	 * Flag every pending stream in the connection recv list that they are
+	 * ready to be used by the viewer.
+	 */
+	set_viewer_ready_flag(cmd);
+
+	reply.ret_code = htobe32(LTTNG_OK);
+	send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
+	if (send_ret < 0) {
+		ERR("Relay sending sent_stream reply");
+		ret = send_ret;
+	} else {
+		/* Success. */
+		ret = 0;
+	}
+
+end_no_session:
+	return ret;
+}
+
 /*
  * Process the commands received on the control socket
  */
@@ -1882,6 +2143,9 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
 	case RELAYD_SEND_INDEX:
 		ret = relay_recv_index(recv_hdr, cmd);
 		break;
+	case RELAYD_STREAMS_SENT:
+		ret = relay_streams_sent(recv_hdr, cmd);
+		break;
 	case RELAYD_UPDATE_SYNC_INFO:
 	default:
 		ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
@@ -2174,6 +2438,7 @@ int relay_add_connection(int fd, struct lttng_poll_event *events,
 		PERROR("read relay cmd pipe");
 		goto error_read;
 	}
+	CDS_INIT_LIST_HEAD(&relay_connection->recv_head);
 
 	/*
 	 * Only used by the control side and the reference is copied inside each
@@ -2225,8 +2490,17 @@ void relay_del_connection(struct lttng_ht *relay_connections_ht,
 	assert(!ret);
 
 	if (relay_connection->type == RELAY_CONTROL) {
+		struct relay_stream_recv_handle *node, *tmp_node;
+
 		relay_delete_session(relay_connection, sessions_ht);
 		lttng_ht_destroy(relay_connection->ctf_traces_ht);
+
+		/* Clean up recv list. */
+		cds_list_for_each_entry_safe(node, tmp_node,
+				&relay_connection->recv_head, node) {
+			cds_list_del(&node->node);
+			free(node);
+		}
 	}
 
 	call_rcu(&relay_connection->rcu_node, deferred_free_connection);
@@ -2555,7 +2829,7 @@ int main(int argc, char **argv)
 
 	/* Parse arguments */
 	progname = argv[0];
-	if ((ret = parse_args(argc, argv)) < 0) {
+	if ((ret = set_options(argc, argv)) < 0) {
 		goto exit;
 	}
 
@@ -2592,7 +2866,7 @@ int main(int argc, char **argv)
 
 	/* Check if daemon is UID = 0 */
 	if (relayd_uid == 0) {
-		if (control_uri->port < 1024 || data_uri->port < 1024) {
+		if (control_uri->port < 1024 || data_uri->port < 1024 || live_uri->port < 1024) {
 			ERR("Need to be root to use ports < 1024");
 			ret = -1;
 			goto exit;