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=True;
   7 def verbose (*args):
   8     if verbose_flag:
   9         print (args)
  10
  11 def class_attributes (classname):
  12     """ locates various attributes defined in the row class """
  13     topmodule = __import__ ('PLC.%ss'%classname)
  14     module = topmodule.__dict__['%ss'%classname]
  15     # local row-like class, e.g. Node
  16     row_class = module.__dict__['%s'%classname]
  17     # local tab-like class, e.g. Nodes
  18     table_class = module.__dict__['%ss'%classname]
  19
  20     return {'row_class':row_class,
  21             'table_class':table_class,
  22             'class_id': row_class.__dict__['class_id'],
  23             'class_key': row_class.__dict__['class_key'],
  24             'foreign_fields': row_class.__dict__['foreign_fields'],
  25             'foreign_xrefs': row_class.__dict__['foreign_xrefs'],
  26             }
  27
  28 class Cache:
  29
  30     # an attempt to provide genericity in the caching algorithm
  31
  32     # the Peer object we are syncing with
  33     def __init__ (self, api, peer, peer_server, auth):
  34
  35         import PLC.Peers
  36
  37         self.api = api
  38         assert isinstance(peer,PLC.Peers.Peer)
  39         self.peer = peer
  40         self.peer_server = peer_server
  41         self.auth = auth
  42
  43     class Transcoder:
  44
  45         def __init__ (self, api, classname, alien_objects):
  46             self.api = api
  47             attrs = class_attributes (classname)
  48             self.class_id = attrs['class_id']
  49             self.class_key = attrs['class_key']
  50
  51             # cannot use dict, it's acquired by xmlrpc and is untyped
  52             d = {}
  53             for x in alien_objects:
  54                 d[x[self.class_id]]=x
  55             self.alien_objects_byid = d
  56
  57             local_objects = attrs['table_class'] (api)
  58             self.local_objects_byname = local_objects.dict(self.class_key)
  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.class_id]
  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     # lambda_ignore : the alien objects are ignored if this returns true
 117     def update_table (self,
 118                       classname,
 119                       alien_object_list,
 120                       alien_xref_objs_dict = {},
 121                       lambda_ignore=lambda x:False):
 122
 123         peer = self.peer
 124         peer_id = peer['peer_id']
 125
 126         attrs = class_attributes (classname)
 127         row_class = attrs['row_class']
 128         table_class = attrs['table_class']
 129         class_id = attrs['class_id']
 130         class_key = attrs['class_key']
 131         foreign_fields = attrs['foreign_fields']
 132         foreign_xrefs = attrs['foreign_xrefs']
 133
 134
 135         ### get current local table
 136         # get ALL local objects so as to cope with
 137         # (*) potential moves between plcs
 138         # (*) or naming conflicts
 139         local_objects = table_class (self.api)
 140         ### index upon class_key for future searches
 141         local_objects_index = local_objects.dict(class_key)
 142         verbose ('update_table',classname,local_objects_index.keys())
 143
 144         ### mark entries for this peer outofdate
 145         old_count=0;
 146         for local_object in local_objects:
 147             if local_object['peer_id'] == peer_id:
 148                 local_object.uptodate=False
 149                 old_count += 1
 150             else:
 151                 local_object.uptodate=True
 152
 153         new_count=0
 154         # scan the peer's local objects
 155         for alien_object in alien_object_list:
 156
 157             ### ignore system-wide slices
 158             if lambda_ignore(alien_object):
 159                 continue
 160
 161             object_name = alien_object[class_key]
 162             verbose ('update_table - Considering',object_name)
 163
 164             # create or update
 165             try:
 166                 ### We know about this object already
 167                 local_object = local_objects_index[object_name]
 168                 if local_object ['peer_id'] is None:
 169                     print 'We are in trouble here'
 170                     print 'The %s object named %s is natively defined twice'%(classname,object_name)
 171                     print 'Once on this PLC and once on peer %d'%peer_id
 172                     print 'We dont raise an exception so that the remaining updates can still take place'
 173                     continue
 174                 if local_object['peer_id'] != peer_id:
 175                     ### the object has changed its plc,
 176                     ### Note, this is not problematic here because both definitions are remote
 177                     ### we can assume the object just moved
 178                     ### needs to update peer_id though
 179                     local_object['peer_id'] = peer_id
 180                 verbose ('update_table FOUND',object_name)
 181             except:
 182                 ### create a new entry
 183                 local_object = row_class(self.api,
 184                                           {class_key :object_name,'peer_id':peer_id})
 185                 # insert in index
 186                 local_objects_index[class_key]=local_object
 187                 verbose ('update_table CREATED',object_name)
 188
 189             # go on with update
 190             for field in foreign_fields:
 191                 local_object[field]=alien_object[field]
 192
 193             # this row is now valid
 194             local_object.uptodate=True
 195             new_count += 1
 196             local_object.sync()
 197
 198             # manage cross-refs
 199             for related_class,xrefspec in foreign_xrefs.iteritems():
 200                 field=xrefspec['field']
 201                 alien_xref_obj_list = alien_xref_objs_dict[related_class]
 202                 alien_value = alien_object[field]
 203                 ### yyy optimize objects allocations for transcoders and xreftables
 204                 if isinstance (alien_value,list):
 205                     verbose ('update_table list-transcoding ',related_class,' aliens=',alien_value,)
 206                     transcoder = Cache.Transcoder(self.api,related_class,alien_xref_obj_list)
 207                     local_values=[]
 208                     for a in alien_value:
 209                         try:
 210                             local_values.append(transcoder.transcode(a))
 211                         except:
 212                             # could not transcode - might be from another peer that we dont know about..
 213                             pass
 214                     verbose (" trasncoded as ",local_values)
 215                     xref_table = Cache.XrefTable (self.api,xrefspec['table'],classname,related_class)
 216                     # newly created objects dont have xrefs yet
 217                     try:
 218                         former_xrefs=local_object[xrefspec['field']]
 219                     except:
 220                         former_xrefs=[]
 221                     xref_table.update_item (local_object[class_id],
 222                                             former_xrefs,
 223                                             local_values)
 224                 elif isinstance (alien_value,int):
 225                     new_value = transcoder.transcode(alien_value)
 226                     local_object[field] = new_value
 227                     local_object.sync()
 228
 229         ### delete entries that are not uptodate
 230         for local_object in local_objects:
 231             if not local_object.uptodate:
 232                 local_object.delete()
 233
 234         self.api.db.commit()
 235
 236         ### return delta in number of objects
 237         return new_count-old_count
 238
 239     def refresh_nodes (self, peer_get_nodes):
 240         """
 241         refreshes the foreign_nodes and peer_node tables
 242         expected input is the current list of local nodes
 243         as returned from the peer by GetNodes {'peer_id':None}
 244
 245         returns the number of new nodes (can be negative)
 246         """
 247
 248         return self.update_table ('Node', peer_get_nodes)
 249
 250     def refresh_slices (self, peer_get_slices, peer_foreign_nodes):
 251         """
 252         refreshes the foreign_slices and peer_slice tables
 253         expected input is the current list of slices as returned by GetSlices
 254
 255         returns the number of new slices on this peer (can be negative)
 256         """
 257
 258         # xxx use 'system' flag for slices
 259         return self.update_table ('Slice', peer_get_slices,
 260                                   {'Node':peer_foreign_nodes},
 261                                   lambda x: x['creator_person_id']==1)
 262
 263     def refresh_peer (self):
 264
 265         peer_local_nodes = self.peer_server.GetNodes(self.auth,None,None,'local')
 266         peer_foreign_nodes = self.peer_server.GetNodes(self.auth,None,None,'foreign')
 267         peer_local_slices = self.peer_server.GetSlices(self.auth,{'peer_id':None})
 268
 269         from PLC.Nodes import Nodes
 270         verbose ('local nodes before refresh',len(Nodes(self.api)))
 271         nb_new_nodes = self.refresh_nodes(peer_local_nodes)
 272         verbose ('local nodes after refresh',len(Nodes(self.api)))
 273
 274         # rough and temporary
 275         nb_new_slices = self.refresh_slices(peer_local_slices,peer_local_nodes+peer_foreign_nodes)
 276
 277         return {'plcname':self.api.config.PLC_NAME,
 278                 'new_nodes':nb_new_nodes,
 279                 'new_slices':nb_new_slices}
 280