X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/c436da861845723ab18f56dba96052cf2569e539..5be63dcb589e10fbfc11fdd85dcb382708852baa:/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 0b0553cb84..4810768ded 100755
--- a/sdk/cli/bin/arv-run-pipeline-instance
+++ b/sdk/cli/bin/arv-run-pipeline-instance
@@ -187,7 +187,9 @@ if $options[:instance]
     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."
+  puts "error: you must supply a --template or --instance."
+  p.educate
+  abort
 end
 
 if $options[:run_here] == $options[:submit]
@@ -226,10 +228,10 @@ class PipelineInstance
                              :parameters => {
                                :uuid => uuid
                              },
-                             :body => {
-                               :api_token => ENV['ARVADOS_API_TOKEN']
-                             },
-                             :authenticated => false)
+                             :authenticated => false,
+                             :headers => {
+                               authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN']
+                             })
     j = JSON.parse result.body, :symbolize_names => true
     unless j.is_a? Hash and j[:uuid]
       debuglog "Failed to get pipeline_instance: #{j[:errors] rescue nil}", 0
@@ -242,10 +244,12 @@ class PipelineInstance
   def self.create(attributes)
     result = $client.execute(:api_method => $arvados.pipeline_instances.create,
                              :body => {
-                               :api_token => ENV['ARVADOS_API_TOKEN'],
                                :pipeline_instance => attributes
                              },
-                             :authenticated => false)
+                             :authenticated => false,
+                             :headers => {
+                               authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN']
+                             })
     j = JSON.parse result.body, :symbolize_names => true
     unless j.is_a? Hash and j[:uuid]
       abort "Failed to create pipeline_instance: #{j[:errors] rescue nil} #{j.inspect}"
@@ -259,10 +263,12 @@ class PipelineInstance
                                :uuid => @pi[:uuid]
                              },
                              :body => {
-                               :api_token => ENV['ARVADOS_API_TOKEN'],
                                :pipeline_instance => @attributes_to_update.to_json
                              },
-                             :authenticated => false)
+                             :authenticated => false,
+                             :headers => {
+                               authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN']
+                             })
     j = JSON.parse result.body, :symbolize_names => true
     unless j.is_a? Hash and j[:uuid]
       debuglog "Failed to save pipeline_instance: #{j[:errors] rescue nil}", 0
@@ -291,20 +297,24 @@ class JobCache
     @cache ||= {}
     result = $client.execute(:api_method => $arvados.jobs.get,
                              :parameters => {
-                               :api_token => ENV['ARVADOS_API_TOKEN'],
                                :uuid => uuid
                              },
-                             :authenticated => false)
+                             :authenticated => false,
+                             :headers => {
+                               authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN']
+                             })
     @cache[uuid] = JSON.parse result.body, :symbolize_names => true
   end
   def self.where(conditions)
     result = $client.execute(:api_method => $arvados.jobs.list,
                              :parameters => {
-                               :api_token => ENV['ARVADOS_API_TOKEN'],
                                :limit => 10000,
                                :where => conditions.to_json
                              },
-                             :authenticated => false)
+                             :authenticated => false,
+                             :headers => {
+                               authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN']
+                             })
     list = JSON.parse result.body, :symbolize_names => true
     if list and list[:items].is_a? Array
       list[:items]
@@ -315,11 +325,13 @@ class JobCache
   def self.create(job, create_params)
     @cache ||= {}
     result = $client.execute(:api_method => $arvados.jobs.create,
-                             :parameters => {
-                               :api_token => ENV['ARVADOS_API_TOKEN'],
+                             :body => {
                                :job => job.to_json
                              }.merge(create_params),
-                             :authenticated => false)
+                             :authenticated => false,
+                             :headers => {
+                               authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN']
+                             })
     j = JSON.parse result.body, :symbolize_names => true
     if j.is_a? Hash and j[:uuid]
       @cache[j[:uuid]] = j
@@ -348,10 +360,12 @@ class WhRunPipelineInstance
     else
       result = $client.execute(:api_method => $arvados.pipeline_templates.get,
                                :parameters => {
-                                 :api_token => ENV['ARVADOS_API_TOKEN'],
                                  :uuid => template
                                },
-                               :authenticated => false)
+                               :authenticated => false,
+                               :headers => {
+                                 authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN']
+                               })
       @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}"
@@ -413,19 +427,30 @@ class WhRunPipelineInstance
   end
 
   def setup_instance
-    @instance ||= PipelineInstance.
-      create(:components => @components,
+    if $options[:submit]
+      @instance ||= PipelineInstance.
+        create(:components => @components,
+              :pipeline_template_uuid => @template[:uuid],
+              :state => 'New')
+    else
+      @instance ||= PipelineInstance.
+        create(:components => @components,
              :pipeline_template_uuid => @template[:uuid],
-             :active => true)
+             :state => 'RunningOnClient')
+    end
     self
   end
 
   def run
     moretodo = true
+    interrupted = false
+
+    job_creation_failed = 0
     while moretodo
       moretodo = false
       @components.each do |cname, c|
         job = nil
+        owner_uuid = @instance[:owner_uuid]
         # Is the job satisfying this component already known to be
         # finished? (Already meaning "before we query API server about
         # the job's current state")
@@ -444,6 +469,7 @@ class WhRunPipelineInstance
             :repository => c[:repository],
             :nondeterministic => c[:nondeterministic],
             :output_is_persistent => c[:output_is_persistent] || false,
+            :owner_uuid => owner_uuid,
             # TODO: Delete the following three attributes when
             # supporting pre-20140418 API servers is no longer
             # important. New API servers take these as flags that
@@ -462,7 +488,8 @@ class WhRunPipelineInstance
             debuglog "component #{cname} new job #{job[:uuid]}"
             c[:job] = job
           else
-            debuglog "component #{cname} new job failed"
+            debuglog "component #{cname} new job failed", 0
+            job_creation_failed += 1
           end
         end
 
@@ -516,7 +543,8 @@ class WhRunPipelineInstance
                     tail_kind: 'arvados#user',
                     tail_uuid: @my_user_uuid,
                     head_kind: 'arvados#collection',
-                    head_uuid: wanted
+                    head_uuid: wanted,
+                    owner_uuid: owner_uuid
                   }
                   debuglog "added link, uuid #{newlink[:uuid]}"
                 end
@@ -532,19 +560,24 @@ class WhRunPipelineInstance
         end
       end
       @instance[:components] = @components
-      @instance[:active] = moretodo
       report_status
 
       if @options[:no_wait]
         moretodo = false
       end
 
+      # If job creation fails, just give up on this pipeline instance.
+      if job_creation_failed > 0
+        moretodo = false
+      end
+
       if moretodo
         begin
           sleep 10
         rescue Interrupt
           debuglog "interrupt", 0
-          abort
+          interrupted = true
+          break
         end
       end
     end
@@ -554,7 +587,7 @@ class WhRunPipelineInstance
     failed = 0
     @components.each do |cname, c|
       if c[:job]
-        if c[:job][:finished_at]
+        if c[:job][:finished_at] or c[:job][:cancelled_at] or (c[:job][:running] == false and c[:job][:success] == false)
           ended += 1
           if c[:job][:success] == true
             succeeded += 1
@@ -565,17 +598,39 @@ class WhRunPipelineInstance
       end
     end
 
-    if ended == @components.length or failed > 0
-      @instance[:active] = false
-      @instance[:success] = (succeeded == @components.length)
+    success = (succeeded == @components.length)
+
+    # A job create call failed. Just give up.
+    if job_creation_failed > 0
+      debuglog "job creation failed - giving up on this pipeline instance", 0
+      success = false
+      failed += 1
     end
 
+    if interrupted
+     if success
+        @instance[:state] = 'Complete'
+     else
+        @instance[:state] = 'Paused'
+      end
+    else
+      if ended == @components.length or failed > 0
+        @instance[:state] = success ? 'Complete' : 'Failed'
+      end
+    end
+
+    debuglog "pipeline instance state is #{@instance[:state]}"
+
+    # set components_summary
+    components_summary = {"todo" => @components.length - ended, "done" => succeeded, "failed" => failed}
+    @instance[:components_summary] = components_summary
+
     @instance.save
   end
 
   def cleanup
-    if @instance
-      @instance[:active] = false
+    if @instance and @instance[:state] == 'RunningOnClient'
+      @instance[:state] = 'Paused'
       @instance.save
     end
   end