[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(struct consumer_socket *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,
			channel->stream_count);

	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(struct consumer_socket *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);
	assert(sock);

	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",
			1);

	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,
			session->id);

	/* 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(struct consumer_socket *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);
	assert(sock);

	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,
			session->id);

	/* 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(struct consumer_socket *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);
	assert(sock);

	/* 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(struct consumer_socket *sock,
		struct ltt_kernel_session *session)
{
	int ret;
	struct ltt_kernel_channel *chan;

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

	/* 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	*/
f50f23d9 DG	34	int kernel_consumer_add_channel(struct consumer_socket *sock,
f50f23d9 DG	35	struct ltt_kernel_channel *channel)
00e2e675 DG	36	{
	37	int ret;
	38	struct lttcomm_consumer_msg lkm;
	39
	40	/* Safety net */
	41	assert(channel);
	42
	43	DBG("Kernel consumer adding channel %s to kernel consumer",
	44	channel->channel->name);
	45
	46	/* Prep channel message structure */
	47	consumer_init_channel_comm_msg(&lkm,
	48	LTTNG_CONSUMER_ADD_CHANNEL,
	49	channel->fd,
	50	channel->channel->attr.subbuf_size,
	51	0, /* Kernel */
c30aaa51 MD	52	channel->channel->name,
c30aaa51 MD	53	channel->stream_count);
00e2e675 DG	54
	55	ret = consumer_send_channel(sock, &lkm);
	56	if (ret < 0) {
	57	goto error;
	58	}
	59
	60	error:
	61	return ret;
	62	}
	63
	64	/*
	65	* Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
	66	*/
f50f23d9 DG	67	int kernel_consumer_add_metadata(struct consumer_socket *sock,
f50f23d9 DG	68	struct ltt_kernel_session *session)
00e2e675 DG	69	{
00e2e675 DG	70	int ret;
a7d9a3e7	71	char tmp_path[PATH_MAX];
00e2e675 DG	72	const char *pathname;
00e2e675 DG	73	struct lttcomm_consumer_msg lkm;
a7d9a3e7	74	struct consumer_output *consumer;
00e2e675 DG	75
	76	/* Safety net */
	77	assert(session);
	78	assert(session->consumer);
f50f23d9	79	assert(sock);
00e2e675 DG	80
	81	DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd);
	82
	83	/* Get consumer output pointer */
a7d9a3e7	84	consumer = session->consumer;
00e2e675	85
a7d9a3e7 DG	86	/* Get the right path name destination */
	87	if (consumer->type == CONSUMER_DST_LOCAL) {
	88	/* Set application path to the destination path */
	89	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
	90	consumer->dst.trace_path, consumer->subdir);
	91	if (ret < 0) {
	92	PERROR("snprintf metadata path");
	93	goto error;
	94	}
	95	pathname = tmp_path;
	96
	97	/* Create directory */
	98	ret = run_as_mkdir_recursive(pathname, S_IRWXU \| S_IRWXG,
	99	session->uid, session->gid);
	100	if (ret < 0) {
	101	if (ret != -EEXIST) {
	102	ERR("Trace directory creation error");
	103	goto error;
	104	}
	105	}
	106	DBG3("Kernel local consumer tracefile path: %s", pathname);
00e2e675	107	} else {
a7d9a3e7 DG	108	ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
	109	if (ret < 0) {
	110	PERROR("snprintf metadata path");
	111	goto error;
	112	}
	113	pathname = tmp_path;
	114	DBG3("Kernel network consumer subdir path: %s", pathname);
00e2e675 DG	115	}
	116
	117	/* Prep channel message structure */
	118	consumer_init_channel_comm_msg(&lkm,
	119	LTTNG_CONSUMER_ADD_CHANNEL,
	120	session->metadata->fd,
	121	session->metadata->conf->attr.subbuf_size,
	122	0, /* for kernel */
c30aaa51 MD	123	"metadata",
c30aaa51 MD	124	1);
00e2e675 DG	125
	126	ret = consumer_send_channel(sock, &lkm);
	127	if (ret < 0) {
	128	goto error;
	129	}
	130
	131	/* Prep stream message structure */
	132	consumer_init_stream_comm_msg(&lkm,
	133	LTTNG_CONSUMER_ADD_STREAM,
	134	session->metadata->fd,
	135	session->metadata_stream_fd,
	136	LTTNG_CONSUMER_ACTIVE_STREAM,
	137	DEFAULT_KERNEL_CHANNEL_OUTPUT,
	138	0, /* Kernel */
	139	session->uid,
	140	session->gid,
a7d9a3e7	141	consumer->net_seq_index,
00e2e675 DG	142	1, /* Metadata flag set */
00e2e675 DG	143	"metadata",
ca22feea DG	144	pathname,
ca22feea DG	145	session->id);
00e2e675 DG	146
00e2e675 DG	147	/* Send stream and file descriptor */
a7d9a3e7	148	ret = consumer_send_stream(sock, consumer, &lkm,
00e2e675 DG	149	&session->metadata_stream_fd, 1);
	150	if (ret < 0) {
	151	goto error;
	152	}
	153
	154	error:
	155	return ret;
	156	}
	157
	158	/*
	159	* Sending a single stream to the consumer with command ADD_STREAM.
	160	*/
f50f23d9 DG	161	int kernel_consumer_add_stream(struct consumer_socket *sock,
	162	struct ltt_kernel_channel channel, struct ltt_kernel_stream stream,
	163	struct ltt_kernel_session *session)
00e2e675 DG	164	{
00e2e675 DG	165	int ret;
a7d9a3e7	166	char tmp_path[PATH_MAX];
00e2e675 DG	167	const char *pathname;
00e2e675 DG	168	struct lttcomm_consumer_msg lkm;
a7d9a3e7	169	struct consumer_output *consumer;
00e2e675 DG	170
	171	assert(channel);
	172	assert(stream);
	173	assert(session);
	174	assert(session->consumer);
f50f23d9	175	assert(sock);
00e2e675 DG	176
	177	DBG("Sending stream %d of channel %s to kernel consumer",
	178	stream->fd, channel->channel->name);
	179
	180	/* Get consumer output pointer */
a7d9a3e7	181	consumer = session->consumer;
00e2e675	182
a7d9a3e7 DG	183	/* Get the right path name destination */
	184	if (consumer->type == CONSUMER_DST_LOCAL) {
	185	/* Set application path to the destination path */
	186	ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
	187	consumer->dst.trace_path, consumer->subdir);
	188	if (ret < 0) {
	189	PERROR("snprintf stream path");
	190	goto error;
	191	}
	192	pathname = tmp_path;
	193	DBG3("Kernel local consumer tracefile path: %s", pathname);
00e2e675	194	} else {
a7d9a3e7 DG	195	ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
	196	if (ret < 0) {
	197	PERROR("snprintf stream path");
	198	goto error;
	199	}
	200	pathname = tmp_path;
	201	DBG3("Kernel network consumer subdir path: %s", pathname);
00e2e675 DG	202	}
	203
	204	/* Prep stream consumer message */
	205	consumer_init_stream_comm_msg(&lkm, LTTNG_CONSUMER_ADD_STREAM,
	206	channel->fd,
	207	stream->fd,
	208	stream->state,
	209	channel->channel->attr.output,
	210	0, /* Kernel */
	211	session->uid,
	212	session->gid,
a7d9a3e7	213	consumer->net_seq_index,
00e2e675 DG	214	0, /* Metadata flag unset */
00e2e675 DG	215	stream->name,
ca22feea DG	216	pathname,
ca22feea DG	217	session->id);
00e2e675 DG	218
00e2e675 DG	219	/* Send stream and file descriptor */
a7d9a3e7	220	ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
00e2e675 DG	221	if (ret < 0) {
	222	goto error;
	223	}
	224
	225	error:
	226	return ret;
	227	}
	228
f1e16794 DG	229	/*
	230	* Send all stream fds of kernel channel to the consumer.
	231	*/
f50f23d9	232	int kernel_consumer_send_channel_stream(struct consumer_socket *sock,
00e2e675	233	struct ltt_kernel_channel channel, struct ltt_kernel_session session)
f1e16794	234	{
00e2e675	235	int ret;
f1e16794	236	struct ltt_kernel_stream *stream;
00e2e675 DG	237
	238	/* Safety net */
	239	assert(channel);
	240	assert(session);
	241	assert(session->consumer);
f50f23d9	242	assert(sock);
00e2e675 DG	243
	244	/* Bail out if consumer is disabled */
	245	if (!session->consumer->enabled) {
f73fabfd	246	ret = LTTNG_OK;
00e2e675 DG	247	goto error;
00e2e675 DG	248	}
f1e16794 DG	249
	250	DBG("Sending streams of channel %s to kernel consumer",
	251	channel->channel->name);
	252
00e2e675	253	ret = kernel_consumer_add_channel(sock, channel);
f1e16794	254	if (ret < 0) {
f1e16794 DG	255	goto error;
	256	}
	257
	258	/* Send streams */
	259	cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
	260	if (!stream->fd) {
	261	continue;
	262	}
00e2e675 DG	263
	264	/* Add stream on the kernel consumer side. */
	265	ret = kernel_consumer_add_stream(sock, channel, stream, session);
f1e16794	266	if (ret < 0) {
f1e16794 DG	267	goto error;
	268	}
	269	}
	270
f1e16794 DG	271	error:
	272	return ret;
	273	}
	274
	275	/*
	276	* Send all stream fds of the kernel session to the consumer.
	277	*/
f50f23d9 DG	278	int kernel_consumer_send_session(struct consumer_socket *sock,
f50f23d9 DG	279	struct ltt_kernel_session *session)
f1e16794 DG	280	{
	281	int ret;
	282	struct ltt_kernel_channel *chan;
f1e16794	283
00e2e675 DG	284	/* Safety net */
	285	assert(session);
	286	assert(session->consumer);
f50f23d9	287	assert(sock);
f1e16794	288
00e2e675 DG	289	/* Bail out if consumer is disabled */
00e2e675 DG	290	if (!session->consumer->enabled) {
f73fabfd	291	ret = LTTNG_OK;
00e2e675	292	goto error;
f1e16794 DG	293	}
f1e16794 DG	294
00e2e675 DG	295	DBG("Sending session stream to kernel consumer");
00e2e675 DG	296
f1e16794	297	if (session->metadata_stream_fd >= 0) {
00e2e675	298	ret = kernel_consumer_add_metadata(sock, session);
f1e16794	299	if (ret < 0) {
f1e16794 DG	300	goto error;
	301	}
	302
00e2e675 DG	303	/* Flag that at least the metadata has been sent to the consumer. */
00e2e675 DG	304	session->consumer_fds_sent = 1;
f1e16794 DG	305	}
f1e16794 DG	306
00e2e675	307	/* Send channel and streams of it */
f1e16794	308	cds_list_for_each_entry(chan, &session->channel_list.head, list) {
00e2e675	309	ret = kernel_consumer_send_channel_stream(sock, chan, session);
f1e16794 DG	310	if (ret < 0) {
	311	goto error;
	312	}
	313	}
	314
00e2e675	315	DBG("Kernel consumer FDs of metadata and channel streams sent");
f1e16794 DG	316
	317	return 0;
	318
	319	error:
	320	return ret;
	321	}