[lttng-tools.git] / src / bin / lttng-sessiond / kernel-consumer.c

/*
 * Copyright (C) 2012 - David Goulet <dgoulet@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, as
 * published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>

#include <common/common.h>
#include <common/defaults.h>

#include "consumer.h"
#include "kernel-consumer.h"

/*
 * Sending a single channel to the consumer with command ADD_CHANNEL.
 */
int kernel_consumer_add_channel(int sock, struct ltt_kernel_channel *channel)
{
	int ret;
	struct lttcomm_consumer_msg lkm;

	/* Safety net */
	assert(channel);

	DBG("Kernel consumer adding channel %s to kernel consumer",
			channel->channel->name);

	/* Prep channel message structure */
	consumer_init_channel_comm_msg(&lkm,
			LTTNG_CONSUMER_ADD_CHANNEL,
			channel->fd,
			channel->channel->attr.subbuf_size,
			0, /* Kernel */
			channel->channel->name);

	ret = consumer_send_channel(sock, &lkm);
	if (ret < 0) {
		goto error;
	}

error:
	return ret;
}

/*
 * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 */
int kernel_consumer_add_metadata(int sock, struct ltt_kernel_session *session)
{
	int ret;
	char tmp_path[PATH_MAX];
	const char *pathname;
	struct lttcomm_consumer_msg lkm;
	struct consumer_output *consumer;

	/* Safety net */
	assert(session);
	assert(session->consumer);

	DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd);

	/* Get consumer output pointer */
	consumer = session->consumer;

	/* Get the right path name destination */
	if (consumer->type == CONSUMER_DST_LOCAL) {
		/* Set application path to the destination path */
		ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
				consumer->dst.trace_path, consumer->subdir);
		if (ret < 0) {
			PERROR("snprintf metadata path");
			goto error;
		}
		pathname = tmp_path;

		/* Create directory */
		ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG,
				session->uid, session->gid);
		if (ret < 0) {
			if (ret != -EEXIST) {
				ERR("Trace directory creation error");
				goto error;
			}
		}
		DBG3("Kernel local consumer tracefile path: %s", pathname);
	} else {
		ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
		if (ret < 0) {
			PERROR("snprintf metadata path");
			goto error;
		}
		pathname = tmp_path;
		DBG3("Kernel network consumer subdir path: %s", pathname);
	}

	/* Prep channel message structure */
	consumer_init_channel_comm_msg(&lkm,
			LTTNG_CONSUMER_ADD_CHANNEL,
			session->metadata->fd,
			session->metadata->conf->attr.subbuf_size,
			0, /* for kernel */
			"metadata");

	ret = consumer_send_channel(sock, &lkm);
	if (ret < 0) {
		goto error;
	}

	/* Prep stream message structure */
	consumer_init_stream_comm_msg(&lkm,
			LTTNG_CONSUMER_ADD_STREAM,
			session->metadata->fd,
			session->metadata_stream_fd,
			LTTNG_CONSUMER_ACTIVE_STREAM,
			DEFAULT_KERNEL_CHANNEL_OUTPUT,
			0, /* Kernel */
			session->uid,
			session->gid,
			consumer->net_seq_index,
			1, /* Metadata flag set */
			"metadata",
			pathname);

	/* Send stream and file descriptor */
	ret = consumer_send_stream(sock, consumer, &lkm,
			&session->metadata_stream_fd, 1);
	if (ret < 0) {
		goto error;
	}

error:
	return ret;
}

/*
 * Sending a single stream to the consumer with command ADD_STREAM.
 */
int kernel_consumer_add_stream(int sock, struct ltt_kernel_channel *channel,
		struct ltt_kernel_stream *stream, struct ltt_kernel_session *session)
{
	int ret;
	char tmp_path[PATH_MAX];
	const char *pathname;
	struct lttcomm_consumer_msg lkm;
	struct consumer_output *consumer;

	assert(channel);
	assert(stream);
	assert(session);
	assert(session->consumer);

	DBG("Sending stream %d of channel %s to kernel consumer",
			stream->fd, channel->channel->name);

	/* Get consumer output pointer */
	consumer = session->consumer;

	/* Get the right path name destination */
	if (consumer->type == CONSUMER_DST_LOCAL) {
		/* Set application path to the destination path */
		ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
				consumer->dst.trace_path, consumer->subdir);
		if (ret < 0) {
			PERROR("snprintf stream path");
			goto error;
		}
		pathname = tmp_path;
		DBG3("Kernel local consumer tracefile path: %s", pathname);
	} else {
		ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
		if (ret < 0) {
			PERROR("snprintf stream path");
			goto error;
		}
		pathname = tmp_path;
		DBG3("Kernel network consumer subdir path: %s", pathname);
	}

	/* Prep stream consumer message */
	consumer_init_stream_comm_msg(&lkm, LTTNG_CONSUMER_ADD_STREAM,
			channel->fd,
			stream->fd,
			stream->state,
			channel->channel->attr.output,
			0, /* Kernel */
			session->uid,
			session->gid,
			consumer->net_seq_index,
			0, /* Metadata flag unset */
			stream->name,
			pathname);

	/* Send stream and file descriptor */
	ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
	if (ret < 0) {
		goto error;
	}

error:
	return ret;
}

/*
 * Send all stream fds of kernel channel to the consumer.
 */
int kernel_consumer_send_channel_stream(int sock,
		struct ltt_kernel_channel *channel, struct ltt_kernel_session *session)
{
	int ret;
	struct ltt_kernel_stream *stream;

	/* Safety net */
	assert(channel);
	assert(session);
	assert(session->consumer);

	/* Bail out if consumer is disabled */
	if (!session->consumer->enabled) {
		ret = LTTNG_OK;
		goto error;
	}

	DBG("Sending streams of channel %s to kernel consumer",
			channel->channel->name);

	ret = kernel_consumer_add_channel(sock, channel);
	if (ret < 0) {
		goto error;
	}

	/* Send streams */
	cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
		if (!stream->fd) {
			continue;
		}

		/* Add stream on the kernel consumer side. */
		ret = kernel_consumer_add_stream(sock, channel, stream, session);
		if (ret < 0) {
			goto error;
		}
	}

error:
	return ret;
}

/*
 * Send all stream fds of the kernel session to the consumer.
 */
int kernel_consumer_send_session(int sock, struct ltt_kernel_session *session)
{
	int ret;
	struct ltt_kernel_channel *chan;

	/* Safety net */
	assert(session);
	assert(session->consumer);

	/* Bail out if consumer is disabled */
	if (!session->consumer->enabled) {
		ret = LTTNG_OK;
		goto error;
	}

	DBG("Sending session stream to kernel consumer");

	if (session->metadata_stream_fd >= 0) {
		ret = kernel_consumer_add_metadata(sock, session);
		if (ret < 0) {
			goto error;
		}

		/* Flag that at least the metadata has been sent to the consumer. */
		session->consumer_fds_sent = 1;
	}

	/* Send channel and streams of it */
	cds_list_for_each_entry(chan, &session->channel_list.head, list) {
		ret = kernel_consumer_send_channel_stream(sock, chan, session);
		if (ret < 0) {
			goto error;
		}
	}

	DBG("Kernel consumer FDs of metadata and channel streams sent");

	return 0;

error:
	return ret;
}
Commit	Line	Data
f1e16794 DG	1	/*
	2	* Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
	3	*
	4	* This program is free software; you can redistribute it and/or modify it
	5	* under the terms of the GNU General Public License, version 2 only, as
	6	* published by the Free Software Foundation.
	7	*
	8	* This program is distributed in the hope that it will be useful, but WITHOUT
	9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	10	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
	11	* more details.
	12	*
	13	* You should have received a copy of the GNU General Public License along with
	14	* this program; if not, write to the Free Software Foundation, Inc., 51
	15	* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	16	*/
	17
	18	#define _GNU_SOURCE
	19	#include <stdio.h>
	20	#include <stdlib.h>
	21	#include <string.h>
	22	#include <sys/stat.h>
	23	#include <unistd.h>
	24
	25	#include <common/common.h>
	26	#include <common/defaults.h>
f1e16794	27
00e2e675	28	#include "consumer.h"
f1e16794 DG	29	#include "kernel-consumer.h"
f1e16794 DG	30
00e2e675 DG	31	/*
	32	* Sending a single channel to the consumer with command ADD_CHANNEL.
	33	*/
	34	int kernel_consumer_add_channel(int sock, struct ltt_kernel_channel *channel)
	35	{
	36	int ret;
	37	struct lttcomm_consumer_msg lkm;
	38
	39	/* Safety net */
	40	assert(channel);
	41
	42	DBG("Kernel consumer adding channel %s to kernel consumer",
	43	channel->channel->name);
	44
	45	/* Prep channel message structure */
	46	consumer_init_channel_comm_msg(&lkm,
	47	LTTNG_CONSUMER_ADD_CHANNEL,
	48	channel->fd,
	49	channel->channel->attr.subbuf_size,
	50	0, /* Kernel */
	51	channel->channel->name);
	52
	53	ret = consumer_send_channel(sock, &lkm);
	54	if (ret < 0) {
	55	goto error;
	56	}
	57
	58	error:
	59	return ret;
	60	}
	61
	62	/*
	63	* Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
	64	*/
	65	int kernel_consumer_add_metadata(int sock, struct ltt_kernel_session *session)
	66	{
	67	int ret;
a7d9a3e7	68	char tmp_path[PATH_MAX];
00e2e675 DG	69	const char *pathname;
00e2e675 DG	70	struct lttcomm_consumer_msg lkm;
a7d9a3e7	71	struct consumer_output *consumer;
00e2e675 DG	72
	73	/* Safety net */
	74	assert(session);
	75	assert(session->consumer);
	76
	77	DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd);
	78
	79	/* Get consumer output pointer */
a7d9a3e7	80	consumer = session->consumer;
00e2e675	81
a7d9a3e7 DG	82	/* Get the right path name destination */
	83	if (consumer->type == CONSUMER_DST_LOCAL) {
	84	/* Set application path to the destination path */
	85	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
	86	consumer->dst.trace_path, consumer->subdir);
	87	if (ret < 0) {
	88	PERROR("snprintf metadata path");
	89	goto error;
	90	}
	91	pathname = tmp_path;
	92
	93	/* Create directory */
	94	ret = run_as_mkdir_recursive(pathname, S_IRWXU \| S_IRWXG,
	95	session->uid, session->gid);
	96	if (ret < 0) {
	97	if (ret != -EEXIST) {
	98	ERR("Trace directory creation error");
	99	goto error;
	100	}
	101	}
	102	DBG3("Kernel local consumer tracefile path: %s", pathname);
00e2e675	103	} else {
a7d9a3e7 DG	104	ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
	105	if (ret < 0) {
	106	PERROR("snprintf metadata path");
	107	goto error;
	108	}
	109	pathname = tmp_path;
	110	DBG3("Kernel network consumer subdir path: %s", pathname);
00e2e675 DG	111	}
	112
	113	/* Prep channel message structure */
	114	consumer_init_channel_comm_msg(&lkm,
	115	LTTNG_CONSUMER_ADD_CHANNEL,
	116	session->metadata->fd,
	117	session->metadata->conf->attr.subbuf_size,
	118	0, /* for kernel */
	119	"metadata");
	120
	121	ret = consumer_send_channel(sock, &lkm);
	122	if (ret < 0) {
	123	goto error;
	124	}
	125
	126	/* Prep stream message structure */
	127	consumer_init_stream_comm_msg(&lkm,
	128	LTTNG_CONSUMER_ADD_STREAM,
	129	session->metadata->fd,
	130	session->metadata_stream_fd,
	131	LTTNG_CONSUMER_ACTIVE_STREAM,
	132	DEFAULT_KERNEL_CHANNEL_OUTPUT,
	133	0, /* Kernel */
	134	session->uid,
	135	session->gid,
a7d9a3e7	136	consumer->net_seq_index,
00e2e675 DG	137	1, /* Metadata flag set */
	138	"metadata",
	139	pathname);
	140
	141	/* Send stream and file descriptor */
a7d9a3e7	142	ret = consumer_send_stream(sock, consumer, &lkm,
00e2e675 DG	143	&session->metadata_stream_fd, 1);
	144	if (ret < 0) {
	145	goto error;
	146	}
	147
	148	error:
	149	return ret;
	150	}
	151
	152	/*
	153	* Sending a single stream to the consumer with command ADD_STREAM.
	154	*/
	155	int kernel_consumer_add_stream(int sock, struct ltt_kernel_channel *channel,
	156	struct ltt_kernel_stream stream, struct ltt_kernel_session session)
	157	{
	158	int ret;
a7d9a3e7	159	char tmp_path[PATH_MAX];
00e2e675 DG	160	const char *pathname;
00e2e675 DG	161	struct lttcomm_consumer_msg lkm;
a7d9a3e7	162	struct consumer_output *consumer;
00e2e675 DG	163
	164	assert(channel);
	165	assert(stream);
	166	assert(session);
	167	assert(session->consumer);
	168
	169	DBG("Sending stream %d of channel %s to kernel consumer",
	170	stream->fd, channel->channel->name);
	171
	172	/* Get consumer output pointer */
a7d9a3e7	173	consumer = session->consumer;
00e2e675	174
a7d9a3e7 DG	175	/* Get the right path name destination */
	176	if (consumer->type == CONSUMER_DST_LOCAL) {
	177	/* Set application path to the destination path */
	178	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
	179	consumer->dst.trace_path, consumer->subdir);
	180	if (ret < 0) {
	181	PERROR("snprintf stream path");
	182	goto error;
	183	}
	184	pathname = tmp_path;
	185	DBG3("Kernel local consumer tracefile path: %s", pathname);
00e2e675	186	} else {
a7d9a3e7 DG	187	ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
	188	if (ret < 0) {
	189	PERROR("snprintf stream path");
	190	goto error;
	191	}
	192	pathname = tmp_path;
	193	DBG3("Kernel network consumer subdir path: %s", pathname);
00e2e675 DG	194	}
	195
	196	/* Prep stream consumer message */
	197	consumer_init_stream_comm_msg(&lkm, LTTNG_CONSUMER_ADD_STREAM,
	198	channel->fd,
	199	stream->fd,
	200	stream->state,
	201	channel->channel->attr.output,
	202	0, /* Kernel */
	203	session->uid,
	204	session->gid,
a7d9a3e7	205	consumer->net_seq_index,
00e2e675 DG	206	0, /* Metadata flag unset */
	207	stream->name,
	208	pathname);
	209
	210	/* Send stream and file descriptor */
a7d9a3e7	211	ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
00e2e675 DG	212	if (ret < 0) {
	213	goto error;
	214	}
	215
	216	error:
	217	return ret;
	218	}
	219
f1e16794 DG	220	/*
	221	* Send all stream fds of kernel channel to the consumer.
	222	*/
00e2e675 DG	223	int kernel_consumer_send_channel_stream(int sock,
00e2e675 DG	224	struct ltt_kernel_channel channel, struct ltt_kernel_session session)
f1e16794	225	{
00e2e675	226	int ret;
f1e16794	227	struct ltt_kernel_stream *stream;
00e2e675 DG	228
	229	/* Safety net */
	230	assert(channel);
	231	assert(session);
	232	assert(session->consumer);
	233
	234	/* Bail out if consumer is disabled */
	235	if (!session->consumer->enabled) {
f73fabfd	236	ret = LTTNG_OK;
00e2e675 DG	237	goto error;
00e2e675 DG	238	}
f1e16794 DG	239
	240	DBG("Sending streams of channel %s to kernel consumer",
	241	channel->channel->name);
	242
00e2e675	243	ret = kernel_consumer_add_channel(sock, channel);
f1e16794	244	if (ret < 0) {
f1e16794 DG	245	goto error;
	246	}
	247
	248	/* Send streams */
	249	cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
	250	if (!stream->fd) {
	251	continue;
	252	}
00e2e675 DG	253
	254	/* Add stream on the kernel consumer side. */
	255	ret = kernel_consumer_add_stream(sock, channel, stream, session);
f1e16794	256	if (ret < 0) {
f1e16794 DG	257	goto error;
	258	}
	259	}
	260
f1e16794 DG	261	error:
	262	return ret;
	263	}
	264
	265	/*
	266	* Send all stream fds of the kernel session to the consumer.
	267	*/
00e2e675	268	int kernel_consumer_send_session(int sock, struct ltt_kernel_session *session)
f1e16794 DG	269	{
	270	int ret;
	271	struct ltt_kernel_channel *chan;
f1e16794	272
00e2e675 DG	273	/* Safety net */
	274	assert(session);
	275	assert(session->consumer);
f1e16794	276
00e2e675 DG	277	/* Bail out if consumer is disabled */
00e2e675 DG	278	if (!session->consumer->enabled) {
f73fabfd	279	ret = LTTNG_OK;
00e2e675	280	goto error;
f1e16794 DG	281	}
f1e16794 DG	282
00e2e675 DG	283	DBG("Sending session stream to kernel consumer");
00e2e675 DG	284
f1e16794	285	if (session->metadata_stream_fd >= 0) {
00e2e675	286	ret = kernel_consumer_add_metadata(sock, session);
f1e16794	287	if (ret < 0) {
f1e16794 DG	288	goto error;
	289	}
	290
00e2e675 DG	291	/* Flag that at least the metadata has been sent to the consumer. */
00e2e675 DG	292	session->consumer_fds_sent = 1;
f1e16794 DG	293	}
f1e16794 DG	294
00e2e675	295	/* Send channel and streams of it */
f1e16794	296	cds_list_for_each_entry(chan, &session->channel_list.head, list) {
00e2e675	297	ret = kernel_consumer_send_channel_stream(sock, chan, session);
f1e16794 DG	298	if (ret < 0) {
	299	goto error;
	300	}
	301	}
	302
00e2e675	303	DBG("Kernel consumer FDs of metadata and channel streams sent");
f1e16794 DG	304
	305	return 0;
	306
	307	error:
	308	return ret;
	309	}