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                 if isinstance (alien_value,list):
 217                     verbose ('update_table list-transcoding ',xref_classname,' aliens=',alien_value,)
 218                     transcoder = accessories[xref_classname]['transcoder']
 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                     new_value = transcoder.transcode(alien_value)
 238                     local_object[field] = new_value
 239                     local_object.sync()
 240
 241         ### delete entries that are not uptodate
 242         for local_object in local_objects:
 243             if not local_object.uptodate:
 244                 local_object.delete()
 245
 246         self.api.db.commit()
 247
 248         ### return delta in number of objects
 249         return new_count-old_count
 250
 251     def refresh_nodes (self, peer_get_nodes):
 252         """
 253         refreshes the foreign_nodes and peer_node tables
 254         expected input is the current list of local nodes
 255         as returned from the peer by GetNodes {'peer_id':None}
 256
 257         returns the number of new nodes (can be negative)
 258         """
 259
 260         return self.update_table ('Node', peer_get_nodes)
 261
 262     def refresh_slices (self, peer_get_slices, peer_foreign_nodes):
 263         """
 264         refreshes the foreign_slices and peer_slice tables
 265         expected input is the current list of slices as returned by GetSlices
 266
 267         returns the number of new slices on this peer (can be negative)
 268         """
 269
 270         # xxx use 'system' flag for finding system slices
 271         return self.update_table ('Slice', peer_get_slices,
 272                                   {'Node':peer_foreign_nodes},
 273                                   lambda x: x['creator_person_id']==1)
 274
 275     def refresh_peer (self):
 276
 277         peer_local_slices = self.peer_server.GetSlices(self.auth,{'peer_id':None})
 278
 279         # refresh keys
 280         peer_local_keys = self.peer_server.GetKeys(self.auth,{'peer_id':None})
 281         nb_new_keys = self.update_table('Key', peer_local_keys)
 282
 283         # refresh nodes
 284         peer_local_nodes = self.peer_server.GetNodes(self.auth,{'peer_id':None})
 285         nb_new_nodes = self.update_table('Node', peer_local_nodes)
 286
 287         # refresh persons
 288         peer_local_persons = self.peer_server.GetPersons(self.auth,{'peer_id':None})
 289         # xxx ideally get our own persons only
 290         # requires to know remote peer's peer_id for ourselves, mmhh
 291         peer_all_keys = peer_local_keys + self.peer_server.GetKeys(self.auth,{'~peer_id':None})
 292         nb_new_persons = self.update_table ('Person', peer_local_persons,
 293                                             { 'Key': peer_all_keys} )
 294
 295         # refresh slices
 296         def is_system_slice (slice):
 297             return slice['creator_person_id'] == 1
 298         # xxx would ideally get our own nodes only,
 299         peer_all_nodes = peer_local_nodes+self.peer_server.GetNodes(self.auth,{'~peer_id':None})
 300
 301         nb_new_slices = self.update_table ('Slice', peer_local_slices,
 302                                            {'Node':peer_all_nodes},
 303                                            is_system_slice)
 304
 305
 306         return {'plcname':self.api.config.PLC_NAME,
 307                 'new_keys':nb_new_keys,
 308                 'new_nodes':nb_new_nodes,
 309                 'new_persons':nb_new_persons,
 310                 'new_slices':nb_new_slices}
 311