perl - Dance アプリケーションで AnyEvent を実行する

Question

追跡している数千台のマシン (自分のマシン) に対してノンブロッキングSSHを実行したいと考えています。Dancerアプリケーションを稼働させており、AnyEvent::timerを使用してSSHを実行したいと考えています。コマンドを非同期に実行します (各マシンには独自のポーリング間隔があり、あるマシンが別のマシンでSSH作業が完了するまで待機する必要はありません)。

同期環境で非同期に動作する最良の方法は何ですか?

Answer 1

私は良い考えではありませんでしたが、それは可能です。スクリプトをリモートで実行するための大きな Dancer アプリケーションがあり、フォークと Net::SSH2 で実行しています。スレッドで試してみましたが、スレッドセーフではないモジュールがいくつかあるため、フォークを使用することをお勧めします。

私のブログhttp://perlondancer.blogspot.mx/2014/04/executing-remote-commands-from-dancer.htmlにいくつかのコメントがあり、この要点には以下のコード例があります: https://gist.github. com/johandry/11197516

#!/usr/bin/env perl

use strict;
use warnings;
use Dancer;
use Net::SSH2;

sub execCommand ($$) {
  my ( $ssh2, $cmd ) = @_;

  my %args=(
    timeout => 1_000,  # polling timeout
    bufsize => 10_240, # read buffer size when polling
  );

  $ssh2->blocking(1); #needed for ssh->channel
  my $chan=$ssh2->channel(); # create SSH2 channel
  if ($ssh2->error()) {
    return (undef, undef, 100);
  }

  # exec $cmd (caveat: only one command line can be executed over this channel. No "ls -l;whoami" combo. Use ssh->shell instead.
  unless ($chan->exec($cmd)) {
    return (undef, undef, 500);
  }

  # defin polling context: will poll stdout (in) and stderr (ext)
  my @poll = ( { handle => $chan, events => ['in','ext'] } );

  my %std=();     # hash of strings. store stdout/stderr results
  $ssh2->blocking( 0 ); # needed for channel->poll
  while(!$chan->eof) { # there still something to read from channel
      $ssh2->poll( $args{'timeout'}, [ @poll ] ); # if any event, it will be store into $poll;

      my( $n, $buf ); # number of bytes read (n) into buffer (buf)
      foreach my $poll ( @poll ) { # for each event
          foreach my $ev ( qw( in ext ) ) { #for each stdout/stderr
              next unless $poll->{revents}{$ev};

              #there are something to read here, into $std{$ev} hash
              if( $n = $chan->read( $buf, $args{'bufsize'}, $ev eq 'ext' ) ) { #got n byte into buf for stdout ($ev='in') or stderr ($ev='ext')
                  $std{$ev}.=$buf;
              }
          } #done foreach
      }
  }
  $chan->wait_closed(); #not really needed but cleaner

  my $exit_code=$chan->exit_status();
  $chan->close(); #not really needed but cleaner

  $ssh2->blocking(1); # set it back for sanity (future calls)

  return ($std{'in'},$std{'ext'},$exit_code);
}   

sub execute ($$$$) {
  my ($ip, $username, $password, $cmd) = @_;
  my $pid = fork();

  if ($pid) {
    # This is the parent (DANCER)
    debug "Process started with PID $pid\n";
  } elsif ( $pid == 0 ) {
    # This is the child 
    my $ssh2 = Net::SSH2->new();
    $ssh2->connect( $ip ) or debug("Cannot connect to $ip");
    my $publicKeyFile  = './id_rsa.pub'; # path(setting('appdir'), 'db', 'id_rsa.pub'); # I prefer to copy the public key in your app dir due to permissions issues
    my $privateKeyFile = './id_rsa';     # path(setting('appdir'), 'db', 'id_rsa'); # I prefer to copy the private key in your app dir due to permissions issues
    if ( $ssh2->auth_publickey( $username, $publicKeyFile, $privateKeyFile, $password ) ) {
      my ($stdout, $stderr, $exitcode) = execCommand($ssh2, $cmd);
    } else {
      debug "Could not authenticate to $ip with $username";
    }
    $ssh2->disconnect();
  } else {
    debug "Could not fork: $!\n";
  }
}

set logger => "console";
set log    => "core";
set show_errors => 1;

get '/uptime/:ip' => sub {

  my $username = "the username";
  my $password = "the password";

  execute(param('ip'), $username, $password, "uptime > /tmp/dancer_example.txt");

  return 'uptime is running';
};

dance;

true;