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: one of L{constants.IEC_ALL} 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 for idx, disk in enumerate(instance.disks): 1179 self._feedback_fn("Creating a snapshot of disk/%s on node %s" % 1180 (idx, src_node_name)) 1181 1182 # result.payload will be a snapshot of an lvm leaf of the one we 1183 # passed 1184 result = self._lu.rpc.call_blockdev_snapshot(src_node, (disk, instance)) 1185 new_dev = False 1186 msg = result.fail_msg 1187 if msg: 1188 self._lu.LogWarning("Could not snapshot disk/%s on node %s: %s", 1189 idx, src_node_name, msg) 1190 elif (not isinstance(result.payload, (tuple, list)) or 1191 len(result.payload) != 2): 1192 self._lu.LogWarning("Could not snapshot disk/%s on node %s: invalid" 1193 " result '%s'", idx, src_node_name, result.payload) 1194 else: 1195 disk_id = tuple(result.payload) 1196 disk_params = constants.DISK_LD_DEFAULTS[constants.DT_PLAIN].copy() 1197 new_dev = objects.Disk(dev_type=constants.DT_PLAIN, size=disk.size, 1198 logical_id=disk_id, iv_name=disk.iv_name, 1199 params=disk_params) 1200 1201 self._snap_disks.append(new_dev) 1202 1203 assert len(self._snap_disks) == len(instance.disks) 1204 assert len(self._removed_snaps) == len(instance.disks)

1205

1206 - def _RemoveSnapshot(self, disk_index):

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

1229

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

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

1290

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

1292 """Inter-cluster instance export. 1293 1294 @type disk_info: list 1295 @param disk_info: Per-disk destination information 1296 @type key_name: string 1297 @param key_name: Name of X509 key to use 1298 @type dest_ca_pem: string 1299 @param dest_ca_pem: Destination X509 CA in PEM format 1300 @type compress: string 1301 @param compress: one of L{constants.IEC_ALL} 1302 @type timeouts: L{ImportExportTimeouts} 1303 @param timeouts: Timeouts for this import 1304 1305 """ 1306 instance = self._instance 1307 1308 assert len(disk_info) == len(instance.disks) 1309 1310 cbs = _RemoteExportCb(self._feedback_fn, len(instance.disks)) 1311 1312 ieloop = ImportExportLoop(self._lu) 1313 try: 1314 for idx, (dev, (host, port, magic)) in enumerate(zip(instance.disks, 1315 disk_info)): 1316 # Decide whether to use IPv6 1317 ipv6 = netutils.IP6Address.IsValid(host) 1318 1319 opts = objects.ImportExportOptions(key_name=key_name, 1320 ca_pem=dest_ca_pem, 1321 magic=magic, 1322 compress=compress, 1323 ipv6=ipv6) 1324 1325 self._feedback_fn("Sending disk %s to %s:%s" % (idx, host, port)) 1326 finished_fn = compat.partial(self._TransferFinished, idx) 1327 ieloop.Add(DiskExport(self._lu, instance.primary_node, 1328 opts, host, port, instance, "disk%d" % idx, 1329 constants.IEIO_SCRIPT, ((dev, instance), idx), 1330 timeouts, cbs, private=(idx, finished_fn))) 1331 1332 ieloop.Run() 1333 finally: 1334 ieloop.FinalizeAll() 1335 1336 return (True, cbs.disk_results)

1337

1338 - def _TransferFinished(self, idx):

1339 """Called once a transfer has finished. 1340 1341 @type idx: number 1342 @param idx: Disk index 1343 1344 """ 1345 logging.debug("Transfer %s finished", idx) 1346 self._RemoveSnapshot(idx)

1347

1348 - def Cleanup(self):

1349 """Remove all snapshots. 1350 1351 """ 1352 assert len(self._removed_snaps) == len(self._instance.disks) 1353 for idx in range(len(self._instance.disks)): 1354 self._RemoveSnapshot(idx)

1355

1356 1357 -class _RemoteImportCb(ImportExportCbBase):

1358 - def __init__(self, feedback_fn, cds, x509_cert_pem, disk_count, 1359 external_address):

1360 """Initializes this class. 1361 1362 @type cds: string 1363 @param cds: Cluster domain secret 1364 @type x509_cert_pem: string 1365 @param x509_cert_pem: CA used for signing import key 1366 @type disk_count: number 1367 @param disk_count: Number of disks 1368 @type external_address: string 1369 @param external_address: External address of destination node 1370 1371 """ 1372 ImportExportCbBase.__init__(self) 1373 self._feedback_fn = feedback_fn 1374 self._cds = cds 1375 self._x509_cert_pem = x509_cert_pem 1376 self._disk_count = disk_count 1377 self._external_address = external_address 1378 1379 self._dresults = [None] * disk_count 1380 self._daemon_port = [None] * disk_count 1381 1382 self._salt = utils.GenerateSecret(8)

1383 1384 @property

1385 - def disk_results(self):

1386 """Returns per-disk results. 1387 1388 """ 1389 return self._dresults

1390

1391 - def _CheckAllListening(self):

1392 """Checks whether all daemons are listening. 1393 1394 If all daemons are listening, the information is sent to the client. 1395 1396 """ 1397 if not compat.all(dp is not None for dp in self._daemon_port): 1398 return 1399 1400 host = self._external_address 1401 1402 disks = [] 1403 for idx, (port, magic) in enumerate(self._daemon_port): 1404 disks.append(ComputeRemoteImportDiskInfo(self._cds, self._salt, 1405 idx, host, port, magic)) 1406 1407 assert len(disks) == self._disk_count 1408 1409 self._feedback_fn(constants.ELOG_REMOTE_IMPORT, { 1410 "disks": disks, 1411 "x509_ca": self._x509_cert_pem, 1412 })

1413

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

1415 """Called when daemon started listening. 1416 1417 """ 1418 (idx, ) = private 1419 1420 self._feedback_fn("Disk %s is now listening" % idx) 1421 1422 assert self._daemon_port[idx] is None 1423 1424 self._daemon_port[idx] = (ie.listen_port, ie.magic) 1425 1426 self._CheckAllListening()

1427

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

1429 """Called when a connection has been established. 1430 1431 """ 1432 (idx, ) = private 1433 1434 self._feedback_fn("Disk %s is now receiving data" % idx)

1435

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

1437 """Called when a transfer has finished. 1438 1439 """ 1440 (idx, ) = private 1441 1442 # Daemon is certainly no longer listening 1443 self._daemon_port[idx] = None 1444 1445 if ie.success: 1446 self._feedback_fn("Disk %s finished receiving data" % idx) 1447 else: 1448 self._feedback_fn(("Disk %s failed to receive data: %s" 1449 " (recent output: %s)") % 1450 (idx, ie.final_message, ie.recent_output)) 1451 1452 self._dresults[idx] = bool(ie.success)

1453

1454 1455 -def RemoteImport(lu, feedback_fn, instance, pnode, source_x509_ca, 1456 cds, compress, timeouts):

1457 """Imports an instance from another cluster. 1458 1459 @param lu: Logical unit instance 1460 @param feedback_fn: Feedback function 1461 @type instance: L{objects.Instance} 1462 @param instance: Instance object 1463 @type pnode: L{objects.Node} 1464 @param pnode: Primary node of instance as an object 1465 @type source_x509_ca: OpenSSL.crypto.X509 1466 @param source_x509_ca: Import source's X509 CA 1467 @type cds: string 1468 @param cds: Cluster domain secret 1469 @type compress: string 1470 @param compress: one of L{constants.IEC_ALL} 1471 @type timeouts: L{ImportExportTimeouts} 1472 @param timeouts: Timeouts for this import 1473 1474 """ 1475 source_ca_pem = OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, 1476 source_x509_ca) 1477 1478 magic_base = utils.GenerateSecret(6) 1479 1480 # Decide whether to use IPv6 1481 ipv6 = netutils.IP6Address.IsValid(pnode.primary_ip) 1482 1483 # Create crypto key 1484 result = lu.rpc.call_x509_cert_create(instance.primary_node, 1485 constants.RIE_CERT_VALIDITY) 1486 result.Raise("Can't create X509 key and certificate on %s" % result.node) 1487 1488 (x509_key_name, x509_cert_pem) = result.payload 1489 try: 1490 # Load certificate 1491 x509_cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, 1492 x509_cert_pem) 1493 1494 # Sign certificate 1495 signed_x509_cert_pem = \ 1496 utils.SignX509Certificate(x509_cert, cds, utils.GenerateSecret(8)) 1497 1498 cbs = _RemoteImportCb(feedback_fn, cds, signed_x509_cert_pem, 1499 len(instance.disks), pnode.primary_ip) 1500 1501 ieloop = ImportExportLoop(lu) 1502 try: 1503 for idx, dev in enumerate(instance.disks): 1504 magic = _GetInstDiskMagic(magic_base, instance.name, idx) 1505 1506 # Import daemon options 1507 opts = objects.ImportExportOptions(key_name=x509_key_name, 1508 ca_pem=source_ca_pem, 1509 magic=magic, 1510 compress=compress, 1511 ipv6=ipv6) 1512 1513 ieloop.Add(DiskImport(lu, instance.primary_node, opts, instance, 1514 "disk%d" % idx, 1515 constants.IEIO_SCRIPT, ((dev, instance), idx), 1516 timeouts, cbs, private=(idx, ))) 1517 1518 ieloop.Run() 1519 finally: 1520 ieloop.FinalizeAll() 1521 finally: 1522 # Remove crypto key and certificate 1523 result = lu.rpc.call_x509_cert_remove(instance.primary_node, x509_key_name) 1524 result.Raise("Can't remove X509 key and certificate on %s" % result.node) 1525 1526 return cbs.disk_results

1527

1528 1529 -def _GetImportExportHandshakeMessage(version):

1530 """Returns the handshake message for a RIE protocol version. 1531 1532 @type version: number 1533 1534 """ 1535 return "%s:%s" % (version, constants.RIE_HANDSHAKE)

1536

1537 1538 -def ComputeRemoteExportHandshake(cds):

1539 """Computes the remote import/export handshake. 1540 1541 @type cds: string 1542 @param cds: Cluster domain secret 1543 1544 """ 1545 salt = utils.GenerateSecret(8) 1546 msg = _GetImportExportHandshakeMessage(constants.RIE_VERSION) 1547 return (constants.RIE_VERSION, utils.Sha1Hmac(cds, msg, salt=salt), salt)

1548

1549 1550 -def CheckRemoteExportHandshake(cds, handshake):

1551 """Checks the handshake of a remote import/export. 1552 1553 @type cds: string 1554 @param cds: Cluster domain secret 1555 @type handshake: sequence 1556 @param handshake: Handshake sent by remote peer 1557 1558 """ 1559 try: 1560 (version, hmac_digest, hmac_salt) = handshake 1561 except (TypeError, ValueError), err: 1562 return "Invalid data: %s" % err 1563 1564 if not utils.VerifySha1Hmac(cds, _GetImportExportHandshakeMessage(version), 1565 hmac_digest, salt=hmac_salt): 1566 return "Hash didn't match, clusters don't share the same domain secret" 1567 1568 if version != constants.RIE_VERSION: 1569 return ("Clusters don't have the same remote import/export protocol" 1570 " (local=%s, remote=%s)" % 1571 (constants.RIE_VERSION, version)) 1572 1573 return None

1574

1575 1576 -def _GetRieDiskInfoMessage(disk_index, host, port, magic):

1577 """Returns the hashed text for import/export disk information. 1578 1579 @type disk_index: number 1580 @param disk_index: Index of disk (included in hash) 1581 @type host: string 1582 @param host: Hostname 1583 @type port: number 1584 @param port: Daemon port 1585 @type magic: string 1586 @param magic: Magic value 1587 1588 """ 1589 return "%s:%s:%s:%s" % (disk_index, host, port, magic)

1590

1591 1592 -def CheckRemoteExportDiskInfo(cds, disk_index, disk_info):

1593 """Verifies received disk information for an export. 1594 1595 @type cds: string 1596 @param cds: Cluster domain secret 1597 @type disk_index: number 1598 @param disk_index: Index of disk (included in hash) 1599 @type disk_info: sequence 1600 @param disk_info: Disk information sent by remote peer 1601 1602 """ 1603 try: 1604 (host, port, magic, hmac_digest, hmac_salt) = disk_info 1605 except (TypeError, ValueError), err: 1606 raise errors.GenericError("Invalid data: %s" % err) 1607 1608 if not (host and port and magic): 1609 raise errors.GenericError("Missing destination host, port or magic") 1610 1611 msg = _GetRieDiskInfoMessage(disk_index, host, port, magic) 1612 1613 if not utils.VerifySha1Hmac(cds, msg, hmac_digest, salt=hmac_salt): 1614 raise errors.GenericError("HMAC is wrong") 1615 1616 if netutils.IP6Address.IsValid(host) or netutils.IP4Address.IsValid(host): 1617 destination = host 1618 else: 1619 destination = netutils.Hostname.GetNormalizedName(host) 1620 1621 return (destination, 1622 utils.ValidateServiceName(port), 1623 magic)

1624

1625 1626 -def ComputeRemoteImportDiskInfo(cds, salt, disk_index, host, port, magic):

1627 """Computes the signed disk information for a remote import. 1628 1629 @type cds: string 1630 @param cds: Cluster domain secret 1631 @type salt: string 1632 @param salt: HMAC salt 1633 @type disk_index: number 1634 @param disk_index: Index of disk (included in hash) 1635 @type host: string 1636 @param host: Hostname 1637 @type port: number 1638 @param port: Daemon port 1639 @type magic: string 1640 @param magic: Magic value 1641 1642 """ 1643 msg = _GetRieDiskInfoMessage(disk_index, host, port, magic) 1644 hmac_digest = utils.Sha1Hmac(cds, msg, salt=salt) 1645 return (host, port, magic, hmac_digest, salt)

1646

1647 1648 -def CalculateGroupIPolicy(cluster, group):

1649 """Calculate instance policy for group. 1650 1651 """ 1652 return cluster.SimpleFillIPolicy(group.ipolicy)

1653

1654 1655 -def ComputeDiskSize(disk_template, disks):

1656 """Compute disk size requirements according to disk template 1657 1658 """ 1659 # Required free disk space as a function of disk and swap space 1660 req_size_dict = { 1661 constants.DT_DISKLESS: 0, 1662 constants.DT_PLAIN: sum(d[constants.IDISK_SIZE] for d in disks), 1663 # 128 MB are added for drbd metadata for each disk 1664 constants.DT_DRBD8: 1665 sum(d[constants.IDISK_SIZE] + constants.DRBD_META_SIZE for d in disks), 1666 constants.DT_FILE: sum(d[constants.IDISK_SIZE] for d in disks), 1667 constants.DT_SHARED_FILE: sum(d[constants.IDISK_SIZE] for d in disks), 1668 constants.DT_GLUSTER: sum(d[constants.IDISK_SIZE] for d in disks), 1669 constants.DT_BLOCK: 0, 1670 constants.DT_RBD: sum(d[constants.IDISK_SIZE] for d in disks), 1671 constants.DT_EXT: sum(d[constants.IDISK_SIZE] for d in disks), 1672 } 1673 1674 if disk_template not in req_size_dict: 1675 raise errors.ProgrammerError("Disk template '%s' size requirement" 1676 " is unknown" % disk_template) 1677 1678 return req_size_dict[disk_template]

1679

Source Code for Module ganeti.masterd.instance