swarm_manager

#!/usr/local/bin/perl
# ===========================================================================
#
#                            PUBLIC DOMAIN NOTICE
#                     Center for Information Technology (CIT)
#                        National Institute of Health (NIH)
#
#  This software/database is a "United States Government Work" under the
#  terms of the United States Copyright Act.  It was written as part of
#  the author's official duties as a United States Government employee and
#  thus cannot be copyrighted.  This software is freely available
#  to the public for use.  The Center for Information Technology, The
#  National Institutes of Health, and the U.S. Government have not placed
#  any restriction on its use or reproduction.
#
#  Although all reasonable efforts have been taken to ensure the accuracy
#  and reliability of the software and data, CIT, NIH and the U.S.
#  Government do not and cannot warrant the performance or results that
#  may be obtained by using this software or data. CIT, NIH and the U.S.
#  Government disclaim all warranties, express or implied, including
#  warranties of performance, merchantability or fitness for any particular
#  purpose.
#
#  Please cite the author and the "NIH Biowulf Cluster" in any work or product
#  based on this material.
#
# ===========================================================================
use strict;
use HPCNIH::Staff::MySQL::Catalog;
use HPCNIH::Util::EmailTools;
use HPCNIH::Util::TimeTools;
use Fcntl ":flock";
use HPCNIH::Util::Tabulate;
use Data::Dumper;
use Date::Parse qw(str2time);
use HPCNIH::Util::PrintColors;
use HPCNIH::Util::Stopwatch;
use HPCNIH::Slurm::CurrentJobs;
use Time::Piece;
use POSIX qw(setsid strftime);
use DBI;
use FileHandle;
use File::Basename;
use File::Find;
use FindBin qw($RealBin);
use Cwd 'abs_path';
use Sort::ArrayOfArrays;
$|=1;

my %PAR;
@{$PAR{fields}} = ('id','tempname','jobid','user','p','num','active','create_time','end_time','delete_time');
$PAR{tempdir_index} = "/usr/local/logs/swarm_tempdir.idx";
tie $PAR{stopwatch},'HPCNIH::Util::Stopwatch';
tie $PAR{stopwatch_full},'HPCNIH::Util::Stopwatch';
$PAR{stopwatch} = 0; # reset the stopwatch
$PAR{stopwatch_full} = 0; # reset the stopwatch_full
$PAR{jobs_table} = "jobs";
$PAR{swarms_table} = "swarms";
$PAR{logfile} = "/usr/local/logs/swarm_cleanup.log";
$PAR{max_days_ago} = 7; # how many days back should we look for swarms in index, logfile, jobs, and sacct?
$PAR{'delete-age'} = 5; # how old should a swarm be to be deleted?
$PAR{'purge-age'} = $PAR{max_days_ago} + 3; # delete_time must be older than the max_days_ago
$PAR{now} = time();
my $OLDDATA;
my $NEWDATA;

my %OPT;
getOptions();

eval {
# Catch signals
  local $SIG{ALRM} = sub { die " TIMEOUT: $OPT{timeout} seconds elapsed" };
  local $SIG{INT} = sub { die " INT: don't interrupt me!" };
  local $SIG{KILL} = sub { die " KILL: arrggg!" };

# Set the alarm to go off -- do nothing if $OPT{timeout} is undefined
  alarm $OPT{timeout};

#--------------------------------------------------------------------------------------------------
# Stuff that needs a timeout goes here
#--------------------------------------------------------------------------------------------------

  $PAR{dbh} = DBI->connect("DBI:mysql:;mysql_read_default_group=dashboard;mysql_read_default_file=/usr/local/etc/my.cnf;mysql_connect_timeout=10",undef,undef,{RaiseError=>0,PrintError=>0,AutoCommit=>0}) or die();

  if ($OPT{wipeout}) {
    _do_mysql("DROP TABLE IF EXISTS `$PAR{swarms_table}`");
    create_swarms_table();
    exit;
  }

  $PAR{routine_message} = printSwarmUsage() if $OPT{routine};
  $PAR{NOW} = time();
  read_swarms_table();
  parse_index_and_logfile() if $OPT{update};
  parse_sbatch_logs() if $OPT{update};
  parse_jobs_table() if $OPT{update};
  parse_sacct() if $OPT{update};
  update_swarms_table() if $OPT{update};
  delete_swarms() if $OPT{delete};
  purge_swarms() if $OPT{purge};
  remove_empty_swarm_directories() if ($OPT{delete});
  print_data_level_1() if ($OPT{verbose} == 1);
  print_data_level_2() if ($OPT{verbose} > 1);
  add_to_message((sprintf "TOTAL TIME TAKEN = %.2f",$PAR{stopwatch_full}),1) if $OPT{timing};

  if ($OPT{routine}) {
    while ((time()-$PAR{NOW}) <= 360) {
      sleep 10;
    }
    $PAR{routine_message} .= printSwarmUsage();
  }

  $PAR{message} = $PAR{routine_message} if $OPT{routine};
  if ($OPT{emailTo}) {
    sendEmail(subject=>"swarm_cleanup.log",message=>$PAR{message},emailTo=>$OPT{emailTo},Bcc=>$OPT{Bcc},debug=>$OPT{'dry-run'},provenance=>1);
  }
  else {
    print $PAR{message};
  }

  $PAR{dbh}->disconnect();

#--------------------------------------------------------------------------------------------------
# Done
#--------------------------------------------------------------------------------------------------

  alarm 0;
  1;  # default return value from eval
};

# Take action if either an error is given or the timeout was reached
if ( $@ ) {
  print STDERR $@;
  if ($OPT{emailTo}) {
    sendEmail(subject=>"ERROR: $0",message=>$@,emailTo=>"hooverdm",debug=>$OPT{'dry-run'},provenance=>1);
  }
  exit 1;
}

#==================================================================================================
sub getOptions
{ 
  use Getopt::Long qw(:config no_ignore_case);
  Getopt::Long::Configure("bundling"); # allows option bundling
  GetOptions(
    'help' => \$OPT{help},
    'h' => \$OPT{help},
    'dry-run' => \$OPT{'dry-run'},
    'd' => \$OPT{'dry-run'},
    'verbose=i' => \$OPT{verbose},
    'v=i' => \$OPT{verbose},
    'delete' => \$OPT{delete},
    'delete-age=i' => \$OPT{'delete-age'}, 
    'purge' => \$OPT{purge},
    'wipeout' => \$OPT{wipeout},
    'routine' => \$OPT{routine},
    'update' => \$OPT{update},
    'timing' => \$OPT{timing},
    'timeout=i' => \$OPT{timeout}, 
    'email=s'=> \$OPT{emailTo},
    'bcc=s' => \$OPT{Bcc}, 
    'hours' => \$OPT{hours}, 
    'minutes' => \$OPT{minutes}, 
    'human' => \$OPT{human}, 
    'u=s' => \$OPT{user}, 
    'user=s' => \$OPT{user}, 
    'j=i' => \$OPT{jobid}, 
    'jobid=i' => \$OPT{jobid}, 
    't=s' => \$OPT{tempdir}, 
    'tempdir=s' => \$OPT{tempdir}, 
    'show-sql' => \$OPT{"show-sql"}, 
  ) || printOptions();

  printOptions() if $OPT{help};

  $OPT{update} = 1 if $OPT{delete};

  add_to_message("Running in dry-run mode",1) if $OPT{'dry-run'};
  $OPT{verbose} = 2 if ((defined $OPT{'dry-run'}) && (not defined $OPT{verbose}));
  $OPT{verbose} = 1 unless (defined $OPT{verbose});
  $OPT{verbose} = 0 if ((defined $OPT{verbose}) && ($OPT{verbose} < 0));

  if ($OPT{routine}) {
    $OPT{update} = 1;
    $OPT{delete} = 1;
    $OPT{purge} = 1;
    undef $OPT{verbose};
  }

  $OPT{'show-sql'} = 1 if ((defined $OPT{verbose}) && ($OPT{verbose} > 2));
  $PAR{'delete-age'} = $OPT{'delete-age'} if ($OPT{'delete-age'});
}
#==================================================================================================
sub remove_empty_swarm_directories
{
  find (\&empty_swarm_dirs,"/spin1/swarm");
  return unless $PAR{empty_directories};
  foreach my $d (sort @{$PAR{empty_directories}}) {
    rmdir $d;
  }
}
#==================================================================================================
sub empty_swarm_dirs{
  if ($File::Find::name) { # be safe
    if (-d $File::Find::name) { # only think about directories
      my $n = ($File::Find::name) =~ tr!/!!; # count slashes in file 
      return $File::Find::prune = 1 if $n > 3; # don't go more than 3 levels
      my $number_of_files = grep $_ => glob("$_/*"); # find how many files are inside
      if ( $number_of_files == 0 ) { push @{$PAR{empty_directories}}, $File::Find::name; }
    }
  }
}
#==================================================================================================
sub read_swarms_table
{
  $PAR{stopwatch} = 0; # reset the stopwatch

# Do some filtering if needed
  my @f;
  push @f,"`user`='$OPT{user}'" if $OPT{user};
  push @f,"`jobid`='$OPT{jobid}'" if $OPT{jobid};
  push @f,"`tempname`='$OPT{tempdir}'" if $OPT{tempdir};
  my $filter;
  if (@f) {
    $filter = join " AND ",@f;
    $filter = "WHERE ".$filter;
  }

# Pull out everything from jobs table, and determine the tempdir name
  $PAR{stopwatch} = 0; # reset the stopwatch
  my $sql = "SELECT * FROM dashboard.$PAR{swarms_table} $filter";
  my $a = $PAR{dbh}->selectall_hashref($sql,["id"]);
  add_to_message($sql,1) if $OPT{'show-sql'};
  foreach my $id (keys %{$a}) {
    my $t = $a->{$id}->{tempname};
    foreach my $k (@{$PAR{fields}}) {
      $OLDDATA->{$a->{$id}->{tempname}}{$k} = $a->{$id}->{$k} if (defined $a->{$id}->{$k});
      $NEWDATA->{$a->{$id}->{tempname}}{$k} = $a->{$id}->{$k} if (defined $a->{$id}->{$k});
    }
    $PAR{jobid2tempname}{$a->{$id}->{jobid}}=$t if defined $a->{$id}->{jobid};
  }

  add_to_message((sprintf "read_swarms_table = %.2f",$PAR{stopwatch}),1) if $OPT{timing};
}
#==================================================================================================
sub parse_index_and_logfile
{
  $PAR{stopwatch} = 0; # reset the stopwatch
  open INDEXFILE, "<$PAR{tempdir_index}";
  add_to_message("reading $PAR{tempdir_index}",1) if ($OPT{verbose} > 3);
  while (<INDEXFILE>) {
    chomp(my $line = $_);
    MATCH1: while ($line=~/(\d{10}),([^,]+),(\w{10}),(\d+),([12])/g) { # Hiccups in the write performance of /spin1
      my ($time,$user,$t,$num,$p) = ($1,$2,$3,$4,$5);

# Filters
      next MATCH1 if ((defined $OPT{user}) && ($OPT{user} ne $user));
      next MATCH1 if ((defined $OPT{tempdir}) && ($OPT{tempdir} ne $t));
      if (defined $OPT{jobid}) {
        if (defined $PAR{jobid2tempname}{$OPT{jobid}}) {
          next MATCH1 if ($t ne $PAR{jobid2tempname}{$OPT{jobid}});
        }
        else {
          next MATCH1;
        }
      }

# Only find jobs created within the last max_days_ago -- this takes a few seconds
      unless ($OPT{tempdir} || $OPT{jobid}) {
        next MATCH1 if ((($PAR{now}-$time)/86400)>$PAR{max_days_ago});
      }

      $NEWDATA->{$t}{tempname}=$t;
      _update_value($NEWDATA->{$t},"create_time",$time);
      _update_value($NEWDATA->{$t},"user",$user);
      _update_value($NEWDATA->{$t},"p",$p);
      _update_value($NEWDATA->{$t},"num",$num);
    }
  }
  close INDEXFILE;

  my @files;
  chomp(my $x = `/bin/find /usr/local/logs/swarm_log_archives -type f -mtime -$PAR{max_days_ago}`);
  @files = split /\n/,$x;
  @files = sort @files; 
  push @files,"/usr/local/logs/swarm.log";
  LOG: foreach my $f (@files) {
    if (open LOGFILE, "<$f") {
      add_to_message("reading $f",1) if ($OPT{verbose} > 3);
      LINE: while (<LOGFILE>) {
        my $line = $_;
        if ($line=~/^date=([^;]+);\s.+jobid=([^;]+);\suser=([^;]+);\s.+njobs=([^;]+);\stempname=([^;]+);/) {
          my ($time,$jobid,$user,$num,$t) = (str2time($1),$2,$3,$4,$5);
          next LINE if ((defined $OPT{user}) && ($OPT{user} ne $user));
          next LINE if ((defined $OPT{jobid}) && ($OPT{jobid} ne $jobid));
          next LINE if ((defined $OPT{tempdir}) && ($OPT{tempdir} ne $t));

# Only find jobs created within the last max_days_ago -- this takes a few seconds
          unless ($OPT{tempdir} || $OPT{jobid}) {
            next LINE if ((($PAR{now}-$time)/86400)>$PAR{max_days_ago});
          }
          my $p = 1;
          $p = 2 if ($line=~/; command=.+-p 2 /);
          $NEWDATA->{$t}{tempname}=$t;
          _update_value($NEWDATA->{$t},"create_time",$time);
          _update_value($NEWDATA->{$t},"user",$user);
          _update_value($NEWDATA->{$t},"num",$num);
          _update_value($NEWDATA->{$t},"p",$p);
          _update_value($NEWDATA->{$t},"jobid",$jobid);
          $PAR{jobid2tempname}{$jobid}=$t if ((defined $jobid) && (defined $t));

          if ($OPT{jobid} || $OPT{tempdir}) {
            add_to_message("found $OPT{jobid}/$OPT{tempdir} in $f",1) if ($OPT{verbose} > 3);
            close LOGFILE;
            last LOG;
          }
        }
      }
      close LOGFILE;
    }
    else {
      add_to_message("failed to read $f",1) if ($OPT{verbose} > 3);
    }
  }
  add_to_message((sprintf "parse_index_and_logfile = %.2f",$PAR{stopwatch}),1) if $OPT{timing};
}
#=================================================================================================
sub parse_sbatch_logs
{
  $PAR{stopwatch} = 0; # reset the stopwatch

  my @files;
  chomp(my $x = `/bin/find /usr/local/logs/sbatch_log_archives -type f -mtime -$PAR{max_days_ago}`);
  @files = split /\n/,$x;
  @files = sort @files;
  push @files,"/usr/local/logs/sbatch.log";
 
  LOG: foreach my $f (@files) {
    if (open LOGFILE, "<$f") {
      add_to_message("reading $f",1) if ($OPT{verbose} > 3);
      LINE: while (<LOGFILE>) {
        my $line = $_;

#20180220 13:43:23 cn3167 SUBM[61905830]: clarkmg /data/clarkmg/ica_dualreg sbatch --array=0-0 --output=/data/clarkmg/ica_dualreg/model1_48_dualreg_split_output0016/scripts+logs/drD_%A_%a.o --error=/data/clarkmg/ica_dualreg/model1_48_dualreg_split_output0016/scripts+logs/drD_%A_%a.e --cpus-per-task=1 --dependency=afterany:61905773 --job-name=drD --mem=4096 --partition=norm --time=02:00:00 /spin1/swarm/clarkmg/mThglAnXQp/swarm.batch

        if ($line=~/^(\d{4})(\d{2})(\d{2}) (\d\d:\d\d:\d\d)\s\w+\sSUBM\[(\d+)\]:\s(\w+)\s+.+array=0-(\d+)\s.+\/spin1\/swarm\/\w+\/(\w+)\/swarm\.batch/) {
          my ($time,$jobid,$user,$num,$t) = (str2time("$1-$2-$3T$4"),$5,$6,$7,$8);
          next LINE if ((defined $OPT{user}) && ($OPT{user} ne $user));
          next LINE if ((defined $OPT{jobid}) && ($OPT{jobid} ne $jobid));
          next LINE if ((defined $OPT{tempdir}) && ($OPT{tempdir} ne $t));

# Only find jobs created within the last max_days_ago -- this takes a few seconds
          unless ($OPT{tempdir} || $OPT{jobid}) {
            next LINE if ((($PAR{now}-$time)/86400)>$PAR{max_days_ago});
          }
          my $p = 1;
          $p = 2 if ($line=~/; command=.+-p 2 /);
          $NEWDATA->{$t}{tempname}=$t;
          _update_value($NEWDATA->{$t},"create_time",$time);
          _update_value($NEWDATA->{$t},"user",$user);
          _update_value($NEWDATA->{$t},"num",$num);
          _update_value($NEWDATA->{$t},"p",$p);
          _update_value($NEWDATA->{$t},"jobid",$jobid);
          $PAR{jobid2tempname}{$jobid}=$t if ((defined $jobid) && (defined $t));

          if ($OPT{jobid} || $OPT{tempdir}) {
            add_to_message("found $OPT{jobid}/$OPT{tempdir} in $f",1) if ($OPT{verbose} > 3);
            close LOGFILE;
            last LOG;
          }
        }
      }
      close LOGFILE;
    }
    else {
      add_to_message("failed to read $f",1) if ($OPT{verbose} > 3);
    }
  }

  add_to_message((sprintf "parse_sbatch_logs = %.2f",$PAR{stopwatch}),1) if $OPT{timing};
}
#=================================================================================================
sub parse_jobs_table
{
# Don't be stupid
  return if ($OPT{tempdir} && not defined $NEWDATA->{$OPT{tempdir}}{jobid});

  $PAR{stopwatch} = 0; # reset the stopwatch
  my $JOBS;

# Do some filtering if needed
  my @f;
  push @f,"`user`='$OPT{user}'" if $OPT{user};
  push @f,"`jobid`='$OPT{jobid}'" if $OPT{jobid};

# Only find jobs created within the last max_days_ago -- this may take 30 seconds
  unless ($OPT{tempdir} || $OPT{jobid}) {
    push @f,"`submit_time`>".($PAR{now}-($PAR{max_days_ago}*86400));
  }
  my $filter;
  if (@f) {
    $filter = join " AND ",@f;
    $filter = "WHERE ".$filter;
  }

  my $sql = "SELECT jobid,jobidarray,state,end_time FROM dashboard.$PAR{jobs_table} $filter";
  add_to_message($sql,1) if $OPT{"show-sql"};
  my $sth = $PAR{dbh}->prepare($sql);
  $sth->execute;
  JOB: while (my $d = $sth->fetchrow_hashref()) {
    if ((defined $d->{jobidarray}) && ($d->{jobidarray}=~/^(\d+)_/)) {
      my $j = $1;
      $JOBS->{$j}{count}++;
      if (defined $d->{state}) { $JOBS->{$j}{state}{$d->{state}}=1; }
      if (defined $d->{end_time}) { $JOBS->{$j}{end_time}{$d->{end_time}}=1; }
    }
    else {
      $JOBS->{$d->{jobid}}{count}++;
      if (defined $d->{state}) { $JOBS->{$d->{jobid}}{state}{$d->{state}}=1; }
      if (defined $d->{end_time}) { $JOBS->{$d->{jobid}}{end_time}{$d->{end_time}}=1; }
    }
  }
  $sth->finish;
  foreach my $j (keys %{$JOBS}) { 
    if (defined $PAR{jobid2tempname}{$j}) {
      my ($active,$end_time) = find_active_and_end_time($JOBS,$j); 
      _update_value($NEWDATA->{$PAR{jobid2tempname}{$j}},"active",$active);
      _update_value($NEWDATA->{$PAR{jobid2tempname}{$j}},"end_time",$end_time);
    }
  }
  add_to_message((sprintf "parse_jobs_table = %.2f",$PAR{stopwatch}),1) if $OPT{timing};
}
#===============================================================================================================================
sub parse_sacct
{
# Don't be stupid
  return if ($OPT{tempdir} && not defined $NEWDATA->{$OPT{tempdir}}{jobid});

  $PAR{stopwatch} = 0; # reset the stopwatch

# Do some filtering if needed
  my @f;
  push @f,"--user $OPT{user}" if $OPT{user};
  push @f,"--jobs $OPT{jobid}" if $OPT{jobid};
  push @f,"--jobs $NEWDATA->{$OPT{tempdir}}{jobid}" if $OPT{tempdir};
  my $filter;
  if (@f) {
    $filter = join " ",@f;
  }

# Only find jobs created within the last max_days_ago -- for 7 days, this takes at least 4 minutes
  my $range;
  if ($OPT{tempdir}) {
    my $since = strftime("%Y-%m-%dT%H:%M:%S",(localtime($NEWDATA->{$OPT{tempdir}}{create_time}))[0 .. 5]);
    my $until = strftime("%Y-%m-%dT%H:%M:%S",(localtime($PAR{now}))[0 .. 5]);
    $range = "-S $since -E $until";
  }
  elsif ($OPT{jobid}) {
    my $since = strftime("%Y-%m-%dT%H:%M:%S",(localtime($NEWDATA->{$PAR{jobid2tempname}{$OPT{jobid}}}{create_time}))[0 .. 5]);
    my $until = strftime("%Y-%m-%dT%H:%M:%S",(localtime($PAR{now}))[0 .. 5]);
    $range = "-S $since -E $until";
  }
  else {
    my $since = strftime("%Y-%m-%dT%H:%M:%S",(localtime($PAR{now}-(86400*$PAR{max_days_ago})))[0 .. 5]);
    my $until = strftime("%Y-%m-%dT%H:%M:%S",(localtime($PAR{now}))[0 .. 5]);
    $range = "-S $since -E $until";
  }
  my $cmd = "sacct $range --format jobid,state,end -s PENDING --noheader $filter";
  #my $cmd = "sacct $range --format jobid,state,end --noheader $filter";
  add_to_message($cmd,1) if ($OPT{verbose} > 2);
  chomp(my $s = `$cmd`);
  my $JOBS;
  foreach my $line (split /\n/,$s) {
    if ($line =~ /^(\d+)\S*\s+(\S+)\s+(\S+)/) {
      my ($j,$s,$t) = ($1,$2,$3);
      if (defined $j) {
        $JOBS->{$j}{line}=$line;
        $JOBS->{$j}{count}++;
        if (defined $s) { $JOBS->{$j}{state}{$s}=1; }
        if (defined $t) {
          if ($t =~ /(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})/) {
            my $tx = str2time($1);
            $JOBS->{$j}{end_time}{$tx}=1; 
          }
        }
      }
    }
  }
  foreach my $j (keys %{$JOBS}) {
    if (defined $PAR{jobid2tempname}{$j}) {
      my ($active,$end_time) = find_active_and_end_time($JOBS,$j); 
      _update_value($NEWDATA->{$PAR{jobid2tempname}{$j}},"active",$active);
      _update_value($NEWDATA->{$PAR{jobid2tempname}{$j}},"end_time",$end_time);
    }
  }

# Ugh.  Check through old swarms (twice max_days_ago) that may be erroneously marked as R/P.
  my $ancient = time() - (86400 * ($PAR{max_days_ago}*2));
  foreach my $tempname (keys %{$NEWDATA}) {
    next if ($OPT{user} && ($NEWDATA->{$tempname}{user} ne $OPT{user}));
    next if ($OPT{jobid} && ($NEWDATA->{$tempname}{jobid} ne $OPT{jobid}));
    next if ($OPT{tempdir} && ($tempname ne $OPT{tempdir}));
    next unless ((defined $NEWDATA->{$tempname}{active}) && ($NEWDATA->{$tempname}{active} != 0)); # only evaluate R/P swarms
    next unless ($NEWDATA->{$tempname}{create_time} < $ancient); # only evaluate ancient swarms

# Use sacct -- faster than Elasticsearch
    my $cmd = "sacct -j $NEWDATA->{$tempname}{jobid} --format jobid,state,end --noheader";
    add_to_message($cmd,1) if ($OPT{verbose} > 2);
    chomp(my $s = `$cmd`);
# swarm was found in sacct
    if ($s=~/\w/) {
      add_to_message($s,1) if ($OPT{verbose} > 2);
      my $ANCIENT_JOBS;
      foreach my $line (split /\n/,$s) {
        if ($line =~ /^(\d+)\S*\s+(\S+)\s+(\S+)/) {
          my ($j,$s,$t) = ($1,$2,$3);
          if (defined $j) {
            $ANCIENT_JOBS->{$j}{line}=$line;
            $ANCIENT_JOBS->{$j}{count}++;
            if (defined $s) { $ANCIENT_JOBS->{$j}{state}{$s}=1; }
            if (defined $t) {
              if ($t =~ /(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})/) {
                my $tx = str2time($1);
                $ANCIENT_JOBS->{$j}{end_time}{$tx}=1; 
              }
            }
          }
        }
      }
      foreach my $j (keys %{$ANCIENT_JOBS}) {
        if (defined $PAR{jobid2tempname}{$j}) {
          my ($active,$end_time) = find_active_and_end_time($ANCIENT_JOBS,$j); 
          _update_value($NEWDATA->{$PAR{jobid2tempname}{$j}},"active",$active);
          _update_value($NEWDATA->{$PAR{jobid2tempname}{$j}},"end_time",$end_time);
        }
      }
    }
# WTF -- swarm slipped past and out of slurmdb
    else {
      add_to_message(" -- sacct returned nothing, marking as finished --",1) if ($OPT{verbose} > 2);
      _update_value($NEWDATA->{$tempname},"active",0);
      _update_value($NEWDATA->{$tempname},"end_time",$ancient);
    }
  }

  add_to_message((sprintf "parse_sacct = %.2f",$PAR{stopwatch}),1) if $OPT{timing};
}
#===============================================================================================================================
sub find_active_and_end_time
{
  my ($x,$j) = @_;

# Find the accumulated state and the latest end_time, if possible
  my $active;
  my $end_time;
  foreach my $s (keys %{$x->{$j}{state}}) {
# active is 1 if running or pending, 0 otherwise
    if ($s eq 'PENDING' || $s eq 'RUNNING') {
      $active+=1;
    }
    else {
      $active+=0; 
    }
  }
# Only set end_time if the job is known to have finished
  if ((defined $active) && ($active == 0)) {
    foreach my $t (keys %{$x->{$j}{end_time}}) {
      if ((not defined $end_time) || ($end_time < $t)) {
        $end_time = $t;
      }
    }
  }
  return ($active,$end_time);
}
#===============================================================================================================================
sub _update_value
{
  my ($h,$key,$value) = @_;
  return unless ((defined $h) && (defined $key) && (defined $value));
  $h->{$key} = $value;
}
#=================================================================================================
sub print_data_level_1
{
  $PAR{stopwatch} = 0; # reset the stopwatch
  my $x;
  foreach my $t (keys %{$NEWDATA}) {
    next if ($OPT{user} && ($NEWDATA->{$t}{user} ne $OPT{user}));
    next if ($OPT{jobid} && ($NEWDATA->{$t}{jobid} ne $OPT{jobid}));
    next if ($OPT{tempdir} && ($t ne $OPT{tempdir}));
    $x->{unfinished}++ if ((defined $NEWDATA->{$t}{active}) && ($NEWDATA->{$t}{active} != 0));
    $x->{finished}++ if ((defined $NEWDATA->{$t}{active}) && ($NEWDATA->{$t}{active} == 0));
    $x->{unknown}++ if (not defined $NEWDATA->{$t}{active});
  }
  add_to_message(sprintf ("R/P=%d,F=%d,U=%d", $x->{unfinished}, $x->{finished}, $x->{unknown},));
  add_to_message((sprintf "print_data_level_1 = %.2f",$PAR{stopwatch}),1) if $OPT{timing};
}
#=================================================================================================
sub print_data_level_2
{
  $PAR{stopwatch} = 0; # reset the stopwatch
  print_data_level_1();
  foreach my $t (keys %{$NEWDATA}) {
    next if ($OPT{user} && ($NEWDATA->{$t}{user} ne $OPT{user}));
    next if ($OPT{jobid} && ($NEWDATA->{$t}{jobid} ne $OPT{jobid}));
    next if ($OPT{tempdir} && ($t ne $OPT{tempdir}));
    my $state;
    if ((defined $NEWDATA->{$t}{active}) && ($NEWDATA->{$t}{active} != 0)) { $state = 'R/P'; }
    elsif ((defined $NEWDATA->{$t}{active}) && ($NEWDATA->{$t}{active} == 0)) { $state = 'F'; }
    else { $state = 'U'; }
# Format jobid
    my $jobid;
    if (not defined $NEWDATA->{$t}{jobid}) { $jobid = -1; }
    else { $jobid = $NEWDATA->{$t}{jobid}; }
# Format create_time
    my $create_time;
    if (not defined $NEWDATA->{$t}{create_time}) { $create_time = -1; }
    else { $create_time = $NEWDATA->{$t}{create_time}; }
    if (($OPT{human}) && (defined $create_time) && ($create_time > 0)) {
      $create_time = strftime("%Y-%m-%dT%H:%M:%S",(localtime($create_time))[0 .. 5]);
    }
# Format end_time
    my $end_time;
    if (not defined $NEWDATA->{$t}{end_time}) { $end_time = -1; }
    else { $end_time = $NEWDATA->{$t}{end_time}; }
    if (($OPT{human}) && (defined $end_time) && ($end_time > 0)) {
      $end_time = strftime("%Y-%m-%dT%H:%M:%S",(localtime($end_time))[0 .. 5]);
    }
# Format delete_time
    my $delete_time;
    if (not defined $NEWDATA->{$t}{delete_time}) { $delete_time = -1; }
    else { $delete_time = $NEWDATA->{$t}{delete_time}; }
    if (($OPT{human}) && (defined $delete_time) && ($delete_time > 0)) {
      $delete_time = strftime("%Y-%m-%dT%H:%M:%S",(localtime($delete_time))[0 .. 5]);
    }
    my $format = "%10s\t%8d\t%1s\t%10s\t%10s\t%10s\t%s";
    if ($OPT{human}) {
      $format = "%10s\t%8d\t%1s\t%19s\t%19s\t%19s\t%s";
    }
    add_to_message(sprintf ($format,
      $t,
      $jobid,
      $state,
      $create_time,
      $end_time,
      $delete_time,
      $NEWDATA->{$t}{user},
    ));
  }
  add_to_message((sprintf "print_data_level_2 = %.2f",$PAR{stopwatch}),1) if $OPT{timing};
}
#=================================================================================================
sub dieWithError
{
  my $message = shift;
  die "ERROR: $message\n";
}
#=================================================================================================
sub add_to_message
{
  my $msg = shift;
  my $immediate = shift;
  if ($immediate) {
    print $msg."\n";
  }
  else {
    $PAR{message} .= $msg."\n";
  }
}
#=================================================================================================
sub create_swarms_table
{
  my $zzz = $PAR{dbh}->selectall_hashref("SHOW TABLES","Tables_in_dashboard");
  if (! $zzz->{$PAR{swarms_table}}) {
    my $sql=<<EOF;
CREATE TABLE `$PAR{swarms_table}` (
  `id` INT(11) NOT NULL AUTO_INCREMENT,
  `tempname` VARCHAR(32) NOT NULL,
  `jobid` INT(11) DEFAULT NULL,
  `user` VARCHAR(32) NOT NULL,
  `p` TINYINT NOT NULL,
  `num` INT(8) NOT NULL,
  `create_time` INT(10) NOT NULL,
  `active` TINYINT DEFAULT NULL,
  `end_time` INT(10) DEFAULT NULL,
  `delete_time` INT(10) DEFAULT NULL,
  PRIMARY KEY `id` (`id`),
  UNIQUE KEY `tempname` (`tempname`),
  UNIQUE KEY `jobid` (`jobid`),
  KEY `user` (`user`),
  KEY `active` (`active`),
  KEY `create_time` (`create_time`),
  KEY `end_time` (`end_time`),
  KEY `delete_time` (`delete_time`)
)
EOF
    _do_mysql($sql);
    add_to_message("created $PAR{swarms_table} table",1);
  }
  undef $zzz;
}
#=================================================================================================
sub _do_mysql
# This subroutine accepts an array of SQL statements and attempts to run all of them atomically.
# If one fails, they all fail.  The script will not die, but will generate a warning in the logfile.
{
  my @sqlarray = @_;
  return unless @sqlarray;

# Run the SQL statements, concatenating any errors
  my $err;
  foreach my $sql (@sqlarray) {
    add_to_message($sql,1) if $OPT{"show-sql"};
    $PAR{dbh}->do($sql) unless $OPT{'dry-run'};
    $err .= $PAR{dbh}->errstr."\n" if $PAR{dbh}->errstr;
  }

# Something bad has happened
  if ($err) {

# Rollback the changes
    $PAR{dbh}->rollback unless $OPT{'dry-run'};

# Record dying message
    add_to_message($err,1);
  }
  else {
# Commit the changes
    $PAR{dbh}->commit unless $OPT{'dry-run'};
  }
  undef @sqlarray;

  return;
}
#=================================================================================================
sub _mysql_quote
{
  my $v = shift;
  if ($v=~/^\d+$/) {
    return $v;
  }
  else {
    return "'".$v."'";
  }
}
#=================================================================================================
sub update_swarms_table
{
  my @sqlarray;
  foreach my $t (keys %{$NEWDATA}) {
    if (not defined $OLDDATA->{$t}) {
      my $sql = _create_insert($t);
      push @sqlarray,$sql if (defined $sql);
    }
    else {
      my $sql = _create_update($t);
      push @sqlarray,$sql if (defined $sql);
    }
  }
  _do_mysql(@sqlarray) if @sqlarray;
}
#==================================================================================================
sub _create_insert
{
  my $t = shift;
  my @fields;
  my @values;

  foreach my $f (@{$PAR{fields}}) {
    next if ($f eq 'id');
    if (defined $NEWDATA->{$t}{$f}) {
      push @fields,"`".$f."`";
      push @values,"'".$NEWDATA->{$t}{$f}."'";
    }
  }

  my $sql = "INSERT INTO $PAR{swarms_table} (";
  $sql .= join ",",@fields;
  $sql .= ") VALUES (";
  $sql .= join ",",@values;
  $sql .= ")";
  $sql .= " ON DUPLICATE KEY UPDATE ";

  my @pairs;
  foreach my $f (@{$PAR{fields}}) {
    next if ($f eq 'id');
    next if ($f eq 'jobid');
    next if ($f eq 'template');
    if (defined $NEWDATA->{$t}{$f}) {
      push @pairs,"`$f`='".$NEWDATA->{$t}{$f}."'";
    }
  }
  $sql .= join ",",@pairs;

  return $sql;
}
#==================================================================================================
sub _create_update
{
  my $t = shift;
  my @changes;
  foreach my $k (@{$PAR{fields}}) {
    next if ($k eq 'id');
    if (((defined $NEWDATA->{$t}{$k}) && (not defined $OLDDATA->{$t}{$k})) ||
        ((defined $NEWDATA->{$t}{$k}) && ($OLDDATA->{$t}{$k} ne $NEWDATA->{$t}{$k}))) {
      push @changes,'`'.$k.'`='._mysql_quote($NEWDATA->{$t}{$k});
    }
  }
  if (@changes) {
    return "UPDATE `$PAR{swarms_table}` SET ".join(",",@changes)." WHERE `id`=$OLDDATA->{$t}{id}";
  }
  else {
    return;
  }
}
#==================================================================================================
sub delete_swarms
{
  $PAR{stopwatch} = 0; # reset the stopwatch
# Choose our victims
  my @victims;
  foreach my $t (keys %{$NEWDATA}) {
    next if ($OPT{user} && ($NEWDATA->{$t}{user} ne $OPT{user}));
    next if ($OPT{jobid} && ($NEWDATA->{$t}{jobid} ne $OPT{jobid}));
    next if ($OPT{tempdir} && ($t ne $OPT{tempdir}));
    push @victims, $t if _deleteable($t);
  }

# Delete the victims in clumps of 20
  my $list;
  my $loglist;
  my @sqlarray;
  foreach my $i (0..$#victims) {
    my $t = $victims[$i];

    $list .= " /spin1/swarm/".$NEWDATA->{$t}{user}."/".$t if ($t);
    $list .= " /spin1/swarm/".$NEWDATA->{$t}{user}."/".$NEWDATA->{$t}{jobid} if ($NEWDATA->{$t}{jobid});

    my $print_list = "rm -rf /spin1/swarm/".$NEWDATA->{$t}{user}."/".$t;
    $print_list .= " /spin1/swarm/".$NEWDATA->{$t}{user}."/".$NEWDATA->{$t}{jobid} if ($NEWDATA->{$t}{jobid});

    $loglist .= $print_list."\n";

    if (defined $OLDDATA->{$t}{id}) {
      push @sqlarray,"UPDATE `$PAR{swarms_table}` SET `delete_time`=$PAR{now} WHERE `id`=$OLDDATA->{$t}{id}";
    }
    else {
      push @sqlarray,"UPDATE `$PAR{swarms_table}` SET `delete_time`=$PAR{now} WHERE `tempname`='$t'";
    }

    add_to_message($print_list) if (defined $OPT{verbose} && $OPT{verbose} > 1);
    if ((!($i%20)) || ($i == $#victims)) {
# construct the command
      my $cmd = "rm -rf $list";
      undef $list;
# delete the directory and symlink and capture the exit code
      my $ret = system($cmd) if (!$OPT{'dry-run'});
# update swarms table and print to logfile if all went well
      _do_mysql(@sqlarray) unless $ret;
      undef @sqlarray;
      print_to_logfile($loglist) unless $ret;
      undef $loglist;
    }
  }
  if ($OPT{verbose} > 1) {
    if (!@victims) {
      add_to_message("Nothing to delete");
    }
  }
  add_to_message((sprintf "delete_swarms = %.2f",$PAR{stopwatch}),1) if $OPT{timing};

  if ($OPT{routine}) {
    $PAR{routine_message} .= "======================================================================\n";
    $PAR{routine_message} .= "Swarm directories scanned: ".scalar(keys %{$NEWDATA})."\n";
    $PAR{routine_message} .= "Swarm directories deleted: ".scalar(@victims)."\n";
    $PAR{routine_message} .= "======================================================================\n";
  }
}
#=================================================================================================
sub purge_swarms
{
  $PAR{stopwatch} = 0; # reset the stopwatch
# Choose our victims
  my @sqlarray;
  foreach my $t (keys %{$NEWDATA}) {
    next if ($OPT{user} && ($NEWDATA->{$t}{user} ne $OPT{user}));
    next if ($OPT{jobid} && ($NEWDATA->{$t}{jobid} ne $OPT{jobid}));
    next if ($OPT{tempdir} && ($t ne $OPT{tempdir}));
    next unless _purgeable($t);
    if (defined $OLDDATA->{$t}{id}) {
      push @sqlarray,"DELETE FROM `$PAR{swarms_table}` WHERE `id`=$OLDDATA->{$t}{id}";
    }
    else {
      push @sqlarray,"DELETE FROM `$PAR{swarms_table}` WHERE `tempname`='$t'";
    }
  }
  if (@sqlarray) {
    _do_mysql(@sqlarray);
  }
  elsif ($OPT{verbose} > 1) {
    add_to_message("Nothing to purge");
  }
  add_to_message((sprintf "purge_swarms = %.2f",$PAR{stopwatch}),1) if $OPT{timing};
}
#=================================================================================================
sub _deleteable
{
  my $t = shift;
  if (defined $NEWDATA->{$t}{delete_time}) {
    return 0;
  } # already deleted
  if ((defined $NEWDATA->{$t}{end_time}) && ((($PAR{now}-$NEWDATA->{$t}{end_time})/86400)>$PAR{'delete-age'})) {
    return 1;
  } # finished
  if ((not defined $NEWDATA->{$t}{active}) && ((($PAR{now}-$NEWDATA->{$t}{create_time})/86400)>$PAR{'delete-age'})) {
    return 1;
  } # unknown
  return 0;
}
#=================================================================================================
sub _purgeable
{
  my $t = shift;
  return 1 if ((defined $NEWDATA->{$t}{delete_time}) && ((($PAR{now}-$NEWDATA->{$t}{delete_time})/86400)>$PAR{'purge-age'}));
  return;
}
#=================================================================================================
sub printSwarmUsage
{
  my $cat = HPCNIH::Staff::MySQL::Catalog->new(catalog=>"quota_spin1");
  my $x = $cat->get_current(entity=>"/spin1/swarm");
  my $y = $x->{'/spin1/swarm'};
  my $string = sprintf("/swarm usage: %6.2f GB (%4.1f%%), %7d files (%4.1f%%)\n",
      ( $y->{dusage}/1024/1024 ),
      ( ($y->{dusage}/$y->{dquota})*100 ),
      ( $y->{fusage} ),
      ( ($y->{fusage}/$y->{fquota})*100 ),
  );
  return $string;
}
#=================================================================================================
sub print_to_logfile
{
  my $message = shift;
  return unless $message;
  chomp($message);
  if ($OPT{'dry-run'}) {
    my $date = strftime("%FT%T", (localtime(time))[0 .. 5]);
    foreach my $m (split /\n/,$message) {
      print "$date\t$m\n";
    }
  }
  else {
    my $LOGFILE = FileHandle->new($PAR{logfile},">>");
    my $date = strftime("%FT%T", (localtime(time))[0 .. 5]);
    foreach my $m (split /\n/,$message) {
      print $LOGFILE "$date $m\n";
    }
    $LOGFILE->flush;
    undef $LOGFILE;
  }
}
#=================================================================================================
sub printOptions
{
  my $msg = shift;
  warn "\n$msg\n" if $msg;

  print STDERR <<EOF;

Usage: $0 [ options ]

Description:

  Display the status of swarms, and possibly delete swarm directories

Options:

  --update             update the swarms table
  --delete             update and then delete expired swarm directories that
                       have ended at least $PAR{'delete-age'} days ago
  --delete-age         days past ending for deletion (default = $PAR{'delete-age'})
  --purge              remove rows in the swarms table that have been 
                       deleted at least $PAR{'purge-age'} days ago
  --routine            enables --update, --delete, and --purge and simplifies
                       output

  --timeout <seconds>  give a timeout for the script

Filters:

  -u/--user            filter for those swarms owned by this user
  -j/--jobid           filter for the swarm with this jobid
  -t/--tempdir         filter for the tempdir created for swarm

Display modifiers:

  --show-sql           show sql statements
  --human              show times as human readable
  --timing             print timing information

Reporting:

  --email <emails>     comma-delimited list of email addresses to receive
                       reports.  If --dry-run is given, the emails will NOT be
                       sent.
  --bcc <emails>       same as --email, except as bcc

Mode modifiers:

  -h, --help           print options list
  -d, --dry-run        don't make any changes, just show wnat might happen
  -v, --verbose        increase verbosity level (0,1,2,3)

Last modification date: 03 Jul 2018 (David Hoover)

EOF
  
  exit;
}
#=================================================================================================