5416: Merge branch 'master' into 5416-arv-git-httpd
[arvados.git] / sdk / cli / bin / arv-run-pipeline-instance
1 #!/usr/bin/env ruby
2
3 # == Synopsis
4 #
5 #  arv-run-pipeline-instance --template pipeline-template-uuid [options] [--] [parameters]
6 #  arv-run-pipeline-instance --instance pipeline-instance-uuid [options]
7 #
8 # Satisfy a pipeline template by finding or submitting a mapreduce job
9 # for each pipeline component.
10 #
11 # == Options
12 #
13 # [--template uuid] Use the specified pipeline template.
14 #
15 # [--template path] Load the pipeline template from the specified
16 #                   local file.
17 #
18 # [--instance uuid] Use the specified pipeline instance.
19 #
20 # [-n, --dry-run] Do not start any new jobs or wait for existing jobs
21 #                 to finish. Just find out whether jobs are finished,
22 #                 queued, or running for each component
23 #
24 # [--submit] Do not try to satisfy any components. Just
25 #                          create an instance, print its UUID to
26 #                          stdout, and exit.
27 #
28 # [--no-wait] Make only as much progress as possible without entering
29 #             a sleep/poll loop.
30 #
31 # [--no-reuse] Do not reuse existing jobs to satisfy pipeline
32 #              components. Submit a new job for every component.
33 #
34 # [--debug] Print extra debugging information on stderr.
35 #
36 # [--debug-level N] Increase amount of debugging information. Default
37 #                   1, possible range 0..3.
38 #
39 # [--status-text path] Print plain text status report to a file or
40 #                      fifo. Default: /dev/stdout
41 #
42 # [--status-json path] Print JSON status report to a file or
43 #                      fifo. Default: /dev/null
44 #
45 # [--description] Description for the pipeline instance.
46 #
47 # == Parameters
48 #
49 # [param_name=param_value]
50 #
51 # [param_name param_value] Set (or override) the default value for
52 #                          every parameter with the given name.
53 #
54 # [component_name::param_name=param_value]
55 # [component_name::param_name param_value]
56 # [--component_name::param_name=param_value]
57 # [--component_name::param_name param_value] Set the value of a
58 #                                            parameter for a single
59 #                                            component.
60 #
61 class WhRunPipelineInstance
62 end
63
64 if RUBY_VERSION < '1.9.3' then
65   abort <<-EOS
66 #{$0.gsub(/^\.\//,'')} requires Ruby version 1.9.3 or higher.
67   EOS
68 end
69
70 begin
71   require 'arvados'
72   require 'rubygems'
73   require 'json'
74   require 'pp'
75   require 'trollop'
76   require 'google/api_client'
77 rescue LoadError => l
78   puts $:
79   abort <<-EOS
80 #{$0}: fatal: #{l.message}
81 Some runtime dependencies may be missing.
82 Try: gem install arvados pp google-api-client json trollop
83   EOS
84 end
85
86 def debuglog(message, verbosity=1)
87   $stderr.puts "#{File.split($0).last} #{$$}: #{message}" if $debuglevel >= verbosity
88 end
89
90 # Parse command line options (the kind that control the behavior of
91 # this program, that is, not the pipeline component parameters).
92
93 p = Trollop::Parser.new do
94   version __FILE__
95   opt(:dry_run,
96       "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.",
97       :type => :boolean,
98       :short => :n)
99   opt(:status_text,
100       "Store plain text status in given file.",
101       :short => :none,
102       :type => :string,
103       :default => '/dev/stdout')
104   opt(:status_json,
105       "Store json-formatted pipeline in given file.",
106       :short => :none,
107       :type => :string,
108       :default => '/dev/null')
109   opt(:no_wait,
110       "Do not wait for jobs to finish. Just look up status, submit new jobs if needed, and exit.",
111       :short => :none,
112       :type => :boolean)
113   opt(:no_reuse,
114       "Do not reuse existing jobs to satisfy pipeline components. Submit a new job for every component.",
115       :short => :none,
116       :type => :boolean)
117   opt(:debug,
118       "Print extra debugging information on stderr.",
119       :type => :boolean)
120   opt(:debug_level,
121       "Set debug verbosity level.",
122       :short => :none,
123       :type => :integer)
124   opt(:template,
125       "UUID of pipeline template, or path to local pipeline template file.",
126       :short => :none,
127       :type => :string)
128   opt(:instance,
129       "UUID of pipeline instance.",
130       :short => :none,
131       :type => :string)
132   opt(:submit,
133       "Submit the pipeline instance to the server, and exit. Let the Crunch dispatch service satisfy the components by finding/running jobs.",
134       :short => :none,
135       :type => :boolean)
136   opt(:run_pipeline_here,
137       "Manage the pipeline instance in-process. Submit jobs to Crunch as needed. Do not exit until the pipeline finishes (or fails).",
138       :short => :none,
139       :type => :boolean)
140   opt(:run_jobs_here,
141       "Run jobs in the local terminal session instead of submitting them to Crunch. Implies --run-pipeline-here. Note: this results in a significantly different job execution environment, and some Crunch features are not supported. It can be necessary to modify a pipeline in order to make it run this way.",
142       :short => :none,
143       :type => :boolean)
144   opt(:run_here,
145       "Synonym for --run-jobs-here.",
146       :short => :none,
147       :type => :boolean)
148   opt(:description,
149       "Description for the pipeline instance.",
150       :short => :none,
151       :type => :string)
152   stop_on [:'--']
153 end
154 $options = Trollop::with_standard_exception_handling p do
155   p.parse ARGV
156 end
157 $debuglevel = $options[:debug_level] || ($options[:debug] && 1) || 0
158
159 $options[:run_jobs_here] ||= $options[:run_here] # old flag name
160 $options[:run_pipeline_here] ||= $options[:run_jobs_here] # B requires A
161
162 if $options[:instance]
163   if $options[:template] or $options[:submit]
164     abort "#{$0}: syntax error: --instance cannot be combined with --template or --submit."
165   end
166 elsif not $options[:template]
167   puts "error: you must supply a --template or --instance."
168   p.educate
169   abort
170 end
171
172 if $options[:run_pipeline_here] == $options[:submit]
173   abort "#{$0}: error: you must supply --run-pipeline-here, --run-jobs-here, or --submit."
174 end
175
176 # Set up the API client.
177
178 $arv = Arvados.new api_version: 'v1'
179 $client = $arv.client
180 $arvados = $arv.arvados_api
181
182 class PipelineInstance
183   def self.find(uuid)
184     result = $client.execute(:api_method => $arvados.pipeline_instances.get,
185                              :parameters => {
186                                :uuid => uuid
187                              },
188                              :authenticated => false,
189                              :headers => {
190                                authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN']
191                              })
192     j = JSON.parse result.body, :symbolize_names => true
193     unless j.is_a? Hash and j[:uuid]
194       debuglog "Failed to get pipeline_instance: #{j[:errors] rescue nil}", 0
195       nil
196     else
197       debuglog "Retrieved pipeline_instance #{j[:uuid]}"
198       self.new(j)
199     end
200   end
201   def self.create(attributes)
202     result = $client.execute(:api_method => $arvados.pipeline_instances.create,
203                              :body_object => {
204                                :pipeline_instance => attributes
205                              },
206                              :authenticated => false,
207                              :headers => {
208                                authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN']
209                              })
210     j = JSON.parse result.body, :symbolize_names => true
211     unless j.is_a? Hash and j[:uuid]
212       abort "\n#{Time.now} -- pipeline_template #{@template[:uuid]}\nFailed to create pipeline_instance: #{j[:errors] rescue nil} #{j.inspect}"
213     end
214     debuglog "Created pipeline instance: #{j[:uuid]}"
215     self.new(j)
216   end
217   def save
218     result = $client.execute(:api_method => $arvados.pipeline_instances.update,
219                              :parameters => {
220                                :uuid => @pi[:uuid]
221                              },
222                              :body_object => {
223                                :pipeline_instance => @attributes_to_update
224                              },
225                              :authenticated => false,
226                              :headers => {
227                                authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN']
228                              })
229     j = JSON.parse result.body, :symbolize_names => true
230     unless j.is_a? Hash and j[:uuid]
231       debuglog "Failed to save pipeline_instance: #{j[:errors] rescue nil}", 0
232       nil
233     else
234       @attributes_to_update = {}
235       @pi = j
236     end
237   end
238   def []=(x,y)
239     @attributes_to_update[x] = y
240     @pi[x] = y
241   end
242   def [](x)
243     @pi[x]
244   end
245
246   def log_stderr(msg)
247     $arv.log.create log: {
248       event_type: 'stderr',
249       object_uuid: self[:uuid],
250       owner_uuid: self[:owner_uuid],
251       properties: {"text" => msg},
252     }
253   end
254
255   protected
256   def initialize(j)
257     @attributes_to_update = {}
258     @pi = j
259   end
260 end
261
262 class JobCache
263   def self.get(uuid)
264     @cache ||= {}
265     result = $client.execute(:api_method => $arvados.jobs.get,
266                              :parameters => {
267                                :uuid => uuid
268                              },
269                              :authenticated => false,
270                              :headers => {
271                                authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN']
272                              })
273     @cache[uuid] = JSON.parse result.body, :symbolize_names => true
274   end
275   def self.where(conditions)
276     result = $client.execute(:api_method => $arvados.jobs.list,
277                              :parameters => {
278                                :limit => 10000,
279                                :where => conditions.to_json
280                              },
281                              :authenticated => false,
282                              :headers => {
283                                authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN']
284                              })
285     list = JSON.parse result.body, :symbolize_names => true
286     if list and list[:items].is_a? Array
287       list[:items]
288     else
289       []
290     end
291   end
292   def self.create(pipeline, component, job, create_params)
293     @cache ||= {}
294
295     body = {job: no_nil_values(job)}.merge(no_nil_values(create_params))
296
297     result = $client.execute(:api_method => $arvados.jobs.create,
298                              :body_object => body,
299                              :authenticated => false,
300                              :headers => {
301                                authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN']
302                              })
303     j = JSON.parse result.body, :symbolize_names => true
304     if j.is_a? Hash and j[:uuid]
305       @cache[j[:uuid]] = j
306     else
307       debuglog "create job: #{j[:errors] rescue nil} with attributes #{body}", 0
308
309       msg = ""
310       j[:errors].each do |err|
311         msg += "Error creating job for component #{component}: #{err}\n"
312       end
313       msg += "Job submission was: #{body.to_json}"
314
315       pipeline.log_stderr(msg)
316       nil
317     end
318   end
319
320   protected
321
322   def self.no_nil_values(hash)
323     hash.reject { |key, value| value.nil? }
324   end
325 end
326
327 class WhRunPipelineInstance
328   attr_reader :instance
329
330   def initialize(_options)
331     @options = _options
332   end
333
334   def fetch_template(template)
335     if template.match /[^-0-9a-z]/
336       # Doesn't look like a uuid -- use it as a filename.
337       @template = JSON.parse File.read(template), :symbolize_names => true
338     else
339       result = $client.execute(:api_method => $arvados.pipeline_templates.get,
340                                :parameters => {
341                                  :uuid => template
342                                },
343                                :authenticated => false,
344                                :headers => {
345                                  authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN']
346                                })
347       @template = JSON.parse result.body, :symbolize_names => true
348       if !@template[:uuid]
349         abort "#{$0}: fatal: failed to retrieve pipeline template #{template} #{@template[:errors].inspect rescue nil}"
350       end
351     end
352     self
353   end
354
355   def fetch_instance(instance_uuid)
356     @instance = PipelineInstance.find(instance_uuid)
357     @template = @instance
358     self
359   end
360
361   def apply_parameters(params_args)
362     params_args.shift if params_args[0] == '--'
363     params = {}
364     while !params_args.empty?
365       if (re = params_args[0].match /^(--)?([^-].*?)=(.+)/)
366         params[re[2]] = re[3]
367         params_args.shift
368       elsif params_args.size > 1
369         param = params_args.shift.sub /^--/, ''
370         params[param] = params_args.shift
371       else
372         abort "\n#{Time.now} -- pipeline_template #{@template[:uuid]}\nSyntax error: I do not know what to do with arg \"#{params_args[0]}\""
373       end
374     end
375
376     if not @template[:components].is_a?(Hash)
377       abort "\n#{Time.now} -- pipeline_template #{@template[:uuid]}\nSyntax error: Template missing \"components\" hash"
378     end
379     @components = @template[:components].dup
380
381     bad_components = @components.each_pair.select do |cname, cspec|
382       not cspec.is_a?(Hash)
383     end
384     if bad_components.any?
385       abort "\n#{Time.now} -- pipeline_template #{@template[:uuid]}\nSyntax error: Components not specified with hashes: #{bad_components.map(&:first).join(', ')}"
386     end
387
388     bad_components = @components.each_pair.select do |cname, cspec|
389       not cspec[:script_parameters].is_a?(Hash)
390     end
391     if bad_components.any?
392       abort "\n#{Time.now} -- pipeline_template #{@template[:uuid]}\nSyntax error: Components missing \"script_parameters\" hashes: #{bad_components.map(&:first).join(', ')}"
393     end
394
395     errors = []
396     @components.each do |componentname, component|
397       component[:script_parameters].each do |parametername, parameter|
398         parameter = { :value => parameter } unless parameter.is_a? Hash
399         value =
400           (params["#{componentname}::#{parametername}"] ||
401            parameter[:value] ||
402            (parameter[:output_of].nil? &&
403             (params[parametername.to_s] ||
404              parameter[:default])) ||
405            nil)
406         if value.nil? and
407             ![false,'false',0,'0'].index parameter[:required]
408           if parameter[:output_of]
409             if not @components[parameter[:output_of].intern]
410               errors << [componentname, parametername, "output_of refers to nonexistent component '#{parameter[:output_of]}'"]
411             end
412             next
413           end
414           errors << [componentname, parametername, "required parameter is missing"]
415         end
416         debuglog "parameter #{componentname}::#{parametername} == #{value}"
417
418         component[:script_parameters][parametername] =
419           parameter.dup.merge(value: value)
420       end
421     end
422     if !errors.empty?
423       abort "\n#{Time.now} -- pipeline_template #{@template[:uuid]}\nErrors:\n#{errors.collect { |c,p,e| "#{c}::#{p} - #{e}\n" }.join ""}"
424     end
425     debuglog "options=" + @options.pretty_inspect
426     self
427   end
428
429   def setup_instance
430     if @instance
431       @instance[:properties][:run_options] ||= {}
432       if @options[:no_reuse]
433         # override properties of existing instance
434         @instance[:properties][:run_options][:enable_job_reuse] = false
435       else
436         # Default to "enable reuse" if not specified. (This code path
437         # can go away when old clients go away.)
438         if @instance[:properties][:run_options][:enable_job_reuse].nil?
439           @instance[:properties][:run_options][:enable_job_reuse] = true
440         end
441       end
442     else
443       description = $options[:description]
444       description = ("Created at #{Time.now.localtime}" + (@template[:name].andand.size.andand>0 ? " using the pipeline template *#{@template[:name]}*" : "")) if !description
445       @instance = PipelineInstance.
446         create(components: @components,
447                properties: {
448                  run_options: {
449                    enable_job_reuse: !@options[:no_reuse]
450                  }
451                },
452                pipeline_template_uuid: @template[:uuid],
453                description: description,
454                state: ($options[:submit] ? 'RunningOnServer' : 'RunningOnClient'))
455     end
456     self
457   end
458
459   def run
460     moretodo = true
461     interrupted = false
462
463     if @instance[:started_at].nil?
464       @instance[:started_at] = Time.now
465     end
466
467     job_creation_failed = 0
468     while moretodo
469       moretodo = false
470       @components.each do |cname, c|
471         job = nil
472         owner_uuid = @instance[:owner_uuid]
473         # Is the job satisfying this component already known to be
474         # finished? (Already meaning "before we query API server about
475         # the job's current state")
476         c_already_finished = (c[:job] &&
477                               c[:job][:uuid] &&
478                               ["Complete", "Failed", "Cancelled"].include?(c[:job][:state]))
479         if !c[:job] and
480             c[:script_parameters].select { |pname, p| p.is_a? Hash and p[:output_of]}.empty?
481           # No job yet associated with this component and is component inputs
482           # are fully specified (any output_of script_parameters are resolved
483           # to real value)
484           my_submit_id = "instance #{@instance[:uuid]} rand #{rand(2**64).to_s(36)}"
485           job = JobCache.create(@instance, cname, {
486             :script => c[:script],
487             :script_parameters => Hash[c[:script_parameters].map do |key, spec|
488                                          [key, spec[:value]]
489                                        end],
490             :script_version => c[:script_version],
491             :repository => c[:repository],
492             :nondeterministic => c[:nondeterministic],
493             :runtime_constraints => c[:runtime_constraints],
494             :owner_uuid => owner_uuid,
495             :is_locked_by_uuid => (@options[:run_jobs_here] ? owner_uuid : nil),
496             :submit_id => my_submit_id,
497             :state => (if @options[:run_jobs_here] then "Running" else "Queued" end)
498           }, {
499             # This is the right place to put these attributes when
500             # dealing with new API servers.
501             :minimum_script_version => c[:minimum_script_version],
502             :exclude_script_versions => c[:exclude_minimum_script_versions],
503             :find_or_create => (@instance[:properties][:run_options].andand[:enable_job_reuse] &&
504                                 !c[:nondeterministic]),
505             :filters => c[:filters]
506           })
507           if job
508             debuglog "component #{cname} new job #{job[:uuid]}"
509             c[:job] = job
510             c[:run_in_process] = (@options[:run_jobs_here] and
511                                   job[:submit_id] == my_submit_id)
512           else
513             debuglog "component #{cname} new job failed", 0
514             job_creation_failed += 1
515           end
516         end
517
518         if c[:job] and c[:run_in_process] and not ["Complete", "Failed", "Cancelled"].include? c[:job][:state]
519           report_status
520           begin
521             require 'open3'
522             Open3.popen3("arv-crunch-job", "--force-unlock",
523                          "--job", c[:job][:uuid]) do |stdin, stdout, stderr, wait_thr|
524               debuglog "arv-crunch-job pid #{wait_thr.pid} started", 0
525               stdin.close
526               while true
527                 rready, wready, = IO.select([stdout, stderr], [])
528                 break if !rready[0]
529                 begin
530                   buf = rready[0].read_nonblock(2**20)
531                 rescue EOFError
532                   break
533                 end
534                 (rready[0] == stdout ? $stdout : $stderr).write(buf)
535               end
536               stdout.close
537               stderr.close
538               debuglog "arv-crunch-job pid #{wait_thr.pid} exit #{wait_thr.value.to_i}", 0
539             end
540             if not $arv.job.get(uuid: c[:job][:uuid])[:finished_at]
541               raise Exception.new("arv-crunch-job did not set finished_at.")
542             end
543           rescue Exception => e
544             debuglog "Interrupted (#{e}). Failing job.", 0
545             $arv.job.update(uuid: c[:job][:uuid],
546                             job: {
547                               state: "Failed"
548                             })
549           end
550         end
551
552         if c[:job] and c[:job][:uuid]
553           if ["Running", "Queued"].include?(c[:job][:state])
554             # Job is running (or may be soon) so update copy of job record
555             c[:job] = JobCache.get(c[:job][:uuid])
556           end
557
558           if c[:job][:state] == "Complete"
559             # Populate script_parameters of other components waiting for
560             # this job
561             @components.each do |c2name, c2|
562               c2[:script_parameters].each do |pname, p|
563                 if p.is_a? Hash and p[:output_of] == cname.to_s
564                   debuglog "parameter #{c2name}::#{pname} == #{c[:job][:output]}"
565                   c2[:script_parameters][pname] = {value: c[:job][:output]}
566                   moretodo = true
567                 end
568               end
569             end
570             unless c_already_finished
571               # This is my first time discovering that the job
572               # succeeded. (At the top of this loop, I was still
573               # waiting for it to finish.)
574
575               if @instance[:name].andand.length.andand > 0
576                 pipeline_name = @instance[:name]
577               elsif @template.andand[:name].andand.length.andand > 0
578                 pipeline_name = @template[:name]
579               else
580                 pipeline_name = @instance[:uuid]
581               end
582               if c[:output_name] != false
583                 # Create a collection located in the same project as the pipeline with the contents of the output.
584                 portable_data_hash = c[:job][:output]
585                 collections = $arv.collection.list(limit: 1,
586                                                    filters: [['portable_data_hash', '=', portable_data_hash]],
587                                                    select: ["portable_data_hash", "manifest_text"]
588                                                    )[:items]
589                 if collections.any?
590                   name = c[:output_name] || "Output #{portable_data_hash[0..7]} of #{cname} of #{pipeline_name}"
591
592                   # check if there is a name collision.
593                   name_collisions = $arv.collection.list(filters: [["owner_uuid", "=", owner_uuid],
594                                                                    ["name", "=", name]])[:items]
595
596                   newcollection_actual = nil
597                   if name_collisions.any? and name_collisions.first[:portable_data_hash] == portable_data_hash
598                     # There is already a collection with the same name and the
599                     # same contents, so just point to that.
600                     newcollection_actual = name_collisions.first
601                   end
602
603                   if newcollection_actual.nil?
604                     # Did not find a collection with the same name (or the
605                     # collection has a different portable data hash) so create
606                     # a new collection with ensure_unique_name: true.
607                     newcollection = {
608                       owner_uuid: owner_uuid,
609                       name: name,
610                       portable_data_hash: collections.first[:portable_data_hash],
611                       manifest_text: collections.first[:manifest_text]
612                     }
613                     debuglog "Creating collection #{newcollection}", 0
614                     newcollection_actual = $arv.collection.create collection: newcollection, ensure_unique_name: true
615                   end
616
617                   c[:output_uuid] = newcollection_actual[:uuid]
618                 else
619                   debuglog "Could not find a collection with portable data hash #{portable_data_hash}", 0
620                 end
621               end
622             end
623           elsif ["Queued", "Running"].include? c[:job][:state]
624             # Job is running or queued to run, so indicate that pipeline
625             # should continue to run
626             moretodo = true
627           elsif c[:job][:state] == "Cancelled"
628             debuglog "component #{cname} job #{c[:job][:uuid]} cancelled."
629             moretodo = false
630           elsif c[:job][:state] == "Failed"
631             moretodo = false
632           end
633         end
634       end
635       @instance[:components] = @components
636       report_status
637
638       if @options[:no_wait]
639         moretodo = false
640       end
641
642       # If job creation fails, just give up on this pipeline instance.
643       if job_creation_failed > 0
644         moretodo = false
645       end
646
647       if moretodo
648         begin
649           sleep 10
650         rescue Interrupt
651           debuglog "interrupt", 0
652           interrupted = true
653           break
654         end
655       end
656     end
657
658     c_in_state = @components.values.group_by { |c|
659       c[:job] and c[:job][:state]
660     }
661     succeeded = c_in_state["Complete"].andand.count || 0
662     failed = (c_in_state["Failed"].andand.count || 0) + (c_in_state["Cancelled"].andand.count || 0)
663     ended = succeeded + failed
664
665     success = (succeeded == @components.length)
666
667     # A job create call failed. Just give up.
668     if job_creation_failed > 0
669       debuglog "job creation failed - giving up on this pipeline instance", 0
670       success = false
671       failed += 1
672     end
673
674     if interrupted
675      if success
676         @instance[:state] = 'Complete'
677      else
678         @instance[:state] = 'Paused'
679       end
680     else
681       if ended == @components.length or failed > 0
682         @instance[:state] = success ? 'Complete' : 'Failed'
683       end
684     end
685
686     if @instance[:finished_at].nil? and ['Complete', 'Failed'].include? @instance[:state]
687       @instance[:finished_at] = Time.now
688     end
689
690     debuglog "pipeline instance state is #{@instance[:state]}"
691
692     # set components_summary
693     components_summary = {"todo" => @components.length - ended, "done" => succeeded, "failed" => failed}
694     @instance[:components_summary] = components_summary
695
696     @instance.save
697   end
698
699   def cleanup
700     if @instance and @instance[:state] == 'RunningOnClient'
701       @instance[:state] = 'Paused'
702       @instance.save
703     end
704   end
705
706   def uuid
707     @instance[:uuid]
708   end
709
710   protected
711
712   def report_status
713     @instance.save
714
715     if @options[:status_json] != '/dev/null'
716       File.open(@options[:status_json], 'w') do |f|
717         f.puts @components.pretty_inspect
718       end
719     end
720
721     if @options[:status_text] != '/dev/null'
722       File.open(@options[:status_text], 'w') do |f|
723         f.puts ""
724         f.puts "#{Time.now} -- pipeline_instance #{@instance[:uuid]}"
725         namewidth = @components.collect { |cname, c| cname.size }.max
726         @components.each do |cname, c|
727           jstatus = if !c[:job]
728                       "-"
729                     else case c[:job][:state]
730                          when "Running"
731                            "#{c[:job][:tasks_summary].inspect}"
732                          when "Complete"
733                            c[:job][:output]
734                          when "Cancelled"
735                            "cancelled #{c[:job][:cancelled_at]}"
736                          when "Failed"
737                            "failed #{c[:job][:finished_at]}"
738                          when "Queued"
739                            "queued #{c[:job][:created_at]}"
740                          end
741                     end
742           f.puts "#{cname.to_s.ljust namewidth} #{c[:job] ? c[:job][:uuid] : '-'.ljust(27)} #{jstatus}"
743         end
744       end
745     end
746   end
747
748   def abort(msg)
749     if @instance
750       if ["New", "Ready", "RunningOnClient",
751           "RunningOnServer"].include?(@instance[:state])
752         @instance[:state] = "Failed"
753         @instance[:finished_at] = Time.now
754         @instance.save
755       end
756       @instance.log_stderr(msg)
757     end
758     Kernel::abort(msg)
759   end
760 end
761
762 runner = WhRunPipelineInstance.new($options)
763 begin
764   if $options[:template]
765     runner.fetch_template($options[:template])
766   else
767     runner.fetch_instance($options[:instance])
768   end
769   runner.apply_parameters(p.leftovers)
770   runner.setup_instance
771   if $options[:submit]
772     runner.instance.save
773     puts runner.instance[:uuid]
774   else
775     runner.run
776   end
777 rescue Exception => e
778   runner.cleanup
779   raise e
780 end