1 class PipelineInstance < ArvadosModel
4 include CommonApiTemplate
5 serialize :components, Hash
6 serialize :properties, Hash
7 serialize :components_summary, Hash
8 belongs_to :pipeline_template, :foreign_key => :pipeline_template_uuid, :primary_key => :uuid
10 before_validation :bootstrap_components
11 before_validation :update_state
12 before_validation :verify_status
13 before_validation :update_timestamps_when_state_changes
14 before_create :set_state_before_save
15 before_save :set_state_before_save
17 api_accessible :user, extend: :common do |t|
18 t.add :pipeline_template_uuid
23 t.add :components_summary
29 # Supported states for a pipeline instance
34 (RunningOnServer = 'RunningOnServer'),
35 (RunningOnClient = 'RunningOnClient'),
38 (Complete = 'Complete'),
41 def self.limit_index_columns_read
45 # if all components have input, the pipeline is Ready
46 def components_look_ready?
47 if !self.components || self.components.empty?
51 all_components_have_input = true
52 self.components.each do |name, component|
53 component['script_parameters'].andand.each do |parametername, parameter|
54 parameter = { 'value' => parameter } unless parameter.is_a? Hash
55 if parameter['value'].nil? and parameter['required']
56 if parameter['output_of']
59 all_components_have_input = false
64 return all_components_have_input
71 components['steps'].collect do |step|
73 row = [nrow, step['name']]
74 if step['complete'] and step['complete'] != 0
75 if step['output_data_locator']
86 row << (step['warehousejob']['id'] rescue nil)
87 row << (step['warehousejob']['revision'] rescue nil)
88 row << step['output_data_locator']
89 row << (Time.parse(step['warehousejob']['finishtime']) rescue nil)
99 return 0 if t.size < 1
100 t.collect { |r| r[2] }.inject(0.0) { |sum,a| sum += a } / t.size
104 self.where("state = 'RunningOnServer'")
107 def cancel(cascade: false, need_transaction: true)
109 ActiveRecord::Base.transaction do
110 cancel(cascade: cascade, need_transaction: false)
115 if self.state.in?([RunningOnServer, RunningOnClient])
118 elsif self.state != Paused
119 raise InvalidStateTransitionError
124 # cancel all child jobs
125 children = self.components.andand.collect{|_, c| c['job']}.compact.collect{|j| j['uuid']}.compact
127 return if children.empty?
129 Job.where(uuid: children, state: [Job::Queued, Job::Running]).each do |job|
130 job.cancel(cascade: cascade, need_transaction: false)
135 def bootstrap_components
136 if pipeline_template and (!components or components.empty?)
137 self.components = pipeline_template.components.deep_dup
142 if components and progress_ratio == 1.0
143 self.state = Complete
148 changed_attributes = self.changed
150 if new_record? or 'components'.in? changed_attributes
152 if (self.state == New) and self.components_look_ready?
157 if self.state.in?(States)
160 errors.add :state, "'#{state.inspect} must be one of: [#{States.join ', '}]"
165 def set_state_before_save
166 if self.components_look_ready? && (!self.state || self.state == New)
171 def update_timestamps_when_state_changes
172 return if not (state_changed? or new_record?)
175 when RunningOnServer, RunningOnClient
176 self.started_at ||= db_current_time
177 when Failed, Complete
178 current_time = db_current_time
179 self.started_at ||= current_time
180 self.finished_at ||= current_time