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 my $tasks_this_level = 0;
 661 foreach my $id (@jobstep_todo) {
 662   $tasks_this_level++ if ($jobstep[$id]->{level} == $level);
 663 }
 664
 665
 666 THISROUND:
 667 for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
 668 {
 669   my $id = $jobstep_todo[$todo_ptr];
 670   my $Jobstep = $jobstep[$id];
 671   if ($Jobstep->{level} != $level)
 672   {
 673     next;
 674   }
 675
 676   pipe $reader{$id}, "writer" or croak ($!);
 677   my $flags = fcntl ($reader{$id}, F_GETFL, 0) or croak ($!);
 678   fcntl ($reader{$id}, F_SETFL, $flags | O_NONBLOCK) or croak ($!);
 679
 680   my $childslot = $freeslot[0];
 681   my $childnode = $slot[$childslot]->{node};
 682   my $childslotname = join (".",
 683                             $slot[$childslot]->{node}->{name},
 684                             $slot[$childslot]->{cpu});
 685
 686   my $childpid = fork();
 687   if ($childpid == 0)
 688   {
 689     $SIG{'INT'} = 'DEFAULT';
 690     $SIG{'QUIT'} = 'DEFAULT';
 691     $SIG{'TERM'} = 'DEFAULT';
 692
 693     foreach (values (%reader))
 694     {
 695       close($_);
 696     }
 697     fcntl ("writer", F_SETFL, 0) or croak ($!); # no close-on-exec
 698     open(STDOUT,">&writer");
 699     open(STDERR,">&writer");
 700
 701     undef $dbh;
 702     undef $sth;
 703
 704     delete $ENV{"GNUPGHOME"};
 705     $ENV{"TASK_UUID"} = $Jobstep->{'arvados_task'}->{'uuid'};
 706     $ENV{"TASK_QSEQUENCE"} = $id;
 707     $ENV{"TASK_SEQUENCE"} = $level;
 708     $ENV{"JOB_SCRIPT"} = $Job->{script};
 709     while (my ($param, $value) = each %{$Job->{script_parameters}}) {
 710       $param =~ tr/a-z/A-Z/;
 711       $ENV{"JOB_PARAMETER_$param"} = $value;
 712     }
 713     $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
 714     $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
 715     $ENV{"TASK_WORK"} = $ENV{"CRUNCH_TMP"}."/task/$childslotname";
 716     $ENV{"HOME"} = $ENV{"TASK_WORK"};
 717     $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}.".keep";
 718     $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 719     $ENV{"CRUNCH_NODE_SLOTS"} = $slot[$childslot]->{node}->{ncpus};
 720     $ENV{"PATH"} = $ENV{"CRUNCH_INSTALL"} . "/bin:" . $ENV{"PATH"};
 721
 722     $ENV{"GZIP"} = "-n";
 723
 724     my $max_node_concurrent_tasks = $ENV{CRUNCH_NODE_SLOTS};
 725     if ($tasks_this_level < $max_node_concurrent_tasks) {
 726       $max_node_concurrent_tasks = $tasks_this_level;
 727     }
 728
 729     my @srunargs = (
 730       "srun",
 731       "--nodelist=".$childnode->{name},
 732       qw(-n1 -c1 -N1 -D), $ENV{'TMPDIR'},
 733       "--job-name=$job_id.$id.$$",
 734         );
 735     my $command =
 736         "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
 737         ."mkdir -p $ENV{CRUNCH_TMP} $ENV{JOB_WORK} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
 738         ."&& cd $ENV{CRUNCH_TMP} "
 739         # These environment variables get used explicitly later in
 740         # $command.  No tool is expected to read these values directly.
 741         .q{&& MEM=$(awk '($1 == "MemTotal:"){print $2}' </proc/meminfo) }
 742         .q{&& SWAP=$(awk '($1 == "SwapTotal:"){print $2}' </proc/meminfo) }
 743         ."&& MEMLIMIT=\$(( (\$MEM * 95) / ($max_node_concurrent_tasks * 100) )) "
 744         ."&& let SWAPLIMIT=\$MEMLIMIT+\$SWAP ";
 745     $command .= "&& exec arv-mount --by-id --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
 746     if ($docker_hash)
 747     {
 748       my $cidfile = "$ENV{CRUNCH_TMP}/$Jobstep->{arvados_task}->{uuid}-$Jobstep->{failures}.cid";
 749       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$cidfile -poll=10000 ";
 750       $command .= "$docker_bin run --rm=true --attach=stdout --attach=stderr --attach=stdin -i --user=crunch --cidfile=$cidfile --sig-proxy ";
 751       # We only set memory limits if Docker lets us limit both memory and swap.
 752       # Memory limits alone have been supported longer, but subprocesses tend
 753       # to get SIGKILL if they exceed that without any swap limit set.
 754       # See #5642 for additional background.
 755       if ($docker_limitmem) {
 756         $command .= "--memory=\${MEMLIMIT}k --memory-swap=\${SWAPLIMIT}k ";
 757       }
 758
 759       # Dynamically configure the container to use the host system as its
 760       # DNS server.  Get the host's global addresses from the ip command,
 761       # and turn them into docker --dns options using gawk.
 762       $command .=
 763           q{$(ip -o address show scope global |
 764               gawk 'match($4, /^([0-9\.:]+)\//, x){print "--dns", x[1]}') };
 765
 766       # The source tree and $destdir directory (which we have
 767       # installed on the worker host) are available in the container,
 768       # under the same path.
 769       $command .= "--volume=\Q$ENV{CRUNCH_SRC}:$ENV{CRUNCH_SRC}:ro\E ";
 770       $command .= "--volume=\Q$ENV{CRUNCH_INSTALL}:$ENV{CRUNCH_INSTALL}:ro\E ";
 771
 772       # Currently, we make arv-mount's mount point appear at /keep
 773       # inside the container (instead of using the same path as the
 774       # host like we do with CRUNCH_SRC and CRUNCH_INSTALL). However,
 775       # crunch scripts and utilities must not rely on this. They must
 776       # use $TASK_KEEPMOUNT.
 777       $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E ";
 778       $ENV{TASK_KEEPMOUNT} = "/keep";
 779
 780       # TASK_WORK is almost exactly like a docker data volume: it
 781       # starts out empty, is writable, and persists until no
 782       # containers use it any more. We don't use --volumes-from to
 783       # share it with other containers: it is only accessible to this
 784       # task, and it goes away when this task stops.
 785       #
 786       # However, a docker data volume is writable only by root unless
 787       # the mount point already happens to exist in the container with
 788       # different permissions. Therefore, we [1] assume /tmp already
 789       # exists in the image and is writable by the crunch user; [2]
 790       # avoid putting TASK_WORK inside CRUNCH_TMP (which won't be
 791       # writable if they are created by docker while setting up the
 792       # other --volumes); and [3] create $TASK_WORK inside the
 793       # container using $build_script.
 794       $command .= "--volume=/tmp ";
 795       $ENV{"TASK_WORK"} = "/tmp/crunch-job-task-work/$childslotname";
 796       $ENV{"HOME"} = $ENV{"TASK_WORK"};
 797       $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 798
 799       # TODO: Share a single JOB_WORK volume across all task
 800       # containers on a given worker node, and delete it when the job
 801       # ends (and, in case that doesn't work, when the next job
 802       # starts).
 803       #
 804       # For now, use the same approach as TASK_WORK above.
 805       $ENV{"JOB_WORK"} = "/tmp/crunch-job-work";
 806
 807       while (my ($env_key, $env_val) = each %ENV)
 808       {
 809         if ($env_key =~ /^(ARVADOS|CRUNCH|JOB|TASK)_/) {
 810           $command .= "--env=\Q$env_key=$env_val\E ";
 811         }
 812       }
 813       $command .= "--env=\QHOME=$ENV{HOME}\E ";
 814       $command .= "\Q$docker_hash\E ";
 815       $command .= "stdbuf --output=0 --error=0 ";
 816       $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 817     } else {
 818       # Non-docker run
 819       $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 ";
 820       $command .= "stdbuf --output=0 --error=0 ";
 821       $command .= "perl - $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 822     }
 823
 824     my @execargs = ('bash', '-c', $command);
 825     srun (\@srunargs, \@execargs, undef, $build_script);
 826     # exec() failed, we assume nothing happened.
 827     die "srun() failed on build script\n";
 828   }
 829   close("writer");
 830   if (!defined $childpid)
 831   {
 832     close $reader{$id};
 833     delete $reader{$id};
 834     next;
 835   }
 836   shift @freeslot;
 837   $proc{$childpid} = { jobstep => $id,
 838                        time => time,
 839                        slot => $childslot,
 840                        jobstepname => "$job_id.$id.$childpid",
 841                      };
 842   croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
 843   $slot[$childslot]->{pid} = $childpid;
 844
 845   Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
 846   Log ($id, "child $childpid started on $childslotname");
 847   $Jobstep->{starttime} = time;
 848   $Jobstep->{node} = $childnode->{name};
 849   $Jobstep->{slotindex} = $childslot;
 850   delete $Jobstep->{stderr};
 851   delete $Jobstep->{finishtime};
 852
 853   $Jobstep->{'arvados_task'}->{started_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{starttime});
 854   $Jobstep->{'arvados_task'}->save;
 855
 856   splice @jobstep_todo, $todo_ptr, 1;
 857   --$todo_ptr;
 858
 859   $progress_is_dirty = 1;
 860
 861   while (!@freeslot
 862          ||
 863          (@slot > @freeslot && $todo_ptr+1 > $#jobstep_todo))
 864   {
 865     last THISROUND if $main::please_freeze || defined($main::success);
 866     if ($main::please_info)
 867     {
 868       $main::please_info = 0;
 869       freeze();
 870       create_output_collection();
 871       save_meta(1);
 872       update_progress_stats();
 873     }
 874     my $gotsome
 875         = readfrompipes ()
 876         + reapchildren ();
 877     if (!$gotsome)
 878     {
 879       check_refresh_wanted();
 880       check_squeue();
 881       update_progress_stats();
 882       select (undef, undef, undef, 0.1);
 883     }
 884     elsif (time - $progress_stats_updated >= 30)
 885     {
 886       update_progress_stats();
 887     }
 888     if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
 889         ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
 890     {
 891       my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
 892           .($thisround_failed+$thisround_succeeded)
 893           .") -- giving up on this round";
 894       Log (undef, $message);
 895       last THISROUND;
 896     }
 897
 898     # move slots from freeslot to holdslot (or back to freeslot) if necessary
 899     for (my $i=$#freeslot; $i>=0; $i--) {
 900       if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
 901         push @holdslot, (splice @freeslot, $i, 1);
 902       }
 903     }
 904     for (my $i=$#holdslot; $i>=0; $i--) {
 905       if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
 906         push @freeslot, (splice @holdslot, $i, 1);
 907       }
 908     }
 909
 910     # give up if no nodes are succeeding
 911     if (!grep { $_->{node}->{losing_streak} == 0 &&
 912                     $_->{node}->{hold_count} < 4 } @slot) {
 913       my $message = "Every node has failed -- giving up on this round";
 914       Log (undef, $message);
 915       last THISROUND;
 916     }
 917   }
 918 }
 919
 920
 921 push @freeslot, splice @holdslot;
 922 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
 923
 924
 925 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
 926 while (%proc)
 927 {
 928   if ($main::please_continue) {
 929     $main::please_continue = 0;
 930     goto THISROUND;
 931   }
 932   $main::please_info = 0, freeze(), create_output_collection(), save_meta(1) if $main::please_info;
 933   readfrompipes ();
 934   if (!reapchildren())
 935   {
 936     check_refresh_wanted();
 937     check_squeue();
 938     update_progress_stats();
 939     select (undef, undef, undef, 0.1);
 940     killem (keys %proc) if $main::please_freeze;
 941   }
 942 }
 943
 944 update_progress_stats();
 945 freeze_if_want_freeze();
 946
 947
 948 if (!defined $main::success)
 949 {
 950   if (@jobstep_todo &&
 951       $thisround_succeeded == 0 &&
 952       ($thisround_failed == 0 || $thisround_failed > 4))
 953   {
 954     my $message = "stop because $thisround_failed tasks failed and none succeeded";
 955     Log (undef, $message);
 956     $main::success = 0;
 957   }
 958   if (!@jobstep_todo)
 959   {
 960     $main::success = 1;
 961   }
 962 }
 963
 964 goto ONELEVEL if !defined $main::success;
 965
 966
 967 release_allocation();
 968 freeze();
 969 my $collated_output = &create_output_collection();
 970
 971 if (!$collated_output) {
 972   Log (undef, "Failed to write output collection");
 973 }
 974 else {
 975   Log(undef, "job output $collated_output");
 976   $Job->update_attributes('output' => $collated_output);
 977 }
 978
 979 Log (undef, "finish");
 980
 981 save_meta();
 982
 983 my $final_state;
 984 if ($collated_output && $main::success) {
 985   $final_state = 'Complete';
 986 } else {
 987   $final_state = 'Failed';
 988 }
 989 $Job->update_attributes('state' => $final_state);
 990
 991 exit (($final_state eq 'Complete') ? 0 : 1);
 992
 993
 994
 995 sub update_progress_stats
 996 {
 997   $progress_stats_updated = time;
 998   return if !$progress_is_dirty;
 999   my ($todo, $done, $running) = (scalar @jobstep_todo,
1000                                  scalar @jobstep_done,
1001                                  scalar @slot - scalar @freeslot - scalar @holdslot);
1002   $Job->{'tasks_summary'} ||= {};
1003   $Job->{'tasks_summary'}->{'todo'} = $todo;
1004   $Job->{'tasks_summary'}->{'done'} = $done;
1005   $Job->{'tasks_summary'}->{'running'} = $running;
1006   $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
1007   Log (undef, "status: $done done, $running running, $todo todo");
1008   $progress_is_dirty = 0;
1009 }
1010
1011
1012
1013 sub reapchildren
1014 {
1015   my $pid = waitpid (-1, WNOHANG);
1016   return 0 if $pid <= 0;
1017
1018   my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
1019                   . "."
1020                   . $slot[$proc{$pid}->{slot}]->{cpu});
1021   my $jobstepid = $proc{$pid}->{jobstep};
1022   my $elapsed = time - $proc{$pid}->{time};
1023   my $Jobstep = $jobstep[$jobstepid];
1024
1025   my $childstatus = $?;
1026   my $exitvalue = $childstatus >> 8;
1027   my $exitinfo = "exit ".exit_status_s($childstatus);
1028   $Jobstep->{'arvados_task'}->reload;
1029   my $task_success = $Jobstep->{'arvados_task'}->{success};
1030
1031   Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
1032
1033   if (!defined $task_success) {
1034     # task did not indicate one way or the other --> fail
1035     $Jobstep->{'arvados_task'}->{success} = 0;
1036     $Jobstep->{'arvados_task'}->save;
1037     $task_success = 0;
1038   }
1039
1040   if (!$task_success)
1041   {
1042     my $temporary_fail;
1043     $temporary_fail ||= $Jobstep->{node_fail};
1044     $temporary_fail ||= ($exitvalue == TASK_TEMPFAIL);
1045
1046     ++$thisround_failed;
1047     ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
1048
1049     # Check for signs of a failed or misconfigured node
1050     if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
1051         2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
1052       # Don't count this against jobstep failure thresholds if this
1053       # node is already suspected faulty and srun exited quickly
1054       if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
1055           $elapsed < 5) {
1056         Log ($jobstepid, "blaming failure on suspect node " .
1057              $slot[$proc{$pid}->{slot}]->{node}->{name});
1058         $temporary_fail ||= 1;
1059       }
1060       ban_node_by_slot($proc{$pid}->{slot});
1061     }
1062
1063     Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
1064                              ++$Jobstep->{'failures'},
1065                              $temporary_fail ? 'temporary ' : 'permanent',
1066                              $elapsed));
1067
1068     if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
1069       # Give up on this task, and the whole job
1070       $main::success = 0;
1071     }
1072     # Put this task back on the todo queue
1073     push @jobstep_todo, $jobstepid;
1074     $Job->{'tasks_summary'}->{'failed'}++;
1075   }
1076   else
1077   {
1078     ++$thisround_succeeded;
1079     $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
1080     $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
1081     push @jobstep_done, $jobstepid;
1082     Log ($jobstepid, "success in $elapsed seconds");
1083   }
1084   $Jobstep->{exitcode} = $childstatus;
1085   $Jobstep->{finishtime} = time;
1086   $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
1087   $Jobstep->{'arvados_task'}->save;
1088   process_stderr ($jobstepid, $task_success);
1089   Log ($jobstepid, sprintf("task output (%d bytes): %s",
1090                            length($Jobstep->{'arvados_task'}->{output}),
1091                            $Jobstep->{'arvados_task'}->{output}));
1092
1093   close $reader{$jobstepid};
1094   delete $reader{$jobstepid};
1095   delete $slot[$proc{$pid}->{slot}]->{pid};
1096   push @freeslot, $proc{$pid}->{slot};
1097   delete $proc{$pid};
1098
1099   if ($task_success) {
1100     # Load new tasks
1101     my $newtask_list = [];
1102     my $newtask_results;
1103     do {
1104       $newtask_results = api_call(
1105         "job_tasks/list",
1106         'where' => {
1107           'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
1108         },
1109         'order' => 'qsequence',
1110         'offset' => scalar(@$newtask_list),
1111       );
1112       push(@$newtask_list, @{$newtask_results->{items}});
1113     } while (@{$newtask_results->{items}});
1114     foreach my $arvados_task (@$newtask_list) {
1115       my $jobstep = {
1116         'level' => $arvados_task->{'sequence'},
1117         'failures' => 0,
1118         'arvados_task' => $arvados_task
1119       };
1120       push @jobstep, $jobstep;
1121       push @jobstep_todo, $#jobstep;
1122     }
1123   }
1124
1125   $progress_is_dirty = 1;
1126   1;
1127 }
1128
1129 sub check_refresh_wanted
1130 {
1131   my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
1132   if (@stat && $stat[9] > $latest_refresh) {
1133     $latest_refresh = scalar time;
1134     my $Job2 = api_call("jobs/get", uuid => $jobspec);
1135     for my $attr ('cancelled_at',
1136                   'cancelled_by_user_uuid',
1137                   'cancelled_by_client_uuid',
1138                   'state') {
1139       $Job->{$attr} = $Job2->{$attr};
1140     }
1141     if ($Job->{'state'} ne "Running") {
1142       if ($Job->{'state'} eq "Cancelled") {
1143         Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
1144       } else {
1145         Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
1146       }
1147       $main::success = 0;
1148       $main::please_freeze = 1;
1149     }
1150   }
1151 }
1152
1153 sub check_squeue
1154 {
1155   # return if the kill list was checked <4 seconds ago
1156   if (defined $squeue_kill_checked && $squeue_kill_checked > time - 4)
1157   {
1158     return;
1159   }
1160   $squeue_kill_checked = time;
1161
1162   # use killem() on procs whose killtime is reached
1163   for (keys %proc)
1164   {
1165     if (exists $proc{$_}->{killtime}
1166         && $proc{$_}->{killtime} <= time)
1167     {
1168       killem ($_);
1169     }
1170   }
1171
1172   # return if the squeue was checked <60 seconds ago
1173   if (defined $squeue_checked && $squeue_checked > time - 60)
1174   {
1175     return;
1176   }
1177   $squeue_checked = time;
1178
1179   if (!$have_slurm)
1180   {
1181     # here is an opportunity to check for mysterious problems with local procs
1182     return;
1183   }
1184
1185   # get a list of steps still running
1186   my @squeue = `squeue -s -h -o '%i %j' && echo ok`;
1187   chop @squeue;
1188   if ($squeue[-1] ne "ok")
1189   {
1190     return;
1191   }
1192   pop @squeue;
1193
1194   # which of my jobsteps are running, according to squeue?
1195   my %ok;
1196   foreach (@squeue)
1197   {
1198     if (/^(\d+)\.(\d+) (\S+)/)
1199     {
1200       if ($1 eq $ENV{SLURM_JOBID})
1201       {
1202         $ok{$3} = 1;
1203       }
1204     }
1205   }
1206
1207   # which of my active child procs (>60s old) were not mentioned by squeue?
1208   foreach (keys %proc)
1209   {
1210     if ($proc{$_}->{time} < time - 60
1211         && !exists $ok{$proc{$_}->{jobstepname}}
1212         && !exists $proc{$_}->{killtime})
1213     {
1214       # kill this proc if it hasn't exited in 30 seconds
1215       $proc{$_}->{killtime} = time + 30;
1216     }
1217   }
1218 }
1219
1220
1221 sub release_allocation
1222 {
1223   if ($have_slurm)
1224   {
1225     Log (undef, "release job allocation");
1226     system "scancel $ENV{SLURM_JOBID}";
1227   }
1228 }
1229
1230
1231 sub readfrompipes
1232 {
1233   my $gotsome = 0;
1234   foreach my $job (keys %reader)
1235   {
1236     my $buf;
1237     while (0 < sysread ($reader{$job}, $buf, 8192))
1238     {
1239       print STDERR $buf if $ENV{CRUNCH_DEBUG};
1240       $jobstep[$job]->{stderr} .= $buf;
1241       preprocess_stderr ($job);
1242       if (length ($jobstep[$job]->{stderr}) > 16384)
1243       {
1244         substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1245       }
1246       $gotsome = 1;
1247     }
1248   }
1249   return $gotsome;
1250 }
1251
1252
1253 sub preprocess_stderr
1254 {
1255   my $job = shift;
1256
1257   while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1258     my $line = $1;
1259     substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1260     Log ($job, "stderr $line");
1261     if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1262       # whoa.
1263       $main::please_freeze = 1;
1264     }
1265     elsif ($line =~ /(srun: error: (Node failure on|Unable to create job step|.*: Communication connection failure))|arvados.errors.Keep/) {
1266       $jobstep[$job]->{node_fail} = 1;
1267       ban_node_by_slot($jobstep[$job]->{slotindex});
1268     }
1269   }
1270 }
1271
1272
1273 sub process_stderr
1274 {
1275   my $job = shift;
1276   my $task_success = shift;
1277   preprocess_stderr ($job);
1278
1279   map {
1280     Log ($job, "stderr $_");
1281   } split ("\n", $jobstep[$job]->{stderr});
1282 }
1283
1284 sub fetch_block
1285 {
1286   my $hash = shift;
1287   my $keep;
1288   if (!open($keep, "-|", "arv-get", "--retries", retry_count(), $hash)) {
1289     Log(undef, "fetch_block run error from arv-get $hash: $!");
1290     return undef;
1291   }
1292   my $output_block = "";
1293   while (1) {
1294     my $buf;
1295     my $bytes = sysread($keep, $buf, 1024 * 1024);
1296     if (!defined $bytes) {
1297       Log(undef, "fetch_block read error from arv-get: $!");
1298       $output_block = undef;
1299       last;
1300     } elsif ($bytes == 0) {
1301       # sysread returns 0 at the end of the pipe.
1302       last;
1303     } else {
1304       # some bytes were read into buf.
1305       $output_block .= $buf;
1306     }
1307   }
1308   close $keep;
1309   if ($?) {
1310     Log(undef, "fetch_block arv-get exited " . exit_status_s($?));
1311     $output_block = undef;
1312   }
1313   return $output_block;
1314 }
1315
1316 # Create a collection by concatenating the output of all tasks (each
1317 # task's output is either a manifest fragment, a locator for a
1318 # manifest fragment stored in Keep, or nothing at all). Return the
1319 # portable_data_hash of the new collection.
1320 sub create_output_collection
1321 {
1322   Log (undef, "collate");
1323
1324   my ($child_out, $child_in);
1325   my $pid = open2($child_out, $child_in, 'python', '-c', q{
1326 import arvados
1327 import sys
1328 print (arvados.api("v1").collections().
1329        create(body={"manifest_text": sys.stdin.read()}).
1330        execute(num_retries=int(sys.argv[1]))["portable_data_hash"])
1331 }, retry_count());
1332
1333   my $task_idx = -1;
1334   my $manifest_size = 0;
1335   for (@jobstep)
1336   {
1337     ++$task_idx;
1338     my $output = $_->{'arvados_task'}->{output};
1339     next if (!defined($output));
1340     my $next_write;
1341     if ($output =~ /^[0-9a-f]{32}(\+\S+)*$/) {
1342       $next_write = fetch_block($output);
1343     } else {
1344       $next_write = $output;
1345     }
1346     if (defined($next_write)) {
1347       if (!defined(syswrite($child_in, $next_write))) {
1348         # There's been an error writing.  Stop the loop.
1349         # We'll log details about the exit code later.
1350         last;
1351       } else {
1352         $manifest_size += length($next_write);
1353       }
1354     } else {
1355       my $uuid = $_->{'arvados_task'}->{'uuid'};
1356       Log (undef, "Error retrieving '$output' output by task $task_idx ($uuid)");
1357       $main::success = 0;
1358     }
1359   }
1360   close($child_in);
1361   Log(undef, "collated output manifest text to send to API server is $manifest_size bytes with access tokens");
1362
1363   my $joboutput;
1364   my $s = IO::Select->new($child_out);
1365   if ($s->can_read(120)) {
1366     sysread($child_out, $joboutput, 1024 * 1024);
1367     waitpid($pid, 0);
1368     if ($?) {
1369       Log(undef, "output collection creation exited " . exit_status_s($?));
1370       $joboutput = undef;
1371     } else {
1372       chomp($joboutput);
1373     }
1374   } else {
1375     Log (undef, "timed out while creating output collection");
1376     foreach my $signal (2, 2, 2, 15, 15, 9) {
1377       kill($signal, $pid);
1378       last if waitpid($pid, WNOHANG) == -1;
1379       sleep(1);
1380     }
1381   }
1382   close($child_out);
1383
1384   return $joboutput;
1385 }
1386
1387
1388 sub killem
1389 {
1390   foreach (@_)
1391   {
1392     my $sig = 2;                # SIGINT first
1393     if (exists $proc{$_}->{"sent_$sig"} &&
1394         time - $proc{$_}->{"sent_$sig"} > 4)
1395     {
1396       $sig = 15;                # SIGTERM if SIGINT doesn't work
1397     }
1398     if (exists $proc{$_}->{"sent_$sig"} &&
1399         time - $proc{$_}->{"sent_$sig"} > 4)
1400     {
1401       $sig = 9;                 # SIGKILL if SIGTERM doesn't work
1402     }
1403     if (!exists $proc{$_}->{"sent_$sig"})
1404     {
1405       Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1406       kill $sig, $_;
1407       select (undef, undef, undef, 0.1);
1408       if ($sig == 2)
1409       {
1410         kill $sig, $_;     # srun wants two SIGINT to really interrupt
1411       }
1412       $proc{$_}->{"sent_$sig"} = time;
1413       $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1414     }
1415   }
1416 }
1417
1418
1419 sub fhbits
1420 {
1421   my($bits);
1422   for (@_) {
1423     vec($bits,fileno($_),1) = 1;
1424   }
1425   $bits;
1426 }
1427
1428
1429 # Send log output to Keep via arv-put.
1430 #
1431 # $log_pipe_in and $log_pipe_out are the input and output filehandles to the arv-put pipe.
1432 # $log_pipe_pid is the pid of the arv-put subprocess.
1433 #
1434 # The only functions that should access these variables directly are:
1435 #
1436 # log_writer_start($logfilename)
1437 #     Starts an arv-put pipe, reading data on stdin and writing it to
1438 #     a $logfilename file in an output collection.
1439 #
1440 # log_writer_send($txt)
1441 #     Writes $txt to the output log collection.
1442 #
1443 # log_writer_finish()
1444 #     Closes the arv-put pipe and returns the output that it produces.
1445 #
1446 # log_writer_is_active()
1447 #     Returns a true value if there is currently a live arv-put
1448 #     process, false otherwise.
1449 #
1450 my ($log_pipe_in, $log_pipe_out, $log_pipe_pid);
1451
1452 sub log_writer_start($)
1453 {
1454   my $logfilename = shift;
1455   $log_pipe_pid = open2($log_pipe_out, $log_pipe_in,
1456                         'arv-put',
1457                         '--portable-data-hash',
1458                         '--project-uuid', $Job->{owner_uuid},
1459                         '--retries', '3',
1460                         '--name', $logfilename,
1461                         '--filename', $logfilename,
1462                         '-');
1463 }
1464
1465 sub log_writer_send($)
1466 {
1467   my $txt = shift;
1468   print $log_pipe_in $txt;
1469 }
1470
1471 sub log_writer_finish()
1472 {
1473   return unless $log_pipe_pid;
1474
1475   close($log_pipe_in);
1476   my $arv_put_output;
1477
1478   my $s = IO::Select->new($log_pipe_out);
1479   if ($s->can_read(120)) {
1480     sysread($log_pipe_out, $arv_put_output, 1024);
1481     chomp($arv_put_output);
1482   } else {
1483     Log (undef, "timed out reading from 'arv-put'");
1484   }
1485
1486   waitpid($log_pipe_pid, 0);
1487   $log_pipe_pid = $log_pipe_in = $log_pipe_out = undef;
1488   if ($?) {
1489     Log("log_writer_finish: arv-put exited ".exit_status_s($?))
1490   }
1491
1492   return $arv_put_output;
1493 }
1494
1495 sub log_writer_is_active() {
1496   return $log_pipe_pid;
1497 }
1498
1499 sub Log                         # ($jobstep_id, $logmessage)
1500 {
1501   if ($_[1] =~ /\n/) {
1502     for my $line (split (/\n/, $_[1])) {
1503       Log ($_[0], $line);
1504     }
1505     return;
1506   }
1507   my $fh = select STDERR; $|=1; select $fh;
1508   my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1509   $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1510   $message .= "\n";
1511   my $datetime;
1512   if (log_writer_is_active() || -t STDERR) {
1513     my @gmtime = gmtime;
1514     $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1515                          $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1516   }
1517   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1518
1519   if (log_writer_is_active()) {
1520     log_writer_send($datetime . " " . $message);
1521   }
1522 }
1523
1524
1525 sub croak
1526 {
1527   my ($package, $file, $line) = caller;
1528   my $message = "@_ at $file line $line\n";
1529   Log (undef, $message);
1530   freeze() if @jobstep_todo;
1531   create_output_collection() if @jobstep_todo;
1532   cleanup();
1533   save_meta();
1534   die;
1535 }
1536
1537
1538 sub cleanup
1539 {
1540   return unless $Job;
1541   if ($Job->{'state'} eq 'Cancelled') {
1542     $Job->update_attributes('finished_at' => scalar gmtime);
1543   } else {
1544     $Job->update_attributes('state' => 'Failed');
1545   }
1546 }
1547
1548
1549 sub save_meta
1550 {
1551   my $justcheckpoint = shift; # false if this will be the last meta saved
1552   return if $justcheckpoint;  # checkpointing is not relevant post-Warehouse.pm
1553   return unless log_writer_is_active();
1554
1555   my $loglocator = log_writer_finish();
1556   Log (undef, "log manifest is $loglocator");
1557   $Job->{'log'} = $loglocator;
1558   $Job->update_attributes('log', $loglocator);
1559 }
1560
1561
1562 sub freeze_if_want_freeze
1563 {
1564   if ($main::please_freeze)
1565   {
1566     release_allocation();
1567     if (@_)
1568     {
1569       # kill some srun procs before freeze+stop
1570       map { $proc{$_} = {} } @_;
1571       while (%proc)
1572       {
1573         killem (keys %proc);
1574         select (undef, undef, undef, 0.1);
1575         my $died;
1576         while (($died = waitpid (-1, WNOHANG)) > 0)
1577         {
1578           delete $proc{$died};
1579         }
1580       }
1581     }
1582     freeze();
1583     create_output_collection();
1584     cleanup();
1585     save_meta();
1586     exit 1;
1587   }
1588 }
1589
1590
1591 sub freeze
1592 {
1593   Log (undef, "Freeze not implemented");
1594   return;
1595 }
1596
1597
1598 sub thaw
1599 {
1600   croak ("Thaw not implemented");
1601 }
1602
1603
1604 sub freezequote
1605 {
1606   my $s = shift;
1607   $s =~ s/\\/\\\\/g;
1608   $s =~ s/\n/\\n/g;
1609   return $s;
1610 }
1611
1612
1613 sub freezeunquote
1614 {
1615   my $s = shift;
1616   $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1617   return $s;
1618 }
1619
1620
1621 sub srun
1622 {
1623   my $srunargs = shift;
1624   my $execargs = shift;
1625   my $opts = shift || {};
1626   my $stdin = shift;
1627   my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1628
1629   $Data::Dumper::Terse = 1;
1630   $Data::Dumper::Indent = 0;
1631   my $show_cmd = Dumper($args);
1632   $show_cmd =~ s/(TOKEN\\*=)[^\s\']+/${1}[...]/g;
1633   $show_cmd =~ s/\n/ /g;
1634   warn "starting: $show_cmd\n";
1635
1636   if (defined $stdin) {
1637     my $child = open STDIN, "-|";
1638     defined $child or die "no fork: $!";
1639     if ($child == 0) {
1640       print $stdin or die $!;
1641       close STDOUT or die $!;
1642       exit 0;
1643     }
1644   }
1645
1646   return system (@$args) if $opts->{fork};
1647
1648   exec @$args;
1649   warn "ENV size is ".length(join(" ",%ENV));
1650   die "exec failed: $!: @$args";
1651 }
1652
1653
1654 sub ban_node_by_slot {
1655   # Don't start any new jobsteps on this node for 60 seconds
1656   my $slotid = shift;
1657   $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1658   $slot[$slotid]->{node}->{hold_count}++;
1659   Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1660 }
1661
1662 sub must_lock_now
1663 {
1664   my ($lockfile, $error_message) = @_;
1665   open L, ">", $lockfile or croak("$lockfile: $!");
1666   if (!flock L, LOCK_EX|LOCK_NB) {
1667     croak("Can't lock $lockfile: $error_message\n");
1668   }
1669 }
1670
1671 sub find_docker_image {
1672   # Given a Keep locator, check to see if it contains a Docker image.
1673   # If so, return its stream name and Docker hash.
1674   # If not, return undef for both values.
1675   my $locator = shift;
1676   my ($streamname, $filename);
1677   my $image = api_call("collections/get", uuid => $locator);
1678   if ($image) {
1679     foreach my $line (split(/\n/, $image->{manifest_text})) {
1680       my @tokens = split(/\s+/, $line);
1681       next if (!@tokens);
1682       $streamname = shift(@tokens);
1683       foreach my $filedata (grep(/^\d+:\d+:/, @tokens)) {
1684         if (defined($filename)) {
1685           return (undef, undef);  # More than one file in the Collection.
1686         } else {
1687           $filename = (split(/:/, $filedata, 3))[2];
1688         }
1689       }
1690     }
1691   }
1692   if (defined($filename) and ($filename =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
1693     return ($streamname, $1);
1694   } else {
1695     return (undef, undef);
1696   }
1697 }
1698
1699 sub retry_count {
1700   # Calculate the number of times an operation should be retried,
1701   # assuming exponential backoff, and that we're willing to retry as
1702   # long as tasks have been running.  Enforce a minimum of 3 retries.
1703   my ($starttime, $endtime, $timediff, $retries);
1704   if (@jobstep) {
1705     $starttime = $jobstep[0]->{starttime};
1706     $endtime = $jobstep[-1]->{finishtime};
1707   }
1708   if (!defined($starttime)) {
1709     $timediff = 0;
1710   } elsif (!defined($endtime)) {
1711     $timediff = time - $starttime;
1712   } else {
1713     $timediff = ($endtime - $starttime) - (time - $endtime);
1714   }
1715   if ($timediff > 0) {
1716     $retries = int(log($timediff) / log(2));
1717   } else {
1718     $retries = 1;  # Use the minimum.
1719   }
1720   return ($retries > 3) ? $retries : 3;
1721 }
1722
1723 sub retry_op {
1724   # Pass in two function references.
1725   # This method will be called with the remaining arguments.
1726   # If it dies, retry it with exponential backoff until it succeeds,
1727   # or until the current retry_count is exhausted.  After each failure
1728   # that can be retried, the second function will be called with
1729   # the current try count (0-based), next try time, and error message.
1730   my $operation = shift;
1731   my $retry_callback = shift;
1732   my $retries = retry_count();
1733   foreach my $try_count (0..$retries) {
1734     my $next_try = time + (2 ** $try_count);
1735     my $result = eval { $operation->(@_); };
1736     if (!$@) {
1737       return $result;
1738     } elsif ($try_count < $retries) {
1739       $retry_callback->($try_count, $next_try, $@);
1740       my $sleep_time = $next_try - time;
1741       sleep($sleep_time) if ($sleep_time > 0);
1742     }
1743   }
1744   # Ensure the error message ends in a newline, so Perl doesn't add
1745   # retry_op's line number to it.
1746   chomp($@);
1747   die($@ . "\n");
1748 }
1749
1750 sub api_call {
1751   # Pass in a /-separated API method name, and arguments for it.
1752   # This function will call that method, retrying as needed until
1753   # the current retry_count is exhausted, with a log on the first failure.
1754   my $method_name = shift;
1755   my $log_api_retry = sub {
1756     my ($try_count, $next_try_at, $errmsg) = @_;
1757     $errmsg =~ s/\s*\bat \Q$0\E line \d+\.?\s*//;
1758     $errmsg =~ s/\s/ /g;
1759     $errmsg =~ s/\s+$//;
1760     my $retry_msg;
1761     if ($next_try_at < time) {
1762       $retry_msg = "Retrying.";
1763     } else {
1764       my $next_try_fmt = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($next_try_at);
1765       $retry_msg = "Retrying at $next_try_fmt.";
1766     }
1767     Log(undef, "API method $method_name failed: $errmsg. $retry_msg");
1768   };
1769   my $method = $arv;
1770   foreach my $key (split(/\//, $method_name)) {
1771     $method = $method->{$key};
1772   }
1773   return retry_op(sub { $method->execute(@_); }, $log_api_retry, @_);
1774 }
1775
1776 sub exit_status_s {
1777   # Given a $?, return a human-readable exit code string like "0" or
1778   # "1" or "0 with signal 1" or "1 with signal 11".
1779   my $exitcode = shift;
1780   my $s = $exitcode >> 8;
1781   if ($exitcode & 0x7f) {
1782     $s .= " with signal " . ($exitcode & 0x7f);
1783   }
1784   if ($exitcode & 0x80) {
1785     $s .= " with core dump";
1786   }
1787   return $s;
1788 }
1789
1790 sub handle_readall {
1791   # Pass in a glob reference to a file handle.
1792   # Read all its contents and return them as a string.
1793   my $fh_glob_ref = shift;
1794   local $/ = undef;
1795   return <$fh_glob_ref>;
1796 }
1797
1798 sub tar_filename_n {
1799   my $n = shift;
1800   return sprintf("%s/git.%s.%d.tar", $ENV{CRUNCH_TMP}, $job_id, $n);
1801 }
1802
1803 sub add_git_archive {
1804   # Pass in a git archive command as a string or list, a la system().
1805   # This method will save its output to be included in the archive sent to the
1806   # build script.
1807   my $git_input;
1808   $git_tar_count++;
1809   if (!open(GIT_ARCHIVE, ">", tar_filename_n($git_tar_count))) {
1810     croak("Failed to save git archive: $!");
1811   }
1812   my $git_pid = open2(">&GIT_ARCHIVE", $git_input, @_);
1813   close($git_input);
1814   waitpid($git_pid, 0);
1815   close(GIT_ARCHIVE);
1816   if ($?) {
1817     croak("Failed to save git archive: git exited " . exit_status_s($?));
1818   }
1819 }
1820
1821 sub combined_git_archive {
1822   # Combine all saved tar archives into a single archive, then return its
1823   # contents in a string.  Return undef if no archives have been saved.
1824   if ($git_tar_count < 1) {
1825     return undef;
1826   }
1827   my $base_tar_name = tar_filename_n(1);
1828   foreach my $tar_to_append (map { tar_filename_n($_); } (2..$git_tar_count)) {
1829     my $tar_exit = system("tar", "-Af", $base_tar_name, $tar_to_append);
1830     if ($tar_exit != 0) {
1831       croak("Error preparing build archive: tar -A exited " .
1832             exit_status_s($tar_exit));
1833     }
1834   }
1835   if (!open(GIT_TAR, "<", $base_tar_name)) {
1836     croak("Could not open build archive: $!");
1837   }
1838   my $tar_contents = handle_readall(\*GIT_TAR);
1839   close(GIT_TAR);
1840   return $tar_contents;
1841 }
1842
1843 __DATA__
1844 #!/usr/bin/perl
1845 #
1846 # This is crunch-job's internal dispatch script.  crunch-job running on the API
1847 # server invokes this script on individual compute nodes, or localhost if we're
1848 # running a job locally.  It gets called in two modes:
1849 #
1850 # * No arguments: Installation mode.  Read a tar archive from the DATA
1851 #   file handle; it includes the Crunch script's source code, and
1852 #   maybe SDKs as well.  Those should be installed in the proper
1853 #   locations.  This runs outside of any Docker container, so don't try to
1854 #   introspect Crunch's runtime environment.
1855 #
1856 # * With arguments: Crunch script run mode.  This script should set up the
1857 #   environment, then run the command specified in the arguments.  This runs
1858 #   inside any Docker container.
1859
1860 use Fcntl ':flock';
1861 use File::Path qw( make_path remove_tree );
1862 use POSIX qw(getcwd);
1863
1864 use constant TASK_TEMPFAIL => 111;
1865
1866 # Map SDK subdirectories to the path environments they belong to.
1867 my %SDK_ENVVARS = ("perl/lib" => "PERLLIB", "ruby/lib" => "RUBYLIB");
1868
1869 my $destdir = $ENV{"CRUNCH_SRC"};
1870 my $commit = $ENV{"CRUNCH_SRC_COMMIT"};
1871 my $repo = $ENV{"CRUNCH_SRC_URL"};
1872 my $install_dir = $ENV{"CRUNCH_INSTALL"} || (getcwd() . "/opt");
1873 my $job_work = $ENV{"JOB_WORK"};
1874 my $task_work = $ENV{"TASK_WORK"};
1875
1876 for my $dir ($destdir, $job_work, $task_work) {
1877   if ($dir) {
1878     make_path $dir;
1879     -e $dir or die "Failed to create temporary directory ($dir): $!";
1880   }
1881 }
1882
1883 if ($task_work) {
1884   remove_tree($task_work, {keep_root => 1});
1885 }
1886
1887 open(STDOUT_ORIG, ">&", STDOUT);
1888 open(STDERR_ORIG, ">&", STDERR);
1889 open(STDOUT, ">>", "$destdir.log");
1890 open(STDERR, ">&", STDOUT);
1891
1892 ### Crunch script run mode
1893 if (@ARGV) {
1894   # We want to do routine logging during task 0 only.  This gives the user
1895   # the information they need, but avoids repeating the information for every
1896   # task.
1897   my $Log;
1898   if ($ENV{TASK_SEQUENCE} eq "0") {
1899     $Log = sub {
1900       my $msg = shift;
1901       printf STDERR_ORIG "[Crunch] $msg\n", @_;
1902     };
1903   } else {
1904     $Log = sub { };
1905   }
1906
1907   my $python_src = "$install_dir/python";
1908   my $venv_dir = "$job_work/.arvados.venv";
1909   my $venv_built = -e "$venv_dir/bin/activate";
1910   if ((!$venv_built) and (-d $python_src) and can_run("virtualenv")) {
1911     shell_or_die(undef, "virtualenv", "--quiet", "--system-site-packages",
1912                  "--python=python2.7", $venv_dir);
1913     shell_or_die(TASK_TEMPFAIL, "$venv_dir/bin/pip", "--quiet", "install", "-I", $python_src);
1914     $venv_built = 1;
1915     $Log->("Built Python SDK virtualenv");
1916   }
1917
1918   my $pip_bin = "pip";
1919   if ($venv_built) {
1920     $Log->("Running in Python SDK virtualenv");
1921     $pip_bin = "$venv_dir/bin/pip";
1922     my $orig_argv = join(" ", map { quotemeta($_); } @ARGV);
1923     @ARGV = ("/bin/sh", "-ec",
1924              ". \Q$venv_dir/bin/activate\E; exec $orig_argv");
1925   } elsif (-d $python_src) {
1926     $Log->("Warning: virtualenv not found inside Docker container default " .
1927            "\$PATH. Can't install Python SDK.");
1928   }
1929
1930   my $pkgs = `(\Q$pip_bin\E freeze 2>/dev/null | grep arvados) || dpkg-query --show '*arvados*'`;
1931   if ($pkgs) {
1932     $Log->("Using Arvados SDK:");
1933     foreach my $line (split /\n/, $pkgs) {
1934       $Log->($line);
1935     }
1936   } else {
1937     $Log->("Arvados SDK packages not found");
1938   }
1939
1940   while (my ($sdk_dir, $sdk_envkey) = each(%SDK_ENVVARS)) {
1941     my $sdk_path = "$install_dir/$sdk_dir";
1942     if (-d $sdk_path) {
1943       if ($ENV{$sdk_envkey}) {
1944         $ENV{$sdk_envkey} = "$sdk_path:" . $ENV{$sdk_envkey};
1945       } else {
1946         $ENV{$sdk_envkey} = $sdk_path;
1947       }
1948       $Log->("Arvados SDK added to %s", $sdk_envkey);
1949     }
1950   }
1951
1952   close(STDOUT);
1953   close(STDERR);
1954   open(STDOUT, ">&", STDOUT_ORIG);
1955   open(STDERR, ">&", STDERR_ORIG);
1956   exec(@ARGV);
1957   die "Cannot exec `@ARGV`: $!";
1958 }
1959
1960 ### Installation mode
1961 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
1962 flock L, LOCK_EX;
1963 if (readlink ("$destdir.commit") eq $commit && -d $destdir) {
1964   # This version already installed -> nothing to do.
1965   exit(0);
1966 }
1967
1968 unlink "$destdir.commit";
1969 mkdir $destdir;
1970
1971 if (!open(TARX, "|-", "tar", "-xC", $destdir)) {
1972   die "Error launching 'tar -xC $destdir': $!";
1973 }
1974 # If we send too much data to tar in one write (> 4-5 MiB), it stops, and we
1975 # get SIGPIPE.  We must feed it data incrementally.
1976 my $tar_input;
1977 while (read(DATA, $tar_input, 65536)) {
1978   print TARX $tar_input;
1979 }
1980 if(!close(TARX)) {
1981   die "'tar -xC $destdir' exited $?: $!";
1982 }
1983
1984 mkdir $install_dir;
1985
1986 my $sdk_root = "$destdir/.arvados.sdk/sdk";
1987 if (-d $sdk_root) {
1988   foreach my $sdk_lang (("python",
1989                          map { (split /\//, $_, 2)[0]; } keys(%SDK_ENVVARS))) {
1990     if (-d "$sdk_root/$sdk_lang") {
1991       if (!rename("$sdk_root/$sdk_lang", "$install_dir/$sdk_lang")) {
1992         die "Failed to install $sdk_lang SDK: $!";
1993       }
1994     }
1995   }
1996 }
1997
1998 my $python_dir = "$install_dir/python";
1999 if ((-d $python_dir) and can_run("python2.7") and
2000     (system("python2.7", "$python_dir/setup.py", "--quiet", "egg_info") != 0)) {
2001   # egg_info failed, probably when it asked git for a build tag.
2002   # Specify no build tag.
2003   open(my $pysdk_cfg, ">>", "$python_dir/setup.cfg");
2004   print $pysdk_cfg "\n[egg_info]\ntag_build =\n";
2005   close($pysdk_cfg);
2006 }
2007
2008 if (-e "$destdir/crunch_scripts/install") {
2009     shell_or_die (undef, "$destdir/crunch_scripts/install", $install_dir);
2010 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
2011     # Old version
2012     shell_or_die (undef, "./tests/autotests.sh", $install_dir);
2013 } elsif (-e "./install.sh") {
2014     shell_or_die (undef, "./install.sh", $install_dir);
2015 }
2016
2017 if ($commit) {
2018     unlink "$destdir.commit.new";
2019     symlink ($commit, "$destdir.commit.new") or die "$destdir.commit.new: $!";
2020     rename ("$destdir.commit.new", "$destdir.commit") or die "$destdir.commit: $!";
2021 }
2022
2023 close L;
2024
2025 sub can_run {
2026   my $command_name = shift;
2027   open(my $which, "-|", "which", $command_name);
2028   while (<$which>) { }
2029   close($which);
2030   return ($? == 0);
2031 }
2032
2033 sub shell_or_die
2034 {
2035   my $exitcode = shift;
2036
2037   if ($ENV{"DEBUG"}) {
2038     print STDERR "@_\n";
2039   }
2040   if (system (@_) != 0) {
2041     my $err = $!;
2042     my $code = $?;
2043     my $exitstatus = sprintf("exit %d signal %d", $code >> 8, $code & 0x7f);
2044     open STDERR, ">&STDERR_ORIG";
2045     system ("cat $destdir.log >&2");
2046     warn "@_ failed ($err): $exitstatus";
2047     if (defined($exitcode)) {
2048       exit $exitcode;
2049     }
2050     else {
2051       exit (($code >> 8) || 1);
2052     }
2053   }
2054 }
2055
2056 __DATA__