-
Notifications
You must be signed in to change notification settings - Fork 2
/
odin.py
269 lines (239 loc) · 8.95 KB
/
odin.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
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
216
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
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
import os
import time
import random
import concurrent.futures
import json
import paramiko
from multipledispatch import dispatch
class Odin:
def __init__(self, machine_dict):
self.machine_dict = machine_dict
def _random_node(self):
"""
Module to select a random node from the
existing node_dict
"""
return random.choice(list(self.node_dict.keys()))
def connect_node(self, node, timeout=None):
"""
Function to establish connection with the given node.
"""
if timeout is None:
timeout_opt = {}
else:
timeout_opt = {'timeout': timeout}
node_ssh_client = paramiko.SSHClient()
node_ssh_client.load_host_keys(
os.path.expanduser('~/.ssh/known_hosts'))
try:
node_ssh_client.connect(
hostname=node,
username='root',
**timeout_opt
)
except Exception as e:
self.connect_flag = False
raise e
self.node_dict[node] = node_ssh_client
def establish_connection(self, timeout=15):
"""
Function to establish connection with the given
set of hosts.
"""
self.node_dict = {}
self.connect_flag = True
for node in self.machine_dict:
self.connect_node(node, timeout)
def deconstruct_connection(self):
"""
Function to close the existing connections.
"""
if not self.connect_flag:
return
for node in self.machine_dict:
if self.node_dict[node]:
(self.node_dict[node]).close()
return
@dispatch(str)
def execute_command(self, cmd):
"""
Module to handle random node execution.
Returns:
ret: A dictionary consisting
- Flag : Flag to check if connection failed
- msg : message
- error_msg: error message
- error_code: error code returned
- cmd : command that got executed
- node : node on which the command got executed
"""
return self.execute_command(cmd, self._random_node())
@dispatch(str, str)
def execute_command(self, cmd, node):
"""
Function to execute command in the given node.
Returns:
ret: A dictionary consisting
- Flag : Flag to check if connection failed
- msg : message
- error_msg: error message
- error_code: error code returned
- cmd : command that got executed
- node : node on which the command got executed
"""
ret_dict = {}
if not self.connect_flag:
ret_dict['Flag'] = False
return ret_dict
try:
_, stdout, stderr = self.node_dict[node].exec_command(cmd)
except Exception:
# Reconnection to be done.
self.connect_node(node)
# On rebooting the node
_, stdout, stderr = self.node_dict[node].exec_command(cmd)
if stdout.channel.recv_exit_status() != 0:
ret_dict['Flag'] = False
ret_dict['msg'] = stdout.readlines()
ret_dict['error_msg'] = stderr.readlines()
if isinstance(ret_dict['error_msg'], list):
ret_dict['error_msg'] = "".join(ret_dict['error_msg'])
else:
ret_dict['msg'] = stdout.readlines()
ret_dict['Flag'] = True
ret_dict['node'] = node
ret_dict['cmd'] = cmd
ret_dict['error_code'] = stdout.channel.recv_exit_status()
return ret_dict
@dispatch(str)
def execute_command_async(self, cmd: str) -> dict:
"""
Module to handle random node async execution.
Returns:
ret: A dictionary consisting
- cmd : Command requested
- node : Node wherein it was run
- stdout : The stdout handle
- stderr : The stderr handle
"""
return self.execute_command_async(cmd, self._random_node())
@dispatch(str, str)
def execute_command_async(self, cmd: str, node: str) -> dict:
"""
Function to execute command asynchronously in the given node.
Args:
cmd (string): Command to be executed.
node (string) : The node ip wherein the command is to be run.
Returns:
ret: A dictionary consisting
- cmd : Command requested
- node : Node wherein the command was run
- stdout : The stdout handle
- stderr : The stderr handle
"""
async_obj = {}
if not self.connect_flag:
return async_obj
try:
stdin, stdout, stderr = self.node_dict[node].exec_command(cmd)
except Exception:
# Reconnection to be done.
self.connect_node(node)
# On rebooting the node
stdin, stdout, stderr = self.node_dict[node].exec_command(cmd)
async_obj = {"cmd": cmd, "node": node, "stdout": stdout,
"stderr": stderr, "stdin": stdin}
return async_obj
def check_async_command_status(self, async_obj: dict) -> bool:
"""
A check to see if the async execution of a command which
was dispatched has been finished.
Args:
async_obj (dict) : Contains the details about the async command,
with keys -> 'stdout', 'stderr', 'cmd', 'node'
Returns:
Bool : True if the operations is completed or else False.
"""
return async_obj["stdout"].channel.exit_status_ready()
def collect_async_result(self, async_obj: dict) -> dict:
"""
Collect the async command's execution result after it ends.
Args:
async_obj (dict) : Contains the details about the async command,
with keys -> 'stdout', 'stderr', 'cmd', 'node'
Returns:
dict: Returns the resultant dictionary
"""
ret_dict = {}
if async_obj['stdout'].channel.recv_exit_status() != 0:
ret_dict['Flag'] = False
ret_dict['msg'] = async_obj['stdout'].readlines()
ret_dict['error_msg'] = async_obj['stderr'].readlines()
if isinstance(ret_dict['error_msg'], list):
ret_dict['error_msg'] = "".join(ret_dict['error_msg'])
else:
ret_dict['msg'] = async_obj['stdout'].readlines()
ret_dict['Flag'] = True
ret_dict['node'] = async_obj['node']
ret_dict['cmd'] = async_obj['cmd']
ret_dict['error_code'] = async_obj['stdout'].channel.recv_exit_status()
return ret_dict
def wait_till_async_command_ends(self, async_obj: dict) -> dict:
"""
Stay put till the async command finished it's execution and
provide the required return value.
Args:
async_obj (dict) : Contains the details about the async command,
with keys -> 'stdout', 'stderr', 'cmd', 'node'
Returns:
dict: Returns the resultant dictionary after the command ends.
"""
ret_dict = {}
while not async_obj['stdout'].channel.exit_status_ready():
time.sleep(1)
ret_dict = self.collect_async_result(async_obj)
return ret_dict
@dispatch(str)
def execute_command_multinode(self, cmd):
"""
Function to execute command in multiple nodes parallely
when node list isn't given.
"""
return self.execute_command_multinode(cmd, list(self.node_dict.keys()))
@dispatch(str, list)
def execute_command_multinode(self, cmd, node_list):
"""
Function to execute command in multiple nodes
parallely.
"""
ret_val = []
with concurrent.futures.ThreadPoolExecutor(
max_workers=len(node_list)) as executor:
future_exec = {executor.submit(
self.execute_command, cmd, node): node for node in node_list}
for future_handle in concurrent.futures.as_completed(future_exec):
try:
ret_val.append(future_handle.result())
except Exception as exc:
print(f"Generated exception : {exc}")
return ret_val
def transfer_file_from_local(self, source_path, dest_path, dest_node):
"""
Method to transfer a given file from the source node to the dest node.
Args:
source_path (str)
dest_path (str)
dest_node (str)
"""
sftp = self.node_dict[dest_node].open_sftp()
sftp.put(source_path, dest_path)
sftp.close()
def reboot_node(self, node):
"""
Reboot of a node is a special case and the normal execute_command
will throw an error.
Arg:
node (str)
"""
cmd = f'ssh -t root@{node} "reboot" > /dev/null 2>&1'
ret = os.system(cmd)