Module: Mobilize::Ssh

Defined in:
lib/mobilize-ssh.rb,
lib/mobilize-ssh/version.rb,
lib/mobilize-ssh/handlers/ssh.rb,
lib/mobilize-ssh/helpers/ssh_helper.rb

Constant Summary collapse

VERSION = 
"1.382"

Class Method Summary collapse

Class Method Details

.config ⇒ Object 



3
4
5
 
# File 'lib/mobilize-ssh/helpers/ssh_helper.rb', line 3

def self.config
  Base.config('ssh')
end

.default_node ⇒ Object 



23
24
25
 
# File 'lib/mobilize-ssh/helpers/ssh_helper.rb', line 23

def self.default_node
  self.nodes.first
end

.deploy(node, user_name, unique_name, command, file_hash) ⇒ Object 



17
18
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
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 17

def Ssh.deploy(node,user_name,unique_name,command,file_hash)
  loc_dir = Ssh.pop_loc_dir(unique_name,file_hash)
  Ssh.fire!(node,"rm -rf #{unique_name} && mkdir -p #{unique_name} && chown -R #{Ssh.node_owner(node)} #{unique_name}")
  if loc_dir
    Ssh.scp(node,loc_dir,".")
    #make sure loc_dir is removed
    FileUtils.rm_r(loc_dir,:force=>true)
  end
  #create cmd_file in unique_name
  cmd_path = "#{unique_name}/cmd.sh"
  Ssh.write(node,command,cmd_path)
  #move folder to user's home, change ownership
  user_dir = "/home/#{user_name}/"
  mobilize_dir = "#{user_dir}mobilize/"
  deploy_dir = "#{mobilize_dir}#{unique_name}/"
  deploy_cmd_path = "#{deploy_dir}cmd.sh"
  deploy_cmd = "sudo mkdir -p #{mobilize_dir} && " +
               "sudo rm -rf  #{mobilize_dir}#{unique_name} && " +
               "sudo mv #{unique_name} #{mobilize_dir} && " +
               "sudo chown -R #{user_name} #{mobilize_dir} && " +
               "sudo chmod -R 0700 #{user_name} #{mobilize_dir}"
  Ssh.fire!(node,deploy_cmd)
  #need to use bash or we get no tee
  full_cmd = "/bin/bash -l -c '(cd #{deploy_dir} && sh #{deploy_cmd_path} > >(tee stdout) 2> >(tee stderr >&2))'"
  #fire_cmd runs sh on cmd_path, optionally with sudo su
  fire_cmd = %{sudo su #{user_name} -c "#{full_cmd}"}
  return fire_cmd
end

.file_hash_by_stage_path(stage_path, gdrive_slot) ⇒ Object 



188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 188

def Ssh.file_hash_by_stage_path(stage_path,gdrive_slot)
  file_hash = {}
  s = Stage.where(:path=>stage_path).first
  u = s.job.runner.user
  user_name = Ssh.user_name_by_stage_path(stage_path)
  s.sources(gdrive_slot).each do |sdst|
                   split_path = sdst.path.split("/")
                   #if path is to stage output, name with stage name
                   file_name = if (split_path.last == "out" and (1..5).to_a.map{|n| "stage#{n.to_s}"}.include?(split_path[-2].to_s))
                                 #<jobname>/stage1/out
                                 "#{split_path[-2]}.out"
                               elsif (1..5).to_a.map{|n| "stage#{n.to_s}"}.include?(split_path.last[-6..-1])
                                 #runner<jobname>stage1
                               "#{split_path.last[-6..-1]}.out"
                               else
                                 split_path.last
                               end
                   if ["gsheet","gfile"].include?(sdst.handler)
                     #google drive sources are always read as the user
                     #with the apportioned slot
                     file_hash[file_name] = sdst.read(u.name,gdrive_slot)
                   else
                     #other sources should be read by su-user
                     file_hash[file_name] = sdst.read(user_name)
                   end
                 end
  return file_hash
end

.fire!(node, cmd) ⇒ Object 



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 127

def Ssh.fire!(node,cmd)
  puts "#{Time.now.utc}--Ssh on #{node}: #{cmd}"
  name,key,port,user = Ssh.host(node).ie{|h| ['name','key','port','user'].map{|k| h[k]}}
  key_path = "#{Base.root}/#{key}"
  opts = {:port=>(port || 22),:keys=>key_path}
  response = if Ssh.needs_gateway?(node)
               gname,gkey,gport,guser = Ssh.gateway(node).ie{|h| ['name','key','port','user'].map{|k| h[k]}}
               gkey_path = "#{Base.root}/#{gkey}"
               gopts = {:port=>(gport || 22),:keys=>gkey_path}
               Net::SSH::Gateway.run(gname,guser,name,user,cmd,gopts,opts)
             else
               Net::SSH.start(name,user,opts) do |ssh|
                 ssh.run(cmd)
               end
             end
  response
end

.gateway(node) ⇒ Object 



11
12
13
 
# File 'lib/mobilize-ssh/helpers/ssh_helper.rb', line 11

def self.gateway(node)
  self.config['nodes'][node]['gateway']
end

.home_dir ⇒ Object 



12
13
14
 
# File 'lib/mobilize-ssh.rb', line 12

def Ssh.home_dir
  File.expand_path('..',File.dirname(__FILE__))
end

.host(node) ⇒ Object 



7
8
9
 
# File 'lib/mobilize-ssh/helpers/ssh_helper.rb', line 7

def self.host(node)
  self.config['nodes'][node]['host']
end

.needs_gateway?(node) ⇒ Boolean

determine if current machine is on host domain, needs gateway if one is provided and it is not

Returns:

  • (Boolean) 


36
37
38
39
40
41
42
43
44
 
# File 'lib/mobilize-ssh/helpers/ssh_helper.rb', line 36

def self.needs_gateway?(node)
  return false if self.skip_gateway?(node)
  begin
    host_domain_name = self.host(node)['name'].split(".")[-2..-1].join(".")
    return true if self.gateway(node) and Socket.domain_name != host_domain_name
  rescue
    return false
  end
end

.node_owner(node) ⇒ Object 



27
28
29
 
# File 'lib/mobilize-ssh/helpers/ssh_helper.rb', line 27

def self.node_owner(node)
  self.host(node)['user']
end

.nodes ⇒ Object 



19
20
21
 
# File 'lib/mobilize-ssh/helpers/ssh_helper.rb', line 19

def self.nodes
  self.config['nodes'].keys
end

.path_to_dst(path, stage_path, gdrive_slot) ⇒ Object

converts a source path or target path to a dst in the context of handler and stage



47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 47

def Ssh.path_to_dst(path,stage_path,gdrive_slot)
  has_handler = true if path.index("://")
  red_path = path.split("://").last
  #is user has a handler, their first path node is a node name,
  #or there are more than 2 path nodes, try to find Ssh file
  if has_handler or Ssh.nodes.include?(red_path.split("/").first) or red_path.split("/").length > 2
    user_name = Ssh.user_name_by_stage_path(stage_path)
    ssh_url = Ssh.url_by_path(red_path,user_name)
    return Dataset.find_or_create_by_url(ssh_url)
  end
  #otherwise, use Gsheet
  return Gsheet.path_to_dst(red_path,stage_path,gdrive_slot)
end

.pop_loc_dir(unique_name, file_hash) ⇒ Object 



5
6
7
8
9
10
11
12
13
14
15
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 5

def Ssh.pop_loc_dir(unique_name,file_hash)
  loc_dir = "/tmp/#{unique_name}"
  `rm -rf #{loc_dir} && mkdir -p #{loc_dir}`
  file_hash.each do |fname,fdata|
    fpath = "#{loc_dir}/#{fname}"
    #for now, only gz is binary
    mode = fname.ends_with?(".gz") ? "wb" : "w"
    File.open(fpath,mode) {|f| f.print(fdata)}
  end
  return loc_dir if file_hash.keys.length>0
end

.read_by_dataset_path(dst_path, user_name, *args) ⇒ Object 



145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 145

def Ssh.read_by_dataset_path(dst_path,user_name,*args)
  #expects node as first part of path
  node,path = dst_path.split("/").ie{|pa| [pa.first,pa[1..-1].join("/")]}
  #slash in front of path
  response = Ssh.run(node,"cat /#{path}",user_name)
  if response['exit_code'] == 0
    return response['stdout']
  else
    raise "Unable to read ssh://#{dst_path} with error: #{response['stderr']}"
  end
end

.run(node, command, user_name, stage_path = nil, file_hash = {}, run_params = nil) ⇒ Object 



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
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 100

def Ssh.run(node,command,user_name,stage_path=nil,file_hash={},run_params=nil)
  file_hash ||= {}
  run_params ||={}
  #replace any params in the file_hash and command
  run_params.each do |k,v|
    command.gsub!("@#{k}",v)
    file_hash.each do |name,data|
      data.gsub!("@#{k}",v)
    end
  end
  #make sure the dir for this command is unique
  unique_name = if stage_path
                 stage_path.downcase.alphanunderscore
               else
                 [user_name,node,command,file_hash.keys.to_s,Time.now.to_f.to_s].join.to_md5
               end
  fire_cmd = Ssh.deploy(node, user_name, unique_name, command, file_hash)
  result = Ssh.fire!(node,fire_cmd)
  #clear out the md5 folders and those not requested to keep
  s = Stage.find_by_path(stage_path) if stage_path
  unless s and s.params['save_logs']
    rm_cmd = "sudo rm -rf /home/#{user_name}/mobilize/#{unique_name}"
    Ssh.fire!(node,rm_cmd)
  end
  return result
end

.run_by_stage_path(stage_path) ⇒ Object 



217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 217

def Ssh.run_by_stage_path(stage_path)
  gdrive_slot = Gdrive.slot_worker_by_path(stage_path)
  #return blank response if there are no slots available
  return nil unless gdrive_slot
  s = Stage.where(:path=>stage_path).first
  u = s.job.runner.user
  params = s.params
  node, command = [params['node'],params['cmd']]
  node ||= Ssh.default_node
  user_name = Ssh.user_name_by_stage_path(stage_path)
  #do not allow server commands from non-sudoers for the special server node
  if node=='server' and !Ssh.sudoers(node).include?(u.name)
    raise "You do not have permission to run commands on the mobilize server"
  end
  file_hash = Ssh.file_hash_by_stage_path(stage_path,gdrive_slot)
  Gdrive.unslot_worker_by_path(stage_path)
  run_params = params['params']
  result = Ssh.run(node,command,user_name,stage_path,file_hash,run_params)
  #use Gridfs to cache result
  response = {}
  response['out_url'] = Dataset.write_by_url("gridfs://#{s.path}/out",result['stdout'].to_s,Gdrive.owner_name)
  response['err_url'] = Dataset.write_by_url("gridfs://#{s.path}/err",result['stderr'].to_s,Gdrive.owner_name) if result['stderr'].to_s.length>0
  #is an error if there is no out and there is an err, regardless of signal
  result['exit_code'] = 500 if result['stdout'].to_s.strip.length==0 and result['stderr'].to_s.strip.length>0
  response['signal'] = result['exit_code']
  response
end

.scp(node, from_path, to_path) ⇒ Object 



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 83

def Ssh.scp(node,from_path,to_path)
  name,key,port,user = Ssh.host(node).ie{|h| ['name','key','port','user'].map{|k| h[k]}}
  key_path = "#{Base.root}/#{key}"
  opts = {:port=>(port || 22),:keys=>key_path}
  if Ssh.needs_gateway?(node)
    gname,gkey,gport,guser = Ssh.gateway(node).ie{|h| ['name','key','port','user'].map{|k| h[k]}}
    gkey_path = "#{Base.root}/#{gkey}"
    gopts = {:port=>(gport || 22),:keys=>gkey_path}
    return Net::SSH::Gateway.sync(gname,guser,name,user,from_path,to_path,gopts,opts)
  else
    Net::SCP.start(name,user,opts) do |scp|
      scp.upload!(from_path,to_path,:recursive=>true)
    end
  end
  return true
end

.skip_gateway?(node) ⇒ Boolean

Returns:

  • (Boolean) 


31
32
33
 
# File 'lib/mobilize-ssh/helpers/ssh_helper.rb', line 31

def self.skip_gateway?(node)
  self.config['nodes'][node]['skip_gateway']
end

.sudoers(node) ⇒ Object 



15
16
17
 
# File 'lib/mobilize-ssh/helpers/ssh_helper.rb', line 15

def self.sudoers(node)
  self.config['nodes'][node]['sudoers']
end

.tmp_file(fdata, binary = false, fpath = nil) ⇒ Object 



165
166
167
168
169
170
171
172
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 165

def Ssh.tmp_file(fdata,binary=false,fpath=nil)
  #creates a file under tmp/files with an md5 from the data
  tmp_file_path = fpath || "#{Dir.mktmpdir}/#{(fdata + Time.now.utc.to_f.to_s).to_md5}"
  write_mode = binary ? "wb" : "w"
  #write data to path
  File.open(tmp_file_path,write_mode) {|f| f.print(fdata)}
  return tmp_file_path
end

.url_by_path(path, user_name) ⇒ Object 



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 61

def Ssh.url_by_path(path,user_name)
  node = path.split("/").first.to_s
  if Ssh.nodes.include?(node)
    #cut node out of path
    path = "/" + path.split("/")[1..-1].join("/")
  else
    node = Ssh.default_node
    path = path.starts_with?("/") ? path : "/#{path}"
  end
  url = "ssh://#{node}#{path}"
  begin
    response = Ssh.run(node, "head -1 #{path}", user_name)
    if response['exit_code'] != 0
      raise "Unable to find #{url} with error: #{response['stderr']}"
    else
      return "ssh://#{node}#{path}"
    end
  rescue => exc
    raise Exception, "Unable to find #{url} with error: #{exc.to_s}", exc.backtrace
  end
end

.user_name_by_stage_path(stage_path, node = nil) ⇒ Object 



174
175
176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 174

def Ssh.user_name_by_stage_path(stage_path,node=nil)
  s = Stage.where(:path=>stage_path).first
  u = s.job.runner.user
  user_name = s.params['user']
  node = s.params['node']
  node = Ssh.default_node unless Ssh.nodes.include?(node)
  if user_name and !Ssh.sudoers(node).include?(u.name)
    raise "#{u.name} does not have su permissions for this node"
  elsif user_name.nil?
    user_name = u.name
  end
  return user_name
end

.write(node, fdata, to_path, binary = false) ⇒ Object 



157
158
159
160
161
162
163
 
# File 'lib/mobilize-ssh/handlers/ssh.rb', line 157

def Ssh.write(node,fdata,to_path,binary=false)
  from_path = Ssh.tmp_file(fdata,binary)
  Ssh.scp(node,from_path,to_path)
  #make sure local is removed
  FileUtils.rm_r(from_path,:force=>true)
  return true
end