File Coverage

blib/lib/App/Sv.pm

Criterion	Covered	Total	%
statement	57	278	20.5
branch	28	152	18.4
condition	5	51	9.8
subroutine	10	54	18.5
pod	2	2	100.0
total	102	537	18.9

package App::Sv;
# ABSTRACT: Event-based multi-process supervisor
our $VERSION = '0.014';

use 5.008001;
use strict;
use warnings;

use Carp 'croak';
use POSIX;
use AnyEvent;
use AnyEvent::Socket;
use AnyEvent::Handle;
use App::Sv::Log;

# Constructor
sub new {
	my $class = shift;
	my $conf;

	if (@_ && scalar @_ > 1) {
		croak "Odd number of arguments to App::Sv" if @_ % 2 != 0;
		foreach my $id (0..$#_) {
			$conf->{$_[$id]} = $_[$id+1] if $id % 2 == 0;
		}
	}
	else {
		$conf = shift @_;
	}

	my $run = $conf->{run};
	$conf->{global}->{log} = delete $conf->{log};
	croak "Commands must be passed as a hash ref" if ref $run ne 'HASH';
	croak "Missing command list" if !scalar (keys %$run);

	# set defaults
	my $defaults = {
		start_retries => 8,
		restart_delay => 1,
		start_wait => 1,
		stop_wait => 0,
		setsid => 1
	};
	# check options
	foreach my $svc (keys %$run) {
		if (!$run->{$svc}) {
			croak "Missing command for \'$svc\'";
		}
		elsif (!ref $run->{$svc}) {
			$run->{$svc} = { cmd => $run->{$svc} };
		}
		elsif (ref $run->{$svc} eq 'CODE') {
			$run->{$svc} = { code => $run->{$svc} };
		}
		elsif (ref $run->{$svc} eq 'ARRAY') {
			if (!$run->{$svc}->[0]) {
				croak "Missing command for \'$svc\'";
			}
			elsif (!ref $run->{$svc}->[0]) {
				$run->{$svc} = { cmd => $run->{$svc} };
			}
			elsif (ref $run->{$svc}->[0] eq 'CODE') {
				$run->{$svc} = { code => $run->{$svc} };
			}
		}

		if (ref $run->{$svc} eq 'HASH') {
			if (!$run->{$svc}->{cmd} && !$run->{$svc}->{code}) {
				croak "Missing command for \'$svc\'"
			}
			$run->{$svc}->{name} = $svc;
			foreach my $opt (keys %$defaults) {
				if (!defined $run->{$svc}->{$opt}) {
					$run->{$svc}->{$opt} = $defaults->{$opt};
				}
				elsif ($opt =~ /delay|wait/ && $run->{$svc}->{$opt} <= 0) {
					$run->{$svc}->{$opt} = $defaults->{$opt};
				}
			}
		}
		else {
			croak "Missing command for \'$svc\'";
		}
	}

	return bless { run => $run, conf => $conf->{global} }, $class;
}

# Start everything
sub run {
	my $self = shift;
	my $cv = AE::cv;

	# signal watchers
	my $int_s = AE::signal 'INT' => sub {
		$self->_signal_all_svc('INT', $cv);
	};
	my $hup_s = AE::signal 'HUP' => sub {
		$self->_signal_all_svc('HUP', $cv);
	};
	my $term_s = AE::signal 'TERM' => sub {
		$self->_signal_all_svc('TERM');
		$cv->send
	};
	# set global umask
	umask oct($self->{conf}->{umask}) if $self->{conf}->{umask};
	# initialize logger
	$self->{log} = App::Sv::Log->new($self->{conf}->{log});
	# open controling socket; load commands
	$self->_listener() if $self->{conf}->{listen};
	$self->{cmds} = $self->_client_cmds() if ref $self->{server} eq 'Guard';

	# start all services
	foreach my $key (keys %{ $self->{run} }) {
		my $svc = $self->{run}->{$key};
		$self->_start_svc($svc);
	}

	$cv->recv;
}

sub _start_svc {
	my ($self, $svc) = @_;

	my $debug = $self->{log}->logger(8);
	my $warn = $self->{log}->logger(5);
	$svc->{state} = 'start';
	if ($svc->{start_count}) {
		$svc->{start_count}++;
	}
	else {
		$svc->{start_count} = 1;
	}

	$debug->("Starting '$svc->{name}' attempt $svc->{start_count}");
	my $pid = fork();
	if (!defined $pid) {
		$warn->("Failed to fork '$svc->{name}': $!");
		$self->_restart_svc($svc);
		return;
	}

	if ($pid == 0) {
		# child
		# set egid/euid
		if ($svc->{group}) {
			$svc->{gid} = getgrnam($svc->{group});
			$) = $svc->{gid};
		}
		if ($svc->{user}) {
			$svc->{uid} = getpwnam($svc->{user});
			$> = $svc->{uid};
		}
		# set process umask
		umask oct($svc->{umask}) if $svc->{umask};
		# change working directory
		if ($svc->{cwd}) {
			chdir $svc->{cwd}
				or $warn->("Failed cwd for '$svc->{name}': $!");
		}
		# set environment
		%ENV = %{$svc->{env}} if $svc->{env} && ref $svc->{env} eq 'HASH';
		# set session id
		if ($svc->{setsid}) {
			$svc->{pgrp} = POSIX::setsid()
				or $warn->("Failed setsid for '$svc->{name}': $!");
		};
		# start process
		if ($svc->{cmd} && !ref $svc->{cmd}) {
			$debug->("Executing command '$svc->{name}'");
			exec($svc->{cmd});
		}
		elsif ($svc->{cmd} && ref $svc->{cmd} eq 'ARRAY') {
			$debug->("Executing command '$svc->{name}'");
			exec(@{$svc->{cmd}});
		}
		elsif ($svc->{code} && ref $svc->{code} eq 'CODE') {
			$debug->("Executing code '$svc->{name}'");
			$svc->{code}->();
		}
		elsif ($svc->{code} && ref $svc->{code} eq 'ARRAY') {
			my $code = shift @{$svc->{code}};
			if (ref $code eq 'CODE') {
				$debug->("Executing code '$svc->{name}'");
				$code->(@{$svc->{code}});
			}
		}
		POSIX::_exit(1);
	}
	else {
		# parent
		$debug->("Watching pid $pid for '$svc->{name}'");
		$svc->{pid} = $pid;
		$svc->{watcher} = AE::child $pid, sub {
			$self->_child_exited($svc, @_);
		};
		$svc->{start_ts} = time;
		my $t; $t = AE::timer $svc->{start_wait}, 0, sub {
			$self->_check_svc_up($svc);
			undef $t;
		};
	}

	return $pid;
}

sub _child_exited {
	my ($self, $svc, undef, $status) = @_;

	my $debug = $self->{log}->logger(8);
	$debug->("Child $svc->{pid} exited, status $status: '$svc->{name}'");
	delete $svc->{watcher};
	delete $svc->{pid};
	$svc->{last_status} = $status >> 8;
	if ($svc->{state} eq 'stop') {
		delete $svc->{start_count};
		$svc->{state} = 'down';
	}
	elsif ($svc->{once}) {
		delete $svc->{start_count};
		$svc->{state} = 'fatal';
	}
	else {
		$self->_restart_svc($svc);
	}
}

sub _restart_svc {
	my ($self, $svc) = @_;

	if ($svc->{start_retries}) {
		if ($svc->{start_count} &&
			($svc->{start_count} >= $svc->{start_retries})) {
			$svc->{state} = 'fatal';
			return;
		}
	}
	else {
		$svc->{state} = 'fatal';
		return;
	}
	my $debug = $self->{log}->logger(8);
	$svc->{state} = 'restart';
	$debug->("Restarting '$svc->{name}' in $svc->{restart_delay} seconds");
	my $t; $t = AE::timer $svc->{restart_delay}, 0, sub {
		$self->_start_svc($svc);
		undef $t;
	};
}

sub _check_svc_up {
	my ($self, $svc) = @_;

	return unless $svc->{state} eq 'start';
	if (!$svc->{pid}) {
		$svc->{state} = 'fail';
		return;
	}
	delete $svc->{start_count};
	$svc->{state} = 'up';
}

sub _stop_svc {
	my ($self, $svc) = @_;

	$svc->{state} = 'stop';
	my $st = $self->_signal_svc($svc, 'TERM');
	if ($svc->{stop_wait} && $svc->{stop_wait} > 0) {
		my $t; $t = AE::timer $svc->{stop_wait}, 0, sub {
			$self->_check_svc_down($svc);
			undef $t;
		};
	}

	return $st;
}

sub _check_svc_down {
	my ($self, $svc) = @_;

	return unless $svc->{state} eq 'stop';
	if ($svc->{pid}) {
		my $st = $self->_signal_svc($svc, 'KILL');
	}
}

sub _signal_svc {
	my ($self, $svc, $sig) = @_;

	return unless ($svc->{pid} && $sig);
	my $debug = $self->{log}->logger(8);
	$debug->("Sent signal $sig to pid $svc->{pid}");
	my $st = kill($sig, $svc->{pid});

	return $st;
}

sub _signal_all_svc {
	my ($self, $sig, $cv) = @_;

	my $debug = $self->{log}->logger(8);
	$debug->("Received signal $sig");
	my $is_any_alive = 0;
	foreach my $key (keys %{ $self->{run} }) {
		my $svc = $self->{run}->{$key};
		next unless my $pid = $svc->{pid};
		$debug->("... sent signal $sig to pid $pid");
		$is_any_alive++;
		kill($sig, $pid);
	}

	return if $cv and $is_any_alive;

	$debug->('Exiting...');
	$cv->send if $cv;
}

# Contolling socket
sub _listener {
	my $self = shift;

	my $debug = $self->{log}->logger(8);
	my ($host, $port) = parse_hostport($self->{conf}->{listen});
	croak "Socket \'$port\' already in use" if ($host eq 'unix/' && -e $port);

	$self->{server} = tcp_server $host, $port,
	sub { $self->_client_conn(@_) },
	sub {
		my ($fh, $host, $port) = @_;
		$debug->("Listening at $host:$port");
	};
}

sub _client_conn {
	my ($self, $fh, $host, $port) = @_;

	return unless $fh;
	my $debug = $self->{log}->logger(8);
	$debug->("New connection to $host:$port");

	my $hdl; $hdl = AnyEvent::Handle->new(
		fh => $fh,
		timeout => 30,
		rbuf_max => 64,
		wbuf_max => 64,
		on_read => sub { $self->_client_input($hdl) },
		on_eof => sub { $self->_client_disconn($hdl) },
		on_timeout => sub { $self->_client_error($hdl, undef, 'Timeout') },
		on_error => sub { $self->_client_error($hdl, undef, $!) }
	);
	$self->{conn}->{fileno($fh)} = $hdl;

	return $fh;
}

sub _client_input {
	my ($self, $hdl) = @_;

	$hdl->push_read(line => sub {
		my ($hdl, $ln) = @_;

		my $client = $self->{conn}->{fileno($hdl->fh)};
		my $cmds = $self->{cmds};
		if ($ln) {
			# generic commands
			$hdl->push_write("\n");
			if ($ln =~ /^(\.|quit)$/) {
				$self->_client_disconn($hdl);
			}
			elsif ($ln eq 'status') {
				$self->_status($hdl);
			}
			elsif (index($ln, ' ') >= 0) {
				my ($sw, $svc) = split(' ', $ln);
				if ($sw && $svc) {
					my $st;
					if ($self->{run}->{$svc} && ref $cmds->{$sw} eq 'CODE') {
						$svc = $self->{run}->{$svc};
						$st = $cmds->{$sw}->($svc);
					}
					else {
						$hdl->push_write("$ln unknown\n");
						return;
					}
					# response
					$st = ref $st eq 'ARRAY' ? join(' ', @$st) : $st;
					$st = $st ? $st : 'fail';
					$hdl->push_write("$ln $st\n") if $st;
				}
			}
			else {
				$hdl->push_write("$ln unknown\n");
			}
		}
	});
}

sub _client_disconn {
	my ($self, $hdl) = @_;

	my $debug = $self->{log}->logger(8);
	delete $self->{conn}->{fileno($hdl->fh)};
	$hdl->destroy();
	$debug->("Connection closed");
}

sub _client_error {
	my ($self, $hdl, $fatal, $msg) = @_;

	my $debug = $self->{log}->logger(8);
	delete $self->{conn}->{fileno($hdl->fh)};
	$debug->("Connection error: $msg");
	$hdl->destroy();
}

sub _client_cmds {
	my $self = shift;

	my $cmds = {
		up => sub {
			unless ($_[0]->{pid}) {
				delete $_[0]->{once};
				return $self->_start_svc($_[0]);
			}
		},
		once => sub {
			unless ($_[0]->{pid}) {
				$_[0]->{once} = 1;
				return $self->_start_svc($_[0]);
			}
		},
		down => sub {
			return $self->_stop_svc($_[0]) if $_[0]->{pid};
		},
		pause => sub {
			return $self->_signal_svc($_[0], 'STOP') if $_[0]->{pid};
		},
		cont => sub {
			return $self->_signal_svc($_[0], 'CONT') if $_[0]->{pid};
		},
		hup => sub {
			return $self->_signal_svc($_[0], 'HUP') if $_[0]->{pid};
		},
		alarm => sub {
			return $self->_signal_svc($_[0], 'ALRM') if $_[0]->{pid};
		},
		int => sub {
			return $self->_signal_svc($_[0], 'INT') if $_[0]->{pid};
		},
		quit => sub {
			return $self->_signal_svc($_[0], 'QUIT') if $_[0]->{pid};
		},
		usr1 => sub {
			return $self->_signal_svc($_[0], 'USR1') if $_[0]->{pid};
		},
		usr2 => sub {
			return $self->_signal_svc($_[0], 'USR2') if $_[0]->{pid};
		},
		term => sub {
			return $self->_signal_svc($_[0], 'TERM') if $_[0]->{pid};
		},
		kill => sub {
			return $self->_signal_svc($_[0], 'KILL') if $_[0]->{pid};
		},
		status => sub {
			if ($_[0]->{pid} && $_[0]->{start_ts}) {
				return([
					$_[0]->{state},
					$_[0]->{pid},
					time - $_[0]->{start_ts}
				]);
			}
			elsif ($_[0]->{start_count}) {
				return([
					$_[0]->{state},
					$_[0]->{start_count}
				]);
			}
			else {
				return $_[0]->{state};
			}
		}
	};

	return $cmds;
}

# Commands status
sub _status {
	my ($self, $hdl) = @_;

	return unless ($hdl && ref $self->{cmds}->{status} eq 'CODE');
	foreach my $key (keys %{ $self->{run} }) {
		my $st = $self->{cmds}->{status}->($self->{run}->{$key});
		$st = ref $st eq 'ARRAY' ? join(' ', @$st) : $st;
		$hdl->push_write("$key $st\n");
	}
	$hdl->push_write("\n");
}

1;

__END__