sdk/cli/bin/crunch-job

   1 #!/usr/bin/perl
   2 # -*- mode: perl; perl-indent-level: 2; indent-tabs-mode: nil; -*-
   3
   4 =head1 NAME
   5
   6 crunch-job: Execute job steps, save snapshots as requested, collate output.
   7
   8 =head1 SYNOPSIS
   9
  10 Obtain job details from Arvados, run tasks on compute nodes (typically
  11 invoked by scheduler on controller):
  12
  13  crunch-job --job x-y-z --git-dir /path/to/repo/.git
  14
  15 Obtain job details from command line, run tasks on local machine
  16 (typically invoked by application or developer on VM):
  17
  18  crunch-job --job '{"script_version":"/path/to/working/tree","script":"scriptname",...}'
  19
  20  crunch-job --job '{"repository":"https://github.com/curoverse/arvados.git","script_version":"master","script":"scriptname",...}'
  21
  22 =head1 OPTIONS
  23
  24 =over
  25
  26 =item --force-unlock
  27
  28 If the job is already locked, steal the lock and run it anyway.
  29
  30 =item --git-dir
  31
  32 Path to a .git directory (or a git URL) where the commit given in the
  33 job's C<script_version> attribute is to be found. If this is I<not>
  34 given, the job's C<repository> attribute will be used.
  35
  36 =item --job-api-token
  37
  38 Arvados API authorization token to use during the course of the job.
  39
  40 =item --no-clear-tmp
  41
  42 Do not clear per-job/task temporary directories during initial job
  43 setup. This can speed up development and debugging when running jobs
  44 locally.
  45
  46 =item --job
  47
  48 UUID of the job to run, or a JSON-encoded job resource without a
  49 UUID. If the latter is given, a new job object will be created.
  50
  51 =back
  52
  53 =head1 RUNNING JOBS LOCALLY
  54
  55 crunch-job's log messages appear on stderr along with the job tasks'
  56 stderr streams. The log is saved in Keep at each checkpoint and when
  57 the job finishes.
  58
  59 If the job succeeds, the job's output locator is printed on stdout.
  60
  61 While the job is running, the following signals are accepted:
  62
  63 =over
  64
  65 =item control-C, SIGINT, SIGQUIT
  66
  67 Save a checkpoint, terminate any job tasks that are running, and stop.
  68
  69 =item SIGALRM
  70
  71 Save a checkpoint and continue.
  72
  73 =item SIGHUP
  74
  75 Refresh node allocation (i.e., check whether any nodes have been added
  76 or unallocated) and attributes of the Job record that should affect
  77 behavior (e.g., cancel job if cancelled_at becomes non-nil).
  78
  79 =back
  80
  81 =cut
  82
  83
  84 use strict;
  85 use POSIX ':sys_wait_h';
  86 use POSIX qw(strftime);
  87 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
  88 use Arvados;
  89 use Cwd qw(realpath);
  90 use Data::Dumper;
  91 use Digest::MD5 qw(md5_hex);
  92 use Getopt::Long;
  93 use IPC::Open2;
  94 use IO::Select;
  95 use File::Temp;
  96 use Fcntl ':flock';
  97 use File::Path qw( make_path remove_tree );
  98
  99 use constant TASK_TEMPFAIL => 111;
 100 use constant EX_TEMPFAIL => 75;
 101
 102 $ENV{"TMPDIR"} ||= "/tmp";
 103 unless (defined $ENV{"CRUNCH_TMP"}) {
 104   $ENV{"CRUNCH_TMP"} = $ENV{"TMPDIR"} . "/crunch-job";
 105   if ($ENV{"USER"} ne "crunch" && $< != 0) {
 106     # use a tmp dir unique for my uid
 107     $ENV{"CRUNCH_TMP"} .= "-$<";
 108   }
 109 }
 110
 111 # Create the tmp directory if it does not exist
 112 if ( ! -d $ENV{"CRUNCH_TMP"} ) {
 113   make_path $ENV{"CRUNCH_TMP"} or die "Failed to create temporary working directory: " . $ENV{"CRUNCH_TMP"};
 114 }
 115
 116 $ENV{"JOB_WORK"} = $ENV{"CRUNCH_TMP"} . "/work";
 117 $ENV{"CRUNCH_INSTALL"} = "$ENV{CRUNCH_TMP}/opt";
 118 $ENV{"CRUNCH_WORK"} = $ENV{"JOB_WORK"}; # deprecated
 119 mkdir ($ENV{"JOB_WORK"});
 120
 121 my $force_unlock;
 122 my $git_dir;
 123 my $jobspec;
 124 my $job_api_token;
 125 my $no_clear_tmp;
 126 my $resume_stash;
 127 GetOptions('force-unlock' => \$force_unlock,
 128            'git-dir=s' => \$git_dir,
 129            'job=s' => \$jobspec,
 130            'job-api-token=s' => \$job_api_token,
 131            'no-clear-tmp' => \$no_clear_tmp,
 132            'resume-stash=s' => \$resume_stash,
 133     );
 134
 135 if (defined $job_api_token) {
 136   $ENV{ARVADOS_API_TOKEN} = $job_api_token;
 137 }
 138
 139 my $have_slurm = exists $ENV{SLURM_JOBID} && exists $ENV{SLURM_NODELIST};
 140 my $local_job = 0;
 141
 142
 143 $SIG{'USR1'} = sub
 144 {
 145   $main::ENV{CRUNCH_DEBUG} = 1;
 146 };
 147 $SIG{'USR2'} = sub
 148 {
 149   $main::ENV{CRUNCH_DEBUG} = 0;
 150 };
 151
 152
 153
 154 my $arv = Arvados->new('apiVersion' => 'v1');
 155
 156 my $Job;
 157 my $job_id;
 158 my $dbh;
 159 my $sth;
 160 my @jobstep;
 161
 162 my $User = api_call("users/current");
 163
 164 if ($jobspec =~ /^[-a-z\d]+$/)
 165 {
 166   # $jobspec is an Arvados UUID, not a JSON job specification
 167   $Job = api_call("jobs/get", uuid => $jobspec);
 168   if (!$force_unlock) {
 169     # Claim this job, and make sure nobody else does
 170     eval { api_call("jobs/lock", uuid => $Job->{uuid}); };
 171     if ($@) {
 172       Log(undef, "Error while locking job, exiting ".EX_TEMPFAIL);
 173       exit EX_TEMPFAIL;
 174     };
 175   }
 176 }
 177 else
 178 {
 179   $Job = JSON::decode_json($jobspec);
 180
 181   if (!$resume_stash)
 182   {
 183     map { croak ("No $_ specified") unless $Job->{$_} }
 184     qw(script script_version script_parameters);
 185   }
 186
 187   $Job->{'is_locked_by_uuid'} = $User->{'uuid'};
 188   $Job->{'started_at'} = gmtime;
 189   $Job->{'state'} = 'Running';
 190
 191   $Job = api_call("jobs/create", job => $Job);
 192 }
 193 $job_id = $Job->{'uuid'};
 194
 195 my $keep_logfile = $job_id . '.log.txt';
 196 log_writer_start($keep_logfile);
 197
 198 $Job->{'runtime_constraints'} ||= {};
 199 $Job->{'runtime_constraints'}->{'max_tasks_per_node'} ||= 0;
 200 my $max_ncpus = $Job->{'runtime_constraints'}->{'max_tasks_per_node'};
 201
 202 my $gem_versions = `gem list --quiet arvados-cli 2>/dev/null`;
 203 if ($? == 0) {
 204   $gem_versions =~ s/^arvados-cli \(/ with arvados-cli Gem version(s) /;
 205   chomp($gem_versions);
 206   chop($gem_versions);  # Closing parentheses
 207 } else {
 208   $gem_versions = "";
 209 }
 210 Log(undef,
 211     "running from " . ((-e $0) ? realpath($0) : "stdin") . $gem_versions);
 212
 213 Log (undef, "check slurm allocation");
 214 my @slot;
 215 my @node;
 216 # Should use $ENV{SLURM_TASKS_PER_NODE} instead of sinfo? (eg. "4(x3),2,4(x2)")
 217 my @sinfo;
 218 if (!$have_slurm)
 219 {
 220   my $localcpus = 0 + `grep -cw ^processor /proc/cpuinfo` || 1;
 221   push @sinfo, "$localcpus localhost";
 222 }
 223 if (exists $ENV{SLURM_NODELIST})
 224 {
 225   push @sinfo, `sinfo -h --format='%c %N' --nodes=\Q$ENV{SLURM_NODELIST}\E`;
 226 }
 227 foreach (@sinfo)
 228 {
 229   my ($ncpus, $slurm_nodelist) = split;
 230   $ncpus = $max_ncpus if $max_ncpus && $ncpus > $max_ncpus;
 231
 232   my @nodelist;
 233   while ($slurm_nodelist =~ s/^([^\[,]+?(\[.*?\])?)(,|$)//)
 234   {
 235     my $nodelist = $1;
 236     if ($nodelist =~ /\[((\d+)(-(\d+))?(,(\d+)(-(\d+))?)*)\]/)
 237     {
 238       my $ranges = $1;
 239       foreach (split (",", $ranges))
 240       {
 241         my ($a, $b);
 242         if (/(\d+)-(\d+)/)
 243         {
 244           $a = $1;
 245           $b = $2;
 246         }
 247         else
 248         {
 249           $a = $_;
 250           $b = $_;
 251         }
 252         push @nodelist, map {
 253           my $n = $nodelist;
 254           $n =~ s/\[[-,\d]+\]/$_/;
 255           $n;
 256         } ($a..$b);
 257       }
 258     }
 259     else
 260     {
 261       push @nodelist, $nodelist;
 262     }
 263   }
 264   foreach my $nodename (@nodelist)
 265   {
 266     Log (undef, "node $nodename - $ncpus slots");
 267     my $node = { name => $nodename,
 268                  ncpus => $ncpus,
 269                  losing_streak => 0,
 270                  hold_until => 0 };
 271     foreach my $cpu (1..$ncpus)
 272     {
 273       push @slot, { node => $node,
 274                     cpu => $cpu };
 275     }
 276   }
 277   push @node, @nodelist;
 278 }
 279
 280
 281
 282 # Ensure that we get one jobstep running on each allocated node before
 283 # we start overloading nodes with concurrent steps
 284
 285 @slot = sort { $a->{cpu} <=> $b->{cpu} } @slot;
 286
 287
 288 $Job->update_attributes(
 289   'tasks_summary' => { 'failed' => 0,
 290                        'todo' => 1,
 291                        'running' => 0,
 292                        'done' => 0 });
 293
 294 Log (undef, "start");
 295 $SIG{'INT'} = sub { $main::please_freeze = 1; };
 296 $SIG{'QUIT'} = sub { $main::please_freeze = 1; };
 297 $SIG{'TERM'} = \&croak;
 298 $SIG{'TSTP'} = sub { $main::please_freeze = 1; };
 299 $SIG{'ALRM'} = sub { $main::please_info = 1; };
 300 $SIG{'CONT'} = sub { $main::please_continue = 1; };
 301 $SIG{'HUP'} = sub { $main::please_refresh = 1; };
 302
 303 $main::please_freeze = 0;
 304 $main::please_info = 0;
 305 $main::please_continue = 0;
 306 $main::please_refresh = 0;
 307 my $jobsteps_must_output_keys = 0;      # becomes 1 when any task outputs a key
 308
 309 grep { $ENV{$1} = $2 if /^(NOCACHE.*?)=(.*)/ } split ("\n", $$Job{knobs});
 310 $ENV{"CRUNCH_JOB_UUID"} = $job_id;
 311 $ENV{"JOB_UUID"} = $job_id;
 312
 313
 314 my @jobstep_todo = ();
 315 my @jobstep_done = ();
 316 my @jobstep_tomerge = ();
 317 my $jobstep_tomerge_level = 0;
 318 my $squeue_checked;
 319 my $squeue_kill_checked;
 320 my $latest_refresh = scalar time;
 321
 322
 323
 324 if (defined $Job->{thawedfromkey})
 325 {
 326   thaw ($Job->{thawedfromkey});
 327 }
 328 else
 329 {
 330   my $first_task = api_call("job_tasks/create", job_task => {
 331     'job_uuid' => $Job->{'uuid'},
 332     'sequence' => 0,
 333     'qsequence' => 0,
 334     'parameters' => {},
 335   });
 336   push @jobstep, { 'level' => 0,
 337                    'failures' => 0,
 338                    'arvados_task' => $first_task,
 339                  };
 340   push @jobstep_todo, 0;
 341 }
 342
 343
 344 if (!$have_slurm)
 345 {
 346   must_lock_now("$ENV{CRUNCH_TMP}/.lock", "a job is already running here.");
 347 }
 348
 349 my $build_script = handle_readall(\*DATA);
 350 my $nodelist = join(",", @node);
 351 my $git_tar_count = 0;
 352
 353 if (!defined $no_clear_tmp) {
 354   # Clean out crunch_tmp/work, crunch_tmp/opt, crunch_tmp/src*
 355   Log (undef, "Clean work dirs");
 356
 357   my $cleanpid = fork();
 358   if ($cleanpid == 0)
 359   {
 360     # Find FUSE mounts that look like Keep mounts (the mount path has the
 361     # word "keep") and unmount them.  Then clean up work directories.
 362     # TODO: When #5036 is done and widely deployed, we can get rid of the
 363     # regular expression and just unmount everything with type fuse.keep.
 364     srun (["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}],
 365           ['bash', '-ec', 'mount -t fuse,fuse.keep | awk \'($3 ~ /\ykeep\y/){print $3}\' | xargs -r -n 1 fusermount -u -z; sleep 1; rm -rf $JOB_WORK $CRUNCH_INSTALL $CRUNCH_TMP/task $CRUNCH_TMP/src* $CRUNCH_TMP/*.cid']);
 366     exit (1);
 367   }
 368   while (1)
 369   {
 370     last if $cleanpid == waitpid (-1, WNOHANG);
 371     freeze_if_want_freeze ($cleanpid);
 372     select (undef, undef, undef, 0.1);
 373   }
 374   Log (undef, "Cleanup command exited ".exit_status_s($?));
 375 }
 376
 377 # If this job requires a Docker image, install that.
 378 my $docker_bin = "/usr/bin/docker.io";
 379 my ($docker_locator, $docker_stream, $docker_hash, $docker_limitmem);
 380 if ($docker_locator = $Job->{docker_image_locator}) {
 381   ($docker_stream, $docker_hash) = find_docker_image($docker_locator);
 382   if (!$docker_hash)
 383   {
 384     croak("No Docker image hash found from locator $docker_locator");
 385   }
 386   $docker_stream =~ s/^\.//;
 387   my $docker_install_script = qq{
 388 if ! $docker_bin images -q --no-trunc --all | grep -qxF \Q$docker_hash\E; then
 389     arv-get \Q$docker_locator$docker_stream/$docker_hash.tar\E | $docker_bin load
 390 fi
 391 };
 392   my $docker_pid = fork();
 393   if ($docker_pid == 0)
 394   {
 395     srun (["srun", "--nodelist=" . join(',', @node)],
 396           ["/bin/sh", "-ec", $docker_install_script]);
 397     exit ($?);
 398   }
 399   while (1)
 400   {
 401     last if $docker_pid == waitpid (-1, WNOHANG);
 402     freeze_if_want_freeze ($docker_pid);
 403     select (undef, undef, undef, 0.1);
 404   }
 405   if ($? != 0)
 406   {
 407     croak("Installing Docker image from $docker_locator exited "
 408           .exit_status_s($?));
 409   }
 410
 411   # Determine whether this version of Docker supports memory+swap limits.
 412   srun(["srun", "--nodelist=" . $node[0]],
 413        ["/bin/sh", "-ec", "$docker_bin run --help | grep -qe --memory-swap="],
 414       {fork => 1});
 415   $docker_limitmem = ($? == 0);
 416
 417   if ($Job->{arvados_sdk_version}) {
 418     # The job also specifies an Arvados SDK version.  Add the SDKs to the
 419     # tar file for the build script to install.
 420     Log(undef, sprintf("Packing Arvados SDK version %s for installation",
 421                        $Job->{arvados_sdk_version}));
 422     add_git_archive("git", "--git-dir=$git_dir", "archive",
 423                     "--prefix=.arvados.sdk/",
 424                     $Job->{arvados_sdk_version}, "sdk");
 425   }
 426 }
 427
 428 if (!defined $git_dir && $Job->{'script_version'} =~ m{^/}) {
 429   # If script_version looks like an absolute path, *and* the --git-dir
 430   # argument was not given -- which implies we were not invoked by
 431   # crunch-dispatch -- we will use the given path as a working
 432   # directory instead of resolving script_version to a git commit (or
 433   # doing anything else with git).
 434   $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{'script_version'};
 435   $ENV{"CRUNCH_SRC"} = $Job->{'script_version'};
 436 }
 437 else {
 438   # Resolve the given script_version to a git commit sha1. Also, if
 439   # the repository is remote, clone it into our local filesystem: this
 440   # ensures "git archive" will work, and is necessary to reliably
 441   # resolve a symbolic script_version like "master^".
 442   $ENV{"CRUNCH_SRC"} = "$ENV{CRUNCH_TMP}/src";
 443
 444   Log (undef, "Looking for version ".$Job->{script_version}." from repository ".$Job->{repository});
 445
 446   $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{script_version};
 447
 448   # If we're running under crunch-dispatch, it will have already
 449   # pulled the appropriate source tree into its own repository, and
 450   # given us that repo's path as $git_dir.
 451   #
 452   # If we're running a "local" job, we might have to fetch content
 453   # from a remote repository.
 454   #
 455   # (Currently crunch-dispatch gives a local path with --git-dir, but
 456   # we might as well accept URLs there too in case it changes its
 457   # mind.)
 458   my $repo = $git_dir || $Job->{'repository'};
 459
 460   # Repository can be remote or local. If remote, we'll need to fetch it
 461   # to a local dir before doing `git log` et al.
 462   my $repo_location;
 463
 464   if ($repo =~ m{://|^[^/]*:}) {
 465     # $repo is a git url we can clone, like git:// or https:// or
 466     # file:/// or [user@]host:repo.git. Note "user/name@host:foo" is
 467     # not recognized here because distinguishing that from a local
 468     # path is too fragile. If you really need something strange here,
 469     # use the ssh:// form.
 470     $repo_location = 'remote';
 471   } elsif ($repo =~ m{^\.*/}) {
 472     # $repo is a local path to a git index. We'll also resolve ../foo
 473     # to ../foo/.git if the latter is a directory. To help
 474     # disambiguate local paths from named hosted repositories, this
 475     # form must be given as ./ or ../ if it's a relative path.
 476     if (-d "$repo/.git") {
 477       $repo = "$repo/.git";
 478     }
 479     $repo_location = 'local';
 480   } else {
 481     # $repo is none of the above. It must be the name of a hosted
 482     # repository.
 483     my $arv_repo_list = api_call("repositories/list",
 484                                  'filters' => [['name','=',$repo]]);
 485     my @repos_found = @{$arv_repo_list->{'items'}};
 486     my $n_found = $arv_repo_list->{'serverResponse'}->{'items_available'};
 487     if ($n_found > 0) {
 488       Log(undef, "Repository '$repo' -> "
 489           . join(", ", map { $_->{'uuid'} } @repos_found));
 490     }
 491     if ($n_found != 1) {
 492       croak("Error: Found $n_found repositories with name '$repo'.");
 493     }
 494     $repo = $repos_found[0]->{'fetch_url'};
 495     $repo_location = 'remote';
 496   }
 497   Log(undef, "Using $repo_location repository '$repo'");
 498   $ENV{"CRUNCH_SRC_URL"} = $repo;
 499
 500   # Resolve given script_version (we'll call that $treeish here) to a
 501   # commit sha1 ($commit).
 502   my $treeish = $Job->{'script_version'};
 503   my $commit;
 504   if ($repo_location eq 'remote') {
 505     # We minimize excess object-fetching by re-using the same bare
 506     # repository in CRUNCH_TMP/.git for multiple crunch-jobs -- we
 507     # just keep adding remotes to it as needed.
 508     my $local_repo = $ENV{'CRUNCH_TMP'}."/.git";
 509     my $gitcmd = "git --git-dir=\Q$local_repo\E";
 510
 511     # Set up our local repo for caching remote objects, making
 512     # archives, etc.
 513     if (!-d $local_repo) {
 514       make_path($local_repo) or croak("Error: could not create $local_repo");
 515     }
 516     # This works (exits 0 and doesn't delete fetched objects) even
 517     # if $local_repo is already initialized:
 518     `$gitcmd init --bare`;
 519     if ($?) {
 520       croak("Error: $gitcmd init --bare exited ".exit_status_s($?));
 521     }
 522
 523     # If $treeish looks like a hash (or abbrev hash) we look it up in
 524     # our local cache first, since that's cheaper. (We don't want to
 525     # do that with tags/branches though -- those change over time, so
 526     # they should always be resolved by the remote repo.)
 527     if ($treeish =~ /^[0-9a-f]{7,40}$/s) {
 528       # Hide stderr because it's normal for this to fail:
 529       my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E 2>/dev/null`;
 530       if ($? == 0 &&
 531           # Careful not to resolve a branch named abcdeff to commit 1234567:
 532           $sha1 =~ /^$treeish/ &&
 533           $sha1 =~ /^([0-9a-f]{40})$/s) {
 534         $commit = $1;
 535         Log(undef, "Commit $commit already present in $local_repo");
 536       }
 537     }
 538
 539     if (!defined $commit) {
 540       # If $treeish isn't just a hash or abbrev hash, or isn't here
 541       # yet, we need to fetch the remote to resolve it correctly.
 542
 543       # First, remove all local heads. This prevents a name that does
 544       # not exist on the remote from resolving to (or colliding with)
 545       # a previously fetched branch or tag (possibly from a different
 546       # remote).
 547       remove_tree("$local_repo/refs/heads", {keep_root => 1});
 548
 549       Log(undef, "Fetching objects from $repo to $local_repo");
 550       `$gitcmd fetch --no-progress --tags ''\Q$repo\E \Q+refs/heads/*:refs/heads/*\E`;
 551       if ($?) {
 552         croak("Error: `$gitcmd fetch` exited ".exit_status_s($?));
 553       }
 554     }
 555
 556     # Now that the data is all here, we will use our local repo for
 557     # the rest of our git activities.
 558     $repo = $local_repo;
 559   }
 560
 561   my $gitcmd = "git --git-dir=\Q$repo\E";
 562   my $sha1 = `$gitcmd rev-list -n1 ''\Q$treeish\E`;
 563   unless ($? == 0 && $sha1 =~ /^([0-9a-f]{40})$/) {
 564     croak("`$gitcmd rev-list` exited "
 565           .exit_status_s($?)
 566           .", '$treeish' not found. Giving up.");
 567   }
 568   $commit = $1;
 569   Log(undef, "Version $treeish is commit $commit");
 570
 571   if ($commit ne $Job->{'script_version'}) {
 572     # Record the real commit id in the database, frozentokey, logs,
 573     # etc. -- instead of an abbreviation or a branch name which can
 574     # become ambiguous or point to a different commit in the future.
 575     if (!$Job->update_attributes('script_version' => $commit)) {
 576       croak("Error: failed to update job's script_version attribute");
 577     }
 578   }
 579
 580   $ENV{"CRUNCH_SRC_COMMIT"} = $commit;
 581   add_git_archive("$gitcmd archive ''\Q$commit\E");
 582 }
 583
 584 my $git_archive = combined_git_archive();
 585 if (!defined $git_archive) {
 586   Log(undef, "Skip install phase (no git archive)");
 587   if ($have_slurm) {
 588     Log(undef, "Warning: This probably means workers have no source tree!");
 589   }
 590 }
 591 else {
 592   Log(undef, "Run install script on all workers");
 593
 594   my @srunargs = ("srun",
 595                   "--nodelist=$nodelist",
 596                   "-D", $ENV{'TMPDIR'}, "--job-name=$job_id");
 597   my @execargs = ("sh", "-c",
 598                   "mkdir -p $ENV{CRUNCH_INSTALL} && cd $ENV{CRUNCH_TMP} && perl -");
 599
 600   my $installpid = fork();
 601   if ($installpid == 0)
 602   {
 603     srun (\@srunargs, \@execargs, {}, $build_script . $git_archive);
 604     exit (1);
 605   }
 606   while (1)
 607   {
 608     last if $installpid == waitpid (-1, WNOHANG);
 609     freeze_if_want_freeze ($installpid);
 610     select (undef, undef, undef, 0.1);
 611   }
 612   my $install_exited = $?;
 613   Log (undef, "Install script exited ".exit_status_s($install_exited));
 614   foreach my $tar_filename (map { tar_filename_n($_); } (1..$git_tar_count)) {
 615     unlink($tar_filename);
 616   }
 617   exit (1) if $install_exited != 0;
 618 }
 619
 620 foreach (qw (script script_version script_parameters runtime_constraints))
 621 {
 622   Log (undef,
 623        "$_ " .
 624        (ref($Job->{$_}) ? JSON::encode_json($Job->{$_}) : $Job->{$_}));
 625 }
 626 foreach (split (/\n/, $Job->{knobs}))
 627 {
 628   Log (undef, "knob " . $_);
 629 }
 630
 631
 632
 633 $main::success = undef;
 634
 635
 636
 637 ONELEVEL:
 638
 639 my $thisround_succeeded = 0;
 640 my $thisround_failed = 0;
 641 my $thisround_failed_multiple = 0;
 642
 643 @jobstep_todo = sort { $jobstep[$a]->{level} <=> $jobstep[$b]->{level}
 644                        or $a <=> $b } @jobstep_todo;
 645 my $level = $jobstep[$jobstep_todo[0]]->{level};
 646 Log (undef, "start level $level");
 647
 648
 649
 650 my %proc;
 651 my @freeslot = (0..$#slot);
 652 my @holdslot;
 653 my %reader;
 654 my $progress_is_dirty = 1;
 655 my $progress_stats_updated = 0;
 656
 657 update_progress_stats();
 658
 659
 660
 661 THISROUND:
 662 my $tasks_this_level = 0;
 663 foreach my $id (@jobstep_todo) {
 664   $tasks_this_level++ if ($jobstep[$id]->{level} == $level);
 665 }
 666 for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
 667 {
 668   my $id = $jobstep_todo[$todo_ptr];
 669   my $Jobstep = $jobstep[$id];
 670   if ($Jobstep->{level} != $level)
 671   {
 672     next;
 673   }
 674
 675   pipe $reader{$id}, "writer" or croak ($!);
 676   my $flags = fcntl ($reader{$id}, F_GETFL, 0) or croak ($!);
 677   fcntl ($reader{$id}, F_SETFL, $flags | O_NONBLOCK) or croak ($!);
 678
 679   my $childslot = $freeslot[0];
 680   my $childnode = $slot[$childslot]->{node};
 681   my $childslotname = join (".",
 682                             $slot[$childslot]->{node}->{name},
 683                             $slot[$childslot]->{cpu});
 684
 685   my $childpid = fork();
 686   if ($childpid == 0)
 687   {
 688     $SIG{'INT'} = 'DEFAULT';
 689     $SIG{'QUIT'} = 'DEFAULT';
 690     $SIG{'TERM'} = 'DEFAULT';
 691
 692     foreach (values (%reader))
 693     {
 694       close($_);
 695     }
 696     fcntl ("writer", F_SETFL, 0) or croak ($!); # no close-on-exec
 697     open(STDOUT,">&writer");
 698     open(STDERR,">&writer");
 699
 700     undef $dbh;
 701     undef $sth;
 702
 703     delete $ENV{"GNUPGHOME"};
 704     $ENV{"TASK_UUID"} = $Jobstep->{'arvados_task'}->{'uuid'};
 705     $ENV{"TASK_QSEQUENCE"} = $id;
 706     $ENV{"TASK_SEQUENCE"} = $level;
 707     $ENV{"JOB_SCRIPT"} = $Job->{script};
 708     while (my ($param, $value) = each %{$Job->{script_parameters}}) {
 709       $param =~ tr/a-z/A-Z/;
 710       $ENV{"JOB_PARAMETER_$param"} = $value;
 711     }
 712     $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
 713     $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
 714     $ENV{"TASK_WORK"} = $ENV{"CRUNCH_TMP"}."/task/$childslotname";
 715     $ENV{"HOME"} = $ENV{"TASK_WORK"};
 716     $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}.".keep";
 717     $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 718     $ENV{"CRUNCH_NODE_SLOTS"} = $slot[$childslot]->{node}->{ncpus};
 719     $ENV{"PATH"} = $ENV{"CRUNCH_INSTALL"} . "/bin:" . $ENV{"PATH"};
 720
 721     $ENV{"GZIP"} = "-n";
 722
 723     my $max_node_concurrent_tasks = $ENV{CRUNCH_NODE_SLOTS};
 724     if ($tasks_this_level < $max_node_concurrent_tasks) {
 725       $max_node_concurrent_tasks = $tasks_this_level;
 726     }
 727
 728     my @srunargs = (
 729       "srun",
 730       "--nodelist=".$childnode->{name},
 731       qw(-n1 -c1 -N1 -D), $ENV{'TMPDIR'},
 732       "--job-name=$job_id.$id.$$",
 733         );
 734     my $command =
 735         "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
 736         ."mkdir -p $ENV{CRUNCH_TMP} $ENV{JOB_WORK} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
 737         ."&& cd $ENV{CRUNCH_TMP} "
 738         # These environment variables get used explicitly later in
 739         # $command.  No tool is expected to read these values directly.
 740         .q{&& MEM=$(awk '($1 == "MemTotal:"){print $2}' </proc/meminfo) }
 741         .q{&& SWAP=$(awk '($1 == "SwapTotal:"){print $2}' </proc/meminfo) }
 742         ."&& MEMLIMIT=\$(( (\$MEM * 95) / ($max_node_concurrent_tasks * 100) )) "
 743         ."&& let SWAPLIMIT=\$MEMLIMIT+\$SWAP ";
 744     $command .= "&& exec arv-mount --by-id --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
 745     if ($docker_hash)
 746     {
 747       my $cidfile = "$ENV{CRUNCH_TMP}/$Jobstep->{arvados_task}->{uuid}-$Jobstep->{failures}.cid";
 748       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$cidfile -poll=10000 ";
 749       $command .= "$docker_bin run --rm=true --attach=stdout --attach=stderr --attach=stdin -i --user=crunch --cidfile=$cidfile --sig-proxy ";
 750       # We only set memory limits if Docker lets us limit both memory and swap.
 751       # Memory limits alone have been supported longer, but subprocesses tend
 752       # to get SIGKILL if they exceed that without any swap limit set.
 753       # See #5642 for additional background.
 754       if ($docker_limitmem) {
 755         $command .= "--memory=\${MEMLIMIT}k --memory-swap=\${SWAPLIMIT}k ";
 756       }
 757
 758       # Dynamically configure the container to use the host system as its
 759       # DNS server.  Get the host's global addresses from the ip command,
 760       # and turn them into docker --dns options using gawk.
 761       $command .=
 762           q{$(ip -o address show scope global |
 763               gawk 'match($4, /^([0-9\.:]+)\//, x){print "--dns", x[1]}') };
 764
 765       # The source tree and $destdir directory (which we have
 766       # installed on the worker host) are available in the container,
 767       # under the same path.
 768       $command .= "--volume=\Q$ENV{CRUNCH_SRC}:$ENV{CRUNCH_SRC}:ro\E ";
 769       $command .= "--volume=\Q$ENV{CRUNCH_INSTALL}:$ENV{CRUNCH_INSTALL}:ro\E ";
 770
 771       # Currently, we make arv-mount's mount point appear at /keep
 772       # inside the container (instead of using the same path as the
 773       # host like we do with CRUNCH_SRC and CRUNCH_INSTALL). However,
 774       # crunch scripts and utilities must not rely on this. They must
 775       # use $TASK_KEEPMOUNT.
 776       $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E ";
 777       $ENV{TASK_KEEPMOUNT} = "/keep";
 778
 779       # TASK_WORK is almost exactly like a docker data volume: it
 780       # starts out empty, is writable, and persists until no
 781       # containers use it any more. We don't use --volumes-from to
 782       # share it with other containers: it is only accessible to this
 783       # task, and it goes away when this task stops.
 784       #
 785       # However, a docker data volume is writable only by root unless
 786       # the mount point already happens to exist in the container with
 787       # different permissions. Therefore, we [1] assume /tmp already
 788       # exists in the image and is writable by the crunch user; [2]
 789       # avoid putting TASK_WORK inside CRUNCH_TMP (which won't be
 790       # writable if they are created by docker while setting up the
 791       # other --volumes); and [3] create $TASK_WORK inside the
 792       # container using $build_script.
 793       $command .= "--volume=/tmp ";
 794       $ENV{"TASK_WORK"} = "/tmp/crunch-job-task-work/$childslotname";
 795       $ENV{"HOME"} = $ENV{"TASK_WORK"};
 796       $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 797
 798       # TODO: Share a single JOB_WORK volume across all task
 799       # containers on a given worker node, and delete it when the job
 800       # ends (and, in case that doesn't work, when the next job
 801       # starts).
 802       #
 803       # For now, use the same approach as TASK_WORK above.
 804       $ENV{"JOB_WORK"} = "/tmp/crunch-job-work";
 805
 806       while (my ($env_key, $env_val) = each %ENV)
 807       {
 808         if ($env_key =~ /^(ARVADOS|CRUNCH|JOB|TASK)_/) {
 809           $command .= "--env=\Q$env_key=$env_val\E ";
 810         }
 811       }
 812       $command .= "--env=\QHOME=$ENV{HOME}\E ";
 813       $command .= "\Q$docker_hash\E ";
 814       $command .= "stdbuf --output=0 --error=0 ";
 815       $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 816     } else {
 817       # Non-docker run
 818       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 ";
 819       $command .= "stdbuf --output=0 --error=0 ";
 820       $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 821     }
 822
 823     my @execargs = ('bash', '-c', $command);
 824     srun (\@srunargs, \@execargs, undef, $build_script);
 825     # exec() failed, we assume nothing happened.
 826     die "srun() failed on build script\n";
 827   }
 828   close("writer");
 829   if (!defined $childpid)
 830   {
 831     close $reader{$id};
 832     delete $reader{$id};
 833     next;
 834   }
 835   shift @freeslot;
 836   $proc{$childpid} = { jobstep => $id,
 837                        time => time,
 838                        slot => $childslot,
 839                        jobstepname => "$job_id.$id.$childpid",
 840                      };
 841   croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
 842   $slot[$childslot]->{pid} = $childpid;
 843
 844   Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
 845   Log ($id, "child $childpid started on $childslotname");
 846   $Jobstep->{starttime} = time;
 847   $Jobstep->{node} = $childnode->{name};
 848   $Jobstep->{slotindex} = $childslot;
 849   delete $Jobstep->{stderr};
 850   delete $Jobstep->{finishtime};
 851
 852   $Jobstep->{'arvados_task'}->{started_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{starttime});
 853   $Jobstep->{'arvados_task'}->save;
 854
 855   splice @jobstep_todo, $todo_ptr, 1;
 856   --$todo_ptr;
 857
 858   $progress_is_dirty = 1;
 859
 860   while (!@freeslot
 861          ||
 862          (@slot > @freeslot && $todo_ptr+1 > $#jobstep_todo))
 863   {
 864     last THISROUND if $main::please_freeze || defined($main::success);
 865     if ($main::please_info)
 866     {
 867       $main::please_info = 0;
 868       freeze();
 869       create_output_collection();
 870       save_meta(1);
 871       update_progress_stats();
 872     }
 873     my $gotsome
 874         = readfrompipes ()
 875         + reapchildren ();
 876     if (!$gotsome)
 877     {
 878       check_refresh_wanted();
 879       check_squeue();
 880       update_progress_stats();
 881       select (undef, undef, undef, 0.1);
 882     }
 883     elsif (time - $progress_stats_updated >= 30)
 884     {
 885       update_progress_stats();
 886     }
 887     if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
 888         ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
 889     {
 890       my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
 891           .($thisround_failed+$thisround_succeeded)
 892           .") -- giving up on this round";
 893       Log (undef, $message);
 894       last THISROUND;
 895     }
 896
 897     # move slots from freeslot to holdslot (or back to freeslot) if necessary
 898     for (my $i=$#freeslot; $i>=0; $i--) {
 899       if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
 900         push @holdslot, (splice @freeslot, $i, 1);
 901       }
 902     }
 903     for (my $i=$#holdslot; $i>=0; $i--) {
 904       if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
 905         push @freeslot, (splice @holdslot, $i, 1);
 906       }
 907     }
 908
 909     # give up if no nodes are succeeding
 910     if (!grep { $_->{node}->{losing_streak} == 0 &&
 911                     $_->{node}->{hold_count} < 4 } @slot) {
 912       my $message = "Every node has failed -- giving up on this round";
 913       Log (undef, $message);
 914       last THISROUND;
 915     }
 916   }
 917 }
 918
 919
 920 push @freeslot, splice @holdslot;
 921 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
 922
 923
 924 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
 925 while (%proc)
 926 {
 927   if ($main::please_continue) {
 928     $main::please_continue = 0;
 929     goto THISROUND;
 930   }
 931   $main::please_info = 0, freeze(), create_output_collection(), save_meta(1) if $main::please_info;
 932   readfrompipes ();
 933   if (!reapchildren())
 934   {
 935     check_refresh_wanted();
 936     check_squeue();
 937     update_progress_stats();
 938     select (undef, undef, undef, 0.1);
 939     killem (keys %proc) if $main::please_freeze;
 940   }
 941 }
 942
 943 update_progress_stats();
 944 freeze_if_want_freeze();
 945
 946
 947 if (!defined $main::success)
 948 {
 949   if (@jobstep_todo &&
 950       $thisround_succeeded == 0 &&
 951       ($thisround_failed == 0 || $thisround_failed > 4))
 952   {
 953     my $message = "stop because $thisround_failed tasks failed and none succeeded";
 954     Log (undef, $message);
 955     $main::success = 0;
 956   }
 957   if (!@jobstep_todo)
 958   {
 959     $main::success = 1;
 960   }
 961 }
 962
 963 goto ONELEVEL if !defined $main::success;
 964
 965
 966 release_allocation();
 967 freeze();
 968 my $collated_output = &create_output_collection();
 969
 970 if (!$collated_output) {
 971   Log (undef, "Failed to write output collection");
 972 }
 973 else {
 974   Log(undef, "job output $collated_output");
 975   $Job->update_attributes('output' => $collated_output);
 976 }
 977
 978 Log (undef, "finish");
 979
 980 save_meta();
 981
 982 my $final_state;
 983 if ($collated_output && $main::success) {
 984   $final_state = 'Complete';
 985 } else {
 986   $final_state = 'Failed';
 987 }
 988 $Job->update_attributes('state' => $final_state);
 989
 990 exit (($final_state eq 'Complete') ? 0 : 1);
 991
 992
 993
 994 sub update_progress_stats
 995 {
 996   $progress_stats_updated = time;
 997   return if !$progress_is_dirty;
 998   my ($todo, $done, $running) = (scalar @jobstep_todo,
 999                                  scalar @jobstep_done,
1000                                  scalar @slot - scalar @freeslot - scalar @holdslot);
1001   $Job->{'tasks_summary'} ||= {};
1002   $Job->{'tasks_summary'}->{'todo'} = $todo;
1003   $Job->{'tasks_summary'}->{'done'} = $done;
1004   $Job->{'tasks_summary'}->{'running'} = $running;
1005   $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
1006   Log (undef, "status: $done done, $running running, $todo todo");
1007   $progress_is_dirty = 0;
1008 }
1009
1010
1011
1012 sub reapchildren
1013 {
1014   my $pid = waitpid (-1, WNOHANG);
1015   return 0 if $pid <= 0;
1016
1017   my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
1018                   . "."
1019                   . $slot[$proc{$pid}->{slot}]->{cpu});
1020   my $jobstepid = $proc{$pid}->{jobstep};
1021   my $elapsed = time - $proc{$pid}->{time};
1022   my $Jobstep = $jobstep[$jobstepid];
1023
1024   my $childstatus = $?;
1025   my $exitvalue = $childstatus >> 8;
1026   my $exitinfo = "exit ".exit_status_s($childstatus);
1027   $Jobstep->{'arvados_task'}->reload;
1028   my $task_success = $Jobstep->{'arvados_task'}->{success};
1029
1030   Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
1031
1032   if (!defined $task_success) {
1033     # task did not indicate one way or the other --> fail
1034     $Jobstep->{'arvados_task'}->{success} = 0;
1035     $Jobstep->{'arvados_task'}->save;
1036     $task_success = 0;
1037   }
1038
1039   if (!$task_success)
1040   {
1041     my $temporary_fail;
1042     $temporary_fail ||= $Jobstep->{node_fail};
1043     $temporary_fail ||= ($exitvalue == TASK_TEMPFAIL);
1044
1045     ++$thisround_failed;
1046     ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
1047
1048     # Check for signs of a failed or misconfigured node
1049     if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
1050         2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
1051       # Don't count this against jobstep failure thresholds if this
1052       # node is already suspected faulty and srun exited quickly
1053       if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
1054           $elapsed < 5) {
1055         Log ($jobstepid, "blaming failure on suspect node " .
1056              $slot[$proc{$pid}->{slot}]->{node}->{name});
1057         $temporary_fail ||= 1;
1058       }
1059       ban_node_by_slot($proc{$pid}->{slot});
1060     }
1061
1062     Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
1063                              ++$Jobstep->{'failures'},
1064                              $temporary_fail ? 'temporary ' : 'permanent',
1065                              $elapsed));
1066
1067     if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
1068       # Give up on this task, and the whole job
1069       $main::success = 0;
1070     }
1071     # Put this task back on the todo queue
1072     push @jobstep_todo, $jobstepid;
1073     $Job->{'tasks_summary'}->{'failed'}++;
1074   }
1075   else
1076   {
1077     ++$thisround_succeeded;
1078     $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
1079     $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
1080     push @jobstep_done, $jobstepid;
1081     Log ($jobstepid, "success in $elapsed seconds");
1082   }
1083   $Jobstep->{exitcode} = $childstatus;
1084   $Jobstep->{finishtime} = time;
1085   $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
1086   $Jobstep->{'arvados_task'}->save;
1087   process_stderr ($jobstepid, $task_success);
1088   Log ($jobstepid, sprintf("task output (%d bytes): %s",
1089                            length($Jobstep->{'arvados_task'}->{output}),
1090                            $Jobstep->{'arvados_task'}->{output}));
1091
1092   close $reader{$jobstepid};
1093   delete $reader{$jobstepid};
1094   delete $slot[$proc{$pid}->{slot}]->{pid};
1095   push @freeslot, $proc{$pid}->{slot};
1096   delete $proc{$pid};
1097
1098   if ($task_success) {
1099     # Load new tasks
1100     my $newtask_list = [];
1101     my $newtask_results;
1102     do {
1103       $newtask_results = api_call(
1104         "job_tasks/list",
1105         'where' => {
1106           'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
1107         },
1108         'order' => 'qsequence',
1109         'offset' => scalar(@$newtask_list),
1110       );
1111       push(@$newtask_list, @{$newtask_results->{items}});
1112     } while (@{$newtask_results->{items}});
1113     foreach my $arvados_task (@$newtask_list) {
1114       my $jobstep = {
1115         'level' => $arvados_task->{'sequence'},
1116         'failures' => 0,
1117         'arvados_task' => $arvados_task
1118       };
1119       push @jobstep, $jobstep;
1120       push @jobstep_todo, $#jobstep;
1121     }
1122   }
1123
1124   $progress_is_dirty = 1;
1125   1;
1126 }
1127
1128 sub check_refresh_wanted
1129 {
1130   my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1131   if (@stat && $stat[9] > $latest_refresh) {
1132     $latest_refresh = scalar time;
1133     my $Job2 = api_call("jobs/get", uuid => $jobspec);
1134     for my $attr ('cancelled_at',
1135                   'cancelled_by_user_uuid',
1136                   'cancelled_by_client_uuid',
1137                   'state') {
1138       $Job->{$attr} = $Job2->{$attr};
1139     }
1140     if ($Job->{'state'} ne "Running") {
1141       if ($Job->{'state'} eq "Cancelled") {
1142         Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
1143       } else {
1144         Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
1145       }
1146       $main::success = 0;
1147       $main::please_freeze = 1;
1148     }
1149   }
1150 }
1151
1152 sub check_squeue
1153 {
1154   # return if the kill list was checked <4 seconds ago
1155   if (defined $squeue_kill_checked && $squeue_kill_checked > time - 4)
1156   {
1157     return;
1158   }
1159   $squeue_kill_checked = time;
1160
1161   # use killem() on procs whose killtime is reached
1162   for (keys %proc)
1163   {
1164     if (exists $proc{$_}->{killtime}
1165         && $proc{$_}->{killtime} <= time)
1166     {
1167       killem ($_);
1168     }
1169   }
1170
1171   # return if the squeue was checked <60 seconds ago
1172   if (defined $squeue_checked && $squeue_checked > time - 60)
1173   {
1174     return;
1175   }
1176   $squeue_checked = time;
1177
1178   if (!$have_slurm)
1179   {
1180     # here is an opportunity to check for mysterious problems with local procs
1181     return;
1182   }
1183
1184   # get a list of steps still running
1185   my @squeue = `squeue -s -h -o '%i %j' && echo ok`;
1186   chop @squeue;
1187   if ($squeue[-1] ne "ok")
1188   {
1189     return;
1190   }
1191   pop @squeue;
1192
1193   # which of my jobsteps are running, according to squeue?
1194   my %ok;
1195   foreach (@squeue)
1196   {
1197     if (/^(\d+)\.(\d+) (\S+)/)
1198     {
1199       if ($1 eq $ENV{SLURM_JOBID})
1200       {
1201         $ok{$3} = 1;
1202       }
1203     }
1204   }
1205
1206   # which of my active child procs (>60s old) were not mentioned by squeue?
1207   foreach (keys %proc)
1208   {
1209     if ($proc{$_}->{time} < time - 60
1210         && !exists $ok{$proc{$_}->{jobstepname}}
1211         && !exists $proc{$_}->{killtime})
1212     {
1213       # kill this proc if it hasn't exited in 30 seconds
1214       $proc{$_}->{killtime} = time + 30;
1215     }
1216   }
1217 }
1218
1219
1220 sub release_allocation
1221 {
1222   if ($have_slurm)
1223   {
1224     Log (undef, "release job allocation");
1225     system "scancel $ENV{SLURM_JOBID}";
1226   }
1227 }
1228
1229
1230 sub readfrompipes
1231 {
1232   my $gotsome = 0;
1233   foreach my $job (keys %reader)
1234   {
1235     my $buf;
1236     while (0 < sysread ($reader{$job}, $buf, 8192))
1237     {
1238       print STDERR $buf if $ENV{CRUNCH_DEBUG};
1239       $jobstep[$job]->{stderr} .= $buf;
1240       preprocess_stderr ($job);
1241       if (length ($jobstep[$job]->{stderr}) > 16384)
1242       {
1243         substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1244       }
1245       $gotsome = 1;
1246     }
1247   }
1248   return $gotsome;
1249 }
1250
1251
1252 sub preprocess_stderr
1253 {
1254   my $job = shift;
1255
1256   while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1257     my $line = $1;
1258     substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1259     Log ($job, "stderr $line");
1260     if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1261       # whoa.
1262       $main::please_freeze = 1;
1263     }
1264     elsif ($line =~ /(srun: error: (Node failure on|Unable to create job step|.*: Communication connection failure))|arvados.errors.Keep/) {
1265       $jobstep[$job]->{node_fail} = 1;
1266       ban_node_by_slot($jobstep[$job]->{slotindex});
1267     }
1268   }
1269 }
1270
1271
1272 sub process_stderr
1273 {
1274   my $job = shift;
1275   my $task_success = shift;
1276   preprocess_stderr ($job);
1277
1278   map {
1279     Log ($job, "stderr $_");
1280   } split ("\n", $jobstep[$job]->{stderr});
1281 }
1282
1283 sub fetch_block
1284 {
1285   my $hash = shift;
1286   my $keep;
1287   if (!open($keep, "-|", "arv-get", "--retries", retry_count(), $hash)) {
1288     Log(undef, "fetch_block run error from arv-get $hash: $!");
1289     return undef;
1290   }
1291   my $output_block = "";
1292   while (1) {
1293     my $buf;
1294     my $bytes = sysread($keep, $buf, 1024 * 1024);
1295     if (!defined $bytes) {
1296       Log(undef, "fetch_block read error from arv-get: $!");
1297       $output_block = undef;
1298       last;
1299     } elsif ($bytes == 0) {
1300       # sysread returns 0 at the end of the pipe.
1301       last;
1302     } else {
1303       # some bytes were read into buf.
1304       $output_block .= $buf;
1305     }
1306   }
1307   close $keep;
1308   if ($?) {
1309     Log(undef, "fetch_block arv-get exited " . exit_status_s($?));
1310     $output_block = undef;
1311   }
1312   return $output_block;
1313 }
1314
1315 # Create a collection by concatenating the output of all tasks (each
1316 # task's output is either a manifest fragment, a locator for a
1317 # manifest fragment stored in Keep, or nothing at all). Return the
1318 # portable_data_hash of the new collection.
1319 sub create_output_collection
1320 {
1321   Log (undef, "collate");
1322
1323   my ($child_out, $child_in);
1324   my $pid = open2($child_out, $child_in, 'python', '-c', q{
1325 import arvados
1326 import sys
1327 print (arvados.api("v1").collections().
1328        create(body={"manifest_text": sys.stdin.read()}).
1329        execute(num_retries=int(sys.argv[1]))["portable_data_hash"])
1330 }, retry_count());
1331
1332   my $task_idx = -1;
1333   my $manifest_size = 0;
1334   for (@jobstep)
1335   {
1336     ++$task_idx;
1337     my $output = $_->{'arvados_task'}->{output};
1338     next if (!defined($output));
1339     my $next_write;
1340     if ($output =~ /^[0-9a-f]{32}(\+\S+)*$/) {
1341       $next_write = fetch_block($output);
1342     } else {
1343       $next_write = $output;
1344     }
1345     if (defined($next_write)) {
1346       if (!defined(syswrite($child_in, $next_write))) {
1347         # There's been an error writing.  Stop the loop.
1348         # We'll log details about the exit code later.
1349         last;
1350       } else {
1351         $manifest_size += length($next_write);
1352       }
1353     } else {
1354       my $uuid = $_->{'arvados_task'}->{'uuid'};
1355       Log (undef, "Error retrieving '$output' output by task $task_idx ($uuid)");
1356       $main::success = 0;
1357     }
1358   }
1359   close($child_in);
1360   Log(undef, "collated output manifest text to send to API server is $manifest_size bytes with access tokens");
1361
1362   my $joboutput;
1363   my $s = IO::Select->new($child_out);
1364   if ($s->can_read(120)) {
1365     sysread($child_out, $joboutput, 1024 * 1024);
1366     waitpid($pid, 0);
1367     if ($?) {
1368       Log(undef, "output collection creation exited " . exit_status_s($?));
1369       $joboutput = undef;
1370     } else {
1371       chomp($joboutput);
1372     }
1373   } else {
1374     Log (undef, "timed out while creating output collection");
1375     foreach my $signal (2, 2, 2, 15, 15, 9) {
1376       kill($signal, $pid);
1377       last if waitpid($pid, WNOHANG) == -1;
1378       sleep(1);
1379     }
1380   }
1381   close($child_out);
1382
1383   return $joboutput;
1384 }
1385
1386
1387 sub killem
1388 {
1389   foreach (@_)
1390   {
1391     my $sig = 2;                # SIGINT first
1392     if (exists $proc{$_}->{"sent_$sig"} &&
1393         time - $proc{$_}->{"sent_$sig"} > 4)
1394     {
1395       $sig = 15;                # SIGTERM if SIGINT doesn't work
1396     }
1397     if (exists $proc{$_}->{"sent_$sig"} &&
1398         time - $proc{$_}->{"sent_$sig"} > 4)
1399     {
1400       $sig = 9;                 # SIGKILL if SIGTERM doesn't work
1401     }
1402     if (!exists $proc{$_}->{"sent_$sig"})
1403     {
1404       Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1405       kill $sig, $_;
1406       select (undef, undef, undef, 0.1);
1407       if ($sig == 2)
1408       {
1409         kill $sig, $_;     # srun wants two SIGINT to really interrupt
1410       }
1411       $proc{$_}->{"sent_$sig"} = time;
1412       $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1413     }
1414   }
1415 }
1416
1417
1418 sub fhbits
1419 {
1420   my($bits);
1421   for (@_) {
1422     vec($bits,fileno($_),1) = 1;
1423   }
1424   $bits;
1425 }
1426
1427
1428 # Send log output to Keep via arv-put.
1429 #
1430 # $log_pipe_in and $log_pipe_out are the input and output filehandles to the arv-put pipe.
1431 # $log_pipe_pid is the pid of the arv-put subprocess.
1432 #
1433 # The only functions that should access these variables directly are:
1434 #
1435 # log_writer_start($logfilename)
1436 #     Starts an arv-put pipe, reading data on stdin and writing it to
1437 #     a $logfilename file in an output collection.
1438 #
1439 # log_writer_send($txt)
1440 #     Writes $txt to the output log collection.
1441 #
1442 # log_writer_finish()
1443 #     Closes the arv-put pipe and returns the output that it produces.
1444 #
1445 # log_writer_is_active()
1446 #     Returns a true value if there is currently a live arv-put
1447 #     process, false otherwise.
1448 #
1449 my ($log_pipe_in, $log_pipe_out, $log_pipe_pid);
1450
1451 sub log_writer_start($)
1452 {
1453   my $logfilename = shift;
1454   $log_pipe_pid = open2($log_pipe_out, $log_pipe_in,
1455                         'arv-put',
1456                         '--portable-data-hash',
1457                         '--project-uuid', $Job->{owner_uuid},
1458                         '--retries', '3',
1459                         '--name', $logfilename,
1460                         '--filename', $logfilename,
1461                         '-');
1462 }
1463
1464 sub log_writer_send($)
1465 {
1466   my $txt = shift;
1467   print $log_pipe_in $txt;
1468 }
1469
1470 sub log_writer_finish()
1471 {
1472   return unless $log_pipe_pid;
1473
1474   close($log_pipe_in);
1475   my $arv_put_output;
1476
1477   my $s = IO::Select->new($log_pipe_out);
1478   if ($s->can_read(120)) {
1479     sysread($log_pipe_out, $arv_put_output, 1024);
1480     chomp($arv_put_output);
1481   } else {
1482     Log (undef, "timed out reading from 'arv-put'");
1483   }
1484
1485   waitpid($log_pipe_pid, 0);
1486   $log_pipe_pid = $log_pipe_in = $log_pipe_out = undef;
1487   if ($?) {
1488     Log("log_writer_finish: arv-put exited ".exit_status_s($?))
1489   }
1490
1491   return $arv_put_output;
1492 }
1493
1494 sub log_writer_is_active() {
1495   return $log_pipe_pid;
1496 }
1497
1498 sub Log                         # ($jobstep_id, $logmessage)
1499 {
1500   if ($_[1] =~ /\n/) {
1501     for my $line (split (/\n/, $_[1])) {
1502       Log ($_[0], $line);
1503     }
1504     return;
1505   }
1506   my $fh = select STDERR; $|=1; select $fh;
1507   my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1508   $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1509   $message .= "\n";
1510   my $datetime;
1511   if (log_writer_is_active() || -t STDERR) {
1512     my @gmtime = gmtime;
1513     $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1514                          $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1515   }
1516   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1517
1518   if (log_writer_is_active()) {
1519     log_writer_send($datetime . " " . $message);
1520   }
1521 }
1522
1523
1524 sub croak
1525 {
1526   my ($package, $file, $line) = caller;
1527   my $message = "@_ at $file line $line\n";
1528   Log (undef, $message);
1529   freeze() if @jobstep_todo;
1530   create_output_collection() if @jobstep_todo;
1531   cleanup();
1532   save_meta();
1533   die;
1534 }
1535
1536
1537 sub cleanup
1538 {
1539   return unless $Job;
1540   if ($Job->{'state'} eq 'Cancelled') {
1541     $Job->update_attributes('finished_at' => scalar gmtime);
1542   } else {
1543     $Job->update_attributes('state' => 'Failed');
1544   }
1545 }
1546
1547
1548 sub save_meta
1549 {
1550   my $justcheckpoint = shift; # false if this will be the last meta saved
1551   return if $justcheckpoint;  # checkpointing is not relevant post-Warehouse.pm
1552   return unless log_writer_is_active();
1553
1554   my $loglocator = log_writer_finish();
1555   Log (undef, "log manifest is $loglocator");
1556   $Job->{'log'} = $loglocator;
1557   $Job->update_attributes('log', $loglocator);
1558 }
1559
1560
1561 sub freeze_if_want_freeze
1562 {
1563   if ($main::please_freeze)
1564   {
1565     release_allocation();
1566     if (@_)
1567     {
1568       # kill some srun procs before freeze+stop
1569       map { $proc{$_} = {} } @_;
1570       while (%proc)
1571       {
1572         killem (keys %proc);
1573         select (undef, undef, undef, 0.1);
1574         my $died;
1575         while (($died = waitpid (-1, WNOHANG)) > 0)
1576         {
1577           delete $proc{$died};
1578         }
1579       }
1580     }
1581     freeze();
1582     create_output_collection();
1583     cleanup();
1584     save_meta();
1585     exit 1;
1586   }
1587 }
1588
1589
1590 sub freeze
1591 {
1592   Log (undef, "Freeze not implemented");
1593   return;
1594 }
1595
1596
1597 sub thaw
1598 {
1599   croak ("Thaw not implemented");
1600 }
1601
1602
1603 sub freezequote
1604 {
1605   my $s = shift;
1606   $s =~ s/\\/\\\\/g;
1607   $s =~ s/\n/\\n/g;
1608   return $s;
1609 }
1610
1611
1612 sub freezeunquote
1613 {
1614   my $s = shift;
1615   $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1616   return $s;
1617 }
1618
1619
1620 sub srun
1621 {
1622   my $srunargs = shift;
1623   my $execargs = shift;
1624   my $opts = shift || {};
1625   my $stdin = shift;
1626   my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1627
1628   $Data::Dumper::Terse = 1;
1629   $Data::Dumper::Indent = 0;
1630   my $show_cmd = Dumper($args);
1631   $show_cmd =~ s/(TOKEN\\*=)[^\s\']+/${1}[...]/g;
1632   $show_cmd =~ s/\n/ /g;
1633   warn "starting: $show_cmd\n";
1634
1635   if (defined $stdin) {
1636     my $child = open STDIN, "-|";
1637     defined $child or die "no fork: $!";
1638     if ($child == 0) {
1639       print $stdin or die $!;
1640       close STDOUT or die $!;
1641       exit 0;
1642     }
1643   }
1644
1645   return system (@$args) if $opts->{fork};
1646
1647   exec @$args;
1648   warn "ENV size is ".length(join(" ",%ENV));
1649   die "exec failed: $!: @$args";
1650 }
1651
1652
1653 sub ban_node_by_slot {
1654   # Don't start any new jobsteps on this node for 60 seconds
1655   my $slotid = shift;
1656   $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1657   $slot[$slotid]->{node}->{hold_count}++;
1658   Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1659 }
1660
1661 sub must_lock_now
1662 {
1663   my ($lockfile, $error_message) = @_;
1664   open L, ">", $lockfile or croak("$lockfile: $!");
1665   if (!flock L, LOCK_EX|LOCK_NB) {
1666     croak("Can't lock $lockfile: $error_message\n");
1667   }
1668 }
1669
1670 sub find_docker_image {
1671   # Given a Keep locator, check to see if it contains a Docker image.
1672   # If so, return its stream name and Docker hash.
1673   # If not, return undef for both values.
1674   my $locator = shift;
1675   my ($streamname, $filename);
1676   my $image = api_call("collections/get", uuid => $locator);
1677   if ($image) {
1678     foreach my $line (split(/\n/, $image->{manifest_text})) {
1679       my @tokens = split(/\s+/, $line);
1680       next if (!@tokens);
1681       $streamname = shift(@tokens);
1682       foreach my $filedata (grep(/^\d+:\d+:/, @tokens)) {
1683         if (defined($filename)) {
1684           return (undef, undef);  # More than one file in the Collection.
1685         } else {
1686           $filename = (split(/:/, $filedata, 3))[2];
1687         }
1688       }
1689     }
1690   }
1691   if (defined($filename) and ($filename =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
1692     return ($streamname, $1);
1693   } else {
1694     return (undef, undef);
1695   }
1696 }
1697
1698 sub retry_count {
1699   # Calculate the number of times an operation should be retried,
1700   # assuming exponential backoff, and that we're willing to retry as
1701   # long as tasks have been running.  Enforce a minimum of 3 retries.
1702   my ($starttime, $endtime, $timediff, $retries);
1703   if (@jobstep) {
1704     $starttime = $jobstep[0]->{starttime};
1705     $endtime = $jobstep[-1]->{finishtime};
1706   }
1707   if (!defined($starttime)) {
1708     $timediff = 0;
1709   } elsif (!defined($endtime)) {
1710     $timediff = time - $starttime;
1711   } else {
1712     $timediff = ($endtime - $starttime) - (time - $endtime);
1713   }
1714   if ($timediff > 0) {
1715     $retries = int(log($timediff) / log(2));
1716   } else {
1717     $retries = 1;  # Use the minimum.
1718   }
1719   return ($retries > 3) ? $retries : 3;
1720 }
1721
1722 sub retry_op {
1723   # Pass in two function references.
1724   # This method will be called with the remaining arguments.
1725   # If it dies, retry it with exponential backoff until it succeeds,
1726   # or until the current retry_count is exhausted.  After each failure
1727   # that can be retried, the second function will be called with
1728   # the current try count (0-based), next try time, and error message.
1729   my $operation = shift;
1730   my $retry_callback = shift;
1731   my $retries = retry_count();
1732   foreach my $try_count (0..$retries) {
1733     my $next_try = time + (2 ** $try_count);
1734     my $result = eval { $operation->(@_); };
1735     if (!$@) {
1736       return $result;
1737     } elsif ($try_count < $retries) {
1738       $retry_callback->($try_count, $next_try, $@);
1739       my $sleep_time = $next_try - time;
1740       sleep($sleep_time) if ($sleep_time > 0);
1741     }
1742   }
1743   # Ensure the error message ends in a newline, so Perl doesn't add
1744   # retry_op's line number to it.
1745   chomp($@);
1746   die($@ . "\n");
1747 }
1748
1749 sub api_call {
1750   # Pass in a /-separated API method name, and arguments for it.
1751   # This function will call that method, retrying as needed until
1752   # the current retry_count is exhausted, with a log on the first failure.
1753   my $method_name = shift;
1754   my $log_api_retry = sub {
1755     my ($try_count, $next_try_at, $errmsg) = @_;
1756     $errmsg =~ s/\s*\bat \Q$0\E line \d+\.?\s*//;
1757     $errmsg =~ s/\s/ /g;
1758     $errmsg =~ s/\s+$//;
1759     my $retry_msg;
1760     if ($next_try_at < time) {
1761       $retry_msg = "Retrying.";
1762     } else {
1763       my $next_try_fmt = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($next_try_at);
1764       $retry_msg = "Retrying at $next_try_fmt.";
1765     }
1766     Log(undef, "API method $method_name failed: $errmsg. $retry_msg");
1767   };
1768   my $method = $arv;
1769   foreach my $key (split(/\//, $method_name)) {
1770     $method = $method->{$key};
1771   }
1772   return retry_op(sub { $method->execute(@_); }, $log_api_retry, @_);
1773 }
1774
1775 sub exit_status_s {
1776   # Given a $?, return a human-readable exit code string like "0" or
1777   # "1" or "0 with signal 1" or "1 with signal 11".
1778   my $exitcode = shift;
1779   my $s = $exitcode >> 8;
1780   if ($exitcode & 0x7f) {
1781     $s .= " with signal " . ($exitcode & 0x7f);
1782   }
1783   if ($exitcode & 0x80) {
1784     $s .= " with core dump";
1785   }
1786   return $s;
1787 }
1788
1789 sub handle_readall {
1790   # Pass in a glob reference to a file handle.
1791   # Read all its contents and return them as a string.
1792   my $fh_glob_ref = shift;
1793   local $/ = undef;
1794   return <$fh_glob_ref>;
1795 }
1796
1797 sub tar_filename_n {
1798   my $n = shift;
1799   return sprintf("%s/git.%s.%d.tar", $ENV{CRUNCH_TMP}, $job_id, $n);
1800 }
1801
1802 sub add_git_archive {
1803   # Pass in a git archive command as a string or list, a la system().
1804   # This method will save its output to be included in the archive sent to the
1805   # build script.
1806   my $git_input;
1807   $git_tar_count++;
1808   if (!open(GIT_ARCHIVE, ">", tar_filename_n($git_tar_count))) {
1809     croak("Failed to save git archive: $!");
1810   }
1811   my $git_pid = open2(">&GIT_ARCHIVE", $git_input, @_);
1812   close($git_input);
1813   waitpid($git_pid, 0);
1814   close(GIT_ARCHIVE);
1815   if ($?) {
1816     croak("Failed to save git archive: git exited " . exit_status_s($?));
1817   }
1818 }
1819
1820 sub combined_git_archive {
1821   # Combine all saved tar archives into a single archive, then return its
1822   # contents in a string.  Return undef if no archives have been saved.
1823   if ($git_tar_count < 1) {
1824     return undef;
1825   }
1826   my $base_tar_name = tar_filename_n(1);
1827   foreach my $tar_to_append (map { tar_filename_n($_); } (2..$git_tar_count)) {
1828     my $tar_exit = system("tar", "-Af", $base_tar_name, $tar_to_append);
1829     if ($tar_exit != 0) {
1830       croak("Error preparing build archive: tar -A exited " .
1831             exit_status_s($tar_exit));
1832     }
1833   }
1834   if (!open(GIT_TAR, "<", $base_tar_name)) {
1835     croak("Could not open build archive: $!");
1836   }
1837   my $tar_contents = handle_readall(\*GIT_TAR);
1838   close(GIT_TAR);
1839   return $tar_contents;
1840 }
1841
1842 __DATA__
1843 #!/usr/bin/perl
1844 #
1845 # This is crunch-job's internal dispatch script.  crunch-job running on the API
1846 # server invokes this script on individual compute nodes, or localhost if we're
1847 # running a job locally.  It gets called in two modes:
1848 #
1849 # * No arguments: Installation mode.  Read a tar archive from the DATA
1850 #   file handle; it includes the Crunch script's source code, and
1851 #   maybe SDKs as well.  Those should be installed in the proper
1852 #   locations.  This runs outside of any Docker container, so don't try to
1853 #   introspect Crunch's runtime environment.
1854 #
1855 # * With arguments: Crunch script run mode.  This script should set up the
1856 #   environment, then run the command specified in the arguments.  This runs
1857 #   inside any Docker container.
1858
1859 use Fcntl ':flock';
1860 use File::Path qw( make_path remove_tree );
1861 use POSIX qw(getcwd);
1862
1863 use constant TASK_TEMPFAIL => 111;
1864
1865 # Map SDK subdirectories to the path environments they belong to.
1866 my %SDK_ENVVARS = ("perl/lib" => "PERLLIB", "ruby/lib" => "RUBYLIB");
1867
1868 my $destdir = $ENV{"CRUNCH_SRC"};
1869 my $commit = $ENV{"CRUNCH_SRC_COMMIT"};
1870 my $repo = $ENV{"CRUNCH_SRC_URL"};
1871 my $install_dir = $ENV{"CRUNCH_INSTALL"} || (getcwd() . "/opt");
1872 my $job_work = $ENV{"JOB_WORK"};
1873 my $task_work = $ENV{"TASK_WORK"};
1874
1875 for my $dir ($destdir, $job_work, $task_work) {
1876   if ($dir) {
1877     make_path $dir;
1878     -e $dir or die "Failed to create temporary directory ($dir): $!";
1879   }
1880 }
1881
1882 if ($task_work) {
1883   remove_tree($task_work, {keep_root => 1});
1884 }
1885
1886 open(STDOUT_ORIG, ">&", STDOUT);
1887 open(STDERR_ORIG, ">&", STDERR);
1888 open(STDOUT, ">>", "$destdir.log");
1889 open(STDERR, ">&", STDOUT);
1890
1891 ### Crunch script run mode
1892 if (@ARGV) {
1893   # We want to do routine logging during task 0 only.  This gives the user
1894   # the information they need, but avoids repeating the information for every
1895   # task.
1896   my $Log;
1897   if ($ENV{TASK_SEQUENCE} eq "0") {
1898     $Log = sub {
1899       my $msg = shift;
1900       printf STDERR_ORIG "[Crunch] $msg\n", @_;
1901     };
1902   } else {
1903     $Log = sub { };
1904   }
1905
1906   my $python_src = "$install_dir/python";
1907   my $venv_dir = "$job_work/.arvados.venv";
1908   my $venv_built = -e "$venv_dir/bin/activate";
1909   if ((!$venv_built) and (-d $python_src) and can_run("virtualenv")) {
1910     shell_or_die(undef, "virtualenv", "--quiet", "--system-site-packages",
1911                  "--python=python2.7", $venv_dir);
1912     shell_or_die(TASK_TEMPFAIL, "$venv_dir/bin/pip", "--quiet", "install", "-I", $python_src);
1913     $venv_built = 1;
1914     $Log->("Built Python SDK virtualenv");
1915   }
1916
1917   my $pip_bin = "pip";
1918   if ($venv_built) {
1919     $Log->("Running in Python SDK virtualenv");
1920     $pip_bin = "$venv_dir/bin/pip";
1921     my $orig_argv = join(" ", map { quotemeta($_); } @ARGV);
1922     @ARGV = ("/bin/sh", "-ec",
1923              ". \Q$venv_dir/bin/activate\E; exec $orig_argv");
1924   } elsif (-d $python_src) {
1925     $Log->("Warning: virtualenv not found inside Docker container default " .
1926            "\$PATH. Can't install Python SDK.");
1927   }
1928
1929   my $pkgs = `(\Q$pip_bin\E freeze 2>/dev/null | grep arvados) || dpkg-query --show '*arvados*'`;
1930   if ($pkgs) {
1931     $Log->("Using Arvados SDK:");
1932     foreach my $line (split /\n/, $pkgs) {
1933       $Log->($line);
1934     }
1935   } else {
1936     $Log->("Arvados SDK packages not found");
1937   }
1938
1939   while (my ($sdk_dir, $sdk_envkey) = each(%SDK_ENVVARS)) {
1940     my $sdk_path = "$install_dir/$sdk_dir";
1941     if (-d $sdk_path) {
1942       if ($ENV{$sdk_envkey}) {
1943         $ENV{$sdk_envkey} = "$sdk_path:" . $ENV{$sdk_envkey};
1944       } else {
1945         $ENV{$sdk_envkey} = $sdk_path;
1946       }
1947       $Log->("Arvados SDK added to %s", $sdk_envkey);
1948     }
1949   }
1950
1951   close(STDOUT);
1952   close(STDERR);
1953   open(STDOUT, ">&", STDOUT_ORIG);
1954   open(STDERR, ">&", STDERR_ORIG);
1955   exec(@ARGV);
1956   die "Cannot exec `@ARGV`: $!";
1957 }
1958
1959 ### Installation mode
1960 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
1961 flock L, LOCK_EX;
1962 if (readlink ("$destdir.commit") eq $commit && -d $destdir) {
1963   # This version already installed -> nothing to do.
1964   exit(0);
1965 }
1966
1967 unlink "$destdir.commit";
1968 mkdir $destdir;
1969
1970 if (!open(TARX, "|-", "tar", "-xC", $destdir)) {
1971   die "Error launching 'tar -xC $destdir': $!";
1972 }
1973 # If we send too much data to tar in one write (> 4-5 MiB), it stops, and we
1974 # get SIGPIPE.  We must feed it data incrementally.
1975 my $tar_input;
1976 while (read(DATA, $tar_input, 65536)) {
1977   print TARX $tar_input;
1978 }
1979 if(!close(TARX)) {
1980   die "'tar -xC $destdir' exited $?: $!";
1981 }
1982
1983 mkdir $install_dir;
1984
1985 my $sdk_root = "$destdir/.arvados.sdk/sdk";
1986 if (-d $sdk_root) {
1987   foreach my $sdk_lang (("python",
1988                          map { (split /\//, $_, 2)[0]; } keys(%SDK_ENVVARS))) {
1989     if (-d "$sdk_root/$sdk_lang") {
1990       if (!rename("$sdk_root/$sdk_lang", "$install_dir/$sdk_lang")) {
1991         die "Failed to install $sdk_lang SDK: $!";
1992       }
1993     }
1994   }
1995 }
1996
1997 my $python_dir = "$install_dir/python";
1998 if ((-d $python_dir) and can_run("python2.7") and
1999     (system("python2.7", "$python_dir/setup.py", "--quiet", "egg_info") != 0)) {
2000   # egg_info failed, probably when it asked git for a build tag.
2001   # Specify no build tag.
2002   open(my $pysdk_cfg, ">>", "$python_dir/setup.cfg");
2003   print $pysdk_cfg "\n[egg_info]\ntag_build =\n";
2004   close($pysdk_cfg);
2005 }
2006
2007 if (-e "$destdir/crunch_scripts/install") {
2008     shell_or_die (undef, "$destdir/crunch_scripts/install", $install_dir);
2009 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
2010     # Old version
2011     shell_or_die (undef, "./tests/autotests.sh", $install_dir);
2012 } elsif (-e "./install.sh") {
2013     shell_or_die (undef, "./install.sh", $install_dir);
2014 }
2015
2016 if ($commit) {
2017     unlink "$destdir.commit.new";
2018     symlink ($commit, "$destdir.commit.new") or die "$destdir.commit.new: $!";
2019     rename ("$destdir.commit.new", "$destdir.commit") or die "$destdir.commit: $!";
2020 }
2021
2022 close L;
2023
2024 sub can_run {
2025   my $command_name = shift;
2026   open(my $which, "-|", "which", $command_name);
2027   while (<$which>) { }
2028   close($which);
2029   return ($? == 0);
2030 }
2031
2032 sub shell_or_die
2033 {
2034   my $exitcode = shift;
2035
2036   if ($ENV{"DEBUG"}) {
2037     print STDERR "@_\n";
2038   }
2039   if (system (@_) != 0) {
2040     my $err = $!;
2041     my $code = $?;
2042     my $exitstatus = sprintf("exit %d signal %d", $code >> 8, $code & 0x7f);
2043     open STDERR, ">&STDERR_ORIG";
2044     system ("cat $destdir.log >&2");
2045     warn "@_ failed ($err): $exitstatus";
2046     if (defined($exitcode)) {
2047       exit $exitcode;
2048     }
2049     else {
2050       exit (($code >> 8) || 1);
2051     }
2052   }
2053 }
2054
2055 __DATA__