ganeti.confd.client

1 # 2 # 3 4 # Copyright (C) 2009 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-msg=E0203 49 50 # E0203: Access to member %r before its definition, since we use 51 # objects.py which doesn't explicitely 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 66 67 -class ConfdAsyncUDPClient(daemon.AsyncUDPSocket):

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

74 - def __init__(self, client):

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

83 84 # this method is overriding a daemon.AsyncUDPSocket method

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

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

87

88 89 -class _Request(object):

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

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

100 self.request = request 101 self.args = args 102 self.expiry = expiry 103 self.sent = frozenset(sent) 104 self.rcvd = set()

105

106 107 -class ConfdClient:

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

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

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

148

149 - def UpdatePeerList(self, peers):

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

162

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

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

175

176 - def _UnpackReply(self, payload):

177 in_payload = confd.UnpackMagic(payload) 178 (dict_answer, salt) = serializer.LoadSignedJson(in_payload, self._hmac_key) 179 answer = objects.ConfdReply.FromDict(dict_answer) 180 return answer, salt

181

182 - def ExpireRequests(self):

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

197

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

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

252

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

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

289

290 - def FlushSendQueue(self):

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

298

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

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

309 310 @staticmethod

311 - def _NeededReplies(peer_cnt):

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

335

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

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

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

386 387 388 # UPCALL_REPLY: server reply upcall 389 # has all ConfdUpcallPayload fields populated 390 UPCALL_REPLY = 1 391 # UPCALL_EXPIRE: internal library request expire 392 # has only salt, type, orig_request and extra_args 393 UPCALL_EXPIRE = 2 394 CONFD_UPCALL_TYPES = frozenset([ 395 UPCALL_REPLY, 396 UPCALL_EXPIRE, 397 ])

398 399 400 -class ConfdUpcallPayload(objects.ConfigObject):

401 """Callback argument for confd replies 402 403 @type salt: string 404 @ivar salt: salt associated with the query 405 @type type: one of confd.client.CONFD_UPCALL_TYPES 406 @ivar type: upcall type (server reply, expired request, ...) 407 @type orig_request: L{objects.ConfdRequest} 408 @ivar orig_request: original request 409 @type server_reply: L{objects.ConfdReply} 410 @ivar server_reply: server reply 411 @type server_ip: string 412 @ivar server_ip: answering server ip address 413 @type server_port: int 414 @ivar server_port: answering server port 415 @type extra_args: any 416 @ivar extra_args: 'args' argument of the SendRequest function 417 @type client: L{ConfdClient} 418 @ivar client: current confd client instance 419 420 """ 421 __slots__ = [ 422 "salt", 423 "type", 424 "orig_request", 425 "server_reply", 426 "server_ip", 427 "server_port", 428 "extra_args", 429 "client", 430 ]

431

432 433 -class ConfdClientRequest(objects.ConfdRequest):

434 """This is the client-side version of ConfdRequest. 435 436 This version of the class helps creating requests, on the client side, by 437 filling in some default values. 438 439 """

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

441 objects.ConfdRequest.__init__(self, **kwargs) 442 if not self.rsalt: 443 self.rsalt = utils.NewUUID() 444 if not self.protocol: 445 self.protocol = constants.CONFD_PROTOCOL_VERSION 446 if self.type not in constants.CONFD_REQS: 447 raise errors.ConfdClientError("Invalid request type")

448

449 450 -class ConfdFilterCallback:

451 """Callback that calls another callback, but filters duplicate results. 452 453 @ivar consistent: a dictionary indexed by salt; for each salt, if 454 all responses ware identical, this will be True; this is the 455 expected state on a healthy cluster; on inconsistent or 456 partitioned clusters, this might be False, if we see answers 457 with the same serial but different contents 458 459 """

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

461 """Constructor for ConfdFilterCallback 462 463 @type callback: f(L{ConfdUpcallPayload}) 464 @param callback: function to call when getting answers 465 @type logger: logging.Logger 466 @param logger: optional logger for internal conditions 467 468 """ 469 if not callable(callback): 470 raise errors.ProgrammerError("callback must be callable") 471 472 self._callback = callback 473 self._logger = logger 474 # answers contains a dict of salt -> answer 475 self._answers = {} 476 self.consistent = {}

477

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

479 if not self._logger: 480 return 481 482 if new_reply.serial > old_reply.serial: 483 self._logger.debug("Filtering confirming answer, with newer" 484 " serial for query %s" % salt) 485 elif new_reply.serial == old_reply.serial: 486 if new_reply.answer != old_reply.answer: 487 self._logger.warning("Got incoherent answers for query %s" 488 " (serial: %s)" % (salt, new_reply.serial)) 489 else: 490 self._logger.debug("Filtering confirming answer, with same" 491 " serial for query %s" % salt) 492 else: 493 self._logger.debug("Filtering outdated answer for query %s" 494 " serial: (%d < %d)" % (salt, old_reply.serial, 495 new_reply.serial))

496

497 - def _HandleExpire(self, up):

498 # if we have no answer we have received none, before the expiration. 499 if up.salt in self._answers: 500 del self._answers[up.salt] 501 if up.salt in self.consistent: 502 del self.consistent[up.salt]

503

504 - def _HandleReply(self, up):

505 """Handle a single confd reply, and decide whether to filter it. 506 507 @rtype: boolean 508 @return: True if the reply should be filtered, False if it should be passed 509 on to the up-callback 510 511 """ 512 filter_upcall = False 513 salt = up.salt 514 if salt not in self.consistent: 515 self.consistent[salt] = True 516 if salt not in self._answers: 517 # first answer for a query (don't filter, and record) 518 self._answers[salt] = up.server_reply 519 elif up.server_reply.serial > self._answers[salt].serial: 520 # newer answer (record, and compare contents) 521 old_answer = self._answers[salt] 522 self._answers[salt] = up.server_reply 523 if up.server_reply.answer == old_answer.answer: 524 # same content (filter) (version upgrade was unrelated) 525 filter_upcall = True 526 self._LogFilter(salt, up.server_reply, old_answer) 527 # else: different content, pass up a second answer 528 else: 529 # older or same-version answer (duplicate or outdated, filter) 530 if (up.server_reply.serial == self._answers[salt].serial and 531 up.server_reply.answer != self._answers[salt].answer): 532 self.consistent[salt] = False 533 filter_upcall = True 534 self._LogFilter(salt, up.server_reply, self._answers[salt]) 535 536 return filter_upcall

537

538 - def __call__(self, up):

539 """Filtering callback 540 541 @type up: L{ConfdUpcallPayload} 542 @param up: upper callback 543 544 """ 545 filter_upcall = False 546 if up.type == UPCALL_REPLY: 547 filter_upcall = self._HandleReply(up) 548 elif up.type == UPCALL_EXPIRE: 549 self._HandleExpire(up) 550 551 if not filter_upcall: 552 self._callback(up)

553

554 555 -class ConfdCountingCallback:

556 """Callback that calls another callback, and counts the answers 557 558 """

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

560 """Constructor for ConfdCountingCallback 561 562 @type callback: f(L{ConfdUpcallPayload}) 563 @param callback: function to call when getting answers 564 @type logger: logging.Logger 565 @param logger: optional logger for internal conditions 566 567 """ 568 if not callable(callback): 569 raise errors.ProgrammerError("callback must be callable") 570 571 self._callback = callback 572 self._logger = logger 573 # answers contains a dict of salt -> count 574 self._answers = {}

575

576 - def RegisterQuery(self, salt):

577 if salt in self._answers: 578 raise errors.ProgrammerError("query already registered") 579 self._answers[salt] = 0

580

581 - def AllAnswered(self):

582 """Have all the registered queries received at least an answer? 583 584 """ 585 return compat.all(self._answers.values())

586

587 - def _HandleExpire(self, up):

588 # if we have no answer we have received none, before the expiration. 589 if up.salt in self._answers: 590 del self._answers[up.salt]

591

592 - def _HandleReply(self, up):

593 """Handle a single confd reply, and decide whether to filter it. 594 595 @rtype: boolean 596 @return: True if the reply should be filtered, False if it should be passed 597 on to the up-callback 598 599 """ 600 if up.salt in self._answers: 601 self._answers[up.salt] += 1

602

603 - def __call__(self, up):

604 """Filtering callback 605 606 @type up: L{ConfdUpcallPayload} 607 @param up: upper callback 608 609 """ 610 if up.type == UPCALL_REPLY: 611 self._HandleReply(up) 612 elif up.type == UPCALL_EXPIRE: 613 self._HandleExpire(up) 614 self._callback(up)

615

616 617 -class StoreResultCallback:

618 """Callback that simply stores the most recent answer. 619 620 @ivar _answers: dict of salt to (have_answer, reply) 621 622 """ 623 _NO_KEY = (False, None) 624

625 - def __init__(self):

626 """Constructor for StoreResultCallback 627 628 """ 629 # answers contains a dict of salt -> best result 630 self._answers = {}

631

632 - def GetResponse(self, salt):

633 """Return the best match for a salt 634 635 """ 636 return self._answers.get(salt, self._NO_KEY)

637

638 - def _HandleExpire(self, up):

639 """Expiration handler. 640 641 """ 642 if up.salt in self._answers and self._answers[up.salt] == self._NO_KEY: 643 del self._answers[up.salt]

644

645 - def _HandleReply(self, up):

646 """Handle a single confd reply, and decide whether to filter it. 647 648 """ 649 self._answers[up.salt] = (True, up)

650

651 - def __call__(self, up):

652 """Filtering callback 653 654 @type up: L{ConfdUpcallPayload} 655 @param up: upper callback 656 657 """ 658 if up.type == UPCALL_REPLY: 659 self._HandleReply(up) 660 elif up.type == UPCALL_EXPIRE: 661 self._HandleExpire(up)

662

663 664 -def GetConfdClient(callback):

665 """Return a client configured using the given callback. 666 667 This is handy to abstract the MC list and HMAC key reading. 668 669 @attention: This should only be called on nodes which are part of a 670 cluster, since it depends on a valid (ganeti) data directory; 671 for code running outside of a cluster, you need to create the 672 client manually 673 674 """ 675 ss = ssconf.SimpleStore() 676 mc_file = ss.KeyToFilename(constants.SS_MASTER_CANDIDATES_IPS) 677 mc_list = utils.ReadFile(mc_file).splitlines() 678 hmac_key = utils.ReadFile(constants.CONFD_HMAC_KEY) 679 return ConfdClient(hmac_key, mc_list, callback)

680

Source Code for Module ganeti.confd.client