[ViewVC] Annotation of: cvsroot/COMP/JOBROBOT/TaskSource

#!/usr/bin/env perl

##H This drop box agent initiates tasks on all published datasets.
##H A task is an application to run on a dataset at a particular site.
##H It keeps a record of tasks created and avoids submitting a task too
##H frequently.
##H
##H Usage:
##H
##H TaskSource
##H
##H -state             agent state directory, including inbox
##H -next              next agent to pass the drops to; can be given several times
##H -wait              time to wait in seconds between work scans
##H -sec_between_inits seconds between submission of same dataset/tier combination
##H -siteconfig        file with list of valid SE and their schedule limits
##H -dbsdlspath        path to DBS/DLS discovery ascii files
##H -max-site-queue    default value for maximum events scheduled per site
##H -max-jobs          maximum number of jobs per project, -1 to deactivate
##H -scheduler         scheduler
##H -events            number of events per job
##H -pattern           DBS discovery pattern
##H -validtiers        perl string pattern with valid tiers (DIGI|RECO)
##H -cmssw             CMSSW directory to use for CRAB
##H -simcfg            CMSSW parameter-set for tier SIM
##H -simoutput         putput (comma separated) for tier SIM
##H -digicfg           CMSSW parameter-set for tier DIGI
##H -digioutput        putput (comma separated) for tier DIGI
##H -recocfg           CMSSW parameter-set for tier RECO
##H -recooutput        putput (comma separated) for tier RECO
###H -gencfg            CMSSW parameter-set for tier GEN
###H -genoutput         putput (comma separater) for tier GEN

BEGIN {
  use strict; use warnings; $^W=1;
  our $me = $0; $me =~ s|.*/||;
  our $home = $0; $home =~ s|/[^/]+$||; $home ||= "."; $home .= "/../PHEDEX/Toolkit/Common";
  unshift(@INC, $home);
}

######################################################################
use UtilsHelp;
my %args = (WAITTIME => 600, SECS_PER_EVENT => 1., MAX_SITE_QUEUE => 10);
while (scalar @ARGV) {
  if ($ARGV[0] eq '-state' && scalar @ARGV > 1) {
    shift (@ARGV); $args{DROPDIR}= shift(@ARGV);
  } elsif ($ARGV[0] eq '-next' && scalar @ARGV > 1) {
    shift (@ARGV); push (@{$args{NEXTDIR}}, shift(@ARGV));
  } elsif ($ARGV[0] eq '-wait' && scalar @ARGV > 1) {
    shift (@ARGV); $args{WAITTIME} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-secs_between_inits' && scalar @ARGV > 1) {
    shift (@ARGV); $args{SECS_BETWEEN_INITS} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-siteconfig' && scalar @ARGV > 1) {
    shift (@ARGV); $args{SITECONFIG} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-dbsdlspath' && scalar @ARGV > 1) {
    shift (@ARGV); $args{DBSDLSPATH} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-max-site-queue' && scalar @ARGV > 1) {
    shift (@ARGV); $args{MAX_SITE_QUEUE} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-max-jobs' && scalar @ARGV > 1) {
    shift (@ARGV); $args{MAX_JOBS} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-scheduler' && scalar @ARGV > 1) {
    shift (@ARGV); $args{SCHEDULER} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-events' && scalar @ARGV > 1) {
    shift (@ARGV); $args{EVENTS} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-pattern' && scalar @ARGV > 1) {
    shift (@ARGV); $args{PATTERN} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-validtiers' && scalar @ARGV > 1) {
    shift (@ARGV); $args{VALIDTIERS} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-cmssw' && scalar @ARGV > 1) {
    shift (@ARGV); $args{CMSSW} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-simcfg' && scalar @ARGV > 1) {
    shift (@ARGV); $args{SIMCFG} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-simoutput' && scalar @ARGV > 1) {
    shift (@ARGV); $args{SIMOUTPUT} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-digicfg' && scalar @ARGV > 1) {
    shift (@ARGV); $args{DIGICFG} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-digioutput' && scalar @ARGV > 1) {
    shift (@ARGV); $args{DIGIOUTPUT} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-recocfg' && scalar @ARGV > 1) {
    shift (@ARGV); $args{RECOCFG} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-recooutput' && scalar @ARGV > 1) {
    shift (@ARGV); $args{RECOOUTPUT} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-usercfg' && scalar @ARGV > 1) {
    shift (@ARGV); $args{USERCFG} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-useroutput' && scalar @ARGV > 1) {
    shift (@ARGV); $args{USEROUTPUT} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-gencfg' && scalar @ARGV > 1) {
    shift (@ARGV); $args{GENCFG} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-genoutput' && scalar @ARGV > 1) {
    shift (@ARGV); $args{GENOUTPUT} = shift(@ARGV);
  } elsif ($ARGV[0] eq '-h') {
    &usage();
  } else {
    last;
  }
}
        
if (@ARGV || !$args{DROPDIR}) {
  die "Insufficient parameters, use -h for help.\n";
}

(new TaskSource (%args))->process();

######################################################################
# Routines specific to this agent.
package TaskSource; use strict; use warnings; use base 'UtilsAgent';
use File::Path;
use UtilsCommand;
use UtilsLogging;
use UtilsTiming;
use UtilsNet;
use POSIX;

sub new
  {
    my $proto = shift;
    my $class = ref($proto) || $proto;
    my $self = $class->SUPER::new(@_);
    my %params = (MAX_SITE_QUEUE     => undef,  # max number of jobs per site
                  MAX_JOBS           => undef,  # max number of jobs per project, -1 to deactivate
                  SECS_BETWEEN_INITS => 600,    # path to siteconfig file
                  SITECONFIG         => undef,  # path to siteconfig file
                  DBSDLSPATH         => "",     # path to DBSDLS discovery ascii files
                  EVENTS             => 1000,   # number of events per job
                  PATTERN            => "*",    # DBS discovery pattern
                  VALIDTIERS         => undef,  # perl string pattern with valid tiers (DIGI|RECO)
                  CMSSW              => "_none_", # CMSSW directory to use
                  SIMCFG             => undef,  # CMSSW parameter-set for SIM tier
                  DIGICFG            => undef,  # CMSSW parameter-set for DIGI tier
                  RECOCFG            => undef,  # CMSSW parameter-set for RECO tier
                  USERCFG            => undef,  # CMSSW parameter-set for USER tier
                  GENCFG            => undef,  # CMSSW parameter-set for GEN tier
                  SIMOUTPUT          => undef,  # output (comma separated) for SIM tier
                  DIGIOUTPUT         => undef,  # output (comma separated) for DIGI tier
                  RECOOUTPUT         => undef,  # output (comma separated) for RECO tier
                  USEROUTPUT         => undef,  # output (comma separated) for USER tier
                  GENOUTPUT         => undef,  # output (comma separated) for GEN tier
                  SCHEDULER          => "edg"); # standard scheduler
    my %args = (@_);
    map { $self->{$_} = defined $args{$_} ? $args{$_} : $params{$_} } keys %params;
    bless $self, $class;
    return $self;
  }

sub init
  {
    my ($self) = @_;
    $self->{TASKREPO} = "$self->{DROPDIR}/tasks";
    -d "$self->{TASKREPO}"
      || mkdir "$self->{TASKREPO}"
        || die  "$self->{TASKREPO}: cannot create directory: $!\n";

    # Determine if links supports -dump-width option
    $self->{LINKS_OPTS} = [];
    open (LINKS_HELP, "links -help 2>/dev/null |");
    if ( grep(/-no-numbering/, <LINKS_HELP>) ) {
      push(@{$self->{LINKS_OPTS}}, qw(-dump-width 300 -no-numbering 1));
    } elsif ( grep(/-dump-width/, <LINKS_HELP>) ) {
      push(@{$self->{LINKS_OPTS}}, qw(-dump-width 300));
    }
    close (LINKS_HELP);
  }

# Find out how many jobs are pending for each site.  This is
# insensitive to the job type, and we only check once in the
# beginning to avoid favouring one dataset over another --
# once we decide to proceed for a site, we submit jobs for 
# all datasets.
sub getSiteStatus
  {
    my ($self) = @_;
    my %result = ();
    my $taskrepo = $self->{TASKREPO};
    foreach my $site (<$taskrepo/*/*>) {
      my ($sitename) = ($site =~ m|.*/(.*)|);
      foreach my $d (<$site/*/*>) {
        next if (-f "$d/TASK_FINISHED.txt");
        if (! -f "$d/JOB_CREATE_LOG.txt") {
          $result{$sitename}{C} ||= 0;
          $result{$sitename}{C}++;
          next;
        }
            
        my $f = (<$d/crab_*/share/db/jobs>)[0];
        next if ! defined $f;
            
        foreach my $status (split(/\n/, &input($f) || '')) {
          my @statusarray = split('\|', $status);
          $result{$sitename}{$statusarray[1]} ||= 0;
          $result{$sitename}{$statusarray[1]}++;
        }
      }
    }
    
    return %result;
  }

sub createHashFromString
  {
    my ($self, $string) = @_;
    my $cmd = "mktemp /tmp/hash_XXXXXX";
    open (HASHFILE, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
    my $hash_file_name;
    foreach my $hash_line (split(/\n/, <HASHFILE> || '')) {
      chomp($hash_line);
      $hash_file_name = $hash_line;
    }
    $cmd = "echo $string >> $hash_file_name";
    system $cmd;
#    open (FILLHASHFILE, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
    $cmd = "cksum $hash_file_name";
    open (CKSUMHASHFILE, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
    my ($hash, $dummy_one, $dummy_two);
    foreach my $hash_output (split(/\n/, <CKSUMHASHFILE> || '')) {
      chomp($hash_output);
      ($hash, $dummy_one, $dummy_two) = split(/\s+/,$hash_output);
    }
    $cmd = "rm -f $hash_file_name";
    system $cmd;
#    open (RMHASHFILE, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
    
    return $hash;
  }

sub idle
  {
    &logmsg ("DBS/DLS discovery");
    my ($self, @pending) = @_;
    eval {
      # Get status of how busy the sites are.  We obtain this only once
      # in order to not favour datasets "early on" in the list.
      my %sitestats = $self->getSiteStatus ();
      if (keys %sitestats) {
        my @load;
        foreach my $site (sort keys %sitestats) {
          my $running = ($sitestats{$site}{R} || 0);
          my $submitted = ($sitestats{$site}{B} || 0);
          my $scheduled = ($sitestats{$site}{S} || 0);
          my $created = ($sitestats{$site}{C} || 0);
          my $pending = $created + $submitted + $scheduled + $running;
          push (@load, "$site" . join("", map { " $_=$sitestats{$site}{$_}" }
                                      sort keys %{$sitestats{$site}}) . " L=$pending");
        }
        &logmsg ("current load: ", join ("; ", @load));
      }
      
      # DBS/DLS discovery
      # if ascii DBS/DLS discovery file exists in DBSDLSPATH with name: dlsdls_query_$cleaned_pattern.txt
      # where cleaned pattern is the pattern with / replaced by _
      # take ascii file, otherwise do discovery
      my $cleaned_pattern = $self->{PATTERN};
      $cleaned_pattern    =~ s/\//_/g;
      my $patternfile     = $self->{DBSDLSPATH} . "/dbsdls_query_" . $cleaned_pattern . ".txt";
      if ( -e "$patternfile" ) {
        &logmsg("Use DBS/DLS ascii file: $patternfile");
        open (PUBLISHED,"< $patternfile");
      } else {
        my $cmd = $ENV{PYTHONSCRIPT} . "/DBS2DLSQuery.py --pattern \"$self->{PATTERN}\"";
        open (PUBLISHED, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
      }
      while (<PUBLISHED>) {
        chomp;
        &timeStart($self->{STARTTIME});
        my ($datapath, $site, $totalevents, $events, $secsinit, $cmssw,
            $mycfg) = split(/\s+/, $_);
        # read in siteconfig dynamically
        if ( $self->{SITECONFIG} ) {
          open(siteconfig_file,"$self->{SITECONFIG}");
          my @line_array = <siteconfig_file>;
          my $validsites = "(";
          for (my $i = 0; $i < scalar @line_array; ++$i) {
            if (defined $line_array[$i] ) {
              if ( $line_array[$i] !~ /^#/ && $line_array[$i] !~ /^\n/ ) {
                my ($tempsite, $tempshortsite, $templimit) = split(/\s+/,$line_array[$i]);
                chomp($tempsite);
                $validsites = $validsites . $tempsite . "|";
              }
            }
          }
          chop($validsites);
          $validsites = $validsites . ")";
          close siteconfig_file;
          if ( $validsites eq ")" ) {
            next;
          } else {
            next if ($validsites && $site !~ /$validsites/);
          }
        }
        # if defined use number of events per job from pattern file
        $events ||= $self->{EVENTS};
        $secsinit ||= $self->{SECS_BETWEEN_INITS};
        $cmssw ||= $self->{CMSSW};
        $mycfg ||= "";
        # check if totalevents is larger than $EVENTS * $MAX_JOBS
        if ($self->{MAX_JOBS} != -1) {
          if ( $self->{MAX_JOBS} * $events < $totalevents ) {
            $totalevents = $self->{MAX_JOBS} * $events;
          }
        }
        $self->createTask($datapath, $site, $totalevents, $events,
                          $secsinit, $cmssw, $mycfg, %sitestats);
      }
    };
    do { chomp ($@); &alert ($@); } if $@;

    $self->nap ($self->{WAITTIME});
  }

sub createTask()
  {
    my ($self) = shift(@_);
    my ($datasetpath, $site, $totalevents, $events, $secsinit,
        $cmssw, $mycfg, %sitestats) = @_;

    # Line changed for supporting the DBS2 name format.
    #my($n, $dataset, $tier, $owner) = split(/\//, $datasetpath);
    my($n, $dataset, $owner, $comp_tier) = split(/\//, $datasetpath);
    my($tier) = (split(/-/, $comp_tier))[-1] || $comp_tier;

    # filter on tier
    next if ($self->{VALIDTIERS} && $tier !~ /$self->{VALIDTIERS}/);

    my ($cfg, $output);
    if ($mycfg) {
        $cfg = $mycfg;
    } else {
        if ($tier =~ /RECO/) {
            $cfg = $self->{RECOCFG};
            $output = $self->{RECOOUTPUT};
        } elsif ($tier =~ /USER/) {
            $cfg = $self->{USERCFG};
            $output = $self->{USEROUTPUT};
        } elsif ($tier =~ /DIGI/) {
            $cfg = $self->{DIGICFG};
            $output = $self->{DIGIOUTPUT};
        } elsif ($tier =~ /SIM/) {
            $cfg = $self->{SIMCFG};
            $output = $self->{SIMOUTPUT};
        } elsif ($tier =~ /GEN/) {
            $cfg = $self->{GENCFG};
            $output = $self->{GENOUTPUT};
        } else {
          &logmsg ("No valid tier, defaults for cfg and output");
          $cfg = "robot.cfg";
          $output = "";
        }
    }

    # take site specific or if not found default max site queue
    # read in from siteconfig file
    my %sitelimits = ();
    my %siteNameTranslation = ();
    if ( $self->{SITECONFIG} ) {
      open(siteconfig_file,"$self->{SITECONFIG}");
      my @line_array = <siteconfig_file>;
      for (my $i = 0; $i < scalar @line_array; ++$i) {
        if (defined $line_array[$i] ) {
          if ( $line_array[$i] !~ /^#/ && $line_array[$i] !~ /^\n/ ) {
            my ($tempsite, $tempshortsite, $templimit) = split(/\s+/,$line_array[$i]);
            chomp($tempsite);
            chomp($tempshortsite);
            chomp($templimit);
            $sitelimits{$tempsite} = $templimit;
            $siteNameTranslation{$tempsite} = $tempshortsite;
          }
        }
      }
      close siteconfig_file;
    }

    # If the site isn't too busy already, ignore.
    my $running = ($sitestats{$siteNameTranslation{$site}}{R} || 0);
    my $scheduled = ($sitestats{$siteNameTranslation{$site}}{S} || 0);
    my $created = ($sitestats{$siteNameTranslation{$site}}{C} || 0);
    my $pending = $created + $scheduled + $running;
    my $max_queue = $sitelimits{$site} || $self->{MAX_SITE_QUEUE};
    if ( $pending > $max_queue ) {
      &logmsg("$siteNameTranslation{$site} : Task not created. number of R+S+C jobs ($running + $scheduled + $created = $pending) exceeds site limit of $site ($max_queue)");
      next;
    }
    
    # Find out what is already pending for this task.  First find all
    # existing tasks in the repository, the latest generation.
    my $datestamp = strftime ("%y%m%d", gmtime(time()));
    my $taskdir = "$self->{TASKREPO}/$datestamp/$siteNameTranslation{$site}/$tier";
    
    # create short unique taskbase from $datestamp.$site.$dataset.$tier.$owner
    my $taskbase = $self->createHashFromString ("$datestamp.$site.$dataset.$tier.$owner");
    chomp($taskbase);

    my @existing = sort { $a <=> $b } map { /.*\.(\d+)$/ } <$taskdir/$taskbase.*>;
    my $curgen = pop(@existing) || 0;
    my $nextgen = $curgen + 1;
    
    # OK to create the task if enough time has passed from previous
    # task creation, or there is no previous task.
    if (! -f "$taskdir/$taskbase.$curgen/crab.cfg"
      || (((stat("$taskdir/$taskbase.$curgen/crab.cfg"))[9]
           < time() - $secsinit)))
      {
        my $mydir = $0; $mydir =~ s|/[^/]+$||;
        my $drop = sprintf("%s.%03d", $taskbase, $nextgen);
        my $ret = &runcmd ("$mydir/CrabJobs",
                           "-cfg", "$mydir/$cfg",
                           "-datasetpath", $datasetpath,
                           "-output", $output,
                           "-totalevents", $totalevents,
                           "-whitelist", $site,
                           "-cmssw", $cmssw,
                           "-eventsperjob", $events,
                           "-scheduler", $self->{SCHEDULER},
                           "-jobname", "$taskdir/$drop");
        die "$drop: failed to create task: @{[&runerror($ret)]}\n" if $ret;
        
        &output ("$taskdir/$drop/TASK_INIT.txt",
                 &mytimeofday () . "\n");
        
        my $dropdir = "$self->{WORKDIR}/$drop";
        mkdir "$dropdir" || die "$dropdir: cannot create: $!\n";
        if (&output ("$dropdir/task", "$taskdir/$drop")) {
          &touch ("$dropdir/done");
          $self->relayDrop ($drop);
          &logmsg("stats: $drop $datestamp $site $dataset $tier $owner @{[&formatElapsedTime($self->{STARTTIME})]} success");
        } else {
          &alert ("$drop $datestamp $site $dataset $tier $owner: failed to create drop");
          &rmtree ([ "$self->{WORKDIR}/$drop" ]);
        }
      }
  }
Revision:	1.35
Committed:	Thu Aug 28 12:13:47 2008 UTC (16 years, 8 months ago) by asciaba
Branch:	MAIN
Changes since 1.34:	+2 -1 lines
Log Message:	Supports status B
#	User	Rev	Content
1	gutsche	1.1	#!/usr/bin/env perl
2
3			##H This drop box agent initiates tasks on all published datasets.
4			##H A task is an application to run on a dataset at a particular site.
5			##H It keeps a record of tasks created and avoids submitting a task too
6			##H frequently.
7			##H
8			##H Usage:
9			##H
10	gutsche	1.21	##H TaskSource
11			##H
12			##H -state agent state directory, including inbox
13			##H -next next agent to pass the drops to; can be given several times
14			##H -wait time to wait in seconds between work scans
15			##H -sec_between_inits seconds between submission of same dataset/tier combination
16			##H -siteconfig file with list of valid SE and their schedule limits
17	gutsche	1.25	##H -dbsdlspath path to DBS/DLS discovery ascii files
18	gutsche	1.21	##H -max-site-queue default value for maximum events scheduled per site
19	gutsche	1.22	##H -max-jobs maximum number of jobs per project, -1 to deactivate
20	gutsche	1.21	##H -scheduler scheduler
21			##H -events number of events per job
22			##H -pattern DBS discovery pattern
23			##H -validtiers perl string pattern with valid tiers (DIGI\|RECO)
24	asciaba	1.32	##H -cmssw CMSSW directory to use for CRAB
25	gutsche	1.21	##H -simcfg CMSSW parameter-set for tier SIM
26			##H -simoutput putput (comma separated) for tier SIM
27			##H -digicfg CMSSW parameter-set for tier DIGI
28			##H -digioutput putput (comma separated) for tier DIGI
29			##H -recocfg CMSSW parameter-set for tier RECO
30			##H -recooutput putput (comma separated) for tier RECO
31	belforte	1.27	###H -gencfg CMSSW parameter-set for tier GEN
32			###H -genoutput putput (comma separater) for tier GEN
33	gutsche	1.1
34			BEGIN {
35			use strict; use warnings; $^W=1;
36			our $me = $0; $me =~ s\|.*/\|\|;
37			our $home = $0; $home =~ s\|/[^/]+$\|\|; $home \|\|= "."; $home .= "/../PHEDEX/Toolkit/Common";
38			unshift(@INC, $home);
39			}
40
41			######################################################################
42			use UtilsHelp;
43	gutsche	1.21	my %args = (WAITTIME => 600, SECS_PER_EVENT => 1., MAX_SITE_QUEUE => 10);
44			while (scalar @ARGV) {
45			if ($ARGV[0] eq '-state' && scalar @ARGV > 1) {
46			shift (@ARGV); $args{DROPDIR}= shift(@ARGV);
47			} elsif ($ARGV[0] eq '-next' && scalar @ARGV > 1) {
48			shift (@ARGV); push (@{$args{NEXTDIR}}, shift(@ARGV));
49			} elsif ($ARGV[0] eq '-wait' && scalar @ARGV > 1) {
50			shift (@ARGV); $args{WAITTIME} = shift(@ARGV);
51			} elsif ($ARGV[0] eq '-secs_between_inits' && scalar @ARGV > 1) {
52			shift (@ARGV); $args{SECS_BETWEEN_INITS} = shift(@ARGV);
53			} elsif ($ARGV[0] eq '-siteconfig' && scalar @ARGV > 1) {
54			shift (@ARGV); $args{SITECONFIG} = shift(@ARGV);
55	gutsche	1.25	} elsif ($ARGV[0] eq '-dbsdlspath' && scalar @ARGV > 1) {
56			shift (@ARGV); $args{DBSDLSPATH} = shift(@ARGV);
57	gutsche	1.21	} elsif ($ARGV[0] eq '-max-site-queue' && scalar @ARGV > 1) {
58			shift (@ARGV); $args{MAX_SITE_QUEUE} = shift(@ARGV);
59	gutsche	1.22	} elsif ($ARGV[0] eq '-max-jobs' && scalar @ARGV > 1) {
60			shift (@ARGV); $args{MAX_JOBS} = shift(@ARGV);
61	gutsche	1.21	} elsif ($ARGV[0] eq '-scheduler' && scalar @ARGV > 1) {
62			shift (@ARGV); $args{SCHEDULER} = shift(@ARGV);
63			} elsif ($ARGV[0] eq '-events' && scalar @ARGV > 1) {
64			shift (@ARGV); $args{EVENTS} = shift(@ARGV);
65			} elsif ($ARGV[0] eq '-pattern' && scalar @ARGV > 1) {
66			shift (@ARGV); $args{PATTERN} = shift(@ARGV);
67			} elsif ($ARGV[0] eq '-validtiers' && scalar @ARGV > 1) {
68			shift (@ARGV); $args{VALIDTIERS} = shift(@ARGV);
69	asciaba	1.32	} elsif ($ARGV[0] eq '-cmssw' && scalar @ARGV > 1) {
70			shift (@ARGV); $args{CMSSW} = shift(@ARGV);
71	gutsche	1.21	} elsif ($ARGV[0] eq '-simcfg' && scalar @ARGV > 1) {
72			shift (@ARGV); $args{SIMCFG} = shift(@ARGV);
73			} elsif ($ARGV[0] eq '-simoutput' && scalar @ARGV > 1) {
74			shift (@ARGV); $args{SIMOUTPUT} = shift(@ARGV);
75			} elsif ($ARGV[0] eq '-digicfg' && scalar @ARGV > 1) {
76			shift (@ARGV); $args{DIGICFG} = shift(@ARGV);
77			} elsif ($ARGV[0] eq '-digioutput' && scalar @ARGV > 1) {
78			shift (@ARGV); $args{DIGIOUTPUT} = shift(@ARGV);
79			} elsif ($ARGV[0] eq '-recocfg' && scalar @ARGV > 1) {
80			shift (@ARGV); $args{RECOCFG} = shift(@ARGV);
81			} elsif ($ARGV[0] eq '-recooutput' && scalar @ARGV > 1) {
82			shift (@ARGV); $args{RECOOUTPUT} = shift(@ARGV);
83	belforte	1.31	} elsif ($ARGV[0] eq '-usercfg' && scalar @ARGV > 1) {
84			shift (@ARGV); $args{USERCFG} = shift(@ARGV);
85			} elsif ($ARGV[0] eq '-useroutput' && scalar @ARGV > 1) {
86			shift (@ARGV); $args{USEROUTPUT} = shift(@ARGV);
87	belforte	1.27	} elsif ($ARGV[0] eq '-gencfg' && scalar @ARGV > 1) {
88			shift (@ARGV); $args{GENCFG} = shift(@ARGV);
89			} elsif ($ARGV[0] eq '-genoutput' && scalar @ARGV > 1) {
90			shift (@ARGV); $args{GENOUTPUT} = shift(@ARGV);
91	gutsche	1.21	} elsif ($ARGV[0] eq '-h') {
92			&usage();
93			} else {
94			last;
95			}
96	gutsche	1.1	}
97
98	gutsche	1.21	if (@ARGV \|\| !$args{DROPDIR}) {
99			die "Insufficient parameters, use -h for help.\n";
100	gutsche	1.1	}
101
102			(new TaskSource (%args))->process();
103
104			######################################################################
105			# Routines specific to this agent.
106			package TaskSource; use strict; use warnings; use base 'UtilsAgent';
107			use File::Path;
108			use UtilsCommand;
109			use UtilsLogging;
110			use UtilsTiming;
111			use UtilsNet;
112			use POSIX;
113
114			sub new
115	gutsche	1.21	{
116	gutsche	1.1	my $proto = shift;
117			my $class = ref($proto) \|\| $proto;
118			my $self = $class->SUPER::new(@_);
119	gutsche	1.21	my %params = (MAX_SITE_QUEUE => undef, # max number of jobs per site
120	gutsche	1.22	MAX_JOBS => undef, # max number of jobs per project, -1 to deactivate
121	gutsche	1.21	SECS_BETWEEN_INITS => 600, # path to siteconfig file
122			SITECONFIG => undef, # path to siteconfig file
123	gutsche	1.25	DBSDLSPATH => "", # path to DBSDLS discovery ascii files
124	gutsche	1.21	EVENTS => 1000, # number of events per job
125			PATTERN => "*", # DBS discovery pattern
126			VALIDTIERS => undef, # perl string pattern with valid tiers (DIGI\|RECO)
127	asciaba	1.32	CMSSW => "_none_", # CMSSW directory to use
128	gutsche	1.21	SIMCFG => undef, # CMSSW parameter-set for SIM tier
129			DIGICFG => undef, # CMSSW parameter-set for DIGI tier
130			RECOCFG => undef, # CMSSW parameter-set for RECO tier
131	belforte	1.31	USERCFG => undef, # CMSSW parameter-set for USER tier
132	belforte	1.27	GENCFG => undef, # CMSSW parameter-set for GEN tier
133	gutsche	1.21	SIMOUTPUT => undef, # output (comma separated) for SIM tier
134			DIGIOUTPUT => undef, # output (comma separated) for DIGI tier
135			RECOOUTPUT => undef, # output (comma separated) for RECO tier
136	belforte	1.31	USEROUTPUT => undef, # output (comma separated) for USER tier
137	belforte	1.27	GENOUTPUT => undef, # output (comma separated) for GEN tier
138	gutsche	1.21	SCHEDULER => "edg"); # standard scheduler
139	gutsche	1.1	my %args = (@_);
140			map { $self->{$_} = defined $args{$_} ? $args{$_} : $params{$_} } keys %params;
141			bless $self, $class;
142			return $self;
143	gutsche	1.21	}
144	gutsche	1.1
145			sub init
146	gutsche	1.21	{
147	gutsche	1.1	my ($self) = @_;
148			$self->{TASKREPO} = "$self->{DROPDIR}/tasks";
149			-d "$self->{TASKREPO}"
150	gutsche	1.21	\|\| mkdir "$self->{TASKREPO}"
151	gutsche	1.1	\|\| die "$self->{TASKREPO}: cannot create directory: $!\n";
152
153			# Determine if links supports -dump-width option
154			$self->{LINKS_OPTS} = [];
155			open (LINKS_HELP, "links -help 2>/dev/null \|");
156	gutsche	1.4	if ( grep(/-no-numbering/, <LINKS_HELP>) ) {
157			push(@{$self->{LINKS_OPTS}}, qw(-dump-width 300 -no-numbering 1));
158			} elsif ( grep(/-dump-width/, <LINKS_HELP>) ) {
159			push(@{$self->{LINKS_OPTS}}, qw(-dump-width 300));
160	gutsche	1.1	}
161			close (LINKS_HELP);
162	gutsche	1.21	}
163	gutsche	1.1
164			# Find out how many jobs are pending for each site. This is
165			# insensitive to the job type, and we only check once in the
166			# beginning to avoid favouring one dataset over another --
167			# once we decide to proceed for a site, we submit jobs for
168			# all datasets.
169			sub getSiteStatus
170	gutsche	1.18	{
171	gutsche	1.1	my ($self) = @_;
172			my %result = ();
173			my $taskrepo = $self->{TASKREPO};
174	gutsche	1.21	foreach my $site (<$taskrepo//>) {
175			my ($sitename) = ($site =~ m\|./(.)\|);
176			foreach my $d (<$site//>) {
177	belforte	1.34	next if (-f "$d/TASK_FINISHED.txt");
178	gutsche	1.21	if (! -f "$d/JOB_CREATE_LOG.txt") {
179			$result{$sitename}{C} \|\|= 0;
180			$result{$sitename}{C}++;
181			next;
182			}
183	gutsche	1.18
184	gutsche	1.21	my $f = (<$d/crab_*/share/db/jobs>)[0];
185			next if ! defined $f;
186	gutsche	1.18
187	gutsche	1.21	foreach my $status (split(/\n/, &input($f) \|\| '')) {
188			my @statusarray = split('\\|', $status);
189			$result{$sitename}{$statusarray[1]} \|\|= 0;
190			$result{$sitename}{$statusarray[1]}++;
191			}
192	gutsche	1.18	}
193	gutsche	1.21	}
194	gutsche	1.18
195	gutsche	1.1	return %result;
196	gutsche	1.18	}
197	gutsche	1.1
198	gutsche	1.21	sub createHashFromString
199			{
200			my ($self, $string) = @_;
201			my $cmd = "mktemp /tmp/hash_XXXXXX";
202			open (HASHFILE, "$cmd 2>/dev/null \|") or die "cannot run `$cmd': $!\n";
203			my $hash_file_name;
204			foreach my $hash_line (split(/\n/, <HASHFILE> \|\| '')) {
205			chomp($hash_line);
206			$hash_file_name = $hash_line;
207			}
208			$cmd = "echo $string >> $hash_file_name";
209			system $cmd;
210			# open (FILLHASHFILE, "$cmd 2>/dev/null \|") or die "cannot run `$cmd': $!\n";
211			$cmd = "cksum $hash_file_name";
212			open (CKSUMHASHFILE, "$cmd 2>/dev/null \|") or die "cannot run `$cmd': $!\n";
213			my ($hash, $dummy_one, $dummy_two);
214			foreach my $hash_output (split(/\n/, <CKSUMHASHFILE> \|\| '')) {
215			chomp($hash_output);
216	asciaba	1.33	($hash, $dummy_one, $dummy_two) = split(/\s+/,$hash_output);
217	gutsche	1.21	}
218			$cmd = "rm -f $hash_file_name";
219			system $cmd;
220			# open (RMHASHFILE, "$cmd 2>/dev/null \|") or die "cannot run `$cmd': $!\n";
221
222			return $hash;
223			}
224
225	gutsche	1.1	sub idle
226			{
227	gutsche	1.21	&logmsg ("DBS/DLS discovery");
228	gutsche	1.1	my ($self, @pending) = @_;
229			eval {
230			# Get status of how busy the sites are. We obtain this only once
231			# in order to not favour datasets "early on" in the list.
232			my %sitestats = $self->getSiteStatus ();
233	gutsche	1.21	if (keys %sitestats) {
234			my @load;
235			foreach my $site (sort keys %sitestats) {
236	belforte	1.31	my $running = ($sitestats{$site}{R} \|\| 0);
237	asciaba	1.35	my $submitted = ($sitestats{$site}{B} \|\| 0);
238	belforte	1.31	my $scheduled = ($sitestats{$site}{S} \|\| 0);
239			my $created = ($sitestats{$site}{C} \|\| 0);
240	asciaba	1.35	my $pending = $created + $submitted + $scheduled + $running;
241	gutsche	1.21	push (@load, "$site" . join("", map { " $_=$sitestats{$site}{$_}" }
242	belforte	1.31	sort keys %{$sitestats{$site}}) . " L=$pending");
243	gutsche	1.1	}
244	gutsche	1.21	&logmsg ("current load: ", join ("; ", @load));
245			}
246	gutsche	1.1
247	gutsche	1.25	# DBS/DLS discovery
248			# if ascii DBS/DLS discovery file exists in DBSDLSPATH with name: dlsdls_query_$cleaned_pattern.txt
249			# where cleaned pattern is the pattern with / replaced by _
250			# take ascii file, otherwise do discovery
251			my $cleaned_pattern = $self->{PATTERN};
252			$cleaned_pattern =~ s/\//_/g;
253			my $patternfile = $self->{DBSDLSPATH} . "/dbsdls_query_" . $cleaned_pattern . ".txt";
254			if ( -e "$patternfile" ) {
255			&logmsg("Use DBS/DLS ascii file: $patternfile");
256			open (PUBLISHED,"< $patternfile");
257			} else {
258	marcelo	1.28	my $cmd = $ENV{PYTHONSCRIPT} . "/DBS2DLSQuery.py --pattern \"$self->{PATTERN}\"";
259	gutsche	1.25	open (PUBLISHED, "$cmd 2>/dev/null \|") or die "cannot run `$cmd': $!\n";
260			}
261	gutsche	1.21	while (<PUBLISHED>) {
262			chomp;
263			&timeStart($self->{STARTTIME});
264	asciaba	1.32	my ($datapath, $site, $totalevents, $events, $secsinit, $cmssw,
265	asciaba	1.33	$mycfg) = split(/\s+/, $_);
266	gutsche	1.21	# read in siteconfig dynamically
267			if ( $self->{SITECONFIG} ) {
268			open(siteconfig_file,"$self->{SITECONFIG}");
269			my @line_array = <siteconfig_file>;
270			my $validsites = "(";
271			for (my $i = 0; $i < scalar @line_array; ++$i) {
272			if (defined $line_array[$i] ) {
273			if ( $line_array[$i] !~ /^#/ && $line_array[$i] !~ /^\n/ ) {
274	asciaba	1.33	my ($tempsite, $tempshortsite, $templimit) = split(/\s+/,$line_array[$i]);
275	gutsche	1.21	chomp($tempsite);
276			$validsites = $validsites . $tempsite . "\|";
277	gutsche	1.19	}
278			}
279	gutsche	1.5	}
280	gutsche	1.21	chop($validsites);
281			$validsites = $validsites . ")";
282			close siteconfig_file;
283	gutsche	1.24	if ( $validsites eq ")" ) {
284			next;
285			} else {
286			next if ($validsites && $site !~ /$validsites/);
287			}
288	gutsche	1.5	}
289	asciaba	1.32	# if defined use number of events per job from pattern file
290			$events \|\|= $self->{EVENTS};
291			$secsinit \|\|= $self->{SECS_BETWEEN_INITS};
292			$cmssw \|\|= $self->{CMSSW};
293			$mycfg \|\|= "";
294	gutsche	1.22	# check if totalevents is larger than $EVENTS * $MAX_JOBS
295			if ($self->{MAX_JOBS} != -1) {
296	asciaba	1.32	if ( $self->{MAX_JOBS} * $events < $totalevents ) {
297			$totalevents = $self->{MAX_JOBS} * $events;
298	gutsche	1.22	}
299			}
300	asciaba	1.32	$self->createTask($datapath, $site, $totalevents, $events,
301			$secsinit, $cmssw, $mycfg, %sitestats);
302	gutsche	1.1	}
303			};
304			do { chomp ($@); &alert ($@); } if $@;
305
306			$self->nap ($self->{WAITTIME});
307			}
308
309			sub createTask()
310			{
311	corvo	1.3	my ($self) = shift(@_);
312	asciaba	1.32	my ($datasetpath, $site, $totalevents, $events, $secsinit,
313			$cmssw, $mycfg, %sitestats) = @_;
314	gutsche	1.5
315	marcelo	1.28	# Line changed for supporting the DBS2 name format.
316			#my($n, $dataset, $tier, $owner) = split(/\//, $datasetpath);
317			my($n, $dataset, $owner, $comp_tier) = split(/\//, $datasetpath);
318	asciaba	1.33	my($tier) = (split(/-/, $comp_tier))[-1] \|\| $comp_tier;
319	gutsche	1.5
320	gutsche	1.21	# filter on tier
321			next if ($self->{VALIDTIERS} && $tier !~ /$self->{VALIDTIERS}/);
322
323			my ($cfg, $output);
324	asciaba	1.32	if ($mycfg) {
325			$cfg = $mycfg;
326	gutsche	1.21	} else {
327	asciaba	1.32	if ($tier =~ /RECO/) {
328			$cfg = $self->{RECOCFG};
329			$output = $self->{RECOOUTPUT};
330			} elsif ($tier =~ /USER/) {
331			$cfg = $self->{USERCFG};
332			$output = $self->{USEROUTPUT};
333			} elsif ($tier =~ /DIGI/) {
334			$cfg = $self->{DIGICFG};
335			$output = $self->{DIGIOUTPUT};
336			} elsif ($tier =~ /SIM/) {
337			$cfg = $self->{SIMCFG};
338			$output = $self->{SIMOUTPUT};
339			} elsif ($tier =~ /GEN/) {
340			$cfg = $self->{GENCFG};
341			$output = $self->{GENOUTPUT};
342			} else {
343			&logmsg ("No valid tier, defaults for cfg and output");
344			$cfg = "robot.cfg";
345			$output = "";
346			}
347	gutsche	1.5	}
348
349	gutsche	1.8	# take site specific or if not found default max site queue
350	gutsche	1.20	# read in from siteconfig file
351			my %sitelimits = ();
352	gutsche	1.21	my %siteNameTranslation = ();
353	gutsche	1.20	if ( $self->{SITECONFIG} ) {
354			open(siteconfig_file,"$self->{SITECONFIG}");
355			my @line_array = <siteconfig_file>;
356			for (my $i = 0; $i < scalar @line_array; ++$i) {
357			if (defined $line_array[$i] ) {
358			if ( $line_array[$i] !~ /^#/ && $line_array[$i] !~ /^\n/ ) {
359	asciaba	1.33	my ($tempsite, $tempshortsite, $templimit) = split(/\s+/,$line_array[$i]);
360	gutsche	1.20	chomp($tempsite);
361	gutsche	1.21	chomp($tempshortsite);
362	gutsche	1.20	chomp($templimit);
363			$sitelimits{$tempsite} = $templimit;
364	gutsche	1.21	$siteNameTranslation{$tempsite} = $tempshortsite;
365	gutsche	1.20	}
366			}
367			}
368			close siteconfig_file;
369			}
370	gutsche	1.21
371			# If the site isn't too busy already, ignore.
372	belforte	1.30	my $running = ($sitestats{$siteNameTranslation{$site}}{R} \|\| 0);
373			my $scheduled = ($sitestats{$siteNameTranslation{$site}}{S} \|\| 0);
374			my $created = ($sitestats{$siteNameTranslation{$site}}{C} \|\| 0);
375			my $pending = $created + $scheduled + $running;
376	gutsche	1.20	my $max_queue = $sitelimits{$site} \|\| $self->{MAX_SITE_QUEUE};
377	gutsche	1.26	if ( $pending > $max_queue ) {
378	belforte	1.30	&logmsg("$siteNameTranslation{$site} : Task not created. number of R+S+C jobs ($running + $scheduled + $created = $pending) exceeds site limit of $site ($max_queue)");
379	gutsche	1.4	next;
380			}
381
382			# Find out what is already pending for this task. First find all
383			# existing tasks in the repository, the latest generation.
384			my $datestamp = strftime ("%y%m%d", gmtime(time()));
385	gutsche	1.21	my $taskdir = "$self->{TASKREPO}/$datestamp/$siteNameTranslation{$site}/$tier";
386
387			# create short unique taskbase from $datestamp.$site.$dataset.$tier.$owner
388			my $taskbase = $self->createHashFromString ("$datestamp.$site.$dataset.$tier.$owner");
389			chomp($taskbase);
390
391	gutsche	1.4	my @existing = sort { $a <=> $b } map { /.\.(\d+)$/ } <$taskdir/$taskbase.>;
392			my $curgen = pop(@existing) \|\| 0;
393			my $nextgen = $curgen + 1;
394
395			# OK to create the task if enough time has passed from previous
396			# task creation, or there is no previous task.
397			if (! -f "$taskdir/$taskbase.$curgen/crab.cfg"
398	gutsche	1.21	\|\| (((stat("$taskdir/$taskbase.$curgen/crab.cfg"))[9]
399	asciaba	1.32	< time() - $secsinit)))
400	gutsche	1.4	{
401			my $mydir = $0; $mydir =~ s\|/[^/]+$\|\|;
402			my $drop = sprintf("%s.%03d", $taskbase, $nextgen);
403	gutsche	1.21	my $ret = &runcmd ("$mydir/CrabJobs",
404			"-cfg", "$mydir/$cfg",
405			"-datasetpath", $datasetpath,
406	gutsche	1.4	"-output", $output,
407	gutsche	1.21	"-totalevents", $totalevents,
408			"-whitelist", $site,
409	asciaba	1.32	"-cmssw", $cmssw,
410			"-eventsperjob", $events,
411	gutsche	1.4	"-scheduler", $self->{SCHEDULER},
412	gutsche	1.21	"-jobname", "$taskdir/$drop");
413	gutsche	1.4	die "$drop: failed to create task: @{[&runerror($ret)]}\n" if $ret;
414	gutsche	1.21
415	gutsche	1.4	&output ("$taskdir/$drop/TASK_INIT.txt",
416			&mytimeofday () . "\n");
417
418			my $dropdir = "$self->{WORKDIR}/$drop";
419			mkdir "$dropdir" \|\| die "$dropdir: cannot create: $!\n";
420	gutsche	1.21	if (&output ("$dropdir/task", "$taskdir/$drop")) {
421			&touch ("$dropdir/done");
422			$self->relayDrop ($drop);
423	gutsche	1.23	&logmsg("stats: $drop $datestamp $site $dataset $tier $owner @{[&formatElapsedTime($self->{STARTTIME})]} success");
424	gutsche	1.21	} else {
425	gutsche	1.23	&alert ("$drop $datestamp $site $dataset $tier $owner: failed to create drop");
426	gutsche	1.21	&rmtree ([ "$self->{WORKDIR}/$drop" ]);
427			}
428	gutsche	1.4	}
429			}