ganeti.cmdlib.backup

1 # 2 # 3 4 # Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 Google Inc. 5 # 6 # This program is free software; you can redistribute it and/or modify 7 # it under the terms of the GNU General Public License as published by 8 # the Free Software Foundation; either version 2 of the License, or 9 # (at your option) any later version. 10 # 11 # This program is distributed in the hope that it will be useful, but 12 # WITHOUT ANY WARRANTY; without even the implied warranty of 13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 14 # General Public License for more details. 15 # 16 # You should have received a copy of the GNU General Public License 17 # along with this program; if not, write to the Free Software 18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 19 # 02110-1301, USA. 20 21 22 """Logical units dealing with backup operations.""" 23 24 import OpenSSL 25 import logging 26 27 from ganeti import compat 28 from ganeti import constants 29 from ganeti import errors 30 from ganeti import locking 31 from ganeti import masterd 32 from ganeti import qlang 33 from ganeti import query 34 from ganeti import utils 35 36 from ganeti.cmdlib.base import QueryBase, NoHooksLU, LogicalUnit 37 from ganeti.cmdlib.common import GetWantedNodes, ShareAll, CheckNodeOnline, \ 38 ExpandNodeName 39 from ganeti.cmdlib.instance_storage import StartInstanceDisks, \ 40 ShutdownInstanceDisks 41 from ganeti.cmdlib.instance_utils import GetClusterDomainSecret, \ 42 BuildInstanceHookEnvByObject, CheckNodeNotDrained, RemoveInstance 43 44

45 -class ExportQuery(QueryBase):

46 FIELDS = query.EXPORT_FIELDS 47 48 #: The node name is not a unique key for this query 49 SORT_FIELD = "node" 50

51 - def ExpandNames(self, lu):

52 lu.needed_locks = {} 53 54 # The following variables interact with _QueryBase._GetNames 55 if self.names: 56 self.wanted = GetWantedNodes(lu, self.names) 57 else: 58 self.wanted = locking.ALL_SET 59 60 self.do_locking = self.use_locking 61 62 if self.do_locking: 63 lu.share_locks = ShareAll() 64 lu.needed_locks = { 65 locking.LEVEL_NODE: self.wanted, 66 } 67 68 if not self.names: 69 lu.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET

70

71 - def DeclareLocks(self, lu, level):

72 pass

73

74 - def _GetQueryData(self, lu):

75 """Computes the list of nodes and their attributes. 76 77 """ 78 # Locking is not used 79 # TODO 80 assert not (compat.any(lu.glm.is_owned(level) 81 for level in locking.LEVELS 82 if level != locking.LEVEL_CLUSTER) or 83 self.do_locking or self.use_locking) 84 85 nodes = self._GetNames(lu, lu.cfg.GetNodeList(), locking.LEVEL_NODE) 86 87 result = [] 88 89 for (node, nres) in lu.rpc.call_export_list(nodes).items(): 90 if nres.fail_msg: 91 result.append((node, None)) 92 else: 93 result.extend((node, expname) for expname in nres.payload) 94 95 return result

96 97

98 -class LUBackupQuery(NoHooksLU):

99 """Query the exports list 100 101 """ 102 REQ_BGL = False 103

104 - def CheckArguments(self):

105 self.expq = ExportQuery(qlang.MakeSimpleFilter("node", self.op.nodes), 106 ["node", "export"], self.op.use_locking)

107

108 - def ExpandNames(self):

109 self.expq.ExpandNames(self)

110

111 - def DeclareLocks(self, level):

112 self.expq.DeclareLocks(self, level)

113

114 - def Exec(self, feedback_fn):

115 result = {} 116 117 for (node, expname) in self.expq.OldStyleQuery(self): 118 if expname is None: 119 result[node] = False 120 else: 121 result.setdefault(node, []).append(expname) 122 123 return result

124 125

126 -class LUBackupPrepare(NoHooksLU):

127 """Prepares an instance for an export and returns useful information. 128 129 """ 130 REQ_BGL = False 131

132 - def ExpandNames(self):

133 self._ExpandAndLockInstance()

134

135 - def CheckPrereq(self):

136 """Check prerequisites. 137 138 """ 139 instance_name = self.op.instance_name 140 141 self.instance = self.cfg.GetInstanceInfo(instance_name) 142 assert self.instance is not None, \ 143 "Cannot retrieve locked instance %s" % self.op.instance_name 144 CheckNodeOnline(self, self.instance.primary_node) 145 146 self._cds = GetClusterDomainSecret()

147

148 - def Exec(self, feedback_fn):

149 """Prepares an instance for an export. 150 151 """ 152 instance = self.instance 153 154 if self.op.mode == constants.EXPORT_MODE_REMOTE: 155 salt = utils.GenerateSecret(8) 156 157 feedback_fn("Generating X509 certificate on %s" % instance.primary_node) 158 result = self.rpc.call_x509_cert_create(instance.primary_node, 159 constants.RIE_CERT_VALIDITY) 160 result.Raise("Can't create X509 key and certificate on %s" % result.node) 161 162 (name, cert_pem) = result.payload 163 164 cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, 165 cert_pem) 166 167 return { 168 "handshake": masterd.instance.ComputeRemoteExportHandshake(self._cds), 169 "x509_key_name": (name, utils.Sha1Hmac(self._cds, name, salt=salt), 170 salt), 171 "x509_ca": utils.SignX509Certificate(cert, self._cds, salt), 172 } 173 174 return None

175 176

177 -class LUBackupExport(LogicalUnit):

178 """Export an instance to an image in the cluster. 179 180 """ 181 HPATH = "instance-export" 182 HTYPE = constants.HTYPE_INSTANCE 183 REQ_BGL = False 184

185 - def CheckArguments(self):

186 """Check the arguments. 187 188 """ 189 self.x509_key_name = self.op.x509_key_name 190 self.dest_x509_ca_pem = self.op.destination_x509_ca 191 192 if self.op.mode == constants.EXPORT_MODE_REMOTE: 193 if not self.x509_key_name: 194 raise errors.OpPrereqError("Missing X509 key name for encryption", 195 errors.ECODE_INVAL) 196 197 if not self.dest_x509_ca_pem: 198 raise errors.OpPrereqError("Missing destination X509 CA", 199 errors.ECODE_INVAL)

200

201 - def ExpandNames(self):

202 self._ExpandAndLockInstance() 203 204 # Lock all nodes for local exports 205 if self.op.mode == constants.EXPORT_MODE_LOCAL: 206 # FIXME: lock only instance primary and destination node 207 # 208 # Sad but true, for now we have do lock all nodes, as we don't know where 209 # the previous export might be, and in this LU we search for it and 210 # remove it from its current node. In the future we could fix this by: 211 # - making a tasklet to search (share-lock all), then create the 212 # new one, then one to remove, after 213 # - removing the removal operation altogether 214 self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET 215 216 # Allocations should be stopped while this LU runs with node locks, but 217 # it doesn't have to be exclusive 218 self.share_locks[locking.LEVEL_NODE_ALLOC] = 1 219 self.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET

220

221 - def DeclareLocks(self, level):

222 """Last minute lock declaration."""

223 # All nodes are locked anyway, so nothing to do here. 224

225 - def BuildHooksEnv(self):

226 """Build hooks env. 227 228 This will run on the master, primary node and target node. 229 230 """ 231 env = { 232 "EXPORT_MODE": self.op.mode, 233 "EXPORT_NODE": self.op.target_node, 234 "EXPORT_DO_SHUTDOWN": self.op.shutdown, 235 "SHUTDOWN_TIMEOUT": self.op.shutdown_timeout, 236 # TODO: Generic function for boolean env variables 237 "REMOVE_INSTANCE": str(bool(self.op.remove_instance)), 238 } 239 240 env.update(BuildInstanceHookEnvByObject(self, self.instance)) 241 242 return env

243

244 - def BuildHooksNodes(self):

245 """Build hooks nodes. 246 247 """ 248 nl = [self.cfg.GetMasterNode(), self.instance.primary_node] 249 250 if self.op.mode == constants.EXPORT_MODE_LOCAL: 251 nl.append(self.op.target_node) 252 253 return (nl, nl)

254

255 - def CheckPrereq(self):

256 """Check prerequisites. 257 258 This checks that the instance and node names are valid. 259 260 """ 261 instance_name = self.op.instance_name 262 263 self.instance = self.cfg.GetInstanceInfo(instance_name) 264 assert self.instance is not None, \ 265 "Cannot retrieve locked instance %s" % self.op.instance_name 266 CheckNodeOnline(self, self.instance.primary_node) 267 268 if (self.op.remove_instance and 269 self.instance.admin_state == constants.ADMINST_UP and 270 not self.op.shutdown): 271 raise errors.OpPrereqError("Can not remove instance without shutting it" 272 " down before", errors.ECODE_STATE) 273 274 if self.op.mode == constants.EXPORT_MODE_LOCAL: 275 self.op.target_node = ExpandNodeName(self.cfg, self.op.target_node) 276 self.dst_node = self.cfg.GetNodeInfo(self.op.target_node) 277 assert self.dst_node is not None 278 279 CheckNodeOnline(self, self.dst_node.name) 280 CheckNodeNotDrained(self, self.dst_node.name) 281 282 self._cds = None 283 self.dest_disk_info = None 284 self.dest_x509_ca = None 285 286 elif self.op.mode == constants.EXPORT_MODE_REMOTE: 287 self.dst_node = None 288 289 if len(self.op.target_node) != len(self.instance.disks): 290 raise errors.OpPrereqError(("Received destination information for %s" 291 " disks, but instance %s has %s disks") % 292 (len(self.op.target_node), instance_name, 293 len(self.instance.disks)), 294 errors.ECODE_INVAL) 295 296 cds = GetClusterDomainSecret() 297 298 # Check X509 key name 299 try: 300 (key_name, hmac_digest, hmac_salt) = self.x509_key_name 301 except (TypeError, ValueError), err: 302 raise errors.OpPrereqError("Invalid data for X509 key name: %s" % err, 303 errors.ECODE_INVAL) 304 305 if not utils.VerifySha1Hmac(cds, key_name, hmac_digest, salt=hmac_salt): 306 raise errors.OpPrereqError("HMAC for X509 key name is wrong", 307 errors.ECODE_INVAL) 308 309 # Load and verify CA 310 try: 311 (cert, _) = utils.LoadSignedX509Certificate(self.dest_x509_ca_pem, cds) 312 except OpenSSL.crypto.Error, err: 313 raise errors.OpPrereqError("Unable to load destination X509 CA (%s)" % 314 (err, ), errors.ECODE_INVAL) 315 316 (errcode, msg) = utils.VerifyX509Certificate(cert, None, None) 317 if errcode is not None: 318 raise errors.OpPrereqError("Invalid destination X509 CA (%s)" % 319 (msg, ), errors.ECODE_INVAL) 320 321 self.dest_x509_ca = cert 322 323 # Verify target information 324 disk_info = [] 325 for idx, disk_data in enumerate(self.op.target_node): 326 try: 327 (host, port, magic) = \ 328 masterd.instance.CheckRemoteExportDiskInfo(cds, idx, disk_data) 329 except errors.GenericError, err: 330 raise errors.OpPrereqError("Target info for disk %s: %s" % 331 (idx, err), errors.ECODE_INVAL) 332 333 disk_info.append((host, port, magic)) 334 335 assert len(disk_info) == len(self.op.target_node) 336 self.dest_disk_info = disk_info 337 338 else: 339 raise errors.ProgrammerError("Unhandled export mode %r" % 340 self.op.mode) 341 342 # instance disk type verification 343 # TODO: Implement export support for file-based disks 344 for disk in self.instance.disks: 345 if disk.dev_type == constants.LD_FILE: 346 raise errors.OpPrereqError("Export not supported for instances with" 347 " file-based disks", errors.ECODE_INVAL)

348

349 - def _CleanupExports(self, feedback_fn):

350 """Removes exports of current instance from all other nodes. 351 352 If an instance in a cluster with nodes A..D was exported to node C, its 353 exports will be removed from the nodes A, B and D. 354 355 """ 356 assert self.op.mode != constants.EXPORT_MODE_REMOTE 357 358 nodelist = self.cfg.GetNodeList() 359 nodelist.remove(self.dst_node.name) 360 361 # on one-node clusters nodelist will be empty after the removal 362 # if we proceed the backup would be removed because OpBackupQuery 363 # substitutes an empty list with the full cluster node list. 364 iname = self.instance.name 365 if nodelist: 366 feedback_fn("Removing old exports for instance %s" % iname) 367 exportlist = self.rpc.call_export_list(nodelist) 368 for node in exportlist: 369 if exportlist[node].fail_msg: 370 continue 371 if iname in exportlist[node].payload: 372 msg = self.rpc.call_export_remove(node, iname).fail_msg 373 if msg: 374 self.LogWarning("Could not remove older export for instance %s" 375 " on node %s: %s", iname, node, msg)

376

377 - def Exec(self, feedback_fn):

378 """Export an instance to an image in the cluster. 379 380 """ 381 assert self.op.mode in constants.EXPORT_MODES 382 383 instance = self.instance 384 src_node = instance.primary_node 385 386 if self.op.shutdown: 387 # shutdown the instance, but not the disks 388 feedback_fn("Shutting down instance %s" % instance.name) 389 result = self.rpc.call_instance_shutdown(src_node, instance, 390 self.op.shutdown_timeout, 391 self.op.reason) 392 # TODO: Maybe ignore failures if ignore_remove_failures is set 393 result.Raise("Could not shutdown instance %s on" 394 " node %s" % (instance.name, src_node)) 395 396 # set the disks ID correctly since call_instance_start needs the 397 # correct drbd minor to create the symlinks 398 for disk in instance.disks: 399 self.cfg.SetDiskID(disk, src_node) 400 401 activate_disks = not instance.disks_active 402 403 if activate_disks: 404 # Activate the instance disks if we'exporting a stopped instance 405 feedback_fn("Activating disks for %s" % instance.name) 406 StartInstanceDisks(self, instance, None) 407 408 try: 409 helper = masterd.instance.ExportInstanceHelper(self, feedback_fn, 410 instance) 411 412 helper.CreateSnapshots() 413 try: 414 if (self.op.shutdown and 415 instance.admin_state == constants.ADMINST_UP and 416 not self.op.remove_instance): 417 assert not activate_disks 418 feedback_fn("Starting instance %s" % instance.name) 419 result = self.rpc.call_instance_start(src_node, 420 (instance, None, None), False, 421 self.op.reason) 422 msg = result.fail_msg 423 if msg: 424 feedback_fn("Failed to start instance: %s" % msg) 425 ShutdownInstanceDisks(self, instance) 426 raise errors.OpExecError("Could not start instance: %s" % msg) 427 428 if self.op.mode == constants.EXPORT_MODE_LOCAL: 429 (fin_resu, dresults) = helper.LocalExport(self.dst_node) 430 elif self.op.mode == constants.EXPORT_MODE_REMOTE: 431 connect_timeout = constants.RIE_CONNECT_TIMEOUT 432 timeouts = masterd.instance.ImportExportTimeouts(connect_timeout) 433 434 (key_name, _, _) = self.x509_key_name 435 436 dest_ca_pem = \ 437 OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, 438 self.dest_x509_ca) 439 440 (fin_resu, dresults) = helper.RemoteExport(self.dest_disk_info, 441 key_name, dest_ca_pem, 442 timeouts) 443 finally: 444 helper.Cleanup() 445 446 # Check for backwards compatibility 447 assert len(dresults) == len(instance.disks) 448 assert compat.all(isinstance(i, bool) for i in dresults), \ 449 "Not all results are boolean: %r" % dresults 450 451 finally: 452 if activate_disks: 453 feedback_fn("Deactivating disks for %s" % instance.name) 454 ShutdownInstanceDisks(self, instance) 455 456 if not (compat.all(dresults) and fin_resu): 457 failures = [] 458 if not fin_resu: 459 failures.append("export finalization") 460 if not compat.all(dresults): 461 fdsk = utils.CommaJoin(idx for (idx, dsk) in enumerate(dresults) 462 if not dsk) 463 failures.append("disk export: disk(s) %s" % fdsk) 464 465 raise errors.OpExecError("Export failed, errors in %s" % 466 utils.CommaJoin(failures)) 467 468 # At this point, the export was successful, we can cleanup/finish 469 470 # Remove instance if requested 471 if self.op.remove_instance: 472 feedback_fn("Removing instance %s" % instance.name) 473 RemoveInstance(self, feedback_fn, instance, 474 self.op.ignore_remove_failures) 475 476 if self.op.mode == constants.EXPORT_MODE_LOCAL: 477 self._CleanupExports(feedback_fn) 478 479 return fin_resu, dresults

480 481

482 -class LUBackupRemove(NoHooksLU):

483 """Remove exports related to the named instance. 484 485 """ 486 REQ_BGL = False 487

488 - def ExpandNames(self):

489 self.needed_locks = { 490 # We need all nodes to be locked in order for RemoveExport to work, but 491 # we don't need to lock the instance itself, as nothing will happen to it 492 # (and we can remove exports also for a removed instance) 493 locking.LEVEL_NODE: locking.ALL_SET, 494 495 # Removing backups is quick, so blocking allocations is justified 496 locking.LEVEL_NODE_ALLOC: locking.ALL_SET, 497 } 498 499 # Allocations should be stopped while this LU runs with node locks, but it 500 # doesn't have to be exclusive 501 self.share_locks[locking.LEVEL_NODE_ALLOC] = 1

502

503 - def Exec(self, feedback_fn):

504 """Remove any export. 505 506 """ 507 instance_name = self.cfg.ExpandInstanceName(self.op.instance_name) 508 # If the instance was not found we'll try with the name that was passed in. 509 # This will only work if it was an FQDN, though. 510 fqdn_warn = False 511 if not instance_name: 512 fqdn_warn = True 513 instance_name = self.op.instance_name 514 515 locked_nodes = self.owned_locks(locking.LEVEL_NODE) 516 exportlist = self.rpc.call_export_list(locked_nodes) 517 found = False 518 for node in exportlist: 519 msg = exportlist[node].fail_msg 520 if msg: 521 self.LogWarning("Failed to query node %s (continuing): %s", node, msg) 522 continue 523 if instance_name in exportlist[node].payload: 524 found = True 525 result = self.rpc.call_export_remove(node, instance_name) 526 msg = result.fail_msg 527 if msg: 528 logging.error("Could not remove export for instance %s" 529 " on node %s: %s", instance_name, node, msg) 530 531 if fqdn_warn and not found: 532 feedback_fn("Export not found. If trying to remove an export belonging" 533 " to a deleted instance please use its Fully Qualified" 534 " Domain Name.")

535

Source Code for Module ganeti.cmdlib.backup