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 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                 verbose ('indexing',x)
  55                 d[x[self.class_id]]=x
  56             self.alien_objects_byid = d
  57
  58             local_objects = attrs['table_class'] (api)
  59             self.local_objects_byname = local_objects.dict(self.class_key)
  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.class_id]
  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     # 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         class_id = attrs['class_id']
 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         accessories={}
 137         for xref_classname,xref_spec in foreign_xrefs.iteritems():
 138             d={}
 139             d['transcoder']=Cache.Transcoder (self.api,xref_classname,alien_xref_objs_dict[xref_classname])
 140             d['xref_table'] =Cache.XrefTable (self.api,xref_spec['table'],classname,xref_classname)
 141             accessories[xref_classname]=d
 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         local_objects_index = local_objects.dict(class_key)
 150         verbose ('update_table',classname,local_objects_index.keys())
 151
 152         ### mark entries for this peer outofdate
 153         old_count=0;
 154         for local_object in local_objects:
 155             if local_object['peer_id'] == peer_id:
 156                 local_object.uptodate=False
 157                 old_count += 1
 158             else:
 159                 local_object.uptodate=True
 160
 161         new_count=0
 162         # scan the peer's local objects
 163         for alien_object in alien_object_list:
 164
 165             ### ignore, e.g. system-wide slices
 166             if lambda_ignore(alien_object):
 167                 continue
 168
 169             object_name = alien_object[class_key]
 170             verbose ('update_table - Considering',object_name)
 171
 172             # create or update
 173             try:
 174                 ### We know about this object already
 175                 local_object = local_objects_index[object_name]
 176                 if local_object ['peer_id'] is None:
 177                     print 'We are in trouble here'
 178                     print 'The %s object named %s is natively defined twice'%(classname,object_name)
 179                     print 'Once on this PLC and once on peer %d'%peer_id
 180                     print 'We dont raise an exception so that the remaining updates can still take place'
 181                     continue
 182                 if local_object['peer_id'] != peer_id:
 183                     ### the object has changed its plc,
 184                     ### Note, this is not problematic here because both definitions are remote
 185                     ### we can assume the object just moved
 186                     ### needs to update peer_id though
 187                     local_object['peer_id'] = peer_id
 188                 verbose ('update_table FOUND',object_name)
 189             except:
 190                 ### create a new entry
 191                 local_object = row_class(self.api,
 192                                           {class_key :object_name,'peer_id':peer_id})
 193                 # insert in index
 194                 local_objects_index[class_key]=local_object
 195                 verbose ('update_table CREATED',object_name)
 196
 197             # go on with update
 198             for field in foreign_fields:
 199                 local_object[field]=alien_object[field]
 200
 201             # this row is now valid
 202             local_object.uptodate=True
 203             new_count += 1
 204             local_object.sync()
 205
 206             # manage cross-refs
 207             for xref_classname,xref_spec in foreign_xrefs.iteritems():
 208                 field=xref_spec['field']
 209                 alien_xref_obj_list = alien_xref_objs_dict[xref_classname]
 210                 alien_value = alien_object[field]
 211                 if isinstance (alien_value,list):
 212                     verbose ('update_table list-transcoding ',xref_classname,' aliens=',alien_value,)
 213                     transcoder = accessories[xref_classname]['transcoder']
 214                     local_values=[]
 215                     for a in alien_value:
 216                         try:
 217                             local_values.append(transcoder.transcode(a))
 218                         except:
 219                             # could not transcode - might be from another peer that we dont know about..
 220                             pass
 221                     verbose (" transcoded as ",local_values)
 222                     xref_table = accessories[xref_classname]['xref_table']
 223                     # newly created objects dont have xrefs yet
 224                     try:
 225                         former_xrefs=local_object[xref_spec['field']]
 226                     except:
 227                         former_xrefs=[]
 228                     xref_table.update_item (local_object[class_id],
 229                                             former_xrefs,
 230                                             local_values)
 231                 elif isinstance (alien_value,int):
 232                     new_value = transcoder.transcode(alien_value)
 233                     local_object[field] = new_value
 234                     local_object.sync()
 235
 236         ### delete entries that are not uptodate
 237         for local_object in local_objects:
 238             if not local_object.uptodate:
 239                 local_object.delete()
 240
 241         self.api.db.commit()
 242
 243         ### return delta in number of objects
 244         return new_count-old_count
 245
 246     def refresh_nodes (self, peer_get_nodes):
 247         """
 248         refreshes the foreign_nodes and peer_node tables
 249         expected input is the current list of local nodes
 250         as returned from the peer by GetNodes {'peer_id':None}
 251
 252         returns the number of new nodes (can be negative)
 253         """
 254
 255         return self.update_table ('Node', peer_get_nodes)
 256
 257     def refresh_slices (self, peer_get_slices, peer_foreign_nodes):
 258         """
 259         refreshes the foreign_slices and peer_slice tables
 260         expected input is the current list of slices as returned by GetSlices
 261
 262         returns the number of new slices on this peer (can be negative)
 263         """
 264
 265         # xxx use 'system' flag for finding system slices
 266         return self.update_table ('Slice', peer_get_slices,
 267                                   {'Node':peer_foreign_nodes},
 268                                   lambda x: x['creator_person_id']==1)
 269
 270     def refresh_peer (self):
 271
 272         peer_local_nodes = self.peer_server.GetNodes(self.auth,None,None,'local')
 273         peer_foreign_nodes = self.peer_server.GetNodes(self.auth,None,None,'foreign')
 274         peer_local_slices = self.peer_server.GetSlices(self.auth,{'peer_id':None})
 275
 276         from PLC.Nodes import Nodes
 277         verbose ('local nodes before refresh',len(Nodes(self.api)))
 278         nb_new_nodes = self.refresh_nodes(peer_local_nodes)
 279         verbose ('local nodes after refresh',len(Nodes(self.api)))
 280
 281         # rough and temporary
 282         nb_new_slices = self.refresh_slices(peer_local_slices,peer_local_nodes+peer_foreign_nodes)
 283
 284         return {'plcname':self.api.config.PLC_NAME,
 285                 'new_nodes':nb_new_nodes,
 286                 'new_slices':nb_new_slices}
 287