PLC/Cache.py

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