Class: OodCore::Job::Adapters::Flux

Inherits:
OodCore::Job::Adapter show all
Defined in:
lib/ood_core/job/adapters/flux.rb

Overview

An adapter object that describes the communication with a Flux resource manager for job management.

Job ids are stored and returned in decimal. Flux prints ids in F58 (e.g. "ƒiFkbMhFM") by default, which doesn't belong in URLs, file paths or batch connect session files. Every flux command accepts decimal ids as input.

Defined Under Namespace

Modules: JobId Classes: Batch

Constant Summary collapse

STATE_MAP =

Maps Flux job states (RFC 21) to OOD job states. A pending job with urgency 0 is held, see #get_state.

{
  "NEW"      => :queued,
  "DEPEND"   => :queued,
  "PRIORITY" => :queued,
  "SCHED"    => :queued,
  "RUN"      => :running,
  "CLEANUP"  => :running,
  "INACTIVE" => :completed
}.freeze
HOLD_URGENCY =

Urgency Flux gives a held job

0
UNLIMITED_DURATION =

Flux reports "no time limit" as a duration of 0

0
SYSTEM_PATH =

PATH for jobs submitted with a cleared environment

"/usr/local/bin:/usr/bin:/bin:/usr/local/sbin:/usr/sbin:/sbin".freeze

Instance Method Summary collapse

Methods inherited from OodCore::Job::Adapter

#accounts, #cluster_info, #info_all_each, #info_historic, #info_where_owner_each, #job_name_illegal_chars, #nodes, #queues, #sanitize_job_name

Constructor Details

#initialize(opts = {}) ⇒ Flux

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of Flux.

Parameters:

  • opts (#to_h) (defaults to: {}) —

    the options defining this adapter

Options Hash (opts):

  • :flux (Batch) —

    The Flux batch object

See Also:



74
75
76
77
78
 
# File 'lib/ood_core/job/adapters/flux.rb', line 74

def initialize(opts = {})
  o = opts.to_h.symbolize_keys

  @flux = o.fetch(:flux) { raise ArgumentError, "No flux object specified. Missing argument: flux" }
end

Instance Method Details

#delete(id) ⇒ void

This method returns an undefined value.

Delete the submitted job

Parameters:

  • id (#to_s) —

    the id of the job

Raises:

See Also:



211
212
213
214
215
 
# File 'lib/ood_core/job/adapters/flux.rb', line 211

def delete(id)
  @flux.delete_job(id.to_s)
rescue Batch::Error => e
  raise JobAdapterError, e.message unless unknown_job?(e)
end

#directive_prefix ⇒ Object 



217
218
219
 
# File 'lib/ood_core/job/adapters/flux.rb', line 217

def directive_prefix
  "# flux:"
end

#hold(id) ⇒ void

This method returns an undefined value.

Put the submitted job on hold

Parameters:

  • id (#to_s) —

    the id of the job

Raises:

See Also:



189
190
191
192
193
 
# File 'lib/ood_core/job/adapters/flux.rb', line 189

def hold(id)
  @flux.hold_job(id.to_s)
rescue Batch::Error => e
  raise JobAdapterError, e.message unless unknown_job?(e)
end

#info(id) ⇒ Info

Retrieve job info from the resource manager

Parameters:

  • id (#to_s) —

    the id of the job

Returns:

  • (Info) —

    information describing submitted job

Raises:

See Also:



164
165
166
167
168
169
170
171
172
173
 
# File 'lib/ood_core/job/adapters/flux.rb', line 164

def info(id)
  id = id.to_s
  job = @flux.get_job(id)
  job.nil? ? Info.new(id: id, status: :completed) : parse_job_info(job)
rescue Batch::Error => e
  # Flux forgets inactive jobs eventually, so an unknown job is done
  raise JobAdapterError, e.message unless unknown_job?(e)

  Info.new(id: id, status: :completed)
end

#info_all(attrs: nil) ⇒ Array<Info>

Retrieve info for all active jobs from the resource manager

Returns:

  • (Array<Info>) —

    information describing submitted jobs

Raises:

See Also:



141
142
143
144
145
 
# File 'lib/ood_core/job/adapters/flux.rb', line 141

def info_all(attrs: nil)
  @flux.get_jobs.map { |job| parse_job_info(job) }
rescue Batch::Error => e
  raise JobAdapterError, e.message
end

#info_where_owner(owner, attrs: nil) ⇒ Array<Info>

Retrieve info for all active jobs of a given owner or owners

Parameters:

  • owner (#to_s, Array<#to_s>) —

    the owner(s) of the jobs

Returns:

  • (Array<Info>) —

    information describing submitted jobs

Raises:



151
152
153
154
155
156
157
 
# File 'lib/ood_core/job/adapters/flux.rb', line 151

def info_where_owner(owner, attrs: nil)
  Array.wrap(owner).map(&:to_s).uniq.flat_map do |user|
    @flux.get_jobs(owner: user).map { |job| parse_job_info(job) }
  end
rescue Batch::Error => e
  raise JobAdapterError, e.message
end

#release(id) ⇒ void

This method returns an undefined value.

Release the job that is on hold

Parameters:

  • id (#to_s) —

    the id of the job

Raises:

See Also:



200
201
202
203
204
 
# File 'lib/ood_core/job/adapters/flux.rb', line 200

def release(id)
  @flux.release_job(id.to_s)
rescue Batch::Error => e
  raise JobAdapterError, e.message unless unknown_job?(e)
end

#status(id) ⇒ Status

Retrieve job status from resource manager

Parameters:

  • id (#to_s) —

    the id of the job

Returns:

Raises:

See Also:



180
181
182
 
# File 'lib/ood_core/job/adapters/flux.rb', line 180

def status(id)
  info(id).status
end

#submit(script, after: [], afterok: [], afternotok: [], afterany: []) ⇒ String

Submit a job with the attributes defined in the job template instance

Parameters:

  • script (Script) —

    script object that describes the script and attributes for the submitted job

  • after (#to_s, Array<#to_s>) (defaults to: []) —

    this job may be scheduled for execution at any point after dependent jobs have started execution

  • afterok (#to_s, Array<#to_s>) (defaults to: []) —

    this job may be scheduled for execution only after dependent jobs have terminated with no errors

  • afternotok (#to_s, Array<#to_s>) (defaults to: []) —

    this job may be scheduled for execution only after dependent jobs have terminated with errors

  • afterany (#to_s, Array<#to_s>) (defaults to: []) —

    this job may be scheduled for execution after dependent jobs have terminated

Returns:

  • (String) —

    the job id, in decimal

Raises:

See Also:



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/ood_core/job/adapters/flux.rb', line 94

def submit(script, after: [], afterok: [], afternotok: [], afterany: [])
  validate_native(script, Array)
  unless script.job_array_request.nil?
    raise JobAdapterError, "Flux does not support job arrays"
  end

  args = []
  args.concat ["--urgency=hold"] if script.submit_as_hold
  args.concat ["--cwd", script.workdir.to_s] unless script.workdir.nil?
  args.concat ["--job-name", script.job_name] unless script.job_name.nil?
  args.concat ["--input", script.input_path.to_s] unless script.input_path.nil?
  args.concat ["--output", script.output_path.to_s] unless script.output_path.nil?
  args.concat ["--error", script.error_path.to_s] unless script.error_path.nil?
  args.concat ["-q", script.queue_name] unless script.queue_name.nil?
  args.concat ["--bank", script.accounting_id] unless script.accounting_id.nil?
  args.concat ["-t", "#{script.wall_time.to_i}s"] unless script.wall_time.nil?
  # flux batch won't submit without a size, so default to one slot
  # (one core), like a bare sbatch
  if script.cores
    args.concat ["-n", script.cores.to_s]
  elsif !size_in_native?(script.native)
    args.concat ["-n", "1"]
  end
  unless script.start_time.nil?
    # Relative, so it's right even if OnDemand and Flux disagree on timezone
    offset = [(script.start_time.to_time - Time.now).round, 0].max
    args.concat ["--begin-time", "+#{offset}s"]
  end

  args.concat dependency_args("afterstart", after)
  args.concat dependency_args("afterok", afterok)
  args.concat dependency_args("afternotok", afternotok)
  args.concat dependency_args("afterany", afterany)

  args.concat env_args(script.job_environment || {}, script.copy_environment?)

  args.concat script.native if script.native

  @flux.submit_string(script_content(script), args: args)
rescue Batch::Error => e
  raise JobAdapterError, e.message
end

#supports_job_arrays? ⇒ Boolean

Whether the adapter supports job arrays

Returns:

  • (Boolean) —
    • false


223
224
225
 
# File 'lib/ood_core/job/adapters/flux.rb', line 223

def supports_job_arrays?
  false
end

#supports_job_dependencies? ⇒ Boolean

Whether the adapter supports job dependencies

Returns:

  • (Boolean) —
    • true


229
230
231
 
# File 'lib/ood_core/job/adapters/flux.rb', line 229

def supports_job_dependencies?
  true
end