--- /dev/null
+---
+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>
layout: default
navsection: installguide
title: Install the Arvados workbench application
-navorder: 1
+navorder: 2
---
{% include alert-stub.html %}
use Arvados::ResourceProxyList;
use Arvados::Request;
+$Arvados::VERSION = 0.1;
+
sub new
{
my $class = shift;
{
my $self = shift;
$self->{'ua'} = new LWP::UserAgent(@_);
- $self->{'ua'}->agent ("libarvados-perl/".$Warehouse::VERSION);
+ $self->{'ua'}->agent ("libarvados-perl/".$Arvados::VERSION);
$self;
}
$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'}}) {
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;
config.git_repositories_dir = '/var/cache/git'
- config.whjobmanager_wrapper = :none
+ config.crunch_job_wrapper = :none
# config.dnsmasq_conf_dir = '/etc/dnsmasq.d'
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'
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'
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`.
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
"--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 ' '}"
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
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
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
$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
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'
use strict;
-use DBI;
use POSIX ':sys_wait_h';
use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
use Arvados;
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]+$/;
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
}
-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;
{
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;
}
projects / arvados.git / commitdiff