Merge branch 'master' into 9998-unsigned_manifest
[arvados.git] / services / api / app / models / pipeline_instance.rb
1 class PipelineInstance < ArvadosModel
2   include HasUuid
3   include KindAndEtag
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
9
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
16
17   api_accessible :user, extend: :common do |t|
18     t.add :pipeline_template_uuid
19     t.add :name
20     t.add :components
21     t.add :properties
22     t.add :state
23     t.add :components_summary
24     t.add :description
25     t.add :started_at
26     t.add :finished_at
27   end
28
29   # Supported states for a pipeline instance
30   States =
31     [
32      (New = 'New'),
33      (Ready = 'Ready'),
34      (RunningOnServer = 'RunningOnServer'),
35      (RunningOnClient = 'RunningOnClient'),
36      (Paused = 'Paused'),
37      (Failed = 'Failed'),
38      (Complete = 'Complete'),
39     ]
40
41   # if all components have input, the pipeline is Ready
42   def components_look_ready?
43     if !self.components || self.components.empty?
44       return false
45     end
46
47     all_components_have_input = true
48     self.components.each do |name, component|
49       component['script_parameters'].andand.each do |parametername, parameter|
50         parameter = { 'value' => parameter } unless parameter.is_a? Hash
51         if parameter['value'].nil? and parameter['required']
52           if parameter['output_of']
53             next
54           end
55           all_components_have_input = false
56           break
57         end
58       end
59     end
60     return all_components_have_input
61   end
62
63   def progress_table
64     begin
65       # v0 pipeline format
66       nrow = -1
67       components['steps'].collect do |step|
68         nrow += 1
69         row = [nrow, step['name']]
70         if step['complete'] and step['complete'] != 0
71           if step['output_data_locator']
72             row << 1.0
73           else
74             row << 0.0
75           end
76         else
77           row << 0.0
78           if step['failed']
79             self.state = Failed
80           end
81         end
82         row << (step['warehousejob']['id'] rescue nil)
83         row << (step['warehousejob']['revision'] rescue nil)
84         row << step['output_data_locator']
85         row << (Time.parse(step['warehousejob']['finishtime']) rescue nil)
86         row
87       end
88     rescue
89       []
90     end
91   end
92
93   def progress_ratio
94     t = progress_table
95     return 0 if t.size < 1
96     t.collect { |r| r[2] }.inject(0.0) { |sum,a| sum += a } / t.size
97   end
98
99   def self.queue
100     self.where("state = 'RunningOnServer'")
101   end
102
103   protected
104   def bootstrap_components
105     if pipeline_template and (!components or components.empty?)
106       self.components = pipeline_template.components.deep_dup
107     end
108   end
109
110   def update_state
111     if components and progress_ratio == 1.0
112       self.state = Complete
113     end
114   end
115
116   def verify_status
117     changed_attributes = self.changed
118
119     if new_record? or 'components'.in? changed_attributes
120       self.state ||= New
121       if (self.state == New) and self.components_look_ready?
122         self.state = Ready
123       end
124     end
125
126     if self.state.in?(States)
127       true
128     else
129       errors.add :state, "'#{state.inspect} must be one of: [#{States.join ', '}]"
130       false
131     end
132   end
133
134   def set_state_before_save
135     if self.components_look_ready? && (!self.state || self.state == New)
136       self.state = Ready
137     end
138   end
139
140   def update_timestamps_when_state_changes
141     return if not (state_changed? or new_record?)
142
143     case state
144     when RunningOnServer, RunningOnClient
145       self.started_at ||= db_current_time
146     when Failed, Complete
147       current_time = db_current_time
148       self.started_at ||= current_time
149       self.finished_at ||= current_time
150     end
151   end
152
153 end