Class: Bolt::Transport::Orch

Inherits:
Base
  • Object
show all
Defined in:
lib/bolt/transport/orch.rb,
lib/bolt/transport/orch/connection.rb

Defined Under Namespace

Classes: Connection

Constant Summary collapse

CONF_FILE = 
File.expand_path('~/.puppetlabs/client-tools/orchestrator.conf')
BOLT_COMMAND_TASK = 
Struct.new(:name).new('bolt_shim::command').freeze
BOLT_SCRIPT_TASK = 
Struct.new(:name).new('bolt_shim::script').freeze
BOLT_UPLOAD_TASK = 
Struct.new(:name).new('bolt_shim::upload').freeze
PROVIDED_FEATURES = 
['puppet-agent'].freeze

Constants inherited from Base

Base::ENVIRONMENT_METHODS, Base::STDIN_METHODS

Instance Attribute Summary collapse

Attributes inherited from Base

#logger

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Base

#assert_batch_size_one, #filter_options, #run_command, #run_script, #run_task, #upload, #with_events

Constructor Details

#initialize(*args) ⇒ Orch

Returns a new instance of Orch.



34
35
36
37
 
# File 'lib/bolt/transport/orch.rb', line 34

def initialize(*args)
  @connections = {}
  super
end

Instance Attribute Details

#plan_context=(value) ⇒ Object (writeonly)

Sets the attribute plan_context

Parameters:

  • value —

    the value to set the attribute plan_context to.



19
20
21
 
# File 'lib/bolt/transport/orch.rb', line 19

def plan_context=(value)
  @plan_context = value
end

Class Method Details

.options ⇒ Object 



21
22
23
 
# File 'lib/bolt/transport/orch.rb', line 21

def self.options
  %w[service-url cacert token-file task-environment local-validation]
end

.validate(options) ⇒ Object 



27
28
29
30
31
32
 
# File 'lib/bolt/transport/orch.rb', line 27

def self.validate(options)
  validation_flag = options['local-validation']
  unless !!validation_flag == validation_flag
    raise Bolt::ValidationError, 'local-validation option must be a Boolean true or false'
  end
end

Instance Method Details

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



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/bolt/transport/orch.rb', line 88

def batch_command(targets, command, options = {}, &callback)
  params = {
    'command' => command
  }
  results = run_task_job(targets,
                         BOLT_COMMAND_TASK,
                         params,
                         options,
                         &callback)
  callback ||= proc {}
  results.map! { |result| unwrap_bolt_result(result.target, result) }
  results.each do |result|
    callback.call(type: :node_result, result: result)
  end
end

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



104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/bolt/transport/orch.rb', line 104

def batch_script(targets, script, arguments, options = {}, &callback)
  content = File.open(script, &:read)
  content = Base64.encode64(content)
  params = {
    'content' => content,
    'arguments' => arguments
  }
  callback ||= proc {}
  results = run_task_job(targets, BOLT_SCRIPT_TASK, params, options, &callback)
  results.map! { |result| unwrap_bolt_result(result.target, result) }
  results.each do |result|
    callback.call(type: :node_result, result: result)
  end
end

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



166
167
168
169
170
171
172
 
# File 'lib/bolt/transport/orch.rb', line 166

def batch_task(targets, task, arguments, options = {}, &callback)
  callback ||= proc {}
  results = run_task_job(targets, task, arguments, options, &callback)
  results.each do |result|
    callback.call(type: :node_result, result: result)
  end
end

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



119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/bolt/transport/orch.rb', line 119

def batch_upload(targets, source, destination, options = {}, &callback)
  content = File.open(source, &:read)
  content = Base64.encode64(content)
  mode = File.stat(source).mode
  params = {
    'path' => destination,
    'content' => content,
    'mode' => mode
  }
  callback ||= proc {}
  results = run_task_job(targets, BOLT_UPLOAD_TASK, params, options, &callback)
  results.map! do |result|
    if result.error_hash
      result
    else
      Bolt::Result.for_upload(result.target, source, destination)
    end
  end
  results.each do |result|
    callback&.call(type: :node_result, result: result)
  end
end

#batches(targets) ⇒ Object 



142
143
144
 
# File 'lib/bolt/transport/orch.rb', line 142

def batches(targets)
  targets.group_by { |target| Connection.get_key(target.options) }.values
end

#finish_plan(result) ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/bolt/transport/orch.rb', line 39

def finish_plan(result)
  if result.is_a? Bolt::PlanResult
    @connections.each_value do |conn|
      begin
        conn.finish_plan(result)
      rescue StandardError => e
        @logger.debug("Failed to finish plan on #{conn.key}: #{e.message}")
      end
    end
  end
end

#get_connection(conn_opts) ⇒ Object

It’s safe to create connections here for now because the batches/threads are per connection.



53
54
55
56
57
58
59
 
# File 'lib/bolt/transport/orch.rb', line 53

def get_connection(conn_opts)
  key = Connection.get_key(conn_opts)
  unless (conn = @connections[key])
    conn = @connections[key] = Connection.new(conn_opts, @plan_context, logger)
  end
  conn
end

#process_run_results(targets, results) ⇒ Object 



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
 
# File 'lib/bolt/transport/orch.rb', line 61

def process_run_results(targets, results)
  targets_by_name = Hash[targets.map(&:host).zip(targets)]
  results.map do |node_result|
    target = targets_by_name[node_result['name']]
    state = node_result['state']
    result = node_result['result']

    # If it's finished or already has a proper error simply pass it to the
    # the result otherwise make sure an error is generated
    if state == 'finished' || (result && result['_error'])
      Bolt::Result.new(target, value: result)
    elsif state == 'skipped'
      Bolt::Result.new(
        target,
        value: { '_error' => {
          'kind' => 'puppetlabs.tasks/skipped-node',
          'msg' => "Node #{target.host} was skipped",
          'details' => {}
        } }
      )
    else
      # Make a generic error with a unkown exit_code
      Bolt::Result.for_task(target, result.to_json, '', 'unknown')
    end
  end
end

#run_task_job(targets, task, arguments, options) ⇒ Object 



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/bolt/transport/orch.rb', line 146

def run_task_job(targets, task, arguments, options)
  targets.each do |target|
    yield(type: :node_start, target: target) if block_given?
  end

  begin
    results = get_connection(targets.first.options).run_task(targets, task, arguments, options)

    process_run_results(targets, results)
  rescue OrchestratorClient::ApiError => e
    targets.map do |target|
      Bolt::Result.new(target, error: e.data)
    end
  rescue StandardError => e
    targets.map do |target|
      Bolt::Result.from_exception(target, e)
    end
  end
end

#unwrap_bolt_result(target, result) ⇒ Object

run_task generates a result that makes sense for a generic task which needs to be unwrapped to extract stdout/stderr/exitcode.



177
178
179
180
181
182
183
184
 
# File 'lib/bolt/transport/orch.rb', line 177

def unwrap_bolt_result(target, result)
  if result.error_hash
    # something went wrong return the failure
    return result
  end

  Bolt::Result.for_command(target, result.value['stdout'], result.value['stderr'], result.value['exit_code'])
end