X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b0af0440838407be80c6208b15f969bc2942008e..1b8120689041941db937883c3132a9c6819b56c5:/sdk/cli/bin/arv-run-pipeline-instance?ds=sidebyside

diff --git a/sdk/cli/bin/arv-run-pipeline-instance b/sdk/cli/bin/arv-run-pipeline-instance
index 090be61ba4..60d5c38cf9 100755
--- a/sdk/cli/bin/arv-run-pipeline-instance
+++ b/sdk/cli/bin/arv-run-pipeline-instance
@@ -312,19 +312,19 @@ class JobCache
       []
     end
   end
-  def self.create(attributes)
+  def self.create(job, create_params)
     @cache ||= {}
     result = $client.execute(:api_method => $arvados.jobs.create,
                              :parameters => {
                                :api_token => ENV['ARVADOS_API_TOKEN'],
-                               :job => attributes.to_json
-                             },
+                               :job => job.to_json
+                             }.merge(create_params),
                              :authenticated => false)
     j = JSON.parse result.body, :symbolize_names => true
     if j.is_a? Hash and j[:uuid]
       @cache[j[:uuid]] = j
     else
-      debuglog "create job: #{j[:errors] rescue nil} with attribute #{attributes}", 0
+      debuglog "create job: #{j[:errors] rescue nil} with attributes #{job}", 0
       nil
     end
   end
@@ -416,16 +416,21 @@ class WhRunPipelineInstance
     @instance ||= PipelineInstance.
       create(:components => @components,
              :pipeline_template_uuid => @template[:uuid],
-             :active => true)
+             :state => 'RunningOnClient')
     self
   end
 
   def run
     moretodo = true
+    interrupted = false
+
     while moretodo
       moretodo = false
       @components.each do |cname, c|
         job = nil
+        # Is the job satisfying this component already known to be
+        # finished? (Already meaning "before we query API server about
+        # the job's current state")
         c_already_finished = (c[:job] &&
                               c[:job][:uuid] &&
                               !c[:job][:success].nil?)
@@ -434,15 +439,27 @@ class WhRunPipelineInstance
           # 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],
-                            :repository => c[:repository],
-                            :minimum_script_version => c[:minimum_script_version],
-                            :exclude_script_versions => c[:exclude_minimum_script_versions],
-                            :nondeterministic => c[:nondeterministic],
-                            :no_reuse => @options[:no_reuse],
-                            :output_is_persistent => c[:output_is_persistent] || false})
+          job = JobCache.create({
+            :script => c[:script],
+            :script_parameters => c[:script_parameters],
+            :script_version => c[:script_version],
+            :repository => c[:repository],
+            :nondeterministic => c[:nondeterministic],
+            :output_is_persistent => c[:output_is_persistent] || false,
+            # TODO: Delete the following three attributes when
+            # supporting pre-20140418 API servers is no longer
+            # important. New API servers take these as flags that
+            # control behavior of create, rather than job attributes.
+            :minimum_script_version => c[:minimum_script_version],
+            :exclude_script_versions => c[:exclude_minimum_script_versions],
+            :no_reuse => @options[:no_reuse] || c[:nondeterministic],
+          }, {
+            # This is the right place to put these attributes when
+            # dealing with new API servers.
+            :minimum_script_version => c[:minimum_script_version],
+            :exclude_script_versions => c[:exclude_minimum_script_versions],
+            :find_or_create => !(@options[:no_reuse] || c[:nondeterministic]),
+          })
           if job
             debuglog "component #{cname} new job #{job[:uuid]}"
             c[:job] = job
@@ -455,7 +472,7 @@ class WhRunPipelineInstance
           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])            
+            c[:job] = JobCache.get(c[:job][:uuid])
           end
 
           if c[:job][:success]
@@ -471,10 +488,12 @@ class WhRunPipelineInstance
               end
             end
             unless c_already_finished
+              # This is my first time discovering that the job
+              # succeeded. (At the top of this loop, I was still
+              # waiting for it to finish.)
               if c[:output_is_persistent]
-                # This is my first time discovering that the job
-                # succeeded. I need to make sure a resources/wants
-                # link is in place to protect the output from garbage
+                # I need to make sure a resources/wants link is in
+                # place to protect the output from garbage
                 # collection. (Normally Crunch does this for me, but
                 # here I might be reusing the output of someone else's
                 # job and I need to make sure it's understood that the
@@ -527,7 +546,9 @@ class WhRunPipelineInstance
           sleep 10
         rescue Interrupt
           debuglog "interrupt", 0
-          abort
+          interrupted = true
+          break
+          #abort
         end
       end
     end
@@ -547,12 +568,31 @@ class WhRunPipelineInstance
         end
       end
     end
-    
-    if ended == @components.length or failed > 0
-      @instance[:active] = false
-      @instance[:success] = (succeeded == @components.length)
+
+    success = (succeeded == @components.length)
+
+    if interrupted
+     if success
+        @instance[:active] = false
+        @instance[:success] = success
+        @instance[:state] = "Complete"
+     else
+        @instance[:active] = nil
+        @instance[:success] = nil
+        @instance[:state] = 'Paused'
+      end
+    else
+      if ended == @components.length or failed > 0
+        @instance[:active] = false
+        @instance[:success] = success
+        @instance[:state] = success ? "Complete" : "Failed"
+      end
     end
 
+    # set components_summary
+    components_summary = {"todo" => @components.length - ended, "done" => succeeded, "failed" => failed}
+    @instance[:components_summary] = components_summary
+
     @instance.save
   end