Class: Proxy::RemoteExecution::Ssh::Connector

Inherits:
Object
  • Object
show all
Defined in:
lib/smart_proxy_remote_execution_ssh/connector.rb

Overview

Service that handles running external commands for Actions::Command Dynflow action. It runs just one (actor) thread for all the commands running in the system and updates the Dynflow actions periodically.

Constant Summary collapse

MAX_PROCESS_RETRIES = 
3

Instance Method Summary collapse

Constructor Details

#initialize(host, user, options = {}) ⇒ Connector

Returns a new instance of Connector.



11
12
13
14
15
16
17
 
# File 'lib/smart_proxy_remote_execution_ssh/connector.rb', line 11

def initialize(host, user, options = {})
  @host = host
  @user = user
  @logger = options[:logger] || Logger.new($stderr)
  @client_private_key_file = options[:client_private_key_file]
  @known_hosts_file = options[:known_hosts_file]
end

Instance Method Details

#async_run(command) ⇒ Object

Initiates run of the remote command and yields the data when available. The yielding doesn’t happen automatically, but as part of calling the ‘refresh` method.



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
52
53
54
 
# File 'lib/smart_proxy_remote_execution_ssh/connector.rb', line 22

def async_run(command)
  started = false
  session.open_channel do |channel|
    channel.request_pty

    channel.on_data { |ch, data| yield CommandUpdate::StdoutData.new(data) }

    channel.on_extended_data { |ch, type, data| yield CommandUpdate::StderrData.new(data) }

    # standard exit of the command
    channel.on_request("exit-status") { |ch, data| yield CommandUpdate::StatusData.new(data.read_long) }

    # on signal: sedning the signal value (such as 'TERM')
    channel.on_request("exit-signal") do |ch, data|
      yield(CommandUpdate::StatusData.new(data.read_string))
      ch.close
      # wait for the channel to finish so that we know at the end
      # that the session is inactive
      ch.wait
    end

    channel.exec(command) do |ch, success|
      started = true
      unless success
        CommandUpdate.encode_exception("Error initializing command #{command}", e).each do |data|
          yield data
        end
      end
    end
  end
  session.process(0) until started
  return true
end

#close ⇒ Object 



118
119
120
121
 
# File 'lib/smart_proxy_remote_execution_ssh/connector.rb', line 118

def close
  @logger.debug("closing session to #{@user}@#{@host}")
  @session.close unless @session.nil? || @session.closed?
end

#ensure_remote_directory(path) ⇒ Object 



111
112
113
114
115
116
 
# File 'lib/smart_proxy_remote_execution_ssh/connector.rb', line 111

def ensure_remote_directory(path)
  exit_code, output = run("mkdir -p #{path}")
  if exit_code != 0
    raise "Unable to create directory on remote system #{path}: exit code: #{exit_code}\n #{output}"
  end
end

#refresh ⇒ Object

calls the callback registered in the ‘async_run` when some data for the session are available



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/smart_proxy_remote_execution_ssh/connector.rb', line 83

def refresh
  return if @session.nil?
  tries = 0
  begin
    session.process(0)
  rescue => e
    @logger.error("Error while processing ssh channel: #{e.class} #{e.message}\n #{e.backtrace.join("\n")}")
    tries += 1
    if tries <= MAX_PROCESS_RETRIES
      retry
    else
      raise e
    end
  end
end

#run(command) ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/smart_proxy_remote_execution_ssh/connector.rb', line 56

def run(command)
  output = ""
  exit_status = nil
  channel = session.open_channel do |ch|
    ch.on_data { |data| output.concat(data) }

    ch.on_extended_data { |_, _, data| output.concat(data) }

    ch.on_request("exit-status") { |_, data| exit_status = data.read_long }

    # on signal: sedning the signal value (such as 'TERM')
    ch.on_request("exit-signal") do |_, data|
      exit_status = data.read_string
      ch.close
      ch.wait
    end

    ch.exec command do |_, success|
      raise "could not execute command" unless success
    end
  end
  channel.wait
  return exit_status, output
end

#upload_file(local_path, remote_path) ⇒ Object 



99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/smart_proxy_remote_execution_ssh/connector.rb', line 99

def upload_file(local_path, remote_path)
  ensure_remote_directory(File.dirname(remote_path))
  scp = Net::SCP.new(session)
  upload_channel = scp.upload(local_path, remote_path)
  upload_channel.wait
ensure
  if upload_channel
    upload_channel.close
    upload_channel.wait
  end
end