Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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 NoDataLinstorCallException(LinstorCallException): 

72 pass 

73 

74 

75class NoPathLinstorCallException(LinstorCallException): 

76 pass 

77 

78 

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

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

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

82 ), priority=util.LOG_DEBUG) 

83 

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

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

86 target_host, device_path, vdi_uuid, remote_method, next_target, e 

87 ), priority=util.LOG_DEBUG) 

88 

89def linstorhostcall(local_method, remote_method=None): 

90 if not remote_method: 

91 remote_method = local_method 

92 

93 def decorated(response_parser): 

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

95 self = args[0] 

96 vdi_uuid = args[1] 

97 

98 device_path = self._linstor.build_device_path( 

99 self._linstor.get_volume_name(vdi_uuid) 

100 ) 

101 

102 if not self._session: 

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

104 

105 remote_args = { 

106 'devicePath': device_path, 

107 'groupName': self._linstor.group_name, 

108 'vdiType': self._vdi_type 

109 } 

110 remote_args.update(**kwargs) 

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

112 

113 this_host_ref = util.get_this_host_ref(self._session) 

114 def call_method(host_label, host_ref): 

115 if host_ref == this_host_ref: 

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

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

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

119 return response_parser(self, vdi_uuid, response) 

120 

121 # 1. Try on attached host. 

122 try: 

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

124 if host_ref_attached: 

125 return call_method('attached host', host_ref_attached) 

126 except Exception as e: 

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

128 

129 # 2. Try on master host. 

130 try: 

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

132 except Exception as e: 

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

134 

135 # 3. Try on a primary. 

136 hosts = self._get_hosts(remote_method, device_path) 

137 

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

139 if primary_hostname: 

140 try: 

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

142 except Exception as remote_e: 

143 self._raise_openers_exception(device_path, remote_e) 

144 

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

146 

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

148 try: 

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

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

151 ), None)) 

152 except Exception as remote_e: 

153 self._raise_openers_exception(device_path, remote_e) 

154 

155 return wrapper 

156 return decorated 

157 

158 

159def linstormodifier(): 

160 def decorated(func): 

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

162 self = args[0] 

163 

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

165 self._linstor.invalidate_resource_cache() 

166 return ret 

167 return wrapper 

168 return decorated 

169 

170 

171class LinstorCowUtil: 

172 class Chain(object): 

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

174 self._files = files 

175 self._leaf_path = leaf_path 

176 

177 @property 

178 def leaf_path(self) -> str: 

179 return self._leaf_path 

180 

181 def close(self) -> None: 

182 for file in self._files: 

183 with contextlib.suppress(Exception): 

184 file.close() 

185 

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

187 self._session = session 

188 self._linstor = linstor 

189 self._cowutil = getCowUtil(vdi_type) 

190 self._vdi_type = vdi_type 

191 

192 @property 

193 def cowutil(self) -> CowUtil: 

194 return self._cowutil 

195 

196 def create_chain_paths( 

197 self, 

198 vdi_uuid: str, 

199 readonly=False, 

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

201 ) -> Chain: 

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

203 # Useful for the snapshot code algorithm. 

204 

205 files: List[IO] = [] 

206 

207 leaf_path = self._linstor.get_device_path(vdi_uuid) 

208 path = leaf_path 

209 try: 

210 while True: 

211 if not util.pathexists(path): 

212 raise xs_errors.XenError( 

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

214 ) 

215 

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

217 def check_volume_usable(): 

218 while True: 

219 try: 

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

221 except (IOError, OSError) as e: 

222 if e.errno == errno.ENODATA: 

223 time.sleep(2) 

224 continue 

225 if e.errno in (errno.EROFS, errno.EMEDIUMTYPE): 

226 openers = self._linstor.get_volume_openers(vdi_uuid) 

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

228 if cb_openers: 

229 cb_openers(vdi_uuid, openers) 

230 raise 

231 break 

232 util.retry(check_volume_usable, 15, 2) 

233 

234 vdi_uuid = self.get_info(vdi_uuid).parentUuid 

235 if not vdi_uuid: 

236 break 

237 path = self._linstor.get_device_path(vdi_uuid) 

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

239 except Exception as e: 

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

241 raise e 

242 

243 return self.Chain(files, leaf_path) 

244 

245 # -------------------------------------------------------------------------- 

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

247 # -------------------------------------------------------------------------- 

248 

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

250 kwargs = { 

251 'ignoreMissingFooter': ignore_missing_footer, 

252 'fast': fast 

253 } 

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

255 

256 @linstorhostcall('check') 

257 def _check(self, vdi_uuid, response): 

258 return CowUtil.CheckResult(response) 

259 

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

261 kwargs = { 

262 'includeParent': include_parent, 

263 'resolveParent': False 

264 } 

265 

266 try: 

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

268 except Exception as e: 

269 # Backward compatibility with non-QCOW2 versions. 

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

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

272 raise 

273 

274 @linstorhostcall('getInfo') 

275 def _get_info(self, vdi_uuid, response): 

276 return self._get_info_impl(vdi_uuid, response) 

277 

278 # Backward compatibility with non-QCOW2 versions. 

279 @linstorhostcall('getVHDInfo') 

280 def _get_vhd_info(self, vdi_uuid, response): 

281 return self._get_info_impl(vdi_uuid, response) 

282 

283 def _get_info_impl(self, vdi_uuid, response): 

284 obj = json.loads(response) 

285 

286 image_info = CowImageInfo(vdi_uuid) 

287 image_info.sizeVirt = obj['sizeVirt'] 

288 image_info.sizePhys = obj['sizePhys'] 

289 if 'parentPath' in obj: 

290 image_info.parentPath = obj['parentPath'] 

291 image_info.parentUuid = obj['parentUuid'] 

292 image_info.hidden = obj['hidden'] 

293 image_info.path = obj['path'] 

294 

295 return image_info 

296 

297 @linstorhostcall('hasParent') 

298 def has_parent(self, vdi_uuid, response): 

299 return util.strtobool(response) 

300 

301 def get_parent(self, vdi_uuid): 

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

303 

304 @linstorhostcall('getParent') 

305 def _get_parent(self, vdi_uuid, response): 

306 return response 

307 

308 @linstorhostcall('getSizeVirt') 

309 def get_size_virt(self, vdi_uuid, response): 

310 return int(response) 

311 

312 @linstorhostcall('getMaxResizeSize') 

313 def get_max_resize_size(self, vdi_uuid, response): 

314 return int(response) 

315 

316 @linstorhostcall('getSizePhys') 

317 def get_size_phys(self, vdi_uuid, response): 

318 return int(response) 

319 

320 @linstorhostcall('getAllocatedSize') 

321 def get_allocated_size(self, vdi_uuid, response): 

322 return int(response) 

323 

324 @linstorhostcall('getDepth') 

325 def get_depth(self, vdi_uuid, response): 

326 return int(response) 

327 

328 @linstorhostcall('getKeyHash') 

329 def get_key_hash(self, vdi_uuid, response): 

330 return response or None 

331 

332 @linstorhostcall('getBlockBitmap') 

333 def get_block_bitmap(self, vdi_uuid, response): 

334 return base64.b64decode(response) 

335 

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

337 def get_drbd_size(self, vdi_uuid, response): 

338 return int(response) 

339 

340 def _get_drbd_size(self, path): 

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

342 if ret == 0: 

343 return int(stdout.strip()) 

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

345 

346 # -------------------------------------------------------------------------- 

347 # Setters: only used locally. 

348 # -------------------------------------------------------------------------- 

349 

350 @linstormodifier() 

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

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

353 

354 @linstormodifier() 

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

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

357 

358 @linstormodifier() 

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

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

361 

362 @linstormodifier() 

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

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

365 

366 @linstormodifier() 

367 def set_key(self, path, key_hash): 

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

369 

370 @linstormodifier() 

371 def kill_data(self, path): 

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

373 

374 @linstormodifier() 

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

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

377 

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

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

380 new_size = LinstorVolumeManager.round_up_volume_size(new_size) 

381 if new_size <= old_size: 

382 return 

383 

384 util.SMlog( 

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

386 .format(vdi_path, new_size, old_size) 

387 ) 

388 

389 journaler.create( 

390 LinstorJournaler.INFLATE, vdi_uuid, old_size 

391 ) 

392 self._linstor.resize_volume(vdi_uuid, new_size) 

393 

394 result_size = self.get_drbd_size(vdi_uuid) 

395 if result_size < new_size: 

396 util.SMlog( 

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

398 .format(new_size, result_size) 

399 ) 

400 

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

402 self.set_size_phys(vdi_path, result_size, False) 

403 journaler.remove(LinstorJournaler.INFLATE, vdi_uuid) 

404 

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

406 if zeroize: 

407 assert old_size > self._cowutil.getFooterSize() 

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

409 

410 new_size = LinstorVolumeManager.round_up_volume_size(new_size) 

411 if new_size >= old_size: 

412 return 

413 

414 util.SMlog( 

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

416 .format(vdi_path, new_size, old_size) 

417 ) 

418 

419 self.set_size_phys(vdi_path, new_size) 

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

421 

422 # -------------------------------------------------------------------------- 

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

424 # -------------------------------------------------------------------------- 

425 

426 @linstormodifier() 

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

428 kwargs = { 

429 'size': size, 

430 'jFile': jFile 

431 } 

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

433 

434 @linstormodifier() 

435 def set_size_virt_fast(self, path, size): 

436 kwargs = { 

437 'size': size 

438 } 

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

440 

441 @linstormodifier() 

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

443 kwargs = { 

444 'parentPath': str(parentPath), 

445 'parentRaw': parentRaw 

446 } 

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

448 

449 @linstormodifier() 

450 def force_coalesce(self, path): 

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

452 

453 @linstormodifier() 

454 def force_repair(self, path): 

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

456 

457 @linstormodifier() 

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

459 kwargs = { 

460 'newSize': newSize, 

461 'oldSize': oldSize, 

462 'zeroize': zeroize 

463 } 

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

465 

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

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

468 

469 # -------------------------------------------------------------------------- 

470 # Helpers. 

471 # -------------------------------------------------------------------------- 

472 

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

474 volume_size = virtual_size 

475 if VdiType.isCowImage(self._vdi_type): 

476 # All LINSTOR VDIs have the metadata area preallocated for 

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

478 meta_overhead = self._cowutil.calcOverheadEmpty( 

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

480 ) 

481 bitmap_overhead = self._cowutil.calcOverheadBitmap(virtual_size) 

482 volume_size += meta_overhead + bitmap_overhead 

483 elif self._vdi_type != VdiType.RAW: 

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

485 

486 return LinstorVolumeManager.round_up_volume_size(volume_size) 

487 

488 def _extract_uuid(self, device_path): 

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

490 return self._linstor.get_volume_uuid_from_device_path( 

491 device_path.rstrip('\n') 

492 ) 

493 

494 def _get_hosts(self, remote_method, device_path): 

495 try: 

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

497 except Exception as e: 

498 raise xs_errors.XenError( 

499 'VDIUnavailable', 

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

501 .format(remote_method, device_path, e) 

502 ) 

503 

504 # -------------------------------------------------------------------------- 

505 

506 @staticmethod 

507 def _find_host_ref_from_hostname(hosts, hostname): 

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

509 

510 def _raise_openers_exception(self, device_path, e): 

511 if isinstance(e, util.CommandException): 

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

513 else: 

514 e_str = str(e) 

515 

516 try: 

517 volume_uuid = self._linstor.get_volume_uuid_from_device_path( 

518 device_path 

519 ) 

520 e_wrapper = Exception( 

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

522 self._linstor.get_volume_openers(volume_uuid) 

523 ) 

524 ) 

525 except Exception as illformed_e: 

526 e_wrapper = Exception( 

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

528 ) 

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

530 raise e_wrapper # pylint: disable = E0702 

531 

532 def _sanitize_local_method(self, local_method): 

533 if isinstance(local_method, str): 

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

535 return local_method 

536 

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

538 local_method = self._sanitize_local_method(local_method) 

539 

540 try: 

541 def local_call(): 

542 try: 

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

544 except util.CommandException as e: 

545 if e.code in (errno.EROFS, errno.EMEDIUMTYPE): 

546 raise ErofsLinstorCallException(e) # Break retry calls. 

547 if e.code == errno.ENODATA: 

548 raise NoDataLinstorCallException(e) 

549 if e.code == errno.ENOENT: 

550 raise NoPathLinstorCallException(e) 

551 raise e 

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

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

554 except util.CommandException as e: 

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

556 raise e 

557 

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

559 try: 

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

561 except ErofsLinstorCallException as e: 

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

563 self._raise_openers_exception(device_path, e.cmd_err) 

564 

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

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

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

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

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

570 

571 local_method = self._sanitize_local_method(local_method) 

572 

573 # A. Try to write locally... 

574 try: 

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

576 except Exception: 

577 pass 

578 

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

580 

581 # B. Execute the command on another host. 

582 # B.1. Get host list. 

583 hosts = self._get_hosts(remote_method, device_path) 

584 

585 # B.2. Prepare remote args. 

586 remote_args = { 

587 'devicePath': device_path, 

588 'groupName': self._linstor.group_name, 

589 'vdiType': self._vdi_type 

590 } 

591 remote_args.update(**kwargs) 

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

593 

594 volume_uuid = self._linstor.get_volume_uuid_from_device_path( 

595 device_path 

596 ) 

597 parent_volume_uuid = None 

598 if use_parent: 

599 parent_volume_uuid = self.get_parent(volume_uuid) 

600 

601 openers_uuid = parent_volume_uuid if use_parent else volume_uuid 

602 

603 # B.3. Call! 

604 def remote_call(): 

605 try: 

606 openers = self._linstor.get_volume_openers(openers_uuid) 

607 except Exception as e: 

608 raise xs_errors.XenError( 

609 'VDIUnavailable', 

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

611 .format(remote_method, device_path, e) 

612 ) 

613 

614 no_host_found = True 

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

616 if not host_openers: 

617 continue 

618 

619 host_ref = self._find_host_ref_from_hostname(hosts, hostname) 

620 if not host_ref: 

621 continue 

622 

623 no_host_found = False 

624 try: 

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

626 except Exception: 

627 pass 

628 

629 if no_host_found: 

630 try: 

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

632 except Exception as e: 

633 self._raise_openers_exception(device_path, e) 

634 

635 raise xs_errors.XenError( 

636 'VDIUnavailable', 

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

638 .format(remote_method, device_path, openers) 

639 ) 

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

641 

642 def _zeroize(self, path, size): 

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

644 raise xs_errors.XenError( 

645 'EIO', 

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

647 ) 

648 

649class MultiLinstorCowUtil: 

650 class ExecutorData(threading.local): 

651 def __init__(self): 

652 self.clear() 

653 

654 def clear(self): 

655 self.session = None 

656 self.linstor = None 

657 self.vdi_type_to_cowutil = {} 

658 

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

660 self._uri = uri 

661 self._group_name = group_name 

662 self._loads: List[util.ApiSession] = [] 

663 self._executor_data = self.ExecutorData() 

664 

665 def __del__(self): 

666 self._cleanup() 

667 

668 def run(self, func, user_data_list): 

669 def wrapper(func, user_data): 

670 if not self._executor_data.session: 

671 self._init_executor_thread() 

672 return func(user_data, self) 

673 

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

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

676 

677 def get_local_cowutil(self, vdi_type): 

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

679 if not instance: 

680 instance = LinstorCowUtil( 

681 self._executor_data.session, 

682 self._executor_data.linstor, 

683 vdi_type 

684 ) 

685 self._executor_data.vdi_type_to_cowutil[vdi_type] = instance 

686 return instance 

687 

688 def _init_executor_thread(self): 

689 api_session = util.ApiSession("SM-linstor-multi-cow-util") 

690 try: 

691 linstor = LinstorVolumeManager( 

692 self._uri, 

693 self._group_name, 

694 repair=False, 

695 logger=util.SMlog 

696 ) 

697 self._executor_data.linstor = linstor 

698 self._executor_data.session = api_session.session 

699 except: 

700 self._executor_data.clear() 

701 raise 

702 self._loads.append(api_session) 

703 

704 def _cleanup(self): 

705 for load in self._loads: 

706 try: 

707 load.logout() 

708 except Exception as e: 

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

710 self._loads.clear()