Class: Bolt::Executor

Inherits:
Object
  • Object
show all
Defined in:
lib/bolt/executor.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config = Bolt::Config.new, noop = nil, plan_logging = 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
 
# File 'lib/bolt/executor.rb', line 19

def initialize(config = Bolt::Config.new, noop = nil, plan_logging = false)
  @config = config
  @logger = Logging.logger[self]

  @transports = {
    'ssh' => Concurrent::Delay.new { Bolt::Transport::SSH.new(config[:transports][:ssh] || {}) },
    'winrm' => Concurrent::Delay.new { Bolt::Transport::WinRM.new(config[:transports][:winrm] || {}) },
    'pcp' => Concurrent::Delay.new { Bolt::Transport::Orch.new(config[:transports][:pcp] || {}) }
  }

  # If a specific elevated log level has been requested, honor that.
  # Otherwise, escalate the log level to "info" if running in plan mode, so
  # that certain progress messages will be visible.
  default_log_level = plan_logging ? :info : :notice
  @logger.level = @config[:log_level] || default_log_level
  @noop = noop
  @run_as = nil
  @pool = Concurrent::CachedThreadPool.new(max_threads: @config[:concurrency])
  @logger.debug { "Started with #{@config[:concurrency]} max thread(s)" }
  @notifier = Bolt::Notifier.new
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

#batch_execute(targets) ⇒ 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.



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/bolt/executor.rb', line 60

def batch_execute(targets)
  promises = targets.group_by(&:protocol).flat_map do |protocol, _protocol_targets|
    transport = transport(protocol)
    transport.batches(targets).flat_map do |batch|
      batch_promises = Hash[Array(batch).map { |target| [target, Concurrent::Promise.new(executor: :immediate)] }]
      # 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|
        begin
          results = yield transport, batch
          Array(results).each do |result|
            result_promises[result.target].set(result)
          end
        # NotImplementedError can be thrown if the transport is implemented improperly
        rescue StandardError, NotImplementedError => e
          result_promises.each do |target, promise|
            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(error))
          end
        end
      end
      batch_promises.values
    end
  end
  ResultSet.new(promises.map(&:value))
end

#file_upload(targets, source, destination, options = {}, &callback) ⇒ Object 



142
143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/bolt/executor.rb', line 142

def file_upload(targets, source, destination, options = {}, &callback)
  @logger.info("Starting file upload from #{source} to #{destination} on #{targets.map(&:uri)}")
  notify = proc { |event| @notifier.notify(callback, event) if callback }
  options = { '_run_as' => run_as }.merge(options) if run_as

  results = batch_execute(targets) do |transport, batch|
    transport.batch_upload(batch, source, destination, options, &notify)
  end

  @logger.info(summary('upload', source, results))
  @notifier.shutdown
  results
end

#run_command(targets, command, options = {}, &callback) ⇒ Object 



97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/bolt/executor.rb', line 97

def run_command(targets, command, options = {}, &callback)
  @logger.info("Starting command run '#{command}' on #{targets.map(&:uri)}")
  notify = proc { |event| @notifier.notify(callback, event) if callback }
  options = { '_run_as' => run_as }.merge(options) if run_as

  results = batch_execute(targets) do |transport, batch|
    transport.batch_command(batch, command, options, &notify)
  end

  @logger.info(summary('command', command, results))
  @notifier.shutdown
  results
end

#run_script(targets, script, arguments, options = {}, &callback) ⇒ Object 



111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/bolt/executor.rb', line 111

def run_script(targets, script, arguments, options = {}, &callback)
  @logger.info("Starting script run #{script} on #{targets.map(&:uri)}")
  @logger.debug("Arguments: #{arguments}")
  notify = proc { |event| @notifier.notify(callback, event) if callback }
  options = { '_run_as' => run_as }.merge(options) if run_as

  results = batch_execute(targets) do |transport, batch|
    transport.batch_script(batch, script, arguments, options, &notify)
  end

  @logger.info(summary('script', script, results))
  @notifier.shutdown
  results
end

#run_task(targets, task, arguments, options = {}, &callback) ⇒ Object 



126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/bolt/executor.rb', line 126

def run_task(targets, task, arguments, options = {}, &callback)
  task_name = task.name
  @logger.info("Starting task #{task_name} on #{targets.map(&:uri)}")
  @logger.debug("Arguments: #{arguments} Input method: #{task.input_method}")
  notify = proc { |event| @notifier.notify(callback, event) if callback }
  options = { '_run_as' => run_as }.merge(options) if run_as

  results = batch_execute(targets) do |transport, batch|
    transport.batch_task(batch, task, arguments, options, &notify)
  end

  @logger.info(summary('task', task_name, results))
  @notifier.shutdown
  results
end

#transport(transport) ⇒ Object 



41
42
43
 
# File 'lib/bolt/executor.rb', line 41

def transport(transport)
  @transports[transport || 'ssh'].value
end