From 1a03dd7a7d10a4843848ff60957c03110a09df43 Mon Sep 17 00:00:00 2001
From: Tom Clegg <tom@clinicalfuture.com>
Date: Tue, 18 Jun 2013 17:22:04 -0400
Subject: [PATCH] port crunch dispatcher from whjobmanager to crunch-job

---
 doc/install/install-crunch-dispatch.textile   |  65 +++++++++
 doc/install/install-workbench-app.md          |   2 +-
 sdk/perl/lib/Arvados.pm                       |   2 +
 sdk/perl/lib/Arvados/Request.pm               |   3 +-
 sdk/perl/lib/Arvados/ResourceMethod.pm        |   4 +-
 .../environments/development.rb.example       |   2 +-
 .../api/config/environments/production.rb     |   2 +-
 services/api/config/environments/test.rb      |   2 +-
 .../{dispatch_jobs.rb => crunch-dispatch.rb}  | 135 +++++-------------
 services/crunch/crunch-job                    | 107 ++++----------
 10 files changed, 139 insertions(+), 185 deletions(-)
 create mode 100644 doc/install/install-crunch-dispatch.textile
 rename services/api/script/{dispatch_jobs.rb => crunch-dispatch.rb} (56%)

diff --git a/doc/install/install-crunch-dispatch.textile b/doc/install/install-crunch-dispatch.textile
new file mode 100644
index 0000000000..a1af5e3025
--- /dev/null
+++ b/doc/install/install-crunch-dispatch.textile
@@ -0,0 +1,65 @@
+---
+layout: default
+navsection: installguide
+title: Install the Crunch dispatcher
+navorder: 3
+---
+
+{% include alert-stub.html %}
+
+h1. Crunch setup
+
+h4. Perl SDK dependencies
+
+* @apt-get install libjson-perl libwww-perl libio-socket-ssl-perl libipc-system-simple-perl@
+
+h4. Repositories
+
+Crunch scripts must be in git repositories in @/var/cache/git/*/.git@ (or whatever is configured in @services/api/config/environments/production.rb@).
+
+h4. Importing commits
+
+@services/api/script/import_commits.rb production@ must run periodically. Example @/var/service/arvados_import_commits/run@ script for daemontools or runit:
+
+<pre>
+#!/bin/sh
+set -e
+while sleep 60
+do
+  cd /path/to/arvados/services/api
+  setuidgid www-data env RAILS_ENV=production bundle exec ./script/import_commits.rb
+done
+</pre>
+
+Once you have imported some commits, you should be able to create a new job:
+
+<pre>
+read newjob <<EOF; arv job create --job "$newjob"
+{"script_parameters":{"input":"f815ec01d5d2f11cb12874ab2ed50daa"},
+ "script_version":"master",
+ "script":"hash"}
+EOF
+</pre>
+
+Without getting this error:
+
+<pre>
+ArgumentError: Specified script_version does not resolve to a commit
+</pre>
+
+h4. Running jobs
+
+* @services/api/script/crunch-dispatch.rb@ must be running.
+* @crunch-dispatch.rb@ needs @services/crunch/crunch-job@ in its @PATH@.
+* @crunch-job@ needs @sdk/perl/lib@ and @warehouse-apps/libwarehouse-perl/lib@ in its @PERLLIB@
+
+Example @/var/service/arvados_crunch_dispatch/run@ script:
+
+<pre>
+#!/bin/sh
+set -e
+PATH=$PATH:/path/to/arvados/services/crunch
+export PERLLIB=/path/to/arvados/sdk/perl/lib:/path/to/warehouse-apps/libwarehouse-perl/lib
+cd /path/to/arvados/services/api
+RAILS_ENV=production bundle exec ./script/crunch-dispatch.rb
+</pre>
diff --git a/doc/install/install-workbench-app.md b/doc/install/install-workbench-app.md
index 8f12ce2e29..468e161ef3 100644
--- a/doc/install/install-workbench-app.md
+++ b/doc/install/install-workbench-app.md
@@ -2,7 +2,7 @@
 layout: default
 navsection: installguide
 title: Install the Arvados workbench application
-navorder: 1
+navorder: 2
 ---
 
 {% include alert-stub.html %}
diff --git a/sdk/perl/lib/Arvados.pm b/sdk/perl/lib/Arvados.pm
index 18800290a7..97a44c86dd 100644
--- a/sdk/perl/lib/Arvados.pm
+++ b/sdk/perl/lib/Arvados.pm
@@ -67,6 +67,8 @@ use Arvados::ResourceProxy;
 use Arvados::ResourceProxyList;
 use Arvados::Request;
 
+$Arvados::VERSION = 0.1;
+
 sub new
 {
     my $class = shift;
diff --git a/sdk/perl/lib/Arvados/Request.pm b/sdk/perl/lib/Arvados/Request.pm
index 4902b75c37..0faed28d1a 100644
--- a/sdk/perl/lib/Arvados/Request.pm
+++ b/sdk/perl/lib/Arvados/Request.pm
@@ -18,7 +18,7 @@ sub _init
 {
     my $self = shift;
     $self->{'ua'} = new LWP::UserAgent(@_);
-    $self->{'ua'}->agent ("libarvados-perl/".$Warehouse::VERSION);
+    $self->{'ua'}->agent ("libarvados-perl/".$Arvados::VERSION);
     $self;
 }
 
@@ -35,6 +35,7 @@ sub process_request
     $req{$self->{'method'}} = $self->{'uri'};
     $self->{'req'} = new HTTP::Request (%req);
     $self->{'req'}->header('Authorization' => ('OAuth2 ' . $self->{'authToken'})) if $self->{'authToken'};
+    $self->{'req'}->header('Accept' => 'application/json');
     my %content;
     my ($p, $v);
     while (($p, $v) = each %{$self->{'queryParams'}}) {
diff --git a/sdk/perl/lib/Arvados/ResourceMethod.pm b/sdk/perl/lib/Arvados/ResourceMethod.pm
index 76c1ac8691..fd577752b2 100644
--- a/sdk/perl/lib/Arvados/ResourceMethod.pm
+++ b/sdk/perl/lib/Arvados/ResourceMethod.pm
@@ -80,7 +80,9 @@ sub execute
         croak("Unsupported parameter(s) passed to API call /$path: \"" . join('", "', keys %extra_params) . '"');
     }
     my $r = $self->{'resourceAccessor'}->{'api'}->new_request;
-    $r->set_uri($self->{'resourceAccessor'}->{'api'}->{'discoveryDocument'}->{'baseUrl'} . "/" . $path);
+    my $base_uri = $self->{'resourceAccessor'}->{'api'}->{'discoveryDocument'}->{'baseUrl'};
+    $base_uri =~ s:/$::;
+    $r->set_uri($base_uri . "/" . $path);
     $r->set_method($method->{'httpMethod'});
     $r->set_auth_token($self->{'resourceAccessor'}->{'api'}->{'authToken'});
     $r->set_query_params(\%body_params) if %body_params;
diff --git a/services/api/config/environments/development.rb.example b/services/api/config/environments/development.rb.example
index ba8d2d43c4..24e9ccaa5c 100644
--- a/services/api/config/environments/development.rb.example
+++ b/services/api/config/environments/development.rb.example
@@ -32,7 +32,7 @@ Server::Application.configure do
 
   config.git_repositories_dir = '/var/cache/git'
 
-  config.whjobmanager_wrapper = :none
+  config.crunch_job_wrapper = :none
 
   # config.dnsmasq_conf_dir = '/etc/dnsmasq.d'
 
diff --git a/services/api/config/environments/production.rb b/services/api/config/environments/production.rb
index a3c154d364..affa94ec69 100644
--- a/services/api/config/environments/production.rb
+++ b/services/api/config/environments/production.rb
@@ -60,7 +60,7 @@ Server::Application.configure do
 
   config.git_repositories_dir = '/var/cache/git'
 
-  config.whjobmanager_wrapper = :slurm_immediate
+  config.crunch_job_wrapper = :slurm_immediate
 
   # config.dnsmasq_conf_dir = '/etc/dnsmasq.d'
 
diff --git a/services/api/config/environments/test.rb b/services/api/config/environments/test.rb
index 9c3afdeda5..c370b96746 100644
--- a/services/api/config/environments/test.rb
+++ b/services/api/config/environments/test.rb
@@ -39,7 +39,7 @@ Server::Application.configure do
 
   config.git_repositories_dir = '/var/cache/git'
 
-  config.whjobmanager_wrapper = :slurm_immediate
+  config.crunch_job_wrapper = :slurm_immediate
 
   # config.dnsmasq_conf_dir = '/etc/dnsmasq.d'
 
diff --git a/services/api/script/dispatch_jobs.rb b/services/api/script/crunch-dispatch.rb
similarity index 56%
rename from services/api/script/dispatch_jobs.rb
rename to services/api/script/crunch-dispatch.rb
index f937747303..f749f22a77 100755
--- a/services/api/script/dispatch_jobs.rb
+++ b/services/api/script/crunch-dispatch.rb
@@ -29,7 +29,7 @@ class Dispatcher
   end
 
   def start_jobs
-    if Server::Application.config.whjobmanager_wrapper.to_s.match /^slurm/
+    if Server::Application.config.crunch_job_wrapper.to_s.match /^slurm/
       @idle_slurm_nodes = 0
       begin
         `sinfo`.
@@ -56,7 +56,7 @@ class Dispatcher
       next if !take(job)
 
       cmd_args = nil
-      case Server::Application.config.whjobmanager_wrapper
+      case Server::Application.config.crunch_job_wrapper
       when :none
         cmd_args = []
       when :slurm_immediate
@@ -67,27 +67,27 @@ class Dispatcher
                     "--job-name=#{job.uuid}",
                     "--nodes=#{min_nodes}"]
       else
-        raise "Unknown whjobmanager_wrapper: #{Server::Application.config.whjobmanager_wrapper}"
+        raise "Unknown crunch_job_wrapper: #{Server::Application.config.crunch_job_wrapper}"
       end
 
-      cmd_args << 'whjobmanager'
-      cmd_args << "id=#{job.uuid}"
-      cmd_args << "mrfunction=#{job.script}"
-      job.script_parameters.each do |k,v|
-        k = k.to_s
-        if k == 'input'
-          k = 'inputkey'
-        else
-          k = k.upcase
-        end
-        cmd_args << "#{k}=#{v}"
-      end
-      cmd_args << "revision=#{job.script_version}"
-
-      begin
-        cmd_args << "stepspernode=#{job.resource_limits['max_tasks_per_node'].to_i}"
-      rescue
-        # OK if limit is not specified. OK to ignore if not integer.
+      job_auth = ApiClientAuthorization.
+        new(user: User.where('uuid=?', job.modified_by_user).first,
+            api_client_id: 0)
+      job_auth.save
+
+      cmd_args << 'crunch-job'
+      cmd_args << '--job-api-token'
+      cmd_args << job_auth.api_token
+      cmd_args << '--job'
+      cmd_args << job.uuid
+
+      commit = Commit.where(sha1: job.script_version).first
+      if commit
+        cmd_args << '--git-dir'
+        cmd_args << File.
+          join(Rails.configuration.git_repositories_dir,
+               commit.repository_name,
+               '.git')
       end
 
       $stderr.puts "dispatch: #{cmd_args.join ' '}"
@@ -110,27 +110,21 @@ class Dispatcher
         job: job,
         stderr_buf: '',
         started: false,
-        sent_int: 0
+        sent_int: 0,
+        job_auth: job_auth
       }
       i.close
     end
   end
 
   def take(job)
-    lock_ok = false
-    ActiveRecord::Base.transaction do
-      job.reload
-      if job.is_locked_by.nil? and
-          job.update_attributes(is_locked_by: sysuser.uuid)
-        lock_ok = true
-      end
-    end
-    lock_ok
+    # no-op -- let crunch-job take care of locking.
+    true
   end
 
   def untake(job)
-    job.reload
-    job.update_attributes is_locked_by: nil
+    # no-op -- let crunch-job take care of locking.
+    true
   end
 
   def read_pipes
@@ -153,7 +147,7 @@ class Dispatcher
       if stderr_buf
         j[:stderr_buf] << stderr_buf
         if j[:stderr_buf].index "\n"
-          lines = j[:stderr_buf].lines "\n"
+          lines = j[:stderr_buf].lines("\n").to_a
           if j[:stderr_buf][-1] == "\n"
             j[:stderr_buf] = ''
           else
@@ -162,58 +156,6 @@ class Dispatcher
           lines.each do |line|
             $stderr.print "#{job_uuid} ! " unless line.index(job_uuid)
             $stderr.puts line
-            line.chomp!
-            if (re = line.match(/#{job_uuid} (\d+) (\S*) (.*)/))
-              ignorethis, whjmpid, taskid, message = re.to_a
-              if taskid == '' and message == 'start'
-                $stderr.puts "dispatch: noticed #{job_uuid} started"
-                j[:started] = true
-                ActiveRecord::Base.transaction do
-                  j[:job].reload
-                  j[:job].update_attributes running: true, started_at: Time.now
-                end
-              elsif taskid == '' and (re = message.match /^revision (\S+)$/)
-                $stderr.puts "dispatch: noticed #{job_uuid} version #{re[1]}"
-                ActiveRecord::Base.transaction do
-                  j[:job].reload
-                  j[:job].script_version = re[1]
-                  j[:job].save
-                end
-              elsif taskid == '' and (re = message.match /^outputkey (\S+)$/)
-                $stderr.puts "dispatch: noticed #{job_uuid} output #{re[1]}"
-                j[:output] = re[1]
-              elsif taskid == '' and (re = message.match /^meta key is (\S+)$/)
-                $stderr.puts "dispatch: noticed #{job_uuid} log #{re[1]}"
-                j[:log] = re[1]
-                ActiveRecord::Base.transaction do
-                  j[:job].reload
-                  j[:job].update_attributes log: j[:log]
-                end
-              elsif taskid.match(/^\d+/) and (re = message.match /^failure /)
-                $stderr.puts "dispatch: noticed #{job_uuid} task fail"
-                ActiveRecord::Base.transaction do
-                  j[:job].reload
-                  j[:job].tasks_summary ||= {}
-                  j[:job].tasks_summary[:failed] ||= 0
-                  j[:job].tasks_summary[:failed] += 1
-                  j[:job].save
-                end
-              elsif (re = message.match(/^status: (\d+) done, (\d+) running, (\d+) todo/))
-                $stderr.puts "dispatch: noticed #{job_uuid} #{message}"
-                ActiveRecord::Base.transaction do
-                  j[:job].reload
-                  j[:job].tasks_summary ||= {}
-                  j[:job].tasks_summary[:done] = re[1].to_i
-                  j[:job].tasks_summary[:running] = re[2].to_i
-                  j[:job].tasks_summary[:todo] = re[3].to_i
-                  j[:job].save
-                end
-                if re[2].to_i == 0 and re[3].to_i == 0
-                  $stderr.puts "dispatch: noticed #{job_uuid} succeeded"
-                  j[:success] = true
-                end
-              end
-            end
           end
         end
       end
@@ -265,23 +207,12 @@ class Dispatcher
       $stderr.puts j_done[:stderr_buf] + "\n"
     end
 
-    j_done[:wait_thr].value          # wait the thread
+    # Wait the thread
+    j_done[:wait_thr].value
+
+    # Invalidate the per-job auth token
+    j_done[:job_auth].update_attributes expires_at: Time.now
 
-    if !j_done[:started]
-      # If the job never really started (due to a scheduling
-      # failure), just put it back in the queue
-      untake(job_done)
-      $stderr.puts "dispatch: job #{job_done.uuid} requeued"
-    else
-      # Otherwise, mark the job as finished
-      ActiveRecord::Base.transaction do
-        job_done.reload
-        job_done.log = j_done[:log]
-        job_done.output = j_done[:output]
-        job_done.success = j_done[:success]
-        job_done.assert_finished
-      end
-    end
     @running.delete job_done.uuid
   end
 
diff --git a/services/crunch/crunch-job b/services/crunch/crunch-job
index 55d4a27e4d..76ce4e4da0 100755
--- a/services/crunch/crunch-job
+++ b/services/crunch/crunch-job
@@ -10,13 +10,31 @@ crunch-job: Execute job steps, save snapshots as requested, collate output.
 Obtain job details from Arvados, run tasks on compute nodes (typically
 invoked by scheduler on controller):
 
- crunch-job --uuid x-y-z
+ crunch-job --job x-y-z
 
 Obtain job details from command line, run tasks on local machine
 (typically invoked by application or developer on VM):
 
  crunch-job --job '{"script_version":"/path/to/tree","script":"scriptname",...}'
 
+=head1 OPTIONS
+
+=over
+
+=item --force-unlock
+
+If the job is already locked, steal the lock and run it anyway.
+
+=item --git-dir
+
+Path to .git directory where the specified commit is found.
+
+=item --job-api-token
+
+Arvados API authorization token to use during the course of the job.
+
+=back
+
 =head1 RUNNING JOBS LOCALLY
 
 crunch-job's log messages appear on stderr along with the job tasks'
@@ -48,7 +66,6 @@ or unallocated). Currently this is a no-op.
 
 
 use strict;
-use DBI;
 use POSIX ':sys_wait_h';
 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
 use Arvados;
@@ -63,13 +80,21 @@ $ENV{"CRUNCH_WORK"} = $ENV{"CRUNCH_TMP"} . "/work";
 mkdir ($ENV{"CRUNCH_TMP"});
 
 my $force_unlock;
+my $git_dir;
 my $jobspec;
+my $job_api_token;
 my $resume_stash;
 GetOptions('force-unlock' => \$force_unlock,
+           'git-dir=s' => \$git_dir,
            'job=s' => \$jobspec,
+           'job-api-token=s' => \$job_api_token,
            'resume-stash=s' => \$resume_stash,
     );
 
+if (defined $job_api_token) {
+  $ENV{ARVADOS_API_TOKEN} = $job_api_token;
+}
+
 my $have_slurm = exists $ENV{SLURM_JOBID} && exists $ENV{SLURM_NODELIST};
 my $job_has_uuid = $jobspec =~ /^[-a-z\d]+$/;
 
@@ -333,8 +358,9 @@ else
 
   my $commit;
   my $treeish = $Job->{'script_version'};
-  my $repo = $ENV{'CRUNCH_DEFAULT_GIT_DIR'};
-  # Todo: let script_version specify alternate repo
+  my $repo = $git_dir || $ENV{'CRUNCH_DEFAULT_GIT_DIR'};
+  # Todo: let script_version specify repository instead of expecting
+  # parent process to figure it out.
   $ENV{"CRUNCH_SRC_URL"} = $repo;
 
   # Create/update our clone of the remote git repo
@@ -1070,34 +1096,6 @@ sub Log				# ($jobstep_id, $logmessage)
 }
 
 
-sub reconnect_database
-{
-  return if !$job_has_uuid;
-  return if ($dbh && $dbh->do ("select now()"));
-  for (1..16)
-  {
-    $dbh = DBI->connect(@$Warehouse::Server::DatabaseDSN);
-    if ($dbh) {
-      $dbh->{InactiveDestroy} = 1;
-      return;
-    }
-    warn ($DBI::errstr);
-    sleep $_;
-  }
-  croak ($DBI::errstr) if !$dbh;
-}
-
-
-sub dbh_do
-{
-  return 1 if !$job_has_uuid;
-  my $ret = $dbh->do (@_);
-  return $ret unless (!$ret && $DBI::errstr =~ /server has gone away/);
-  reconnect_database();
-  return $dbh->do (@_);
-}
-
-
 sub croak
 {
   my ($package, $file, $line) = caller;
@@ -1169,51 +1167,6 @@ sub freeze
 {
   Log (undef, "Freeze not implemented");
   return;
-
-  my $whc;                      # todo
-  Log (undef, "freeze");
-
-  my $freezer = new Warehouse::Stream (whc => $whc);
-  $freezer->clear;
-  $freezer->name (".");
-  $freezer->write_start ("state.txt");
-
-  $freezer->write_data (join ("\n",
-			      "job $Job->{uuid}",
-			      map
-			      {
-				$_ . "=" . freezequote($Job->{$_})
-			      } grep { $_ ne "id" } keys %$Job) . "\n\n");
-
-  foreach my $Jobstep (@jobstep)
-  {
-    my $str = join ("\n",
-		    map
-		    {
-		      $_ . "=" . freezequote ($Jobstep->{$_})
-		    } grep {
-                      $_ !~ /^stderr|slotindex|node_fail/
-                    } keys %$Jobstep);
-    $freezer->write_data ($str."\n\n");
-  }
-  if (@jobstep_tomerge)
-  {
-    $freezer->write_data
-	("merge $jobstep_tomerge_level "
-	 . freezequote (join ("\n",
-			      map { freezequote ($_) } @jobstep_tomerge))
-	 . "\n\n");
-  }
-
-  $freezer->write_finish;
-  my $frozentokey = $freezer->as_key;
-  undef $freezer;
-  Log (undef, "frozento key is $frozentokey");
-  dbh_do ("update mrjob set frozentokey=? where id=?", undef,
-	  $frozentokey, $job_id);
-  my $kfrozentokey = $whc->store_in_keep (hash => $frozentokey, nnodes => 3);
-  Log (undef, "frozento+K key is $kfrozentokey");
-  return $frozentokey;
 }
 
 
-- 
2.30.2