X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/4d84c7d2c66ea255db24e2b4159bcafed29ef00d..7ea5d6161e4eab67795fbe2758d98822a9a03f13:/sdk/cli/bin/crunch-job diff --git a/sdk/cli/bin/crunch-job b/sdk/cli/bin/crunch-job index 8b4717734a..d5edf0beed 100755 --- a/sdk/cli/bin/crunch-job +++ b/sdk/cli/bin/crunch-job @@ -82,6 +82,7 @@ use IPC::Open2; use IO::Select; use File::Temp; use Fcntl ':flock'; +use File::Path qw( make_path ); $ENV{"TMPDIR"} ||= "/tmp"; unless (defined $ENV{"CRUNCH_TMP"}) { @@ -91,20 +92,17 @@ unless (defined $ENV{"CRUNCH_TMP"}) { $ENV{"CRUNCH_TMP"} .= "-$<"; } } + +# Create the tmp directory if it does not exist +if ( ! -d $ENV{"CRUNCH_TMP"} ) { + make_path $ENV{"CRUNCH_TMP"} or die "Failed to create temporary working directory: " . $ENV{"CRUNCH_TMP"}; +} + $ENV{"JOB_WORK"} = $ENV{"CRUNCH_TMP"} . "/work"; $ENV{"CRUNCH_INSTALL"} = "$ENV{CRUNCH_TMP}/opt"; $ENV{"CRUNCH_WORK"} = $ENV{"JOB_WORK"}; # deprecated mkdir ($ENV{"JOB_WORK"}); -my $arv_cli; - -if (defined $ENV{"ARV_CLI"}) { - $arv_cli = $ENV{"ARV_CLI"}; -} -else { - $arv_cli = 'arv'; -} - my $force_unlock; my $git_dir; my $jobspec; @@ -152,17 +150,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 111 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'} . ", exiting 111"); + exit(111); } 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(111); } if ($Job->{'running'}) { - croak("Job 'running' flag is already set"); + Log(undef, "Job 'running' flag is already set"); + exit(111); } 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(111); } } } @@ -275,7 +281,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 111"); + exit(111); } $Job->update_attributes('started_at' => scalar gmtime, 'running' => 1, @@ -501,13 +508,24 @@ if (!$have_slurm) # If this job requires a Docker image, install that. my $docker_bin = "/usr/bin/docker.io"; -my $docker_image = $Job->{runtime_constraints}->{docker_image} || ""; -if ($docker_image) { +my ($docker_locator, $docker_stream, $docker_hash); +if ($docker_locator = $Job->{docker_image_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_stream/$docker_hash.tar\E | $docker_bin load +fi +}; my $docker_pid = fork(); if ($docker_pid == 0) { - srun (["srun", "--nodelist=" . join(' ', @node)], - [$docker_bin, 'pull', $docker_image]); + srun (["srun", "--nodelist=" . join(',', @node)], + ["/bin/sh", "-ec", $docker_install_script]); exit ($?); } while (1) @@ -516,11 +534,9 @@ if ($docker_image) { freeze_if_want_freeze ($docker_pid); select (undef, undef, undef, 0.1); } - # If the Docker image was specified as a hash, pull will fail. - # Ignore that error. We'll see what happens when we try to run later. - if (($? != 0) && ($docker_image !~ /^[0-9a-fA-F]{5,64}$/)) + if ($? != 0) { - croak("Installing Docker image $docker_image returned exit code $?"); + croak("Installing Docker image from $docker_locator returned exit code $?"); } } @@ -614,6 +630,7 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++) $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name}; $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu}; $ENV{"TASK_WORK"} = $ENV{"JOB_WORK"}."/$id.$$"; + $ENV{"HOME"} = $ENV{"TASK_WORK"}; $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}.".keep"; $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated $ENV{"CRUNCH_NODE_SLOTS"} = $slot[$childslot]->{node}->{ncpus}; @@ -630,7 +647,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{JOB_WORK} $ENV{CRUNCH_TMP} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT}" ."&& cd $ENV{CRUNCH_TMP} "; if ($build_script) { @@ -639,35 +656,50 @@ for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++) "&& perl -"; } $command .= "&& exec arv-mount --allow-other $ENV{TASK_KEEPMOUNT} --exec "; - if ($docker_image) + if ($docker_hash) { $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$ENV{TASK_WORK}/docker.cid -poll=10000 "; - $command .= "$docker_bin run -i -a stdin -a stdout -a stderr --cidfile=$ENV{TASK_WORK}/docker.cid "; + $command .= "$docker_bin run --rm=true --attach=stdout --attach=stderr --user=crunch --cidfile=$ENV{TASK_WORK}/docker.cid "; # Dynamically configure the container to use the host system as its # DNS server. Get the host's global addresses from the ip command, # and turn them into docker --dns options using gawk. $command .= q{$(ip -o address show scope global | gawk 'match($4, /^([0-9\.:]+)\//, x){print "--dns", x[1]}') }; - foreach my $env_key (qw(CRUNCH_SRC CRUNCH_TMP TASK_KEEPMOUNT)) - { - $command .= "-v \Q$ENV{$env_key}:$ENV{$env_key}:rw\E "; - } + $command .= "--volume=\Q$ENV{CRUNCH_SRC}:/tmp/crunch-src:ro\E "; + $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E "; + $command .= "--env=\QHOME=/home/crunch\E "; while (my ($env_key, $env_val) = each %ENV) { - if ($env_key =~ /^(JOB|TASK)_/) { - $command .= "-e \Q$env_key=$env_val\E "; + 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") { + $command .= "--env=\QTASK_KEEPMOUNT=/keep\E "; + } + else { + $command .= "--env=\Q$env_key=$env_val\E "; + } } } - $command .= "\Q$docker_image\E "; + $command .= "--env=\QCRUNCH_NODE_SLOTS=$ENV{CRUNCH_NODE_SLOTS}\E "; + $command .= "--env=\QCRUNCH_SRC=/tmp/crunch-src\E "; + $command .= "\Q$docker_hash\E "; + $command .= "stdbuf --output=0 --error=0 "; + $command .= "/tmp/crunch-src/crunch_scripts/" . $Job->{"script"}; } else { - $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 " + # Non-docker run + $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 "; + $command .= "stdbuf --output=0 --error=0 "; + $command .= "$ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"}; } - $command .= "stdbuf -o0 -e0 "; - $command .= "$ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"}; + 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) @@ -814,10 +846,12 @@ 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', 'keep', 'get', $collated_output) + open(my $orig_manifest, '-|', 'arv-get', $collated_output) or die "failed to get collated manifest: $!"; # Read the original manifest, and strip permission hints from it, # so we can put the result in a Collection. @@ -838,7 +872,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', @@ -970,27 +1005,29 @@ sub reapchildren push @freeslot, $proc{$pid}->{slot}; delete $proc{$pid}; - # Load new tasks - my $newtask_list = []; - my $newtask_results; - do { - $newtask_results = $arv->{'job_tasks'}->{'list'}->execute( - 'where' => { - 'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid} - }, - 'order' => 'qsequence', - 'offset' => scalar(@$newtask_list), - ); - push(@$newtask_list, @{$newtask_results->{items}}); - } while (@{$newtask_results->{items}}); - foreach my $arvados_task (@$newtask_list) { - my $jobstep = { - 'level' => $arvados_task->{'sequence'}, - 'failures' => 0, - 'arvados_task' => $arvados_task - }; - push @jobstep, $jobstep; - push @jobstep_todo, $#jobstep; + if ($task_success) { + # Load new tasks + my $newtask_list = []; + my $newtask_results; + do { + $newtask_results = $arv->{'job_tasks'}->{'list'}->execute( + 'where' => { + 'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid} + }, + 'order' => 'qsequence', + 'offset' => scalar(@$newtask_list), + ); + push(@$newtask_list, @{$newtask_results->{items}}); + } while (@{$newtask_results->{items}}); + foreach my $arvados_task (@$newtask_list) { + my $jobstep = { + 'level' => $arvados_task->{'sequence'}, + 'failures' => 0, + 'arvados_task' => $arvados_task + }; + push @jobstep, $jobstep; + push @jobstep_todo, $#jobstep; + } } $progress_is_dirty = 1; @@ -1155,9 +1192,22 @@ sub fetch_block my $hash = shift; my ($keep, $child_out, $output_block); - my $cmd = "$arv_cli keep get \Q$hash\E"; + 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; } @@ -1167,7 +1217,7 @@ sub collate_output Log (undef, "collate"); my ($child_out, $child_in); - my $pid = open2($child_out, $child_in, $arv_cli, 'keep', 'put', '--raw'); + my $pid = open2($child_out, $child_in, 'arv-put', '--raw'); my $joboutput; for (@jobstep) { @@ -1204,20 +1254,11 @@ sub collate_output sysread($child_out, $joboutput, 64 * 1024 * 1024); chomp($joboutput); } else { - Log (undef, "timed out reading from 'arv keep put'"); + Log (undef, "timed out reading from 'arv-put'"); } } 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; } @@ -1317,7 +1358,7 @@ sub save_meta return if $justcheckpoint; # checkpointing is not relevant post-Warehouse.pm $local_logfile->flush; - my $cmd = "$arv_cli keep put --filename ''\Q$keep_logfile\E " + my $cmd = "arv-put --filename ''\Q$keep_logfile\E " . quotemeta($local_logfile->filename); my $loglocator = `$cmd`; die "system $cmd failed: $?" if $?; @@ -1437,6 +1478,21 @@ sub must_lock_now } } +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; + 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 (undef, undef); +} + __DATA__ #!/usr/bin/perl