Index: /trunk/ippScripts/scripts/rawcheck.pl
===================================================================
--- /trunk/ippScripts/scripts/rawcheck.pl	(revision 36203)
+++ /trunk/ippScripts/scripts/rawcheck.pl	(revision 36203)
@@ -0,0 +1,360 @@
+#!/usr/bin/env perl
+
+use strict;
+use warnings FATAL => qw( all );
+
+use vars qw( $VERSION );
+$VERSION = '0.01';
+
+use Nebulous::Client;
+
+use Getopt::Long qw( GetOptions :config auto_help auto_version );
+use Pod::Usage qw( pod2usage );
+use IPC::Cmd 0.36 qw( can_run run);
+use PS::IPP::Metadata::List qw( parse_md_list );
+use PS::IPP::Config 1.01 qw( :standard );
+
+use Digest::MD5;
+use URI;
+
+my $missing_tools = 0;
+my $regtool  = can_run('regtool') or (warn "Can't find regtool" and $missing_tools = 1);
+
+
+my ($server,$dbname,$exp_id);
+
+$server = $ENV{'NEB_SERVER'} unless $server;
+# $dbname = 'gpc1'; 
+GetOptions(
+    'server|s=s'     => \$server,
+    'dbname'         => \$dbname,
+    'exp_id|x=s'     => \$exp_id,
+) || pod2usage( 2 );
+
+# Option parsing
+pod2usage( -msg => "Unknown option: @ARGV", -exitval => 2 ) if @ARGV;
+pod2usage( -msg => "Required options: --server", -exitval => 2 )
+    unless $server;
+pod2usage( -msg => "missing key", exitval => 2 )
+    unless defined $exp_id;
+
+# Global options:
+
+my $do_ops = 1;
+my %host_mapping = (
+    'XY01' => 'ipp010','XY02' => 'ipp011','XY03' => 'ipp006','XY04' => 'ipp054',
+    'XY05' => 'ipp007','XY06' => 'ipp055','XY10' => 'ipp008','XY11' => 'ipp056',
+    'XY12' => 'ipp009','XY13' => 'ipp057','XY14' => 'ipp058','XY15' => 'ipp059',
+    'XY16' => 'ipp012','XY17' => 'ipp060','XY20' => 'ipp013','XY21' => 'ipp014',
+    'XY22' => 'ipp061','XY23' => 'ipp015','XY24' => 'ipp016','XY25' => 'ipp062',
+    'XY26' => 'ipp064','XY27' => 'ipp065','XY30' => 'ipp066','XY31' => 'ipp020',
+    'XY32' => 'ipp063','XY33' => 'ipp021','XY34' => 'ipp023','XY35' => 'ipp013',
+    'XY36' => 'ipp024','XY37' => 'ipp019','XY40' => 'ipp025','XY41' => 'ipp018',
+    'XY42' => 'ipp017','XY43' => 'ipp015','XY44' => 'ipp027','XY45' => 'ipp028',
+    'XY46' => 'ipp029','XY47' => 'ipp030','XY50' => 'ipp031','XY51' => 'ipp032',
+    'XY52' => 'ipp033','XY53' => 'ipp034','XY54' => 'ipp035','XY55' => 'ipp036',
+    'XY56' => 'ipp037','XY57' => 'ipp038','XY60' => 'ipp039','XY61' => 'ipp040',
+    'XY62' => 'ipp041','XY63' => 'ipp042','XY64' => 'ipp043','XY65' => 'ipp044',
+    'XY66' => 'ipp046','XY67' => 'ipp047','XY71' => 'ipp048','XY72' => 'ipp049',
+    'XY73' => 'ipp050','XY74' => 'ipp051','XY75' => 'ipp052','XY76' => 'ipp053');
+my %backup_hosts = ('ippb00' => 1, 'ippb01' => 1,
+		    'ippb02' => 1, 'ippb03' => 1,
+		    'ippb04' => 1, 'ippb05' => 1
+    );
+my %backup_destinations = ('ippb04' => 1,
+			   'ippb05' => 1);
+my $backup_Nvols = 3;
+
+# Set up nebulous db interface
+my $neb = Nebulous::Client->new(
+    proxy => "$server",
+);
+die "can't connected to Nebulous Server: $server"
+    unless defined $neb;
+
+# Pull data from the gpc1 database
+my $verbose = 0;
+my $mdcParser = PS::IPP::Metadata::Config->new;
+
+my $regtool_cmd = "$regtool -processedimfile -exp_id $exp_id -dbname $dbname";
+my ($success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = 
+    run(command => $regtool_cmd, verbose => 0);
+unless ($success) {
+    $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR);
+    &my_die("Unable to perform regtool -processedimfile: $error_code", $exp_id);
+}
+my $imfiles = $mdcParser->parse_list(join "", @$stdout_buf) or
+    &my_die("Unable to parse metadata from regtool -processedimfile", $exp_id);
+
+foreach my $imfile (@$imfiles) {
+    my $key        = $imfile->{uri};
+    my $data_state = $imfile->{data_state};
+    my $md5sum     = $imfile->{md5sum};
+    my $hostname   = $imfile->{hostname};
+    my $class_id   = $imfile->{class_id};
+    
+
+#    if (($data_state ne 'full')||($data_state ne 'compressed')) { next; } ## skip things that aren't full.
+    
+# Do validation
+
+    # neb-stat level handling
+    my $stat = $neb->stat($key);
+    die "nebulous key: $key not found" unless $stat;   
+    my $instances = $neb->find_instances($key, 'any');
+    die "no instances found" unless $instances;   
+    
+    my $user_copies;
+    eval {
+	$user_copies = $neb->getxattr($key, "user.copies");
+    };
+    unless(defined($user_copies)) {
+	$user_copies = 1;
+    }
+
+    my @validation;
+    my %md5sum_uniq;
+    my $existing_copies = 0;
+    
+    my @existance;
+    my @md5sums;
+    my @files = map {URI->new($_)->file if $_} @$instances;
+    my @diskvols;
+    my @diskhosts;
+    my @quality;
+    my $Ngood = 0;
+    my $quality_mask = 0;
+
+    for (my $i = 0; $i <= $#files; $i++) {
+	if (-e $files[$i]) {
+	    $existance[$i] = 1;
+	    $existing_copies++;
+	    my $fh;
+	    open($fh,$files[$i]);
+	    $md5sums[$i] = Digest::MD5->new->addfile($fh)->hexdigest;
+	    $md5sum_uniq{$md5sums[$i]} = 1;
+	    close($fh);
+	}
+	else {
+	    $existance[$i] = 0;
+	    $md5sums[$i] = 'NON-EXISTANT';
+	    $md5sum_uniq{$md5sums[$i]} = 1;
+	}
+	($diskhosts[$i],$diskvols[$i]) = parse_volume($files[$i]);
+	$validation[$i] = sprintf("% 3d %32s %s %d %s %d",
+				  $existance[$i],
+				  $md5sums[$i],
+				  $files[$i],
+				  $md5sums[$i] eq $md5sum,
+				  $diskhosts[$i],$diskvols[$i]
+	    );
+
+	# Pre-parse decisions
+	if ($md5sums[$i] ne $md5sum) {
+	    $quality[$i] = 0;
+	}
+	elsif ($existance[$i] == 0) {
+	    $quality[$i] = 0;
+	}
+	elsif (is_backup_volume($diskhosts[$i])) {
+	    $quality[$i] = 1;
+	    $quality_mask = $quality_mask | 1;
+	    $Ngood++;
+	}
+	elsif ($diskhosts[$i] eq $host_mapping{$class_id}) {
+	    $quality[$i] = 1;
+	    $quality_mask = $quality_mask | 2;
+	    $Ngood++;
+	}
+	else {
+	    $quality[$i] = 2;
+	    $Ngood++;
+	}
+    }
+
+    print "\n$key $data_state $md5sum $hostname $class_id\n";
+    if (1) {
+
+	print
+	    "object id:             ", @$stat[0], "\n",
+	    "key:                   ", @$stat[1], "\n",
+	    "read lock:             ", @$stat[2], "\n";
+#	    "write lock:            ", @$stat[3], "\n";
+	print 
+	    "epoch:                 ", @$stat[4], "\n",
+	    "mtime:                 ", @$stat[5], "\n";
+	print 
+	    "md5sum count:          ", scalar(keys %md5sum_uniq), "\n";
+	print 
+	    "requested instances:   ", $user_copies, "\n",
+	    "available instances:   ", @$stat[6], "\n",
+	    "existing instances:    ", $existing_copies, "\n",
+	    "total instances:       ", @$stat[7], "\n",
+	    "instance location:\n", " " x 4;
+	print
+	    join("\n" . " " x 4, @validation), "\n";
+    }
+
+
+    # Decide what to do
+    if ($Ngood == 0) {
+	# DO something to attempt to fix this.
+	my $deneb_key = $key; $deneb_key =~ s/.*?gpc/gpc/;
+
+	open(DD,"/home/panstarrs/ipp/local/bin/deneb-locate.py $deneb_key 2> /dev/null |");
+	my $good_file = '';
+	while (<DD>) {
+	    $_ =~ s/^\s+//;
+	    my ($z,undef,$ff) = split /\s+/;
+	    if (($ff)&&(-e $ff)) {
+		my $md_response = `md5sum $ff`;
+		if ($md_response =~ /$md5sum/) {
+		    $good_file = (split /\s+/,$md_response)[1];
+		}
+	    }
+	}
+	close(DD);
+	$quality[0] = 1;
+	print "cp $good_file $files[0]\n";
+	$Ngood = 1;  # We now hand off this single valid instance object to be handled by the Ngood=1 case.
+    }
+
+    if ($Ngood == 1) { # We have only one version
+	if ($quality_mask & 1) { # ANd it's on a backup volume
+	    print "neb-replicate $key\n";
+	    if ($do_ops) {
+		$neb->replicate($key) or die "failed to replicate the single valid copy";
+		if ($@) { die $@; }
+		
+	    }
+	    
+	}
+	else { # And it's not, so put one there
+	    my $rep_vol = get_random_backup_volume();
+	    print "neb-replicate --volume $rep_vol  $key\n";
+	    if ($do_ops) {
+		$neb->replicate($key,$rep_vol) or die "failed to replicate the single valid copy to the backup node";
+		if ($@) { die $@; }
+		
+# Begin my best validation thought
+		system("sync") == 0 or die "Couldn't sync?";
+		my $uris = $neb->find_instances($key,$rep_vol);
+		@$uris = map {URI->new($_)->file if $_} @$uris;
+		my $fh;
+		open($fh,${ $uris }[0]);
+		my $tmpmd5 = Digest::MD5->new->addfile($fh)->hexdigest;
+		close($fh);
+		if ($tmpmd5 ne $md5sum) { 
+		    die "Post-replication md5sum does not match! $tmpmd5 != $md5sum";
+		}
+# End my best validation thought.
+	    }	    
+	}
+    }
+    else { # N >= 2
+	if (!($quality_mask & 1)) { # no backup copy
+	    my $rep_vol = get_random_backup_volume();
+	    print "neb-replicate --volume $rep_vol  $key\n";
+	    if ($do_ops) {
+		$neb->replicate($key,$rep_vol) or die "failed to replicate a copy to the backup node";
+		if ($@) { die $@; }
+
+# Begin my best validation thought
+		system("sync") == 0 or die "Couldn't sync?";
+		my $uris = $neb->find_instances($key,$rep_vol);
+		@$uris = map {URI->new($_)->file if $_} @$uris;
+
+		print $uris . " " . ${ $uris }[0] . " \n";
+		my $fh;
+		open($fh,${ $uris }[0]);
+
+		my $tmpmd5 = Digest::MD5->new->addfile($fh)->hexdigest;
+		close($fh);
+		if ($tmpmd5 ne $md5sum) { 
+		    die "Post-replication md5sum does not match! $tmpmd5 != $md5sum";
+		}
+# End my best validation thought.
+
+	    }	    
+	}
+	if (!($quality_mask & 2)) { # no copy on the requested host, so select the first as valid.
+	    for (my $i = 0; $i <= $#files; $i++) {
+		if ($quality[$i] == 2) {
+		    $quality[$i] = 1;
+		    last;
+		}
+	    }
+	}
+	for (my $i = 0; $i <= $#files; $i++) {
+	    print "$existance[$i] $quality[$i] $md5sums[$i] $md5sum $diskhosts[$i] $diskvols[$i]\n";
+	    if ($quality[$i] != 1) {
+		my $cull_vol = $diskhosts[$i] . "." . $diskvols[$i];
+		print "neb-cull --volume $cull_vol $key\n";
+		if ($do_ops) {
+		    # The tilde here is to force hard volumes.  Don't touch it.
+		    $neb->cull($key,"~${cull_vol}",2) or die "failed to cull a superfluous instance";
+		    if ($@) { die $@; }
+		}
+	    }
+	}
+    }
+#    exit(0);
+}
+
+sub parse_volume {
+    my $filename = shift(@_);
+    my $full_volume   = (split /\//, $filename)[2];
+    my ($hostname,$vol_index) = split /\./, $full_volume; # /;
+    return($hostname,$vol_index);
+}
+
+sub is_backup_volume {
+    my $hostname = shift(@_);
+    if (exists($backup_hosts{$hostname})) {
+	return(1);
+    }
+    return(0);
+}
+
+sub get_random_backup_volume {
+    my $NN = scalar keys %backup_destinations;
+    my $backup_host = (keys %backup_destinations)[int(rand($NN))];
+    my $backup_vol  = int(rand($backup_Nvols));
+    
+    return("${backup_host}.${backup_vol}");
+}
+
+	
+
+
+sub my_die {
+    my $msg = shift(@_);
+    print $msg . "\n";
+    foreach my $a (@_) {
+	print "ARG: $a\n";
+    }
+    die;
+}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+__END__
+
+
Index: /trunk/ippTasks/rawcheck.pro
===================================================================
--- /trunk/ippTasks/rawcheck.pro	(revision 36203)
+++ /trunk/ippTasks/rawcheck.pro	(revision 36203)
@@ -0,0 +1,168 @@
+## replicate.pro : tasks for data replication : -*- sh -*-
+
+# test for required global variables
+check.globals
+
+# load in the local configuration (NEB_USER, etc.)
+module nebulous.site.pro
+
+$LOGSUBDIR = $LOGDIR/rawcheck
+mkdir $LOGSUBDIR
+
+book init rawcheckPending
+
+macro rawcheck.reset
+  book init checkPending
+end
+
+macro rawcheck.status
+  book listbook rawcheckPending
+end
+
+macro rawcheck.on
+  task rawcheck.load
+    active true
+ end
+  task rawcheck.run
+    active true
+ end
+end
+
+macro rawcheck.off
+  task rawcheck.load
+    active false
+ end
+  task rawcheck.run
+    active false
+ end
+end
+
+# this variable will cycle through the known database names
+$rawcheck_DB = 0
+
+
+
+$EXP_ID_START = 0
+$EXP_ID_RANGE = 500000
+
+task           rawcheck.load
+  host         local
+
+  # modify these after the tasks are tested
+  periods      -poll 0.5
+  periods      -exec 5
+  periods      -timeout 1500
+  npending     1
+
+  # silently drop stdout
+  stdout NULL
+  stderr $LOGSUBDIR/rawcheck.log
+
+  task.exec
+      book npages rawcheckPending -var N
+      if ($N > 2000)
+        process_cleanup rawcheckPending
+        break
+     end      
+     
+     $run = regtool -processedexp -state compressed 
+     if ($DB:n == 0)
+      option DEFAULT
+    else
+      # save the DB name for the exit tasks
+      option $DB:$rawcheck_DB
+      $run = $run -dbname $DB:$rawcheck_DB
+      $rawcheck_DB ++
+      if ($rawcheck_DB >= $DB:n) set rawcheck_DB = 0
+    end
+    add_poll_args run
+    add_poll_labels run
+    command $run
+  end
+
+  # success
+  task.exit    0
+    # convert 'stdout' to book format
+    ipptool2book stdout rawcheckPending -key exp_id -uniq -setword dbname $options:0 -setword pantaskState INIT
+        if ($VERBOSE > 2)
+      book listbook rawcheckPending
+    end
+
+    # delete existing entries in the appropriate pantaskStates
+    process_cleanup rawcheckPending
+  end
+
+  # locked list
+  task.exit    default
+    showcommand failure
+  end
+
+  task.exit    crash
+    showcommand crash
+  end
+
+  # operation times out?
+  task.exit    timeout
+    showcommand timeout
+  end
+end
+
+
+task        rawcheck.run
+  periods      -poll 0.5
+  periods      -exec 5
+  periods      -timeout 600
+
+  task.exec
+    periods -exec 5
+
+    book npages rawcheckPending -var N
+    if ($N == 0) break
+    if ($NETWORK == 0) break
+    
+    # look for new objects in rawcheckPending
+    book getpage rawcheckPending 0 -var pageName -key pantaskState INIT
+    if ("$pageName" == "NULL") break
+
+    book setword rawcheckPending $pageName pantaskState RUN
+
+    # XXX what values do I need to get back?
+    book getword rawcheckPending $pageName exp_id         -var EXP_ID
+    book getword rawcheckPending $pageName dbname         -var DBNAME
+
+#    stdout NULL
+    stdout $LOGSUBDIR/rawcheck.log
+    stderr $LOGSUBDIR/rawcheck.log
+
+    # these operations do not require a database to be specified
+    $run = neb-insck --dbname $DBNAME --exp_id $EXP_ID
+
+    # save the pageName for future reference below
+    options $pageName
+
+    # create the command line
+    if ($VERBOSE > 1)
+      echo command $run
+    end
+    periods -exec 0.05
+    command $run
+  end
+
+  # default exit status
+  task.exit default
+    process_exit rawcheckPending $options:0 $JOB_STATUS
+  end
+
+  task.exit    crash
+    showcommand crash
+    book setword rawcheckPending $options:0 pantaskState CRASH
+  end
+
+  # operation timed out?
+  task.exit    timeout
+    showcommand timeout
+    book setword rawcheckPending $options:0 pantaskState TIMEOUT
+  end
+end
+
+