backend.server.rhnVirtualization

153 154 ## 155 # This map defines how to route each event to the appropriate handler. 156 # 157 HANDLERS = { 158 (EventType.EXISTS, TargetType.SYSTEM): '_handle_system_exists', 159 (EventType.EXISTS, TargetType.DOMAIN): '_handle_domain_exists', 160 (EventType.REMOVED, TargetType.DOMAIN): '_handle_domain_removed', 161 (EventType.CRAWL_BEGAN, TargetType.SYSTEM): '_handle_system_crawl_began', 162 (EventType.CRAWL_ENDED, TargetType.SYSTEM): '_handle_system_crawl_ended', 163 (EventType.EXISTS, TargetType.LOG_MSG): '_handle_log_msg_exists' 164 } 165 166 ## 167 # This map defines the absolute required properties for each event type. 168 # 169 REQUIRED_PROPERTIES = { 170 (EventType.EXISTS, TargetType.SYSTEM): (PropertyType.IDENTITY, 171 PropertyType.UUID, ), 172 (EventType.EXISTS, TargetType.DOMAIN): (PropertyType.UUID, ), 173 (EventType.EXISTS, TargetType.LOG_MSG): (PropertyType.MESSAGE, 174 PropertyType.ID, ) 175 } 176 177 ########################################################################### 178 # Public Methods 179 ########################################################################### 180

181 - def __init__(self):

182 pass

183

184 - def handle(self, system_id, notification):

185 186 log_debug(5, "Handling notification:", system_id, notification) 187 188 # First, validate that the notification is in the correct format. If it 189 # is not, we'll bail out. 190 if len(notification) != 4: 191 raise VirtualizationEventError( 192 "Received invalid notification length:", notification, 193 "; len=", len(notification)) 194 195 # Now we are ready to field the notification. Begin by parsing it. 196 (timestamp, action, target, properties) = notification 197 198 event = (action, target) 199 200 # Fetch the appropriate handler. 201 handler = None 202 try: 203 handler = getattr(self, self.HANDLERS[event]) 204 except KeyError: 205 ke = sys.exc_info()[1] 206 raise_with_tb(VirtualizationEventError( 207 "Don't know how to handle virt event:", event), sys.exc_info()[2]) 208 209 # Ensure that the event has any required properties before calling the 210 # handler. 211 if event in self.REQUIRED_PROPERTIES: 212 required_properties = self.REQUIRED_PROPERTIES[event] 213 for required_property in required_properties: 214 if required_property not in properties: 215 raise VirtualizationEventError( 216 "Event does not have required property:", 217 required_property, 218 event) 219 220 # Some properties need to be preprocessed before we can actually 221 # handle the notification. 222 self.__convert_properties(properties) 223 224 # Call the handler. 225 handler(system_id, timestamp, properties)

226 227 ########################################################################### 228 # Protected Methods 229 ########################################################################### 230

231 - def _handle_system_exists(self, system_id, timestamp, properties):

232 uuid = properties[PropertyType.UUID] 233 identity = properties[PropertyType.IDENTITY] 234 virt_type = None 235 236 if PropertyType.TYPE in properties: 237 virt_type = properties[PropertyType.TYPE] 238 else: 239 # presume paravirt if not specified, probably a host 240 virt_type = VirtualizationType.PARA 241 242 row = self.__db_get_system(identity, system_id, uuid) 243 if not row: 244 self.__db_insert_system(identity, system_id, uuid, virt_type) 245 else: 246 self.__db_update_system(identity, system_id, uuid, row) 247 248 self.__notify_listeners(ListenerEvent.GUEST_REGISTERED, 249 row['host_system_id'], 250 system_id)

251

252 - def _handle_domain_exists(self, system_id, timestamp, properties):

253 uuid = properties[PropertyType.UUID] 254 255 row = self.__db_get_domain(system_id, uuid) 256 if not row: 257 self.__db_insert_domain(system_id, uuid, properties) 258 259 # We've noticed a new guest; send a notification down the pipeline. 260 self.__notify_listeners(ListenerEvent.GUEST_DISCOVERED, 261 system_id, 262 uuid) 263 else: 264 self.__db_update_domain(system_id, uuid, properties, row) 265 266 # We'll attempt to detect migration by checking if the host system 267 # ID has changed. 268 if 'host_system_id' in row and \ 269 row['host_system_id'] != system_id: 270 271 self.__notify_listeners(ListenerEvent.GUEST_MIGRATED, 272 row['host_system_id'], 273 system_id, 274 row['virtual_system_id'], 275 uuid)

276

277 - def _handle_domain_removed(self, system_id, timestamp, properties):

278 """ Handle a domain removal. Since we are dealing with virtual domains, we 279 can't really tell whether physical removal took place, so we'll just mark 280 the domain as 'stopped'. 281 """ 282 uuid = properties[PropertyType.UUID] 283 284 row = self.__db_get_domain(system_id, uuid) 285 if len(list(row.keys())) == 0: 286 log_debug(1, "Guest already deleted in satellite: ", properties) 287 return 288 new_properties = {PropertyType.STATE: ServerStateType.STOPPED} 289 self.__db_update_domain(system_id, uuid, new_properties, row)

290

291 - def _handle_system_crawl_began(self, system_id, timestamp, properties):

292 self.__unconfirm_domains(system_id)

293

294 - def _handle_system_crawl_ended(self, system_id, timestamp, properties):

295 self.__remove_unconfirmed_domains(system_id) 296 self.__confirm_domains(system_id)

297

298 - def _handle_log_msg_exists(self, system_id, timestamp, properties):

299 kickstart_session_id = properties[PropertyType.ID] 300 log_message = properties[PropertyType.MESSAGE] 301 302 self.__db_insert_log_message(kickstart_session_id, log_message)

303 304 ########################################################################### 305 # Helper Methods 306 ########################################################################### 307

308 - def __db_get_system(self, identity, system_id, uuid):

309 """ This returns a row from the database that represents a virtual system. 310 If no system could be found, None is returned. 311 """ 312 313 condition = None 314 315 # The SELECT condition is different, depending on whether this system 316 # is a host or a guest. A guest will always have a UUID, while a host 317 # will never have one. Instead, a host should be identified by its 318 # sysid only. 319 # 320 # When IdentityType.GUEST, need to worry about cross-org issues... 321 # 2 states to worry about: 322 # - no prior entry in the VI table; we return nothing, insert happens 323 # - prior entry, we return that one, update happens 324 if identity == IdentityType.HOST: 325 condition = """ 326 vi.uuid is null 327 and vi.host_system_id=:system_id 328 """ 329 elif identity == IdentityType.GUEST: 330 condition = """ 331 ( 332 vi.uuid=:uuid 333 AND (vi.virtual_system_id is null or 334 vi.virtual_system_id = :system_id) 335 ) 336 OR 337 ( 338 vi.uuid is not null and 339 vi.virtual_system_id = :system_id 340 ) 341 """ 342 else: 343 raise VirtualizationEventError( 344 "Unknown identity:", identity) 345 346 select_sql = """ 347 SELECT 348 vi.id as id, 349 vi.host_system_id as host_system_id, 350 vi.virtual_system_id as virtual_system_id, 351 vi.uuid as uuid, 352 vi.confirmed as confirmed 353 FROM 354 rhnVirtualInstance vi 355 WHERE 356 %s 357 """ % (condition) 358 query = rhnSQL.prepare(select_sql) 359 query.execute(system_id=system_id, uuid=uuid) 360 361 row = query.fetchone_dict() or {} 362 363 return row

364

365 - def __db_insert_system(self, identity, system_id, uuid, virt_type):

366 """ Inserts a new system into the database. """ 367 368 # If this system is a host, it's sysid goes into the host_system_id 369 # column. Otherwise, it's sysid goes into the virtual_system_id 370 # column. 371 host_id = None 372 guest_id = None 373 if identity == IdentityType.HOST: 374 host_id = system_id 375 elif identity == IdentityType.GUEST: 376 guest_id = system_id 377 378 # Check to see if this uuid has already been registered to a 379 # host and is confirmed. 380 check_sql = """ 381 select 382 vi.id, 383 vi.host_system_id, 384 vi.confirmed 385 from 386 rhnVirtualInstance vi 387 where 388 vi.uuid = :uuid 389 and confirmed = 1 390 """ 391 392 query = rhnSQL.prepare(check_sql) 393 query.execute(uuid=uuid, system_id=system_id) 394 395 row = query.fetchone_dict() 396 397 if row: 398 # We found a host for this guest, we'll save the value 399 # to use when we create the row in rhnVirtualInstance. 400 host_id = row['host_system_id'] 401 else: 402 # We didn't find a host, this guest will just end up with 403 # no host, and consuming physical entitlements. 404 pass 405 406 else: 407 raise VirtualizationEventError( 408 "Unknown identity:", identity) 409 410 get_id_sql = "SELECT sequence_nextval('rhn_vi_id_seq') as id FROM dual" 411 query = rhnSQL.prepare(get_id_sql) 412 query.execute() 413 row = query.fetchone_dict() or {} 414 415 if not row or 'id' not in row: 416 raise VirtualizationEventError('unable to get virt instance id') 417 418 insert_sql = """ 419 INSERT INTO rhnVirtualInstance 420 (id, host_system_id, virtual_system_id, uuid, confirmed) 421 VALUES 422 (:id, :host_id, :guest_id, :uuid, 1) 423 """ 424 query = rhnSQL.prepare(insert_sql) 425 query.execute(id=row['id'], 426 host_id=host_id, 427 guest_id=guest_id, 428 uuid=uuid) 429 430 # Initialize a dummy info record for this system. 431 insert_sql = """ 432 INSERT INTO rhnVirtualInstanceInfo 433 (instance_id, state, instance_type) 434 VALUES 435 (:id, 436 ( 437 SELECT rvis.id 438 FROM rhnVirtualInstanceState rvis 439 WHERE rvis.label = :state 440 ), 441 ( 442 SELECT rvit.id 443 FROM rhnVirtualInstanceType rvit 444 WHERE rvit.label = :virt_type 445 )) 446 """ 447 query = rhnSQL.prepare(insert_sql) 448 query.execute(id=row['id'], 449 state=ServerStateType.UNKNOWN, 450 virt_type=virt_type)

451

452 - def __db_update_system(self, identity, system_id, uuid, existing_row):

453 """ Updates a system in the database. """ 454 455 # since __db_get_system protects us against crossing the org 456 # boundary, we really don't need to worry much about existing_row's 457 # values... 458 459 new_values_array = [] 460 bindings = {} 461 if not existing_row.get('confirmed'): 462 new_values_array.append("confirmed=1") 463 464 # Some guests may have been unregistered before, and therefore did not 465 # have sysid's. If we got an EXISTS for a guest system, then a guest 466 # must have been registered. Make sure that we update the 467 # virtual_system_id column in the DB to reflect that this guest is now 468 # registered. 469 if identity == IdentityType.GUEST: 470 if existing_row['virtual_system_id'] != system_id: 471 new_values_array.append("virtual_system_id=:sysid") 472 bindings['sysid'] = system_id 473 # note, at this point, it's still possible to have 474 # an entry in rhnVirtualInstance for this uuid w/out 475 # a virtual_system_id; it'd be for a different org 476 if existing_row['uuid'] != uuid: 477 new_values_array.append("uuid=:uuid") 478 bindings['uuid'] = uuid 479 480 # Only touch the database if something changed. 481 if new_values_array: 482 new_values = string.join(new_values_array, ', ') 483 484 bindings['row_id'] = existing_row['id'] 485 486 update_sql = """ 487 UPDATE rhnVirtualInstance SET %s WHERE id=:row_id 488 """ % (new_values) 489 query = rhnSQL.prepare(update_sql) 490 query.execute(**bindings)

491

492 - def __db_get_domain(self, host_id, uuid):

493 select_sql = """ 494 SELECT 495 rvi.id as rvi_id, 496 rvi.host_system_id as host_system_id, 497 rvi.virtual_system_id as virtual_system_id, 498 rvi.confirmed as confirmed, 499 rvii.name as name, 500 rvit.label as instance_type, 501 rvii.memory_size_k as memory_size_k, 502 rvii.instance_id as instance_id, 503 rvii.vcpus as vcpus, 504 rvis.label as state 505 FROM 506 rhnVirtualInstanceInfo rvii, 507 rhnVirtualInstanceType rvit, 508 rhnVirtualInstanceState rvis, 509 rhnVirtualInstance rvi 510 WHERE 511 (rvi.uuid = :uuid or 512 (:uuid is null and 513 rvi.uuid is null)) and 514 rvi.host_system_id = :host_id and 515 rvi.id = rvii.instance_id and 516 rvit.id = rvii.instance_type and 517 rvis.id = rvii.state 518 """ 519 query = rhnSQL.prepare(select_sql) 520 query.execute(host_id=host_id, uuid=uuid) 521 522 row = query.fetchone_dict() or {} 523 524 return row

525

526 - def __db_insert_domain(self, host_id, uuid, properties):

527 """ To create a new domain, we must modify both the rhnVirtualInstance 528 and the rhnVirtualInstanceInfo tables. 529 """ 530 # We'll do rhnVirtualInstance first. 531 get_id_sql = "SELECT sequence_nextval('rhn_vi_id_seq') as id FROM dual" 532 query = rhnSQL.prepare(get_id_sql) 533 query.execute() 534 row = query.fetchone_dict() or {} 535 536 if not row or 'id' not in row: 537 raise VirtualizationEventError('unable to get virt instance id') 538 id = row['id'] 539 540 insert_sql = """ 541 INSERT INTO rhnVirtualInstance 542 (id, host_system_id, virtual_system_id, uuid, confirmed) 543 VALUES 544 (:id, :host_id, null, :uuid, 1) 545 """ 546 query = rhnSQL.prepare(insert_sql) 547 query.execute(id=id, host_id=host_id, uuid=uuid) 548 549 # Now we'll insert into the rhnVirtualInstanceInfo table. 550 551 insert_sql = """ 552 INSERT INTO rhnVirtualInstanceInfo 553 (instance_id, 554 name, 555 vcpus, 556 memory_size_k, 557 instance_type, 558 state) 559 SELECT 560 :id, 561 :name, 562 :vcpus, 563 :memory, 564 rvit.id, 565 rvis.id 566 FROM 567 rhnVirtualInstanceType rvit, 568 rhnVirtualInstanceState rvis 569 WHERE 570 rvit.label=:virt_type and 571 rvis.label=:state 572 """ 573 name = properties[PropertyType.NAME] 574 vcpus = properties[PropertyType.VCPUS] 575 memory = properties[PropertyType.MEMORY] 576 virt_type = properties[PropertyType.TYPE] 577 state = properties[PropertyType.STATE] 578 579 query = rhnSQL.prepare(insert_sql) 580 query.execute(id=id, 581 name=name, 582 vcpus=vcpus, 583 memory=memory, 584 virt_type=virt_type, 585 state=state)

586

587 - def __db_update_domain(self, host_id, uuid, properties, existing_row):

588 589 # First, update the rhnVirtualInstance table. If a guest domain was 590 # registered but its host was not, it is possible that the 591 # rhnVirtualInstance table's host_system_id column is null. We'll 592 # update that now, if need be. 593 594 # __db_get_domain is responsible for ensuring that the org for any 595 # existing_row matches the org for host_id 596 597 new_values_array = [] 598 bindings = {} 599 600 if not existing_row.get('confirmed'): 601 new_values_array.append('confirmed=1') 602 603 if existing_row['host_system_id'] != host_id: 604 new_values_array.append('host_system_id=:host_id') 605 bindings['host_id'] = host_id 606 607 # Only touch the database if something changed. 608 if new_values_array: 609 new_values = string.join(new_values_array, ', ') 610 611 bindings['row_id'] = existing_row['rvi_id'] 612 613 update_sql = """ 614 UPDATE rhnVirtualInstance SET %s WHERE id=:row_id 615 """ % (new_values) 616 query = rhnSQL.prepare(update_sql) 617 618 try: 619 query.execute(**bindings) 620 except rhnSQL.SQLError: 621 e = sys.exc_info()[1] 622 log_error(str(e)) 623 raise_with_tb(VirtualizationEventError(str(e)), sys.exc_info()[2]) 624 625 # Now update the rhnVirtualInstanceInfo table. 626 627 new_values_array = [] 628 bindings = {} 629 630 if PropertyType.NAME in properties and \ 631 existing_row['name'] != properties[PropertyType.NAME]: 632 new_values_array.append('name=:name') 633 bindings['name'] = properties[PropertyType.NAME] 634 635 if PropertyType.VCPUS in properties and \ 636 existing_row['vcpus'] != properties[PropertyType.VCPUS]: 637 new_values_array.append('vcpus=:vcpus') 638 bindings['vcpus'] = properties[PropertyType.VCPUS] 639 640 if PropertyType.MEMORY in properties and \ 641 existing_row['memory_size_k'] != properties[PropertyType.MEMORY]: 642 new_values_array.append('memory_size_k=:memory') 643 bindings['memory'] = properties[PropertyType.MEMORY] 644 645 if PropertyType.TYPE in properties and \ 646 existing_row['instance_type'] != properties[PropertyType.TYPE]: 647 new_values_array.append(""" 648 instance_type = ( 649 select rvit.id 650 from rhnVirtualInstanceType rvit 651 where rvit.label = :virt_type) 652 """) 653 bindings['virt_type'] = properties[PropertyType.TYPE] 654 655 if PropertyType.STATE in properties and \ 656 existing_row['state'] != properties[PropertyType.STATE]: 657 new_values_array.append(""" 658 state = ( 659 SELECT rvis.id 660 FROM rhnVirtualInstanceState rvis 661 WHERE rvis.label = :state) 662 """) 663 bindings['state'] = properties[PropertyType.STATE] 664 665 # Only touch the database if something changed. 666 if new_values_array: 667 new_values = string.join(new_values_array, ', ') 668 669 bindings['row_id'] = existing_row['instance_id'] 670 671 update_sql = """ 672 UPDATE rhnVirtualInstanceInfo SET %s WHERE instance_id=:row_id 673 """ % (new_values) 674 query = rhnSQL.prepare(update_sql) 675 query.execute(**bindings)

676

677 - def __unconfirm_domains(self, system_id):

678 update_sql = """ 679 UPDATE rhnVirtualInstance 680 SET confirmed=0 681 WHERE host_system_id=:sysid 682 """ 683 query = rhnSQL.prepare(update_sql) 684 query.execute(sysid=system_id)

685

686 - def __confirm_domains(self, system_id):

687 update_sql = """ 688 UPDATE rhnVirtualInstance 689 SET confirmed=1 690 WHERE host_system_id=:sysid 691 """ 692 query = rhnSQL.prepare(update_sql) 693 query.execute(sysid=system_id)

694

695 - def __remove_unconfirmed_domains(self, system_id):

696 """ Mark the unconfirmed entries in the RVII table as stopped, since it 697 appears they are no longer running. 698 """ 699 700 update_sql = """ 701 UPDATE rhnVirtualInstanceInfo rvii 702 SET state=( 703 SELECT rvis.id 704 FROM rhnVirtualInstanceState rvis 705 WHERE rvis.label=:state 706 ) 707 WHERE 708 rvii.instance_id IN ( 709 SELECT rvi.id 710 FROM rhnVirtualInstance rvi 711 WHERE rvi.confirmed=0) 712 """ 713 query = rhnSQL.prepare(update_sql) 714 query.execute(state=ServerStateType.STOPPED)

715

716 - def __db_insert_log_message(self, kickstart_session_id, log_message):

717 """ 718 Insert a new installation log message into the database. 719 """ 720 721 # log_message must be 4000 chars or shorter, db constraint 722 log_message = log_message[:4000] 723 724 insert_sql = """ 725 INSERT INTO rhnVirtualInstanceInstallLog 726 (id, log_message, ks_session_id) 727 VALUES 728 (sequence_nextval('rhn_viil_id_seq'), :log_message, :kickstart_session_id) 729 """ 730 query = rhnSQL.prepare(insert_sql) 731 query.execute(log_message=log_message, 732 kickstart_session_id=kickstart_session_id)

733

734 - def __convert_properties(self, properties):

735 """ This function normalizes and converts the values of some properties to 736 format consumable by the server. 737 """ 738 # Attempt to normalize the UUID. 739 if PropertyType.UUID in properties: 740 uuid = properties[PropertyType.UUID] 741 if uuid: 742 uuid_as_number = string.atol(uuid, 16) 743 744 if uuid_as_number == 0: 745 # If the UUID is a bunch of null bytes, we will convert it 746 # to None. This will allow us to interact with the 747 # database properly, since the database assumes a null UUID 748 # when the system is a host. 749 properties[PropertyType.UUID] = None 750 else: 751 # Normalize the UUID. We don't know how it will appear 752 # when it comes from the client, so we'll convert it to a 753 # normal form. 754 # if UUID had leading 0, we must pad 0 again #429192 755 properties[PropertyType.UUID] = "%032x" % uuid_as_number 756 else: 757 properties[PropertyType.UUID] = None 758 759 # The server only cares about certain types of states. 760 if PropertyType.STATE in properties: 761 state = properties[PropertyType.STATE] 762 properties[PropertyType.STATE] = CLIENT_SERVER_STATE_MAP[state] 763 764 # We must send the memory across as a string because XMLRPC can only 765 # handle up to 32 bit numbers. RAM can easily exceed that limit these 766 # days. 767 if PropertyType.MEMORY in properties: 768 memory = properties[PropertyType.MEMORY] 769 properties[PropertyType.MEMORY] = LongType(memory)

770

771 - def __notify_listeners(self, *args):

772 for listener in Listeners.listeners: 773 listener._notify(*args)

Source Code for Module backend.server.rhnVirtualization