apps/workbench/app/models/pipeline_instance.rb

   1 require "arvados/keep"
   2
   3 class PipelineInstance < ArvadosBase
   4   attr_accessor :pipeline_template
   5
   6   def self.goes_in_projects?
   7     true
   8   end
   9
  10   def friendly_link_name lookup=nil
  11     pipeline_name = self.name
  12     if pipeline_name.nil? or pipeline_name.empty?
  13       template = if lookup and lookup[self.pipeline_template_uuid]
  14                    lookup[self.pipeline_template_uuid]
  15                  else
  16                    PipelineTemplate.find?(self.pipeline_template_uuid) if self.pipeline_template_uuid
  17                  end
  18       if template
  19         template.name
  20       else
  21         self.uuid
  22       end
  23     else
  24       pipeline_name
  25     end
  26   end
  27
  28   def content_summary
  29     begin
  30       PipelineTemplate.find(pipeline_template_uuid).name
  31     rescue
  32       super
  33     end
  34   end
  35
  36   def update_job_parameters(new_params)
  37     self.components[:steps].each_with_index do |step, i|
  38       step[:params].each do |param|
  39         if new_params.has_key?(new_param_name = "#{i}/#{param[:name]}") or
  40             new_params.has_key?(new_param_name = "#{step[:name]}/#{param[:name]}") or
  41             new_params.has_key?(new_param_name = param[:name])
  42           param_type = :value
  43           %w(hash data_locator).collect(&:to_sym).each do |ptype|
  44             param_type = ptype if param.has_key? ptype
  45           end
  46           param[param_type] = new_params[new_param_name]
  47         end
  48       end
  49     end
  50   end
  51
  52   def editable_attributes
  53     %w(name description components)
  54   end
  55
  56   def attribute_editable?(name, ever=nil)
  57     if name.to_s == "components"
  58       (ever or %w(New Ready).include?(state)) and super
  59     else
  60       super
  61     end
  62   end
  63
  64   def attributes_for_display
  65     super.reject { |k,v| k == 'components' }
  66   end
  67
  68   def self.creatable?
  69     false
  70   end
  71
  72   def component_input_title(component_name, input_name)
  73     component = components[component_name]
  74     return nil if component.nil?
  75     param_info = component[:script_parameters].andand[input_name.to_sym]
  76     if param_info.is_a?(Hash) and param_info[:title]
  77       param_info[:title]
  78     else
  79       "\"#{input_name.to_s}\" parameter for #{component[:script]} script in #{component_name} component"
  80     end
  81   end
  82
  83   def textile_attributes
  84     [ 'description' ]
  85   end
  86
  87   def job_uuids
  88     components_map { |cspec| cspec[:job][:uuid] rescue nil }
  89   end
  90
  91   def job_log_ids
  92     components_map { |cspec| cspec[:job][:log] rescue nil }
  93   end
  94
  95   def job_ids
  96     components_map { |cspec| cspec[:job][:uuid] rescue nil }
  97   end
  98
  99   def stderr_log_object_uuids
 100     result = job_uuids.values.compact
 101     result << uuid
 102   end
 103
 104   def stderr_log_query(limit=nil)
 105     query = Log.
 106             with_count('none').
 107             where(event_type: "stderr",
 108                   object_uuid: stderr_log_object_uuids).
 109             order("created_at DESC")
 110     unless limit.nil?
 111       query = query.limit(limit)
 112     end
 113     query
 114   end
 115
 116   def stderr_log_lines(limit=2000)
 117     stderr_log_query(limit).results.reverse.
 118       flat_map { |log| log.properties[:text].split("\n") rescue [] }
 119   end
 120
 121   def has_readable_logs?
 122     log_pdhs, log_uuids = job_log_ids.values.compact.partition do |loc_s|
 123       Keep::Locator.parse(loc_s)
 124     end
 125     if log_pdhs.any? and
 126         Collection.where(portable_data_hash: log_pdhs).limit(1).results.any?
 127       true
 128     elsif log_uuids.any? and
 129         Collection.where(uuid: log_uuids).limit(1).results.any?
 130       true
 131     else
 132       stderr_log_query(1).results.any?
 133     end
 134   end
 135
 136   def work_unit(label=nil)
 137     PipelineInstanceWorkUnit.new(self, label || self.name, self.uuid)
 138   end
 139
 140   def cancel
 141     arvados_api_client.api "pipeline_instances/#{self.uuid}/", "cancel", {"cascade" => true}
 142   end
 143
 144   private
 145
 146   def components_map
 147     Hash[components.map { |cname, cspec| [cname, yield(cspec)] }]
 148   end
 149 end