Class: Bolt::Executor
- Inherits:
-
Object
- Object
- Bolt::Executor
- Defined in:
- lib/bolt/executor.rb
Defined Under Namespace
Classes: TimeoutError
Instance Attribute Summary collapse
-
#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”.
- #download_file(targets, source, destination, options = {}) ⇒ Object
- #finish_plan(plan_result) ⇒ Object
-
#initialize(concurrency = 1, analytics = Bolt::Analytics::NoopClient.new, noop = false, modified_concurrency = false) ⇒ 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_function_call(function) ⇒ Object
- #report_transport(transport, count) ⇒ Object
- #report_yaml_plan(plan) ⇒ Object
- #run_command(targets, command, options = {}) ⇒ Object
- #run_plan(scope, plan, params) ⇒ Object
- #run_script(targets, script, arguments, options = {}) ⇒ Object
- #run_task(targets, task, arguments, options = {}) ⇒ Object
- #run_task_with(target_mapping, task, options = {}) ⇒ 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
- #upload_file(targets, source, destination, options = {}) ⇒ Object
- #wait_until(timeout, retry_interval) ⇒ Object
- #wait_until_available(targets, description: 'wait until available', wait_time: 120, retry_interval: 1) ⇒ Object
- #with_node_logging(description, batch) ⇒ Object
- #without_default_logging ⇒ Object
Constructor Details
#initialize(concurrency = 1, analytics = Bolt::Analytics::NoopClient.new, noop = false, modified_concurrency = false) ⇒ Executor
Returns a new instance of Executor.
35 36 37 38 39 40 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 |
# File 'lib/bolt/executor.rb', line 35 def initialize(concurrency = 1, analytics = Bolt::Analytics::NoopClient.new, noop = false, modified_concurrency = false) # 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 @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
#noop ⇒ Object (readonly)
Returns the value of attribute noop.
32 33 34 |
# File 'lib/bolt/executor.rb', line 32 def noop @noop end |
#run_as ⇒ Object
Returns the value of attribute run_as.
33 34 35 |
# File 'lib/bolt/executor.rb', line 33 def run_as @run_as end |
#transports ⇒ Object (readonly)
Returns the value of attribute transports.
32 33 34 |
# File 'lib/bolt/executor.rb', line 32 def transports @transports end |
Instance Method Details
#await_results(promises) ⇒ Object
Create a ResultSet from the results of all promises.
162 163 164 |
# File 'lib/bolt/executor.rb', line 162 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.
173 174 175 176 |
# File 'lib/bolt/executor.rb', line 173 def batch_execute(targets, &block) promises = queue_execute(targets, &block) await_results(promises) end |
#download_file(targets, source, destination, options = {}) ⇒ Object
325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 |
# File 'lib/bolt/executor.rb', line 325 def download_file(targets, source, destination, = {}) 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, , &method(:publish_event)) end end end end |
#finish_plan(plan_result) ⇒ Object
417 418 419 |
# File 'lib/bolt/executor.rb', line 417 def finish_plan(plan_result) transport('pcp').finish_plan(plan_result) end |
#log_action(description, targets) ⇒ Object
178 179 180 181 182 183 184 185 186 187 188 |
# File 'lib/bolt/executor.rb', line 178 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
190 191 192 193 194 195 196 197 198 199 200 201 202 203 |
# File 'lib/bolt/executor.rb', line 190 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
385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 |
# File 'lib/bolt/executor.rb', line 385 def prompt(prompt, ) unless $stdin.tty? raise Bolt::Error.new('STDIN is not a tty, unable to prompt', 'bolt/no-tty-error') end $stderr.print("#{prompt}: ") value = if [:sensitive] $stdin.noecho(&:gets).to_s.chomp else $stdin.gets.to_s.chomp end $stderr.puts if [:sensitive] value end |
#publish_event(event) ⇒ Object
87 88 89 90 91 92 93 94 95 96 |
# File 'lib/bolt/executor.rb', line 87 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| 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.
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 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 |
# File 'lib/bolt/executor.rb', line 109 def queue_execute(targets) if @warn_concurrency && targets.length > @concurrency @warn_concurrency = false @logger.warn("The ulimit is low, which may cause file limit issues. Default concurrency has been set to "\ "'#{@concurrency}' to mitigate those issues, which may 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.") 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
221 222 223 224 225 226 227 228 229 230 231 |
# File 'lib/bolt/executor.rb', line 221 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
217 218 219 |
# File 'lib/bolt/executor.rb', line 217 def report_bundled_content(mode, name) @analytics.report_bundled_content(mode, name) end |
#report_function_call(function) ⇒ Object
213 214 215 |
# File 'lib/bolt/executor.rb', line 213 def report_function_call(function) @analytics&.event('Plan', 'call_function', label: function) end |
#report_transport(transport, count) ⇒ Object
205 206 207 208 209 210 211 |
# File 'lib/bolt/executor.rb', line 205 def report_transport(transport, count) name = transport.class.name.split('::').last.downcase unless @reported_transports.include?(name) @analytics&.event('Transport', 'initialize', label: name, value: count) end @reported_transports.add(name) end |
#report_yaml_plan(plan) ⇒ Object
233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 |
# File 'lib/bolt/executor.rb', line 233 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 |
#run_command(targets, command, options = {}) ⇒ Object
256 257 258 259 260 261 262 263 264 265 266 267 |
# File 'lib/bolt/executor.rb', line 256 def run_command(targets, command, = {}) 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, , &method(:publish_event)) end end end end |
#run_plan(scope, plan, params) ⇒ Object
346 347 348 |
# File 'lib/bolt/executor.rb', line 346 def run_plan(scope, plan, params) plan.call_by_name_with_scope(scope, params, true) end |
#run_script(targets, script, arguments, options = {}) ⇒ Object
269 270 271 272 273 274 275 276 277 278 279 280 |
# File 'lib/bolt/executor.rb', line 269 def run_script(targets, script, arguments, = {}) 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, , &method(:publish_event)) end end end end |
#run_task(targets, task, arguments, options = {}) ⇒ Object
282 283 284 285 286 287 288 289 290 291 292 293 294 |
# File 'lib/bolt/executor.rb', line 282 def run_task(targets, task, arguments, = {}) 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) do transport.batch_task(batch, task, arguments, , &method(:publish_event)) end end end end |
#run_task_with(target_mapping, task, options = {}) ⇒ Object
296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 |
# File 'lib/bolt/executor.rb', line 296 def run_task_with(target_mapping, task, = {}) 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, , &method(:publish_event)) end end end end |
#shutdown ⇒ Object
98 99 100 101 |
# File 'lib/bolt/executor.rb', line 98 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.
413 414 415 |
# File 'lib/bolt/executor.rb', line 413 def start_plan(plan_context) transport('pcp').plan_context = plan_context end |
#subscribe(subscriber, types = nil) ⇒ Object
82 83 84 85 |
# File 'lib/bolt/executor.rb', line 82 def subscribe(subscriber, types = nil) @subscribers[subscriber] = types self end |
#transport(transport) ⇒ Object
74 75 76 77 78 79 80 |
# File 'lib/bolt/executor.rb', line 74 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 |
#upload_file(targets, source, destination, options = {}) ⇒ Object
312 313 314 315 316 317 318 319 320 321 322 323 |
# File 'lib/bolt/executor.rb', line 312 def upload_file(targets, source, destination, = {}) 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, , &method(:publish_event)) end end end end |
#wait_until(timeout, retry_interval) ⇒ Object
377 378 379 380 381 382 383 |
# File 'lib/bolt/executor.rb', line 377 def wait_until(timeout, retry_interval) start = wait_now until yield raise(TimeoutError, 'Timed out waiting for target') if (wait_now - start).to_i >= timeout sleep(retry_interval) end end |
#wait_until_available(targets, description: 'wait until available', wait_time: 120, retry_interval: 1) ⇒ Object
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 352 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) ⇒ Object
249 250 251 252 253 254 |
# File 'lib/bolt/executor.rb', line 249 def with_node_logging(description, batch) @logger.info("#{description} on #{batch.map(&:safe_name)}") result = yield @logger.info(result.to_json) result end |
#without_default_logging ⇒ Object
421 422 423 424 425 426 |
# File 'lib/bolt/executor.rb', line 421 def without_default_logging publish_event(type: :disable_default_output) yield ensure publish_event(type: :enable_default_output) end |