apps/workbench/app/helpers/pipeline_instances_helper.rb

   1 module PipelineInstancesHelper
   2
   3   def pipeline_jobs object=nil
   4     object ||= @object
   5     if object.components[:steps].is_a? Array
   6       pipeline_jobs_oldschool object
   7     elsif object.components.is_a? Hash
   8       pipeline_jobs_newschool object
   9     end
  10   end
  11
  12   def render_pipeline_jobs
  13     pipeline_jobs.collect do |pj|
  14       render_pipeline_job pj
  15     end
  16   end
  17
  18   def render_pipeline_job pj
  19     pj[:progress_bar] = render partial: 'job_progress', locals: {:j => pj[:job]}
  20     pj[:output_link] = link_to_if_arvados_object pj[:output]
  21     pj[:job_link] = link_to_if_arvados_object pj[:job][:uuid]
  22     pj
  23   end
  24
  25   def merge_range timestamps, started_at, finished_at
  26     timestamps.each_index do |i|
  27       if started_at
  28         if started_at >= timestamps[i][0] and finished_at <= timestamps[i][1]
  29           # 'j' started and ended during 'i'
  30           return timestamps
  31         end
  32
  33         if started_at < timestamps[i][0] and finished_at >= timestamps[i][0] and finished_at <= timestamps[i][1]
  34           # 'j' started before 'i' and finished during 'i'
  35           # re-merge range between when 'j' started and 'i' finished
  36           finished_at = timestamps[i][1]
  37           timestamps.delete_at i
  38           return merge_range timestamps, started_at, finished_at
  39         end
  40
  41         if started_at >= timestamps[i][0] and started_at <= timestamps[i][1]
  42           # 'j' started during 'i' and finished sometime after
  43           # move end time of 'i' back
  44           # re-merge range between when 'i' started and 'j' finished
  45           started_at = timestamps[i][0]
  46           timestamps.delete_at i
  47           return merge_range timestamps, started_at, finished_at
  48         end
  49
  50         if finished_at < timestamps[i][0]
  51           # 'j' finished before 'i' started, so insert before 'i'
  52           timestamps.insert i, [started_at, finished_at]
  53           return timestamps
  54         end
  55       end
  56     end
  57
  58     timestamps << [started_at, finished_at]
  59   end
  60
  61   def determine_wallclock_runtime jobs
  62     puts "Begin #{jobs}"
  63     timestamps = []
  64     jobs.each do |j|
  65       insert_at = 0
  66       started_at = j[:started_at]
  67       finished_at = (if j[:finished_at] then j[:finished_at] else Time.now end)
  68       if started_at
  69         timestamps = merge_range timestamps, started_at, finished_at
  70       end
  71     end
  72     timestamps.map { |t| t[1] - t[0] }.reduce(:+)
  73   end
  74
  75   protected
  76
  77   def pipeline_jobs_newschool object
  78     ret = []
  79     i = -1
  80
  81     jobuuids = object.components.values.map { |c|
  82       c[:job][:uuid] if c.is_a?(Hash) and c[:job].is_a?(Hash)
  83     }.compact
  84     job = {}
  85     Job.where(uuid: jobuuids).each do |j|
  86       job[j[:uuid]] = j
  87     end
  88
  89     object.components.each do |cname, c|
  90       i += 1
  91       pj = {index: i, name: cname}
  92       if not c.is_a?(Hash)
  93         ret << pj
  94         next
  95       end
  96       if c[:job] and c[:job][:uuid] and job[c[:job][:uuid]]
  97         pj[:job] = job[c[:job][:uuid]]
  98       else
  99         pj[:job] = c[:job].is_a?(Hash) ? c[:job] : {}
 100       end
 101       pj[:percent_done] = 0
 102       pj[:percent_running] = 0
 103       if pj[:job][:success]
 104         if pj[:job][:output]
 105           pj[:progress] = 1.0
 106           pj[:percent_done] = 100
 107         else
 108           pj[:progress] = 0.0
 109         end
 110       else
 111         if pj[:job][:tasks_summary]
 112           begin
 113             ts = pj[:job][:tasks_summary]
 114             denom = ts[:done].to_f + ts[:running].to_f + ts[:todo].to_f
 115             pj[:progress] = (ts[:done].to_f + ts[:running].to_f/2) / denom
 116             pj[:percent_done] = 100.0 * ts[:done].to_f / denom
 117             pj[:percent_running] = 100.0 * ts[:running].to_f / denom
 118             pj[:progress_detail] = "#{ts[:done]} done #{ts[:running]} run #{ts[:todo]} todo"
 119           rescue
 120             pj[:progress] = 0.5
 121             pj[:percent_done] = 0.0
 122             pj[:percent_running] = 100.0
 123           end
 124         else
 125           pj[:progress] = 0.0
 126         end
 127       end
 128       if pj[:job][:success]
 129         pj[:result] = 'complete'
 130         pj[:labeltype] = 'success'
 131         pj[:complete] = true
 132         pj[:progress] = 1.0
 133       elsif pj[:job][:finished_at]
 134         pj[:result] = 'failed'
 135         pj[:labeltype] = 'danger'
 136         pj[:failed] = true
 137       elsif pj[:job][:started_at]
 138         pj[:result] = 'running'
 139         pj[:labeltype] = 'primary'
 140       elsif pj[:job][:uuid]
 141         pj[:result] = 'queued'
 142         pj[:labeltype] = 'default'
 143       else
 144         pj[:result] = 'none'
 145         pj[:labeltype] = 'default'
 146       end
 147
 148       pj[:job_id] = pj[:job][:uuid]
 149       pj[:script] = pj[:job][:script] || c[:script]
 150       pj[:repository] = pj[:job][:script] || c[:repository]
 151       pj[:script_parameters] = pj[:job][:script_parameters] || c[:script_parameters]
 152       pj[:script_version] = pj[:job][:script_version] || c[:script_version]
 153       pj[:nondeterministic] = pj[:job][:nondeterministic] || c[:nondeterministic]
 154       pj[:output] = pj[:job][:output]
 155       pj[:output_uuid] = c[:output_uuid]
 156       pj[:finished_at] = (Time.parse(pj[:job][:finished_at]) rescue nil)
 157       ret << pj
 158     end
 159     ret
 160   end
 161
 162   def pipeline_jobs_oldschool object
 163     ret = []
 164     object.components[:steps].each_with_index do |step, i|
 165       pj = {index: i, name: step[:name]}
 166       if step[:complete] and step[:complete] != 0
 167         if step[:output_data_locator]
 168           pj[:progress] = 1.0
 169         else
 170           pj[:progress] = 0.0
 171         end
 172       else
 173         if step[:progress] and
 174             (re = step[:progress].match /^(\d+)\+(\d+)\/(\d+)$/)
 175           pj[:progress] = (((re[1].to_f + re[2].to_f/2) / re[3].to_f) rescue 0.5)
 176         else
 177           pj[:progress] = 0.0
 178         end
 179         if step[:failed]
 180           pj[:result] = 'failed'
 181           pj[:failed] = true
 182         end
 183       end
 184       if step[:warehousejob]
 185         if step[:complete]
 186           pj[:result] = 'complete'
 187           pj[:complete] = true
 188           pj[:progress] = 1.0
 189         elsif step[:warehousejob][:finishtime]
 190           pj[:result] = 'failed'
 191           pj[:failed] = true
 192         elsif step[:warehousejob][:starttime]
 193           pj[:result] = 'running'
 194         else
 195           pj[:result] = 'queued'
 196         end
 197       end
 198       pj[:progress_detail] = (step[:progress] rescue nil)
 199       pj[:job_id] = (step[:warehousejob][:id] rescue nil)
 200       pj[:job_link] = pj[:job_id]
 201       pj[:script] = step[:function]
 202       pj[:script_version] = (step[:warehousejob][:revision] rescue nil)
 203       pj[:output] = step[:output_data_locator]
 204       pj[:finished_at] = (Time.parse(step[:warehousejob][:finishtime]) rescue nil)
 205       ret << pj
 206     end
 207     ret
 208   end
 209
 210   def runtime duration, long
 211     hours = 0
 212     minutes = 0
 213     seconds = 0
 214     if duration >= 3600
 215       hours = (duration / 3600).floor
 216       duration -= hours * 3600
 217     end
 218     if duration >= 60
 219       minutes = (duration / 60).floor
 220       duration -= minutes * 60
 221     end
 222     duration = duration.floor
 223
 224     if long
 225       s = ""
 226       if hours > 0 then
 227         s += "#{hours} hour#{'s' if hours != 1} "
 228       end
 229       if minutes > 0 then
 230         s += "#{minutes} minute#{'s' if minutes != 1} "
 231       end
 232       s += "#{duration} second#{'s' if duration != 1}"
 233     else
 234       s = "#{hours}:#{minutes.to_s.rjust(2, '0')}:#{duration.to_s.rjust(2, '0')}"
 235     end
 236     s
 237   end
 238
 239 end