PyFCGI/src/monitor.c

#include "monitor.h"

static pyfcgi_monitor_t pyfcgi_mon;

static void clean_exit(int status)
{
	if(pyfcgi_mon.sockserv)
	{
		if(pyfcgi_mon.sockcli)
		{
			if(shutdown(pyfcgi_mon.sockcli, SHUT_RDWR) < 0)
			{
				pyfcgi_log(LOG_WARNING, "Unable to shutdown clisocket : %s",
					strerror(errno));
			}
			if(close(pyfcgi_mon.sockcli) < 0)
			{
				pyfcgi_log(LOG_WARNING, "Unable to close clisocket : %s",
					strerror(errno));
			}
		}
		pyfcgi_log(LOG_INFO, "Closing listening socket...");
		if(shutdown(pyfcgi_mon.sockserv, SHUT_RDWR) < 0)
		{
			pyfcgi_log(LOG_WARNING, "Unable to shutdown socket : %s",
				strerror(errno));
		}
		if(close(pyfcgi_mon.sockserv) < 0)
		{
			pyfcgi_log(LOG_WARNING, "Unable to close socket : %s",
				strerror(errno));
		}
	}
	exit(status);
}

pid_t pyfcgi_spawn_monitor()
{
	pid_t res;
	struct sigaction act, actdrop;
	int err;

	act.sa_handler = pyfcgi_monitor_sighandler;
	sigemptyset(&act.sa_mask);
	act.sa_flags = 0;
	act.sa_restorer = NULL;
	
	actdrop.sa_handler = pyfcgi_sighandler_drop;
	sigemptyset(&act.sa_mask);
	act.sa_flags = 0;
	act.sa_restorer = NULL;

	if(!PyFCGI_conf.mon_socket)
	{
		pyfcgi_log(LOG_ERR, "No socket url set but pyfcgi_spawn_monitor called !");
		return -1;
	}

	memset(&pyfcgi_mon, 0, sizeof(pyfcgi_mon));

	res = fork();
	if(res == -1)
	{
		pyfcgi_log(LOG_ALERT, "Unable to fork into monitoring server process : %s",
			strerror(errno));
		sleep(1);
	}
	if(!res)
	{ //child process
		pyfcgi_logger_set_ident("StatServ");
		if(sigaction(SIGTERM, &act, NULL))
		{
			err = errno;
			pyfcgi_log(LOG_ERR, "Unable to run sigaction for SIGTERM : %s",
				strerror(err));
			exit(PYFCGI_FATAL);
		}
		if(sigaction(SIGINT, &actdrop, NULL))
		{
			pyfcgi_log(LOG_WARNING, "Unable run sigaction for SIGINT sigaction : %s",
				strerror(errno));
		}
		if(sigaction(SIGALRM, &actdrop, NULL))
		{
			pyfcgi_log(LOG_WARNING, "Unable to restore ALARM sigaction : %s",
				strerror(errno));
		}
		pyfcgi_monitor_loop();
		pyfcgi_log(LOG_ALERT, "Monitor loop should never return but just did it...");
		clean_exit(PYFCGI_FATAL);
	}
	return res;
}

int pyfcgi_monitor_IPC_init()
{
	int ret;
	ret = 0;
	if(!PyFCGI_SEM_OPEN(SEM_WREQS))
	{
		if(pyfcgi_IPC_init(IPC_WREQS)) { ret = -1; }
	}
	if(!PyFCGI_SEM_OPEN(SEM_STATS))
	{
		if(pyfcgi_IPC_init(IPC_SEMST)) { ret = -1; }
	}
	if(!PyFCGI_conf.shm.ptr)
	{
		if(pyfcgi_IPC_init(IPC_SHMST)) { ret = -1; }
	}
	return ret;
}

void pyfcgi_monitor_loop()
{
	pyfcgi_monitor_addr_t addr;
	int *sockargs, reuse;
	socklen_t addrlen;

	sockargs = pyfcgi_mon.sockargs;

	if(pyfcgi_monitor_parse_sock(PyFCGI_conf.mon_socket, sockargs,
		&addr) < 0)
	{
		pyfcgi_log(LOG_WARNING,
			"Unable to parse socket URL, monitoring server not starting...");
		free(PyFCGI_conf.mon_socket);
		PyFCGI_conf.mon_socket = NULL;
		clean_exit(PYFCGI_ERR);
	}

	if( (pyfcgi_mon.sockserv = socket(sockargs[0], sockargs[1],
		sockargs[2])) < 0)
	{
		pyfcgi_log(LOG_ERR, "Unable to create socket : %s",
			strerror(errno));
		sleep(1);
		clean_exit(PYFCGI_ERR);
	}

	if(sockargs[0] == AF_INET)
	{
		addrlen = sizeof(struct sockaddr_in);
	}
	else if(sockargs[0] == AF_INET6)
	{
		addrlen = sizeof(struct sockaddr_in6);
	}
	else
	{
		addrlen = sizeof(struct sockaddr_un);
	}
	reuse = 1;
	if(setsockopt(pyfcgi_mon.sockserv, SOL_SOCKET, SO_REUSEADDR, &reuse,
		sizeof(int)) < 0)
	{
		pyfcgi_log(LOG_WARNING, "Unable to set socket option : %s",
			strerror(errno));
	}
	if(bind(pyfcgi_mon.sockserv, (struct sockaddr*)&addr, addrlen) < 0)
	{
		pyfcgi_log(LOG_ERR, "Unable to bind socket... : %s",
			strerror(errno));
		pyfcgi_log(LOG_INFO, "Retrying in 10s");
		close(pyfcgi_mon.sockserv);
		sleep(10);
		clean_exit(PYFCGI_ERR);
	}
	pyfcgi_log(LOG_INFO, "Listening on %s", PyFCGI_conf.mon_socket);

	//bind successfull, sending to appropriate response loop
	if(pyfcgi_monitor_IPC_init() < 0)
	{
		pyfcgi_log(LOG_ERR, "Unable to initialize IPC");
		sleep(1);
		exit(PYFCGI_ERR);
	}
	pyfcgi_stats_init();

	if(pyfcgi_mon.sockargs[1] == SOCK_DGRAM)
	{
		pyfcgi_monitor_dgram_loop();
	}
	else
	{
		pyfcgi_monitor_stream_loop();
	}
}

void pyfcgi_monitor_stream_loop(pyfcgi_monitor_addr_t addr_serv)
{
	pyfcgi_monitor_addr_t cliaddr;
	socklen_t addrlen;
	int sockcli, err, ret;
	char ipstr[64];
	const char *buff;
	size_t bufflen;

	addrlen = sizeof(cliaddr);
	if(listen(pyfcgi_mon.sockserv, PYFCGI_MONITOR_STREAM_BACKLOG) < 0)
	{
		pyfcgi_log(LOG_ERR, "Unable to listen on socket : %s",
			strerror(errno));
		clean_exit(PYFCGI_ERR);
	}
	
	while(1)
	{
		sockcli = pyfcgi_mon.sockcli = accept(pyfcgi_mon.sockserv,
			(struct sockaddr*)&cliaddr, &addrlen);
		if(sockcli < 0)
		{
			err = errno;
			if(err == EINTR) { continue; }
			break;
		}
		pyfcgi_log(LOG_DEBUG, "New client");
		if(!inet_ntop(pyfcgi_mon.sockargs[0], (void*)&cliaddr, ipstr, 64))
		{
			pyfcgi_log(LOG_WARNING,
				"inet_ntop fails to represent client address : %s",
				strerror(errno));
			strcpy(ipstr, "XXX");
		}
		if(!pyfcgi_stats_format())
		{
			pyfcgi_log(LOG_ERR, "Unable to format stats...");
			close(sockcli);
			continue;
		}
		pyfcgi_stats_buff(&buff, &bufflen);
		pyfcgi_log(LOG_INFO, "Sending stats to %s", ipstr);
		
		do
		{
			ret = send(sockcli, buff, bufflen, 0);
			err = errno;
		}while(ret == -1 && err == EINTR);
		if(ret == -1)
		{
			pyfcgi_log(LOG_WARNING, "Unable to send stats to client : %s",
				strerror(err));
		}
		shutdown(sockcli, SHUT_RDWR);
		close(sockcli);
		pyfcgi_mon.sockcli = 0;
		
		addrlen = sizeof(cliaddr);
	}
	pyfcgi_log(LOG_ERR, "Unable to accept new connection on socket : %s",
		strerror(errno));
}

void pyfcgi_monitor_dgram_loop()
{
	pyfcgi_log(LOG_ERR, "Dgram server not implemented.... exiting server");
	free(PyFCGI_conf.mon_socket);
	PyFCGI_conf.mon_socket = NULL;
	clean_exit(PYFCGI_ERR);
}

int pyfcgi_monitor_check_sock(const char* sockurl)
{
	const char *port;
	short tcp;
	const char *urlorig;

	urlorig = sockurl;

	if(!(tcp = strncasecmp("tcp://", sockurl, 6)) ||
		!strncasecmp("udp://", sockurl, 6))
	{
		sockurl += 6; //first addr chr
	}
	else if(!strncasecmp("unix://", sockurl, 7))
	{
		sockurl += 7;
		if(!strlen(sockurl))
		{
			dprintf(2, "UNIX socket missing file path : '%s'\n",
				urlorig);
			return -1;
		}
		if(strlen(sockurl) > UNIX_SOCKPATH_MAX)
		{
			dprintf(2, "UNIX socket support only path with length <= %d but given path is %ld bytes long : '%s'",
				UNIX_SOCKPATH_MAX, strlen(sockurl), urlorig);
			return -1;
		}
		return 0;
	}
	else
	{
		dprintf(2, "Invalid protocol in '%s'\n", sockurl);
		return -1;
	}
	do { sockurl++; } while(*sockurl && *sockurl != ':');
	if(!sockurl)
	{
		dprintf(2, "%s protocol choosen but not port given : '%s'\n",
			tcp?"TCP":"UDP", sockurl);
		return -1;
	}
	if(!*sockurl)
	{
		dprintf(2, "Port missing in socket URL '%s'\n", urlorig);
		return -1;
	}
	sockurl++;
	port = sockurl;
	while(*sockurl && *sockurl >= '0' && *sockurl <= '9')
	{
		sockurl++;
	}
	if(*sockurl)
	{
		dprintf(2, "Invalid port '%s' in socket URL '%s'\n",
			port, urlorig);
		return -1;
	}
	return 0;
}

int pyfcgi_monitor_parse_sock(const char *sockurl, int sockargs[3],
	pyfcgi_monitor_addr_t *listen_addr)
{
	const char *addr_ptr;
	short tcp;
	int *domain, *type, *protocol;
	struct sockaddr_un *addr_un;

	domain = &sockargs[0];
	type = &sockargs[1];
	protocol = &sockargs[2];

	addr_un = &(listen_addr->un);

	if(!strncasecmp("unix://", sockurl, 7))
	{
		addr_ptr = sockurl + 7;
		*domain = AF_UNIX;
		*type = SOCK_STREAM;
		*protocol = 0;
		addr_un->sun_family = AF_UNIX;
		strncpy(addr_un->sun_path, addr_ptr, UNIX_SOCKPATH_MAX);
		return 0;
	}

	if((tcp = strncasecmp("tcp://", sockurl, 6)) &&
		strncasecmp("udp://", sockurl, 6))
	{ //Unchecked URL??!!
		pyfcgi_log(LOG_ERR, "Invalid protocol in URL : '%s'",
			sockurl);
		return -1;
	}
	addr_ptr = sockurl + 6;
	*type = tcp ? SOCK_DGRAM : SOCK_STREAM;
	*protocol = 0;

	if(pyfcgi_monitor_parse_inet_addr(addr_ptr, *type, listen_addr, domain))
	{
		return -1;
	}
	return 0;
}

int pyfcgi_monitor_parse_inet_addr(const char* addr_str, int socktype,
	pyfcgi_monitor_addr_t *listen_addr, int* domain)
{
	char *addr, *port, *ptr, ipstr[64];
	struct addrinfo *infos, hints, *info;
	short v4, v6;
	int ret;

	// initialize temporary address & port pointers
	addr = strdup(addr_str);
	if(!addr)
	{
		pyfcgi_log(LOG_ALERT, "strdup() failed to copy socket addr : %s",
			strerror(errno));
		return -1;
	}
	ptr = addr;
	do { ptr++; }while(*ptr && *ptr != ':');
	if(!ptr)
	{
		pyfcgi_log(LOG_ERR, "No port found in INET url : %s",
			addr_str);
		goto free_err;
	}
	*ptr = '\0';
	port = ptr+1;

	memset(&hints, 0, sizeof(struct addrinfo));
	v4 = PyFCGI_conf.ipv4;
	v6 = PyFCGI_conf.ipv6;
	hints.ai_family = (v4?AF_INET:(v6?AF_INET6:PF_UNSPEC));
	hints.ai_socktype = socktype;
	hints.ai_flags = AI_CANONNAME;

	if((ret = getaddrinfo(addr, port, &hints, &infos)))
	{
		pyfcgi_log(LOG_ALERT, "getaddrinfo fails on '%s' : %s",
			gai_strerror(ret));
		goto free_err;
	}

	for(info = infos; info != NULL; info = info->ai_next)
	{
		if(info->ai_family == AF_INET)
		{
			memcpy(&listen_addr->in, info->ai_addr,
				info->ai_addrlen);
		}
		else if(info->ai_family == AF_INET6)
		{
			memcpy(&listen_addr->in6, info->ai_addr,
				info->ai_addrlen);
		}
		else
		{
			continue;
		}
		*domain = info->ai_family;
		if(!inet_ntop(*domain, info->ai_addr, ipstr, 64))
		{
			pyfcgi_log(LOG_ERR, "Unable to format IP in string : %s",
				strerror(errno));
			strcpy(ipstr, "IP");
		}
		pyfcgi_log(LOG_DEBUG, "Listen addr resolved to %s(%s)",
			info->ai_canonname,  ipstr);

		freeaddrinfo(infos);
		free(addr);
		return 0;
	}

	pyfcgi_log(LOG_ERR, "Unable to resolve to a valid AF_INET[6] address");
	freeaddrinfo(infos);

free_err:
	free(addr);
	return -1;
}

void pyfcgi_monitor_sighandler(int signum)
{
	pyfcgi_log(LOG_NOTICE, "Received signal %s, exiting",
		strsignal(signum));
	clean_exit(0);
}