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
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
|
import logging
import os
import asyncio
from tempfile import NamedTemporaryFile
from threading import Thread
from contextlib import contextmanager
from io import StringIO
from shlex import quote
from typing import TYPE_CHECKING, Optional, List, Tuple, Dict, Any, Iterator
from orchestrator import OrchestratorError
try:
import asyncssh
except ImportError:
asyncssh = None
if TYPE_CHECKING:
from cephadm.module import CephadmOrchestrator
from asyncssh.connection import SSHClientConnection
logger = logging.getLogger(__name__)
asyncssh_logger = logging.getLogger('asyncssh')
asyncssh_logger.propagate = False
DEFAULT_SSH_CONFIG = """
Host *
User root
StrictHostKeyChecking no
UserKnownHostsFile /dev/null
ConnectTimeout=30
"""
class EventLoopThread(Thread):
def __init__(self) -> None:
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
self._loop.set_debug(True)
super().__init__(target=self._loop.run_forever)
self.start()
def get_result(self, coro) -> Any: # type: ignore
return asyncio.run_coroutine_threadsafe(coro, self._loop).result()
class SSHManager:
def __init__(self, mgr: "CephadmOrchestrator"):
self.mgr: "CephadmOrchestrator" = mgr
self.cons: Dict[str, "SSHClientConnection"] = {}
async def _remote_connection(self,
host: str,
addr: Optional[str] = None,
) -> "SSHClientConnection":
if not self.cons.get(host):
if not addr and host in self.mgr.inventory:
addr = self.mgr.inventory.get_addr(host)
if not addr:
raise OrchestratorError("host address is empty")
assert self.mgr.ssh_user
n = self.mgr.ssh_user + '@' + addr
logger.debug("Opening connection to {} with ssh options '{}'".format(
n, self.mgr._ssh_options))
asyncssh.set_log_level('DEBUG')
asyncssh.set_debug_level(3)
with self.redirect_log(host, addr):
try:
conn = await asyncssh.connect(addr, username=self.mgr.ssh_user, client_keys=[self.mgr.tkey.name], known_hosts=None, config=[self.mgr.ssh_config_fname], preferred_auth=['publickey'])
except OSError:
raise
except asyncssh.Error:
raise
except Exception:
raise
self.cons[host] = conn
self.mgr.offline_hosts_remove(host)
conn = self.cons.get(host)
return conn
@contextmanager
def redirect_log(self, host: str, addr: str) -> Iterator[None]:
log_string = StringIO()
ch = logging.StreamHandler(log_string)
ch.setLevel(logging.DEBUG)
asyncssh_logger.addHandler(ch)
try:
yield
except OSError as e:
self.mgr.offline_hosts.add(host)
log_content = log_string.getvalue()
msg = f"Can't communicate with remote host `{addr}`, possibly because python3 is not installed there. {str(e)}" + \
'\n' + f'Log: {log_content}'
logger.exception(msg)
raise OrchestratorError(msg)
except asyncssh.Error as e:
self.mgr.offline_hosts.add(host)
log_content = log_string.getvalue()
msg = f'Failed to connect to {host} ({addr}). {str(e)}' + '\n' + f'Log: {log_content}'
logger.debug(msg)
raise OrchestratorError(msg)
except Exception as e:
self.mgr.offline_hosts.add(host)
log_content = log_string.getvalue()
logger.exception(str(e))
raise OrchestratorError(
f'Failed to connect to {host} ({addr}): {repr(e)}' + '\n' f'Log: {log_content}')
finally:
log_string.flush()
asyncssh_logger.removeHandler(ch)
def remote_connection(self,
host: str,
addr: Optional[str] = None,
) -> "SSHClientConnection":
return self.mgr.event_loop.get_result(self._remote_connection(host, addr))
async def _execute_command(self,
host: str,
cmd: List[str],
stdin: Optional[bytes] = None,
addr: Optional[str] = None,
) -> Tuple[str, str, int]:
conn = await self._remote_connection(host, addr)
cmd = "sudo " + " ".join(quote(x) for x in cmd)
logger.debug(f'Running command: {cmd}')
try:
r = await conn.run(cmd, input=stdin.decode() if stdin else None)
# handle these Exceptions otherwise you might get a weird error like TypeError: __init__() missing 1 required positional argument: 'reason' (due to the asyncssh error interacting with raise_if_exception)
except (asyncssh.ChannelOpenError, Exception) as e:
# SSH connection closed or broken, will create new connection next call
logger.debug(f'Connection to {host} failed. {str(e)}')
await self._reset_con(host)
self.mgr.offline_hosts.add(host)
raise OrchestratorError(f'Unable to reach remote host {host}. {str(e)}')
out = r.stdout.rstrip('\n')
err = r.stderr.rstrip('\n')
return out, err, r.returncode
def execute_command(self,
host: str,
cmd: List[str],
stdin: Optional[bytes] = None,
addr: Optional[str] = None,
) -> Tuple[str, str, int]:
return self.mgr.event_loop.get_result(self._execute_command(host, cmd, stdin, addr))
async def _check_execute_command(self,
host: str,
cmd: List[str],
stdin: Optional[bytes] = None,
addr: Optional[str] = None,
) -> str:
out, err, code = await self._execute_command(host, cmd, stdin, addr)
if code != 0:
msg = f'Command {cmd} failed. {err}'
logger.debug(msg)
raise OrchestratorError(msg)
return out
def check_execute_command(self,
host: str,
cmd: List[str],
stdin: Optional[bytes] = None,
addr: Optional[str] = None,
) -> str:
return self.mgr.event_loop.get_result(self._check_execute_command(host, cmd, stdin, addr))
async def _write_remote_file(self,
host: str,
path: str,
content: bytes,
mode: Optional[int] = None,
uid: Optional[int] = None,
gid: Optional[int] = None,
addr: Optional[str] = None,
) -> None:
try:
dirname = os.path.dirname(path)
await self._check_execute_command(host, ['mkdir', '-p', dirname], addr=addr)
tmp_path = path + '.new'
await self._check_execute_command(host, ['touch', tmp_path], addr=addr)
if uid is not None and gid is not None and mode is not None:
# shlex quote takes str or byte object, not int
await self._check_execute_command(host, ['chown', '-R', str(uid) + ':' + str(gid), tmp_path], addr=addr)
await self._check_execute_command(host, ['chmod', oct(mode)[2:], tmp_path], addr=addr)
with NamedTemporaryFile(prefix='cephadm-write-remote-file-') as f:
os.fchmod(f.fileno(), 0o600)
f.write(content)
f.flush()
conn = await self._remote_connection(host, addr)
await asyncssh.scp(f.name, (conn, tmp_path))
await self._check_execute_command(host, ['mv', tmp_path, path], addr=addr)
except Exception as e:
msg = f"Unable to write {host}:{path}: {e}"
logger.exception(msg)
raise OrchestratorError(msg)
def write_remote_file(self,
host: str,
path: str,
content: bytes,
mode: Optional[int] = None,
uid: Optional[int] = None,
gid: Optional[int] = None,
addr: Optional[str] = None,
) -> None:
self.mgr.event_loop.get_result(self._write_remote_file(
host, path, content, mode, uid, gid, addr))
async def _reset_con(self, host: str) -> None:
conn = self.cons.get(host)
if conn:
logger.debug(f'_reset_con close {host}')
conn.close()
del self.cons[host]
def reset_con(self, host: str) -> None:
self.mgr.event_loop.get_result(self._reset_con(host))
def _reset_cons(self) -> None:
for host, conn in self.cons.items():
logger.debug(f'_reset_cons close {host}')
conn.close()
self.cons = {}
def _reconfig_ssh(self) -> None:
temp_files = [] # type: list
ssh_options = [] # type: List[str]
# ssh_config
self.mgr.ssh_config_fname = self.mgr.ssh_config_file
ssh_config = self.mgr.get_store("ssh_config")
if ssh_config is not None or self.mgr.ssh_config_fname is None:
if not ssh_config:
ssh_config = DEFAULT_SSH_CONFIG
f = NamedTemporaryFile(prefix='cephadm-conf-')
os.fchmod(f.fileno(), 0o600)
f.write(ssh_config.encode('utf-8'))
f.flush() # make visible to other processes
temp_files += [f]
self.mgr.ssh_config_fname = f.name
if self.mgr.ssh_config_fname:
self.mgr.validate_ssh_config_fname(self.mgr.ssh_config_fname)
ssh_options += ['-F', self.mgr.ssh_config_fname]
self.mgr.ssh_config = ssh_config
# identity
ssh_key = self.mgr.get_store("ssh_identity_key")
ssh_pub = self.mgr.get_store("ssh_identity_pub")
self.mgr.ssh_pub = ssh_pub
self.mgr.ssh_key = ssh_key
if ssh_key and ssh_pub:
self.mgr.tkey = NamedTemporaryFile(prefix='cephadm-identity-')
self.mgr.tkey.write(ssh_key.encode('utf-8'))
os.fchmod(self.mgr.tkey.fileno(), 0o600)
self.mgr.tkey.flush() # make visible to other processes
tpub = open(self.mgr.tkey.name + '.pub', 'w')
os.fchmod(tpub.fileno(), 0o600)
tpub.write(ssh_pub)
tpub.flush() # make visible to other processes
temp_files += [self.mgr.tkey, tpub]
ssh_options += ['-i', self.mgr.tkey.name]
self.mgr._temp_files = temp_files
if ssh_options:
self.mgr._ssh_options = ' '.join(ssh_options)
else:
self.mgr._ssh_options = None
if self.mgr.mode == 'root':
self.mgr.ssh_user = self.mgr.get_store('ssh_user', default='root')
elif self.mgr.mode == 'cephadm-package':
self.mgr.ssh_user = 'cephadm'
self._reset_cons()
|
