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     # the Peer object we are syncing with
  34     def __init__ (self, api, peer, peer_server, auth):
  35
  36         import PLC.Peers
  37
  38         self.api = api
  39         assert isinstance(peer,PLC.Peers.Peer)
  40         self.peer = peer
  41         self.peer_server = peer_server
  42         self.auth = auth
  43
  44     class Transcoder:
  45
  46         def __init__ (self, api, classname, alien_objects):
  47             self.api = api
  48             attrs = class_attributes (classname)
  49             self.primary_key = attrs['primary_key']
  50             self.class_key = attrs['class_key']
  51
  52             # cannot use dict, it's acquired by xmlrpc and is untyped
  53             self.alien_objects_byid = dict( [ (x[self.primary_key],x) for x in alien_objects ] )
  54
  55             # retrieve local objects
  56             local_objects = attrs['table_class'] (api)
  57             self.local_objects_byname = local_objects.dict(self.class_key)
  58
  59             verbose ('Transcoder init :',classname,
  60                      self.alien_objects_byid.keys(),
  61                      self.local_objects_byname.keys())
  62
  63         def transcode (self, alien_id):
  64             """ transforms an alien id into a local one """
  65             # locate alien obj from alien_id
  66             verbose ('entering transcode with alien_id',alien_id,)
  67             alien_object=self.alien_objects_byid[alien_id]
  68             verbose ('located alien_obj',)
  69             name = alien_object [self.class_key]
  70             verbose ('got name',name,)
  71             local_object=self.local_objects_byname[name]
  72             verbose ('found local obj')
  73             local_id=local_object[self.primary_key]
  74             verbose ('and local_id',local_id)
  75             return local_id
  76
  77
  78     class XrefTable:
  79
  80         def __init__ (self, api, tablename, class1, class2):
  81             self.api = api
  82             self.tablename = tablename
  83             self.lowerclass1 = class1.lower()
  84             self.lowerclass2 = class2.lower()
  85
  86         def delete_old_items (self, id1, id2_set):
  87             if id2_set:
  88                 sql = ""
  89                 sql += "DELETE FROM %s WHERE %s_id=%d"%(self.tablename,self.lowerclass1,id1)
  90                 sql += " AND %s_id IN ("%self.lowerclass2
  91                 sql += ",".join([str(i) for i in id2_set])
  92                 sql += ")"
  93                 self.api.db.do (sql)
  94
  95         def insert_new_items (self, id1, id2_set):
  96         ### xxx needs to be optimized
  97         ### tried to figure a way to use a single sql statement
  98         ### like: insert into table (x,y) values (1,2),(3,4);
  99         ### but apparently this is not supported under postgresql
 100             for id2 in id2_set:
 101                 sql = "INSERT INTO %s VALUES (%d,%d)"%(self.tablename,id1,id2)
 102                 self.api.db.do (sql)
 103
 104         def update_item (self, id1, old_id2s, new_id2s):
 105             news = set (new_id2s)
 106             olds = set (old_id2s)
 107             to_delete = olds-news
 108             self.delete_old_items (id1, to_delete)
 109             to_create = news-olds
 110             self.insert_new_items (id1, to_create)
 111             self.api.db.commit()
 112
 113     # classname: the type of objects we are talking about;       e.g. 'Slice'
 114     # peer_object_list list of objects at a given peer -         e.g. peer.GetSlices()
 115     # alien_xref_objs_dict : a dict {'classname':alien_obj_list} e.g. {'Node':peer.GetNodes()}
 116     # his must match the keys in xref_specs
 117     # lambda_ignore : the alien objects are ignored if this returns true
 118     def update_table (self,
 119                       classname,
 120                       alien_object_list,
 121                       alien_xref_objs_dict = {},
 122                       lambda_ignore=lambda x:False):
 123
 124         peer = self.peer
 125         peer_id = peer['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         accessories = dict(
 138             [ (xref_classname,
 139                {'transcoder':Cache.Transcoder (self.api,xref_classname,alien_xref_objs_dict[xref_classname]),
 140                 'xref_table':Cache.XrefTable (self.api,xref_spec['table'],classname,xref_classname)})
 141               for xref_classname,xref_spec in foreign_xrefs.iteritems()])
 142
 143         ### get current local table
 144         # get ALL local objects so as to cope with
 145         # (*) potential moves between plcs
 146         # (*) or naming conflicts
 147         local_objects = table_class (self.api)
 148         ### index upon class_key for future searches
 149         #verbose ('local objects:',local_objects)
 150         verbose ('class_key',class_key)
 151         local_objects_index = local_objects.dict(class_key)
 152         verbose ('update_table',classname,local_objects_index.keys())
 153
 154         ### mark entries for this peer outofdate
 155         old_count=0;
 156         for local_object in local_objects:
 157             if local_object['peer_id'] == peer_id:
 158                 local_object.uptodate=False
 159                 old_count += 1
 160             else:
 161                 local_object.uptodate=True
 162
 163         new_count=0
 164         # scan the peer's local objects
 165         for alien_object in alien_object_list:
 166
 167             object_name = alien_object[class_key]
 168
 169             ### ignore, e.g. system-wide slices
 170             if lambda_ignore(alien_object):
 171                 verbose('Ignoring',object_name)
 172                 continue
 173
 174             verbose ('update_table - Considering',object_name)
 175
 176             # create or update
 177             try:
 178                 ### We know about this object already
 179                 local_object = local_objects_index[object_name]
 180                 if local_object ['peer_id'] is None:
 181                     ### xxx send e-mail
 182                     print '==================== We are in trouble here'
 183                     print 'The %s object named %s is natively defined twice'%(classname,object_name)
 184                     print 'Once on this PLC and once on peer %d'%peer_id
 185                     print 'We dont raise an exception so that the remaining updates can still take place'
 186                     continue
 187                 if local_object['peer_id'] != peer_id:
 188                     ### the object has changed its plc,
 189                     ### Note, this is not problematic here because both definitions are remote
 190                     ### we can assume the object just moved
 191                     ### needs to update peer_id though
 192                     local_object['peer_id'] = peer_id
 193                 verbose ('update_table FOUND',object_name)
 194             except:
 195                 ### create a new entry
 196                 local_object = row_class(self.api,
 197                                           {class_key :object_name,'peer_id':peer_id})
 198                 # insert in index
 199                 local_objects_index[class_key]=local_object
 200                 verbose ('update_table CREATED',object_name)
 201
 202             # go on with update
 203             for field in foreign_fields:
 204                 local_object[field]=alien_object[field]
 205
 206             # this row is now valid
 207             local_object.uptodate=True
 208             new_count += 1
 209             local_object.sync()
 210
 211             # manage cross-refs
 212             for xref_classname,xref_spec in foreign_xrefs.iteritems():
 213                 field=xref_spec['field']
 214                 alien_xref_obj_list = alien_xref_objs_dict[xref_classname]
 215                 alien_value = alien_object[field]
 216                 transcoder = accessories[xref_classname]['transcoder']
 217                 if isinstance (alien_value,list):
 218                     verbose ('update_table list-transcoding ',xref_classname,' aliens=',alien_value,)
 219                     local_values=[]
 220                     for a in alien_value:
 221                         try:
 222                             local_values.append(transcoder.transcode(a))
 223                         except:
 224                             # could not transcode - might be from another peer that we dont know about..
 225                             pass
 226                     verbose (" transcoded as ",local_values)
 227                     xref_table = accessories[xref_classname]['xref_table']
 228                     # newly created objects dont have xrefs yet
 229                     try:
 230                         former_xrefs=local_object[xref_spec['field']]
 231                     except:
 232                         former_xrefs=[]
 233                     xref_table.update_item (local_object[primary_key],
 234                                             former_xrefs,
 235                                             local_values)
 236                 elif isinstance (alien_value,int):
 237                     verbose ('update_table atom-transcoding ',xref_classname,' aliens=',alien_value,)
 238                     new_value = transcoder.transcode(alien_value)
 239                     local_object[field] = new_value
 240                     local_object.sync()
 241
 242         ### delete entries that are not uptodate
 243         for local_object in local_objects:
 244             if not local_object.uptodate:
 245                 local_object.delete()
 246
 247         self.api.db.commit()
 248
 249         ### return delta in number of objects
 250         return new_count-old_count
 251
 252     def get_locals (self, list):
 253         return [x for x in list if x['peer_id'] is None]
 254
 255     def refresh_peer (self):
 256
 257         # so as to minimize the numer of requests
 258         # we get all objects in a single call and sort afterwards
 259         # xxx ideally get objects either local or the ones attached here
 260         # requires to know remote peer's peer_id for ourselves, mmhh..
 261         # does not make any difference in a 2-peer deployment though
 262
 263         # refresh sites
 264         all_sites = self.peer_server.GetSites(self.auth)
 265         local_sites = self.get_locals (all_sites)
 266         nb_new_sites = self.update_table('Site', local_sites)
 267
 268         # refresh keys
 269         all_keys = self.peer_server.GetKeys(self.auth)
 270         local_keys = self.get_locals (all_keys)
 271         nb_new_keys = self.update_table('Key', local_keys)
 272
 273         # refresh nodes
 274         all_nodes = self.peer_server.GetNodes(self.auth)
 275         local_nodes = self.get_locals(all_nodes)
 276         nb_new_nodes = self.update_table('Node', local_nodes,
 277                                          { 'Site' : all_sites } )
 278
 279         # refresh persons
 280         all_persons = self.peer_server.GetPersons(self.auth)
 281         local_persons = self.get_locals(all_persons)
 282         nb_new_persons = self.update_table ('Person', local_persons,
 283                                             { 'Key': all_keys, 'Site' : all_sites } )
 284
 285         # refresh slices
 286         local_slices = self.peer_server.GetSlices(self.auth,{'peer_id':None})
 287
 288         def is_system_slice (slice):
 289             return slice['creator_person_id'] == 1
 290
 291         nb_new_slices = self.update_table ('Slice', local_slices,
 292                                            {'Node': all_nodes, 'Person': all_persons},
 293                                            is_system_slice)
 294
 295         return {'plcname':self.api.config.PLC_NAME,
 296                 'new_sites':nb_new_sites,
 297                 'new_keys':nb_new_keys,
 298                 'new_nodes':nb_new_nodes,
 299                 'new_persons':nb_new_persons,
 300                 'new_slices':nb_new_slices}
 301