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”.
- #finish_plan(plan_result) ⇒ Object
-
#initialize(concurrency = 1, analytics = Bolt::Analytics::NoopClient.new, noop = 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) ⇒ 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 |
# File 'lib/bolt/executor.rb', line 35 def initialize(concurrency = 1, analytics = Bolt::Analytics::NoopClient.new, noop = false) # lazy-load expensive gem code require 'concurrent' @analytics = analytics @logger = Logging.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 @noop = noop @run_as = nil @pool = if concurrency > 0 Concurrent::ThreadPoolExecutor.new(max_threads: concurrency) else Concurrent.global_immediate_executor end @logger.debug { "Started with #{concurrency} max thread(s)" } 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.
147 148 149 |
# File 'lib/bolt/executor.rb', line 147 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.
158 159 160 161 |
# File 'lib/bolt/executor.rb', line 158 def batch_execute(targets, &block) promises = queue_execute(targets, &block) await_results(promises) end |
#finish_plan(plan_result) ⇒ Object
381 382 383 |
# File 'lib/bolt/executor.rb', line 381 def finish_plan(plan_result) transport('pcp').finish_plan(plan_result) end |
#log_action(description, targets) ⇒ Object
163 164 165 166 167 168 169 170 171 172 173 |
# File 'lib/bolt/executor.rb', line 163 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
175 176 177 178 179 180 181 182 183 184 185 186 187 188 |
# File 'lib/bolt/executor.rb', line 175 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
349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 |
# File 'lib/bolt/executor.rb', line 349 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
82 83 84 85 86 87 88 89 90 91 |
# File 'lib/bolt/executor.rb', line 82 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.
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 136 137 138 139 140 141 142 143 144 |
# File 'lib/bolt/executor.rb', line 104 def queue_execute(targets) 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| 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
206 207 208 209 210 211 212 213 214 215 216 |
# File 'lib/bolt/executor.rb', line 206 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
202 203 204 |
# File 'lib/bolt/executor.rb', line 202 def report_bundled_content(mode, name) @analytics.report_bundled_content(mode, name) end |
#report_function_call(function) ⇒ Object
198 199 200 |
# File 'lib/bolt/executor.rb', line 198 def report_function_call(function) @analytics&.event('Plan', 'call_function', label: function) end |
#report_transport(transport, count) ⇒ Object
190 191 192 193 194 195 196 |
# File 'lib/bolt/executor.rb', line 190 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
218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 |
# File 'lib/bolt/executor.rb', line 218 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.debug { "Failed to submit analytics event: #{e.}" } end |
#run_command(targets, command, options = {}) ⇒ Object
241 242 243 244 245 246 247 248 249 250 251 252 |
# File 'lib/bolt/executor.rb', line 241 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
310 311 312 |
# File 'lib/bolt/executor.rb', line 310 def run_plan(scope, plan, params) plan.call_by_name_with_scope(scope, params, true) end |
#run_script(targets, script, arguments, options = {}) ⇒ Object
254 255 256 257 258 259 260 261 262 263 264 265 |
# File 'lib/bolt/executor.rb', line 254 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
267 268 269 270 271 272 273 274 275 276 277 278 279 |
# File 'lib/bolt/executor.rb', line 267 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
281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 |
# File 'lib/bolt/executor.rb', line 281 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
93 94 95 96 |
# File 'lib/bolt/executor.rb', line 93 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.
377 378 379 |
# File 'lib/bolt/executor.rb', line 377 def start_plan(plan_context) transport('pcp').plan_context = plan_context end |
#subscribe(subscriber, types = nil) ⇒ Object
77 78 79 80 |
# File 'lib/bolt/executor.rb', line 77 def subscribe(subscriber, types = nil) @subscribers[subscriber] = types self end |
#transport(transport) ⇒ Object
69 70 71 72 73 74 75 |
# File 'lib/bolt/executor.rb', line 69 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
297 298 299 300 301 302 303 304 305 306 307 308 |
# File 'lib/bolt/executor.rb', line 297 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
341 342 343 344 345 346 347 |
# File 'lib/bolt/executor.rb', line 341 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
316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 |
# File 'lib/bolt/executor.rb', line 316 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
234 235 236 237 238 239 |
# File 'lib/bolt/executor.rb', line 234 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
385 386 387 388 389 390 |
# File 'lib/bolt/executor.rb', line 385 def without_default_logging publish_event(type: :disable_default_output) yield ensure publish_event(type: :enable_default_output) end |