ganeti.masterd.instance

1 # 2 # 3 4 # Copyright (C) 2010, 2011 Google Inc. 5 # All rights reserved. 6 # 7 # Redistribution and use in source and binary forms, with or without 8 # modification, are permitted provided that the following conditions are 9 # met: 10 # 11 # 1. Redistributions of source code must retain the above copyright notice, 12 # this list of conditions and the following disclaimer. 13 # 14 # 2. Redistributions in binary form must reproduce the above copyright 15 # notice, this list of conditions and the following disclaimer in the 16 # documentation and/or other materials provided with the distribution. 17 # 18 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS 19 # IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED 20 # TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR 21 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR 22 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 23 # EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 24 # PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 25 # PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 26 # LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 27 # NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 28 # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 29 30 31 """Instance-related functions and classes for masterd. 32 33 """ 34 35 import logging 36 import time 37 import OpenSSL 38 39 from ganeti import constants 40 from ganeti import errors 41 from ganeti import compat 42 from ganeti import utils 43 from ganeti import objects 44 from ganeti import netutils 45 from ganeti import pathutils

46 47 48 -class _ImportExportError(Exception):

49 """Local exception to report import/export errors. 50 51 """

52

53 54 -class ImportExportTimeouts(object):

55 #: Time until daemon starts writing status file 56 DEFAULT_READY_TIMEOUT = 10 57 58 #: Length of time until errors cause hard failure 59 DEFAULT_ERROR_TIMEOUT = 10 60 61 #: Time after which daemon must be listening 62 DEFAULT_LISTEN_TIMEOUT = 10 63 64 #: Progress update interval 65 DEFAULT_PROGRESS_INTERVAL = 60 66 67 __slots__ = [ 68 "error", 69 "ready", 70 "listen", 71 "connect", 72 "progress", 73 ] 74

75 - def __init__(self, connect, 76 listen=DEFAULT_LISTEN_TIMEOUT, 77 error=DEFAULT_ERROR_TIMEOUT, 78 ready=DEFAULT_READY_TIMEOUT, 79 progress=DEFAULT_PROGRESS_INTERVAL):

80 """Initializes this class. 81 82 @type connect: number 83 @param connect: Timeout for establishing connection 84 @type listen: number 85 @param listen: Timeout for starting to listen for connections 86 @type error: number 87 @param error: Length of time until errors cause hard failure 88 @type ready: number 89 @param ready: Timeout for daemon to become ready 90 @type progress: number 91 @param progress: Progress update interval 92 93 """ 94 self.error = error 95 self.ready = ready 96 self.listen = listen 97 self.connect = connect 98 self.progress = progress

99

100 101 -class ImportExportCbBase(object):

102 """Callbacks for disk import/export. 103 104 """

105 - def ReportListening(self, ie, private, component):

106 """Called when daemon started listening. 107 108 @type ie: Subclass of L{_DiskImportExportBase} 109 @param ie: Import/export object 110 @param private: Private data passed to import/export object 111 @param component: transfer component name 112 113 """

114

115 - def ReportConnected(self, ie, private):

116 """Called when a connection has been established. 117 118 @type ie: Subclass of L{_DiskImportExportBase} 119 @param ie: Import/export object 120 @param private: Private data passed to import/export object 121 122 """

123

124 - def ReportProgress(self, ie, private):

125 """Called when new progress information should be reported. 126 127 @type ie: Subclass of L{_DiskImportExportBase} 128 @param ie: Import/export object 129 @param private: Private data passed to import/export object 130 131 """

132

133 - def ReportFinished(self, ie, private):

134 """Called when a transfer has finished. 135 136 @type ie: Subclass of L{_DiskImportExportBase} 137 @param ie: Import/export object 138 @param private: Private data passed to import/export object 139 140 """

141

142 143 -class _DiskImportExportBase(object):

144 MODE_TEXT = None 145

146 - def __init__(self, lu, node_uuid, opts, 147 instance, component, timeouts, cbs, private=None):

148 """Initializes this class. 149 150 @param lu: Logical unit instance 151 @type node_uuid: string 152 @param node_uuid: Node UUID for import 153 @type opts: L{objects.ImportExportOptions} 154 @param opts: Import/export daemon options 155 @type instance: L{objects.Instance} 156 @param instance: Instance object 157 @type component: string 158 @param component: which part of the instance is being imported 159 @type timeouts: L{ImportExportTimeouts} 160 @param timeouts: Timeouts for this import 161 @type cbs: L{ImportExportCbBase} 162 @param cbs: Callbacks 163 @param private: Private data for callback functions 164 165 """ 166 assert self.MODE_TEXT 167 168 self._lu = lu 169 self.node_uuid = node_uuid 170 self.node_name = lu.cfg.GetNodeName(node_uuid) 171 self._opts = opts.Copy() 172 self._instance = instance 173 self._component = component 174 self._timeouts = timeouts 175 self._cbs = cbs 176 self._private = private 177 178 # Set master daemon's timeout in options for import/export daemon 179 assert self._opts.connect_timeout is None 180 self._opts.connect_timeout = timeouts.connect 181 182 # Parent loop 183 self._loop = None 184 185 # Timestamps 186 self._ts_begin = None 187 self._ts_connected = None 188 self._ts_finished = None 189 self._ts_cleanup = None 190 self._ts_last_progress = None 191 self._ts_last_error = None 192 193 # Transfer status 194 self.success = None 195 self.final_message = None 196 197 # Daemon status 198 self._daemon_name = None 199 self._daemon = None

200 201 @property

202 - def recent_output(self):

203 """Returns the most recent output from the daemon. 204 205 """ 206 if self._daemon: 207 return "\n".join(self._daemon.recent_output) 208 209 return None

210 211 @property

212 - def progress(self):

213 """Returns transfer progress information. 214 215 """ 216 if not self._daemon: 217 return None 218 219 return (self._daemon.progress_mbytes, 220 self._daemon.progress_throughput, 221 self._daemon.progress_percent, 222 self._daemon.progress_eta)

223 224 @property

225 - def magic(self):

226 """Returns the magic value for this import/export. 227 228 """ 229 return self._opts.magic

230 231 @property

232 - def active(self):

233 """Determines whether this transport is still active. 234 235 """ 236 return self.success is None

237 238 @property

239 - def loop(self):

240 """Returns parent loop. 241 242 @rtype: L{ImportExportLoop} 243 244 """ 245 return self._loop

246

247 - def SetLoop(self, loop):

248 """Sets the parent loop. 249 250 @type loop: L{ImportExportLoop} 251 252 """ 253 if self._loop: 254 raise errors.ProgrammerError("Loop can only be set once") 255 256 self._loop = loop

257

258 - def _StartDaemon(self):

259 """Starts the import/export daemon. 260 261 """ 262 raise NotImplementedError()

263

264 - def CheckDaemon(self):

265 """Checks whether daemon has been started and if not, starts it. 266 267 @rtype: string 268 @return: Daemon name 269 270 """ 271 assert self._ts_cleanup is None 272 273 if self._daemon_name is None: 274 assert self._ts_begin is None 275 276 result = self._StartDaemon() 277 if result.fail_msg: 278 raise _ImportExportError("Failed to start %s on %s: %s" % 279 (self.MODE_TEXT, self.node_name, 280 result.fail_msg)) 281 282 daemon_name = result.payload 283 284 logging.info("Started %s '%s' on %s", self.MODE_TEXT, daemon_name, 285 self.node_name) 286 287 self._ts_begin = time.time() 288 self._daemon_name = daemon_name 289 290 return self._daemon_name

291

292 - def GetDaemonName(self):

293 """Returns the daemon name. 294 295 """ 296 assert self._daemon_name, "Daemon has not been started" 297 assert self._ts_cleanup is None 298 return self._daemon_name

299

300 - def Abort(self):

301 """Sends SIGTERM to import/export daemon (if still active). 302 303 """ 304 if self._daemon_name: 305 self._lu.LogWarning("Aborting %s '%s' on %s", 306 self.MODE_TEXT, self._daemon_name, self.node_uuid) 307 result = self._lu.rpc.call_impexp_abort(self.node_uuid, self._daemon_name) 308 if result.fail_msg: 309 self._lu.LogWarning("Failed to abort %s '%s' on %s: %s", 310 self.MODE_TEXT, self._daemon_name, 311 self.node_uuid, result.fail_msg) 312 return False 313 314 return True

315

316 - def _SetDaemonData(self, data):

317 """Internal function for updating status daemon data. 318 319 @type data: L{objects.ImportExportStatus} 320 @param data: Daemon status data 321 322 """ 323 assert self._ts_begin is not None 324 325 if not data: 326 if utils.TimeoutExpired(self._ts_begin, self._timeouts.ready): 327 raise _ImportExportError("Didn't become ready after %s seconds" % 328 self._timeouts.ready) 329 330 return False 331 332 self._daemon = data 333 334 return True

335

336 - def SetDaemonData(self, success, data):

337 """Updates daemon status data. 338 339 @type success: bool 340 @param success: Whether fetching data was successful or not 341 @type data: L{objects.ImportExportStatus} 342 @param data: Daemon status data 343 344 """ 345 if not success: 346 if self._ts_last_error is None: 347 self._ts_last_error = time.time() 348 349 elif utils.TimeoutExpired(self._ts_last_error, self._timeouts.error): 350 raise _ImportExportError("Too many errors while updating data") 351 352 return False 353 354 self._ts_last_error = None 355 356 return self._SetDaemonData(data)

357

358 - def CheckListening(self):

359 """Checks whether the daemon is listening. 360 361 """ 362 raise NotImplementedError()

363

364 - def _GetConnectedCheckEpoch(self):

365 """Returns timeout to calculate connect timeout. 366 367 """ 368 raise NotImplementedError()

369

370 - def CheckConnected(self):

371 """Checks whether the daemon is connected. 372 373 @rtype: bool 374 @return: Whether the daemon is connected 375 376 """ 377 assert self._daemon, "Daemon status missing" 378 379 if self._ts_connected is not None: 380 return True 381 382 if self._daemon.connected: 383 self._ts_connected = time.time() 384 385 # TODO: Log remote peer 386 logging.debug("%s '%s' on %s is now connected", 387 self.MODE_TEXT, self._daemon_name, self.node_uuid) 388 389 self._cbs.ReportConnected(self, self._private) 390 391 return True 392 393 if utils.TimeoutExpired(self._GetConnectedCheckEpoch(), 394 self._timeouts.connect): 395 raise _ImportExportError("Not connected after %s seconds" % 396 self._timeouts.connect) 397 398 return False

399

400 - def _CheckProgress(self):

401 """Checks whether a progress update should be reported. 402 403 """ 404 if ((self._ts_last_progress is None or 405 utils.TimeoutExpired(self._ts_last_progress, 406 self._timeouts.progress)) and 407 self._daemon and 408 self._daemon.progress_mbytes is not None and 409 self._daemon.progress_throughput is not None): 410 self._cbs.ReportProgress(self, self._private) 411 self._ts_last_progress = time.time()

412

413 - def CheckFinished(self):

414 """Checks whether the daemon exited. 415 416 @rtype: bool 417 @return: Whether the transfer is finished 418 419 """ 420 assert self._daemon, "Daemon status missing" 421 422 if self._ts_finished: 423 return True 424 425 if self._daemon.exit_status is None: 426 # TODO: Adjust delay for ETA expiring soon 427 self._CheckProgress() 428 return False 429 430 self._ts_finished = time.time() 431 432 self._ReportFinished(self._daemon.exit_status == 0, 433 self._daemon.error_message) 434 435 return True

436

437 - def _ReportFinished(self, success, message):

438 """Transfer is finished or daemon exited. 439 440 @type success: bool 441 @param success: Whether the transfer was successful 442 @type message: string 443 @param message: Error message 444 445 """ 446 assert self.success is None 447 448 self.success = success 449 self.final_message = message 450 451 if success: 452 logging.info("%s '%s' on %s succeeded", self.MODE_TEXT, 453 self._daemon_name, self.node_uuid) 454 elif self._daemon_name: 455 self._lu.LogWarning("%s '%s' on %s failed: %s", 456 self.MODE_TEXT, self._daemon_name, 457 self._lu.cfg.GetNodeName(self.node_uuid), 458 message) 459 else: 460 self._lu.LogWarning("%s on %s failed: %s", self.MODE_TEXT, 461 self._lu.cfg.GetNodeName(self.node_uuid), message) 462 463 self._cbs.ReportFinished(self, self._private)

464

465 - def _Finalize(self):

466 """Makes the RPC call to finalize this import/export. 467 468 """ 469 return self._lu.rpc.call_impexp_cleanup(self.node_uuid, self._daemon_name)

470

471 - def Finalize(self, error=None):

472 """Finalizes this import/export. 473 474 """ 475 if self._daemon_name: 476 logging.info("Finalizing %s '%s' on %s", 477 self.MODE_TEXT, self._daemon_name, self.node_uuid) 478 479 result = self._Finalize() 480 if result.fail_msg: 481 self._lu.LogWarning("Failed to finalize %s '%s' on %s: %s", 482 self.MODE_TEXT, self._daemon_name, 483 self.node_uuid, result.fail_msg) 484 return False 485 486 # Daemon is no longer running 487 self._daemon_name = None 488 self._ts_cleanup = time.time() 489 490 if error: 491 self._ReportFinished(False, error) 492 493 return True

494

495 496 -class DiskImport(_DiskImportExportBase):

497 MODE_TEXT = "import" 498

499 - def __init__(self, lu, node_uuid, opts, instance, component, 500 dest, dest_args, timeouts, cbs, private=None):

501 """Initializes this class. 502 503 @param lu: Logical unit instance 504 @type node_uuid: string 505 @param node_uuid: Node name for import 506 @type opts: L{objects.ImportExportOptions} 507 @param opts: Import/export daemon options 508 @type instance: L{objects.Instance} 509 @param instance: Instance object 510 @type component: string 511 @param component: which part of the instance is being imported 512 @param dest: I/O destination 513 @param dest_args: I/O arguments 514 @type timeouts: L{ImportExportTimeouts} 515 @param timeouts: Timeouts for this import 516 @type cbs: L{ImportExportCbBase} 517 @param cbs: Callbacks 518 @param private: Private data for callback functions 519 520 """ 521 _DiskImportExportBase.__init__(self, lu, node_uuid, opts, instance, 522 component, timeouts, cbs, private) 523 self._dest = dest 524 self._dest_args = dest_args 525 526 # Timestamps 527 self._ts_listening = None

528 529 @property

530 - def listen_port(self):

531 """Returns the port the daemon is listening on. 532 533 """ 534 if self._daemon: 535 return self._daemon.listen_port 536 537 return None

538

539 - def _StartDaemon(self):

540 """Starts the import daemon. 541 542 """ 543 return self._lu.rpc.call_import_start(self.node_uuid, self._opts, 544 self._instance, self._component, 545 (self._dest, self._dest_args))

546

547 - def CheckListening(self):

548 """Checks whether the daemon is listening. 549 550 @rtype: bool 551 @return: Whether the daemon is listening 552 553 """ 554 assert self._daemon, "Daemon status missing" 555 556 if self._ts_listening is not None: 557 return True 558 559 port = self._daemon.listen_port 560 if port is not None: 561 self._ts_listening = time.time() 562 563 logging.debug("Import '%s' on %s is now listening on port %s", 564 self._daemon_name, self.node_uuid, port) 565 566 self._cbs.ReportListening(self, self._private, self._component) 567 568 return True 569 570 if utils.TimeoutExpired(self._ts_begin, self._timeouts.listen): 571 raise _ImportExportError("Not listening after %s seconds" % 572 self._timeouts.listen) 573 574 return False

575

576 - def _GetConnectedCheckEpoch(self):

577 """Returns the time since we started listening. 578 579 """ 580 assert self._ts_listening is not None, \ 581 ("Checking whether an import is connected is only useful" 582 " once it's been listening") 583 584 return self._ts_listening

585

586 587 -class DiskExport(_DiskImportExportBase):

588 MODE_TEXT = "export" 589

590 - def __init__(self, lu, node_uuid, opts, dest_host, dest_port, 591 instance, component, source, source_args, 592 timeouts, cbs, private=None):

593 """Initializes this class. 594 595 @param lu: Logical unit instance 596 @type node_uuid: string 597 @param node_uuid: Node UUID for import 598 @type opts: L{objects.ImportExportOptions} 599 @param opts: Import/export daemon options 600 @type dest_host: string 601 @param dest_host: Destination host name or IP address 602 @type dest_port: number 603 @param dest_port: Destination port number 604 @type instance: L{objects.Instance} 605 @param instance: Instance object 606 @type component: string 607 @param component: which part of the instance is being imported 608 @param source: I/O source 609 @param source_args: I/O source 610 @type timeouts: L{ImportExportTimeouts} 611 @param timeouts: Timeouts for this import 612 @type cbs: L{ImportExportCbBase} 613 @param cbs: Callbacks 614 @param private: Private data for callback functions 615 616 """ 617 _DiskImportExportBase.__init__(self, lu, node_uuid, opts, instance, 618 component, timeouts, cbs, private) 619 self._dest_host = dest_host 620 self._dest_port = dest_port 621 self._source = source 622 self._source_args = source_args

623

624 - def _StartDaemon(self):

625 """Starts the export daemon. 626 627 """ 628 return self._lu.rpc.call_export_start(self.node_uuid, self._opts, 629 self._dest_host, self._dest_port, 630 self._instance, self._component, 631 (self._source, self._source_args))

632

633 - def CheckListening(self):

634 """Checks whether the daemon is listening. 635 636 """ 637 # Only an import can be listening 638 return True

639

640 - def _GetConnectedCheckEpoch(self):

641 """Returns the time since the daemon started. 642 643 """ 644 assert self._ts_begin is not None 645 646 return self._ts_begin

647

648 649 -def FormatProgress(progress):

650 """Formats progress information for user consumption 651 652 """ 653 (mbytes, throughput, percent, eta) = progress 654 655 parts = [ 656 utils.FormatUnit(mbytes, "h"), 657 658 # Not using FormatUnit as it doesn't support kilobytes 659 "%0.1f MiB/s" % throughput, 660 ] 661 662 if percent is not None: 663 parts.append("%d%%" % percent) 664 665 if eta is not None: 666 parts.append("ETA %s" % utils.FormatSeconds(eta)) 667 668 return utils.CommaJoin(parts)

669

670 671 -class ImportExportLoop(object):

672 MIN_DELAY = 1.0 673 MAX_DELAY = 20.0 674

675 - def __init__(self, lu):

676 """Initializes this class. 677 678 """ 679 self._lu = lu 680 self._queue = [] 681 self._pending_add = []

682

683 - def Add(self, diskie):

684 """Adds an import/export object to the loop. 685 686 @type diskie: Subclass of L{_DiskImportExportBase} 687 @param diskie: Import/export object 688 689 """ 690 assert diskie not in self._pending_add 691 assert diskie.loop is None 692 693 diskie.SetLoop(self) 694 695 # Adding new objects to a staging list is necessary, otherwise the main 696 # loop gets confused if callbacks modify the queue while the main loop is 697 # iterating over it. 698 self._pending_add.append(diskie)

699 700 @staticmethod

701 - def _CollectDaemonStatus(lu, daemons):

702 """Collects the status for all import/export daemons. 703 704 """ 705 daemon_status = {} 706 707 for node_name, names in daemons.iteritems(): 708 result = lu.rpc.call_impexp_status(node_name, names) 709 if result.fail_msg: 710 lu.LogWarning("Failed to get daemon status on %s: %s", 711 node_name, result.fail_msg) 712 continue 713 714 assert len(names) == len(result.payload) 715 716 daemon_status[node_name] = dict(zip(names, result.payload)) 717 718 return daemon_status

719 720 @staticmethod

721 - def _GetActiveDaemonNames(queue):

722 """Gets the names of all active daemons. 723 724 """ 725 result = {} 726 for diskie in queue: 727 if not diskie.active: 728 continue 729 730 try: 731 # Start daemon if necessary 732 daemon_name = diskie.CheckDaemon() 733 except _ImportExportError, err: 734 logging.exception("%s failed", diskie.MODE_TEXT) 735 diskie.Finalize(error=str(err)) 736 continue 737 738 result.setdefault(diskie.node_name, []).append(daemon_name) 739 740 assert len(queue) >= len(result) 741 assert len(queue) >= sum([len(names) for names in result.itervalues()]) 742 743 logging.debug("daemons=%r", result) 744 745 return result

746

747 - def _AddPendingToQueue(self):

748 """Adds all pending import/export objects to the internal queue. 749 750 """ 751 assert compat.all(diskie not in self._queue and diskie.loop == self 752 for diskie in self._pending_add) 753 754 self._queue.extend(self._pending_add) 755 756 del self._pending_add[:]

757

758 - def Run(self):

759 """Utility main loop. 760 761 """ 762 while True: 763 self._AddPendingToQueue() 764 765 # Collect all active daemon names 766 daemons = self._GetActiveDaemonNames(self._queue) 767 if not daemons: 768 break 769 770 # Collection daemon status data 771 data = self._CollectDaemonStatus(self._lu, daemons) 772 773 # Use data 774 delay = self.MAX_DELAY 775 for diskie in self._queue: 776 if not diskie.active: 777 continue 778 779 try: 780 try: 781 all_daemon_data = data[diskie.node_name] 782 except KeyError: 783 result = diskie.SetDaemonData(False, None) 784 else: 785 result = \ 786 diskie.SetDaemonData(True, 787 all_daemon_data[diskie.GetDaemonName()]) 788 789 if not result: 790 # Daemon not yet ready, retry soon 791 delay = min(3.0, delay) 792 continue 793 794 if diskie.CheckFinished(): 795 # Transfer finished 796 diskie.Finalize() 797 continue 798 799 # Normal case: check again in 5 seconds 800 delay = min(5.0, delay) 801 802 if not diskie.CheckListening(): 803 # Not yet listening, retry soon 804 delay = min(1.0, delay) 805 continue 806 807 if not diskie.CheckConnected(): 808 # Not yet connected, retry soon 809 delay = min(1.0, delay) 810 continue 811 812 except _ImportExportError, err: 813 logging.exception("%s failed", diskie.MODE_TEXT) 814 diskie.Finalize(error=str(err)) 815 816 if not compat.any(diskie.active for diskie in self._queue): 817 break 818 819 # Wait a bit 820 delay = min(self.MAX_DELAY, max(self.MIN_DELAY, delay)) 821 logging.debug("Waiting for %ss", delay) 822 time.sleep(delay)

823

824 - def FinalizeAll(self):

825 """Finalizes all pending transfers. 826 827 """ 828 success = True 829 830 for diskie in self._queue: 831 success = diskie.Finalize() and success 832 833 return success

834

835 836 -class _TransferInstCbBase(ImportExportCbBase):

837 - def __init__(self, lu, feedback_fn, instance, timeouts, src_node_uuid, 838 src_cbs, dest_node_uuid, dest_ip):

839 """Initializes this class. 840 841 """ 842 ImportExportCbBase.__init__(self) 843 844 self.lu = lu 845 self.feedback_fn = feedback_fn 846 self.instance = instance 847 self.timeouts = timeouts 848 self.src_node_uuid = src_node_uuid 849 self.src_cbs = src_cbs 850 self.dest_node_uuid = dest_node_uuid 851 self.dest_ip = dest_ip

852

853 854 -class _TransferInstSourceCb(_TransferInstCbBase):

855 - def ReportConnected(self, ie, dtp):

856 """Called when a connection has been established. 857 858 """ 859 assert self.src_cbs is None 860 assert dtp.src_export == ie 861 assert dtp.dest_import 862 863 self.feedback_fn("%s is sending data on %s" % 864 (dtp.data.name, ie.node_name))

865

866 - def ReportProgress(self, ie, dtp):

867 """Called when new progress information should be reported. 868 869 """ 870 progress = ie.progress 871 if not progress: 872 return 873 874 self.feedback_fn("%s sent %s" % (dtp.data.name, FormatProgress(progress)))

875

876 - def ReportFinished(self, ie, dtp):

877 """Called when a transfer has finished. 878 879 """ 880 assert self.src_cbs is None 881 assert dtp.src_export == ie 882 assert dtp.dest_import 883 884 if ie.success: 885 self.feedback_fn("%s finished sending data" % dtp.data.name) 886 else: 887 self.feedback_fn("%s failed to send data: %s (recent output: %s)" % 888 (dtp.data.name, ie.final_message, ie.recent_output)) 889 890 dtp.RecordResult(ie.success) 891 892 cb = dtp.data.finished_fn 893 if cb: 894 cb() 895 896 # TODO: Check whether sending SIGTERM right away is okay, maybe we should 897 # give the daemon a moment to sort things out 898 if dtp.dest_import and not ie.success: 899 dtp.dest_import.Abort()

900

901 902 -class _TransferInstDestCb(_TransferInstCbBase):

903 - def ReportListening(self, ie, dtp, component):

904 """Called when daemon started listening. 905 906 """ 907 assert self.src_cbs 908 assert dtp.src_export is None 909 assert dtp.dest_import 910 assert dtp.export_opts 911 912 self.feedback_fn("%s is now listening, starting export" % dtp.data.name) 913 914 # Start export on source node 915 de = DiskExport(self.lu, self.src_node_uuid, dtp.export_opts, 916 self.dest_ip, ie.listen_port, self.instance, 917 component, dtp.data.src_io, dtp.data.src_ioargs, 918 self.timeouts, self.src_cbs, private=dtp) 919 ie.loop.Add(de) 920 921 dtp.src_export = de

922

923 - def ReportConnected(self, ie, dtp):

924 """Called when a connection has been established. 925 926 """ 927 self.feedback_fn("%s is receiving data on %s" % 928 (dtp.data.name, 929 self.lu.cfg.GetNodeName(self.dest_node_uuid)))

930

931 - def ReportFinished(self, ie, dtp):

932 """Called when a transfer has finished. 933 934 """ 935 if ie.success: 936 self.feedback_fn("%s finished receiving data" % dtp.data.name) 937 else: 938 self.feedback_fn("%s failed to receive data: %s (recent output: %s)" % 939 (dtp.data.name, ie.final_message, ie.recent_output)) 940 941 dtp.RecordResult(ie.success) 942 943 # TODO: Check whether sending SIGTERM right away is okay, maybe we should 944 # give the daemon a moment to sort things out 945 if dtp.src_export and not ie.success: 946 dtp.src_export.Abort()

947

948 949 -class DiskTransfer(object):

950 - def __init__(self, name, src_io, src_ioargs, dest_io, dest_ioargs, 951 finished_fn):

952 """Initializes this class. 953 954 @type name: string 955 @param name: User-visible name for this transfer (e.g. "disk/0") 956 @param src_io: Source I/O type 957 @param src_ioargs: Source I/O arguments 958 @param dest_io: Destination I/O type 959 @param dest_ioargs: Destination I/O arguments 960 @type finished_fn: callable 961 @param finished_fn: Function called once transfer has finished 962 963 """ 964 self.name = name 965 966 self.src_io = src_io 967 self.src_ioargs = src_ioargs 968 969 self.dest_io = dest_io 970 self.dest_ioargs = dest_ioargs 971 972 self.finished_fn = finished_fn

973

974 975 -class _DiskTransferPrivate(object):

976 - def __init__(self, data, success, export_opts):

977 """Initializes this class. 978 979 @type data: L{DiskTransfer} 980 @type success: bool 981 982 """ 983 self.data = data 984 self.success = success 985 self.export_opts = export_opts 986 987 self.src_export = None 988 self.dest_import = None

989

990 - def RecordResult(self, success):

991 """Updates the status. 992 993 One failed part will cause the whole transfer to fail. 994 995 """ 996 self.success = self.success and success

997

998 999 -def _GetInstDiskMagic(base, instance_name, index):

1000 """Computes the magic value for a disk export or import. 1001 1002 @type base: string 1003 @param base: Random seed value (can be the same for all disks of a transfer) 1004 @type instance_name: string 1005 @param instance_name: Name of instance 1006 @type index: number 1007 @param index: Disk index 1008 1009 """ 1010 h = compat.sha1_hash() 1011 h.update(str(constants.RIE_VERSION)) 1012 h.update(base) 1013 h.update(instance_name) 1014 h.update(str(index)) 1015 return h.hexdigest()

1016

1017 1018 -def TransferInstanceData(lu, feedback_fn, src_node_uuid, dest_node_uuid, 1019 dest_ip, compress, instance, all_transfers):

1020 """Transfers an instance's data from one node to another. 1021 1022 @param lu: Logical unit instance 1023 @param feedback_fn: Feedback function 1024 @type src_node_uuid: string 1025 @param src_node_uuid: Source node UUID 1026 @type dest_node_uuid: string 1027 @param dest_node_uuid: Destination node UUID 1028 @type dest_ip: string 1029 @param dest_ip: IP address of destination node 1030 @type compress: string 1031 @param compress: Compression tool to use 1032 @type instance: L{objects.Instance} 1033 @param instance: Instance object 1034 @type all_transfers: list of L{DiskTransfer} instances 1035 @param all_transfers: List of all disk transfers to be made 1036 @rtype: list 1037 @return: List with a boolean (True=successful, False=failed) for success for 1038 each transfer 1039 1040 """ 1041 src_node_name = lu.cfg.GetNodeName(src_node_uuid) 1042 dest_node_name = lu.cfg.GetNodeName(dest_node_uuid) 1043 1044 logging.debug("Source node %s, destination node %s, compression '%s'", 1045 src_node_name, dest_node_name, compress) 1046 1047 timeouts = ImportExportTimeouts(constants.DISK_TRANSFER_CONNECT_TIMEOUT) 1048 src_cbs = _TransferInstSourceCb(lu, feedback_fn, instance, timeouts, 1049 src_node_uuid, None, dest_node_uuid, dest_ip) 1050 dest_cbs = _TransferInstDestCb(lu, feedback_fn, instance, timeouts, 1051 src_node_uuid, src_cbs, dest_node_uuid, 1052 dest_ip) 1053 1054 all_dtp = [] 1055 1056 base_magic = utils.GenerateSecret(6) 1057 1058 ieloop = ImportExportLoop(lu) 1059 try: 1060 for idx, transfer in enumerate(all_transfers): 1061 if transfer: 1062 feedback_fn("Exporting %s from %s to %s" % 1063 (transfer.name, src_node_name, dest_node_name)) 1064 1065 magic = _GetInstDiskMagic(base_magic, instance.name, idx) 1066 opts = objects.ImportExportOptions(key_name=None, ca_pem=None, 1067 compress=compress, magic=magic) 1068 1069 dtp = _DiskTransferPrivate(transfer, True, opts) 1070 1071 di = DiskImport(lu, dest_node_uuid, opts, instance, "disk%d" % idx, 1072 transfer.dest_io, transfer.dest_ioargs, 1073 timeouts, dest_cbs, private=dtp) 1074 ieloop.Add(di) 1075 1076 dtp.dest_import = di 1077 else: 1078 dtp = _DiskTransferPrivate(None, False, None) 1079 1080 all_dtp.append(dtp) 1081 1082 ieloop.Run() 1083 finally: 1084 ieloop.FinalizeAll() 1085 1086 assert len(all_dtp) == len(all_transfers) 1087 assert compat.all((dtp.src_export is None or 1088 dtp.src_export.success is not None) and 1089 (dtp.dest_import is None or 1090 dtp.dest_import.success is not None) 1091 for dtp in all_dtp), \ 1092 "Not all imports/exports are finalized" 1093 1094 return [bool(dtp.success) for dtp in all_dtp]

1095

1096 1097 -class _RemoteExportCb(ImportExportCbBase):

1098 - def __init__(self, feedback_fn, disk_count):

1099 """Initializes this class. 1100 1101 """ 1102 ImportExportCbBase.__init__(self) 1103 self._feedback_fn = feedback_fn 1104 self._dresults = [None] * disk_count

1105 1106 @property

1107 - def disk_results(self):

1108 """Returns per-disk results. 1109 1110 """ 1111 return self._dresults

1112

1113 - def ReportConnected(self, ie, private):

1114 """Called when a connection has been established. 1115 1116 """ 1117 (idx, _) = private 1118 1119 self._feedback_fn("Disk %s is now sending data" % idx)

1120

1121 - def ReportProgress(self, ie, private):

1122 """Called when new progress information should be reported. 1123 1124 """ 1125 (idx, _) = private 1126 1127 progress = ie.progress 1128 if not progress: 1129 return 1130 1131 self._feedback_fn("Disk %s sent %s" % (idx, FormatProgress(progress)))

1132

1133 - def ReportFinished(self, ie, private):

1134 """Called when a transfer has finished. 1135 1136 """ 1137 (idx, finished_fn) = private 1138 1139 if ie.success: 1140 self._feedback_fn("Disk %s finished sending data" % idx) 1141 else: 1142 self._feedback_fn("Disk %s failed to send data: %s (recent output: %s)" % 1143 (idx, ie.final_message, ie.recent_output)) 1144 1145 self._dresults[idx] = bool(ie.success) 1146 1147 if finished_fn: 1148 finished_fn()

1149

1150 1151 -class ExportInstanceHelper(object):

1152 - def __init__(self, lu, feedback_fn, instance):

1153 """Initializes this class. 1154 1155 @param lu: Logical unit instance 1156 @param feedback_fn: Feedback function 1157 @type instance: L{objects.Instance} 1158 @param instance: Instance object 1159 1160 """ 1161 self._lu = lu 1162 self._feedback_fn = feedback_fn 1163 self._instance = instance 1164 1165 self._snap_disks = [] 1166 self._removed_snaps = [False] * len(instance.disks)

1167

1168 - def CreateSnapshots(self):

1169 """Creates an LVM snapshot for every disk of the instance. 1170 1171 """ 1172 assert not self._snap_disks 1173 1174 instance = self._instance 1175 src_node = instance.primary_node 1176 src_node_name = self._lu.cfg.GetNodeName(src_node) 1177 1178 inst_disks = self._lu.cfg.GetInstanceDisks(instance.uuid) 1179 for idx, disk in enumerate(inst_disks): 1180 self._feedback_fn("Creating a snapshot of disk/%s on node %s" % 1181 (idx, src_node_name)) 1182 1183 # result.payload will be a snapshot of an lvm leaf of the one we 1184 # passed 1185 result = self._lu.rpc.call_blockdev_snapshot(src_node, (disk, instance)) 1186 new_dev = False 1187 msg = result.fail_msg 1188 if msg: 1189 self._lu.LogWarning("Could not snapshot disk/%s on node %s: %s", 1190 idx, src_node_name, msg) 1191 elif (not isinstance(result.payload, (tuple, list)) or 1192 len(result.payload) != 2): 1193 self._lu.LogWarning("Could not snapshot disk/%s on node %s: invalid" 1194 " result '%s'", idx, src_node_name, result.payload) 1195 else: 1196 disk_id = tuple(result.payload) 1197 disk_params = constants.DISK_LD_DEFAULTS[constants.DT_PLAIN].copy() 1198 new_dev = objects.Disk(dev_type=constants.DT_PLAIN, size=disk.size, 1199 logical_id=disk_id, iv_name=disk.iv_name, 1200 params=disk_params) 1201 1202 self._snap_disks.append(new_dev) 1203 1204 assert len(self._snap_disks) == len(instance.disks) 1205 assert len(self._removed_snaps) == len(instance.disks)

1206

1207 - def _RemoveSnapshot(self, disk_index):

1208 """Removes an LVM snapshot. 1209 1210 @type disk_index: number 1211 @param disk_index: Index of the snapshot to be removed 1212 1213 """ 1214 disk = self._snap_disks[disk_index] 1215 if disk and not self._removed_snaps[disk_index]: 1216 src_node = self._instance.primary_node 1217 src_node_name = self._lu.cfg.GetNodeName(src_node) 1218 1219 self._feedback_fn("Removing snapshot of disk/%s on node %s" % 1220 (disk_index, src_node_name)) 1221 1222 result = self._lu.rpc.call_blockdev_remove(src_node, 1223 (disk, self._instance)) 1224 if result.fail_msg: 1225 self._lu.LogWarning("Could not remove snapshot for disk/%d from node" 1226 " %s: %s", disk_index, src_node_name, 1227 result.fail_msg) 1228 else: 1229 self._removed_snaps[disk_index] = True

1230

1231 - def LocalExport(self, dest_node, compress):

1232 """Intra-cluster instance export. 1233 1234 @type dest_node: L{objects.Node} 1235 @param dest_node: Destination node 1236 @type compress: string 1237 @param compress: Compression tool to use 1238 1239 """ 1240 instance = self._instance 1241 src_node_uuid = instance.primary_node 1242 1243 assert len(self._snap_disks) == len(instance.disks) 1244 1245 transfers = [] 1246 1247 for idx, dev in enumerate(self._snap_disks): 1248 if not dev: 1249 transfers.append(None) 1250 continue 1251 1252 path = utils.PathJoin(pathutils.EXPORT_DIR, "%s.new" % instance.name, 1253 dev.logical_id[1]) 1254 1255 finished_fn = compat.partial(self._TransferFinished, idx) 1256 1257 if instance.os: 1258 src_io = constants.IEIO_SCRIPT 1259 src_ioargs = ((dev, instance), idx) 1260 else: 1261 src_io = constants.IEIO_RAW_DISK 1262 src_ioargs = (dev, instance) 1263 1264 # FIXME: pass debug option from opcode to backend 1265 dt = DiskTransfer("snapshot/%s" % idx, 1266 src_io, src_ioargs, 1267 constants.IEIO_FILE, (path, ), 1268 finished_fn) 1269 transfers.append(dt) 1270 1271 # Actually export data 1272 dresults = TransferInstanceData(self._lu, self._feedback_fn, 1273 src_node_uuid, dest_node.uuid, 1274 dest_node.secondary_ip, 1275 compress, 1276 instance, transfers) 1277 1278 assert len(dresults) == len(instance.disks) 1279 1280 # Finalize only if all the disks have been exported successfully 1281 if all(dresults): 1282 self._feedback_fn("Finalizing export on %s" % dest_node.name) 1283 result = self._lu.rpc.call_finalize_export(dest_node.uuid, instance, 1284 self._snap_disks) 1285 msg = result.fail_msg 1286 fin_resu = not msg 1287 if msg: 1288 self._lu.LogWarning("Could not finalize export for instance %s" 1289 " on node %s: %s", instance.name, dest_node.name, 1290 msg) 1291 else: 1292 fin_resu = False 1293 self._lu.LogWarning("Some disk exports have failed; there may be " 1294 "leftover data for instance %s on node %s", 1295 instance.name, dest_node.name) 1296 1297 return (fin_resu, dresults)

1298

1299 - def RemoteExport(self, disk_info, key_name, dest_ca_pem, compress, timeouts):

1300 """Inter-cluster instance export. 1301 1302 @type disk_info: list 1303 @param disk_info: Per-disk destination information 1304 @type key_name: string 1305 @param key_name: Name of X509 key to use 1306 @type dest_ca_pem: string 1307 @param dest_ca_pem: Destination X509 CA in PEM format 1308 @type compress: string 1309 @param compress: Compression tool to use 1310 @type timeouts: L{ImportExportTimeouts} 1311 @param timeouts: Timeouts for this import 1312 1313 """ 1314 instance = self._instance 1315 inst_disks = self._lu.cfg.GetInstanceDisks(instance.uuid) 1316 1317 assert len(disk_info) == len(instance.disks) 1318 1319 cbs = _RemoteExportCb(self._feedback_fn, len(instance.disks)) 1320 1321 ieloop = ImportExportLoop(self._lu) 1322 try: 1323 for idx, (dev, (host, port, magic)) in enumerate(zip(inst_disks, 1324 disk_info)): 1325 # Decide whether to use IPv6 1326 ipv6 = netutils.IP6Address.IsValid(host) 1327 1328 opts = objects.ImportExportOptions(key_name=key_name, 1329 ca_pem=dest_ca_pem, 1330 magic=magic, 1331 compress=compress, 1332 ipv6=ipv6) 1333 1334 if instance.os: 1335 src_io = constants.IEIO_SCRIPT 1336 src_ioargs = ((dev, instance), idx) 1337 else: 1338 src_io = constants.IEIO_RAW_DISK 1339 src_ioargs = (dev, instance) 1340 1341 self._feedback_fn("Sending disk %s to %s:%s" % (idx, host, port)) 1342 finished_fn = compat.partial(self._TransferFinished, idx) 1343 ieloop.Add(DiskExport(self._lu, instance.primary_node, 1344 opts, host, port, instance, "disk%d" % idx, 1345 src_io, src_ioargs, 1346 timeouts, cbs, private=(idx, finished_fn))) 1347 1348 ieloop.Run() 1349 finally: 1350 ieloop.FinalizeAll() 1351 1352 return (True, cbs.disk_results)

1353

1354 - def _TransferFinished(self, idx):

1355 """Called once a transfer has finished. 1356 1357 @type idx: number 1358 @param idx: Disk index 1359 1360 """ 1361 logging.debug("Transfer %s finished", idx) 1362 self._RemoveSnapshot(idx)

1363

1364 - def Cleanup(self):

1365 """Remove all snapshots. 1366 1367 """ 1368 assert len(self._removed_snaps) == len(self._instance.disks) 1369 for idx in range(len(self._instance.disks)): 1370 self._RemoveSnapshot(idx)

1371

1372 1373 -class _RemoteImportCb(ImportExportCbBase):

1374 - def __init__(self, feedback_fn, cds, x509_cert_pem, disk_count, 1375 external_address):

1376 """Initializes this class. 1377 1378 @type cds: string 1379 @param cds: Cluster domain secret 1380 @type x509_cert_pem: string 1381 @param x509_cert_pem: CA used for signing import key 1382 @type disk_count: number 1383 @param disk_count: Number of disks 1384 @type external_address: string 1385 @param external_address: External address of destination node 1386 1387 """ 1388 ImportExportCbBase.__init__(self) 1389 self._feedback_fn = feedback_fn 1390 self._cds = cds 1391 self._x509_cert_pem = x509_cert_pem 1392 self._disk_count = disk_count 1393 self._external_address = external_address 1394 1395 self._dresults = [None] * disk_count 1396 self._daemon_port = [None] * disk_count 1397 1398 self._salt = utils.GenerateSecret(8)

1399 1400 @property

1401 - def disk_results(self):

1402 """Returns per-disk results. 1403 1404 """ 1405 return self._dresults

1406

1407 - def _CheckAllListening(self):

1408 """Checks whether all daemons are listening. 1409 1410 If all daemons are listening, the information is sent to the client. 1411 1412 """ 1413 if not compat.all(dp is not None for dp in self._daemon_port): 1414 return 1415 1416 host = self._external_address 1417 1418 disks = [] 1419 for idx, (port, magic) in enumerate(self._daemon_port): 1420 disks.append(ComputeRemoteImportDiskInfo(self._cds, self._salt, 1421 idx, host, port, magic)) 1422 1423 assert len(disks) == self._disk_count 1424 1425 self._feedback_fn(constants.ELOG_REMOTE_IMPORT, { 1426 "disks": disks, 1427 "x509_ca": self._x509_cert_pem, 1428 })

1429

1430 - def ReportListening(self, ie, private, _):

1431 """Called when daemon started listening. 1432 1433 """ 1434 (idx, ) = private 1435 1436 self._feedback_fn("Disk %s is now listening" % idx) 1437 1438 assert self._daemon_port[idx] is None 1439 1440 self._daemon_port[idx] = (ie.listen_port, ie.magic) 1441 1442 self._CheckAllListening()

1443

1444 - def ReportConnected(self, ie, private):

1445 """Called when a connection has been established. 1446 1447 """ 1448 (idx, ) = private 1449 1450 self._feedback_fn("Disk %s is now receiving data" % idx)

1451

1452 - def ReportFinished(self, ie, private):

1453 """Called when a transfer has finished. 1454 1455 """ 1456 (idx, ) = private 1457 1458 # Daemon is certainly no longer listening 1459 self._daemon_port[idx] = None 1460 1461 if ie.success: 1462 self._feedback_fn("Disk %s finished receiving data" % idx) 1463 else: 1464 self._feedback_fn(("Disk %s failed to receive data: %s" 1465 " (recent output: %s)") % 1466 (idx, ie.final_message, ie.recent_output)) 1467 1468 self._dresults[idx] = bool(ie.success)

1469

1470 1471 -def RemoteImport(lu, feedback_fn, instance, pnode, source_x509_ca, 1472 cds, compress, timeouts):

1473 """Imports an instance from another cluster. 1474 1475 @param lu: Logical unit instance 1476 @param feedback_fn: Feedback function 1477 @type instance: L{objects.Instance} 1478 @param instance: Instance object 1479 @type pnode: L{objects.Node} 1480 @param pnode: Primary node of instance as an object 1481 @type source_x509_ca: OpenSSL.crypto.X509 1482 @param source_x509_ca: Import source's X509 CA 1483 @type cds: string 1484 @param cds: Cluster domain secret 1485 @type compress: string 1486 @param compress: Compression tool to use 1487 @type timeouts: L{ImportExportTimeouts} 1488 @param timeouts: Timeouts for this import 1489 1490 """ 1491 source_ca_pem = OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, 1492 source_x509_ca) 1493 1494 magic_base = utils.GenerateSecret(6) 1495 1496 # Decide whether to use IPv6 1497 ipv6 = netutils.IP6Address.IsValid(pnode.primary_ip) 1498 1499 # Create crypto key 1500 result = lu.rpc.call_x509_cert_create(instance.primary_node, 1501 constants.RIE_CERT_VALIDITY) 1502 result.Raise("Can't create X509 key and certificate on %s" % result.node) 1503 1504 (x509_key_name, x509_cert_pem) = result.payload 1505 try: 1506 # Load certificate 1507 x509_cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, 1508 x509_cert_pem) 1509 1510 # Sign certificate 1511 signed_x509_cert_pem = \ 1512 utils.SignX509Certificate(x509_cert, cds, utils.GenerateSecret(8)) 1513 1514 cbs = _RemoteImportCb(feedback_fn, cds, signed_x509_cert_pem, 1515 len(instance.disks), pnode.primary_ip) 1516 1517 ieloop = ImportExportLoop(lu) 1518 inst_disks = lu.cfg.GetInstanceDisks(instance.uuid) 1519 try: 1520 for idx, dev in enumerate(inst_disks): 1521 magic = _GetInstDiskMagic(magic_base, instance.name, idx) 1522 1523 # Import daemon options 1524 opts = objects.ImportExportOptions(key_name=x509_key_name, 1525 ca_pem=source_ca_pem, 1526 magic=magic, 1527 compress=compress, 1528 ipv6=ipv6) 1529 1530 if instance.os: 1531 src_io = constants.IEIO_SCRIPT 1532 src_ioargs = ((dev, instance), idx) 1533 else: 1534 src_io = constants.IEIO_RAW_DISK 1535 src_ioargs = (dev, instance) 1536 1537 ieloop.Add(DiskImport(lu, instance.primary_node, opts, instance, 1538 "disk%d" % idx, 1539 src_io, src_ioargs, 1540 timeouts, cbs, private=(idx, ))) 1541 1542 ieloop.Run() 1543 finally: 1544 ieloop.FinalizeAll() 1545 finally: 1546 # Remove crypto key and certificate 1547 result = lu.rpc.call_x509_cert_remove(instance.primary_node, x509_key_name) 1548 result.Raise("Can't remove X509 key and certificate on %s" % result.node) 1549 1550 return cbs.disk_results

1551

1552 1553 -def _GetImportExportHandshakeMessage(version):

1554 """Returns the handshake message for a RIE protocol version. 1555 1556 @type version: number 1557 1558 """ 1559 return "%s:%s" % (version, constants.RIE_HANDSHAKE)

1560

1561 1562 -def ComputeRemoteExportHandshake(cds):

1563 """Computes the remote import/export handshake. 1564 1565 @type cds: string 1566 @param cds: Cluster domain secret 1567 1568 """ 1569 salt = utils.GenerateSecret(8) 1570 msg = _GetImportExportHandshakeMessage(constants.RIE_VERSION) 1571 return (constants.RIE_VERSION, utils.Sha1Hmac(cds, msg, salt=salt), salt)

1572

1573 1574 -def CheckRemoteExportHandshake(cds, handshake):

1575 """Checks the handshake of a remote import/export. 1576 1577 @type cds: string 1578 @param cds: Cluster domain secret 1579 @type handshake: sequence 1580 @param handshake: Handshake sent by remote peer 1581 1582 """ 1583 try: 1584 (version, hmac_digest, hmac_salt) = handshake 1585 except (TypeError, ValueError), err: 1586 return "Invalid data: %s" % err 1587 1588 if not utils.VerifySha1Hmac(cds, _GetImportExportHandshakeMessage(version), 1589 hmac_digest, salt=hmac_salt): 1590 return "Hash didn't match, clusters don't share the same domain secret" 1591 1592 if version != constants.RIE_VERSION: 1593 return ("Clusters don't have the same remote import/export protocol" 1594 " (local=%s, remote=%s)" % 1595 (constants.RIE_VERSION, version)) 1596 1597 return None

1598

1599 1600 -def _GetRieDiskInfoMessage(disk_index, host, port, magic):

1601 """Returns the hashed text for import/export disk information. 1602 1603 @type disk_index: number 1604 @param disk_index: Index of disk (included in hash) 1605 @type host: string 1606 @param host: Hostname 1607 @type port: number 1608 @param port: Daemon port 1609 @type magic: string 1610 @param magic: Magic value 1611 1612 """ 1613 return "%s:%s:%s:%s" % (disk_index, host, port, magic)

1614

1615 1616 -def CheckRemoteExportDiskInfo(cds, disk_index, disk_info):

1617 """Verifies received disk information for an export. 1618 1619 @type cds: string 1620 @param cds: Cluster domain secret 1621 @type disk_index: number 1622 @param disk_index: Index of disk (included in hash) 1623 @type disk_info: sequence 1624 @param disk_info: Disk information sent by remote peer 1625 1626 """ 1627 try: 1628 (host, port, magic, hmac_digest, hmac_salt) = disk_info 1629 except (TypeError, ValueError), err: 1630 raise errors.GenericError("Invalid data: %s" % err) 1631 1632 if not (host and port and magic): 1633 raise errors.GenericError("Missing destination host, port or magic") 1634 1635 msg = _GetRieDiskInfoMessage(disk_index, host, port, magic) 1636 1637 if not utils.VerifySha1Hmac(cds, msg, hmac_digest, salt=hmac_salt): 1638 raise errors.GenericError("HMAC is wrong") 1639 1640 if netutils.IP6Address.IsValid(host) or netutils.IP4Address.IsValid(host): 1641 destination = host 1642 else: 1643 destination = netutils.Hostname.GetNormalizedName(host) 1644 1645 return (destination, 1646 utils.ValidateServiceName(port), 1647 magic)

1648

1649 1650 -def ComputeRemoteImportDiskInfo(cds, salt, disk_index, host, port, magic):

1651 """Computes the signed disk information for a remote import. 1652 1653 @type cds: string 1654 @param cds: Cluster domain secret 1655 @type salt: string 1656 @param salt: HMAC salt 1657 @type disk_index: number 1658 @param disk_index: Index of disk (included in hash) 1659 @type host: string 1660 @param host: Hostname 1661 @type port: number 1662 @param port: Daemon port 1663 @type magic: string 1664 @param magic: Magic value 1665 1666 """ 1667 msg = _GetRieDiskInfoMessage(disk_index, host, port, magic) 1668 hmac_digest = utils.Sha1Hmac(cds, msg, salt=salt) 1669 return (host, port, magic, hmac_digest, salt)

1670

1671 1672 -def CalculateGroupIPolicy(cluster, group):

1673 """Calculate instance policy for group. 1674 1675 """ 1676 return cluster.SimpleFillIPolicy(group.ipolicy)

1677

1678 1679 -def ComputeDiskSize(disk_template, disks):

1680 """Compute disk size requirements according to disk template 1681 1682 """ 1683 # Required free disk space as a function of disk and swap space 1684 req_size_dict = { 1685 constants.DT_DISKLESS: 0, 1686 constants.DT_PLAIN: sum(d[constants.IDISK_SIZE] for d in disks), 1687 # 128 MB are added for drbd metadata for each disk 1688 constants.DT_DRBD8: 1689 sum(d[constants.IDISK_SIZE] + constants.DRBD_META_SIZE for d in disks), 1690 constants.DT_FILE: sum(d[constants.IDISK_SIZE] for d in disks), 1691 constants.DT_SHARED_FILE: sum(d[constants.IDISK_SIZE] for d in disks), 1692 constants.DT_GLUSTER: sum(d[constants.IDISK_SIZE] for d in disks), 1693 constants.DT_BLOCK: 0, 1694 constants.DT_RBD: sum(d[constants.IDISK_SIZE] for d in disks), 1695 constants.DT_EXT: sum(d[constants.IDISK_SIZE] for d in disks), 1696 } 1697 1698 if disk_template not in req_size_dict: 1699 raise errors.ProgrammerError("Disk template '%s' size requirement" 1700 " is unknown" % disk_template) 1701 1702 return req_size_dict[disk_template]

1703

Source Code for Module ganeti.masterd.instance