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