X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/d13e8de76d48788a9198d4581205995f45464ff3..41a6554ca2281983645cf606ba6291cc2332dced:/sdk/cli/bin/crunch-job diff --git a/sdk/cli/bin/crunch-job b/sdk/cli/bin/crunch-job index 30f06b8cab..638353d76f 100755 --- a/sdk/cli/bin/crunch-job +++ b/sdk/cli/bin/crunch-job @@ -84,6 +84,8 @@ use File::Temp; use Fcntl ':flock'; use File::Path qw( make_path ); +use constant EX_TEMPFAIL => 75; + $ENV{"TMPDIR"} ||= "/tmp"; unless (defined $ENV{"CRUNCH_TMP"}) { $ENV{"CRUNCH_TMP"} = $ENV{"TMPDIR"} . "/crunch-job"; @@ -150,17 +152,25 @@ if ($job_has_uuid) { $Job = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec); if (!$force_unlock) { + # If some other crunch-job process has grabbed this job (or we see + # other evidence that the job is already underway) we exit + # EX_TEMPFAIL so crunch-dispatch (our parent process) doesn't + # mark the job as failed. if ($Job->{'is_locked_by_uuid'}) { - croak("Job is locked: " . $Job->{'is_locked_by_uuid'}); + Log(undef, "Job is locked by " . $Job->{'is_locked_by_uuid'}); + exit EX_TEMPFAIL; } if ($Job->{'success'} ne undef) { - croak("Job 'success' flag (" . $Job->{'success'} . ") is not null"); + Log(undef, "Job 'success' flag (" . $Job->{'success'} . ") is not null"); + exit EX_TEMPFAIL; } if ($Job->{'running'}) { - croak("Job 'running' flag is already set"); + Log(undef, "Job 'running' flag is already set"); + exit EX_TEMPFAIL; } if ($Job->{'started_at'}) { - croak("Job 'started_at' time is already set (" . $Job->{'started_at'} . ")"); + Log(undef, "Job 'started_at' time is already set (" . $Job->{'started_at'} . ")"); + exit EX_TEMPFAIL; } } } @@ -273,7 +283,8 @@ if ($job_has_uuid) # Claim this job, and make sure nobody else does unless ($Job->update_attributes('is_locked_by_uuid' => $User->{'uuid'}) && $Job->{'is_locked_by_uuid'} == $User->{'uuid'}) { - croak("Error while updating / locking job"); + Log(undef, "Error while updating / locking job, exiting ".EX_TEMPFAIL); + exit EX_TEMPFAIL; } $Job->update_attributes('started_at' => scalar gmtime, 'running' => 1, @@ -383,7 +394,7 @@ else if ($cleanpid == 0) { srun (["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}], - ['bash', '-c', 'if mount | grep -q $JOB_WORK/; then sudo /bin/umount $JOB_WORK/* 2>/dev/null; fi; sleep 1; rm -rf $JOB_WORK $CRUNCH_TMP/opt $CRUNCH_TMP/src*']); + ['bash', '-c', 'if mount | grep -q $JOB_WORK/; then for i in $JOB_WORK/*keep; do /bin/fusermount -z -u $i; done; fi; sleep 1; rm -rf $JOB_WORK $CRUNCH_TMP/opt $CRUNCH_TMP/src*']); exit (1); } while (1) @@ -499,16 +510,17 @@ if (!$have_slurm) # If this job requires a Docker image, install that. my $docker_bin = "/usr/bin/docker.io"; -my ($docker_locator, $docker_hash); +my ($docker_locator, $docker_stream, $docker_hash); if ($docker_locator = $Job->{docker_image_locator}) { - $docker_hash = find_docker_hash($docker_locator); + ($docker_stream, $docker_hash) = find_docker_image($docker_locator); if (!$docker_hash) { croak("No Docker image hash found from locator $docker_locator"); } + $docker_stream =~ s/^\.//; my $docker_install_script = qq{ if ! $docker_bin images -q --no-trunc | grep -qxF \Q$docker_hash\E; then - arv-get \Q$docker_locator/$docker_hash.tar\E | $docker_bin load + arv-get \Q$docker_locator$docker_stream/$docker_hash.tar\E | $docker_bin load fi }; my $docker_pid = fork(); @@ -637,7 +649,7 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++) my $build_script_to_send = ""; my $command = "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; " - ."mkdir -p $ENV{JOB_WORK} $ENV{CRUNCH_TMP} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT}" + ."mkdir -p $ENV{CRUNCH_TMP} $ENV{JOB_WORK} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} " ."&& cd $ENV{CRUNCH_TMP} "; if ($build_script) { @@ -662,10 +674,7 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++) while (my ($env_key, $env_val) = each %ENV) { if ($env_key =~ /^(ARVADOS|JOB|TASK)_/) { - if ($env_key eq "TASK_WORK") { - $command .= "--env=\QTASK_WORK=/tmp/crunch-job\E "; - } - elsif ($env_key eq "TASK_KEEPMOUNT") { + if ($env_key eq "TASK_KEEPMOUNT") { $command .= "--env=\QTASK_KEEPMOUNT=/keep\E "; } else { @@ -687,7 +696,9 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++) my @execargs = ('bash', '-c', $command); srun (\@srunargs, \@execargs, undef, $build_script_to_send); - exit (111); + # exec() failed, we assume nothing happened. + Log(undef, "srun() failed on build script"); + die; } close("writer"); if (!defined $childpid) @@ -834,8 +845,10 @@ if ($job_has_uuid) { 'finished_at' => scalar gmtime) } -if ($collated_output) -{ +if (!$collated_output) { + Log(undef, "output undef"); +} +else { eval { open(my $orig_manifest, '-|', 'arv-get', $collated_output) or die "failed to get collated manifest: $!"; @@ -858,7 +871,8 @@ if ($collated_output) 'uuid' => md5_hex($stripped_manifest_text), 'manifest_text' => $orig_manifest_text, }); - $Job->update_attributes('output' => $output->{uuid}); + Log(undef, "output " . $output->{uuid}); + $Job->update_attributes('output' => $output->{uuid}) if $job_has_uuid; if ($Job->{'output_is_persistent'}) { $arv->{'links'}->{'create'}->execute('link' => { 'tail_kind' => 'arvados#user', @@ -878,7 +892,7 @@ if ($collated_output) Log (undef, "finish"); save_meta(); -exit 0; +exit ($Job->{'success'} ? 1 : 0); @@ -1179,7 +1193,20 @@ sub fetch_block my $cmd = "arv-get \Q$hash\E"; open($keep, '-|', $cmd) or die "fetch_block: $cmd: $!"; - sysread($keep, $output_block, 64 * 1024 * 1024); + $output_block = ''; + while (1) { + my $buf; + my $bytes = sysread($keep, $buf, 1024 * 1024); + if (!defined $bytes) { + die "reading from arv-get: $!"; + } elsif ($bytes == 0) { + # sysread returns 0 at the end of the pipe. + last; + } else { + # some bytes were read into buf. + $output_block .= $buf; + } + } close $keep; return $output_block; } @@ -1193,9 +1220,8 @@ sub collate_output my $joboutput; for (@jobstep) { - next if (!exists $_->{'arvados_task'}->{output} || - !$_->{'arvados_task'}->{'success'} || - $_->{'exitcode'} != 0); + next if (!exists $_->{'arvados_task'}->{'output'} || + !$_->{'arvados_task'}->{'success'}); my $output = $_->{'arvados_task'}->{output}; if ($output !~ /^[0-9a-f]{32}(\+\S+)*$/) { @@ -1231,15 +1257,6 @@ sub collate_output } waitpid($pid, 0); - if ($joboutput) - { - Log (undef, "output $joboutput"); - $Job->update_attributes('output' => $joboutput) if $job_has_uuid; - } - else - { - Log (undef, "output undef"); - } return $joboutput; } @@ -1459,19 +1476,19 @@ sub must_lock_now } } -sub find_docker_hash { - # Given a Keep locator, search for a matching link to find the Docker hash - # of the stored image. +sub find_docker_image { + # Given a Keep locator, check to see if it contains a Docker image. + # If so, return its stream name and Docker hash. + # If not, return undef for both values. my $locator = shift; - my $links_result = $arv->{links}->{list}->execute( - filters => [["head_uuid", "=", $locator], - ["link_class", "=", "docker_image_hash"]], - limit => 1); - my $docker_hash; - foreach my $link (@{$links_result->{items}}) { - $docker_hash = lc($link->{name}); + if (my $image = $arv->{collections}->{get}->execute(uuid => $locator)) { + my @file_list = @{$image->{files}}; + if ((scalar(@file_list) == 1) && + ($file_list[0][1] =~ /^([0-9A-Fa-f]{64})\.tar$/)) { + return ($file_list[0][0], $1); + } } - return $docker_hash; + return (undef, undef); } __DATA__ @@ -1480,10 +1497,17 @@ __DATA__ # checkout-and-build use Fcntl ':flock'; +use File::Path qw( make_path ); my $destdir = $ENV{"CRUNCH_SRC"}; my $commit = $ENV{"CRUNCH_SRC_COMMIT"}; my $repo = $ENV{"CRUNCH_SRC_URL"}; +my $task_work = $ENV{"TASK_WORK"}; + +if ($task_work) { + make_path $task_work; + -e $task_work or die "Failed to create temporary working directory ($task_work): $!"; +} open L, ">", "$destdir.lock" or die "$destdir.lock: $!"; flock L, LOCK_EX;