apps/workbench/app/models/pipeline_instance.rb

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