[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, 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(channel);
	assert(session);
	assert(session->consumer);

	consumer = session->consumer;

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

	/* 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,
			channel->fd,
			session->id,
			pathname,
			session->uid,
			session->gid,
			consumer->net_seq_index,
			channel->channel->name,
			channel->stream_count,
			channel->channel->attr.output,
			CONSUMER_CHANNEL_TYPE_DATA);

	health_code_update();

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

	health_code_update();

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->id,
			pathname,
			session->uid,
			session->gid,
			consumer->net_seq_index,
			DEFAULT_METADATA_NAME,
			1,
			DEFAULT_KERNEL_CHANNEL_OUTPUT,
			CONSUMER_CHANNEL_TYPE_METADATA);

	health_code_update();

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

	health_code_update();

	/* Prep stream message structure */
	consumer_init_stream_comm_msg(&lkm,
			LTTNG_CONSUMER_ADD_STREAM,
			session->metadata->fd,
			session->metadata_stream_fd,
			0);	/* CPU: 0 for metadata. */

	health_code_update();

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

	health_code_update();

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;
	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;

	/* Prep stream consumer message */
	consumer_init_stream_comm_msg(&lkm,
			LTTNG_CONSUMER_ADD_STREAM,
			channel->fd,
			stream->fd,
			stream->cpu);

	health_code_update();

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

	health_code_update();

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