view release on metacpan or  search on metacpan

lib/App/RemoteCommand.pm  view on Meta::CPAN

package App::RemoteCommand v1.0.0;
use v5.24;
use warnings;
use experimental qw(lexical_subs signatures);

use App::RemoteCommand::Pool;
use App::RemoteCommand::SSH;
use App::RemoteCommand::Select;
use App::RemoteCommand::Util qw(prompt DEBUG logger);

use File::Basename ();
use File::Copy ();
use File::Temp ();
use Getopt::Long ();
use IO::Select;
use List::Util ();
use POSIX 'strftime';
use Pod::Usage ();
use String::Glob::Permute 'string_glob_permute';

use constant TICK_SECOND => 0.1;

my $SCRIPT = File::Basename::basename($0);
my $SUDO_PROMPT = sprintf "sudo password (asking with %s): ", $SCRIPT;
my $SUDO_FAIL = "Sorry, try again.";

our $TRIAL = 0;

sub new ($class, %option) {
    bless {
        %option,
        pending => [],
        running => App::RemoteCommand::Pool->new,
        select => App::RemoteCommand::Select->new,
    }, $class;
}

sub run ($self, @argv) {
    $self = $self->new unless ref $self;
    $self->parse_options(@argv);
    $self->register;

    local $| = 1;
    my $INT; local $SIG{INT} = sub (@) { $INT++ };
    my $TERM; local $SIG{TERM} = sub (@) { $TERM++ };
    while (1) {
        if ($INT || $TERM) {
            my $signal = $TERM ? "TERM" : "INT";
            warn "\nCatch SIG$signal, try to shutdown gracefully...\n";
            DEBUG and logger "handling signal %s", $signal;
            $self->cancel($signal);
            $INT = $TERM = 0;
        }
        $self->one_tick;
        last if $self->{pending}->@* == 0 && $self->{running}->count == 0;
    }

    my @success = sort grep { $self->{exit}{$_} == 0 } keys $self->{exit}->%*;
    my @fail    = sort grep { $self->{exit}{$_} != 0 } keys $self->{exit}->%*;
    if (!$self->{quiet}) {
        print STDERR "\e[32mSUCCESS\e[m $_\n" for @success;
        print STDERR "\e[31mFAIL\e[m $_\n"    for @fail;
    }
    return @fail ? 1 : 0;
}

sub show_help ($self) {
    open my $fh, '>', \my $out;
    Pod::Usage::pod2usage
        exitval => 'noexit',
        input => $0,
        output => $fh,
        sections => 'SYNOPSIS|OPTIONS|EXAMPLES',
        verbose => 99,
    ;
    $out =~ s/^[ ]{4,6}/  /mg;
    $out =~ s/\n$//;
    print $out;
}

sub parse_options ($self, @argv) {
    my $parser = Getopt::Long::Parser->new(
        config => [qw(no_auto_abbrev no_ignore_case)],
    );
    $parser->getoptionsfromarray(
        \@argv,
        "c|concurrency=i"     => \($self->{concurrency} = 5),
        "h|help"              => sub (@) { $self->show_help; exit 1 },
        "s|script=s"          => \($self->{script}),
        "v|version"           => sub (@) { printf "%s %s\n", __PACKAGE__, __PACKAGE__->VERSION; exit },
        "a|ask-sudo-password" => \(my $ask_sudo_password),
        "H|host-file=s"       => \(my $host_file),
        "sudo-password=s"     => \($self->{sudo_password}),
        "append-hostname!"    => \(my $append_hostname = 1),
        "append-time!"        => \(my $append_time),
        "sudo=s"              => \($self->{sudo_user}),
        "q|quiet"             => \($self->{quiet}),
        "F=s"                 => \($self->{configfile}),
    ) or exit(2);

    my $host_arg = $host_file ? undef : shift @argv;
    if ($self->{script}) {
        $self->{script_arg} = \@argv;
    } else {
        $self->{command} = \@argv;
    }

    if (!($self->{command} || [])->@* && !$self->{script}) {
        warn "COMMANDS or --script option is required\n";
        exit(2);
    }
    if ($self->{script}) {
        my ($tempfh, $tempfile) = File::Temp::tempfile(UNLINK => 1, EXLOCK => 0);
        File::Copy::copy($self->{script}, $tempfh)
            or die "copy $self->{script} to tempfile: $!";
        close $tempfh;
        chmod 0755, $tempfile;
        $self->{script} = $tempfile;
    }

    $self->{format} = $self->make_format(
        append_hostname => $append_hostname,
        append_time => $append_time,
    );

    if ($ask_sudo_password) {
        my $password = prompt $SUDO_PROMPT;
        $self->{sudo_password} = $password;
    }
    $self->{host} = $host_file ? $self->parse_host_file($host_file)
                               : $self->parse_host_arg($host_arg);
    $self;
}

sub cancel ($self, $signal) {
    $self->{pending}->@* = ();
    for my $ssh ($self->{running}->all) {
        $ssh->cancel($signal);
    }
}

sub one_tick ($self) {

    while ($self->{running}->count < $self->{concurrency} and my $ssh = shift $self->{pending}->@*) {
        $self->{running}->add($ssh);
    }

    my @ready = $self->{select}->can_read(TICK_SECOND);
    DEBUG and logger "one tick running %d (watching %d, can_read %d), pending %d",
        $self->{running}->count, $self->{select}->count, scalar @ready, scalar $self->{pending}->@*;

    if ($self->{select}->count == 0) {
        select undef, undef, undef, TICK_SECOND;
    } else {
        $self->_process($_) for @ready;
    }

    # -1: there is no child process
    #  0: all child process are running
    my $pid = waitpid -1, POSIX::WNOHANG;
    my $exit = $?;

    if ($pid > 0 and my $remove = $self->{select}->remove(pid => $pid)) {
        $self->_post_process($remove);
    }

    for my $ssh ($self->{running}->all) {
        my %args = (select => $self->{select}, $pid > 0 ? (pid => $pid, exit => $exit) : ());
        my $is_running = $ssh->one_tick(%args);
        if (!$is_running) {
            $self->{exit}{$ssh->host} = $ssh->exit;
            $self->{running}->remove($ssh);
        }
    }
}

# We close fh explicitly; otherwise it happens that
# perl warns "unnable to close filehandle properly: Input/output error" under ssh proxy
sub _process ($self, $ready) {
    my ($fh, $pid, $host, $buffer) = @{$ready}{qw(fh pid host buffer)};
    my $len = sysread $fh, my $buf, 64*1024;
    my ($errno, $errmsg) = (0+$!, "$!");
    DEBUG and logger " READ %s, pid %d, len %s, err: %s",
        $host, $pid, defined $len ? $len : 'undef', $errmsg || "N/A";
    if ($len) {
        if (my @line = $buffer->add($buf)->get) {
            print $self->{format}->($host, $_) for @line;