Class: Bolt::Executor
- Inherits:
-
Object
- Object
- Bolt::Executor
- Defined in:
- lib/bolt/executor.rb
Defined Under Namespace
Classes: TimeoutError
Instance Attribute Summary collapse
-
#future ⇒ Object
readonly
Returns the value of attribute future.
-
#in_parallel ⇒ Object
readonly
Returns the value of attribute in_parallel.
-
#noop ⇒ Object
readonly
Returns the value of attribute noop.
-
#run_as ⇒ Object
Returns the value of attribute run_as.
-
#transports ⇒ Object
readonly
Returns the value of attribute transports.
Instance Method Summary collapse
-
#await_results(promises) ⇒ Object
Create a ResultSet from the results of all promises.
-
#batch_execute(targets, &block) ⇒ Object
Execute the given block on a list of nodes in parallel, one thread per “batch”.
- #create_yarn(scope, block, object, index) ⇒ Object
- #download_file(targets, source, destination, options = {}, position = []) ⇒ Object
- #finish_plan(plan_result) ⇒ Object
- #handle_event(event) ⇒ Object
-
#initialize(concurrency = 1, analytics = Bolt::Analytics::NoopClient.new, noop = false, modified_concurrency = false, future = {}) ⇒ Executor
constructor
A new instance of Executor.
- #log_action(description, targets) ⇒ Object
- #log_plan(plan_name) ⇒ Object
- #prompt(prompt, options) ⇒ Object
- #publish_event(event) ⇒ Object
-
#queue_execute(targets) ⇒ Object
Starts executing the given block on a list of nodes in parallel, one thread per “batch”.
- #report_apply(statement_count, resource_counts) ⇒ Object
- #report_bundled_content(mode, name) ⇒ Object
- #report_file_source(plan_function, source) ⇒ Object
- #report_function_call(function) ⇒ Object
- #report_yaml_plan(plan) ⇒ Object
- #round_robin(skein) ⇒ Object
- #run_command(targets, command, options = {}, position = []) ⇒ Object
- #run_plan(scope, plan, params) ⇒ Object
- #run_script(targets, script, arguments, options = {}, position = []) ⇒ Object
- #run_task(targets, task, arguments, options = {}, position = [], log_level = :info) ⇒ Object
- #run_task_with(target_mapping, task, options = {}, position = []) ⇒ Object
- #shutdown ⇒ Object
-
#start_plan(plan_context) ⇒ Object
Plan context doesn’t make sense for most transports but it is tightly coupled with the orchestrator transport since the transport behaves differently when a plan is running.
- #subscribe(subscriber, types = nil) ⇒ Object
- #transport(transport) ⇒ Object
- #unsubscribe(subscriber, types = nil) ⇒ Object
- #upload_file(targets, source, destination, options = {}, position = []) ⇒ Object
- #wait_until_available(targets, description: 'wait until available', wait_time: 120, retry_interval: 1) ⇒ Object
- #with_node_logging(description, batch, log_level = :info) ⇒ Object
- #without_default_logging ⇒ Object
Constructor Details
#initialize(concurrency = 1, analytics = Bolt::Analytics::NoopClient.new, noop = false, modified_concurrency = false, future = {}) ⇒ Executor
Returns a new instance of Executor.
41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 |
# File 'lib/bolt/executor.rb', line 41 def initialize(concurrency = 1, analytics = Bolt::Analytics::NoopClient.new, noop = false, modified_concurrency = false, future = {}) # lazy-load expensive gem code require 'concurrent' @analytics = analytics @logger = Bolt::Logger.logger(self) @transports = Bolt::TRANSPORTS.each_with_object({}) do |(key, val), coll| coll[key.to_s] = if key == :remote Concurrent::Delay.new do val.new(self) end else Concurrent::Delay.new do val.new end end end @reported_transports = Set.new @subscribers = {} @publisher = Concurrent::SingleThreadExecutor.new @publisher.post { Thread.current[:name] = 'event-publisher' } @noop = noop @run_as = nil @in_parallel = false @future = future @pool = if concurrency > 0 Concurrent::ThreadPoolExecutor.new(name: 'exec', max_threads: concurrency) else Concurrent.global_immediate_executor end @logger.debug { "Started with #{concurrency} max thread(s)" } @concurrency = concurrency @warn_concurrency = modified_concurrency end |
Instance Attribute Details
#future ⇒ Object (readonly)
Returns the value of attribute future.
38 39 40 |
# File 'lib/bolt/executor.rb', line 38 def future @future end |
#in_parallel ⇒ Object (readonly)
Returns the value of attribute in_parallel.
38 39 40 |
# File 'lib/bolt/executor.rb', line 38 def in_parallel @in_parallel end |
#noop ⇒ Object (readonly)
Returns the value of attribute noop.
38 39 40 |
# File 'lib/bolt/executor.rb', line 38 def noop @noop end |
#run_as ⇒ Object
Returns the value of attribute run_as.
39 40 41 |
# File 'lib/bolt/executor.rb', line 39 def run_as @run_as end |
#transports ⇒ Object (readonly)
Returns the value of attribute transports.
38 39 40 |
# File 'lib/bolt/executor.rb', line 38 def transports @transports end |
Instance Method Details
#await_results(promises) ⇒ Object
Create a ResultSet from the results of all promises.
181 182 183 |
# File 'lib/bolt/executor.rb', line 181 def await_results(promises) ResultSet.new(promises.map(&:value)) end |
#batch_execute(targets, &block) ⇒ Object
Execute the given block on a list of nodes in parallel, one thread per “batch”.
This is the main driver of execution on a list of targets. It first groups targets by transport, then divides each group into batches as defined by the transport. Each batch, along with the corresponding transport, is yielded to the block in turn and the results all collected into a single ResultSet.
192 193 194 195 |
# File 'lib/bolt/executor.rb', line 192 def batch_execute(targets, &block) promises = queue_execute(targets, &block) await_results(promises) end |
#create_yarn(scope, block, object, index) ⇒ Object
376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 |
# File 'lib/bolt/executor.rb', line 376 def create_yarn(scope, block, object, index) fiber = Fiber.new do # Create the new scope newscope = Puppet::Parser::Scope.new(scope.compiler) local = Puppet::Parser::Scope::LocalScope.new # Compress the current scopes into a single vars hash to add to the new scope current_scope = scope.effective_symtable(true) until current_scope.nil? current_scope.instance_variable_get(:@symbols)&.each_pair { |k, v| local[k] = v } current_scope = current_scope.parent end newscope.push_ephemerals([local]) begin result = catch(:return) do args = { block.parameters[0][1].to_s => object } block.closure.call_by_name_with_scope(newscope, args, true) end # If we got a return from the block, get it's value # Otherwise the result is the last line from the block result = result.value if result.is_a?(Puppet::Pops::Evaluator::Return) # Validate the result is a PlanResult unless Puppet::Pops::Types::TypeParser.singleton.parse('Boltlib::PlanResult').instance?(result) raise Bolt::InvalidParallelResult.new(result.to_s, *Puppet::Pops::PuppetStack.top_of_stack) end result rescue Puppet::PreformattedError => e if e.cause.is_a?(Bolt::Error) e.cause else raise e end end end Bolt::Yarn.new(fiber, index) end |
#download_file(targets, source, destination, options = {}, position = []) ⇒ Object
351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 |
# File 'lib/bolt/executor.rb', line 351 def download_file(targets, source, destination, = {}, position = []) description = .fetch(:description, "file download from #{source} to #{destination}") begin FileUtils.mkdir_p(destination) rescue Errno::EEXIST => e = "#{e.}; unable to create destination directory #{destination}" raise Bolt::Error.new(, 'bolt/file-exist-error') end log_action(description, targets) do [:run_as] = run_as if run_as && !.key?(:run_as) batch_execute(targets) do |transport, batch| with_node_logging("Downloading file #{source} to #{destination}", batch) do transport.batch_download(batch, source, destination, , position, &method(:publish_event)) end end end end |
#finish_plan(plan_result) ⇒ Object
532 533 534 |
# File 'lib/bolt/executor.rb', line 532 def finish_plan(plan_result) transport('pcp').finish_plan(plan_result) end |
#handle_event(event) ⇒ Object
418 419 420 421 422 423 |
# File 'lib/bolt/executor.rb', line 418 def handle_event(event) case event[:type] when :node_result @thread_completed = true end end |
#log_action(description, targets) ⇒ Object
197 198 199 200 201 202 203 204 205 206 207 |
# File 'lib/bolt/executor.rb', line 197 def log_action(description, targets) publish_event(type: :step_start, description: description, targets: targets) start_time = Time.now results = yield duration = Time.now - start_time publish_event(type: :step_finish, description: description, result: results, duration: duration) results end |
#log_plan(plan_name) ⇒ Object
209 210 211 212 213 214 215 216 217 218 219 220 221 222 |
# File 'lib/bolt/executor.rb', line 209 def log_plan(plan_name) publish_event(type: :plan_start, plan: plan_name) start_time = Time.now results = nil begin results = yield ensure duration = Time.now - start_time publish_event(type: :plan_finish, plan: plan_name, duration: duration) end results end |
#prompt(prompt, options) ⇒ Object
490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 |
# File 'lib/bolt/executor.rb', line 490 def prompt(prompt, ) unless $stdin.tty? return [:default] if [:default] raise Bolt::Error.new('STDIN is not a tty, unable to prompt', 'bolt/no-tty-error') end @prompting = true if [:default] && ![:sensitive] $stderr.print("#{prompt} [#{[:default]}]: ") else $stderr.print("#{prompt}: ") end value = if [:sensitive] $stdin.noecho(&:gets).to_s.chomp else $stdin.gets.to_s.chomp end @prompting = false $stderr.puts if [:sensitive] value = [:default] if value.empty? value end |
#publish_event(event) ⇒ Object
103 104 105 106 107 108 109 110 111 112 113 114 |
# File 'lib/bolt/executor.rb', line 103 def publish_event(event) @subscribers.each do |subscriber, types| # If types isn't set or if the subscriber is subscribed to # that type of event, publish the event next unless types.nil? || types.include?(event[:type]) @publisher.post(subscriber) do |sub| # Wait for user to input to prompt before printing anything sleep(0.1) while @prompting sub.handle_event(event) end end end |
#queue_execute(targets) ⇒ Object
Starts executing the given block on a list of nodes in parallel, one thread per “batch”.
This is the main driver of execution on a list of targets. It first groups targets by transport, then divides each group into batches as defined by the transport. Yields each batch, along with the corresponding transport, to the block in turn and returns an array of result promises.
127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 |
# File 'lib/bolt/executor.rb', line 127 def queue_execute(targets) if @warn_concurrency && targets.length > @concurrency @warn_concurrency = false msg = "The ulimit is low, which might cause file limit issues. Default concurrency has been set to "\ "'#{@concurrency}' to mitigate those issues, which might cause Bolt to run slow. "\ "Disable this warning by configuring ulimit using 'ulimit -n <limit>' in your shell "\ "configuration, or by configuring Bolt's concurrency. "\ "See https://puppet.com/docs/bolt/latest/bolt_known_issues.html for details." Bolt::Logger.warn("low_ulimit", msg) end targets.group_by(&:transport).flat_map do |protocol, protocol_targets| transport = transport(protocol) report_transport(transport, protocol_targets.count) transport.batches(protocol_targets).flat_map do |batch| batch_promises = Array(batch).each_with_object({}) do |target, h| h[target] = Concurrent::Promise.new(executor: :immediate) end # Pass this argument through to avoid retaining a reference to a # local variable that will change on the next iteration of the loop. @pool.post(batch_promises) do |result_promises| Thread.current[:name] ||= Thread.current.name results = yield transport, batch Array(results).each do |result| result_promises[result.target].set(result) end # NotImplementedError can be thrown if the transport is not implemented improperly rescue StandardError, NotImplementedError => e result_promises.each do |target, promise| # If an exception happens while running, the result won't be logged # by the CLI. Log a warning, as this is probably a problem with the transport. # If batch_* commands are used from the Base transport, then exceptions # normally shouldn't reach here. @logger.warn(e) promise.set(Bolt::Result.from_exception(target, e)) end ensure # Make absolutely sure every promise gets a result to avoid a # deadlock. Use whatever exception is causing this block to # execute, or generate one if we somehow got here without an # exception and some promise is still missing a result. result_promises.each do |target, promise| next if promise.fulfilled? error = $ERROR_INFO || Bolt::Error.new("No result was returned for #{target.uri}", "puppetlabs.bolt/missing-result-error") promise.set(Bolt::Result.from_exception(target, error)) end end batch_promises.values end end end |
#report_apply(statement_count, resource_counts) ⇒ Object
245 246 247 248 249 250 251 252 253 254 255 |
# File 'lib/bolt/executor.rb', line 245 def report_apply(statement_count, resource_counts) data = { statement_count: statement_count } unless resource_counts.empty? sum = resource_counts.inject(0) { |accum, i| accum + i } # Intentionally rounded to an integer. High precision isn't useful. data[:resource_mean] = sum / resource_counts.length end @analytics&.event('Apply', 'ast', **data) end |
#report_bundled_content(mode, name) ⇒ Object
236 237 238 |
# File 'lib/bolt/executor.rb', line 236 def report_bundled_content(mode, name) @analytics.report_bundled_content(mode, name) end |
#report_file_source(plan_function, source) ⇒ Object
240 241 242 243 |
# File 'lib/bolt/executor.rb', line 240 def report_file_source(plan_function, source) label = Pathname.new(source).absolute? ? 'absolute' : 'module' @analytics&.event('Plan', plan_function, label: label) end |
#report_function_call(function) ⇒ Object
232 233 234 |
# File 'lib/bolt/executor.rb', line 232 def report_function_call(function) @analytics&.event('Plan', 'call_function', label: function) end |
#report_yaml_plan(plan) ⇒ Object
257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 |
# File 'lib/bolt/executor.rb', line 257 def report_yaml_plan(plan) steps = plan.steps.count return_type = case plan.return when Bolt::PAL::YamlPlan::EvaluableString 'expression' when nil nil else 'value' end @analytics&.event('Plan', 'yaml', plan_steps: steps, return_type: return_type) rescue StandardError => e @logger.trace { "Failed to submit analytics event: #{e.}" } end |
#round_robin(skein) ⇒ Object
425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 |
# File 'lib/bolt/executor.rb', line 425 def round_robin(skein) subscribe(self, [:node_result]) results = Array.new(skein.length) @in_parallel = true publish_event(type: :stop_spin) until skein.empty? @thread_completed = false r = nil skein.each do |yarn| if yarn.alive? publish_event(type: :stop_spin) r = yarn.resume else results[yarn.index] = yarn.value skein.delete(yarn) end end next unless r == 'unfinished' sleep(0.1) until @thread_completed || skein.empty? end publish_event(type: :stop_spin) @in_parallel = false unsubscribe(self, [:node_result]) results end |
#run_command(targets, command, options = {}, position = []) ⇒ Object
282 283 284 285 286 287 288 289 290 291 292 293 |
# File 'lib/bolt/executor.rb', line 282 def run_command(targets, command, = {}, position = []) description = .fetch(:description, "command '#{command}'") log_action(description, targets) do [:run_as] = run_as if run_as && !.key?(:run_as) batch_execute(targets) do |transport, batch| with_node_logging("Running command '#{command}'", batch) do transport.batch_command(batch, command, , position, &method(:publish_event)) end end end end |
#run_plan(scope, plan, params) ⇒ Object
372 373 374 |
# File 'lib/bolt/executor.rb', line 372 def run_plan(scope, plan, params) plan.call_by_name_with_scope(scope, params, true) end |
#run_script(targets, script, arguments, options = {}, position = []) ⇒ Object
295 296 297 298 299 300 301 302 303 304 305 306 |
# File 'lib/bolt/executor.rb', line 295 def run_script(targets, script, arguments, = {}, position = []) description = .fetch(:description, "script #{script}") log_action(description, targets) do [:run_as] = run_as if run_as && !.key?(:run_as) batch_execute(targets) do |transport, batch| with_node_logging("Running script #{script} with '#{arguments.to_json}'", batch) do transport.batch_script(batch, script, arguments, , position, &method(:publish_event)) end end end end |
#run_task(targets, task, arguments, options = {}, position = [], log_level = :info) ⇒ Object
308 309 310 311 312 313 314 315 316 317 318 319 320 |
# File 'lib/bolt/executor.rb', line 308 def run_task(targets, task, arguments, = {}, position = [], log_level = :info) description = .fetch(:description, "task #{task.name}") log_action(description, targets) do [:run_as] = run_as if run_as && !.key?(:run_as) arguments['_task'] = task.name batch_execute(targets) do |transport, batch| with_node_logging("Running task #{task.name} with '#{arguments.to_json}'", batch, log_level) do transport.batch_task(batch, task, arguments, , position, &method(:publish_event)) end end end end |
#run_task_with(target_mapping, task, options = {}, position = []) ⇒ Object
322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 |
# File 'lib/bolt/executor.rb', line 322 def run_task_with(target_mapping, task, = {}, position = []) targets = target_mapping.keys description = .fetch(:description, "task #{task.name}") log_action(description, targets) do [:run_as] = run_as if run_as && !.key?(:run_as) target_mapping.each_value { |arguments| arguments['_task'] = task.name } batch_execute(targets) do |transport, batch| with_node_logging("Running task #{task.name}'", batch) do transport.batch_task_with(batch, task, target_mapping, , position, &method(:publish_event)) end end end end |
#shutdown ⇒ Object
116 117 118 119 |
# File 'lib/bolt/executor.rb', line 116 def shutdown @publisher.shutdown @publisher.wait_for_termination end |
#start_plan(plan_context) ⇒ Object
Plan context doesn’t make sense for most transports but it is tightly coupled with the orchestrator transport since the transport behaves differently when a plan is running. In order to limit how much this pollutes the transport API we only handle the orchestrator transport here. Since we callt this function without resolving targets this will result in the orchestrator transport always being initialized during plan runs. For now that’s ok.
In the future if other transports need this or if we want a plan stack we’ll need to refactor.
528 529 530 |
# File 'lib/bolt/executor.rb', line 528 def start_plan(plan_context) transport('pcp').plan_context = plan_context end |
#subscribe(subscriber, types = nil) ⇒ Object
90 91 92 93 |
# File 'lib/bolt/executor.rb', line 90 def subscribe(subscriber, types = nil) @subscribers[subscriber] = types self end |
#transport(transport) ⇒ Object
82 83 84 85 86 87 88 |
# File 'lib/bolt/executor.rb', line 82 def transport(transport) impl = @transports[transport || 'ssh'] raise(Bolt::UnknownTransportError, transport) unless impl # If there was an error creating the transport, ensure it gets thrown impl.no_error! impl.value end |
#unsubscribe(subscriber, types = nil) ⇒ Object
95 96 97 98 99 100 101 |
# File 'lib/bolt/executor.rb', line 95 def unsubscribe(subscriber, types = nil) if types.nil? || types.sort == @subscribers[subscriber]&.sort @subscribers.delete(subscriber) elsif @subscribers[subscriber].is_a?(Array) @subscribers[subscriber] = @subscribers[subscriber] - types end end |
#upload_file(targets, source, destination, options = {}, position = []) ⇒ Object
338 339 340 341 342 343 344 345 346 347 348 349 |
# File 'lib/bolt/executor.rb', line 338 def upload_file(targets, source, destination, = {}, position = []) description = .fetch(:description, "file upload from #{source} to #{destination}") log_action(description, targets) do [:run_as] = run_as if run_as && !.key?(:run_as) batch_execute(targets) do |transport, batch| with_node_logging("Uploading file #{source} to #{destination}", batch) do transport.batch_upload(batch, source, destination, , position, &method(:publish_event)) end end end end |
#wait_until_available(targets, description: 'wait until available', wait_time: 120, retry_interval: 1) ⇒ Object
457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 |
# File 'lib/bolt/executor.rb', line 457 def wait_until_available(targets, description: 'wait until available', wait_time: 120, retry_interval: 1) log_action(description, targets) do batch_execute(targets) do |transport, batch| with_node_logging('Waiting until available', batch) do wait_until(wait_time, retry_interval) { transport.batch_connected?(batch) } batch.map { |target| Result.new(target, action: 'wait_until_available', object: description) } rescue TimeoutError => e available, unavailable = batch.partition { |target| transport.batch_connected?([target]) } ( available.map { |target| Result.new(target, action: 'wait_until_available', object: description) } + unavailable.map { |target| Result.from_exception(target, e, action: 'wait_until_available') } ) end end end end |
#with_node_logging(description, batch, log_level = :info) ⇒ Object
273 274 275 276 277 278 279 280 |
# File 'lib/bolt/executor.rb', line 273 def with_node_logging(description, batch, log_level = :info) @logger.send(log_level, "#{description} on #{batch.map(&:safe_name)}") publish_event(type: :start_spin) result = yield publish_event(type: :stop_spin) @logger.send(log_level, result.to_json) result end |
#without_default_logging ⇒ Object
536 537 538 539 540 541 |
# File 'lib/bolt/executor.rb', line 536 def without_default_logging publish_event(type: :disable_default_output) yield ensure publish_event(type: :enable_default_output) end |