# -*-perl-*-

# $Id: Atom.plm,v 1.33 2011/08/30 19:59:45 koczan Exp $

# This module handles the Atom data types and methods

# $Log: Atom.plm,v $
# Revision 1.33  2011/08/30 19:59:45  koczan
# System calls shouldn't be PATH-dependent
#
# Revision 1.32  2011/03/31 23:09:07  koczan
# lastBu and lastBuLowerLevel were horrendously wasteful.
# Have them use their own queries
#
# Revision 1.31  2010/02/12 22:12:25  koczan
# reportFailed now always sends a message
#
# Revision 1.30  2008/01/31 17:35:17  thomas
# Don't need sushiremote_wrapper, just sushiremote
#
# Revision 1.29  2007/02/06 16:45:46  thomas
# printf gets confused on BIG numbers
#
# Revision 1.28  2005/11/01 22:46:04  thomas
# add func readHistoryOffline for history and find commands
#
# Revision 1.27  2005/10/03 19:47:44  thomas
# Add host_status support for atom relation.
#
# Revision 1.26  2005/08/03 16:25:50  thomas
# don't schedule lost+found directories with get_fanout()
#
# Revision 1.25  2005/06/24 21:07:09  thomas
# Add timed history read support
#
# Revision 1.24  2005/03/03 17:23:25  thomas
# Add verbiage to failed atom email re: status Setup or Repair
#
# Revision 1.23  2005/03/02 21:40:10  thomas
# finish transaction in atom->history
#
# Revision 1.22  2005/02/08 21:13:20  thomas
# new version of ssh
#
# Revision 1.21  2004/09/07 19:14:01  thomas
# misc fixes for fanout hack.
# Backup all directories, even ones beginning with '.'
#
# Revision 1.20  2004/04/26 14:59:41  thomas
# add fanout for new mail server
#
# Revision 1.19  2004/04/14 15:30:16  thomas
# fail an atom of atoms
#
# Revision 1.18  2004/03/11 21:58:43  thomas
# Clean up ssh calls to use /s/sushi-0.3.19/depend/bin/ssh
#
# Revision 1.17  2004/03/04 20:40:25  thomas
# [CSL #180910]
#
# Revision 1.16  2004/03/02 20:36:42  thomas
# add atom id to failed atoms req
#
# Revision 1.15  2004/02/24 15:49:21  thomas
# First build for postgres 7.4.1
#
# Revision 1.14  2003/06/09 14:45:11  thomas
# schema version of database
#
# Revision 1.13  2003/01/23 15:15:11  thomas
# Add daily req
#
# Revision 1.12  2002/10/31 21:06:29  thomas
# Show atom log at restore time.
#
# Revision 1.11  2002/10/30 20:11:37  thomas
# *** empty log message ***
#
# Revision 1.10  2002/10/30 19:41:49  thomas
# - Use gmtime, not ctime when converting to db values
# - Add errors for HostAtom checkDump
#
# Revision 1.9  2002/10/07 19:01:21  thomas
# sushi-0.3.3
#
# Revision 1.8  2002/09/23 22:03:00  thomas
# 0.3.2 to production
#
# Revision 1.7  2002/09/10 20:42:57  thomas
# sushi-0.3.1
#
# Revision 1.6  2002/08/05 22:00:01  thomas
# final 0.2.8 commit
#
# Revision 1.5  2002/07/31 14:46:55  thomas
# --like
#
# Revision 1.4  2002/07/23 16:02:27  thomas
# version 0.2.5
#
# Revision 1.3  2002/07/10 14:01:28  thomas
# sushi-0.2.4
#
# Revision 1.2  2002/07/08 15:56:32  thomas
# 0.2.3 release
#
# Revision 1.1.1.1  2002/05/21 14:13:27  thomas
# sushi - backup to disk
#

# The following methods are implemented:

# newAtom     Create a new Atom.  (schedule_group, type, name) must be included.

# readAtoms   Read atoms (by schedule_group, type, name, id, or other) from the DB.
#             (creates new atoms).  Returns a ref to a hash of atoms.

# writeAtom  Write an atom out to the database

# limit   Add, update, get, or remove a constraint to an atom.
#                Specify (name, hashref) to add or change a constraint
#                Specify (name) to read a constraint (returns array)
#                Specify (name, undef) to remove a constraint
#                Specify () to get a list of constraint names (array).

# result       Add, update, get or remove a result record to an atom.
#                (Same calls as limit)

package Sushi::Atom;

use strict;
use Time::localtime;
use CSL::Pg;
use DBI;
use FileHandle;
use Data::Dumper;
use IO::Pipe;
use Compress::Zlib;

use base qw(Sushi::Log);

###############################################################################

# The atom constructor.  See also Atom::Read.

sub newAtom {
  my $invocant = shift;
  my %parms = ( @_ );
  my $self = { debug => $::debug };

  # Copy a few things over from our invocant
  foreach ('debug', 'schedule_group', 'type', 'name', 'id', 'level',
	  'estimated_size', 'estimated_time', 'state', 'data_last_update',
	  'execute_before', 'execute_after', 'host_status') {
    if (ref ($invocant)) {
      defined ($invocant->{$_}) and $self->{$_} = $invocant->{$_};
    }
    defined ($parms{$_}) and $self->{$_} = $parms{$_};
  }
  if (!defined ($self->{schedule_group}) ||
      !defined ($self->{type}) ||
      !defined ($self->{name})) {
    return undef;
  }
  my $class = "Sushi::Atom::$self->{type}";
  bless $self, $class;
  return $self;
}

###############################################################################
# Method to read atoms out of the database.  You must specify at least a
# schedule_group _or_ and atom id.  The return value is an arrayref
# of all matching atoms.

sub readAtoms {
  my $class = shift;
  my $sth = undef;
  my $atom;
  my $res = [];
  my $recs = 0;

  ref $class and return undef;

  my %init = (
	      debug => $::debug,
	      maxrecs => 100000,
	      @_
	     );

  my $log = new Sushi::Log (
			    @_
			   );

  my $query = 'select * from atom where';
  foreach ('schedule_group', 'type', 'name', 'id', 'level',
	   'estimated_size', 'estimated_time', 'state', 'host_status') {
    if (defined $init{$_}) {
      $query .= " $_ = " . $::sudbh->quote($init{$_}) . " and";
    }
  }
  foreach ('data_last_update') {
    if (defined $init{$_}) {
      $query .= " $_ = " . $::sudbh->quote(Sushi::Misc::secs2utc($init{$_})) . " and";
    }
  }
  if (defined $init{execute_before}) {
    $query .= " (execute_before > " . $::sudbh->quote(Sushi::Misc::secs2utc($init{execute_before})) . " or execute_before is null ) and";
  }
  if (defined $init{execute_after}) {
    $query .= " (execute_after < " . $::sudbh->quote(Sushi::Misc::secs2utc($init{execute_after})) . " or execute_after is null ) and";
  }

  $query =~ s/ (and|where)$//;
  $log->log (4, "Atom::readAtoms:dbh $query\n");
  $sth = $::sudbh->prepare ($query);
  if (!defined ($sth)) {
    $log->error ("Atom: failed preparing to read atom $init{id}\n");
    goto fail;
  }
  $sth->execute or goto fail;
  my $row;
  while ($row = $sth->fetchrow_hashref) {
    # Success!
    foreach ('data_last_update', 'execute_before', 'execute_after') {
      defined ($row->{$_}) and
	$row->{$_} = &Sushi::Misc::date2secs ($row->{$_});
    }
    $atom = Sushi::Atom->newAtom (%init, @_, %$row);
    push @$res, $atom;
    # Be done after awhile
    $recs++ > $init{maxrecs} and last;
    # The results and weights data will get added when it is referenced.
  }
  $::sudbh->commit;
  return $res;

 fail:
  $::sudbh->commit;
  return undef;
}

###############################################################################
# Write an Atom to the db.  Both createrow and updaterow are here.

sub writeAtom {
  my $self = shift;
  my ($sth,$cmd, $rows);

  ref($self) or return -1;

  if (!defined ($self->{schedule_group}) ||
      !defined ($self->{type}) ||
      !defined ($self->{name})) {
    $self->error ("Atom::writeAtom: missing required information!\n");
    return 0;
  }
  if (!defined $self->{id}) {
    $self->{host_status}='' unless defined $self->{host_status};
    # Create a record before anything else
    $cmd = "insert into atom (schedule_group, type, name, host_status) values ( " .
      $::sudbh->quote($self->{schedule_group}) . ", " .
	$::sudbh->quote($self->{type}) . ", " .
	  $::sudbh->quote($self->{name}) . ", " .
	    $::sudbh->quote($self->{host_status}) . ")";
    $self->log (4, "Atom::Write:dbh $cmd\n");
    $::sudbh->do ($cmd);
    my $na = Sushi::Atom->readAtoms ( schedule_group => $self->{schedule_group},
				      type  => $self->{type},
				      name  => $self->{name}
				    );
    defined $na or return 0;
    $self->{id} = $na->[0]->{id};
  }
  # Now, prepare the update of the existing record
  $cmd = "";
  foreach ('level', 'estimated_size', 'estimated_time') {
    if (defined $self->{$_}) {
      $cmd .= "$_ = $self->{$_}, ";
    }
  }
  foreach ('state', 'priority') {
    if (defined $self->{$_}) {
      $cmd .= "$_ = " . $::sudbh->quote ($self->{$_}) . ", ";
    }
  }

  foreach ('data_last_update', 'execute_before', 'execute_after') {
    if (defined $self->{$_}) {
      $cmd .= "$_ = " . $::sudbh->quote (Sushi::Misc::secs2utc ($self->{$_})) . ", ";
    }
  }
  # Remove the last separator
  $cmd =~ s/, $//;
  $cmd = "update atom set " . $cmd . " where id = $self->{id}";
  $self->log (4, "Atom::Write:dbh $cmd\n");
  $rows = $::sudbh->do ($cmd);
  $self->writeWeights();
  $self->writeResults();
  $::sudbh->commit();
  return 1;
}

###############################################################################
#
# Delete atoms from the database.
#
# When called as a class method, delete all atoms with the passed 
# characteristics.  When called as an instance method, delete the passed
# record.

sub deleteAtoms {

  my $invocant = shift;
  my $self;

  if (ref($invocant)) {
    $self = $invocant;
  } else {
    $self = {
	     debug => $main::debug,
	     @_
	    };
    bless($self);
  }

  if ((!defined ($self->{schedule_group}) || $self->{schedule_group} eq "")
      && !defined ($self->{id})) {
    $self->error ("Atom::deleteAtoms: no group or id specified!\n");
    return 0;
  }
  my ($cmd, $rows);

  if (defined $self->{id}) {
    $cmd = "delete from atom where id = $self->{id}";
  } else {
    $cmd = "delete from atom where schedule_group = " .
      $::sudbh->quote ($self->{schedule_group}) . " and ";
    foreach ("type", "name") {
      defined $self->{$_} and
	$cmd .= "$_ = " . $::sudbh->quote ($self->{$_}) . " and ";
    }
    foreach ("level") {
      defined $self->{$_} and
	$cmd .= "$_ = $self->{$_} and ";
    }
    $cmd =~ s/and $//;
  }
  $self->log (4, "Atom::delete:dbh $cmd\n");
  $rows = $::sudbh->do ($cmd);
  $::sudbh->commit();
  # (The constraint and result records are automatically deleted)
  return $rows;
}



###############################################################################
###############################################################################
# Weights-related functions

sub readWeights {
  my $self = shift;

  ref($self) or return 0;
  defined $self->{id} or return 0;

  defined $self->{weights} or $self->{weights} = [];

  my $query = "select * from atom_weights where id=$self->{id}";
  $self->log (4, "Atom::readWeights:dbh $query\n");
  my $sth = $::sudbh->prepare ($query);
  if (!defined ($sth)) {
    $self->error ("Atom: failed preparing to read constraints $self->{id}\n");
    return -1;
  }
  $sth->execute;
  while (my $row = $sth->fetchrow_hashref) {
    push @{$self->{weights}}, $row;
  }
  return 1;
}

###############################################################################
# Write constraint (weight) records to the db

sub writeWeights {
  my $self = shift;

  my ($cmd, $rows);
  ref($self) or return 0;
  defined ($self->{id}) or return 0;
  defined $self->{weights} or return 1;

  foreach my $weight (@{$self->{weights}}) {
    if (defined ($weight->{changed}) || defined ($weight->{deleted})) {
      $cmd = "delete from atom_weights where id = $self->{id} and resource_type = " . $::sudbh->quote ($weight->{resource_type}) . " and resource_alias = " . $::sudbh->quote ($weight->{resource_alias});
      $self->log (4, "Atom::writeWeights:deleteloop $cmd\n");
      $rows = $::sudbh->do ($cmd);
    }

    defined $weight->{changed} or next;

    $cmd = "insert into atom_weights ( id, resource_type, resource_alias, resource_name, weight ) values (" .
      $self->{id} . ", " .
	$::sudbh->quote ($weight->{resource_type}) . ", " .
	  $::sudbh->quote ($weight->{resource_alias}) . ", " .
	    $::sudbh->quote ($weight->{resource_name}) . ", " .
	      $weight->{weight} . " )";
    $self->log (4, "Atom::writeWeights:dbh $cmd\n");
    $rows = $::sudbh->do ($cmd);
  }
  return 1;
}


###############################################################################
# Return a copy of all the weights records for this atom

sub getWeights {
  my $self = shift;
  my $key = shift;

  defined $self->{weights} or $self->readWeights;

  my $aref = [];
  foreach my $res (@{$self->{weights}}) {
    defined ($res->{deleted}) or push @$aref, {%$res};
  }
  return $aref;
}

###############################################################################
# Set the hash of the passed index

sub setWeight {
  my $self = shift;
  my $hash = shift;

  defined $self->{weights} or $self->readWeights;

  delete $hash->{deleted};
  $hash->{changed} = 1;

  foreach my $res (@{$self->{weights}}) {
    if ($res->{resource_type} eq $hash->{resource_type} &&
	$res->{resource_alias} eq $hash->{resource_alias} ) {
      $res = $hash;
      return 1;
    }
  }
  push @{$self->{weights}}, $hash;
  return 1;
}

###############################################################################
# Delete a hashed item from an atom.

sub delWeight {
  my $self = shift;
  my $hash = shift;

  defined $self->{weights} or $self->readWeights;

  foreach my $weight (@{$self->{weights}}) {
    if ($weight->{resource_type} eq $hash->{resource_type} &&
	$weight->{resource_alias} eq $hash->{resource_alias} ) {
      $_->{deleted} = 1;
      delete $_->{changed};
      return 1;
    }
  }
  return 0;
}



###############################################################################
###############################################################################
# Result record functions

###############################################################################
# Return a copy of all the current result records

sub getResults {
  my $self = shift;

  defined $self->{results} or $self->readResults;

  my $aref = [];
  foreach my $res (@{$self->{results}}) {
    defined ($res->{deleted}) or push @$aref, {%$res};
  }
  return $aref;
}

###############################################################################

sub setResult {
  my $self = shift;
  my $hash = shift;

  defined $self->{results} or $self->readResults;

  delete $hash->{deleted};
  $hash->{changed} = 1;

  foreach my $res (@{$self->{results}}) {
    if ($res->{timestamp} == $hash->{timestamp}) {
      $res = $hash;
      return 1;
    }
  }
  push @{$self->{results}}, $hash;
  return 1;
}


###############################################################################

sub delResult {
  my $self = shift;
  my $hash = shift;

  defined $self->{results} or $self->readResults;

  foreach (@{$self->{results}}) {
    if ($_->{timestamp} == $hash->{timestamp}) {
      $_->{deleted} = 1;
      delete $_->{changed};
      return 1;
    }
  }
  return 0;
}

###############################################################################

sub readResults {
  my $self = shift;

  ref($self) or return 0;

  defined $self->{id} or return 0;

  defined $self->{results} or $self->{results} = [];

  my $query = "select * from results where id=$self->{id}";
  $self->log (4, "Atom::ReadResults:dbh $query\n");
  my $sth = $::sudbh->prepare ($query);
  if (!defined ($sth)) {
    $self->error ("Atom: failed preparing to read constraints $self->{id}\n");
    return 0;
  }
  $sth->execute;
  while (my $row = $sth->fetchrow_hashref) {
    if (defined $row->{timestamp}) {
      $row->{timestamp} = Sushi::Misc::date2secs ($row->{timestamp});
      push @{$self->{results}}, $row;
    }
  }
  return 1;
}

###############################################################################
# WriteResults -- write result records to the db

sub writeResults {
  my $self = shift;

  my ($cmd, $rows);
  ref($self) or return 0;
  defined ($self->{id}) or return 0;
  defined ($self->{results}) or return 1;

  foreach my $res (@{$self->{results}}) {
    if (defined ($res->{changed}) || defined ($res->{deleted})) {
      # Delete any records that are there now.
      $cmd = "delete from results where id = $self->{id} and timestamp = " .
	$::sudbh->quote (Sushi::Misc::secs2utc ($res->{timestamp}));
      $self->log (4, "Atom::writeResults:deleteloop $cmd\n");
      $rows = $::sudbh->do ($cmd);
    }

    defined $res->{changed} or next;

    my (@tags, @vals);
    foreach ('id', 'result', 'runtime', 'raw_size', 'comp_size', 'afs_volume', 'afs_path', 'checksum', 'dump_stderr', 'exec_host', 'exec_pid', 'on_disk') {
      if (defined ($res->{$_})) {
	push @tags, $_;
	push @vals, $::sudbh->quote ($res->{$_});
      }
    }
    foreach ('timestamp') {
      if (defined ($res->{$_})) {
	push @tags, $_;
	push @vals, $::sudbh->quote (Sushi::Misc::secs2utc($res->{$_}));
	}
    }
    $cmd = "insert into results ( " . join (", ", @tags) . ") values ( " . join (",", @vals) . ")";
    $self->log (4, "Atom::writeResults:dbh $cmd\n");
    $rows = $::sudbh->do ($cmd);
  }
  return 1;
}

###############################################################################
# Read the history for this atom from the database.

sub readHistory {
  my $self = shift;
  my $since = shift;

  ref($self) or return 0;

  my ($query, $sth, $row);
  $query = "select schedule_group, type, name, atom.id, timestamp, result, runtime, raw_size, comp_size, afs_volume, afs_path, checksum, exec_host, exec_pid, level, on_disk from atom atom, results res where " .
	"type = " . $::sudbh->quote($self->{type}) . " and " .
	  "name = " . $::sudbh->quote($self->{name}) . " and " .

	    "atom.id = res.id";

  if (defined $since) {
    $query .= " and timestamp > ".
      $::sudbh->quote(Sushi::Misc::secs2utc($since));
  }

  $self->log (4, "Atom::readHistory:dbh $query\n");
  $sth = $::sudbh->prepare ($query);
  if (!defined ($sth)) {
    $self->error ("Atom: failed preparing to read atom $self->{id}\n");
    return 0;
  }
  $sth->execute;
  defined $self->{hist} or $self->{hist} = [];
  while (defined ($row = $sth->fetchrow_hashref)) {
    if (defined $row->{timestamp}) {
      $row->{timestamp} = Sushi::Misc::date2secs ($row->{timestamp});
      push @{$self->{hist}}, {%$row};
    }
  }

  $sth->finish;
  return 1;
}

###############################################################################
# Return the history records.

sub history {
  my $self = shift;
  my $since = shift;

  ref ($self) or return 0;

  if (!defined $self->{hist}) {
    if (defined $since) {
      $self->readHistory ($since);
    } else {
      $self->readHistory;
    }
  }
  $self->{hist};
}

###############################################################################
# This is for convenience.  Return the most recent complete result for
# the atom at the specified level.

sub lastBu {
  my $self = shift;
  my $level = shift;

  ref($self) or return undef;
  defined ($level) or return undef;

  if (!defined $self->{hist}) {
    # History was taking too much memory, let's use a generated query instead
    my ($query, $sth, $row);
    $query = "select schedule_group, type, name, atom.id, timestamp, result, runtime, raw_size, comp_size, afs_volume, afs_path, checksum, exec_host, exec_pid, level, on_disk from atom atom, results res where " .
	"atom.type = " . $::sudbh->quote($self->{type}) . " and " .
	"atom.name = " . $::sudbh->quote($self->{name}) . " and " .
	"atom.level = $level and " .
	"res.result = 'Success' and res.on_disk = 'Yes' and atom.id = res.id ".
	"order by res.timestamp desc limit 1";
    $self->log (4, "Atom::lastBu:dbh $query\n");
    $sth = $::sudbh->prepare ($query);
    if (!defined ($sth)) {
      $self->error ("Atom::lastBu failed preparing to read atom $self->{id}\n");
      return 0;
    }
    $sth->execute;
    $row = $sth->fetchrow_hashref;
    $sth->finish;
    if (defined $row) {
      $row->{timestamp} = Sushi::Misc::date2secs ($row->{timestamp});
      return $row;
    } else {
      return undef;
    }
  } else {
    my $rec = { timestamp => 0 };

    foreach (@{$self->{hist}}) {
      $rec = $_ if (($_->{result} eq 'Success') && ($_->{on_disk} eq 'Yes') && ($_->{level} == $level) && ($_->{timestamp} > $rec->{timestamp}));
    }

    return $rec if (defined $rec->{result});
    return undef;
  }

  # failsafe
  return undef;
}

###############################################################################
# Return the most recent successful backup record at a lower level

sub lastBuLowerLevel {
  my $self = shift;

  ref($self) or return undef;

  my $level = $self->{level};

  # History was taking too much memory, let's use a generated query instead
  if (!defined $self->{hist}) {
    my ($query, $sth, $row);
    $query = "select schedule_group, type, name, atom.id, timestamp, result, runtime, raw_size, comp_size, afs_volume, afs_path, checksum, exec_host, exec_pid, level, on_disk from atom atom, results res where " .
	"atom.type = " . $::sudbh->quote($self->{type}) . " and " .
	"atom.name = " . $::sudbh->quote($self->{name}) . " and " .
	"atom.level < $level and " .
	"res.result = 'Success' and res.on_disk = 'Yes' and atom.id = res.id ".
	"order by res.timestamp desc limit 1";
    $self->log (4, "Atom::lastBuLowerLevel:dbh $query\n");
    $sth = $::sudbh->prepare ($query);
    if (!defined ($sth)) {
      $self->error ("Atom::lastBuLowerLevel failed preparing to read atom $self->{id}\n");
      return 0;
    }
    $sth->execute;
    $row = $sth->fetchrow_hashref;
    $sth->finish;

    if (defined $row) {
      $row->{timestamp} = Sushi::Misc::date2secs ($row->{timestamp});
      return $row;
    } else {
      return undef;
    }
  } else {
    my $rec = { timestamp => 0 };

    foreach (@{$self->{hist}}) {
      $rec = $_ if (($_->{result} eq 'Success') && ($_->{on_disk} eq 'Yes') && ($_->{level} < $level) && ($_->{timestamp} > $rec->{timestamp}));
    }

    return $rec if (defined $rec->{result});
    return undef;
  }

  # failsafe
  return undef;
}

###############################################################################
# Create a DESTROY method for profiling
# Try to increase the speed of destroying these objects

sub DESTROY {
  my $self = shift;

  # Destroy the sub-bits first
  delete $self->{hist};
  delete $self->{weights};
  delete $self->{results};
  return;
}

###############################################################################

# These are the methods that just set (or report) a variable -- use autoload

sub data_last_update;
sub debug;
sub epsize;
sub estimated_size;
sub estimated_time;
sub execute_after;
sub id;
sub schedule_group;
sub interval;
sub level;
sub name;
sub now;
sub state;
sub type;

sub AUTOLOAD {
  my $self = shift;
  my $value = shift;

  our $AUTOLOAD;
  my $var;

  ref($self) or return undef;

  $AUTOLOAD =~ /::DESTROY$/ and return;

  if ($AUTOLOAD =~ /::(name|debug|dest|destroot|epsize|estimated_size|estimated_time|execute_after|execute_before|id|schedule_group|interval|level|data_last_update|now|state|type)$/) {
    $var = $1;
  } else {
    $main::log->error ("Unimplemented method $AUTOLOAD referenced\n");
    return undef;
  }

  if ($var !~ /^(schedule_group|type|name|id)$/) {
    defined ($value) and $self->{$var} = $value;
  }
  defined ($self->{$var}) or return undef;
  return $self->{$var};
}

###############################################################################
# Give an atom with a known level, fill in esize and etime

sub estimateSize {
  my $self = shift;

  ref($self) or return 0;

  defined $self->{level} or return 0;

  defined $self->{hist} or $self->readHistory;

  if (defined ($self->{epsize}) && ($self->{level} == 0)) {
    $self->{estimated_size} = $self->{epsize};
  }

  my $rc = $self->lastBu ($self->{level});
  my $r0 = $self->lastBu (0);
  if (defined $rc) {
    # Use the most recent bu at the current level
    defined ($self->{estimated_size}) or $self->{estimated_size} = $rc->{raw_size};
    if ($rc->{raw_size}>0) {
      $self->{estimated_time} = int($self->{estimated_size} * $rc->{runtime} / $rc->{raw_size});
    } else {
      $self->{estimated_time} = int($self->{estimated_size} /(100*1024));
    }
  } else {
    # No previous dump at the current level.
    if ($self->{level} == 0 ) {
      defined ($self->{estimated_size}) or $self->{estimated_size} = 100*1024*1024;   # 100 Meg
    } else {
      if (!defined $self->{estimated_size}) {
	if (defined $r0) {
	  $self->{estimated_size} = $r0->{raw_size} * .05;   # Inc at 5% of epoch
	} else {
	  $self->{estimated_size} = 50*1024*1024;   # ~100 Meg * .05
	}
      }
    }
    # WAG for rate -- 100k/second
    $self->{estimated_time} = $self->{estimated_size}/102400;
  }
  $self->{estimated_time}>1 or $self->{estimated_time} = 1;

  return 1;
}


###############################################################################

sub getDest {
  my $self = shift;
  my $timestamp = shift;

  ref ($self) or return undef;
  defined ($timestamp) or return undef;

  (my $name = $self->{name}) =~ s/\//_/g;
  my $timestring = sprintf "%s:%s:%s/%s",
     $self->{schedule_group}, $self->{type}, $name,
       Sushi::Misc::shortTime ($timestamp);
}

###############################################################################

sub twoLetterHash
{
  my $self = shift;
  my $name = $self->{name};

  my ($c0, $c1) = (0,0);
  my ($a1, $b1);

  while ($name ne '') {
    $name =~ /(.)(.*)/;
    $b1 = ord $1; $name = $2;
    $c0 = ($c0 + $b1 ) & 0xFF;
    $c1 = ($c1 + $c0) & 0xFF;
  }
  $a1 = ord 'a';
  $c0 = chr(($c0 & 0xF) + $a1);
  $c1 = chr(($c1 & 0xF) + $a1);
  $c0 . $c1;
}

###############################################################################
# Return the command used to execute the child atom.

sub getChildCmd {
  my $self = shift;
  my $timestamp = shift;

  my @argv = ();

  if (defined (my $exec_host = $self->executeHost)) {
    @argv = ('/s/sushi-0.3.19/depend/bin/ssh', '-2', '-x', '-n');
    if (defined (my $exec_user = $self->executeUser)) {
      push @argv, '-o', 'gssapiauthentication=no', '-l', $exec_user;
    }
    push @argv, $exec_host;
  }

  $::debug>3 and push @argv, "time";

  push @argv, "/s/sushi-0.3.19/bin/sushiremote", '-d', $::debug, $self->{id}, $timestamp;

  return @argv;
}

###############################################################################
# By default, guess that we don't need a host ACL.

sub needHostAcl { undef; }

###############################################################################
# Make sure the place to store the atom exists.  Call the AfsBits routine.

sub createDest {
  return Sushi::AfsBits::createDest ( @_ );
}

###############################################################################
# Return the common name for an atom

sub commonName {
  my $self = shift;

  (defined ($self->{schedule_group}) &&
   defined ($self->{type}) &&
   defined ($self->{name})) or return "(undef)";

  defined ($self->{level}) and return "$self->{schedule_group}:$self->{type}:$self->{name}($self->{level})";
  "$self->{schedule_group}:$self->{type}:$self->{name}(?)";
}

###############################################################################
# Fail some atoms

sub failAtoms {
  my $self = shift;
  my $args = shift;
  my $opts = shift;

  if (!$opts->{force}) {
    print STDERR "--force not specified; atom will not be modified...\n";
  }

  ref ($self) and return $self->failAtom ($opts);

  my ($row, $atom);
  my $rows=0;
  defined ($args->{schedule_group}) or $args->{schedule_group} = Sushi::Misc::dailyGroup();
  foreach $row (@{Sushi::GenDB::fetch (
				       $args,
				       $opts,
				       $::table_defs->{atom})}) {
    $atom = Sushi::Atom::newAtom ($row);
    $rows += $atom->failAtom ($opts);
  }
  print "$rows atoms modified.\n";
  return $rows;
}

# Fail one loaded atom
sub failAtom {
  my $self = shift;
  my $opts = shift;

  ref ($self) && defined ($self->{id}) or die "Atom::failAtom: no id!\n";
  my $rows = 0;
  my $row;
  foreach $row (@{Sushi::GenDB::fetch (
				       {
					id => $self->{id},
					result => 'Success',
				       },
				       $opts,
				       $::table_defs->{results})}) {
    print "Found successful result for " . $self->commonName . " dated " . ctime ($row->{timestamp}) . ".\n";
    $opts->{force} or next;
    print "   Fail(y/n)?";
    my $answer = <STDIN>;
    if ($answer =~ /^\s*[Yy]/) {
      # OK.. Do it
      my $updaterows = Sushi::GenDB::update
	(
	 {
	  id        => $self->{id},
	  timestamp => $row->{timestamp},
	  result    => 'Success',
	 },
	 {},
	 {
	  result   => 'Fail',
	 },
	 $::table_defs->{results}
	);
      print "OK.  Modified $updaterows rows.\n";
      $rows++;
    }
  }
  return $rows;
}

###############################################################################
# Display atoms in human-readable format.

sub displayAtoms {
  my $self = shift;
  my $args = shift;
  my $opts = shift;

  my $newopts = {%$opts};
  exists ($newopts->{fd}) or $newopts->{fd} = undef;

  ref ($self) and return $self->displayAtom ($opts);

  my ($row, $atom);
  my $rows=0;
  foreach $row (@{Sushi::GenDB::fetch (
				       $args,
				       $opts,
				       $::table_defs->{atom})}) {
    $atom = Sushi::Atom::newAtom ($row);
    $atom->displayAtom ($newopts);
    $newopts->{print_headings} = 0;
    $rows++;
  }
  Sushi::GenDB::displayTrailer (
				$opts,
				$::table_defs->{atom},
				$rows
			       );
  return $rows;
}

###############################################################################
# Display one atom in human-readable format

sub displayAtom {
  my $self = shift;
  my $opts = shift;

  my $line;
  Sushi::GenDB::displayHeader (
			       $opts,
			       $::table_defs->{atom},
			      );

  Sushi::GenDB::displayRow (
			    $opts,
			    $::table_defs->{atom},
			    $self
			   );

  if ($opts->{print_weights}) {
    Sushi::GenDB::displayHeader (
			       $opts,
			       $::table_defs->{atom_weights},
			      );
    foreach my $weight (sort @{$self->getWeights}) {
      Sushi::GenDB::displayRow (
				$opts,
				$::table_defs->{atom_weights},
				$weight
			       );
    }
  }
  if ($opts->{print_results} || $opts->{print_output}) {
    Sushi::Misc::printToFile ($opts->{fd}, "\n  Results:\n");
    foreach my $result (sort {$a->{timestamp} <=> $b->{timestamp}} @{$self->getResults}) {
      $line = sprintf "  Time:            %-24s    (%d)\n" .
	"  Result:          %-7s\n" .
	  "  Run Time:        %-7d\n".
	    "  Raw Size:        $result->{raw_size}\n" .
	      "  Compressed Size: $result->{comp_size}\n" .
		"  On disk?:        %-10s\n".
		  "  Afs Volume:      %-16s\n" .
		    "  Afs path:        %-25s\n".
		      "  Execute Host:    %-25s\n".
			"  Execute Pid:     %-8d\n\n",
			  ctime ($result->{timestamp}), $result->{timestamp},
			    $result->{result},
			      $result->{runtime},
				    $result->{on_disk},
				      $result->{afs_volume},
					$result->{afs_path},
					  $result->{exec_host},
					    $result->{exec_pid};
      Sushi::Misc::printToFile ($opts->{fd}, $line);
      if ($opts->{print_output}) {
	Sushi::Misc::printToFile ($opts->{fd}, "(Output follows)\n$result->{dump_stderr}\n(End of Output)\n");
      }
    }
  }
  return 1;
}

###############################################################################
# General signal handling code...
my $signalled = 0;
sub siggobyebyehandler {
  my $sig = shift;
  print STDERR "Caught SIG$sig: shutting down.\n";
  $signalled = 1;
  return;
}

###############################################################################
# Create a restore stream to the specified file, or STDOUT

sub restore {
  my $self = shift;
  my $args = shift;
  my $opts = shift;

  my $timestamp = Sushi::Misc::date2secs ($opts->{timestamp});
  defined ($opts->{force}) or $opts->{force} = 0;
  my $atoms = Sushi::Atom->readAtoms (%$args);
  my $atom = shift @$atoms;
  if (!defined ($atom)) {
    $::log->error ("Could not find atom!\n");
    return 0;
  }
  my $commonName = $atom->commonName;
  # Find the result record
  foreach my $result (@{$atom->getResults}) {

    # Skip record if needed
    if (defined $timestamp) {
      if ($result->{timestamp} != $timestamp) {
	next;
      } elsif (!$opts->{force}) {
	$atom->error ("Requested result result=$result->{result} and --force not specified, quitting.\n");
	return 0;
      }
    } elsif ($result->{result} ne 'Success') {
      next;
    }

    # Either:
    # - There was no timestamp, but this is a successful attempt
    # - We have the timestamp they requested, and --force is there if needed
    # Either way, print a message and continue...
    if ($opts->{force}) {
      $atom->log (1, "Requested atom result=$result->{result}, but --force specified, continuing.\n");
    }
    $atom->log (1, "Using result dated " . ctime ($result->{timestamp}) . "\n");
    # Got the atom -- mount the volume and start the restore
    my $reason = "Restore $commonName:$result->{timestamp}";
    my ($mountpath);
    if (!defined ($mountpath = Sushi::AfsBits::mountVolume ($result->{afs_volume}, $reason))) {
      $atom->error ("Mount afs volume failed for $commonName ($result->{afs_volume})\n");
      return 0;
    }

    my $path = "$mountpath/$result->{afs_path}";
    my $fn = 0;
    my ($buffer, $nr, $gz);
    {
      local $SIG{INT} = \&siggobyebyehandler;
      local $SIG{QUIT} = \&siggobyebyehandler;
      local $SIG{HUP}  = \&siggobyebyehandler;
      local $SIG{PIPE} = \&siggobyebyehandler;

      while (defined ($gz = gzopen("$path/data$fn.gz", "r")) && !$signalled) {
	while (($nr = $gz->gzread($buffer))>0 && !$signalled) {
	  print $buffer;
	}
	$fn++;
      }
    }
    if (!Sushi::AfsBits::unMountVolume ($result->{afs_volume}, $reason)) {
      $atom->error ("Unmount $result->{afs_volume} failed for $commonName.\n");
    }
    if (!$fn) {
      $atom->error ("No data files found.  This probably isn't good(tm).\n");
      return 0;
    }
    print STDERR "\nThe result log from the restored atom follows.\n" .
      "Please review this, as it may contain information about files\n" .
	"whose backup may be incomplete.\n\n$result->{dump_stderr}\n";
    return 1;
  }
}

###############################################################################
# Print the history records for an atom

sub displayHistory {
  my $self = shift;
  my $args = shift;
  my $opts = shift;

  my $line;
  exists ($opts->{fd}) or $opts->{fd} = undef;
  defined ($opts->{print_failed}) or $opts->{print_failed} = 0;
  defined ($opts->{print_stats}) or $opts->{print_stats} = 0;

  my $history = $self->readHistoryOffline ($args, $opts);

  Sushi::GenDB::displayHeader (
			       $opts,
			       $::table_defs->{atom_history}
			      );

  my $rows = 0;
  my $result;
  if (defined ($history) and ref($history)) {
    foreach $result (sort {$a->{timestamp} <=> $b->{timestamp} or $a->{write_time} <=> $b->{write_time}} @$history) {
      if ($result->{result} ne 'Success' && !$opts->{print_failed}) {
	next;
      }
      if ($opts->{print_stats}) {
	$result->{compression} = ($result->{comp_size} ? ($result->{raw_size}/$result->{comp_size}):0);
	
	$result->{raw_kb_sec} = ($result->{runtime} ? ($result->{raw_size}/$result->{runtime}):0);
	$result->{comp_kb_sec} = ($result->{runtime} ? ($result->{comp_size}/$result->{runtime}):0);
      }

      Sushi::GenDB::displayRow (
				$opts,
				$::table_defs->{atom_history},
				$result
			       );
      $rows++;
    }
  }

  Sushi::GenDB::displayTrailer (
				$opts,
				$::table_defs->{atom_history},
				$rows
			       );
  return $rows;
}

###############################################################################
# Read the history of an atom, including any offline records

sub readHistoryOffline {
  my $self = shift;
  my $args = shift;
  my $opts = shift;

  my $atoms = Sushi::Atom->readAtoms (%$args);
  my $atom = shift @$atoms;
  if (!defined ($atom)) {
    $::log->error ("Could not read requested atom!\n");
    return 0;
  }

  my ($offline, $result);
  my $history = [];
  foreach $result (@{$atom->history ($opts->{since})}) {
    push @$history, $result;
  }
  if ($opts->{print_offline}) {
    foreach $offline (@{Sushi::GenDB::fetch (
					     {
					      type => $atom->{type},
					      name => $atom->{name}
					     },
					     {},
					     $::table_defs->{view_results_offline}
					    )}) {
      # Find the result record
      foreach $result (@$history) {
	if ($result->{timestamp} == $offline->{timestamp}) {
	  if (defined $result->{storage_id}) {
	    # Oh no! Create a duplicate result record...
	    my $newres = {%$result};
	    $newres->{storage_id} = $offline->{storage_id};
	    $newres->{write_time} = $offline->{write_time};
	    $newres->{saveset_name} = $offline->{saveset_name};
	    $newres->{disposition} = $offline->{disposition};
	    push @$history, $newres;
	    last;
	  } else {
	    # Put in the information
	    $result->{storage_id} = $offline->{storage_id};
	    $result->{write_time} = $offline->{write_time};
	    $result->{saveset_name} = $offline->{saveset_name};
	    $result->{disposition} = $offline->{disposition};
	    last;
	  }
	}
      }
    }
  }
  $history;
}


###############################################################################

sub failActive {
  my $self = shift;
  my $args = shift;

  my $rows = 0;
  my $list = Sushi::Atom->readAtoms (%$args, state => 'Active');
  my $atom;
  foreach $atom (@$list) {
    if ($atom->state ne 'Active') {
      $atom->error ("Atom::failactive: Non-active atom returned!\n");
      next;
    }
    $atom->log (1, "Failing active atom " . $atom->commonName . ".\n");
    $atom->state ('Done');
    $atom->writeAtom;
    $rows++;
  }
  $rows>0 and $::log->log(1, "Failed $rows active atoms\n");
  return $rows;
}


###############################################################################

sub rescheduleFailed {
  my $self = shift;
  my $args = shift;
  my $opts = shift;

  defined ($args->{schedule_group}) or $args->{schedule_group} = Sushi::Misc::dailyGroup();
  defined ($opts->{donothing}) or $opts->{donothing} = $::donothing;
  my $atoms = Sushi::Atom->readAtoms (
				      %$args,
				      schedule_group => $args->{schedule_group},
				      state          => 'Done'
				     );
  my ($atom, $res);
  my $rows = 0;
  foreach $atom (@$atoms) {
    my $success = 0;
    foreach $res (@{$atom->getResults}) {
      if ($res->{result} eq 'Success') {
	$success = 1;
	last;
      }
    }
    if (!$success) {
      if ($opts->{donothing}) {
	$atom->log (1, "Would reschedule " . $atom->commonName . "\n");
      } else {
	$atom->log (1, "Rescheduling " . $atom->commonName . "\n");
	$atom->state ('Scheduled');
	$atom->writeAtom;
	$rows++;
      }
    }
  }
  return $rows;
}

###############################################################################

sub listFailed {
  my $self = shift;
  my $args = shift;
  my $opts = shift;

  my $rows = 0;
  defined ($args->{schedule_group}) or $args->{schedule_group} = Sushi::Misc::dailyGroup();

  my $atoms = Sushi::Atom->readAtoms (
				      schedule_group => $args->{schedule_group},
				      state => 'Done'
				     );

	print <<__Eof__

Host
Status     Type         Name                                    Id        Level  Result   Failures
---------- -----------  --------------------------------------  --------  -----  -------  ---------
__Eof__
	  ;

  foreach my $atom (sort {$a->{host_status} cmp $b->{host_status}
			    or $a->{type} cmp $b->{type}
			      or $a->{name} cmp $b->{name}} @$atoms) {
    my ($display, $res);
    my ($failures, $success) = (0, "Fail");
    if ($opts->{print_failed}) {
      # Show atom if any failure record
      $display = 0;
      foreach $res (@{$atom->getResults}) {
	$res->{result} eq 'Success' and $success = "Success";
	$res->{result} ne 'Success' and $display = 1;
	$res->{result} eq 'Fail'    and $failures++;
      }
    } else {
      # Show atom only if there isn't a success record
      $display = 1;
      foreach $res (@{$atom->getResults}) {
	$res->{result} eq 'Success' and $success = "Success";
	$res->{result} eq 'Success' and $display = 0;
	$res->{result} eq 'Fail'    and $failures++;
      }
    }
    $display or next;
    printf "%-10s %-11s  %-38s  %8d  %5d  %-8s  %8d\n",
      $atom->{host_status}, $atom->{type}, $atom->{name}, $atom->{id},
	$atom->{level}, $success, $failures;
    $rows++
  }
  print "\n($rows rows)\n\n";
  return;
}

###############################################################################

sub reportFailed {
  my $self = shift;
  my $args = shift;
  my $opts = shift;

  my $rows = 0;
  my $mh;

  defined ($args->{schedule_group}) or $args->{schedule_group} = Sushi::Misc::dailyGroup();
  defined ($opts->{mailto}) or $opts->{mailto} = 'lab';
  my $ctime = ctime (time);
  defined ($opts->{subject}) or $opts->{subject} = "sushi failures - $ctime";

  my $atoms = Sushi::Atom->readAtoms (
				      schedule_group => $args->{schedule_group},
				      state => 'Done'
				     );

  my $priority = '';

  # we always want to email, so just start constructing the email...now.
  if ($opts->{nomail}) {
    $mh = *STDOUT;
  } else {
    $mh = new FileHandle "|mail -s '$opts->{subject}' $opts->{mailto}";
    if (!defined $mh) {
      $::log->error ("Ieeee! Can't send mail to $opts->{mailto}\n");
      return -1;
    }
  }
  print $mh <<__Eof__

*Ding*
This is your daily sushi status/failure ticket.

Here is a list of what has failed.  The 'Days Failed'
column lists the number of days each atom has failed,
including today.

Dumper: Please do the following:

* Check to make sure that all atoms got scheduled and, if any they
  have not yet finished, please make sure they will (i.e. run
  'sushi status' and 'sushi showactive').

* Comment this ticket about any "false positives" (computers that are
  expected to have dump failures, such as those that are being
  reinstalled, repaired, etc).

* Read the existing tickets (if any) about any recurring problem
  computers. Take appropriate action (reboot, etc) and comment the
  existing reqs.

* Create new req for any new problems, and take appropriate action
  (reboot, etc).

* Use your judgment and follow up on anything that has been failing
  and has not been acted for a while.

In case of major problems, contact the main sushi admin.

Please check host atoms against their status in the database.
Hosts with status "Repair" or "Setup" may not be expected to dump,
but they will try to anyway.

When everything has finished. Comment this req, resolve it, and put
it in the dumpers queue.

Host                                                                      #
Status     Type         Name                                    Atom ID   Days
---------- -----------  --------------------------------------  --------  ----
__Eof__
	;

  foreach my $atom (sort {$a->{host_status} cmp $b->{host_status}
			    or $a->{type} cmp $b->{type}
			      or $a->{name} cmp $b->{name}} @$atoms) {
    my $res;
    my $display = 1;
    foreach $res (@{$atom->getResults}) {
      $res->{result} eq 'Success' and $display = 0;
    }
    $display or next;

    # Now get the history and count the number of failures
    my ($group, $days) = ('', -1);
    foreach $res (sort {$b->{schedule_group} cmp $a->{schedule_group}} @{$atom->history}) {
      if ($group ne $res->{schedule_group}) {
	$group = $res->{schedule_group};
	$days++;
      }
      ($res->{result} eq 'Success') and last;
    }
    # Leave whitespace between Production, Setup, etc. priorities
    if ($priority ne $atom->{priority}) {
      print $mh "\n" if length ($priority);
      $priority = $atom->{priority};
    }
    printf $mh "%-10s %-11s  %-38s  %8d  %4d\n",
      $atom->{host_status}, $atom->{type}, $atom->{name}, $atom->{id}, $days;
    $rows++
  }

  if (defined $mh) {
    print $mh "\n($rows rows)\n\n";
    undef $mh;
  }
  return $rows;
}

###############################################################################
# By default, return undef (don't run as another user).
# Atom-specific methods may override.

sub executeUser { undef; }

###############################################################################
# By default, we don't need to create pts users for atoms.

sub createPts { undef; }

###############################################################################
# return the specified host's fstab as an array.  This may get overridden
# for some atom types.

my $sshpid;
my $host;

sub getfstab {
  my $self = shift;
  $host = shift;

  my $epsize = {};

  local $SIG{ALRM} = sub {
    my ($sig) = shift;
    print STDERR "Failed to get fstab for $host in 60 seconds\n";
    kill 9, $sshpid;
    return;
  };

  $::log->log (3, "Getting fstab and df -k output for $host...\n");
  alarm 60;
  if (!($sshpid = open (SSH, "/s/sushi-0.3.19/depend/bin/ssh -2 -x -n -o gssapiauthentication=no -l wsbackup $host cat /etc/fstab|"))) {
    alarm 0;
    $::log->error ("Can't ssh to $host to get /etc/fstab!\n");
    return undef;
  }
  my @fstab = <SSH>;
  close SSH;
  alarm 0;
  my $exitcode = $? >> 8;
  ($exitcode > 0) and $::log->error ("Getting fstab for $host exitted $exitcode\n");
  if (!@fstab) {
    $::log->error ("Empty fstab returned for $host\n");
    return {};
  }

  my ($line, $mtpt, $dump, $usedk);
  foreach (@fstab) {
    /^\s*\#/ and next;
    (undef, $mtpt, undef, undef, $dump) = split;
    if ($dump) {
      $epsize->{$mtpt} = 100 * 1024 * 1024; # default to 100 Meg
    }
  }

  alarm 60;
  if (!($sshpid = open (SSH, "/s/sushi-0.3.19/depend/bin/ssh -2 -x -n -o gssapiauthentication=no -l wsbackup $host df -k|"))) {
    alarm 0;
    $::log->error ("Can't ssh to $host to get df -k output!\n");
    return undef;
  }
  my @df = <SSH>;
  close SSH;
  alarm 0;
  foreach (@df) {
    (undef, undef, $usedk, undef, undef, $mtpt) = split;
    if (exists $epsize->{$mtpt}) {
      $epsize->{$mtpt} = $usedk*1024;
    }
  }
  return $epsize;
}

###############################################################################

# Return an array of all the sub-directories of a mount point on the
# remote host.

sub getfanout {
  my $self = shift;
  my $host = shift;
  my $dir = shift;

  my $epsize = {};

  # Remove trailing '/'
  $dir =~ s:/+$::;

  local $SIG{ALRM} = sub {
    my ($sig) = shift;
    print STDERR "Failed to read $dir for $host in 60 seconds\n";
    kill 9, $sshpid;
    return;
  };

  $::log->log (3, "Reading directory $dir for $host...\n");
  alarm 60;
  if (!($sshpid = open (SSH, "/s/sushi-0.3.19/depend/bin/ssh -2 -x -n -o gssapiauthentication=no -l wsbackup $host /bin/ls -a $dir|"))) {
    alarm 0;
    $::log->error ("Can't ssh to $host to ls $dir!\n");
    return undef;
  }
  my @contents;
  chomp (@contents = grep (!/^(\.\.?|lost\+found)\/?$/, <SSH>));
  close SSH;
  alarm 0;
  my $exitcode = $? >> 8;
  ($exitcode > 0) and $::log->error ("Reading $dir for $host exitted $exitcode\n");
  if (!@contents) {
    $::log->error ("Empty directory $dir returned for $host\n");
    return {};
  }

  foreach (@contents) {
    $epsize->{"$dir/$_"} = 100 * 1024 * 1024; # default to 100 Meg
  }

  return $epsize;
}

###############################################################################

1;
