database.py

   1 # $Id$
   2 # $URL$
   3
   4 """The database houses information on slivers.  This information
   5 reaches the sliver manager in two different ways: one, through the
   6 GetSlivers() call made periodically; two, by users delivering tickets.
   7 The sync() method of the Database class turns this data into reality.
   8
   9 The database format is a dictionary that maps account names to records
  10 (also known as dictionaries).  Inside each record, data supplied or
  11 computed locally is stored under keys that begin with an underscore,
  12 and data from PLC is stored under keys that don't.
  13
  14 In order to maintain service when the node reboots during a network
  15 partition, the database is constantly being dumped to disk.
  16 """
  17
  18 import cPickle
  19 import threading
  20 import time
  21
  22 import accounts
  23 import logger
  24 import tools
  25 import bwmon
  26
  27 # We enforce minimum allocations to keep the clueless from hosing their slivers.
  28 # Disallow disk loans because there's currently no way to punish slivers over quota.
  29 MINIMUM_ALLOCATION = {'cpu_pct': 0, 'cpu_share': 1, 'net_min_rate': 0, 'net_max_rate': 8, 'net_i2_min_rate': 0, 'net_i2_max_rate': 8, 'net_share': 1}
  30 LOANABLE_RESOURCES = MINIMUM_ALLOCATION.keys()
  31
  32 DB_FILE = '/root/sliver_mgr_db.pickle'
  33
  34
  35 # database object and associated lock
  36 db_lock = threading.RLock()
  37 db = None
  38
  39 # these are used in tandem to request a database dump from the dumper daemon
  40 db_cond = threading.Condition(db_lock)
  41 dump_requested = False
  42
  43 # decorator that acquires and releases the database lock before and after the decorated operation
  44 # XXX - replace with "with" statements once we switch to 2.5
  45 def synchronized(fn):
  46     def sync_fn(*args, **kw_args):
  47         db_lock.acquire()
  48         try: return fn(*args, **kw_args)
  49         finally: db_lock.release()
  50     sync_fn.__doc__ = fn.__doc__
  51     sync_fn.__name__ = fn.__name__
  52     return sync_fn
  53
  54
  55 class Database(dict):
  56     def __init__(self):
  57         self._min_timestamp = 0
  58
  59     def _compute_effective_rspecs(self):
  60         """Calculate the effects of loans and store the result in field _rspec.  At the moment, we allow slivers to loan only those resources that they have received directly from PLC.  In order to do the accounting, we store three different rspecs: field 'rspec', which is the resources given by PLC; field '_rspec', which is the actual amount of resources the sliver has after all loans; and variable resid_rspec, which is the amount of resources the sliver has after giving out loans but not receiving any."""
  61         slivers = {}
  62         for name, rec in self.iteritems():
  63             if 'rspec' in rec:
  64                 rec['_rspec'] = rec['rspec'].copy()
  65                 slivers[name] = rec
  66         for rec in slivers.itervalues():
  67             eff_rspec = rec['_rspec']
  68             resid_rspec = rec['rspec'].copy()
  69             for target, resname, amt in rec.get('_loans', []):
  70                 if target in slivers and amt <= resid_rspec[resname] - MINIMUM_ALLOCATION[resname]:
  71                     eff_rspec[resname] -= amt
  72                     resid_rspec[resname] -= amt
  73                     slivers[target]['_rspec'][resname] += amt
  74
  75     def deliver_record(self, rec):
  76         """A record is simply a dictionary with 'name' and 'timestamp' keys.  We keep some persistent private data in the records under keys that start with '_'; thus record updates should not displace such keys."""
  77         if rec['timestamp'] < self._min_timestamp: return
  78         name = rec['name']
  79         old_rec = self.get(name)
  80         if old_rec == None: self[name] = rec
  81         elif rec['timestamp'] > old_rec['timestamp']:
  82             for key in old_rec.keys():
  83                 if not key.startswith('_'): del old_rec[key]
  84             old_rec.update(rec)
  85
  86     def set_min_timestamp(self, ts):
  87         """The ._min_timestamp member is the timestamp on the last comprehensive update.  We use it to determine if a record is stale.  This method should be called whenever new GetSlivers() data comes in."""
  88         self._min_timestamp = ts
  89         for name, rec in self.items():
  90             if rec['timestamp'] < ts: del self[name]
  91
  92     def sync(self):
  93         """Synchronize reality with the database contents.  This method does a lot of things, and it's currently called after every single batch of database changes (a GetSlivers(), a loan, a record).  It may be necessary in the future to do something smarter."""
  94
  95         # delete expired records
  96         now = time.time()
  97         for name, rec in self.items():
  98             if rec.get('expires', now) < now: del self[name]
  99
 100         self._compute_effective_rspecs()
 101
 102         # create and destroy accounts as needed
 103         logger.verbose("database:sync : fetching accounts")
 104         existing_acct_names = accounts.all()
 105         for name in existing_acct_names:
 106             if name not in self:
 107                 logger.verbose("database:sync : ensure_destroy'ing %s"%name)
 108                 accounts.get(name).ensure_destroyed()
 109         for name, rec in self.iteritems():
 110             # protect this; if anything fails for a given sliver
 111             # we still need the other ones to be handled
 112             try:
 113                 sliver = accounts.get(name)
 114                 logger.verbose("database:sync : looping on %s (shell account class from pwd %s)" %(name,sliver._get_class()))
 115                 # Make sure we refresh accounts that are running
 116                 if rec['instantiation'] == 'plc-instantiated':
 117                     logger.verbose ("database.sync : ensure_create'ing 'instantiation' sliver %s"%name)
 118                     sliver.ensure_created(rec)
 119                 elif rec['instantiation'] == 'nm-controller':
 120                     logger.verbose ("database.sync : ensure_create'ing 'nm-controller' sliver %s"%name)
 121                     sliver.ensure_created(rec)
 122                 # Back door to ensure PLC overrides Ticket in delegation.
 123                 elif rec['instantiation'] == 'delegated' and sliver._get_class() != None:
 124                     # if the ticket has been delivered and the nm-contoroller started the slice
 125                     # update rspecs and keep them up to date.
 126                     if sliver.is_running():
 127                         logger.verbose ("database.sync : ensure_create'ing 'delegated' sliver %s"%name)
 128                         sliver.ensure_created(rec)
 129             except:
 130                 logger.log_exc("database.sync failed to handle sliver",name=name)
 131
 132         # Wake up bwmom to update limits.
 133         bwmon.lock.set()
 134         global dump_requested
 135         dump_requested = True
 136         db_cond.notify()
 137
 138
 139 def start():
 140     """The database dumper daemon.  When it starts up, it populates the database with the last dumped database.  It proceeds to handle dump requests forever."""
 141     def run():
 142         global dump_requested
 143         while True:
 144             db_lock.acquire()
 145             while not dump_requested: db_cond.wait()
 146             db_pickle = cPickle.dumps(db, cPickle.HIGHEST_PROTOCOL)
 147             dump_requested = False
 148             db_lock.release()
 149             try: tools.write_file(DB_FILE, lambda f: f.write(db_pickle))
 150             except: logger.log_exc("failed in database.start.run")
 151     global db
 152     try:
 153         f = open(DB_FILE)
 154         try: db = cPickle.load(f)
 155         finally: f.close()
 156     except:
 157         logger.log_exc("failed in database.start")
 158         db = Database()
 159     tools.as_daemon_thread(run)