X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e80fcd1a64c60849112e6af470b620f517eeb996..392c382ddaf8ea4c0c4b6655c7f508de73274d12:/sdk/cli/bin/arv-run-pipeline-instance

diff --git a/sdk/cli/bin/arv-run-pipeline-instance b/sdk/cli/bin/arv-run-pipeline-instance
index e36ac0f031..9f1663d1a3 100755
--- a/sdk/cli/bin/arv-run-pipeline-instance
+++ b/sdk/cli/bin/arv-run-pipeline-instance
@@ -2,8 +2,8 @@
 
 # == Synopsis
 #
-#  wh-run-pipeline-instance --template pipeline-template-uuid [options] [--] [parameters]
-#  wh-run-pipeline-instance --instance pipeline-instance-uuid [options]
+#  arv-run-pipeline-instance --template pipeline-template-uuid [options] [--] [parameters]
+#  arv-run-pipeline-instance --instance pipeline-instance-uuid [options]
 #
 # Satisfy a pipeline template by finding or submitting a mapreduce job
 # for each pipeline component.
@@ -12,24 +12,22 @@
 #
 # [--template uuid] Use the specified pipeline template.
 #
+# [--template path] Load the pipeline template from the specified
+#                   local file.
+#
 # [--instance uuid] Use the specified pipeline instance.
 #
 # [-n, --dry-run] Do not start any new jobs or wait for existing jobs
 #                 to finish. Just find out whether jobs are finished,
 #                 queued, or running for each component
 #
-# [--create-instance-only] Do not try to satisfy any components. Just
+# [--submit] Do not try to satisfy any components. Just
 #                          create an instance, print its UUID to
 #                          stdout, and exit.
 #
 # [--no-wait] Make only as much progress as possible without entering
 #             a sleep/poll loop.
 #
-# [--no-reuse-finished] Do not reuse existing outputs to satisfy
-#                       pipeline components. Always submit a new job
-#                       or use an existing job which has not yet
-#                       finished.
-#
 # [--no-reuse] Do not reuse existing jobs to satisfy pipeline
 #              components. Submit a new job for every component.
 #
@@ -77,13 +75,15 @@ $arvados_api_token = ENV['ARVADOS_API_TOKEN'] or
 
 begin
   require 'rubygems'
-  require 'google/api_client'
   require 'json'
   require 'pp'
   require 'trollop'
-rescue LoadError
+  require 'google/api_client'
+rescue LoadError => l
+  puts $:
   abort <<-EOS
-#{$0}: fatal: some runtime dependencies are missing.
+#{$0}: fatal: #{l.message}
+Some runtime dependencies may be missing.
 Try: gem install pp google-api-client json trollop
   EOS
 end
@@ -128,6 +128,7 @@ end
 # this program, that is, not the pipeline component parameters).
 
 p = Trollop::Parser.new do
+  version __FILE__
   opt(:dry_run,
       "Do not start any new jobs or wait for existing jobs to finish. Just find out whether jobs are finished, queued, or running for each component.",
       :type => :boolean,
@@ -146,10 +147,6 @@ p = Trollop::Parser.new do
       "Do not wait for jobs to finish. Just look up status, submit new jobs if needed, and exit.",
       :short => :none,
       :type => :boolean)
-  opt(:no_reuse_finished,
-      "Do not reuse existing outputs to satisfy pipeline components. Always submit a new job or use an existing job which has not yet finished.",
-      :short => :none,
-      :type => :boolean)
   opt(:no_reuse,
       "Do not reuse existing jobs to satisfy pipeline components. Submit a new job for every component.",
       :short => :none,
@@ -162,17 +159,21 @@ p = Trollop::Parser.new do
       :short => :none,
       :type => :integer)
   opt(:template,
-      "UUID of pipeline template.",
+      "UUID of pipeline template, or path to local pipeline template file.",
       :short => :none,
       :type => :string)
   opt(:instance,
       "UUID of pipeline instance.",
       :short => :none,
       :type => :string)
-  opt(:create_instance_only,
+  opt(:submit,
       "Do not try to satisfy any components. Just create a pipeline instance and output its UUID.",
       :short => :none,
       :type => :boolean)
+  opt(:run_here,
+      "Manage the pipeline in process.",
+      :short => :none,
+      :type => :boolean)
   stop_on [:'--']
 end
 $options = Trollop::with_standard_exception_handling p do
@@ -181,13 +182,33 @@ end
 $debuglevel = $options[:debug_level] || ($options[:debug] && 1) || 0
 
 if $options[:instance]
-  if $options[:template] or $options[:create_instance_only]
-    abort "#{$0}: syntax error: --instance cannot be combined with --template or --create-instance-only."
+  if $options[:template] or $options[:submit]
+    abort "#{$0}: syntax error: --instance cannot be combined with --template or --submit."
   end
 elsif not $options[:template]
   abort "#{$0}: syntax error: you must supply a --template or --instance."
 end
 
+if $options[:run_here] == $options[:submit]
+  abort "#{$0}: syntax error: you must supply either --run-here or --submit."
+end
+
+# Suppress SSL certificate checks if ARVADOS_API_HOST_INSECURE
+
+module Kernel
+  def suppress_warnings
+    original_verbosity = $VERBOSE
+    $VERBOSE = nil
+    result = yield
+    $VERBOSE = original_verbosity
+    return result
+  end
+end
+
+if ENV['ARVADOS_API_HOST_INSECURE']
+  suppress_warnings { OpenSSL::SSL::VERIFY_PEER = OpenSSL::SSL::VERIFY_NONE }
+end
+
 # Set up the API client.
 
 $client ||= Google::APIClient.
@@ -201,9 +222,11 @@ class PipelineInstance
   def self.find(uuid)
     result = $client.execute(:api_method => $arvados.pipeline_instances.get,
                              :parameters => {
-                               :api_token => ENV['ARVADOS_API_TOKEN'],
                                :uuid => uuid
                              },
+                             :body => {
+                               :api_token => ENV['ARVADOS_API_TOKEN']
+                             },
                              :authenticated => false)
     j = JSON.parse result.body, :symbolize_names => true
     unless j.is_a? Hash and j[:uuid]
@@ -216,9 +239,9 @@ class PipelineInstance
   end
   def self.create(attributes)
     result = $client.execute(:api_method => $arvados.pipeline_instances.create,
-                             :parameters => {
+                             :body => {
                                :api_token => ENV['ARVADOS_API_TOKEN'],
-                               :pipeline_instance => attributes.to_json
+                               :pipeline_instance => attributes
                              },
                              :authenticated => false)
     j = JSON.parse result.body, :symbolize_names => true
@@ -231,8 +254,10 @@ class PipelineInstance
   def save
     result = $client.execute(:api_method => $arvados.pipeline_instances.update,
                              :parameters => {
+                               :uuid => @pi[:uuid]
+                             },
+                             :body => {
                                :api_token => ENV['ARVADOS_API_TOKEN'],
-                               :uuid => @pi[:uuid],
                                :pipeline_instance => @attributes_to_update.to_json
                              },
                              :authenticated => false)
@@ -297,7 +322,7 @@ class JobCache
     if j.is_a? Hash and j[:uuid]
       @cache[j[:uuid]] = j
     else
-      debuglog "create job: #{j[:errors] rescue nil}", 0
+      debuglog "create job: #{j[:errors] rescue nil} with attribute #{attributes}", 0
       nil
     end
   end
@@ -310,16 +335,25 @@ class WhRunPipelineInstance
     @options = _options
   end
 
-  def fetch_template(template_uuid)
-    result = $client.execute(:api_method => $arvados.pipeline_templates.get,
-                             :parameters => {
-                               :api_token => ENV['ARVADOS_API_TOKEN'],
-                               :uuid => template_uuid
-                             },
-                             :authenticated => false)
-    @template = JSON.parse result.body, :symbolize_names => true
-    if !@template[:uuid]
-      abort "#{$0}: fatal: failed to retrieve pipeline template #{template_uuid} #{@template[:errors].inspect rescue nil}"
+  def fetch_template(template)
+    if template.match /[^-0-9a-z]/
+      # Doesn't look like a uuid -- use it as a filename.
+      @template = JSON.parse File.read(template), :symbolize_names => true
+      if !@template[:components]
+        abort ("#{$0}: Template loaded from #{template} " +
+               "does not have a \"components\" key")
+      end
+    else
+      result = $client.execute(:api_method => $arvados.pipeline_templates.get,
+                               :parameters => {
+                                 :api_token => ENV['ARVADOS_API_TOKEN'],
+                                 :uuid => template
+                               },
+                               :authenticated => false)
+      @template = JSON.parse result.body, :symbolize_names => true
+      if !@template[:uuid]
+        abort "#{$0}: fatal: failed to retrieve pipeline template #{template} #{@template[:errors].inspect rescue nil}"
+      end
     end
     self
   end
@@ -390,77 +424,34 @@ class WhRunPipelineInstance
       moretodo = false
       @components.each do |cname, c|
         job = nil
-        if !c[:job] and
-            c[:script_parameters].select { |pname, p| p.is_a? Hash }.empty?
-          # Job is fully specified (all parameter values are present) but
-          # no particular job has been found.
-
-          debuglog "component #{cname} ready to satisfy."
-
-          c.delete :wait
-          second_place_job = nil # satisfies component, but not finished yet
-
-          (@options[:no_reuse] ? [] : JobCache.
-           where(script: c[:script],
-                 script_parameters: c[:script_parameters],
-                 script_version_descends_from: c[:script_version_descends_from])
-           ).each do |candidate_job|
-            candidate_params_downcase = Hash[candidate_job[:script_parameters].
-                                             map { |k,v| [k.downcase,v] }]
-            c_params_downcase = Hash[c[:script_parameters].
-                                     map { |k,v| [k.downcase,v] }]
-
-            debuglog "component #{cname} considering job #{candidate_job[:uuid]} version #{candidate_job[:script_version]} parameters #{candidate_params_downcase.inspect}", 3
 
-            unless candidate_params_downcase == c_params_downcase
-              next
-            end
-
-            unless candidate_job[:success] || candidate_job[:running] ||
-                (!candidate_job[:started_at] && !candidate_job[:cancelled_at])
-              debuglog "component #{cname} would be satisfied by job #{candidate_job[:uuid]} if it were running or successful.", 2
-              next
-            end
-
-            if candidate_job[:success]
-              unless @options[:no_reuse_finished]
-                job = candidate_job
-                debuglog "component #{cname} satisfied by job #{job[:uuid]} version #{job[:script_version]}"
-                c[:job] = job
-              end
-            else
-              second_place_job ||= candidate_job
-            end
-            break
-          end
-          if not c[:job] and second_place_job
-            job = second_place_job
-            debuglog "component #{cname} satisfied by job #{job[:uuid]} version #{job[:script_version]}"
+        if !c[:job] and
+            c[:script_parameters].select { |pname, p| p.is_a? Hash and p[:output_of]}.empty?
+          # No job yet associated with this component and is component inputs
+          # are fully specified (any output_of script_parameters are resolved
+          # to real value)
+          job = JobCache.create({:script => c[:script],
+                            :script_parameters => c[:script_parameters],
+                            :script_version => c[:script_version],
+                            :minimum_script_version => c[:minimum_script_version],
+                            :exclude_script_versions => c[:exclude_minimum_script_versions],
+                            :nondeterministic => c[:nondeterministic],
+                            :no_reuse => @options[:no_reuse]})
+          if job
+            debuglog "component #{cname} new job #{job[:uuid]}"
             c[:job] = job
+          else
+            debuglog "component #{cname} new job failed"
           end
-          if not c[:job]
-            debuglog "component #{cname} not satisfied by any existing job."
-            if !@options[:dry_run]
-              debuglog "component #{cname} new job."
-              job = JobCache.create(:script => c[:script],
-                                    :script_parameters => c[:script_parameters],
-                                    :resource_limits => c[:resource_limits] || {},
-                                    :script_version => c[:script_version] || 'master')
-              if job
-                debuglog "component #{cname} new job #{job[:uuid]}"
-                c[:job] = job
-              else
-                debuglog "component #{cname} new job failed"
-              end
-            end
-          end
-        else
-          c[:wait] = true
         end
+
         if c[:job] and c[:job][:uuid]
-          if not c[:job][:finished_at] and not c[:job][:cancelled_at]
-            c[:job] = JobCache.get(c[:job][:uuid])
+          if (c[:job][:running] or
+              not (c[:job][:finished_at] or c[:job][:cancelled_at]))
+            # Job is running so update copy of job record
+            c[:job] = JobCache.get(c[:job][:uuid])            
           end
+
           if c[:job][:success]
             # Populate script_parameters of other components waiting for
             # this job
@@ -469,12 +460,14 @@ class WhRunPipelineInstance
                 if p.is_a? Hash and p[:output_of] == cname.to_s
                   debuglog "parameter #{c2name}::#{pname} == #{c[:job][:output]}"
                   c2[:script_parameters][pname] = c[:job][:output]
+                  moretodo = true
                 end
               end
             end
           elsif c[:job][:running] ||
               (!c[:job][:started_at] && !c[:job][:cancelled_at])
-            moretodo ||= !@options[:no_wait]
+            # Job is still running
+            moretodo = true
           elsif c[:job][:cancelled_at]
             debuglog "component #{cname} job #{c[:job][:uuid]} cancelled."
           end
@@ -483,6 +476,11 @@ class WhRunPipelineInstance
       @instance[:components] = @components
       @instance[:active] = moretodo
       report_status
+
+      if @options[:no_wait]
+        moretodo = false
+      end
+
       if moretodo
         begin
           sleep 10
@@ -492,7 +490,28 @@ class WhRunPipelineInstance
         end
       end
     end
-    @instance[:success] = @components.reject { |cname,c| c[:job] and c[:job][:success] }.empty?
+
+    ended = 0
+    succeeded = 0
+    failed = 0
+    @components.each do |cname, c|
+      if c[:job]
+        if c[:job][:finished_at]
+          ended += 1
+          if c[:job][:success] == true
+            succeeded += 1
+          elsif c[:job][:success] == false
+            failed += 1
+          end
+        end
+      end
+    end
+    
+    if ended == @components.length or failed > 0
+      @instance[:active] = false
+      @instance[:success] = (succeeded == @components.length)
+    end
+
     @instance.save
   end
 
@@ -520,6 +539,7 @@ class WhRunPipelineInstance
 
     if @options[:status_text] != '/dev/null'
       File.open(@options[:status_text], 'w') do |f|
+        f.puts ""
         f.puts "#{Time.now} -- pipeline_instance #{@instance[:uuid]}"
         namewidth = @components.collect { |cname, c| cname.size }.max
         @components.each do |cname, c|
@@ -554,7 +574,7 @@ begin
   end
   runner.apply_parameters(p.leftovers)
   runner.setup_instance
-  if $options[:create_instance_only]
+  if $options[:submit]
     runner.instance.save
     puts runner.instance[:uuid]
   else