[lttng-tools.git] / src / bin / lttng-consumerd / health-consumerd.c

/*
 * Copyright (C) 2013 - Mathieu Desnoyers <mathieu.desnoyers@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
#define _LGPL_SOURCE
#include <fcntl.h>
#include <getopt.h>
#include <grp.h>
#include <limits.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ipc.h>
#include <sys/resource.h>
#include <sys/shm.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <urcu/list.h>
#include <poll.h>
#include <unistd.h>
#include <sys/mman.h>
#include <assert.h>
#include <config.h>
#include <urcu/compiler.h>
#include <ulimit.h>
#include <inttypes.h>

#include <common/defaults.h>
#include <common/common.h>
#include <common/consumer.h>
#include <common/consumer-timer.h>
#include <common/compat/poll.h>
#include <common/sessiond-comm/sessiond-comm.h>
#include <common/utils.h>

#include "lttng-consumerd.h"
#include "health-consumerd.h"

/* Global health check unix path */
static char health_unix_sock_path[PATH_MAX];

int health_quit_pipe[2];

/*
 * Check if the thread quit pipe was triggered.
 *
 * Return 1 if it was triggered else 0;
 */
static
int check_health_quit_pipe(int fd, uint32_t events)
{
	if (fd == health_quit_pipe[0] && (events & LPOLLIN)) {
		return 1;
	}

	return 0;
}

/*
 * Send data on a unix socket using the liblttsessiondcomm API.
 *
 * Return lttcomm error code.
 */
static int send_unix_sock(int sock, void *buf, size_t len)
{
	/* Check valid length */
	if (len == 0) {
		return -1;
	}

	return lttcomm_send_unix_sock(sock, buf, len);
}

static
int setup_health_path(void)
{
	int is_root, ret = 0;
	enum lttng_consumer_type type;
	const char *home_path;

	type = lttng_consumer_get_type();
	is_root = !getuid();

	if (is_root) {
		if (strlen(health_unix_sock_path) != 0) {
			goto end;
		}
		switch (type) {
		case LTTNG_CONSUMER_KERNEL:
			snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
				DEFAULT_GLOBAL_KCONSUMER_HEALTH_UNIX_SOCK);
			break;
		case LTTNG_CONSUMER64_UST:
			snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
				DEFAULT_GLOBAL_USTCONSUMER64_HEALTH_UNIX_SOCK);
			break;
		case LTTNG_CONSUMER32_UST:
			snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
				DEFAULT_GLOBAL_USTCONSUMER32_HEALTH_UNIX_SOCK);
			break;
		default:
			ret = -EINVAL;
			goto end;
		}
	} else {
		home_path = utils_get_home_dir();
		if (home_path == NULL) {
			/* TODO: Add --socket PATH option */
			ERR("Can't get HOME directory for sockets creation.");
			ret = -EPERM;
			goto end;
		}

		/* Set health check Unix path */
		if (strlen(health_unix_sock_path) != 0) {
			goto end;
		}
		switch (type) {
		case LTTNG_CONSUMER_KERNEL:
			snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
				DEFAULT_HOME_KCONSUMER_HEALTH_UNIX_SOCK, home_path);
			break;
		case LTTNG_CONSUMER64_UST:
			snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
				DEFAULT_HOME_USTCONSUMER64_HEALTH_UNIX_SOCK, home_path);
			break;
		case LTTNG_CONSUMER32_UST:
			snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
				DEFAULT_HOME_USTCONSUMER32_HEALTH_UNIX_SOCK, home_path);
			break;
		default:
			ret = -EINVAL;
			goto end;
		}
	}
end:
	return ret;
}

/*
 * Thread managing health check socket.
 */
void *thread_manage_health(void *data)
{
	int sock = -1, new_sock = -1, ret, i, pollfd, err = -1;
	uint32_t revents, nb_fd;
	struct lttng_poll_event events;
	struct health_comm_msg msg;
	struct health_comm_reply reply;
	int is_root;

	DBG("[thread] Manage health check started");

	setup_health_path();

	rcu_register_thread();

	/* We might hit an error path before this is created. */
	lttng_poll_init(&events);

	/* Create unix socket */
	sock = lttcomm_create_unix_sock(health_unix_sock_path);
	if (sock < 0) {
		ERR("Unable to create health check Unix socket");
		ret = -1;
		goto error;
	}

	is_root = !getuid();
	if (is_root) {
		/* lttng health client socket path permissions */
		ret = chown(health_unix_sock_path, 0,
				utils_get_group_id(tracing_group_name));
		if (ret < 0) {
			ERR("Unable to set group on %s", health_unix_sock_path);
			PERROR("chown");
			ret = -1;
			goto error;
		}

		ret = chmod(health_unix_sock_path,
				S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
		if (ret < 0) {
			ERR("Unable to set permissions on %s", health_unix_sock_path);
			PERROR("chmod");
			ret = -1;
			goto error;
		}
	}

	/*
	 * Set the CLOEXEC flag. Return code is useless because either way, the
	 * show must go on.
	 */
	(void) utils_set_fd_cloexec(sock);

	ret = lttcomm_listen_unix_sock(sock);
	if (ret < 0) {
		goto error;
	}

	/* Size is set to 1 for the consumer_channel pipe */
	ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
	if (ret < 0) {
		ERR("Poll set creation failed");
		goto error;
	}

	ret = lttng_poll_add(&events, health_quit_pipe[0], LPOLLIN);
	if (ret < 0) {
		goto error;
	}

	/* Add the application registration socket */
	ret = lttng_poll_add(&events, sock, LPOLLIN | LPOLLPRI);
	if (ret < 0) {
		goto error;
	}

	/* Perform prior memory accesses before decrementing ready */
	cmm_smp_mb__before_uatomic_dec();
	uatomic_dec(&lttng_consumer_ready);

	while (1) {
		DBG("Health check ready");

		/* Inifinite blocking call, waiting for transmission */
restart:
		ret = lttng_poll_wait(&events, -1);
		if (ret < 0) {
			/*
			 * Restart interrupted system call.
			 */
			if (errno == EINTR) {
				goto restart;
			}
			goto error;
		}

		nb_fd = ret;

		for (i = 0; i < nb_fd; i++) {
			/* Fetch once the poll data */
			revents = LTTNG_POLL_GETEV(&events, i);
			pollfd = LTTNG_POLL_GETFD(&events, i);

			if (!revents) {
				/* No activity for this FD (poll implementation). */
				continue;
			}

			/* Thread quit pipe has been closed. Killing thread. */
			ret = check_health_quit_pipe(pollfd, revents);
			if (ret) {
				err = 0;
				goto exit;
			}

			/* Event on the registration socket */
			if (pollfd == sock) {
				if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
					ERR("Health socket poll error");
					goto error;
				}
			}
		}

		new_sock = lttcomm_accept_unix_sock(sock);
		if (new_sock < 0) {
			goto error;
		}

		/*
		 * Set the CLOEXEC flag. Return code is useless because either way, the
		 * show must go on.
		 */
		(void) utils_set_fd_cloexec(new_sock);

		DBG("Receiving data from client for health...");
		ret = lttcomm_recv_unix_sock(new_sock, (void *)&msg, sizeof(msg));
		if (ret <= 0) {
			DBG("Nothing recv() from client... continuing");
			ret = close(new_sock);
			if (ret) {
				PERROR("close");
			}
			new_sock = -1;
			continue;
		}

		rcu_thread_online();

		assert(msg.cmd == HEALTH_CMD_CHECK);

		memset(&reply, 0, sizeof(reply));
		for (i = 0; i < NR_HEALTH_CONSUMERD_TYPES; i++) {
			/*
			 * health_check_state return 0 if thread is in
			 * error.
			 */
			if (!health_check_state(health_consumerd, i)) {
				reply.ret_code |= 1ULL << i;
			}
		}

		DBG("Health check return value %" PRIx64, reply.ret_code);

		ret = send_unix_sock(new_sock, (void *) &reply, sizeof(reply));
		if (ret < 0) {
			ERR("Failed to send health data back to client");
		}

		/* End of transmission */
		ret = close(new_sock);
		if (ret) {
			PERROR("close");
		}
		new_sock = -1;
	}

exit:
error:
	if (err) {
		ERR("Health error occurred in %s", __func__);
	}
	DBG("Health check thread dying");
	unlink(health_unix_sock_path);
	if (sock >= 0) {
		ret = close(sock);
		if (ret) {
			PERROR("close");
		}
	}

	lttng_poll_clean(&events);

	rcu_unregister_thread();
	return NULL;
}
Commit	Line	Data
5c635c72 MD	1	/*
	2	* Copyright (C) 2013 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	3	*
	4	* This program is free software; you can redistribute it and/or modify
	5	* it under the terms of the GNU General Public License, version 2 only,
	6	* as published by the Free Software Foundation.
	7	*
	8	* This program is distributed in the hope that it will be useful,
	9	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	10	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	11	* GNU General Public License for more details.
	12	*
	13	* You should have received a copy of the GNU General Public License along
	14	* with this program; if not, write to the Free Software Foundation, Inc.,
	15	* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	16	*/
	17
	18	#define _GNU_SOURCE
6c1c0768	19	#define _LGPL_SOURCE
5c635c72 MD	20	#include <fcntl.h>
	21	#include <getopt.h>
	22	#include <grp.h>
	23	#include <limits.h>
	24	#include <pthread.h>
	25	#include <signal.h>
	26	#include <stdio.h>
	27	#include <stdlib.h>
	28	#include <string.h>
	29	#include <sys/ipc.h>
	30	#include <sys/resource.h>
	31	#include <sys/shm.h>
	32	#include <sys/socket.h>
	33	#include <sys/stat.h>
	34	#include <sys/types.h>
	35	#include <urcu/list.h>
	36	#include <poll.h>
	37	#include <unistd.h>
	38	#include <sys/mman.h>
	39	#include <assert.h>
	40	#include <config.h>
	41	#include <urcu/compiler.h>
	42	#include <ulimit.h>
6c71277b	43	#include <inttypes.h>
5c635c72 MD	44
	45	#include <common/defaults.h>
	46	#include <common/common.h>
	47	#include <common/consumer.h>
	48	#include <common/consumer-timer.h>
	49	#include <common/compat/poll.h>
	50	#include <common/sessiond-comm/sessiond-comm.h>
	51	#include <common/utils.h>
	52
	53	#include "lttng-consumerd.h"
	54	#include "health-consumerd.h"
	55
	56	/* Global health check unix path */
	57	static char health_unix_sock_path[PATH_MAX];
	58
	59	int health_quit_pipe[2];
	60
	61	/*
	62	* Check if the thread quit pipe was triggered.
	63	*
	64	* Return 1 if it was triggered else 0;
	65	*/
	66	static
	67	int check_health_quit_pipe(int fd, uint32_t events)
	68	{
	69	if (fd == health_quit_pipe[0] && (events & LPOLLIN)) {
	70	return 1;
	71	}
	72
	73	return 0;
	74	}
	75
	76	/*
	77	* Send data on a unix socket using the liblttsessiondcomm API.
	78	*
	79	* Return lttcomm error code.
	80	*/
	81	static int send_unix_sock(int sock, void *buf, size_t len)
	82	{
	83	/* Check valid length */
	84	if (len == 0) {
	85	return -1;
	86	}
	87
	88	return lttcomm_send_unix_sock(sock, buf, len);
	89	}
	90
	91	static
	92	int setup_health_path(void)
	93	{
	94	int is_root, ret = 0;
	95	enum lttng_consumer_type type;
	96	const char *home_path;
	97
	98	type = lttng_consumer_get_type();
	99	is_root = !getuid();
	100
	101	if (is_root) {
	102	if (strlen(health_unix_sock_path) != 0) {
	103	goto end;
	104	}
	105	switch (type) {
	106	case LTTNG_CONSUMER_KERNEL:
	107	snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
108	DEFAULT_GLOBAL_KCONSUMER_HEALTH_UNIX_SOCK);
109	break;
110	case LTTNG_CONSUMER64_UST:
111	snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
112	DEFAULT_GLOBAL_USTCONSUMER64_HEALTH_UNIX_SOCK);
113	break;
114	case LTTNG_CONSUMER32_UST:
115	snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
116	DEFAULT_GLOBAL_USTCONSUMER32_HEALTH_UNIX_SOCK);
117	break;
118	default:
119	ret = -EINVAL;
120	goto end;
121	}
122	} else {
5c635c72 MD	123	home_path = utils_get_home_dir();
	124	if (home_path == NULL) {
	125	/* TODO: Add --socket PATH option */
	126	ERR("Can't get HOME directory for sockets creation.");
	127	ret = -EPERM;
	128	goto end;
	129	}
	130
5c635c72 MD	131	/* Set health check Unix path */
	132	if (strlen(health_unix_sock_path) != 0) {
	133	goto end;
	134	}
	135	switch (type) {
	136	case LTTNG_CONSUMER_KERNEL:
	137	snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
dbc8403d	138	DEFAULT_HOME_KCONSUMER_HEALTH_UNIX_SOCK, home_path);
5c635c72 MD	139	break;
	140	case LTTNG_CONSUMER64_UST:
	141	snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
dbc8403d	142	DEFAULT_HOME_USTCONSUMER64_HEALTH_UNIX_SOCK, home_path);
5c635c72 MD	143	break;
	144	case LTTNG_CONSUMER32_UST:
	145	snprintf(health_unix_sock_path, sizeof(health_unix_sock_path),
dbc8403d	146	DEFAULT_HOME_USTCONSUMER32_HEALTH_UNIX_SOCK, home_path);
5c635c72 MD	147	break;
	148	default:
	149	ret = -EINVAL;
	150	goto end;
	151	}
	152	}
5c635c72 MD	153	end:
	154	return ret;
	155	}
	156
	157	/*
	158	* Thread managing health check socket.
	159	*/
	160	void thread_manage_health(void data)
	161	{
	162	int sock = -1, new_sock = -1, ret, i, pollfd, err = -1;
	163	uint32_t revents, nb_fd;
	164	struct lttng_poll_event events;
	165	struct health_comm_msg msg;
	166	struct health_comm_reply reply;
6c71277b	167	int is_root;
5c635c72 MD	168
	169	DBG("[thread] Manage health check started");
	170
	171	setup_health_path();
	172
	173	rcu_register_thread();
	174
	175	/* We might hit an error path before this is created. */
	176	lttng_poll_init(&events);
	177
	178	/* Create unix socket */
	179	sock = lttcomm_create_unix_sock(health_unix_sock_path);
	180	if (sock < 0) {
	181	ERR("Unable to create health check Unix socket");
	182	ret = -1;
	183	goto error;
	184	}
	185
6c71277b MD	186	is_root = !getuid();
	187	if (is_root) {
	188	/* lttng health client socket path permissions */
	189	ret = chown(health_unix_sock_path, 0,
	190	utils_get_group_id(tracing_group_name));
	191	if (ret < 0) {
	192	ERR("Unable to set group on %s", health_unix_sock_path);
	193	PERROR("chown");
	194	ret = -1;
	195	goto error;
	196	}
	197
	198	ret = chmod(health_unix_sock_path,
	199	S_IRUSR \| S_IWUSR \| S_IRGRP \| S_IWGRP);
	200	if (ret < 0) {
	201	ERR("Unable to set permissions on %s", health_unix_sock_path);
	202	PERROR("chmod");
	203	ret = -1;
	204	goto error;
	205	}
	206	}
	207
5c635c72 MD	208	/*
	209	* Set the CLOEXEC flag. Return code is useless because either way, the
	210	* show must go on.
	211	*/
	212	(void) utils_set_fd_cloexec(sock);
	213
	214	ret = lttcomm_listen_unix_sock(sock);
	215	if (ret < 0) {
	216	goto error;
	217	}
	218
	219	/* Size is set to 1 for the consumer_channel pipe */
	220	ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
	221	if (ret < 0) {
	222	ERR("Poll set creation failed");
	223	goto error;
	224	}
	225
	226	ret = lttng_poll_add(&events, health_quit_pipe[0], LPOLLIN);
	227	if (ret < 0) {
	228	goto error;
	229	}
	230
	231	/* Add the application registration socket */
	232	ret = lttng_poll_add(&events, sock, LPOLLIN \| LPOLLPRI);
	233	if (ret < 0) {
	234	goto error;
	235	}
	236
748b7b07 MD	237	/* Perform prior memory accesses before decrementing ready */
	238	cmm_smp_mb__before_uatomic_dec();
	239	uatomic_dec(&lttng_consumer_ready);
	240
5c635c72 MD	241	while (1) {
	242	DBG("Health check ready");
	243
	244	/* Inifinite blocking call, waiting for transmission */
	245	restart:
	246	ret = lttng_poll_wait(&events, -1);
	247	if (ret < 0) {
	248	/*
	249	* Restart interrupted system call.
	250	*/
	251	if (errno == EINTR) {
	252	goto restart;
	253	}
	254	goto error;
	255	}
	256
	257	nb_fd = ret;
	258
	259	for (i = 0; i < nb_fd; i++) {
	260	/* Fetch once the poll data */
	261	revents = LTTNG_POLL_GETEV(&events, i);
	262	pollfd = LTTNG_POLL_GETFD(&events, i);
	263
fd20dac9 MD	264	if (!revents) {
	265	/* No activity for this FD (poll implementation). */
	266	continue;
	267	}
	268
5c635c72 MD	269	/* Thread quit pipe has been closed. Killing thread. */
	270	ret = check_health_quit_pipe(pollfd, revents);
	271	if (ret) {
	272	err = 0;
	273	goto exit;
	274	}
	275
	276	/* Event on the registration socket */
	277	if (pollfd == sock) {
	278	if (revents & (LPOLLERR \| LPOLLHUP \| LPOLLRDHUP)) {
	279	ERR("Health socket poll error");
	280	goto error;
	281	}
	282	}
	283	}
	284
	285	new_sock = lttcomm_accept_unix_sock(sock);
	286	if (new_sock < 0) {
	287	goto error;
	288	}
	289
	290	/*
	291	* Set the CLOEXEC flag. Return code is useless because either way, the
	292	* show must go on.
	293	*/
	294	(void) utils_set_fd_cloexec(new_sock);
	295
	296	DBG("Receiving data from client for health...");
	297	ret = lttcomm_recv_unix_sock(new_sock, (void *)&msg, sizeof(msg));
	298	if (ret <= 0) {
	299	DBG("Nothing recv() from client... continuing");
	300	ret = close(new_sock);
	301	if (ret) {
	302	PERROR("close");
	303	}
	304	new_sock = -1;
	305	continue;
	306	}
	307
	308	rcu_thread_online();
	309
	310	assert(msg.cmd == HEALTH_CMD_CHECK);
	311
53efb85a	312	memset(&reply, 0, sizeof(reply));
6c71277b MD	313	for (i = 0; i < NR_HEALTH_CONSUMERD_TYPES; i++) {
	314	/*
	315	* health_check_state return 0 if thread is in
	316	* error.
	317	*/
	318	if (!health_check_state(health_consumerd, i)) {
	319	reply.ret_code \|= 1ULL << i;
	320	}
5c635c72 MD	321	}
5c635c72 MD	322
6137f630	323	DBG("Health check return value %" PRIx64, reply.ret_code);
5c635c72 MD	324
	325	ret = send_unix_sock(new_sock, (void *) &reply, sizeof(reply));
	326	if (ret < 0) {
	327	ERR("Failed to send health data back to client");
	328	}
	329
	330	/* End of transmission */
	331	ret = close(new_sock);
	332	if (ret) {
	333	PERROR("close");
	334	}
	335	new_sock = -1;
	336	}
	337
	338	exit:
	339	error:
	340	if (err) {
	341	ERR("Health error occurred in %s", __func__);
	342	}
	343	DBG("Health check thread dying");
	344	unlink(health_unix_sock_path);
	345	if (sock >= 0) {
	346	ret = close(sock);
	347	if (ret) {
	348	PERROR("close");
	349	}
	350	}
	351
	352	lttng_poll_clean(&events);
	353
	354	rcu_unregister_thread();
	355	return NULL;
	356	}