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
- #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_script(targets, script, arguments, options = {}) ⇒ Object
- #run_task(targets, task, arguments, 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.
19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 |
# File 'lib/bolt/executor.rb', line 19 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.
16 17 18 |
# File 'lib/bolt/executor.rb', line 16 def noop @noop end |
#run_as ⇒ Object
Returns the value of attribute run_as.
17 18 19 |
# File 'lib/bolt/executor.rb', line 17 def run_as @run_as end |
#transports ⇒ Object (readonly)
Returns the value of attribute transports.
16 17 18 |
# File 'lib/bolt/executor.rb', line 16 def transports @transports end |
Instance Method Details
#await_results(promises) ⇒ Object
Create a ResultSet from the results of all promises.
131 132 133 |
# File 'lib/bolt/executor.rb', line 131 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.
142 143 144 145 |
# File 'lib/bolt/executor.rb', line 142 def batch_execute(targets, &block) promises = queue_execute(targets, &block) await_results(promises) end |
#finish_plan(plan_result) ⇒ Object
327 328 329 |
# File 'lib/bolt/executor.rb', line 327 def finish_plan(plan_result) transport('pcp').finish_plan(plan_result) end |
#log_action(description, targets) ⇒ Object
147 148 149 150 151 152 153 154 155 156 157 |
# File 'lib/bolt/executor.rb', line 147 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
159 160 161 162 163 164 165 166 167 168 169 170 171 172 |
# File 'lib/bolt/executor.rb', line 159 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 |
#publish_event(event) ⇒ Object
66 67 68 69 70 71 72 73 74 75 |
# File 'lib/bolt/executor.rb', line 66 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.
88 89 90 91 92 93 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 |
# File 'lib/bolt/executor.rb', line 88 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
190 191 192 193 194 195 196 197 198 199 200 |
# File 'lib/bolt/executor.rb', line 190 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
186 187 188 |
# File 'lib/bolt/executor.rb', line 186 def report_bundled_content(mode, name) @analytics.report_bundled_content(mode, name) end |
#report_function_call(function) ⇒ Object
182 183 184 |
# File 'lib/bolt/executor.rb', line 182 def report_function_call(function) @analytics&.event('Plan', 'call_function', label: function) end |
#report_transport(transport, count) ⇒ Object
174 175 176 177 178 179 180 |
# File 'lib/bolt/executor.rb', line 174 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
202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 |
# File 'lib/bolt/executor.rb', line 202 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
225 226 227 228 229 230 231 232 233 234 235 236 |
# File 'lib/bolt/executor.rb', line 225 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_script(targets, script, arguments, options = {}) ⇒ Object
238 239 240 241 242 243 244 245 246 247 248 249 |
# File 'lib/bolt/executor.rb', line 238 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
251 252 253 254 255 256 257 258 259 260 261 262 263 |
# File 'lib/bolt/executor.rb', line 251 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 |
#shutdown ⇒ Object
77 78 79 80 |
# File 'lib/bolt/executor.rb', line 77 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.
323 324 325 |
# File 'lib/bolt/executor.rb', line 323 def start_plan(plan_context) transport('pcp').plan_context = plan_context end |
#subscribe(subscriber, types = nil) ⇒ Object
61 62 63 64 |
# File 'lib/bolt/executor.rb', line 61 def subscribe(subscriber, types = nil) @subscribers[subscriber] = types self end |
#transport(transport) ⇒ Object
53 54 55 56 57 58 59 |
# File 'lib/bolt/executor.rb', line 53 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
265 266 267 268 269 270 271 272 273 274 275 276 |
# File 'lib/bolt/executor.rb', line 265 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
305 306 307 308 309 310 311 |
# File 'lib/bolt/executor.rb', line 305 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
280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 |
# File 'lib/bolt/executor.rb', line 280 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) } rescue TimeoutError => e available, unavailable = batch.partition { |target| transport.batch_connected?([target]) } ( available.map { |target| Result.new(target) } + unavailable.map { |target| Result.from_exception(target, e) } ) end end end end |
#with_node_logging(description, batch) ⇒ Object
218 219 220 221 222 223 |
# File 'lib/bolt/executor.rb', line 218 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
331 332 333 334 335 336 |
# File 'lib/bolt/executor.rb', line 331 def without_default_logging publish_event(type: :disable_default_output) yield ensure publish_event(type: :enable_default_output) end |