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     timestamps = []
  63     jobs.each do |j|
  64       insert_at = 0
  65       started_at = j[:started_at]
  66       finished_at = (if j[:finished_at] then j[:finished_at] else Time.now end)
  67       if started_at
  68         timestamps = merge_range timestamps, started_at, finished_at
  69       end
  70     end
  71     timestamps.map { |t| t[1] - t[0] }.reduce(:+) || 0
  72   end
  73
  74   protected
  75
  76   def pipeline_jobs_newschool object
  77     ret = []
  78     i = -1
  79
  80     jobuuids = object.components.values.map { |c|
  81       c[:job][:uuid] if c.is_a?(Hash) and c[:job].is_a?(Hash)
  82     }.compact
  83     job = {}
  84     Job.where(uuid: jobuuids).each do |j|
  85       job[j[:uuid]] = j
  86     end
  87
  88     object.components.each do |cname, c|
  89       i += 1
  90       pj = {index: i, name: cname}
  91       if not c.is_a?(Hash)
  92         ret << pj
  93         next
  94       end
  95       if c[:job] and c[:job][:uuid] and job[c[:job][:uuid]]
  96         pj[:job] = job[c[:job][:uuid]]
  97       else
  98         pj[:job] = c[:job].is_a?(Hash) ? c[:job] : {}
  99       end
 100       pj[:percent_done] = 0
 101       pj[:percent_running] = 0
 102       if pj[:job][:success]
 103         if pj[:job][:output]
 104           pj[:progress] = 1.0
 105           pj[:percent_done] = 100
 106         else
 107           pj[:progress] = 0.0
 108         end
 109       else
 110         if pj[:job][:tasks_summary]
 111           begin
 112             ts = pj[:job][:tasks_summary]
 113             denom = ts[:done].to_f + ts[:running].to_f + ts[:todo].to_f
 114             pj[:progress] = (ts[:done].to_f + ts[:running].to_f/2) / denom
 115             pj[:percent_done] = 100.0 * ts[:done].to_f / denom
 116             pj[:percent_running] = 100.0 * ts[:running].to_f / denom
 117             pj[:progress_detail] = "#{ts[:done]} done #{ts[:running]} run #{ts[:todo]} todo"
 118           rescue
 119             pj[:progress] = 0.5
 120             pj[:percent_done] = 0.0
 121             pj[:percent_running] = 100.0
 122           end
 123         else
 124           pj[:progress] = 0.0
 125         end
 126       end
 127       if pj[:job][:success]
 128         pj[:result] = 'complete'
 129         pj[:labeltype] = 'success'
 130         pj[:complete] = true
 131         pj[:progress] = 1.0
 132       elsif pj[:job][:finished_at]
 133         pj[:result] = 'failed'
 134         pj[:labeltype] = 'danger'
 135         pj[:failed] = true
 136       elsif pj[:job][:started_at]
 137         pj[:result] = 'running'
 138         pj[:labeltype] = 'primary'
 139       elsif pj[:job][:uuid]
 140         pj[:result] = 'queued'
 141         pj[:labeltype] = 'default'
 142       else
 143         pj[:result] = 'none'
 144         pj[:labeltype] = 'default'
 145       end
 146
 147       pj[:job_id] = pj[:job][:uuid]
 148       pj[:script] = pj[:job][:script] || c[:script]
 149       pj[:repository] = pj[:job][:script] || c[:repository]
 150       pj[:script_parameters] = pj[:job][:script_parameters] || c[:script_parameters]
 151       pj[:script_version] = pj[:job][:script_version] || c[:script_version]
 152       pj[:nondeterministic] = pj[:job][:nondeterministic] || c[:nondeterministic]
 153       pj[:output] = pj[:job][:output]
 154       pj[:output_uuid] = c[:output_uuid]
 155       pj[:finished_at] = (Time.parse(pj[:job][:finished_at]) rescue nil)
 156       ret << pj
 157     end
 158     ret
 159   end
 160
 161   def pipeline_jobs_oldschool object
 162     ret = []
 163     object.components[:steps].each_with_index do |step, i|
 164       pj = {index: i, name: step[:name]}
 165       if step[:complete] and step[:complete] != 0
 166         if step[:output_data_locator]
 167           pj[:progress] = 1.0
 168         else
 169           pj[:progress] = 0.0
 170         end
 171       else
 172         if step[:progress] and
 173             (re = step[:progress].match /^(\d+)\+(\d+)\/(\d+)$/)
 174           pj[:progress] = (((re[1].to_f + re[2].to_f/2) / re[3].to_f) rescue 0.5)
 175         else
 176           pj[:progress] = 0.0
 177         end
 178         if step[:failed]
 179           pj[:result] = 'failed'
 180           pj[:failed] = true
 181         end
 182       end
 183       if step[:warehousejob]
 184         if step[:complete]
 185           pj[:result] = 'complete'
 186           pj[:complete] = true
 187           pj[:progress] = 1.0
 188         elsif step[:warehousejob][:finishtime]
 189           pj[:result] = 'failed'
 190           pj[:failed] = true
 191         elsif step[:warehousejob][:starttime]
 192           pj[:result] = 'running'
 193         else
 194           pj[:result] = 'queued'
 195         end
 196       end
 197       pj[:progress_detail] = (step[:progress] rescue nil)
 198       pj[:job_id] = (step[:warehousejob][:id] rescue nil)
 199       pj[:job_link] = pj[:job_id]
 200       pj[:script] = step[:function]
 201       pj[:script_version] = (step[:warehousejob][:revision] rescue nil)
 202       pj[:output] = step[:output_data_locator]
 203       pj[:finished_at] = (Time.parse(step[:warehousejob][:finishtime]) rescue nil)
 204       ret << pj
 205     end
 206     ret
 207   end
 208
 209   def runtime duration, long
 210     hours = 0
 211     minutes = 0
 212     seconds = 0
 213     if duration >= 3600
 214       hours = (duration / 3600).floor
 215       duration -= hours * 3600
 216     end
 217     if duration >= 60
 218       minutes = (duration / 60).floor
 219       duration -= minutes * 60
 220     end
 221     duration = duration.floor
 222
 223     if long
 224       s = ""
 225       if hours > 0 then
 226         s += "#{hours} hour#{'s' if hours != 1} "
 227       end
 228       if minutes > 0 then
 229         s += "#{minutes} minute#{'s' if minutes != 1} "
 230       end
 231       s += "#{duration} second#{'s' if duration != 1}"
 232     else
 233       s = "#{hours}:#{minutes.to_s.rjust(2, '0')}:#{duration.to_s.rjust(2, '0')}"
 234     end
 235     s
 236   end
 237
 238 end