X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/987c225f9f6845868ee674902090c27a5a064f42..d084ca24b06c598271844d2ba4c8c40f251c0999:/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 b70fd7dad6..b66e9c0526 100755 --- a/sdk/cli/bin/arv-run-pipeline-instance +++ b/sdk/cli/bin/arv-run-pipeline-instance @@ -1,63 +1,8 @@ #!/usr/bin/env ruby - -# == Synopsis -# -# 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. -# -# == Options -# -# [--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 -# -# [--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] Do not reuse existing jobs to satisfy pipeline -# components. Submit a new job for every component. -# -# [--debug] Print extra debugging information on stderr. -# -# [--debug-level N] Increase amount of debugging information. Default -# 1, possible range 0..3. -# -# [--status-text path] Print plain text status report to a file or -# fifo. Default: /dev/stdout -# -# [--status-json path] Print JSON status report to a file or -# fifo. Default: /dev/null -# -# [--description] Description for the pipeline instance. -# -# == Parameters -# -# [param_name=param_value] -# -# [param_name param_value] Set (or override) the default value for -# every parameter with the given name. -# -# [component_name::param_name=param_value] -# [component_name::param_name param_value] -# [--component_name::param_name=param_value] -# [--component_name::param_name param_value] Set the value of a -# parameter for a single -# component. +# Copyright (C) The Arvados Authors. All rights reserved. # +# SPDX-License-Identifier: Apache-2.0 + class WhRunPipelineInstance end @@ -67,12 +12,6 @@ if RUBY_VERSION < '1.9.3' then EOS end -$arvados_api_version = ENV['ARVADOS_API_VERSION'] || 'v1' -$arvados_api_host = ENV['ARVADOS_API_HOST'] or - abort "#{$0}: fatal: ARVADOS_API_HOST environment variable not set." -$arvados_api_token = ENV['ARVADOS_API_TOKEN'] or - abort "#{$0}: fatal: ARVADOS_API_TOKEN environment variable not set." - begin require 'arvados' require 'rubygems' @@ -81,7 +20,7 @@ begin require 'trollop' require 'google/api_client' rescue LoadError => l - puts $: + $stderr.puts $: abort <<-EOS #{$0}: fatal: #{l.message} Some runtime dependencies may be missing. @@ -93,28 +32,33 @@ def debuglog(message, verbosity=1) $stderr.puts "#{File.split($0).last} #{$$}: #{message}" if $debuglevel >= verbosity end -module Kernel - def suppress_warnings - original_verbosity = $VERBOSE - $VERBOSE = nil - result = yield - $VERBOSE = original_verbosity - return result - end -end - -if $arvados_api_host.match /local/ - # You probably don't care about SSL certificate checks if you're - # testing with a dev server. - suppress_warnings { OpenSSL::SSL::VERIFY_PEER = OpenSSL::SSL::VERIFY_NONE } -end - - # Parse command line options (the kind that control the behavior of # this program, that is, not the pipeline component parameters). p = Trollop::Parser.new do version __FILE__ + banner(< :boolean, @@ -172,6 +116,10 @@ p = Trollop::Parser.new do "Description for the pipeline instance.", :short => :none, :type => :string) + opt(:project_uuid, + "UUID of the project for the pipeline instance.", + short: :none, + type: :string) stop_on [:'--'] end $options = Trollop::with_standard_exception_handling p do @@ -187,7 +135,7 @@ if $options[:instance] abort "#{$0}: syntax error: --instance cannot be combined with --template or --submit." end elsif not $options[:template] - puts "error: you must supply a --template or --instance." + $stderr.puts "error: you must supply a --template or --instance." p.educate abort end @@ -196,22 +144,6 @@ if $options[:run_pipeline_here] == $options[:submit] abort "#{$0}: error: you must supply --run-pipeline-here, --run-jobs-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. $arv = Arvados.new api_version: 'v1' @@ -226,7 +158,7 @@ class PipelineInstance }, :authenticated => false, :headers => { - authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN'] + authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN'] }) j = JSON.parse result.body, :symbolize_names => true unless j.is_a? Hash and j[:uuid] @@ -244,7 +176,7 @@ class PipelineInstance }, :authenticated => false, :headers => { - authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN'] + authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN'] }) j = JSON.parse result.body, :symbolize_names => true unless j.is_a? Hash and j[:uuid] @@ -263,7 +195,7 @@ class PipelineInstance }, :authenticated => false, :headers => { - authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN'] + authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN'] }) j = JSON.parse result.body, :symbolize_names => true unless j.is_a? Hash and j[:uuid] @@ -307,7 +239,7 @@ class JobCache }, :authenticated => false, :headers => { - authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN'] + authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN'] }) @cache[uuid] = JSON.parse result.body, :symbolize_names => true end @@ -319,7 +251,7 @@ class JobCache }, :authenticated => false, :headers => { - authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN'] + authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN'] }) list = JSON.parse result.body, :symbolize_names => true if list and list[:items].is_a? Array @@ -328,31 +260,46 @@ class JobCache [] end end + + # create() returns [job, exception]. If both job and exception are + # nil, there was a non-retryable error and the call should not be + # attempted again. def self.create(pipeline, component, job, create_params) @cache ||= {} body = {job: no_nil_values(job)}.merge(no_nil_values(create_params)) - result = $client.execute(:api_method => $arvados.jobs.create, - :body_object => body, - :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] + result = nil + begin + result = $client.execute( + :api_method => $arvados.jobs.create, + :body_object => body, + :authenticated => false, + :headers => { + authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN'] + }) + if result.status == 429 || result.status >= 500 + raise Exception.new("HTTP status #{result.status}") + end + rescue Exception => e + return nil, e + end + j = JSON.parse(result.body, :symbolize_names => true) rescue nil + if result.status == 200 && j.is_a?(Hash) && j[:uuid] @cache[j[:uuid]] = j + return j, nil else - debuglog "create job: #{j[:errors] rescue nil} with attributes #{body}", 0 + errors = j[:errors] rescue [] + debuglog "create job: [#{result.status}] #{errors.inspect} with attributes #{body}", 0 msg = "" - j[:errors].each do |err| + errors.each do |err| msg += "Error creating job for component #{component}: #{err}\n" end msg += "Job submission was: #{body.to_json}" pipeline.log_stderr(msg) - nil + return nil, nil end end @@ -381,7 +328,7 @@ class WhRunPipelineInstance }, :authenticated => false, :headers => { - authorization: 'OAuth2 '+ENV['ARVADOS_API_TOKEN'] + authorization: 'OAuth2 '+$arv.config['ARVADOS_API_TOKEN'] }) @template = JSON.parse result.body, :symbolize_names => true if !@template[:uuid] @@ -435,19 +382,27 @@ class WhRunPipelineInstance @components.each do |componentname, component| component[:script_parameters].each do |parametername, parameter| parameter = { :value => parameter } unless parameter.is_a? Hash - value = - (params["#{componentname}::#{parametername}"] || - parameter[:value] || - (parameter[:output_of].nil? && - (params[parametername.to_s] || - parameter[:default])) || - nil) - if value.nil? and - ![false,'false',0,'0'].index parameter[:required] - if parameter[:output_of] - next + if params.has_key?("#{componentname}::#{parametername}") + value = params["#{componentname}::#{parametername}"] + elsif parameter.has_key?(:value) + value = parameter[:value] + elsif parameter.has_key?(:output_of) + if !@components[parameter[:output_of].intern] + errors << [componentname, parametername, "output_of refers to nonexistent component '#{parameter[:output_of]}'"] + else + # value will be filled in later when the upstream + # component's output becomes known end + next + elsif params.has_key?(parametername.to_s) + value = params[parametername.to_s] + elsif parameter.has_key?(:default) + value = parameter[:default] + elsif [false, 'false', 0, '0'].index(parameter[:required]) + value = nil + else errors << [componentname, parametername, "required parameter is missing"] + next end debuglog "parameter #{componentname}::#{parametername} == #{value}" @@ -456,7 +411,10 @@ class WhRunPipelineInstance end end if !errors.empty? - abort "\n#{Time.now} -- pipeline_template #{@template[:uuid]}\nErrors:\n#{errors.collect { |c,p,e| "#{c}::#{p} - #{e}\n" }.join ""}" + all_errors = errors.collect do |c,p,e| + "#{c}::#{p} - #{e}\n" + end.join("") + abort "\n#{Time.now} -- pipeline_template #{@template[:uuid]}\nErrors:\n#{all_errors}" end debuglog "options=" + @options.pretty_inspect self @@ -476,18 +434,23 @@ class WhRunPipelineInstance end end else - description = $options[:description] - description = ("Created at #{Time.now.localtime}" + (@template[:name].andand.size.andand>0 ? " using the pipeline template *#{@template[:name]}*" : "")) if !description - @instance = PipelineInstance. - create(components: @components, - properties: { - run_options: { - enable_job_reuse: !@options[:no_reuse] - } - }, - pipeline_template_uuid: @template[:uuid], - description: description, - state: ($options[:submit] ? 'RunningOnServer' : 'RunningOnClient')) + description = $options[:description] || + ("Created at #{Time.now.localtime}" + (@template[:name].andand.size.andand>0 ? " using the pipeline template *#{@template[:name]}*" : "")) + instance_body = { + components: @components, + properties: { + run_options: { + enable_job_reuse: !@options[:no_reuse] + } + }, + pipeline_template_uuid: @template[:uuid], + description: description, + state: ($options[:submit] ? 'RunningOnServer' : 'RunningOnClient') + } + if @options[:project_uuid] + instance_body[:owner_uuid] = @options[:project_uuid] + end + @instance = PipelineInstance.create(instance_body) end self end @@ -518,7 +481,7 @@ class WhRunPipelineInstance # are fully specified (any output_of script_parameters are resolved # to real value) my_submit_id = "instance #{@instance[:uuid]} rand #{rand(2**64).to_s(36)}" - job = JobCache.create(@instance, cname, { + job, err = JobCache.create(@instance, cname, { :script => c[:script], :script_parameters => Hash[c[:script_parameters].map do |key, spec| [key, spec[:value]] @@ -545,9 +508,11 @@ class WhRunPipelineInstance c[:job] = job c[:run_in_process] = (@options[:run_jobs_here] and job[:submit_id] == my_submit_id) - else + elsif err.nil? debuglog "component #{cname} new job failed", 0 job_creation_failed += 1 + else + debuglog "component #{cname} new job failed, err=#{err}", 0 end end @@ -712,7 +677,7 @@ class WhRunPipelineInstance @instance[:state] = 'Complete' else @instance[:state] = 'Paused' - end + end else if ended == @components.length or failed > 0 @instance[:state] = success ? 'Complete' : 'Failed'