PLC/Cache.py

   1 import time
   2
   3 from PLC.Faults import *
   4 from PLC.Parameter import Parameter
   5 from PLC.Filter import Filter
   6 from PLC.Table import Row, Table
   7
   8 verbose_flag=False;
   9 #verbose_flag=True;
  10 def verbose (*args):
  11     if verbose_flag:
  12         print (args)
  13
  14 def class_attributes (classname):
  15     """ locates various attributes defined in the row class """
  16     topmodule = __import__ ('PLC.%ss'%classname)
  17     module = topmodule.__dict__['%ss'%classname]
  18     # local row-like class, e.g. Node
  19     row_class = module.__dict__['%s'%classname]
  20     # local tab-like class, e.g. Nodes
  21     table_class = module.__dict__['%ss'%classname]
  22
  23     return {'row_class':row_class,
  24             'table_class':table_class,
  25             'primary_key': row_class.__dict__['primary_key'],
  26             'class_key': row_class.__dict__['class_key'],
  27             'foreign_fields': row_class.__dict__['foreign_fields'],
  28             'foreign_xrefs': row_class.__dict__['foreign_xrefs'],
  29             }
  30
  31 class Cache:
  32
  33     # an attempt to provide genericity in the caching algorithm
  34
  35     def __init__ (self, api, peer_id, peer_server, auth):
  36
  37         self.api = api
  38         self.peer_id = peer_id
  39         self.peer_server = peer_server
  40         self.auth = auth
  41
  42     class Transcoder:
  43
  44         def __init__ (self, api, classname, alien_objects):
  45             self.api = api
  46             attrs = class_attributes (classname)
  47             self.primary_key = attrs['primary_key']
  48             self.class_key = attrs['class_key']
  49
  50             # cannot use dict, it's acquired by xmlrpc and is untyped
  51             self.alien_objects_byid = dict( [ (x[self.primary_key],x) for x in alien_objects ] )
  52
  53             # retrieve local objects
  54             local_objects = attrs['table_class'] (api)
  55             self.local_objects_byname = local_objects.dict(self.class_key)
  56
  57             verbose ('Transcoder init :',classname,
  58                      self.alien_objects_byid.keys(),
  59                      self.local_objects_byname.keys())
  60
  61         def transcode (self, alien_id):
  62             """ transforms an alien id into a local one """
  63             # locate alien obj from alien_id
  64             #verbose ('.entering transcode with alien_id',alien_id,)
  65             alien_object=self.alien_objects_byid[alien_id]
  66             #verbose ('..located alien_obj',)
  67             name = alien_object [self.class_key]
  68             #verbose ('...got name',name,)
  69             local_object=self.local_objects_byname[name]
  70             #verbose ('....found local obj')
  71             local_id=local_object[self.primary_key]
  72             #verbose ('.....and local_id',local_id)
  73             return local_id
  74
  75
  76     # for handling simple n-to-n relation tables, like e.g. slice_node
  77     class XrefTable:
  78
  79         def __init__ (self, api, tablename, class1, class2):
  80             self.api = api
  81             self.tablename = tablename
  82             self.lowerclass1 = class1.lower()
  83             self.lowerclass2 = class2.lower()
  84
  85         def delete_old_items (self, id1, id2_set):
  86             if id2_set:
  87                 sql = ""
  88                 sql += "DELETE FROM %s WHERE %s_id=%d"%(self.tablename,self.lowerclass1,id1)
  89                 sql += " AND %s_id IN ("%self.lowerclass2
  90                 sql += ",".join([str(i) for i in id2_set])
  91                 sql += ")"
  92                 self.api.db.do (sql)
  93
  94         def insert_new_items (self, id1, id2_set):
  95             if id2_set:
  96                 sql = "INSERT INTO %s select %d, %d " % \
  97                         self.tablename, id1, id2[0]
  98                 for id2 in id2_set[1:]:
  99                         sql += " UNION ALL SELECT %d, %d " % \
 100                         (id1,id2)
 101                 self.api.db.do (sql)
 102
 103         def update_item (self, id1, old_id2s, new_id2s):
 104             news = set (new_id2s)
 105             olds = set (old_id2s)
 106             to_delete = olds-news
 107             self.delete_old_items (id1, to_delete)
 108             to_create = news-olds
 109             self.insert_new_items (id1, to_create)
 110             self.api.db.commit()
 111
 112     # classname: the type of objects we are talking about;       e.g. 'Slice'
 113     # peer_object_list list of objects at a given peer -         e.g. peer.GetSlices()
 114     # alien_xref_objs_dict : a dict {'classname':alien_obj_list} e.g. {'Node':peer.GetNodes()}
 115     #    we need an entry for each class mentioned in the class's foreign_xrefs
 116     # lambda_ignore : the alien objects are ignored if this returns true
 117     def update_table (self,
 118                       classname,
 119                       alien_object_list,
 120                       alien_xref_objs_dict = {},
 121                       lambda_ignore=lambda x:False,
 122                       report_name_conflicts = True):
 123
 124         verbose ("============================== entering update_table on",classname)
 125         peer_id=self.peer_id
 126
 127         attrs = class_attributes (classname)
 128         row_class = attrs['row_class']
 129         table_class = attrs['table_class']
 130         primary_key = attrs['primary_key']
 131         class_key = attrs['class_key']
 132         foreign_fields = attrs['foreign_fields']
 133         foreign_xrefs = attrs['foreign_xrefs']
 134
 135         ## allocate transcoders and xreftables once, for each item in foreign_xrefs
 136         # create a dict 'classname' -> {'transcoder' : ..., 'xref_table' : ...}
 137         xref_accessories = dict(
 138             [ (xref['field'],
 139                {'transcoder' : Cache.Transcoder (self.api,xref['class'],alien_xref_objs_dict[xref['class']]),
 140                 'xref_table' : Cache.XrefTable (self.api,xref['table'],classname,xref['class'])})
 141               for xref in foreign_xrefs ])
 142
 143         # the fields that are direct references, like e.g. site_id in Node
 144         # determined lazily, we need an alien_object to do that, and we may have none here
 145         direct_ref_fields = None
 146
 147         ### get current local table
 148         # get ALL local objects so as to cope with
 149         # (*) potential moves between plcs
 150         # (*) or naming conflicts
 151         local_objects = table_class (self.api)
 152         ### index upon class_key for future searches
 153         local_objects_index = local_objects.dict(class_key)
 154
 155         #verbose ('update_table',classname,local_objects_index.keys())
 156
 157         ### mark entries for this peer outofdate
 158         new_count=0
 159         old_count=0;
 160         for local_object in local_objects:
 161             if local_object['peer_id'] == peer_id:
 162                 local_object.uptodate=False
 163                 old_count += 1
 164             else:
 165                 local_object.uptodate=True
 166
 167         # scan the peer's local objects
 168         for alien_object in alien_object_list:
 169
 170             object_name = alien_object[class_key]
 171
 172             ### ignore, e.g. system-wide slices
 173             if lambda_ignore(alien_object):
 174                 verbose('Ignoring',object_name)
 175                 continue
 176
 177             verbose ('update_table (%s) - Considering'%classname,object_name)
 178
 179             # create or update
 180             try:
 181                 ### We know about this object already
 182                 local_object = local_objects_index[object_name]
 183                 if local_object ['peer_id'] is None:
 184                     if report_name_conflicts:
 185                         ### xxx send e-mail
 186                         print '!!!!!!!!!! We are in trouble here'
 187                         print 'The %s object named %s is natively defined twice, '%(classname,object_name),
 188                         print 'once on this PLC and once on peer %d'%peer_id
 189                         print 'We dont raise an exception so that the remaining updates can still take place'
 190                         print '!!!!!!!!!!'
 191                     continue
 192                 if local_object['peer_id'] != peer_id:
 193                     ### the object has changed its plc,
 194                     ### Note, this is not problematic here because both definitions are remote
 195                     ### we can assume the object just moved
 196                     ### needs to update peer_id though
 197                     local_object['peer_id'] = peer_id
 198                 # update all fields as per foreign_fields
 199                 for field in foreign_fields:
 200                     local_object[field]=alien_object[field]
 201                 verbose ('update_table FOUND',object_name)
 202             except:
 203                 ### create a new entry
 204                 local_object = row_class(self.api,
 205                                           {class_key :object_name,'peer_id':peer_id})
 206                 # insert in index
 207                 local_objects_index[class_key]=local_object
 208                 verbose ('update_table CREATED',object_name)
 209                 # update all fields as per foreign_fields
 210                 for field in foreign_fields:
 211                     local_object[field]=alien_object[field]
 212                 # this is tricky; at this point we may have primary_key unspecified,
 213                 # but we need it for handling xrefs below, so we'd like to sync to get one
 214                 # on the other hand some required fields may be still missing so
 215                 #  the DB would refuse to sync in this case (e.g. site_id in Node)
 216                 # so let's fill them with 1 so we can sync, this will be overridden below
 217                 # lazily determine this set of fields now
 218                 if direct_ref_fields is None:
 219                     direct_ref_fields=[]
 220                     for xref in foreign_xrefs:
 221                         field=xref['field']
 222                         verbose('checking field %s for direct_ref'%field)
 223                         if isinstance(alien_object[field],int):
 224                             direct_ref_fields.append(field)
 225                     verbose("FOUND DIRECT REFS",direct_ref_fields)
 226                 for field in direct_ref_fields:
 227                     local_object[field]=1
 228                 verbose('Early sync on',local_object)
 229                 local_object.sync()
 230
 231             # this row is now valid
 232             local_object.uptodate=True
 233             new_count += 1
 234
 235             # manage cross-refs
 236             for xref in foreign_xrefs:
 237                 field=xref['field']
 238                 alien_xref_obj_list = alien_xref_objs_dict[xref['class']]
 239                 alien_value = alien_object[field]
 240                 transcoder = xref_accessories[xref['field']]['transcoder']
 241                 if isinstance (alien_value,list):
 242                     #verbose ('update_table list-transcoding ',xref['class'],' aliens=',alien_value,)
 243                     local_values=[]
 244                     for a in alien_value:
 245                         try:
 246                             local_values.append(transcoder.transcode(a))
 247                         except:
 248                             # could not transcode - might be from another peer that we dont know about..
 249                             pass
 250                     #verbose (" transcoded as ",local_values)
 251                     xref_table = xref_accessories[xref['field']]['xref_table']
 252                     # newly created objects dont have xref fields set yet
 253                     try:
 254                         former_xrefs=local_object[xref['field']]
 255                     except:
 256                         former_xrefs=[]
 257                     xref_table.update_item (local_object[primary_key],
 258                                             former_xrefs,
 259                                             local_values)
 260                 elif isinstance (alien_value,int):
 261                     #verbose ('update_table atom-transcoding ',xref['class'],' aliens=',alien_value,)
 262                     new_value = transcoder.transcode(alien_value)
 263                     local_object[field] = new_value
 264
 265             ### this object is completely updated, let's save it
 266             verbose('FINAL sync on %s:'%object_name,local_object)
 267             local_object.sync()
 268
 269
 270         ### delete entries that are not uptodate
 271         for local_object in local_objects:
 272             if not local_object.uptodate:
 273                 local_object.delete()
 274
 275         self.api.db.commit()
 276
 277         ### return delta in number of objects
 278         return new_count-old_count
 279
 280     # slice attributes exhibit a special behaviour
 281     # because there is no name we can use to retrieve/check for equality
 282     # this object is like a 3-part xref, linking slice_attribute_type, slice,
 283     #    and potentially node, together with a value that can change over time.
 284     # extending the generic model to support a lambda rather than class_key
 285     #    would clearly become overkill
 286     def update_slice_attributes (self,
 287                                  alien_slice_attributes,
 288                                  alien_nodes,
 289                                  alien_slices):
 290
 291         from PLC.SliceAttributeTypes import SliceAttributeTypes
 292         from PLC.SliceAttributes import SliceAttribute, SliceAttributes
 293
 294         # init
 295         peer_id = self.peer_id
 296
 297         # create transcoders
 298         node_xcoder = Cache.Transcoder (self.api, 'Node', alien_nodes)
 299         slice_xcoder= Cache.Transcoder (self.api, 'Slice', alien_slices)
 300         # no need to transcode SliceAttributeTypes, we have a name in the result
 301         local_sat_dict = SliceAttributeTypes(self.api).dict('name')
 302
 303         # load local objects
 304         local_objects = SliceAttributes (self.api,{'peer_id':peer_id})
 305
 306         ### mark entries for this peer outofdate
 307         new_count = 0
 308         old_count=len(local_objects)
 309         for local_object in local_objects:
 310             local_object.uptodate=False
 311
 312         for alien_object in alien_slice_attributes:
 313
 314             verbose('----- update_slice_attributes: considering ...')
 315             verbose('   ',alien_object)
 316
 317             # locate local slice
 318             try:
 319                 slice_id = slice_xcoder.transcode(alien_object['slice_id'])
 320             except:
 321                 verbose('update_slice_attributes: unable to locate slice',
 322                         alien_object['slice_id'])
 323                 continue
 324             # locate slice_attribute_type
 325             try:
 326                 sat_id = local_sat_dict[alien_object['name']]['attribute_type_id']
 327             except:
 328                 verbose('update_slice_attributes: unable to locate slice attribute type',
 329                         alien_object['name'])
 330                 continue
 331             # locate local node if specified
 332             try:
 333                 alien_node_id = alien_object['node_id']
 334                 if alien_node_id is not None:
 335                     node_id = node_xcoder.transcode(alien_node_id)
 336                 else:
 337                     node_id=None
 338             except:
 339                 verbose('update_slice_attributes: unable to locate node',
 340                         alien_object['node_id'])
 341                 continue
 342
 343             # locate the local SliceAttribute if any
 344             try:
 345                 verbose ('searching name=', alien_object['name'],
 346                          'slice_id',slice_id, 'node_id',node_id)
 347                 local_object = SliceAttributes (self.api,
 348                                                 {'name':alien_object['name'],
 349                                                  'slice_id':slice_id,
 350                                                  'node_id':node_id})[0]
 351
 352                 if local_object['peer_id'] != peer_id:
 353                     verbose ('FOUND local sa - skipped')
 354                     continue
 355                 verbose('FOUND already cached sa - setting value')
 356                 local_object['value'] = alien_object['value']
 357             # create it if missing
 358             except:
 359                 local_object = SliceAttribute(self.api,
 360                                               {'peer_id':peer_id,
 361                                                'slice_id':slice_id,
 362                                                'node_id':node_id,
 363                                                'attribute_type_id':sat_id,
 364                                                'value':alien_object['value']})
 365                 verbose('CREATED new sa')
 366             local_object.uptodate=True
 367             new_count += 1
 368             local_object.sync()
 369
 370         for local_object in local_objects:
 371             if not local_object.uptodate:
 372                 local_object.delete()
 373
 374         self.api.db.commit()
 375         ### return delta in number of objects
 376         return new_count-old_count
 377
 378     def refresh_peer (self):
 379
 380         # so as to minimize the numer of requests
 381         # we get all objects in a single call and sort afterwards
 382         # xxx ideally get objects either local or the ones attached here
 383         # requires to know remote peer's peer_id for ourselves, mmhh..
 384         # does not make any difference in a 2-peer deployment though
 385
 386         ### uses GetPeerData to gather all info in a single xmlrpc request
 387
 388         t_start=time.time()
 389         # xxx see also GetPeerData - peer_id arg unused yet
 390         all_data = self.peer_server.GetPeerData (self.auth,0)
 391
 392         t_acquired = time.time()
 393         # refresh sites
 394         plocal_sites = all_data['Sites-local']
 395         all_sites = plocal_sites + all_data['Sites-peer']
 396         nb_new_sites = self.update_table('Site', plocal_sites)
 397
 398         # refresh keys
 399         plocal_keys = all_data['Keys-local']
 400         all_keys = plocal_keys + all_data['Keys-peer']
 401         nb_new_keys = self.update_table('Key', plocal_keys)
 402
 403         # refresh nodes
 404         plocal_nodes = all_data['Nodes-local']
 405         all_nodes = plocal_nodes + all_data['Nodes-peer']
 406         nb_new_nodes = self.update_table('Node', plocal_nodes,
 407                                          { 'Site' : all_sites } )
 408
 409         # refresh persons
 410         plocal_persons = all_data['Persons-local']
 411         all_persons = plocal_persons + all_data['Persons-peer']
 412         nb_new_persons = self.update_table ('Person', plocal_persons,
 413                                             { 'Key': all_keys, 'Site' : all_sites } )
 414
 415         # refresh slice attribute types
 416         plocal_slice_attribute_types = all_data ['SliceAttibuteTypes-local']
 417         nb_new_slice_attribute_types = self.update_table ('SliceAttributeType',
 418                                                           plocal_slice_attribute_types,
 419                                                           report_name_conflicts = False)
 420
 421         # refresh slices
 422         plocal_slices = all_data['Slices-local']
 423         all_slices = plocal_slices + all_data['Slices-peer']
 424
 425         def is_system_slice (slice):
 426             return slice['creator_person_id'] == 1
 427
 428         nb_new_slices = self.update_table ('Slice', plocal_slices,
 429                                            {'Node': all_nodes,
 430                                             'Person': all_persons,
 431                                             'Site': all_sites},
 432                                            is_system_slice)
 433
 434         # refresh slice attributes
 435         plocal_slice_attributes = all_data ['SliceAttributes-local']
 436         nb_new_slice_attributes = self.update_slice_attributes (plocal_slice_attributes,
 437                                                                 all_nodes,
 438                                                                 all_slices)
 439
 440         t_end=time.time()
 441         ### returned as-is by RefreshPeer
 442         return {'plcname':self.api.config.PLC_NAME,
 443                 'new_sites':nb_new_sites,
 444                 'new_keys':nb_new_keys,
 445                 'new_nodes':nb_new_nodes,
 446                 'new_persons':nb_new_persons,
 447                 'new_slice_attribute_types':nb_new_slice_attribute_types,
 448                 'new_slices':nb_new_slices,
 449                 'new_slice_attributes':nb_new_slice_attributes,
 450                 'time_gather': all_data['ellapsed'],
 451                 'time_transmit':t_acquired-t_start-all_data['ellapsed'],
 452                 'time_process':t_end-t_acquired,
 453                 'time_all':t_end-t_start,
 454                 }
 455