pegasus-wms 4.0.1+dfsg-8 / usr / bin / pegasus-run

#!/usr/bin/env perl
#
# Wrapper around pegasus-submit-dag to run a workflow
#
# Usage: pegasus-run rundir
##
#  Copyright 2007-2010 University Of Southern California
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#  http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing,
#  software distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
##
#
# Author: Jens-S. V??ckler voeckler at isi dot edu
# Author: Gaurang Mehta gmehta at isi dot edu
# Revision : $Revision: 4507 $
#
use 5.006;
use strict;
use Carp;
use Cwd;
use File::Spec;
use File::Basename qw(basename dirname);
use Getopt::Long qw(:config bundling no_ignore_case);

# path to load our libs.. 
BEGIN { 
    my $pegasus_config = File::Spec->catfile( dirname($0), 'pegasus-config' );
    eval `$pegasus_config --perl-dump`;
    die("Unable to eval pegasus-config output. $@") if $@;
}

# load our own local modules
use Pegasus::Common;
use Pegasus::Properties qw(%initial); # parses -Dprop=val from @ARGV

sub make_path($) {
    my $p = File::Spec->catfile( Cwd::realpath( dirname($0) ), shift() ); 
    warn "# make_path = $p\n" if $main::DEBUG; 
    $p;
}

sub usage(;$);			# { }

# constants
$main::DEBUG = 0;		# for now
my $condor;			# if set, use Condor for pegasus-monitord
my $grid=0;                     # if set, enable grid checks, disabled by default
my $default_tsd = make_path( 'pegasus-monitord' );
my $dagman = make_path( 'pegasus-dagman' );
my ($tsd,$monitor,$conffile); 
GetOptions( "help|h" => \&usage
	  , "monitor|m=s" => \$tsd
	  , "condor" => \$condor
	  , "debug|d=o" => \$main::DEBUG
	  , "verbose|v+" => \$main::DEBUG
          , "grid!" => \$grid
	  , 'conf|c=s' => \$conffile
	  , "M" => \$monitor
	  );

my $run = shift;
# NEW: Default to cwd if nothing was specified
unless ( defined $run ) { 
    $run = getcwd();
    my $braindb = File::Spec->catfile( $run, $Pegasus::Common::brainbase ); 
    usage("You need to provide a valid run directory. Cannot find pegasus brain file") 
	unless -r $braindb; 
}

# extra sanity
usage( "$run is not a directory." ) unless -d $run;
usage( "$run is not accessible." ) unless -r _; 
my %config = slurp_braindb( $run ) or die "ERROR: open braindb: $!\n";

# pre-condition: The planner writes all properties per workflow into the DAG dir.
my $props = Pegasus::Properties->new( $conffile, File::Spec->catfile($run,$config{properties}) );

# if TSD is empty, fill in from default locations
$tsd ||=  $props->property('pegasus.monitord') || $default_tsd;

#
# --- functions -------------------------------------------------
#

sub usage(;$) {
    my $msg = shift;
    my $flag = ( defined $msg && lc($msg) ne 'help' ); 
    if ( $flag ) { 
	my $tty = -t STDOUT;
	print "\033[1m" if $tty;
	print "ERROR: $msg\n";
	print "\033[0m" if $tty;
    }

    my $basename = basename($0,'.pl');
    print << "EOF";

Usage: $basename [options] [rundir]

SemiMandatory arguments:
  rundir  is the directory where the workflow resides as well as ancilliary
          files related to the workflow. Defaults to current working directory if not specified.

Optional arguments:
 -Dprop=val         Explicit settings of a property (multi-option) (only use if really required)
 -c|--conf fn	    Use file fn as pegasus properties file. (only use for debugging purposes.
                    pegasus-run will pick the correct planned properties by default.
 -h|--help          Print this help message and exit.
 -m|--monitor l     Uses the workflow monitor daemon installed as l, default
                    is $default_tsd.
 -d|--debug lvl     Sets the debug level (verbosity), default is $main::DEBUG. 
 -v|--verbose       Raises debug level by 1, see --debug. 
 --condor           Uses Condor to submit daemons (prototype).
 --nogrid           Disable checking for grids (default).
 --grid             Enable checking for grids. 

EOF
    exit( $flag ? 1 : 0 );
}

sub create_args {
    # purpose: create either a shell daemon, or condor-submitted job
    # warning: It is assumed that condor-jobs require -N to foreground
    # globals: $condor (IN): use Condor (true) or daemon (false) mode
    # paramtr: $daemon (IN): executable location of daemon 
    #          @args (IN): other arguments to daemon
    # returns: argument vector ready to execute
    my $daemon = shift || croak "Need a daemon name";

    my @result;
    if ( $condor ) {
	# use Condor to submit a daemon -- makes it restartable on SH death
	# however, insert -N into the daemon's argument list for foregrounding
	my $condor_submit = find_exec('condor_submit') || 
	    die "ERROR: Unable to find condor_submit\n";
	my $base = 'pegasus-run-' . basename($daemon);
	my $fn = File::Spec->catfile( $run, "$base.sub" );
	local(*SUB);
	open( SUB, ">$fn" ) || die "open $fn: $!\n";

	# only arg
	@result = ( );
	foreach my $k ( keys %initial ) { push( @result, "-D$k=$initial{$k}" ) }
	push( @result, '-N' );
	push( @result, @_ ) if ( @_ > 0 );

	# write submit file for (restartable) pegasus-monitord
	print SUB "initialdir = $run\n";
	print SUB 'executable = ', File::Spec->rel2abs($daemon), "\n";
	print SUB 'arguments  = ', join(' ',@result), "\n";
	print SUB "universe   = local\n";
	print SUB "notification = NEVER\n";
	print SUB "getenv = True\n";
	print SUB "output = $base.out.txt\n";
	print SUB "error  = $base.err.txt\n";
	print SUB "log    = $base.log.txt\n";
	print SUB "queue\n";
	close SUB;
	@result = ( $condor_submit, $fn );
    } else {
	# run daemon directly
	@result = ( $daemon );
	foreach my $k ( keys %initial ) { push( @result, "-D$k=$initial{$k}" ) }
	push( @result, @_ ) if ( @_ > 0 );
    }

    @result;
}


#
# --- main ------------------------------------------------------
#

# sanity check: lower umask
umask 0002;

# where were we...
my $here = File::Spec->curdir();
$SIG{'__DIE__'} = sub {
    chdir($here) if defined $here;
};
chdir($run) || die "ERROR: chdir $run: $!\n";

# sanity check: find the pegasus-monitord daemon
warn "WARN: Either $tsd does not exist or is not executable\n" unless -x $tsd;
print STDERR "# found $tsd\n" if $main::DEBUG;

# Do GRID check if $grid enabled
if ( $grid ) {
    # sanity check: Is there a GLOBUS_LOCATION?

    die ( "ERROR: Your environment setup misses GLOBUS_LOCATION.\n",
	  "Please check carefully that you have sourced the correct setup files!\n" )
	unless exists $ENV{'GLOBUS_LOCATION'};

    # sanity check: find grid-proxy-init from GLOBUS_LOCATION
    my $g_l = $ENV{'GLOBUS_LOCATION'};
    print STDERR "# GLOBUS_LOCATION=$g_l\n" if $main::DEBUG;

    # sanity check: Is G_L part of L_L_P?
    my @llp = grep { /^$g_l/ } split /:/, $ENV{'LD_LIBRARY_PATH'};
    $ENV{'LD_LIBRARY_PATH'}=File::Spec->catfile($ENV{'GLOBUS_LOCATION'},"/lib") if @llp == 0;

    # Find grid-proxy-init (should we use openssl instead?? )
    my $gpi = File::Spec->catfile( $g_l, 'bin', 'grid-proxy-info' );
    die "ERROR: Unable to find $gpi\n" unless -x $gpi;
    print STDERR "# found $gpi\n" if $main::DEBUG;

    # common user error
    # sanity check: Sufficient time left on grid proxy certificate
    open( GPI, "$gpi -timeleft 2>&1|" ) || die "open $gpi: $!\n";
    my $timeleft = <GPI>;
    chomp($timeleft);
    $timeleft += 0;			# make numeric
    close GPI;
    die( "ERROR: $gpi died on signal ", ($? & 127) ) if ( ($? & 127) > 0 );
    die( "ERROR: Grid proxy not initialized, Please generate a new proxy\n" ) if $timeleft == -1;
    die( "ERROR: Grid proxy expired, please refresh\n" ) if $timeleft == 0;
    die( "ERROR: $gpi exited with status ", $?>>8 ) if ( $? != 0 );
    warn( "ERROR: Too little time left ($timeleft s) on grid proxy. Please refresh your proxy\n" ) 
	if $timeleft < 7200;
    print STDERR "# grid proxy has $timeleft s left\n" if $main::DEBUG;
} # end if($grid) checks only if grid option is enabled.

if ( $config{dag} ) {
    # find pegasus-submit-dag
    my $psd = File::Spec->catfile( $config{bindir}, 'pegasus-submit-dag' );
    die "ERROR: Unable to access $psd\n" unless -x $psd;
    print STDERR "# found $psd\n" if $main::DEBUG;

    # sanity check: Is the DAG file there? 
    die "ERROR: Unable to locate $config{dag}\n" unless -r $config{dag};
    print STDERR "# found $config{dag}\n" if $main::DEBUG;

    # NEW: is there a rescue file, or multiple rescue levels?
    my $original;
    my @rescue = check_rescue($run,$config{dag});
    if ( @rescue > 0 ) {
	my (@stat,%rescue,$maxsize);
	foreach my $fn ( @rescue ) {
	    if ( (@stat = stat($fn)) > 0 ) {
		$rescue{$fn} = [ @stat ];
		$maxsize = $stat[7] if $maxsize < $stat[7];
	    }
	}

	print "\n\nDetected the presence of Rescue DAGs:\n";
	my $width = log10($maxsize);
	foreach my $fn ( @rescue ) {
	    printf( " %s %*u %s\n", 
		    isodate($rescue{$fn}[9]), 
		    $width, $rescue{$fn}[7], 
		    basename($fn) );
	}

	# overwrite with "latest" (read: longest basename) rescue DAG
	$original = $config{dag};
	$config{dag} = $rescue[$#rescue];
	print "\nWILL USE ", $config{dag}, "\n\n";
    }

    # find the workflow name and timestamp for pegasus-status
    my $workflow=$config{'pegasus_wf_name'};
    my $time=$config{timestamp};

    # start DAGMan with default throttles
    my @extra = ();
    foreach my $k ( keys %initial ) {
	push( @extra, "-D$k=$initial{$k}" );
    }

    my @args = ( $psd );
    push( @args, @extra ) if @extra > 0;
    push( @args, '-d', 0+$main::DEBUG);
    push( @args, '--grid' ) if $grid;
    push( @args, '--dagman', $dagman, $config{dag} );

    print STDERR "# @args\n" if $main::DEBUG;
    system(@args) == 0 
	or die( "ERROR: Running pegasus-submit-dag failed with ", parse_exit($?));
    print STDERR "# dagman is running\n" if $main::DEBUG;

    if ( $monitor ) {
	if ( -x $tsd ) {
	    # run pegasus-monitord to update job stats until DAGMan finishes
	    my @tsdargs = ($tsd);
	    push( @tsdargs, $config{dag} . '.dagman.out' );

	    print STDERR "# @tsdargs\n" if $main::DEBUG;
	    if ( system(@tsdargs) == 0 ) {
		print STDERR "# $tsd is running\n" if $main::DEBUG;
	    } else {
		warn( "WARN: Running $tsd failed with ", parse_exit($?) );
	    }
	}
    }

    # next step
    if ( @rescue > 0 ) {
	my $n = @rescue + 0;
	print( "\nI found $n rescue DAG", ( $n>1 ? 's' : '' ), 
	       ". I submitted the rescue DAG\n", 
	       $config{dag}, "\ninstead of\n", $original, "\n" );
    }

    my $did=undef;
    my $daglogfile=$config{dag}.".dagman.out";
    if ( open( DID, "<$daglogfile" ) ) {
	while (<DID>) {
	    if ( /condor_scheduniv_exec/ ) {
		# this part was written by a python programmer?
		$did=(split /\./,  (split)[3],2)[1];
		last;
	    }
	}
	close(DID);
    }
    print << "EOF";

Your Workflow has been started and runs in base directory given below

cd $run

*** To monitor the workflow you can run ***

pegasus-status -l $run

*** To remove your workflow run ***
EOF

    print "pegasus-remove -d $did\nor\n" if defined $did; 
    print "pegasus-remove $run\n\n";

} elsif ( $config{type}=="shell" ) {
    # sanity check: Is the SCRIPT file there? 
    die "ERROR: Unable to execute $config{script}\n" unless -x $config{script};
    print STDERR "# found $config{script}\n" if $main::DEBUG;

    my @args=( "/bin/bash", $config{script} );
    system(@args) == 0 
	or die( "ERROR: Running $config{script} failed with ", 
		parse_exit($?) );
}

chdir($here);
exit 0;