lib/bin/bex-queue

   1 #!/usr/bin/perl
   2 # Batch EXecutor -- Show Queued Jobs
   3 # (c) 2011-2015 Martin Mares <mj@ucw.cz>
   4
   5 use strict;
   6 use warnings;
   7 use Getopt::Long;
   8 use POSIX;
   9 use BEX;
  10
  11 my $op_by_job;
  12 my $op_by_host;
  13 my $op_rm;
  14 my $op_move_to;
  15
  16 my $queue_name;
  17 my $attention;
  18 my $filenames;
  19 my $given_job;
  20 my $summary;
  21 my $state_regex;
  22 my $why;
  23
  24 sub usage() {
  25         print <<AMEN ;
  26 Usage: bex queue [<options and actions>] [[!]<machine-or-group> ...]
  27
  28 Actions:
  29     --by-job            Show jobs sorted by job ID (default)
  30 -h, --by-host           Show jobs sorted by host
  31     --rm                Remove jobs from the queue
  32     --move-to=<queue>   Move jobs to a different queue
  33
  34 Options:
  35 -a, --attn              Show only jobs needing attention (e.g., failures)
  36 -f, --filenames         Show filenames of jobs and log files
  37 -j, --job=<id>          Act on the specified job (default: on all)
  38 -q, --queue=<name>      Act on the given queue
  39 -s, --summary           Show only a summary
  40 -S, --state=<regex>     Act only on jobs whose state matches the given regex
  41 -w, --why[=<lines>]     In case of failed jobs, display last few lines of output
  42 AMEN
  43         exit 0;
  44 }
  45
  46 Getopt::Long::Configure("bundling");
  47 GetOptions(
  48         "by-job!" => \$op_by_job,
  49         "h|by-host!" => \$op_by_host,
  50         "rm!" => \$op_rm,
  51         "move-to=s" => \$op_move_to,
  52         "a|attn!" => \$attention,
  53         "f|filenames!" => \$filenames,
  54         "j|job=s" => \$given_job,
  55         "q|queue=s" => \$queue_name,
  56         "s|summary!" => \$summary,
  57         "S|state=s" => \$state_regex,
  58         "w|why:i" => \$why,
  59         "help" => \&usage,
  60 ) or die "Try `bex queue --help' for more information.\n";
  61
  62 my @machines = BEX::Config::parse_machine_list(@ARGV ? @ARGV : '*');
  63 my $queue = BEX::Queue->new($queue_name);
  64 $given_job = $queue->resolve_job_id($given_job) if defined $given_job;
  65
  66 # Status cache
  67 my %status_cache = ();
  68 sub get_status($$) {
  69         my ($m, $j) = @_;
  70         $status_cache{$m}{$j} or $status_cache{$m}{$j} = $queue->read_job_status($m, $j);
  71         return $status_cache{$m}{$j};
  72 }
  73 sub get_stat($$) {
  74         my ($m, $j) = @_;
  75         return get_status($m, $j)->{'Status'} // 'UNKNOWN';
  76 }
  77
  78 # Select jobs
  79 my %jobs = ();
  80 my %machs = ();
  81 for my $m (@machines) {
  82         for my $j ($queue->scan($m)) {
  83                 if (defined $given_job) {
  84                         next if $j ne $given_job;
  85                 }
  86                 if (defined $attention) {
  87                         next if get_stat($m, $j) =~ m{^(NEW|NOPING)$};
  88                 }
  89                 if (defined $state_regex) {
  90                         next unless get_stat($m, $j) =~ m{^($state_regex)$};
  91                 }
  92                 push @{$jobs{$j}}, $m;
  93                 push @{$machs{$m}}, $j;
  94         }
  95 }
  96
  97 sub do_ls();
  98 sub do_rm();
  99 sub do_move_to();
 100
 101 my $ops = 0 + defined($op_by_host) + defined($op_by_job) + defined($op_rm) + defined($op_move_to);
 102 if ($ops > 1) { die "Multiple actions are not allowed\n"; }
 103
 104 if ($op_rm) { do_rm(); }
 105 elsif (defined $op_move_to) { do_move_to(); }
 106 else { do_ls(); }
 107 exit 0;
 108
 109 sub do_ls()
 110 {
 111         my %stat = ();
 112         my %mach_locked = ();
 113         my %cnt_by_job = ();
 114         my %cnt_by_mach = ();
 115         my %why = ();
 116         for my $m (keys %machs) {
 117                 $mach_locked{$m} = $queue->is_locked($m, undef);
 118                 for my $j (@{$machs{$m}}) {
 119                         my $st = get_status($m, $j);
 120                         my $s = get_stat($m, $j);
 121                         my $adir = $queue->attachment_dir($j);
 122                         my $has_adir = -d $adir;
 123                         $cnt_by_job{$j}{$s}++;
 124                         $cnt_by_mach{$m}{$s}++;
 125                         $why{$m}{$j} = "";
 126                         if ($filenames) {
 127                                 $why{$m}{$j} .= "\t\t== Job file: " . $queue->queue_file($m, $j) . "\n";
 128                                 $why{$m}{$j} .= "\t\t== Attachments: $adir\n" if $has_adir;
 129                         }
 130                         if (defined($st->{'Time'}) && defined($st->{'Status'})) {
 131                                 $stat{$m}{$j} = ' [' . $st->{'Status'} . ' on ' .
 132                                                 POSIX::strftime('%Y-%m-%d', localtime $st->{'Time'}) . ']';
 133                                 if (defined($why) && $st->{'Status'} eq 'FAILED') {
 134                                         my $lines = $why ? $why : 3;
 135                                         my $log = $queue->log_file($m, $j);
 136                                         if (-f $log) {
 137                                                 $why{$m}{$j} .= join("", map { "\t\t>> $_" } `tail -n$lines $log`);
 138                                         }
 139                                 }
 140                         } else {
 141                                 $stat{$m}{$j} = '';
 142                         }
 143                         if ($mach_locked{$m} || $queue->is_locked($m, $j)) {
 144                                 $stat{$m}{$j} .= ' [LOCKED]';
 145                         }
 146                         $stat{$m}{$j} .= ' +ATT' if $has_adir;
 147                 }
 148         }
 149
 150         if ($queue->is_locked(undef, undef)) {
 151                 print "### Queue lock present\n\n";
 152         }
 153
 154         if ($summary) {
 155                 if ($op_by_host) {
 156                         for my $m (sort keys %cnt_by_mach) {
 157                                 print "$m: ", join(" ", map { "$_:" . $cnt_by_mach{$m}{$_} } sort keys %{$cnt_by_mach{$m}}), "\n";
 158                         }
 159                 } else {
 160                         for my $j (sort keys %cnt_by_job) {
 161                                 print $queue->job_name($j), ": ", join(" ", map { "$_:" . $cnt_by_job{$j}{$_} } sort keys %{$cnt_by_job{$j}}), "\n";
 162                         }
 163                 }
 164         } else {
 165                 if ($op_by_host) {
 166                         for my $m (sort keys %machs) {
 167                                 print "$m", ($mach_locked{$m} ? ' [LOCKED]' : ''), "\n";
 168                                 for my $j (@{$machs{$m}}) {
 169                                         print "\t" . $queue->job_name($j) . $stat{$m}{$j}, "\n";
 170                                         print $why{$m}{$j};
 171                                 }
 172                         }
 173                 } else {
 174                         for my $j (sort keys %jobs) {
 175                                 print $queue->job_name($j), "\n";
 176                                 for my $m (sort @{$jobs{$j}}) {
 177                                         print "\t$m", $stat{$m}{$j}, "\n";
 178                                         print $why{$m}{$j};
 179                                 }
 180                         }
 181                 }
 182         }
 183 }
 184
 185 sub do_rm()
 186 {
 187         my $err = 0;
 188         for my $m (sort keys %machs) {
 189                 for my $j (sort @{$machs{$m}}) {
 190                         if (!$queue->lock($m, $j)) {
 191                                 print STDERR "Cannot remove $m:", $queue->job_name($j), ", it is locked\n";
 192                                 $err = 1;
 193                         } else {
 194                                 $queue->update_job_status($m, $j, 'REMOVED');
 195                                 $queue->remove($m, $j);
 196                                 print "Removed $m:", $queue->job_name($j), "\n";
 197                         }
 198                 }
 199         }
 200         $queue->unlock;
 201         exit $err;
 202 }
 203
 204 sub do_move_to()
 205 {
 206         my $err = 0;
 207         my $dest = BEX::Queue->new($op_move_to);
 208         $dest->{'Name'} ne $queue->{'Name'} or die "Moving to the same queue is not permitted\n";
 209         for my $j (sort keys %jobs) {
 210                 my $job = BEX::Job->new_from_file($queue->job_file($j));
 211                 for my $m (sort @{$jobs{$j}}) {
 212                         if (!$queue->lock($m, $j)) {
 213                                 print STDERR "Cannot move $m:", $queue->job_name($j), ", it is locked\n";
 214                                 $err = 1;
 215                         } else {
 216                                 my $enq = $dest->enqueue($m, $job);
 217                                 if ($enq) {
 218                                         $dest->update_job_status($m, $job->id, 'NEW', 'Moved to this queue');
 219                                 } else {
 220                                         $dest->log($m, $job->id, 'REQUEUE', 'Moved to this queue');
 221                                 }
 222                                 $queue->update_job_status($m, $job->id, 'REMOVED', 'Moved from this queue');
 223                                 $queue->remove($m, $j);
 224                                 print "Moved $m:", $dest->job_name($j);
 225                                 print " (already queued)" if !$enq;
 226                                 print "\n";
 227                         }
 228                 }
 229         }
 230         $queue->unlock;
 231         exit $err;
 232 }