ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/JOBROBOT/TaskSource
Revision: 1.21
Committed: Fri Oct 13 23:29:34 2006 UTC (18 years, 6 months ago) by gutsche
Branch: MAIN
Changes since 1.20: +200 -470 lines
Log Message:
robot for >= CMSSW_0_8_0 and CRAB_1_4_0

File Contents

# 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     ##H -max-site-queue default value for maximum events scheduled per site
18     ##H -scheduler scheduler
19     ##H -events number of events per job
20     ##H -pattern DBS discovery pattern
21     ##H -validtiers perl string pattern with valid tiers (DIGI|RECO)
22     ##H -simcfg CMSSW parameter-set for tier SIM
23     ##H -simoutput putput (comma separated) for tier SIM
24     ##H -digicfg CMSSW parameter-set for tier DIGI
25     ##H -digioutput putput (comma separated) for tier DIGI
26     ##H -recocfg CMSSW parameter-set for tier RECO
27     ##H -recooutput putput (comma separated) for tier RECO
28 gutsche 1.1
29     BEGIN {
30     use strict; use warnings; $^W=1;
31     our $me = $0; $me =~ s|.*/||;
32     our $home = $0; $home =~ s|/[^/]+$||; $home ||= "."; $home .= "/../PHEDEX/Toolkit/Common";
33     unshift(@INC, $home);
34     }
35    
36     ######################################################################
37     use UtilsHelp;
38 gutsche 1.21 my %args = (WAITTIME => 600, SECS_PER_EVENT => 1., MAX_SITE_QUEUE => 10);
39     while (scalar @ARGV) {
40     if ($ARGV[0] eq '-state' && scalar @ARGV > 1) {
41     shift (@ARGV); $args{DROPDIR}= shift(@ARGV);
42     } elsif ($ARGV[0] eq '-next' && scalar @ARGV > 1) {
43     shift (@ARGV); push (@{$args{NEXTDIR}}, shift(@ARGV));
44     } elsif ($ARGV[0] eq '-wait' && scalar @ARGV > 1) {
45     shift (@ARGV); $args{WAITTIME} = shift(@ARGV);
46     } elsif ($ARGV[0] eq '-secs_between_inits' && scalar @ARGV > 1) {
47     shift (@ARGV); $args{SECS_BETWEEN_INITS} = shift(@ARGV);
48     } elsif ($ARGV[0] eq '-siteconfig' && scalar @ARGV > 1) {
49     shift (@ARGV); $args{SITECONFIG} = shift(@ARGV);
50     } elsif ($ARGV[0] eq '-max-site-queue' && scalar @ARGV > 1) {
51     shift (@ARGV); $args{MAX_SITE_QUEUE} = shift(@ARGV);
52     } elsif ($ARGV[0] eq '-scheduler' && scalar @ARGV > 1) {
53     shift (@ARGV); $args{SCHEDULER} = shift(@ARGV);
54     } elsif ($ARGV[0] eq '-events' && scalar @ARGV > 1) {
55     shift (@ARGV); $args{EVENTS} = shift(@ARGV);
56     } elsif ($ARGV[0] eq '-pattern' && scalar @ARGV > 1) {
57     shift (@ARGV); $args{PATTERN} = shift(@ARGV);
58     } elsif ($ARGV[0] eq '-validtiers' && scalar @ARGV > 1) {
59     shift (@ARGV); $args{VALIDTIERS} = shift(@ARGV);
60     } elsif ($ARGV[0] eq '-simcfg' && scalar @ARGV > 1) {
61     shift (@ARGV); $args{SIMCFG} = shift(@ARGV);
62     } elsif ($ARGV[0] eq '-simoutput' && scalar @ARGV > 1) {
63     shift (@ARGV); $args{SIMOUTPUT} = shift(@ARGV);
64     } elsif ($ARGV[0] eq '-digicfg' && scalar @ARGV > 1) {
65     shift (@ARGV); $args{DIGICFG} = shift(@ARGV);
66     } elsif ($ARGV[0] eq '-digioutput' && scalar @ARGV > 1) {
67     shift (@ARGV); $args{DIGIOUTPUT} = shift(@ARGV);
68     } elsif ($ARGV[0] eq '-recocfg' && scalar @ARGV > 1) {
69     shift (@ARGV); $args{RECOCFG} = shift(@ARGV);
70     } elsif ($ARGV[0] eq '-recooutput' && scalar @ARGV > 1) {
71     shift (@ARGV); $args{RECOOUTPUT} = shift(@ARGV);
72     } elsif ($ARGV[0] eq '-h') {
73     &usage();
74     } else {
75     last;
76     }
77 gutsche 1.1 }
78    
79 gutsche 1.21 if (@ARGV || !$args{DROPDIR}) {
80     die "Insufficient parameters, use -h for help.\n";
81 gutsche 1.1 }
82    
83     (new TaskSource (%args))->process();
84    
85     ######################################################################
86     # Routines specific to this agent.
87     package TaskSource; use strict; use warnings; use base 'UtilsAgent';
88     use File::Path;
89     use UtilsCommand;
90     use UtilsLogging;
91     use UtilsTiming;
92     use UtilsNet;
93     use POSIX;
94    
95     sub new
96 gutsche 1.21 {
97 gutsche 1.1 my $proto = shift;
98     my $class = ref($proto) || $proto;
99     my $self = $class->SUPER::new(@_);
100 gutsche 1.21 my %params = (MAX_SITE_QUEUE => undef, # max number of jobs per site
101     SECS_BETWEEN_INITS => 600, # path to siteconfig file
102     SITECONFIG => undef, # path to siteconfig file
103     EVENTS => 1000, # number of events per job
104     PATTERN => "*", # DBS discovery pattern
105     VALIDTIERS => undef, # perl string pattern with valid tiers (DIGI|RECO)
106     SIMCFG => undef, # CMSSW parameter-set for SIM tier
107     DIGICFG => undef, # CMSSW parameter-set for DIGI tier
108     RECOCFG => undef, # CMSSW parameter-set for RECO tier
109     SIMOUTPUT => undef, # output (comma separated) for SIM tier
110     DIGIOUTPUT => undef, # output (comma separated) for DIGI tier
111     RECOOUTPUT => undef, # output (comma separated) for RECO tier
112     SCHEDULER => "edg"); # standard scheduler
113 gutsche 1.1 my %args = (@_);
114     map { $self->{$_} = defined $args{$_} ? $args{$_} : $params{$_} } keys %params;
115     bless $self, $class;
116     return $self;
117 gutsche 1.21 }
118 gutsche 1.1
119     sub init
120 gutsche 1.21 {
121 gutsche 1.1 my ($self) = @_;
122     $self->{TASKREPO} = "$self->{DROPDIR}/tasks";
123     -d "$self->{TASKREPO}"
124 gutsche 1.21 || mkdir "$self->{TASKREPO}"
125 gutsche 1.1 || die "$self->{TASKREPO}: cannot create directory: $!\n";
126    
127     # Determine if links supports -dump-width option
128     $self->{LINKS_OPTS} = [];
129     open (LINKS_HELP, "links -help 2>/dev/null |");
130 gutsche 1.4 if ( grep(/-no-numbering/, <LINKS_HELP>) ) {
131     push(@{$self->{LINKS_OPTS}}, qw(-dump-width 300 -no-numbering 1));
132     } elsif ( grep(/-dump-width/, <LINKS_HELP>) ) {
133     push(@{$self->{LINKS_OPTS}}, qw(-dump-width 300));
134 gutsche 1.1 }
135     close (LINKS_HELP);
136 gutsche 1.21 }
137 gutsche 1.1
138     # Find out how many jobs are pending for each site. This is
139     # insensitive to the job type, and we only check once in the
140     # beginning to avoid favouring one dataset over another --
141     # once we decide to proceed for a site, we submit jobs for
142     # all datasets.
143     sub getSiteStatus
144 gutsche 1.18 {
145 gutsche 1.1 my ($self) = @_;
146     my %result = ();
147     my $taskrepo = $self->{TASKREPO};
148 gutsche 1.21 foreach my $site (<$taskrepo/*/*>) {
149     my ($sitename) = ($site =~ m|.*/(.*)|);
150     foreach my $d (<$site/*/*>) {
151     if (! -f "$d/JOB_CREATE_LOG.txt") {
152     $result{$sitename}{C} ||= 0;
153     $result{$sitename}{C}++;
154     next;
155     }
156 gutsche 1.18
157 gutsche 1.21 my $f = (<$d/crab_*/share/db/jobs>)[0];
158     next if ! defined $f;
159 gutsche 1.18
160 gutsche 1.21 foreach my $status (split(/\n/, &input($f) || '')) {
161     my @statusarray = split('\|', $status);
162     $result{$sitename}{$statusarray[1]} ||= 0;
163     $result{$sitename}{$statusarray[1]}++;
164     }
165 gutsche 1.18 }
166 gutsche 1.21 }
167 gutsche 1.18
168 gutsche 1.1 return %result;
169 gutsche 1.18 }
170 gutsche 1.1
171 gutsche 1.21 sub createHashFromString
172     {
173     my ($self, $string) = @_;
174     my $cmd = "mktemp /tmp/hash_XXXXXX";
175     open (HASHFILE, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
176     my $hash_file_name;
177     foreach my $hash_line (split(/\n/, <HASHFILE> || '')) {
178     chomp($hash_line);
179     $hash_file_name = $hash_line;
180     }
181     $cmd = "echo $string >> $hash_file_name";
182     system $cmd;
183     # open (FILLHASHFILE, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
184     $cmd = "cksum $hash_file_name";
185     open (CKSUMHASHFILE, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
186     my ($hash, $dummy_one, $dummy_two);
187     foreach my $hash_output (split(/\n/, <CKSUMHASHFILE> || '')) {
188     chomp($hash_output);
189     ($hash, $dummy_one, $dummy_two) = split(/ /,$hash_output);
190     }
191     $cmd = "rm -f $hash_file_name";
192     system $cmd;
193     # open (RMHASHFILE, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
194    
195     return $hash;
196     }
197    
198 gutsche 1.1 sub idle
199     {
200 gutsche 1.21 &logmsg ("DBS/DLS discovery");
201 gutsche 1.1 my ($self, @pending) = @_;
202     eval {
203     # Get status of how busy the sites are. We obtain this only once
204     # in order to not favour datasets "early on" in the list.
205     my %sitestats = $self->getSiteStatus ();
206 gutsche 1.21 if (keys %sitestats) {
207     my @load;
208     foreach my $site (sort keys %sitestats) {
209     push (@load, "$site" . join("", map { " $_=$sitestats{$site}{$_}" }
210     sort keys %{$sitestats{$site}}));
211 gutsche 1.1 }
212 gutsche 1.21 &logmsg ("current load: ", join ("; ", @load));
213     }
214 gutsche 1.1
215     # Invoke links to fetch a formatted web page of published datasets.
216 gutsche 1.21 my $cmd = $ENV{PYTHONSCRIPT} . "/DBSDLSQuery.py --pattern \"$self->{PATTERN}\"";
217     open (PUBLISHED, "$cmd 2>/dev/null |") or die "cannot run `$cmd': $!\n";
218     while (<PUBLISHED>) {
219     chomp;
220     &timeStart($self->{STARTTIME});
221     my ($datapath, $site, $totalevents) = split(/ /, $_);
222     # read in siteconfig dynamically
223     if ( $self->{SITECONFIG} ) {
224     open(siteconfig_file,"$self->{SITECONFIG}");
225     my @line_array = <siteconfig_file>;
226     my $validsites = "(";
227     for (my $i = 0; $i < scalar @line_array; ++$i) {
228     if (defined $line_array[$i] ) {
229     if ( $line_array[$i] !~ /^#/ && $line_array[$i] !~ /^\n/ ) {
230     my ($tempsite, $tempshortsite, $templimit) = split(/ /,$line_array[$i]);
231     chomp($tempsite);
232     $validsites = $validsites . $tempsite . "|";
233 gutsche 1.19 }
234     }
235 gutsche 1.5 }
236 gutsche 1.21 chop($validsites);
237     $validsites = $validsites . ")";
238     close siteconfig_file;
239     next if ($validsites && $site !~ /$validsites/);
240 gutsche 1.5 }
241 gutsche 1.21 $self->createTask($datapath, $site, $totalevents, %sitestats);
242 gutsche 1.1 }
243     };
244     do { chomp ($@); &alert ($@); } if $@;
245    
246     $self->nap ($self->{WAITTIME});
247     }
248    
249     sub createTask()
250     {
251 corvo 1.3 my ($self) = shift(@_);
252 gutsche 1.5 my ($datasetpath, $site, $totalevents, %sitestats) = @_;
253    
254     my($n, $dataset, $tier, $owner) = split(/\//, $datasetpath);
255    
256 gutsche 1.21 # filter on tier
257     next if ($self->{VALIDTIERS} && $tier !~ /$self->{VALIDTIERS}/);
258    
259     my ($cfg, $output);
260     if ($tier =~ /RECO/) {
261     $cfg = $self->{RECOCFG};
262     $output = $self->{RECOOUTPUT};
263     } elsif ($tier =~ /DIGI/) {
264     $cfg = $self->{DIGICFG};
265     $output = $self->{DIGIOUTPUT};
266     } elsif ($tier =~ /SIM/) {
267     $cfg = $self->{SIMCFG};
268     $output = $self->{SIMOUTPUT};
269     } else {
270     &logmsg ("No valid tier, defaults for cfg and output");
271     $cfg = "robot.cfg";
272     $output = "";
273 gutsche 1.5 }
274    
275 gutsche 1.8 # take site specific or if not found default max site queue
276 gutsche 1.20 # read in from siteconfig file
277     my %sitelimits = ();
278 gutsche 1.21 my %siteNameTranslation = ();
279 gutsche 1.20 if ( $self->{SITECONFIG} ) {
280     open(siteconfig_file,"$self->{SITECONFIG}");
281     my @line_array = <siteconfig_file>;
282     for (my $i = 0; $i < scalar @line_array; ++$i) {
283     if (defined $line_array[$i] ) {
284     if ( $line_array[$i] !~ /^#/ && $line_array[$i] !~ /^\n/ ) {
285 gutsche 1.21 my ($tempsite, $tempshortsite, $templimit) = split(/ /,$line_array[$i]);
286 gutsche 1.20 chomp($tempsite);
287 gutsche 1.21 chomp($tempshortsite);
288 gutsche 1.20 chomp($templimit);
289     $sitelimits{$tempsite} = $templimit;
290 gutsche 1.21 $siteNameTranslation{$tempsite} = $tempshortsite;
291 gutsche 1.20 }
292     }
293     }
294     close siteconfig_file;
295     }
296 gutsche 1.21
297     # If the site isn't too busy already, ignore.
298     my $pending = ($sitestats{$siteNameTranslation{$site}}{S} || 0);
299     $pending += ($sitestats{$siteNameTranslation{$site}}{C} || 0);
300 gutsche 1.20 my $max_queue = $sitelimits{$site} || $self->{MAX_SITE_QUEUE};
301 gutsche 1.21 if ( $pending gt $max_queue ) {
302 gutsche 1.4 next;
303     }
304    
305     # Find out what is already pending for this task. First find all
306     # existing tasks in the repository, the latest generation.
307     my $datestamp = strftime ("%y%m%d", gmtime(time()));
308 gutsche 1.21 my $taskdir = "$self->{TASKREPO}/$datestamp/$siteNameTranslation{$site}/$tier";
309    
310     # create short unique taskbase from $datestamp.$site.$dataset.$tier.$owner
311     my $taskbase = $self->createHashFromString ("$datestamp.$site.$dataset.$tier.$owner");
312     chomp($taskbase);
313    
314 gutsche 1.4 my @existing = sort { $a <=> $b } map { /.*\.(\d+)$/ } <$taskdir/$taskbase.*>;
315     my $curgen = pop(@existing) || 0;
316     my $nextgen = $curgen + 1;
317    
318     # OK to create the task if enough time has passed from previous
319     # task creation, or there is no previous task.
320     if (! -f "$taskdir/$taskbase.$curgen/crab.cfg"
321 gutsche 1.21 || (((stat("$taskdir/$taskbase.$curgen/crab.cfg"))[9]
322     < time() - $self->{SECS_BETWEEN_INITS})))
323 gutsche 1.4 {
324     my $mydir = $0; $mydir =~ s|/[^/]+$||;
325     my $drop = sprintf("%s.%03d", $taskbase, $nextgen);
326 gutsche 1.21 my $ret = &runcmd ("$mydir/CrabJobs",
327     "-cfg", "$mydir/$cfg",
328     "-datasetpath", $datasetpath,
329 gutsche 1.4 "-output", $output,
330 gutsche 1.21 "-totalevents", $totalevents,
331     "-whitelist", $site,
332     "-eventsperjob", $self->{EVENTS},
333 gutsche 1.4 "-scheduler", $self->{SCHEDULER},
334 gutsche 1.21 "-jobname", "$taskdir/$drop");
335 gutsche 1.4 die "$drop: failed to create task: @{[&runerror($ret)]}\n" if $ret;
336 gutsche 1.21
337 gutsche 1.4 &output ("$taskdir/$drop/TASK_INIT.txt",
338     &mytimeofday () . "\n");
339    
340     my $dropdir = "$self->{WORKDIR}/$drop";
341     mkdir "$dropdir" || die "$dropdir: cannot create: $!\n";
342 gutsche 1.21 if (&output ("$dropdir/task", "$taskdir/$drop")) {
343     &touch ("$dropdir/done");
344     $self->relayDrop ($drop);
345     &logmsg("stats: $drop @{[&formatElapsedTime($self->{STARTTIME})]} success");
346     } else {
347     &alert ("$drop: failed to create drop");
348     &rmtree ([ "$self->{WORKDIR}/$drop" ]);
349     }
350 gutsche 1.4 }
351     }