ganeti.confd.client

1 # 2 # 3 4 # Copyright (C) 2009, 2010, 2012 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 """Ganeti confd client 23 24 Clients can use the confd client library to send requests to a group of master 25 candidates running confd. The expected usage is through the asyncore framework, 26 by sending queries, and asynchronously receiving replies through a callback. 27 28 This way the client library doesn't ever need to "wait" on a particular answer, 29 and can proceed even if some udp packets are lost. It's up to the user to 30 reschedule queries if they haven't received responses and they need them. 31 32 Example usage:: 33 34 client = ConfdClient(...) # includes callback specification 35 req = confd_client.ConfdClientRequest(type=constants.CONFD_REQ_PING) 36 client.SendRequest(req) 37 # then make sure your client calls asyncore.loop() or daemon.Mainloop.Run() 38 # ... wait ... 39 # And your callback will be called by asyncore, when your query gets a 40 # response, or when it expires. 41 42 You can use the provided ConfdFilterCallback to act as a filter, only passing 43 "newer" answer to your callback, and filtering out outdated ones, or ones 44 confirming what you already got. 45 46 """ 47 48 # pylint: disable=E0203 49 50 # E0203: Access to member %r before its definition, since we use 51 # objects.py which doesn't explicitly initialise its members 52 53 import time 54 import random 55 56 from ganeti import utils 57 from ganeti import constants 58 from ganeti import objects 59 from ganeti import serializer 60 from ganeti import daemon # contains AsyncUDPSocket 61 from ganeti import errors 62 from ganeti import confd 63 from ganeti import ssconf 64 from ganeti import compat 65 from ganeti import netutils 66 from ganeti import pathutils

67 68 69 -class ConfdAsyncUDPClient(daemon.AsyncUDPSocket):

70 """Confd udp asyncore client 71 72 This is kept separate from the main ConfdClient to make sure it's easy to 73 implement a non-asyncore based client library. 74 75 """

76 - def __init__(self, client, family):

77 """Constructor for ConfdAsyncUDPClient 78 79 @type client: L{ConfdClient} 80 @param client: client library, to pass the datagrams to 81 82 """ 83 daemon.AsyncUDPSocket.__init__(self, family) 84 self.client = client

85 86 # this method is overriding a daemon.AsyncUDPSocket method

87 - def handle_datagram(self, payload, ip, port):

88 self.client.HandleResponse(payload, ip, port)

89

90 91 -class _Request(object):

92 """Request status structure. 93 94 @ivar request: the request data 95 @ivar args: any extra arguments for the callback 96 @ivar expiry: the expiry timestamp of the request 97 @ivar sent: the set of contacted peers 98 @ivar rcvd: the set of peers who replied 99 100 """

101 - def __init__(self, request, args, expiry, sent):

102 self.request = request 103 self.args = args 104 self.expiry = expiry 105 self.sent = frozenset(sent) 106 self.rcvd = set()

107

108 109 -class ConfdClient:

110 """Send queries to confd, and get back answers. 111 112 Since the confd model works by querying multiple master candidates, and 113 getting back answers, this is an asynchronous library. It can either work 114 through asyncore or with your own handling. 115 116 @type _requests: dict 117 @ivar _requests: dictionary indexes by salt, which contains data 118 about the outstanding requests; the values are objects of type 119 L{_Request} 120 121 """

122 - def __init__(self, hmac_key, peers, callback, port=None, logger=None):

123 """Constructor for ConfdClient 124 125 @type hmac_key: string 126 @param hmac_key: hmac key to talk to confd 127 @type peers: list 128 @param peers: list of peer nodes 129 @type callback: f(L{ConfdUpcallPayload}) 130 @param callback: function to call when getting answers 131 @type port: integer 132 @param port: confd port (default: use GetDaemonPort) 133 @type logger: logging.Logger 134 @param logger: optional logger for internal conditions 135 136 """ 137 if not callable(callback): 138 raise errors.ProgrammerError("callback must be callable") 139 140 self.UpdatePeerList(peers) 141 self._SetPeersAddressFamily() 142 self._hmac_key = hmac_key 143 self._socket = ConfdAsyncUDPClient(self, self._family) 144 self._callback = callback 145 self._confd_port = port 146 self._logger = logger 147 self._requests = {} 148 149 if self._confd_port is None: 150 self._confd_port = netutils.GetDaemonPort(constants.CONFD)

151

152 - def UpdatePeerList(self, peers):

153 """Update the list of peers 154 155 @type peers: list 156 @param peers: list of peer nodes 157 158 """ 159 # we are actually called from init, so: 160 # pylint: disable=W0201 161 if not isinstance(peers, list): 162 raise errors.ProgrammerError("peers must be a list") 163 # make a copy of peers, since we're going to shuffle the list, later 164 self._peers = list(peers)

165

166 - def _PackRequest(self, request, now=None):

167 """Prepare a request to be sent on the wire. 168 169 This function puts a proper salt in a confd request, puts the proper salt, 170 and adds the correct magic number. 171 172 """ 173 if now is None: 174 now = time.time() 175 tstamp = "%d" % now 176 req = serializer.DumpSignedJson(request.ToDict(), self._hmac_key, tstamp) 177 return confd.PackMagic(req)

178

179 - def _UnpackReply(self, payload):

180 in_payload = confd.UnpackMagic(payload) 181 (dict_answer, salt) = serializer.LoadSignedJson(in_payload, self._hmac_key) 182 answer = objects.ConfdReply.FromDict(dict_answer) 183 return answer, salt

184

185 - def ExpireRequests(self):

186 """Delete all the expired requests. 187 188 """ 189 now = time.time() 190 for rsalt, rq in self._requests.items(): 191 if now >= rq.expiry: 192 del self._requests[rsalt] 193 client_reply = ConfdUpcallPayload(salt=rsalt, 194 type=UPCALL_EXPIRE, 195 orig_request=rq.request, 196 extra_args=rq.args, 197 client=self, 198 ) 199 self._callback(client_reply)

200

201 - def SendRequest(self, request, args=None, coverage=0, async=True):

202 """Send a confd request to some MCs 203 204 @type request: L{objects.ConfdRequest} 205 @param request: the request to send 206 @type args: tuple 207 @param args: additional callback arguments 208 @type coverage: integer 209 @param coverage: number of remote nodes to contact; if default 210 (0), it will use a reasonable default 211 (L{ganeti.constants.CONFD_DEFAULT_REQ_COVERAGE}), if -1 is 212 passed, it will use the maximum number of peers, otherwise the 213 number passed in will be used 214 @type async: boolean 215 @param async: handle the write asynchronously 216 217 """ 218 if coverage == 0: 219 coverage = min(len(self._peers), constants.CONFD_DEFAULT_REQ_COVERAGE) 220 elif coverage == -1: 221 coverage = len(self._peers) 222 223 if coverage > len(self._peers): 224 raise errors.ConfdClientError("Not enough MCs known to provide the" 225 " desired coverage") 226 227 if not request.rsalt: 228 raise errors.ConfdClientError("Missing request rsalt") 229 230 self.ExpireRequests() 231 if request.rsalt in self._requests: 232 raise errors.ConfdClientError("Duplicate request rsalt") 233 234 if request.type not in constants.CONFD_REQS: 235 raise errors.ConfdClientError("Invalid request type") 236 237 random.shuffle(self._peers) 238 targets = self._peers[:coverage] 239 240 now = time.time() 241 payload = self._PackRequest(request, now=now) 242 243 for target in targets: 244 try: 245 self._socket.enqueue_send(target, self._confd_port, payload) 246 except errors.UdpDataSizeError: 247 raise errors.ConfdClientError("Request too big") 248 249 expire_time = now + constants.CONFD_CLIENT_EXPIRE_TIMEOUT 250 self._requests[request.rsalt] = _Request(request, args, expire_time, 251 targets) 252 253 if not async: 254 self.FlushSendQueue()

255

256 - def HandleResponse(self, payload, ip, port):

257 """Asynchronous handler for a confd reply 258 259 Call the relevant callback associated to the current request. 260 261 """ 262 try: 263 try: 264 answer, salt = self._UnpackReply(payload) 265 except (errors.SignatureError, errors.ConfdMagicError), err: 266 if self._logger: 267 self._logger.debug("Discarding broken package: %s" % err) 268 return 269 270 try: 271 rq = self._requests[salt] 272 except KeyError: 273 if self._logger: 274 self._logger.debug("Discarding unknown (expired?) reply: %s" % err) 275 return 276 277 rq.rcvd.add(ip) 278 279 client_reply = ConfdUpcallPayload(salt=salt, 280 type=UPCALL_REPLY, 281 server_reply=answer, 282 orig_request=rq.request, 283 server_ip=ip, 284 server_port=port, 285 extra_args=rq.args, 286 client=self, 287 ) 288 self._callback(client_reply) 289 290 finally: 291 self.ExpireRequests()

292

293 - def FlushSendQueue(self):

294 """Send out all pending requests. 295 296 Can be used for synchronous client use. 297 298 """ 299 while self._socket.writable(): 300 self._socket.handle_write()

301

302 - def ReceiveReply(self, timeout=1):

303 """Receive one reply. 304 305 @type timeout: float 306 @param timeout: how long to wait for the reply 307 @rtype: boolean 308 @return: True if some data has been handled, False otherwise 309 310 """ 311 return self._socket.process_next_packet(timeout=timeout)

312 313 @staticmethod

314 - def _NeededReplies(peer_cnt):

315 """Compute the minimum safe number of replies for a query. 316 317 The algorithm is designed to work well for both small and big 318 number of peers: 319 - for less than three, we require all responses 320 - for less than five, we allow one miss 321 - otherwise, half the number plus one 322 323 This guarantees that we progress monotonically: 1->1, 2->2, 3->2, 324 4->2, 5->3, 6->3, 7->4, etc. 325 326 @type peer_cnt: int 327 @param peer_cnt: the number of peers contacted 328 @rtype: int 329 @return: the number of replies which should give a safe coverage 330 331 """ 332 if peer_cnt < 3: 333 return peer_cnt 334 elif peer_cnt < 5: 335 return peer_cnt - 1 336 else: 337 return int(peer_cnt / 2) + 1

338

339 - def WaitForReply(self, salt, timeout=constants.CONFD_CLIENT_EXPIRE_TIMEOUT):

340 """Wait for replies to a given request. 341 342 This method will wait until either the timeout expires or a 343 minimum number (computed using L{_NeededReplies}) of replies are 344 received for the given salt. It is useful when doing synchronous 345 calls to this library. 346 347 @param salt: the salt of the request we want responses for 348 @param timeout: the maximum timeout (should be less or equal to 349 L{ganeti.constants.CONFD_CLIENT_EXPIRE_TIMEOUT} 350 @rtype: tuple 351 @return: a tuple of (timed_out, sent_cnt, recv_cnt); if the 352 request is unknown, timed_out will be true and the counters 353 will be zero 354 355 """ 356 def _CheckResponse(): 357 if salt not in self._requests: 358 # expired? 359 if self._logger: 360 self._logger.debug("Discarding unknown/expired request: %s" % salt) 361 return MISSING 362 rq = self._requests[salt] 363 if len(rq.rcvd) >= expected: 364 # already got all replies 365 return (False, len(rq.sent), len(rq.rcvd)) 366 # else wait, using default timeout 367 self.ReceiveReply() 368 raise utils.RetryAgain()

369 370 MISSING = (True, 0, 0) 371 372 if salt not in self._requests: 373 return MISSING 374 # extend the expire time with the current timeout, so that we 375 # don't get the request expired from under us 376 rq = self._requests[salt] 377 rq.expiry += timeout 378 sent = len(rq.sent) 379 expected = self._NeededReplies(sent) 380 381 try: 382 return utils.Retry(_CheckResponse, 0, timeout) 383 except utils.RetryTimeout: 384 if salt in self._requests: 385 rq = self._requests[salt] 386 return (True, len(rq.sent), len(rq.rcvd)) 387 else: 388 return MISSING

389

390 - def _SetPeersAddressFamily(self):

391 if not self._peers: 392 raise errors.ConfdClientError("Peer list empty") 393 try: 394 peer = self._peers[0] 395 self._family = netutils.IPAddress.GetAddressFamily(peer) 396 for peer in self._peers[1:]: 397 if netutils.IPAddress.GetAddressFamily(peer) != self._family: 398 raise errors.ConfdClientError("Peers must be of same address family") 399 except errors.IPAddressError: 400 raise errors.ConfdClientError("Peer address %s invalid" % peer)

401 402 403 # UPCALL_REPLY: server reply upcall 404 # has all ConfdUpcallPayload fields populated 405 UPCALL_REPLY = 1 406 # UPCALL_EXPIRE: internal library request expire 407 # has only salt, type, orig_request and extra_args 408 UPCALL_EXPIRE = 2 409 CONFD_UPCALL_TYPES = compat.UniqueFrozenset([ 410 UPCALL_REPLY, 411 UPCALL_EXPIRE, 412 ])

413 414 415 -class ConfdUpcallPayload(objects.ConfigObject):

416 """Callback argument for confd replies 417 418 @type salt: string 419 @ivar salt: salt associated with the query 420 @type type: one of confd.client.CONFD_UPCALL_TYPES 421 @ivar type: upcall type (server reply, expired request, ...) 422 @type orig_request: L{objects.ConfdRequest} 423 @ivar orig_request: original request 424 @type server_reply: L{objects.ConfdReply} 425 @ivar server_reply: server reply 426 @type server_ip: string 427 @ivar server_ip: answering server ip address 428 @type server_port: int 429 @ivar server_port: answering server port 430 @type extra_args: any 431 @ivar extra_args: 'args' argument of the SendRequest function 432 @type client: L{ConfdClient} 433 @ivar client: current confd client instance 434 435 """ 436 __slots__ = [ 437 "salt", 438 "type", 439 "orig_request", 440 "server_reply", 441 "server_ip", 442 "server_port", 443 "extra_args", 444 "client", 445 ]

446

447 448 -class ConfdClientRequest(objects.ConfdRequest):

449 """This is the client-side version of ConfdRequest. 450 451 This version of the class helps creating requests, on the client side, by 452 filling in some default values. 453 454 """

455 - def __init__(self, **kwargs):

456 objects.ConfdRequest.__init__(self, **kwargs) 457 if not self.rsalt: 458 self.rsalt = utils.NewUUID() 459 if not self.protocol: 460 self.protocol = constants.CONFD_PROTOCOL_VERSION 461 if self.type not in constants.CONFD_REQS: 462 raise errors.ConfdClientError("Invalid request type")

463

464 465 -class ConfdFilterCallback:

466 """Callback that calls another callback, but filters duplicate results. 467 468 @ivar consistent: a dictionary indexed by salt; for each salt, if 469 all responses ware identical, this will be True; this is the 470 expected state on a healthy cluster; on inconsistent or 471 partitioned clusters, this might be False, if we see answers 472 with the same serial but different contents 473 474 """

475 - def __init__(self, callback, logger=None):

476 """Constructor for ConfdFilterCallback 477 478 @type callback: f(L{ConfdUpcallPayload}) 479 @param callback: function to call when getting answers 480 @type logger: logging.Logger 481 @param logger: optional logger for internal conditions 482 483 """ 484 if not callable(callback): 485 raise errors.ProgrammerError("callback must be callable") 486 487 self._callback = callback 488 self._logger = logger 489 # answers contains a dict of salt -> answer 490 self._answers = {} 491 self.consistent = {}

492

493 - def _LogFilter(self, salt, new_reply, old_reply):

494 if not self._logger: 495 return 496 497 if new_reply.serial > old_reply.serial: 498 self._logger.debug("Filtering confirming answer, with newer" 499 " serial for query %s" % salt) 500 elif new_reply.serial == old_reply.serial: 501 if new_reply.answer != old_reply.answer: 502 self._logger.warning("Got incoherent answers for query %s" 503 " (serial: %s)" % (salt, new_reply.serial)) 504 else: 505 self._logger.debug("Filtering confirming answer, with same" 506 " serial for query %s" % salt) 507 else: 508 self._logger.debug("Filtering outdated answer for query %s" 509 " serial: (%d < %d)" % (salt, old_reply.serial, 510 new_reply.serial))

511

512 - def _HandleExpire(self, up):

513 # if we have no answer we have received none, before the expiration. 514 if up.salt in self._answers: 515 del self._answers[up.salt] 516 if up.salt in self.consistent: 517 del self.consistent[up.salt]

518

519 - def _HandleReply(self, up):

520 """Handle a single confd reply, and decide whether to filter it. 521 522 @rtype: boolean 523 @return: True if the reply should be filtered, False if it should be passed 524 on to the up-callback 525 526 """ 527 filter_upcall = False 528 salt = up.salt 529 if salt not in self.consistent: 530 self.consistent[salt] = True 531 if salt not in self._answers: 532 # first answer for a query (don't filter, and record) 533 self._answers[salt] = up.server_reply 534 elif up.server_reply.serial > self._answers[salt].serial: 535 # newer answer (record, and compare contents) 536 old_answer = self._answers[salt] 537 self._answers[salt] = up.server_reply 538 if up.server_reply.answer == old_answer.answer: 539 # same content (filter) (version upgrade was unrelated) 540 filter_upcall = True 541 self._LogFilter(salt, up.server_reply, old_answer) 542 # else: different content, pass up a second answer 543 else: 544 # older or same-version answer (duplicate or outdated, filter) 545 if (up.server_reply.serial == self._answers[salt].serial and 546 up.server_reply.answer != self._answers[salt].answer): 547 self.consistent[salt] = False 548 filter_upcall = True 549 self._LogFilter(salt, up.server_reply, self._answers[salt]) 550 551 return filter_upcall

552

553 - def __call__(self, up):

554 """Filtering callback 555 556 @type up: L{ConfdUpcallPayload} 557 @param up: upper callback 558 559 """ 560 filter_upcall = False 561 if up.type == UPCALL_REPLY: 562 filter_upcall = self._HandleReply(up) 563 elif up.type == UPCALL_EXPIRE: 564 self._HandleExpire(up) 565 566 if not filter_upcall: 567 self._callback(up)

568

569 570 -class ConfdCountingCallback:

571 """Callback that calls another callback, and counts the answers 572 573 """

574 - def __init__(self, callback, logger=None):

575 """Constructor for ConfdCountingCallback 576 577 @type callback: f(L{ConfdUpcallPayload}) 578 @param callback: function to call when getting answers 579 @type logger: logging.Logger 580 @param logger: optional logger for internal conditions 581 582 """ 583 if not callable(callback): 584 raise errors.ProgrammerError("callback must be callable") 585 586 self._callback = callback 587 self._logger = logger 588 # answers contains a dict of salt -> count 589 self._answers = {}

590

591 - def RegisterQuery(self, salt):

592 if salt in self._answers: 593 raise errors.ProgrammerError("query already registered") 594 self._answers[salt] = 0

595

596 - def AllAnswered(self):

597 """Have all the registered queries received at least an answer? 598 599 """ 600 return compat.all(self._answers.values())

601

602 - def _HandleExpire(self, up):

603 # if we have no answer we have received none, before the expiration. 604 if up.salt in self._answers: 605 del self._answers[up.salt]

606

607 - def _HandleReply(self, up):

608 """Handle a single confd reply, and decide whether to filter it. 609 610 @rtype: boolean 611 @return: True if the reply should be filtered, False if it should be passed 612 on to the up-callback 613 614 """ 615 if up.salt in self._answers: 616 self._answers[up.salt] += 1

617

618 - def __call__(self, up):

619 """Filtering callback 620 621 @type up: L{ConfdUpcallPayload} 622 @param up: upper callback 623 624 """ 625 if up.type == UPCALL_REPLY: 626 self._HandleReply(up) 627 elif up.type == UPCALL_EXPIRE: 628 self._HandleExpire(up) 629 self._callback(up)

630

631 632 -class StoreResultCallback:

633 """Callback that simply stores the most recent answer. 634 635 @ivar _answers: dict of salt to (have_answer, reply) 636 637 """ 638 _NO_KEY = (False, None) 639

640 - def __init__(self):

641 """Constructor for StoreResultCallback 642 643 """ 644 # answers contains a dict of salt -> best result 645 self._answers = {}

646

647 - def GetResponse(self, salt):

648 """Return the best match for a salt 649 650 """ 651 return self._answers.get(salt, self._NO_KEY)

652

653 - def _HandleExpire(self, up):

654 """Expiration handler. 655 656 """ 657 if up.salt in self._answers and self._answers[up.salt] == self._NO_KEY: 658 del self._answers[up.salt]

659

660 - def _HandleReply(self, up):

661 """Handle a single confd reply, and decide whether to filter it. 662 663 """ 664 self._answers[up.salt] = (True, up)

665

666 - def __call__(self, up):

667 """Filtering callback 668 669 @type up: L{ConfdUpcallPayload} 670 @param up: upper callback 671 672 """ 673 if up.type == UPCALL_REPLY: 674 self._HandleReply(up) 675 elif up.type == UPCALL_EXPIRE: 676 self._HandleExpire(up)

677

678 679 -def GetConfdClient(callback):

680 """Return a client configured using the given callback. 681 682 This is handy to abstract the MC list and HMAC key reading. 683 684 @attention: This should only be called on nodes which are part of a 685 cluster, since it depends on a valid (ganeti) data directory; 686 for code running outside of a cluster, you need to create the 687 client manually 688 689 """ 690 ss = ssconf.SimpleStore() 691 mc_file = ss.KeyToFilename(constants.SS_MASTER_CANDIDATES_IPS) 692 mc_list = utils.ReadFile(mc_file).splitlines() 693 hmac_key = utils.ReadFile(pathutils.CONFD_HMAC_KEY) 694 return ConfdClient(hmac_key, mc_list, callback)

695

Source Code for Module ganeti.confd.client