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             ### ignore, e.g. system-wide slices
 168             if lambda_ignore(alien_object):
 169                 continue
 170
 171             object_name = alien_object[class_key]
 172             verbose ('update_table - Considering',object_name)
 173
 174             # create or update
 175             try:
 176                 ### We know about this object already
 177                 local_object = local_objects_index[object_name]
 178                 if local_object ['peer_id'] is None:
 179                     print 'We are in trouble here'
 180                     print 'The %s object named %s is natively defined twice'%(classname,object_name)
 181                     print 'Once on this PLC and once on peer %d'%peer_id
 182                     print 'We dont raise an exception so that the remaining updates can still take place'
 183                     continue
 184                 if local_object['peer_id'] != peer_id:
 185                     ### the object has changed its plc,
 186                     ### Note, this is not problematic here because both definitions are remote
 187                     ### we can assume the object just moved
 188                     ### needs to update peer_id though
 189                     local_object['peer_id'] = peer_id
 190                 verbose ('update_table FOUND',object_name)
 191             except:
 192                 ### create a new entry
 193                 local_object = row_class(self.api,
 194                                           {class_key :object_name,'peer_id':peer_id})
 195                 # insert in index
 196                 local_objects_index[class_key]=local_object
 197                 verbose ('update_table CREATED',object_name)
 198
 199             # go on with update
 200             for field in foreign_fields:
 201                 local_object[field]=alien_object[field]
 202
 203             # this row is now valid
 204             local_object.uptodate=True
 205             new_count += 1
 206             local_object.sync()
 207
 208             # manage cross-refs
 209             for xref_classname,xref_spec in foreign_xrefs.iteritems():
 210                 field=xref_spec['field']
 211                 alien_xref_obj_list = alien_xref_objs_dict[xref_classname]
 212                 alien_value = alien_object[field]
 213                 if isinstance (alien_value,list):
 214                     verbose ('update_table list-transcoding ',xref_classname,' aliens=',alien_value,)
 215                     transcoder = accessories[xref_classname]['transcoder']
 216                     local_values=[]
 217                     for a in alien_value:
 218                         try:
 219                             local_values.append(transcoder.transcode(a))
 220                         except:
 221                             # could not transcode - might be from another peer that we dont know about..
 222                             pass
 223                     verbose (" transcoded as ",local_values)
 224                     xref_table = accessories[xref_classname]['xref_table']
 225                     # newly created objects dont have xrefs yet
 226                     try:
 227                         former_xrefs=local_object[xref_spec['field']]
 228                     except:
 229                         former_xrefs=[]
 230                     xref_table.update_item (local_object[primary_key],
 231                                             former_xrefs,
 232                                             local_values)
 233                 elif isinstance (alien_value,int):
 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     def refresh_nodes (self, peer_get_nodes):
 249         """
 250         refreshes the foreign_nodes and peer_node tables
 251         expected input is the current list of local nodes
 252         as returned from the peer by GetNodes {'peer_id':None}
 253
 254         returns the number of new nodes (can be negative)
 255         """
 256
 257         return self.update_table ('Node', peer_get_nodes)
 258
 259     def refresh_slices (self, peer_get_slices, peer_foreign_nodes):
 260         """
 261         refreshes the foreign_slices and peer_slice tables
 262         expected input is the current list of slices as returned by GetSlices
 263
 264         returns the number of new slices on this peer (can be negative)
 265         """
 266
 267         # xxx use 'system' flag for finding system slices
 268         return self.update_table ('Slice', peer_get_slices,
 269                                   {'Node':peer_foreign_nodes},
 270                                   lambda x: x['creator_person_id']==1)
 271
 272     def refresh_peer (self):
 273
 274         peer_local_keys = self.peer_server.GetKeys(self.auth,{'peer_id':None})
 275         peer_local_nodes = self.peer_server.GetNodes(self.auth,None,None,'local')
 276         # xxx would ideally get our own nodes only,
 277         # requires to know remote peer's peer_id for ourselves, mmhh
 278         peer_foreign_nodes = self.peer_server.GetNodes(self.auth,None,None,'foreign')
 279         peer_local_slices = self.peer_server.GetSlices(self.auth,{'peer_id':None})
 280
 281         nb_new_keys = self.update_table('Key', peer_local_keys)
 282
 283         nb_new_nodes = self.update_table('Node', peer_local_nodes)
 284
 285         # rough and temporary
 286         nb_new_slices = self.refresh_slices(peer_local_slices,peer_local_nodes+peer_foreign_nodes)
 287
 288         return {'plcname':self.api.config.PLC_NAME,
 289                 'new_keys':nb_new_keys,
 290                 'new_nodes':nb_new_nodes,
 291                 'new_slices':nb_new_slices}
 292