1#!/usr/bin/env python3 

2# 

3# Copyright (C) 2020 Vates SAS - ronan.abhamon@vates.fr 

4# 

5# This program is free software: you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation, either version 3 of the License, or 

8# (at your option) any later version. 

9# This program is distributed in the hope that it will be useful, 

10# but WITHOUT ANY WARRANTY; without even the implied warranty of 

11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

12# GNU General Public License for more details. 

13# 

14# You should have received a copy of the GNU General Public License 

15# along with this program. If not, see <https://www.gnu.org/licenses/>. 

16 

17from sm_typing import Any, Callable, Dict, IO, List, Optional, override 

18 

19from linstorjournaler import LinstorJournaler 

20from linstorvolumemanager import LinstorVolumeManager, LinstorVolumeOpeners 

21 

22from concurrent.futures import ThreadPoolExecutor 

23import base64 

24import contextlib 

25import errno 

26import json 

27import socket 

28import threading 

29import time 

30 

31from cowutil import CowImageInfo, CowUtil, getCowUtil 

32import util 

33import xs_errors 

34 

35from vditype import VdiType 

36 

37MANAGER_PLUGIN = 'linstor-manager' 

38 

39 

40def call_remote_method(session, host_ref, method, args): 

41 try: 

42 response = session.xenapi.host.call_plugin( 

43 host_ref, MANAGER_PLUGIN, method, args 

44 ) 

45 except Exception as e: 

46 util.SMlog('call-plugin on {} ({} with {}) exception: {}'.format( 

47 host_ref, method, args, e 

48 )) 

49 raise util.SMException(str(e)) 

50 

51 util.SMlog('call-plugin on {} ({} with {}) returned: {}'.format( 

52 host_ref, method, args, response 

53 )) 

54 

55 return response 

56 

57 

58class LinstorCallException(util.SMException): 

59 def __init__(self, cmd_err): 

60 self.cmd_err = cmd_err 

61 

62 @override 

63 def __str__(self) -> str: 

64 return str(self.cmd_err) 

65 

66 

67class ErofsLinstorCallException(LinstorCallException): 

68 pass 

69 

70 

71class NoPathLinstorCallException(LinstorCallException): 

72 pass 

73 

74def log_successful_call(target_host, device_path, vdi_uuid, remote_method, response): 

75 util.SMlog('Successful access on {} for device {} ({}): `{}` => {}'.format( 

76 target_host, device_path, vdi_uuid, remote_method, str(response) 

77 ), priority=util.LOG_DEBUG) 

78 

79def log_failed_call(target_host, next_target, device_path, vdi_uuid, remote_method, e): 

80 util.SMlog('Failed to call method on {} for device {} ({}): {}. Trying accessing on {}... (cause: {})'.format( 

81 target_host, device_path, vdi_uuid, remote_method, next_target, e 

82 ), priority=util.LOG_DEBUG) 

83 

84def linstorhostcall(local_method, remote_method=None): 

85 if not remote_method: 

86 remote_method = local_method 

87 

88 def decorated(response_parser): 

89 def wrapper(*args, **kwargs): 

90 self = args[0] 

91 vdi_uuid = args[1] 

92 

93 device_path = self._linstor.build_device_path( 

94 self._linstor.get_volume_name(vdi_uuid) 

95 ) 

96 

97 if not self._session: 

98 return self._call_local_method(local_method, device_path, *args[2:], **kwargs) 

99 

100 remote_args = { 

101 'devicePath': device_path, 

102 'groupName': self._linstor.group_name, 

103 'vdiType': self._vdi_type 

104 } 

105 remote_args.update(**kwargs) 

106 remote_args = {str(key): str(value) for key, value in remote_args.items()} 

107 

108 this_host_ref = util.get_this_host_ref(self._session) 

109 def call_method(host_label, host_ref): 

110 if host_ref == this_host_ref: 

111 return self._call_local_method(local_method, device_path, *args[2:], **kwargs) 

112 response = call_remote_method(self._session, host_ref, remote_method, remote_args) 

113 log_successful_call(host_label, device_path, vdi_uuid, remote_method, response) 

114 return response_parser(self, vdi_uuid, response) 

115 

116 # 1. Try on attached host. 

117 try: 

118 host_ref_attached = next(iter(util.get_hosts_attached_on(self._session, [vdi_uuid])), None) 

119 if host_ref_attached: 

120 return call_method('attached host', host_ref_attached) 

121 except Exception as e: 

122 log_failed_call('attached host', 'master', device_path, vdi_uuid, remote_method, e) 

123 

124 # 2. Try on master host. 

125 try: 

126 return call_method('master', util.get_master_ref(self._session)) 

127 except Exception as e: 

128 log_failed_call('master', 'primary', device_path, vdi_uuid, remote_method, e) 

129 

130 # 3. Try on a primary. 

131 hosts = self._get_hosts(remote_method, device_path) 

132 

133 nodes, primary_hostname = self._linstor.find_up_to_date_diskful_nodes(vdi_uuid) 

134 if primary_hostname: 

135 try: 

136 return call_method('primary', self._find_host_ref_from_hostname(hosts, primary_hostname)) 

137 except Exception as remote_e: 

138 self._raise_openers_exception(device_path, remote_e) 

139 

140 log_failed_call('primary', 'another node', device_path, vdi_uuid, remote_method, 'no primary') 

141 

142 # 4. Try on any host with local data. 

143 try: 

144 return call_method('another node', next(filter(None, 

145 (self._find_host_ref_from_hostname(hosts, hostname) for hostname in nodes) 

146 ), None)) 

147 except Exception as remote_e: 

148 self._raise_openers_exception(device_path, remote_e) 

149 

150 return wrapper 

151 return decorated 

152 

153 

154def linstormodifier(): 

155 def decorated(func): 

156 def wrapper(*args, **kwargs): 

157 self = args[0] 

158 

159 ret = func(*args, **kwargs) 

160 self._linstor.invalidate_resource_cache() 

161 return ret 

162 return wrapper 

163 return decorated 

164 

165 

166class LinstorCowUtil: 

167 class Chain(object): 

168 def __init__(self, files: List[IO], leaf_path: str): 

169 self._files = files 

170 self._leaf_path = leaf_path 

171 

172 @property 

173 def leaf_path(self) -> str: 

174 return self._leaf_path 

175 

176 def close(self) -> None: 

177 for file in self._files: 

178 with contextlib.suppress(Exception): 

179 file.close() 

180 

181 def __init__(self, session, linstor, vdi_type: str): 

182 self._session = session 

183 self._linstor = linstor 

184 self._cowutil = getCowUtil(vdi_type) 

185 self._vdi_type = vdi_type 

186 

187 @property 

188 def cowutil(self) -> CowUtil: 

189 return self._cowutil 

190 

191 def create_chain_paths( 

192 self, 

193 vdi_uuid: str, 

194 readonly=False, 

195 cb_openers: Optional[Callable[[str, LinstorVolumeOpeners], Any]] = None 

196 ) -> Chain: 

197 # OPTIMIZE: Add a limit_to_first_allocated_block param to limit cowutil calls. 

198 # Useful for the snapshot code algorithm. 

199 

200 files: List[IO] = [] 

201 

202 leaf_path = self._linstor.get_device_path(vdi_uuid) 

203 path = leaf_path 

204 try: 

205 while True: 

206 if not util.pathexists(path): 

207 raise xs_errors.XenError( 

208 'VDIUnavailable', opterr='Could not find: {}'.format(path) 

209 ) 

210 

211 # Diskless path can be created on the fly, ensure we can open it. 

212 def check_volume_usable(): 

213 while True: 

214 try: 

215 files.append(open(path, 'r' if readonly else 'r+')) 

216 except (IOError, OSError) as e: 

217 if e.errno == errno.ENODATA: 

218 time.sleep(2) 

219 continue 

220 if e.errno == errno.EROFS or e.errno == errno.EMEDIUMTYPE: 

221 openers = self._linstor.get_volume_openers(vdi_uuid) 

222 util.SMlog(f'Volume not attachable because used. Openers: {openers}') 

223 if cb_openers: 

224 cb_openers(vdi_uuid, openers) 

225 raise 

226 break 

227 util.retry(check_volume_usable, 15, 2) 

228 

229 vdi_uuid = self.get_info(vdi_uuid).parentUuid 

230 if not vdi_uuid: 

231 break 

232 path = self._linstor.get_device_path(vdi_uuid) 

233 readonly = True # Non-leaf is always readonly. 

234 except Exception as e: 

235 self.Chain(files, leaf_path).close() 

236 raise e 

237 

238 return self.Chain(files, leaf_path) 

239 

240 # -------------------------------------------------------------------------- 

241 # Getters: read locally and try on another host in case of failure. 

242 # -------------------------------------------------------------------------- 

243 

244 def check(self, vdi_uuid, ignore_missing_footer=False, fast=False): 

245 kwargs = { 

246 'ignoreMissingFooter': ignore_missing_footer, 

247 'fast': fast 

248 } 

249 return self._check(vdi_uuid, **kwargs) 

250 

251 @linstorhostcall('check') 

252 def _check(self, vdi_uuid, response): 

253 return CowUtil.CheckResult(response) 

254 

255 def get_info(self, vdi_uuid, include_parent=True): 

256 kwargs = { 

257 'includeParent': include_parent, 

258 'resolveParent': False 

259 } 

260 

261 try: 

262 return self._get_info(vdi_uuid, self._extract_uuid, **kwargs) 

263 except Exception as e: 

264 # Backward compatibility with non-QCOW2 versions. 

265 if str(e).startswith("['UNKNOWN_XENAPI_PLUGIN_FUNCTION', 'getInfo']"): 

266 return self._get_vhd_info(vdi_uuid, self._extract_uuid, **kwargs) 

267 raise 

268 

269 @linstorhostcall('getInfo') 

270 def _get_info(self, vdi_uuid, response): 

271 return self._get_info_impl(vdi_uuid, response) 

272 

273 # Backward compatibility with non-QCOW2 versions. 

274 @linstorhostcall('getVHDInfo') 

275 def _get_vhd_info(self, vdi_uuid, response): 

276 return self._get_info_impl(vdi_uuid, response) 

277 

278 def _get_info_impl(self, vdi_uuid, response): 

279 obj = json.loads(response) 

280 

281 image_info = CowImageInfo(vdi_uuid) 

282 image_info.sizeVirt = obj['sizeVirt'] 

283 image_info.sizePhys = obj['sizePhys'] 

284 if 'parentPath' in obj: 

285 image_info.parentPath = obj['parentPath'] 

286 image_info.parentUuid = obj['parentUuid'] 

287 image_info.hidden = obj['hidden'] 

288 image_info.path = obj['path'] 

289 

290 return image_info 

291 

292 @linstorhostcall('hasParent') 

293 def has_parent(self, vdi_uuid, response): 

294 return util.strtobool(response) 

295 

296 def get_parent(self, vdi_uuid): 

297 return self._get_parent(vdi_uuid, self._extract_uuid) 

298 

299 @linstorhostcall('getParent') 

300 def _get_parent(self, vdi_uuid, response): 

301 return response 

302 

303 @linstorhostcall('getSizeVirt') 

304 def get_size_virt(self, vdi_uuid, response): 

305 return int(response) 

306 

307 @linstorhostcall('getMaxResizeSize') 

308 def get_max_resize_size(self, vdi_uuid, response): 

309 return int(response) 

310 

311 @linstorhostcall('getSizePhys') 

312 def get_size_phys(self, vdi_uuid, response): 

313 return int(response) 

314 

315 @linstorhostcall('getAllocatedSize') 

316 def get_allocated_size(self, vdi_uuid, response): 

317 return int(response) 

318 

319 @linstorhostcall('getDepth') 

320 def get_depth(self, vdi_uuid, response): 

321 return int(response) 

322 

323 @linstorhostcall('getKeyHash') 

324 def get_key_hash(self, vdi_uuid, response): 

325 return response or None 

326 

327 @linstorhostcall('getBlockBitmap') 

328 def get_block_bitmap(self, vdi_uuid, response): 

329 return base64.b64decode(response) 

330 

331 @linstorhostcall('_get_drbd_size', 'getDrbdSize') 

332 def get_drbd_size(self, vdi_uuid, response): 

333 return int(response) 

334 

335 def _get_drbd_size(self, path): 

336 (ret, stdout, stderr) = util.doexec(['blockdev', '--getsize64', path]) 

337 if ret == 0: 

338 return int(stdout.strip()) 

339 raise util.SMException('Failed to get DRBD size: {}'.format(stderr)) 

340 

341 # -------------------------------------------------------------------------- 

342 # Setters: only used locally. 

343 # -------------------------------------------------------------------------- 

344 

345 @linstormodifier() 

346 def create(self, path, size, static, msize=0): 

347 return self._call_local_method_or_fail(self._cowutil.create, path, size, static, msize) 

348 

349 @linstormodifier() 

350 def set_size_phys(self, path, size, debug=True): 

351 return self._call_local_method_or_fail(self._cowutil.setSizePhys, path, size, debug) 

352 

353 @linstormodifier() 

354 def set_parent(self, path, parentPath, parentRaw=False): 

355 return self._call_local_method_or_fail(self._cowutil.setParent, path, parentPath, parentRaw) 

356 

357 @linstormodifier() 

358 def set_hidden(self, path, hidden=True): 

359 return self._call_local_method_or_fail(self._cowutil.setHidden, path, hidden) 

360 

361 @linstormodifier() 

362 def set_key(self, path, key_hash): 

363 return self._call_local_method_or_fail(self._cowutil.setKey, path, key_hash) 

364 

365 @linstormodifier() 

366 def kill_data(self, path): 

367 return self._call_local_method_or_fail(self._cowutil.killData, path) 

368 

369 @linstormodifier() 

370 def snapshot(self, path, parent, parentRaw, msize=0, checkEmpty=True): 

371 return self._call_local_method_or_fail(self._cowutil.snapshot, path, parent, parentRaw, msize, checkEmpty) 

372 

373 def inflate(self, journaler, vdi_uuid, vdi_path, new_size, old_size): 

374 # Only inflate if the LINSTOR volume capacity is not enough. 

375 new_size = LinstorVolumeManager.round_up_volume_size(new_size) 

376 if new_size <= old_size: 

377 return 

378 

379 util.SMlog( 

380 'Inflate {} (size={}, previous={})' 

381 .format(vdi_path, new_size, old_size) 

382 ) 

383 

384 journaler.create( 

385 LinstorJournaler.INFLATE, vdi_uuid, old_size 

386 ) 

387 self._linstor.resize_volume(vdi_uuid, new_size) 

388 

389 result_size = self.get_drbd_size(vdi_uuid) 

390 if result_size < new_size: 

391 util.SMlog( 

392 'WARNING: Cannot inflate volume to {}B, result size: {}B' 

393 .format(new_size, result_size) 

394 ) 

395 

396 self._zeroize(vdi_path, result_size - self._cowutil.getFooterSize()) 

397 self.set_size_phys(vdi_path, result_size, False) 

398 journaler.remove(LinstorJournaler.INFLATE, vdi_uuid) 

399 

400 def deflate(self, vdi_path, new_size, old_size, zeroize=False): 

401 if zeroize: 

402 assert old_size > self._cowutil.getFooterSize() 

403 self._zeroize(vdi_path, old_size - self._cowutil.getFooterSize()) 

404 

405 new_size = LinstorVolumeManager.round_up_volume_size(new_size) 

406 if new_size >= old_size: 

407 return 

408 

409 util.SMlog( 

410 'Deflate {} (new size={}, previous={})' 

411 .format(vdi_path, new_size, old_size) 

412 ) 

413 

414 self.set_size_phys(vdi_path, new_size) 

415 # TODO: Change the LINSTOR volume size using linstor.resize_volume. 

416 

417 # -------------------------------------------------------------------------- 

418 # Remote setters: write locally and try on another host in case of failure. 

419 # -------------------------------------------------------------------------- 

420 

421 @linstormodifier() 

422 def set_size_virt(self, path, size, jFile): 

423 kwargs = { 

424 'size': size, 

425 'jFile': jFile 

426 } 

427 return self._call_method(self._cowutil.setSizeVirt, 'setSizeVirt', path, use_parent=False, **kwargs) 

428 

429 @linstormodifier() 

430 def set_size_virt_fast(self, path, size): 

431 kwargs = { 

432 'size': size 

433 } 

434 return self._call_method(self._cowutil.setSizeVirtFast, 'setSizeVirtFast', path, use_parent=False, **kwargs) 

435 

436 @linstormodifier() 

437 def force_parent(self, path, parentPath, parentRaw=False): 

438 kwargs = { 

439 'parentPath': str(parentPath), 

440 'parentRaw': parentRaw 

441 } 

442 return self._call_method(self._cowutil.setParent, 'setParent', path, use_parent=False, **kwargs) 

443 

444 @linstormodifier() 

445 def force_coalesce(self, path): 

446 return int(self._call_method(self._cowutil.coalesce, 'coalesce', path, use_parent=True)) 

447 

448 @linstormodifier() 

449 def force_repair(self, path): 

450 return self._call_method(self._cowutil.repair, 'repair', path, use_parent=False) 

451 

452 @linstormodifier() 

453 def force_deflate(self, path, newSize, oldSize, zeroize): 

454 kwargs = { 

455 'newSize': newSize, 

456 'oldSize': oldSize, 

457 'zeroize': zeroize 

458 } 

459 return self._call_method('_force_deflate', 'deflate', path, use_parent=False, **kwargs) 

460 

461 def _force_deflate(self, path, newSize, oldSize, zeroize): 

462 self.deflate(path, newSize, oldSize, zeroize) 

463 

464 # -------------------------------------------------------------------------- 

465 # Helpers. 

466 # -------------------------------------------------------------------------- 

467 

468 def compute_volume_size(self, virtual_size: int) -> int: 

469 if VdiType.isCowImage(self._vdi_type): 

470 # All LINSTOR VDIs have the metadata area preallocated for 

471 # the maximum possible virtual size (for fast online VDI.resize). 

472 meta_overhead = self._cowutil.calcOverheadEmpty( 

473 max(virtual_size, self._cowutil.getDefaultPreallocationSizeVirt()) 

474 ) 

475 bitmap_overhead = self._cowutil.calcOverheadBitmap(virtual_size) 

476 virtual_size += meta_overhead + bitmap_overhead 

477 else: 

478 raise Exception('Invalid image type: {}'.format(self._vdi_type)) 

479 

480 return LinstorVolumeManager.round_up_volume_size(virtual_size) 

481 

482 def _extract_uuid(self, device_path): 

483 # TODO: Remove new line in the vhdutil module. Not here. 

484 return self._linstor.get_volume_uuid_from_device_path( 

485 device_path.rstrip('\n') 

486 ) 

487 

488 def _get_hosts(self, remote_method, device_path): 

489 try: 

490 return self._session.xenapi.host.get_all_records() 

491 except Exception as e: 

492 raise xs_errors.XenError( 

493 'VDIUnavailable', 

494 opterr='Unable to get host list to run cowutil command `{}` (path={}): {}' 

495 .format(remote_method, device_path, e) 

496 ) 

497 

498 # -------------------------------------------------------------------------- 

499 

500 @staticmethod 

501 def _find_host_ref_from_hostname(hosts, hostname): 

502 return next((ref for ref, rec in hosts.items() if rec['hostname'] == hostname), None) 

503 

504 def _raise_openers_exception(self, device_path, e): 

505 if isinstance(e, util.CommandException): 

506 e_str = 'cmd: `{}`, code: `{}`, reason: `{}`'.format(e.cmd, e.code, e.reason) 

507 else: 

508 e_str = str(e) 

509 

510 try: 

511 volume_uuid = self._linstor.get_volume_uuid_from_device_path( 

512 device_path 

513 ) 

514 e_wrapper = Exception( 

515 e_str + ' (openers: {})'.format( 

516 self._linstor.get_volume_openers(volume_uuid) 

517 ) 

518 ) 

519 except Exception as illformed_e: 

520 e_wrapper = Exception( 

521 e_str + ' (unable to get openers: {})'.format(illformed_e) 

522 ) 

523 util.SMlog('raise opener exception: {}'.format(e_wrapper)) 

524 raise e_wrapper # pylint: disable = E0702 

525 

526 def _sanitize_local_method(self, local_method): 

527 if isinstance(local_method, str): 

528 return getattr(self if local_method.startswith('_') else self._cowutil, local_method) 

529 return local_method 

530 

531 def _call_local_method(self, local_method, device_path, *args, **kwargs): 

532 local_method = self._sanitize_local_method(local_method) 

533 

534 try: 

535 def local_call(): 

536 try: 

537 return local_method(device_path, *args, **kwargs) 

538 except util.CommandException as e: 

539 if e.code == errno.EROFS or e.code == errno.EMEDIUMTYPE: 

540 raise ErofsLinstorCallException(e) # Break retry calls. 

541 if e.code == errno.ENOENT: 

542 raise NoPathLinstorCallException(e) 

543 raise e 

544 # Retry only locally if it's not an EROFS exception. 

545 return util.retry(local_call, 5, 2, exceptions=[util.CommandException]) 

546 except util.CommandException as e: 

547 util.SMlog('failed to execute locally CowUtil (sys {})'.format(e.code)) 

548 raise e 

549 

550 def _call_local_method_or_fail(self, local_method, device_path, *args, **kwargs): 

551 try: 

552 return self._call_local_method(local_method, device_path, *args, **kwargs) 

553 except ErofsLinstorCallException as e: 

554 # Volume is locked on a host, find openers. 

555 self._raise_openers_exception(device_path, e.cmd_err) 

556 

557 def _call_method(self, local_method, remote_method, device_path, use_parent, *args, **kwargs): 

558 # Note: `use_parent` exists to know if the COW image parent is used by the local/remote method. 

559 # Normally in case of failure, if the parent is unused we try to execute the method on 

560 # another host using the DRBD opener list. In the other case, if the parent is required, 

561 # we must check where this last one is open instead of the child. 

562 

563 local_method = self._sanitize_local_method(local_method) 

564 

565 # A. Try to write locally... 

566 try: 

567 return self._call_local_method(local_method, device_path, *args, **kwargs) 

568 except Exception: 

569 pass 

570 

571 util.SMlog('unable to execute `{}` locally, retry using a writable host...'.format(remote_method)) 

572 

573 # B. Execute the command on another host. 

574 # B.1. Get host list. 

575 hosts = self._get_hosts(remote_method, device_path) 

576 

577 # B.2. Prepare remote args. 

578 remote_args = { 

579 'devicePath': device_path, 

580 'groupName': self._linstor.group_name, 

581 'vdiType': self._vdi_type 

582 } 

583 remote_args.update(**kwargs) 

584 remote_args = {str(key): str(value) for key, value in remote_args.items()} 

585 

586 volume_uuid = self._linstor.get_volume_uuid_from_device_path( 

587 device_path 

588 ) 

589 parent_volume_uuid = None 

590 if use_parent: 

591 parent_volume_uuid = self.get_parent(volume_uuid) 

592 

593 openers_uuid = parent_volume_uuid if use_parent else volume_uuid 

594 

595 # B.3. Call! 

596 def remote_call(): 

597 try: 

598 openers = self._linstor.get_volume_openers(openers_uuid) 

599 except Exception as e: 

600 raise xs_errors.XenError( 

601 'VDIUnavailable', 

602 opterr='Unable to get DRBD openers to run CowUtil command `{}` (path={}): {}' 

603 .format(remote_method, device_path, e) 

604 ) 

605 

606 no_host_found = True 

607 for hostname, host_openers in openers.items(): 

608 if not host_openers: 

609 continue 

610 

611 host_ref = self._find_host_ref_from_hostname(hosts, hostname) 

612 if not host_ref: 

613 continue 

614 

615 no_host_found = False 

616 try: 

617 return call_remote_method(self._session, host_ref, remote_method, remote_args) 

618 except Exception: 

619 pass 

620 

621 if no_host_found: 

622 try: 

623 return local_method(device_path, *args, **kwargs) 

624 except Exception as e: 

625 self._raise_openers_exception(device_path, e) 

626 

627 raise xs_errors.XenError( 

628 'VDIUnavailable', 

629 opterr='No valid host found to run CowUtil command `{}` (path=`{}`, openers=`{}`)' 

630 .format(remote_method, device_path, openers) 

631 ) 

632 return util.retry(remote_call, 5, 2) 

633 

634 def _zeroize(self, path, size): 

635 if not util.zeroOut(path, size, self._cowutil.getFooterSize()): 

636 raise xs_errors.XenError( 

637 'EIO', 

638 opterr='Failed to zero out COW image footer {}'.format(path) 

639 ) 

640 

641class MultiLinstorCowUtil: 

642 class ExecutorData(threading.local): 

643 def __init__(self): 

644 self.clear() 

645 

646 def clear(self): 

647 self.session = None 

648 self.linstor = None 

649 self.vdi_type_to_cowutil = {} 

650 

651 class Load: 

652 def __init__(self, session): 

653 self.session = session 

654 

655 def cleanup(self): 

656 if self.session: 

657 self.session.xenapi.session.logout() 

658 self.session = None 

659 

660 def __init__(self, uri, group_name) -> None: 

661 self._uri = uri 

662 self._group_name = group_name 

663 self._loads: List[MultiLinstorCowUtil.Load] = [] 

664 self._executor_data = self.ExecutorData() 

665 

666 def __del__(self): 

667 self._cleanup() 

668 

669 def run(self, func, user_data_list): 

670 def wrapper(func, user_data): 

671 if not self._executor_data.session: 

672 self._init_executor_thread() 

673 return func(user_data, self) 

674 

675 with ThreadPoolExecutor(thread_name_prefix="CowUtil") as executor: 

676 return executor.map(lambda user_data: wrapper(func, user_data), user_data_list) 

677 

678 def get_local_cowutil(self, vdi_type): 

679 instance = self._executor_data.vdi_type_to_cowutil.get(vdi_type) 

680 if not instance: 

681 instance = LinstorCowUtil( 

682 self._executor_data.session, 

683 self._executor_data.linstor, 

684 vdi_type 

685 ) 

686 self._executor_data.vdi_type_to_cowutil[vdi_type] = instance 

687 return instance 

688 

689 def _init_executor_thread(self): 

690 session = util.get_localAPI_session() 

691 load = self.Load(session) 

692 try: 

693 linstor = LinstorVolumeManager( 

694 self._uri, 

695 self._group_name, 

696 repair=False, 

697 logger=util.SMlog 

698 ) 

699 self._executor_data.linstor = linstor 

700 self._executor_data.session = session 

701 except: 

702 self._executor_data.clear() 

703 load.cleanup() 

704 raise 

705 

706 self._loads.append(load) 

707 

708 def _cleanup(self): 

709 for load in self._loads: 

710 try: 

711 load.cleanup() 

712 except Exception as e: 

713 util.SMlog(f"Failed to clean load executor: {e}") 

714 self._loads.clear()