PLC/Methods/RefreshPeer.py

   1 #
   2 # Thierry Parmentelat - INRIA
   3 #
   4 import os
   5 import sys
   6 import fcntl
   7 import time
   8
   9 from PLC.Logger import logger
  10 from PLC.Faults import *
  11 from PLC.Method import Method
  12 from PLC.Parameter import Parameter, Mixed
  13 from PLC.Auth import Auth
  14
  15 from PLC.Peers import Peer, Peers
  16 from PLC.Sites import Site, Sites
  17 from PLC.Persons import Person, Persons
  18 from PLC.KeyTypes import KeyType, KeyTypes
  19 from PLC.Keys import Key, Keys
  20 from PLC.BootStates import BootState, BootStates
  21 from PLC.Nodes import Node, Nodes
  22 from PLC.SliceInstantiations import SliceInstantiations
  23 from PLC.Slices import Slice, Slices
  24 from PLC.Roles import Role, Roles
  25
  26 # settings
  27 # initial version was doing only one final commit
  28 # * set commit_mode to False to get that behaviour
  29 # * set comit_mode to True to get everything synced at once
  30 # the issue with the 'one-commit-at-the-end' approach is
  31 # that the db gets basically totally locked during too long
  32 # causing various issues/crashes in the rest of the system
  33 commit_mode = True
  34
  35 # turn this to False only if both ends have the same db schema
  36 # compatibility mode is a bit slower but probably safer on the long run
  37 compatibility = True
  38
  39 # debugging
  40 # for verbose output
  41 verbose = False
  42 # set to a filename for using cached data when debugging
  43 # WARNING: does not actually connect to the peer in this case
  44 use_cache = None
  45 # for debugging specific entries - display detailed info on selected objs
  46 focus_type = None  # set to e.g. 'Person'
  47 # set to a list of ids (e.g. person_ids) - remote or local ids should work
  48 focus_ids = []
  49 # example
  50 # use_cache="/var/log/peers/getpeerdata.pickle"
  51 # verbose=True
  52 # focus_type='Person'
  53 # focus_ids=[621,1088]
  54
  55 # helpers
  56
  57
  58 def message(to_print=None, verbose_only=False):
  59     if verbose_only and not verbose:
  60         return
  61     logger.info(to_print)
  62
  63
  64 def message_verbose(to_print=None, header='VERBOSE'):
  65     message("%s> %r" % (header, to_print), verbose_only=True)
  66
  67
  68 # to avoid several instances running at the same time
  69 class FileLock:
  70     """
  71     Lock/Unlock file
  72     """
  73
  74     def __init__(self, file_path, expire=60 * 60 * 2):
  75         self.expire = expire
  76         self.fpath = file_path
  77         self.fd = None
  78
  79     def lock(self):
  80         if os.path.exists(self.fpath):
  81             if (time.time() - os.stat(self.fpath).st_ctime) > self.expire:
  82                 try:
  83                     os.unlink(self.fpath)
  84                 except Exception, e:
  85                     message('FileLock.lock(%s) : %s' % (self.fpath, e))
  86                     return False
  87         try:
  88             self.fd = open(self.fpath, 'w')
  89             fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
  90         except IOError, e:
  91             message('FileLock.lock(%s) : %s' % (self.fpath, e))
  92             return False
  93         return True
  94
  95     def unlock(self):
  96         try:
  97             fcntl.flock(self.fd, fcntl.LOCK_UN | fcntl.LOCK_NB)
  98             self.fd.close()
  99         except IOError, e:
 100             message('FileLock.unlock(%s) : %s' % (self.fpath, e))
 101
 102
 103 class RefreshPeer(Method):
 104     """
 105     Fetches site, node, slice, person and key data from the specified peer
 106     and caches it locally; also deletes stale entries.
 107     Upon successful completion, returns a dict reporting various timers.
 108     Faults otherwise.
 109     """
 110
 111     roles = ['admin']
 112
 113     accepts = [
 114         Auth(),
 115         Mixed(Peer.fields['peer_id'],
 116               Peer.fields['peername']),
 117     ]
 118
 119     returns = Parameter(int, "1 if successful")
 120
 121     ignore_site_fields = ['peer_id', 'peer_site_id', 'last_updated', 'date_created',
 122                           'address_ids', 'node_ids', 'person_ids', 'pcu_ids', 'slice_ids']
 123     ignore_key_fields = ['peer_id', 'peer_key_id', 'person_id']
 124     ignore_person_fields = ['peer_id', 'peer_person_id', 'last_updated', 'date_created',
 125                             'key_ids', 'slice_ids', 'person_tag_ids']
 126     ignore_node_fields = ['peer_id', 'peer_node_id', 'last_updated', 'last_contact', 'date_created',
 127                           'node_tag_ids', 'interface_ids', 'slice_ids', 'nodegroup_ids', 'pcu_ids', 'ports']
 128     ignore_slice_fields = ['peer_id', 'peer_slice_id', 'created',
 129                            'person_ids', 'slice_tag_ids', 'node_ids', ]
 130
 131     def call(self, auth, peer_id_or_peername):
 132         ret_val = None
 133         peername = Peers(self.api, [peer_id_or_peername], [
 134                          'peername'])[0]['peername']
 135         file_lock = FileLock("/tmp/refresh-peer-%s.lock" % peername)
 136         if not file_lock.lock():
 137             raise Exception, "Another instance of RefreshPeer is running."
 138         try:
 139             ret_val = self.real_call(auth, peer_id_or_peername)
 140         except Exception, e:
 141             file_lock.unlock()
 142             message("RefreshPeer caught exception - BEG")
 143             import traceback
 144             traceback.print_exc(file=log)
 145             message("RefreshPeer caught exception - END")
 146             raise Exception, e
 147         file_lock.unlock()
 148         return ret_val
 149
 150     def real_call(self, auth, peer_id_or_peername):
 151         # Get peer
 152         peers = Peers(self.api, [peer_id_or_peername])
 153         if not peers:
 154             raise PLCInvalidArgument, "No such peer '%s'" % unicode(peer_id_or_peername)
 155         peer = peers[0]
 156         peer_id = peer['peer_id']
 157
 158         # Connect to peer API
 159         peer.connect()
 160
 161         timers = {}
 162
 163         # Get peer data
 164         start = time.time()
 165         message('RefreshPeer starting up (commit_mode=%r)' % commit_mode)
 166         if not use_cache:
 167             message('Issuing GetPeerData')
 168             peer_tables = peer.GetPeerData()
 169         else:
 170             import pickle
 171             if os.path.isfile(use_cache):
 172                 message("use_cache: WARNING: using cached getpeerdata")
 173                 peer_tables = pickle.load(file(use_cache, 'rb'))
 174             else:
 175                 message("use_cache: issuing getpeerdata")
 176                 peer_tables = peer.GetPeerData()
 177                 message("use_cache: saving in cache %s", use_cache)
 178                 pickle.dump(peer_tables, file(use_cache, 'wb'))
 179
 180         # for smooth federation with 4.2 - ignore fields that are useless
 181         # anyway, and rewrite boot_state
 182         boot_state_rewrite = {'dbg': 'safeboot', 'diag': 'safeboot', 'disable': 'disabled',
 183                               'inst': 'reinstall', 'rins': 'reinstall', 'new': 'reinstall', 'rcnf': 'reinstall'}
 184         for node in peer_tables['Nodes']:
 185             for key in ['nodenetwork_ids', 'dummybox_id']:
 186                 if key in node:
 187                     del node[key]
 188             if node['boot_state'] in boot_state_rewrite:
 189                 node['boot_state'] = boot_state_rewrite[node['boot_state']]
 190         for slice in peer_tables['Slices']:
 191             for key in ['slice_attribute_ids']:
 192                 if key in slice:
 193                     del slice[key]
 194         timers['transport'] = time.time() - start - peer_tables['db_time']
 195         timers['peer_db'] = peer_tables['db_time']
 196         message_verbose('GetPeerData returned -> db=%d transport=%d' %
 197                         (timers['peer_db'], timers['transport']))
 198
 199         def sync(objects, peer_objects, classobj, columns):
 200             """
 201             Synchronizes two dictionaries of objects. objects should
 202             be a dictionary of local objects keyed on their foreign
 203             identifiers. peer_objects should be a dictionary of
 204             foreign objects keyed on their local (i.e., foreign to us)
 205             identifiers. Returns a final dictionary of local objects
 206             keyed on their foreign identifiers.
 207             """
 208
 209             classname = classobj(self.api).__class__.__name__
 210             primary_key = getattr(classobj, 'primary_key')
 211             # display all peer objects of these types while looping
 212             secondary_keys = {'Node': 'hostname', 'Slice': 'name',
 213                               'Person': 'email', 'Site': 'login_base'}
 214             secondary_key = None
 215             if classname in secondary_keys:
 216                 secondary_key = secondary_keys[classname]
 217
 218             message_verbose('Entering sync on %s (%s)' %
 219                             (classname, primary_key))
 220
 221             synced = {}
 222
 223             # Delete stale objects
 224             for peer_object_id, object in objects.iteritems():
 225                 if peer_object_id not in peer_objects:
 226                     object.delete(commit=commit_mode)
 227                     message("%s %s %s deleted" %
 228                             (peer['peername'], classname, object[primary_key]))
 229
 230             total = len(peer_objects)
 231             count = 1
 232
 233             # peer_object_id, peer_object and object are dynamically bound in the loop below...
 234             # (local) object might be None if creating a new one
 235             def in_focus():
 236                 if classname != focus_type:
 237                     return False
 238                 return peer_object_id in focus_ids or \
 239                     (object and primary_key in object and object[
 240                      primary_key] in focus_ids)
 241
 242             def message_focus(message):
 243                 if in_focus():
 244                     # always show remote
 245                     message_verbose("peer_obj : %d [[%r]]" % (peer_object_id, peer_object),
 246                                     header='FOCUS ' + message)
 247                     # show local object if a match was found
 248                     if object:
 249                         message_verbose("local_obj : <<%r>>" % (object),
 250                                         header='FOCUS ' + message)
 251
 252             # the function to compare a local object with its cadidate peer obj
 253             # xxx probably faster when compatibility is False...
 254             def equal_fields(object, peer_object, columns):
 255                 # fast version: must use __eq__() instead of == since
 256                 # peer_object may be a raw dict instead of a Peer object.
 257                 if not compatibility:
 258                     return object.__eq__(peer_object)
 259                 elif not verbose:
 260                     for column in columns:
 261                         #                        if in_focus(): message ('FOCUS comparing column %s'%column)
 262                         if object[column] != peer_object[column]:
 263                             return False
 264                     return True
 265                 else:
 266                     result = True
 267                     for column in columns:
 268                         test = object[column] == peer_object[column]
 269                         if not test:
 270                             result = False
 271                     return result
 272
 273             # Add/update new/existing objects
 274             for peer_object_id, peer_object in peer_objects.iteritems():
 275                 peer_object_name = ""
 276                 if secondary_key:
 277                     peer_object_name = "(%s)" % peer_object[secondary_key]
 278                 message_verbose('%s peer_object_id=%d %s (%d/%d)'
 279                                 % (classname, peer_object_id, peer_object_name, count, total))
 280                 count += 1
 281                 if peer_object_id in synced:
 282                     message("Warning: %s Skipping already added %s: %r" % (
 283                             peer['peername'], classname, peer_object))
 284                     continue
 285
 286                 if peer_object_id in objects:
 287                     # Update existing object
 288                     object = objects[peer_object_id]
 289
 290                     # Replace foreign identifier with existing local
 291                     # identifier temporarily for the purposes of
 292                     # comparison.
 293                     peer_object[primary_key] = object[primary_key]
 294
 295                     if not equal_fields(object, peer_object, columns):
 296                         # Only update intrinsic fields
 297                         object.update(object.db_fields(peer_object))
 298                         message_focus("DIFFERENCES : updated / syncing")
 299                         sync = True
 300                         action = "changed"
 301                     else:
 302                         message_focus("UNCHANGED - left intact / not syncing")
 303                         sync = False
 304                         action = None
 305
 306                     # Restore foreign identifier
 307                     peer_object[primary_key] = peer_object_id
 308                 else:
 309                     object = None
 310                     # Add new object
 311                     object = classobj(self.api, peer_object)
 312                     # Replace foreign identifier with new local identifier
 313                     del object[primary_key]
 314                     message_focus("NEW -- created with clean id - syncing")
 315                     sync = True
 316                     action = "added"
 317
 318                 if sync:
 319                     message_verbose("syncing %s %d - commit_mode=%r"
 320                                     % (classname, peer_object_id, commit_mode))
 321                     try:
 322                         object.sync(commit=commit_mode)
 323                     except PLCInvalidArgument, err:
 324                         # Skip if validation fails
 325                         # XXX Log an event instead of printing to logfile
 326                         message("Warning: %s Skipping invalid %s %r : %r" %
 327                                 (peer['peername'], classname, peer_object, err))
 328                         continue
 329
 330                 synced[peer_object_id] = object
 331
 332                 if action:
 333                     message("%s: (%d/%d) %s %d %s %s"
 334                             % (peer['peername'], count, total, classname,
 335                                object[primary_key], peer_object_name, action))
 336
 337             message_verbose("Exiting sync on %s" % classname)
 338
 339             return synced
 340
 341         # over time, we've had issues with a given column being
 342         # added on one side and not on the other
 343         # this helper function computes the intersection of two list of
 344         # fields/columns
 345         def intersect(l1, l2):
 346             if compatibility:
 347                 return list(set(l1).intersection(set(l2)))
 348             else:
 349                 return l1
 350
 351         # some fields definitely need to be ignored
 352         def ignore(l1, l2):
 353             return list(set(l1).difference(set(l2)))
 354
 355         #
 356         # Synchronize foreign sites
 357         #
 358
 359         start = time.time()
 360
 361         message('Dealing with Sites')
 362
 363         # Compare only the columns returned by the GetPeerData() call
 364         if peer_tables['Sites']:
 365             columns = peer_tables['Sites'][0].keys()
 366             columns = intersect(columns, Site.fields)
 367         else:
 368             columns = None
 369
 370         # Keyed on foreign site_id
 371         old_peer_sites = Sites(
 372             self.api, {'peer_id': peer_id}, columns).dict('peer_site_id')
 373         sites_at_peer = dict([(site['site_id'], site)
 374                               for site in peer_tables['Sites']])
 375
 376         # Synchronize new set (still keyed on foreign site_id)
 377         peer_sites = sync(old_peer_sites, sites_at_peer, Site,
 378                           ignore(columns, RefreshPeer.ignore_site_fields))
 379
 380         for peer_site_id, site in peer_sites.iteritems():
 381             # Bind any newly cached sites to peer
 382             if peer_site_id not in old_peer_sites:
 383                 peer.add_site(site, peer_site_id, commit=commit_mode)
 384                 site['peer_id'] = peer_id
 385                 site['peer_site_id'] = peer_site_id
 386
 387         timers['site'] = time.time() - start
 388
 389         #
 390         # XXX Synchronize foreign key types
 391         #
 392
 393         message('Dealing with Keys')
 394
 395         key_types = KeyTypes(self.api).dict()
 396
 397         #
 398         # Synchronize foreign keys
 399         #
 400
 401         start = time.time()
 402
 403         # Compare only the columns returned by the GetPeerData() call
 404         if peer_tables['Keys']:
 405             columns = peer_tables['Keys'][0].keys()
 406             columns = intersect(columns, Key.fields)
 407         else:
 408             columns = None
 409
 410         # Keyed on foreign key_id
 411         old_peer_keys = Keys(
 412             self.api, {'peer_id': peer_id}, columns).dict('peer_key_id')
 413         keys_at_peer = dict([(key['key_id'], key)
 414                              for key in peer_tables['Keys']])
 415
 416         # Fix up key_type references
 417         for peer_key_id, key in keys_at_peer.items():
 418             if key['key_type'] not in key_types:
 419                 # XXX Log an event instead of printing to logfile
 420                 message("Warning: Skipping invalid %s key %r" %
 421                         (peer['peername'], key))
 422                 del keys_at_peer[peer_key_id]
 423                 continue
 424
 425         # Synchronize new set (still keyed on foreign key_id)
 426         peer_keys = sync(old_peer_keys, keys_at_peer, Key,
 427                          ignore(columns, RefreshPeer.ignore_key_fields))
 428         for peer_key_id, key in peer_keys.iteritems():
 429             # Bind any newly cached keys to peer
 430             if peer_key_id not in old_peer_keys:
 431                 peer.add_key(key, peer_key_id, commit=commit_mode)
 432                 key['peer_id'] = peer_id
 433                 key['peer_key_id'] = peer_key_id
 434
 435         timers['keys'] = time.time() - start
 436
 437         #
 438         # Synchronize foreign users
 439         #
 440
 441         start = time.time()
 442
 443         message('Dealing with Persons')
 444
 445         # Compare only the columns returned by the GetPeerData() call
 446         if peer_tables['Persons']:
 447             columns = peer_tables['Persons'][0].keys()
 448             columns = intersect(columns, Person.fields)
 449         else:
 450             columns = None
 451
 452         # Keyed on foreign person_id
 453         old_peer_persons = Persons(
 454             self.api, {'peer_id': peer_id}, columns).dict('peer_person_id')
 455
 456         # artificially attach the persons returned by GetPeerData to the new peer
 457         # this is because validate_email needs peer_id to be correct when
 458         # checking for duplicates
 459         for person in peer_tables['Persons']:
 460             person['peer_id'] = peer_id
 461         persons_at_peer = dict([(peer_person['person_id'], peer_person)
 462                                 for peer_person in peer_tables['Persons']])
 463
 464         # XXX Do we care about membership in foreign site(s)?
 465
 466         # Synchronize new set (still keyed on foreign person_id)
 467         peer_persons = sync(old_peer_persons, persons_at_peer, Person,
 468                             ignore(columns, RefreshPeer.ignore_person_fields))
 469
 470         # transcoder : retrieve a local key_id from a peer_key_id
 471         key_transcoder = dict([(key['key_id'], peer_key_id)
 472                                for peer_key_id, key in peer_keys.iteritems()])
 473
 474         for peer_person_id, person in peer_persons.iteritems():
 475             # Bind any newly cached users to peer
 476             if peer_person_id not in old_peer_persons:
 477                 peer.add_person(person, peer_person_id, commit=commit_mode)
 478                 person['peer_id'] = peer_id
 479                 person['peer_person_id'] = peer_person_id
 480                 person['key_ids'] = []
 481
 482             # User as viewed by peer
 483             peer_person = persons_at_peer[peer_person_id]
 484
 485             # Foreign keys currently belonging to the user
 486             old_person_key_ids = [key_transcoder[key_id] for key_id in person['key_ids']
 487                                   if key_transcoder[key_id] in peer_keys]
 488
 489             # Foreign keys that should belong to the user
 490             # this is basically peer_person['key_ids'], we just check it makes sense
 491             # (e.g. we might have failed importing it)
 492             person_key_ids = [key_id for key_id in peer_person[
 493                 'key_ids'] if key_id in peer_keys]
 494
 495             # Remove stale keys from user
 496             for key_id in (set(old_person_key_ids) - set(person_key_ids)):
 497                 person.remove_key(peer_keys[key_id], commit=commit_mode)
 498                 message("%s Key %d removed from person %s" %
 499                         (peer['peername'], key_id, person['email']))
 500
 501             # Add new keys to user
 502             for key_id in (set(person_key_ids) - set(old_person_key_ids)):
 503                 message("before add_key, passing person=%r" % person)
 504                 message("before add_key, passing key=%r" % peer_keys[key_id])
 505                 person.add_key(peer_keys[key_id], commit=commit_mode)
 506                 message("%s Key %d added into person %s" %
 507                         (peer['peername'], key_id, person['email']))
 508
 509         timers['persons'] = time.time() - start
 510
 511         #
 512         # XXX Synchronize foreign boot states
 513         #
 514
 515         boot_states = BootStates(self.api).dict()
 516
 517         #
 518         # Synchronize foreign nodes
 519         #
 520
 521         start = time.time()
 522
 523         message('Dealing with Nodes (1)')
 524
 525         # Compare only the columns returned by the GetPeerData() call
 526         if peer_tables['Nodes']:
 527             columns = peer_tables['Nodes'][0].keys()
 528             columns = intersect(columns, Node.fields)
 529         else:
 530             columns = Node.fields
 531
 532         # Keyed on foreign node_id
 533         old_peer_nodes = Nodes(
 534             self.api, {'peer_id': peer_id}, columns).dict('peer_node_id')
 535         nodes_at_peer = dict([(node['node_id'], node)
 536                               for node in peer_tables['Nodes']])
 537
 538         # Fix up site_id and boot_states references
 539         for peer_node_id, node in nodes_at_peer.items():
 540             errors = []
 541             if node['site_id'] not in peer_sites:
 542                 errors.append("invalid site %d" % node['site_id'])
 543             if node['boot_state'] not in boot_states:
 544                 errors.append("invalid boot state %s" % node['boot_state'])
 545             if errors:
 546                 # XXX Log an event instead of printing to logfile
 547                 message("Warning: Skipping invalid %s node %r : " % (peer['peername'], node)
 548                         + ", ".join(errors))
 549                 del nodes_at_peer[peer_node_id]
 550                 continue
 551             else:
 552                 node['site_id'] = peer_sites[node['site_id']]['site_id']
 553
 554         # Synchronize new set
 555         peer_nodes = sync(old_peer_nodes, nodes_at_peer, Node,
 556                           ignore(columns, RefreshPeer.ignore_node_fields))
 557
 558         for peer_node_id, node in peer_nodes.iteritems():
 559             # Bind any newly cached foreign nodes to peer
 560             if peer_node_id not in old_peer_nodes:
 561                 peer.add_node(node, peer_node_id, commit=commit_mode)
 562                 node['peer_id'] = peer_id
 563                 node['peer_node_id'] = peer_node_id
 564
 565         timers['nodes'] = time.time() - start
 566
 567         #
 568         # Synchronize local nodes
 569         #
 570
 571         start = time.time()
 572         message('Dealing with Nodes (2)')
 573
 574         # Keyed on local node_id
 575         local_nodes = Nodes(self.api).dict()
 576
 577         for node in peer_tables['PeerNodes']:
 578             # Foreign identifier for our node as maintained by peer
 579             peer_node_id = node['node_id']
 580             # Local identifier for our node as cached by peer
 581             node_id = node['peer_node_id']
 582             if node_id in local_nodes:
 583                 # Still a valid local node, add it to the synchronized
 584                 # set of local node objects keyed on foreign node_id.
 585                 peer_nodes[peer_node_id] = local_nodes[node_id]
 586
 587         timers['local_nodes'] = time.time() - start
 588
 589         #
 590         # XXX Synchronize foreign slice instantiation states
 591         #
 592
 593         slice_instantiations = SliceInstantiations(self.api).dict()
 594
 595         #
 596         # Synchronize foreign slices
 597         #
 598
 599         start = time.time()
 600
 601         message('Dealing with Slices (1)')
 602
 603         # Compare only the columns returned by the GetPeerData() call
 604         if peer_tables['Slices']:
 605             columns = peer_tables['Slices'][0].keys()
 606             columns = intersect(columns, Slice.fields)
 607         else:
 608             columns = None
 609
 610         # Keyed on foreign slice_id
 611         old_peer_slices = Slices(
 612             self.api, {'peer_id': peer_id}, columns).dict('peer_slice_id')
 613         slices_at_peer = dict([(slice['slice_id'], slice)
 614                                for slice in peer_tables['Slices']])
 615
 616         # Fix up site_id, instantiation, and creator_person_id references
 617         for peer_slice_id, slice in slices_at_peer.items():
 618             errors = []
 619             if slice['site_id'] not in peer_sites:
 620                 errors.append("invalid site %d" % slice['site_id'])
 621             if slice['instantiation'] not in slice_instantiations:
 622                 errors.append("invalid instantiation %s" %
 623                               slice['instantiation'])
 624             if slice['creator_person_id'] not in peer_persons:
 625                 # Just NULL it out
 626                 slice['creator_person_id'] = None
 627             else:
 628                 slice['creator_person_id'] = peer_persons[
 629                     slice['creator_person_id']]['person_id']
 630             if errors:
 631                 message("Warning: Skipping invalid %s slice %r : " % (peer['peername'], slice)
 632                         + ", ".join(errors))
 633                 del slices_at_peer[peer_slice_id]
 634                 continue
 635             else:
 636                 slice['site_id'] = peer_sites[slice['site_id']]['site_id']
 637
 638         # Synchronize new set
 639         peer_slices = sync(old_peer_slices, slices_at_peer, Slice, ignore(
 640             columns, RefreshPeer.ignore_slice_fields))
 641
 642         message('Dealing with Slices (2)')
 643         # transcoder : retrieve a local node_id from a peer_node_id
 644         node_transcoder = dict([(node['node_id'], peer_node_id)
 645                                 for peer_node_id, node in peer_nodes.iteritems()])
 646         person_transcoder = dict([(person['person_id'], peer_person_id)
 647                                   for peer_person_id, person in peer_persons.iteritems()])
 648
 649         for peer_slice_id, slice in peer_slices.iteritems():
 650             # Bind any newly cached foreign slices to peer
 651             if peer_slice_id not in old_peer_slices:
 652                 peer.add_slice(slice, peer_slice_id, commit=commit_mode)
 653                 slice['peer_id'] = peer_id
 654                 slice['peer_slice_id'] = peer_slice_id
 655                 slice['node_ids'] = []
 656                 slice['person_ids'] = []
 657
 658             # Slice as viewed by peer
 659             peer_slice = slices_at_peer[peer_slice_id]
 660
 661             # Nodes that are currently part of the slice
 662             old_slice_node_ids = [node_transcoder[node_id] for node_id in slice['node_ids']
 663                                   if node_id in node_transcoder and node_transcoder[node_id] in peer_nodes]
 664
 665             # Nodes that should be part of the slice
 666             slice_node_ids = [node_id for node_id in peer_slice[
 667                 'node_ids'] if node_id in peer_nodes]
 668
 669             # Remove stale nodes from slice
 670             for node_id in (set(old_slice_node_ids) - set(slice_node_ids)):
 671                 slice.remove_node(peer_nodes[node_id], commit=commit_mode)
 672                 message("%s node %s removed from slice %s" % (
 673                     peer['peername'], peer_nodes[node_id]['hostname'], slice['name']))
 674
 675             # Add new nodes to slice
 676             for node_id in (set(slice_node_ids) - set(old_slice_node_ids)):
 677                 slice.add_node(peer_nodes[node_id], commit=commit_mode)
 678                 message("%s node %s added into slice %s" % (
 679                     peer['peername'], peer_nodes[node_id]['hostname'], slice['name']))
 680
 681             # N.B.: Local nodes that may have been added to the slice
 682             # by hand, are removed. In other words, don't do this.
 683
 684             # Foreign users that are currently part of the slice
 685             # old_slice_person_ids = [ person_transcoder[person_id] for person_id in slice['person_ids'] \
 686             #                if person_transcoder[person_id] in peer_persons]
 687             # An issue occurred with a user who registered on both sites (same email)
 688             # So the remote person could not get cached locally
 689             # The one-line map/filter style is nicer but ineffective here
 690             old_slice_person_ids = []
 691             for person_id in slice['person_ids']:
 692                 if not person_transcoder.has_key(person_id):
 693                     message('WARNING : person_id %d in %s not transcodable (1) - skipped' %
 694                             (person_id, slice['name']))
 695                 elif person_transcoder[person_id] not in peer_persons:
 696                     message('WARNING : person_id %d in %s not transcodable (2) - skipped' %
 697                             (person_id, slice['name']))
 698                 else:
 699                     old_slice_person_ids += [person_transcoder[person_id]]
 700
 701             # Foreign users that should be part of the slice
 702             slice_person_ids = [person_id for person_id in peer_slice[
 703                 'person_ids'] if person_id in peer_persons]
 704
 705             # Remove stale users from slice
 706             for person_id in (set(old_slice_person_ids) - set(slice_person_ids)):
 707                 slice.remove_person(
 708                     peer_persons[person_id], commit=commit_mode)
 709                 message("%s user %s removed from slice %s" % (
 710                     peer['peername'], peer_persons[person_id]['email'], slice['name']))
 711
 712             # Add new users to slice
 713             for person_id in (set(slice_person_ids) - set(old_slice_person_ids)):
 714                 slice.add_person(peer_persons[person_id], commit=commit_mode)
 715                 message("%s user %s added into slice %s" % (
 716                     peer['peername'], peer_persons[person_id]['email'], slice['name']))
 717
 718             # N.B.: Local users that may have been added to the slice
 719             # by hand, are not touched.
 720
 721         timers['slices'] = time.time() - start
 722
 723         #
 724         # Persons x Sites
 725         #
 726         start = time.time()
 727
 728         message('Dealing Sites X Persons relationship')
 729
 730         for peer_site_id, site in peer_sites.iteritems():
 731             # Site as viewed by peer
 732             peer_site = sites_at_peer[peer_site_id]
 733
 734             # Persons that are currently part of the site
 735             old_site_person_ids = [person_transcoder[person_id] for person_id in site['person_ids']
 736                                    if person_id in person_transcoder and person_transcoder[person_id] in peer_persons]
 737
 738             # Perons that should be part of the site
 739             site_person_ids = [person_id for person_id in peer_site[
 740                 'person_ids'] if person_id in peer_persons]
 741
 742             # Remove stale persons from site
 743             for person_id in (set(old_site_person_ids) - set(site_person_ids)):
 744                 site.remove_person(peer_persons[person_id], commit=commit_mode)
 745                 message("%s person %s removed from site %s" % (
 746                     peer['peername'], peer_persons[person_id]['email'], site['login_base']))
 747
 748             # Add new persons to site
 749             for person_id in (set(site_person_ids) - set(old_site_person_ids)):
 750                 site.add_person(peer_persons[person_id], commit=commit_mode)
 751                 message("%s person %s added into site %s" % (
 752                     peer['peername'], peer_persons[person_id]['email'], site['login_base']))
 753
 754         timers['sites-persons'] = time.time() - start
 755
 756         #
 757         # Persons x Roles
 758         #
 759         start = time.time()
 760
 761         message('Dealing with Persons Roles relationship')
 762
 763         roles = Roles(self.api)
 764         roles_dict = dict([(role['role_id'], role) for role in roles])
 765         for peer_person_id, person in peer_persons.iteritems():
 766             # Person as viewed by peer
 767             peer_person = persons_at_peer[peer_person_id]
 768
 769             # Roles that are currently attributed for the person
 770             old_person_role_ids = [role_id for role_id in person['role_ids']]
 771
 772             # Roles that should be attributed to the person
 773             person_role_ids = [role_id for role_id in peer_person['role_ids']]
 774
 775             # Remove stale roles
 776             for role_id in (set(old_person_role_ids) - set(person_role_ids)):
 777                 person.remove_role(roles_dict[role_id], commit=commit_mode)
 778                 message("%s role %s removed from person %s" % (
 779                     peer['peername'], roles_dict[role_id]['name'], person['email']))
 780
 781             # Add new roles to person
 782             for role_id in (set(person_role_ids) - set(old_person_role_ids)):
 783                 person.add_role(roles_dict[role_id], commit=commit_mode)
 784                 message("%s role %s added from person %s" % (
 785                     peer['peername'], roles_dict[role_id]['name'], person['email']))
 786
 787         timers['persons-roles'] = time.time() - start
 788
 789         # Update peer itself and commit
 790         peer.sync(commit=True)
 791
 792         return timers