brun

   1 #!/usr/bin/perl
   2 # Batch EXecutor 2.0 -- Run Queued Jobs
   3 # (c) 2011 Martin Mares <mj@ucw.cz>
   4
   5 use strict;
   6 use warnings;
   7 use Getopt::Long;
   8
   9 use lib 'lib';
  10 use BEX;
  11
  12 my $queue_name;
  13
  14 GetOptions(
  15         "q|queue=s" => \$queue_name,
  16 ) or die <<AMEN ;
  17 Usage: brun [<options>] [[!]<machine-or-class> ...]
  18
  19 Options:
  20 -q, --queue=<name>      Run jobs in the given queue
  21 AMEN
  22
  23 sub ping_machine($) {
  24         my ($mach) = @_;
  25         `ping -c1 -n $mach >/dev/null 2>/dev/null`;
  26         return !$?;
  27 }
  28
  29 sub run_job($$$) {
  30         my ($job, $queue, $mach) = @_;
  31         # FIXME: rsyncing, rsync-only jobs
  32         # FIXME: Locking
  33
  34         my $tmp = $queue->temp_file($mach, $job->{'ID'});
  35         open T, '>', $tmp or die;
  36         if (defined $BEX::Config::job_prolog) {
  37                 open P, $BEX::Config::job_prolog or return "Cannot open prolog: $!";
  38                 while (<P>) { print T; }
  39                 close P;
  40         } else {
  41                 print T "#!/bin/sh\n";
  42         }
  43         print T "# BEX job ", $job->{'ID'}, "\n";
  44         print T $job->{'body'};
  45         if (defined $BEX::Config::job_epilog) {
  46                 open E, $BEX::Config::job_epilog or return "Cannot open epilog: $!";
  47                 while (<E>) { print T; }
  48                 close E;
  49         }
  50         close T;
  51
  52         my $cmd = 't=$(mktemp -t bex-XXXXXXXX) && cat >$t && chmod u+x $t && echo $t';
  53         my $rtmp = `ssh <$tmp $mach '$cmd'`;
  54         !$? && defined($rtmp) && $rtmp ne '' or return "Transfer failed";
  55         chomp $rtmp;
  56
  57         system 'ssh', '-t', $mach, "$rtmp ; e=\$? ; rm -f $rtmp ; exit \$e";
  58         if ($?) {
  59                 return 'Failed';
  60         } else {
  61                 return 'OK';
  62         }
  63 }
  64
  65 my @machines = BEX::Config::parse_machine_list(@ARGV ? @ARGV : '*');
  66 my $queue = BEX::Queue->new($queue_name);
  67 for my $mach (@machines) {
  68         my @q = $queue->scan($mach) or next;
  69         my $ping;
  70         for my $jid (@q) {
  71                 my $job = BEX::Job->new_from_file($queue->job_file($jid));
  72                 my $stat = {
  73                         'Time' => time,
  74                 };
  75                 print "### Running $jid (", $job->attr('Subject'), ") on $mach ###\n";
  76                 $ping //= ping_machine($mach);
  77                 my $s;
  78                 if (!$ping) {
  79                         $s = 'No ping';
  80                 } else {
  81                         $s = run_job($job, $queue, $mach);
  82                 }
  83
  84                 BEX::log("$mach $jid $s");
  85                 if ($s eq 'OK') {
  86                         print "+++ OK\n";
  87                         $queue->remove($mach, $jid);
  88                 } else {
  89                         print "--- $s\n";
  90                         $stat->{'Status'} = $s;
  91                         $queue->write_job_status($mach, $jid, $stat);
  92                 }
  93         }
  94 }