bwmon.py

   1 #!/usr/bin/python
   2 #
   3 # Average bandwidth monitoring script. Run periodically via NM db.sync to
   4 # enforce a soft limit on daily bandwidth usage for each slice. If a
   5 # slice is found to have transmitted 80% of its daily byte limit usage,
   6 # its instantaneous rate will be capped at the bytes remaning in the limit
   7 # over the time remaining in the recording period.
   8 #
   9 # Two separate limits are enforced, one for destinations exempt from
  10 # the node bandwidth cap (i.e. Internet2), and the other for all other destinations.
  11 #
  12 # Mark Huang <mlhuang@cs.princeton.edu>
  13 # Andy Bavier <acb@cs.princeton.edu>
  14 # Faiyaz Ahmed <faiyaza@cs.princeton.edu>
  15 # Copyright (C) 2004-2008 The Trustees of Princeton University
  16 #
  17 # $Id$
  18 #
  19
  20 import os
  21 import sys
  22 import time
  23 import pickle
  24 import socket
  25 import logger
  26 import copy
  27 import threading
  28 import tools
  29
  30 import bwlimit
  31 import database
  32
  33 from sets import Set
  34
  35 # Defaults
  36 debug = False
  37 verbose = False
  38 datafile = "/var/lib/misc/bwmon.dat"
  39
  40 try:
  41     sys.path.append("/etc/planetlab")
  42     from plc_config import *
  43 except:
  44     logger.log("bwmon:  Warning: Configuration file /etc/planetlab/plc_config.py not found")
  45     logger.log("bwmon:  Running in DEBUG mode.  Logging to file and not emailing.")
  46
  47 # Constants
  48 seconds_per_day = 24 * 60 * 60
  49 bits_per_byte = 8
  50
  51 # Burst to line rate (or node cap).  Set by NM. in KBit/s
  52 default_MaxRate = int(bwlimit.get_bwcap() / 1000)
  53 default_Maxi2Rate = int(bwlimit.bwmax / 1000)
  54 # Min rate 8 bits/s
  55 default_MinRate = 0
  56 default_Mini2Rate = 0
  57 # 5.4 Gbyte per day. 5.4 * 1024 k * 1024M * 1024G
  58 # 5.4 Gbyte per day max allowed transfered per recording period
  59 default_MaxKByte = 5662310
  60 # 16.4 Gbyte per day max allowed transfered per recording period to I2
  61 default_Maxi2KByte = 17196646
  62 # Default share quanta
  63 default_Share = 1
  64
  65 # Average over 1 day
  66 period = 1 * seconds_per_day
  67
  68 # Message template
  69 template = \
  70 """
  71 The slice %(slice)s has transmitted more than %(bytes)s from
  72 %(hostname)s to %(class)s destinations
  73 since %(since)s.
  74
  75 Its maximum %(class)s burst rate will be capped at %(new_maxrate)s/s
  76 until %(until)s.
  77
  78 Please reduce the average %(class)s transmission rate
  79 of the slice to %(limit)s per %(period)s.
  80
  81 """.lstrip()
  82
  83 footer = \
  84 """
  85 %(date)s %(hostname)s bwcap %(slice)s
  86 """.lstrip()
  87
  88 def format_bytes(bytes, si = True):
  89     """
  90     Formats bytes into a string
  91     """
  92     if si:
  93         kilo = 1000.
  94     else:
  95         # Officially, a kibibyte
  96         kilo = 1024.
  97
  98     if bytes >= (kilo * kilo * kilo):
  99         return "%.1f GB" % (bytes / (kilo * kilo * kilo))
 100     elif bytes >= 1000000:
 101         return "%.1f MB" % (bytes / (kilo * kilo))
 102     elif bytes >= 1000:
 103         return "%.1f KB" % (bytes / kilo)
 104     else:
 105         return "%.0f bytes" % bytes
 106
 107 def format_period(seconds):
 108     """
 109     Formats a period in seconds into a string
 110     """
 111
 112     if seconds == (24 * 60 * 60):
 113         return "day"
 114     elif seconds == (60 * 60):
 115         return "hour"
 116     elif seconds > (24 * 60 * 60):
 117         return "%.1f days" % (seconds / 24. / 60. / 60.)
 118     elif seconds > (60 * 60):
 119         return "%.1f hours" % (seconds / 60. / 60.)
 120     elif seconds > (60):
 121         return "%.1f minutes" % (seconds / 60.)
 122     else:
 123         return "%.0f seconds" % seconds
 124
 125 def slicemail(slice, subject, body):
 126     '''
 127     Front end to sendmail.  Sends email to slice alias with given subject and body.
 128     '''
 129
 130     sendmail = os.popen("/usr/sbin/sendmail -N never -t -f%s" % PLC_MAIL_SUPPORT_ADDRESS, "w")
 131
 132     # Parsed from MyPLC config
 133     to = [PLC_MAIL_MOM_LIST_ADDRESS]
 134
 135     if slice is not None and slice != "root":
 136         to.append(PLC_MAIL_SLICE_ADDRESS.replace("SLICE", slice))
 137
 138     header = {'from': "%s Support <%s>" % (PLC_NAME, PLC_MAIL_SUPPORT_ADDRESS),
 139               'to': ", ".join(to),
 140               'version': sys.version.split(" ")[0],
 141               'subject': subject}
 142
 143     # Write headers
 144     sendmail.write(
 145 """
 146 Content-type: text/plain
 147 From: %(from)s
 148 Reply-To: %(from)s
 149 To: %(to)s
 150 X-Mailer: Python/%(version)s
 151 Subject: %(subject)s
 152
 153 """.lstrip() % header)
 154
 155     # Write body
 156     sendmail.write(body)
 157     # Done
 158     sendmail.close()
 159
 160
 161 class Slice:
 162     """
 163     Stores the last recorded bandwidth parameters of a slice.
 164
 165     xid - slice context/VServer ID
 166     name - slice name
 167     time - beginning of recording period in UNIX seconds
 168     bytes - low bandwidth bytes transmitted at the beginning of the recording period
 169     i2bytes - high bandwidth bytes transmitted at the beginning of the recording period (for I2 -F)
 170     MaxKByte - total volume of data allowed
 171     ThreshKbyte - After thresh, cap node to (maxkbyte - bytes)/(time left in period)
 172     Maxi2KByte - same as MaxKByte, but for i2
 173     Threshi2Kbyte - same as Threshi2KByte, but for i2
 174     MaxRate - max_rate slice attribute.
 175     Maxi2Rate - max_exempt_rate slice attribute.
 176     Share - Used by Sirius to loan min rates
 177     Sharei2 - Used by Sirius to loan min rates for i2
 178     self.emailed - did slice recv email during this recording period
 179
 180     """
 181
 182     def __init__(self, xid, name, rspec):
 183         self.xid = xid
 184         self.name = name
 185         self.time = 0
 186         self.bytes = 0
 187         self.i2bytes = 0
 188         self.MaxRate = default_MaxRate
 189         self.MinRate = default_MinRate
 190         self.Maxi2Rate = default_Maxi2Rate
 191         self.Mini2Rate = default_Mini2Rate
 192         self.MaxKByte = default_MaxKByte
 193         self.ThreshKByte = (.8 * self.MaxKByte)
 194         self.Maxi2KByte = default_Maxi2KByte
 195         self.Threshi2KByte = (.8 * self.Maxi2KByte)
 196         self.Share = default_Share
 197         self.Sharei2 = default_Share
 198         self.emailed = False
 199         self.capped = False
 200
 201         self.updateSliceAttributes(rspec)
 202         bwlimit.set(xid = self.xid,
 203                 minrate = self.MinRate * 1000,
 204                 maxrate = self.MaxRate * 1000,
 205                 maxexemptrate = self.Maxi2Rate * 1000,
 206                 minexemptrate = self.Mini2Rate * 1000,
 207                 share = self.Share)
 208
 209     def __repr__(self):
 210         return self.name
 211
 212     def updateSliceAttributes(self, rspec):
 213         '''
 214         Use respects from GetSlivers to PLC to populate slice object.  Also
 215         do some sanity checking.
 216         '''
 217
 218         # Sanity check plus policy decision for MinRate:
 219         # Minrate cant be greater than 25% of MaxRate or NodeCap.
 220         MinRate = int(rspec.get("net_min_rate", default_MinRate))
 221         if MinRate > int(.25 * default_MaxRate):
 222             MinRate = int(.25 * default_MaxRate)
 223         if MinRate != self.MinRate:
 224             self.MinRate = MinRate
 225             logger.log("bwmon:  Updating %s: Min Rate = %s" %(self.name, self.MinRate))
 226
 227         MaxRate = int(rspec.get('net_max_rate', bwlimit.get_bwcap() / 1000))
 228         if MaxRate != self.MaxRate:
 229             self.MaxRate = MaxRate
 230             logger.log("bwmon:  Updating %s: Max Rate = %s" %(self.name, self.MaxRate))
 231
 232         Mini2Rate = int(rspec.get('net_i2_min_rate', default_Mini2Rate))
 233         if Mini2Rate != self.Mini2Rate:
 234             self.Mini2Rate = Mini2Rate
 235             logger.log("bwmon:  Updating %s: Min i2 Rate = %s" %(self.name, self.Mini2Rate))
 236
 237         Maxi2Rate = int(rspec.get('net_i2_max_rate', bwlimit.bwmax / 1000))
 238         if Maxi2Rate != self.Maxi2Rate:
 239             self.Maxi2Rate = Maxi2Rate
 240             logger.log("bwmon:  Updating %s: Max i2 Rate = %s" %(self.name, self.Maxi2Rate))
 241
 242         MaxKByte = int(rspec.get('net_max_kbyte', default_MaxKByte))
 243         if MaxKByte != self.MaxKByte:
 244             self.MaxKByte = MaxKByte
 245             logger.log("bwmon:  Updating %s: Max KByte lim = %s" %(self.name, self.MaxKByte))
 246
 247         Maxi2KByte = int(rspec.get('net_i2_max_kbyte', default_Maxi2KByte))
 248         if Maxi2KByte != self.Maxi2KByte:
 249             self.Maxi2KByte = Maxi2KByte
 250             logger.log("bwmon:  Updating %s: Max i2 KByte = %s" %(self.name, self.Maxi2KByte))
 251
 252         ThreshKByte = int(rspec.get('net_thresh_kbyte', (MaxKByte * .8)))
 253         if ThreshKByte != self.ThreshKByte:
 254             self.ThreshKByte = ThreshKByte
 255             logger.log("bwmon:  Updating %s: Thresh KByte = %s" %(self.name, self.ThreshKByte))
 256
 257         Threshi2KByte = int(rspec.get('net_i2_thresh_kbyte', (Maxi2KByte * .8)))
 258         if Threshi2KByte != self.Threshi2KByte:
 259             self.Threshi2KByte = Threshi2KByte
 260             logger.log("bwmon:  Updating %s: i2 Thresh KByte = %s" %(self.name, self.Threshi2KByte))
 261
 262         Share = int(rspec.get('net_share', default_Share))
 263         if Share != self.Share:
 264             self.Share = Share
 265             logger.log("bwmon:  Updating %s: Net Share = %s" %(self.name, self.Share))
 266
 267         Sharei2 = int(rspec.get('net_i2_share', default_Share))
 268         if Sharei2 != self.Sharei2:
 269             self.Sharei2 = Sharei2
 270             logger.log("bwmon:  Updating %s: Net i2 Share = %s" %(self.name, self.i2Share))
 271
 272
 273     def reset(self, runningmaxrate, runningmaxi2rate, usedbytes, usedi2bytes, rspec):
 274         """
 275         Begin a new recording period. Remove caps by restoring limits
 276         to their default values.
 277         """
 278
 279         # Query Node Manager for max rate overrides
 280         self.updateSliceAttributes(rspec)
 281
 282         # Reset baseline time
 283         self.time = time.time()
 284
 285         # Reset baseline byte coutns
 286         self.bytes = usedbytes
 287         self.i2bytes = usedi2bytes
 288
 289         # Reset email
 290         self.emailed = False
 291         # Reset flag
 292         self.capped = False
 293         # Reset rates.
 294         maxrate = self.MaxRate * 1000
 295         maxi2rate = self.Maxi2Rate * 1000
 296         if (self.MaxRate != runningmaxrate) or (self.Maxi2Rate != runningmaxi2rate):
 297             logger.log("bwmon:  %s reset to %s/%s" % \
 298                   (self.name,
 299                    bwlimit.format_tc_rate(maxrate),
 300                    bwlimit.format_tc_rate(maxi2rate)))
 301             bwlimit.set(xid = self.xid,
 302                 minrate = self.MinRate * 1000,
 303                 maxrate = self.MaxRate * 1000,
 304                 maxexemptrate = self.Maxi2Rate * 1000,
 305                 minexemptrate = self.Mini2Rate * 1000,
 306                 share = self.Share)
 307
 308     def notify(self, new_maxrate, new_maxexemptrate, usedbytes, usedi2bytes):
 309         """
 310         Notify the slice it's being capped.
 311         """
 312          # Prepare message parameters from the template
 313         message = ""
 314         params = {'slice': self.name, 'hostname': socket.gethostname(),
 315                   'since': time.asctime(time.gmtime(self.time)) + " GMT",
 316                   'until': time.asctime(time.gmtime(self.time + period)) + " GMT",
 317                   'date': time.asctime(time.gmtime()) + " GMT",
 318                   'period': format_period(period)}
 319
 320         if new_maxrate != self.MaxRate:
 321             # Format template parameters for low bandwidth message
 322             params['class'] = "low bandwidth"
 323             params['bytes'] = format_bytes(usedbytes - self.bytes)
 324             params['limit'] = format_bytes(self.MaxKByte * 1024)
 325             params['new_maxrate'] = bwlimit.format_tc_rate(new_maxrate)
 326
 327             # Cap low bandwidth burst rate
 328             message += template % params
 329             logger.log("bwmon:   ** %(slice)s %(class)s capped at %(new_maxrate)s/s " % params)
 330
 331         if new_maxexemptrate != self.Maxi2Rate:
 332             # Format template parameters for high bandwidth message
 333             params['class'] = "high bandwidth"
 334             params['bytes'] = format_bytes(usedi2bytes - self.i2bytes)
 335             params['limit'] = format_bytes(self.Maxi2KByte * 1024)
 336             params['new_maxexemptrate'] = bwlimit.format_tc_rate(new_maxi2rate)
 337
 338             message += template % params
 339             logger.log("bwmon:   ** %(slice)s %(class)s capped at %(new_maxrate)s/s " % params)
 340
 341         # Notify slice
 342         if message and self.emailed == False:
 343             subject = "pl_mom capped bandwidth of slice %(slice)s on %(hostname)s" % params
 344             if debug:
 345                 logger.log("bwmon:  "+ subject)
 346                 logger.log("bwmon:  "+ message + (footer % params))
 347             else:
 348                 self.emailed = True
 349                 slicemail(self.name, subject, message + (footer % params))
 350
 351
 352     def update(self, runningmaxrate, runningmaxi2rate, usedbytes, usedi2bytes, runningshare, rspec):
 353         """
 354         Update byte counts and check if byte thresholds have been
 355         exceeded. If exceeded, cap to  remaining bytes in limit over remaining in period.
 356         Recalculate every time module runs.
 357         """
 358
 359         # Query Node Manager for max rate overrides
 360         self.updateSliceAttributes(rspec)
 361
 362         # Check shares for Sirius loans.
 363         if runningshare != self.share:
 364             logger.log("bwmon:  Updating share to %s" % self.share)
 365             bwlimit.set(xid = self.xid,
 366                 minrate = self.MinRate * 1000,
 367                 maxrate = self.MaxRate * 1000,
 368                 maxexemptrate = self.Maxi2Rate * 1000,
 369                 minexemptrate = self.Mini2Rate * 1000,
 370                 share = self.Share)
 371
 372         # Prepare message parameters from the template
 373         #message = ""
 374         #params = {'slice': self.name, 'hostname': socket.gethostname(),
 375         #          'since': time.asctime(time.gmtime(self.time)) + " GMT",
 376         #          'until': time.asctime(time.gmtime(self.time + period)) + " GMT",
 377         #          'date': time.asctime(time.gmtime()) + " GMT",
 378         #          'period': format_period(period)}
 379
 380         # Check limits.
 381         if usedbytes >= (self.bytes + (self.ThreshKByte * 1024)):
 382             sum = self.bytes + (self.ThreshKByte * 1024)
 383             maxbyte = self.MaxKByte * 1024
 384             bytesused = usedbytes - self.bytes
 385             timeused = int(time.time() - self.time)
 386             # Calcuate new rate.
 387             new_maxrate = int(((maxbyte - bytesused) * 8)/(period - timeused))
 388             # Never go under MinRate
 389             if new_maxrate < (self.MinRate * 1000):
 390                 new_maxrate = self.MinRate * 1000
 391             # State information.  I'm capped.
 392             self.capped = True
 393         else:
 394             # Sanity Check
 395             new_maxrate = self.MaxRate * 1000
 396             self.capped = False
 397
 398         ## Format template parameters for low bandwidth message
 399         #params['class'] = "low bandwidth"
 400         #params['bytes'] = format_bytes(usedbytes - self.bytes)
 401         #params['limit'] = format_bytes(self.MaxKByte * 1024)
 402         #params['thresh'] = format_bytes(self.ThreshKByte * 1024)
 403         #params['new_maxrate'] = bwlimit.format_tc_rate(new_maxrate)
 404
 405         # Cap low bandwidth burst rate
 406         #if new_maxrate != runningmaxrate:
 407         #    message += template % params
 408         #    logger.log("bwmon:   ** %(slice)s %(class)s capped at %(new_maxrate)s/s " % params)
 409
 410         if usedi2bytes >= (self.i2bytes + (self.Threshi2KByte * 1024)):
 411             maxi2byte = self.Maxi2KByte * 1024
 412             i2bytesused = usedi2bytes - self.i2bytes
 413             timeused = int(time.time() - self.time)
 414             # Calcuate New Rate.
 415             new_maxi2rate = int(((maxi2byte - i2bytesused) * 8)/(period - timeused))
 416             # Never go under MinRate
 417             if new_maxi2rate < (self.Mini2Rate * 1000):
 418                 new_maxi2rate = self.Mini2Rate * 1000
 419             # State information.  I'm capped.
 420             self.capped = True
 421         else:
 422             # Sanity
 423             new_maxi2rate = self.Maxi2Rate * 1000
 424             self.capped = False
 425
 426         # Format template parameters for high bandwidth message
 427         #params['class'] = "high bandwidth"
 428         #params['bytes'] = format_bytes(usedi2bytes - self.i2bytes)
 429         #params['limit'] = format_bytes(self.Maxi2KByte * 1024)
 430         #params['new_maxexemptrate'] = bwlimit.format_tc_rate(new_maxi2rate)
 431
 432         # Cap high bandwidth burst rate
 433         #if new_maxi2rate != runningmaxi2rate:
 434         #    message += template % params
 435         #    logger.log("bwmon:  %(slice)s %(class)s capped at %(new_maxexemptrate)s/s" % params)
 436
 437         # Apply parameters
 438         if new_maxrate != runningmaxrate or new_maxi2rate != runningmaxi2rate:
 439             bwlimit.set(xid = self.xid, maxrate = new_maxrate, maxexemptrate = new_maxi2rate)
 440
 441         # Notify slice
 442         if self.capped == True and self.emailed == False:
 443             self.notify(newmaxrate, newmaxexemptrate, usedbytes, usedi2bytes)
 444         #    subject = "pl_mom capped bandwidth of slice %(slice)s on %(hostname)s" % params
 445         #    if debug:
 446         #        logger.log("bwmon:  "+ subject)
 447         #        logger.log("bwmon:  "+ message + (footer % params))
 448         #    else:
 449         #        self.emailed = True
 450         #        slicemail(self.name, subject, message + (footer % params))
 451
 452 def gethtbs(root_xid, default_xid):
 453     """
 454     Return dict {xid: {*rates}} of running htbs as reported by tc that have names.
 455     Turn off HTBs without names.
 456     """
 457     livehtbs = {}
 458     for params in bwlimit.get():
 459         (xid, share,
 460          minrate, maxrate,
 461          minexemptrate, maxexemptrate,
 462          usedbytes, usedi2bytes) = params
 463
 464         name = bwlimit.get_slice(xid)
 465
 466         if (name is None) \
 467         and (xid != root_xid) \
 468         and (xid != default_xid):
 469             # Orphaned (not associated with a slice) class
 470             name = "%d?" % xid
 471             logger.log("bwmon:  Found orphaned HTB %s. Removing." %name)
 472             bwlimit.off(xid)
 473
 474         livehtbs[xid] = {'share': share,
 475             'minrate': minrate,
 476             'maxrate': maxrate,
 477             'maxexemptrate': maxexemptrate,
 478             'minexemptrate': minexemptrate,
 479             'usedbytes': usedbytes,
 480             'name': name,
 481             'usedi2bytes': usedi2bytes}
 482
 483     return livehtbs
 484
 485 def sync(nmdbcopy):
 486     """
 487     Syncs tc, db, and bwmon.dat.  Then, starts new slices, kills old ones, and updates byte accounts for each running slice.  Sends emails and caps those that went over their limit.
 488     """
 489     # Defaults
 490     global datafile, \
 491         period, \
 492         default_MaxRate, \
 493         default_Maxi2Rate, \
 494         default_MinRate, \
 495         default_MaxKByte,\
 496         default_ThreshKByte,\
 497         default_Maxi2KByte,\
 498         default_Threshi2KByte,\
 499         default_Share,\
 500         verbose
 501
 502     # All slices
 503     names = []
 504     # Incase the limits have changed.
 505     default_MaxRate = int(bwlimit.get_bwcap() / 1000)
 506     default_Maxi2Rate = int(bwlimit.bwmax / 1000)
 507
 508     # Incase default isn't set yet.
 509     if default_MaxRate == -1:
 510         default_MaxRate = 1000000
 511
 512     try:
 513         f = open(datafile, "r+")
 514         logger.log("bwmon:  Loading %s" % datafile)
 515         (version, slices, deaddb) = pickle.load(f)
 516         f.close()
 517         # Check version of data file
 518         if version != "$Id$":
 519             logger.log("bwmon:  Not using old version '%s' data file %s" % (version, datafile))
 520             raise Exception
 521     except Exception:
 522         version = "$Id$"
 523         slices = {}
 524         deaddb = {}
 525
 526     # Get/set special slice IDs
 527     root_xid = bwlimit.get_xid("root")
 528     default_xid = bwlimit.get_xid("default")
 529
 530     # Since root is required for sanity, its not in the API/plc database, so pass {}
 531     # to use defaults.
 532     if root_xid not in slices.keys():
 533         slices[root_xid] = Slice(root_xid, "root", {})
 534         slices[root_xid].reset(0, 0, 0, 0, {})
 535
 536     # Used by bwlimit.  pass {} since there is no rspec (like above).
 537     if default_xid not in slices.keys():
 538         slices[default_xid] = Slice(default_xid, "default", {})
 539         slices[default_xid].reset(0, 0, 0, 0, {})
 540
 541     live = {}
 542     # Get running slivers that should be on this node (from plc). {xid: name}
 543     # db keys on name, bwmon keys on xid.  db doesnt have xid either.
 544     for plcSliver in nmdbcopy.keys():
 545         live[bwlimit.get_xid(plcSliver)] = nmdbcopy[plcSliver]
 546
 547     logger.log("bwmon:  Found %s instantiated slices" % live.keys().__len__())
 548     logger.log("bwmon:  Found %s slices in dat file" % slices.values().__len__())
 549
 550     # Get actual running values from tc.
 551     # Update slice totals and bandwidth. {xid: {values}}
 552     kernelhtbs = gethtbs(root_xid, default_xid)
 553     logger.log("bwmon:  Found %s running HTBs" % kernelhtbs.keys().__len__())
 554
 555     # The dat file has HTBs for slices, but the HTBs aren't running
 556     nohtbslices =  Set(slices.keys()) - Set(kernelhtbs.keys())
 557     logger.log( "bwmon:  Found %s slices in dat but not running." % nohtbslices.__len__() )
 558     # Reset tc counts.
 559     for nohtbslice in nohtbslices:
 560         if live.has_key(nohtbslice):
 561             slices[nohtbslice].reset( 0, 0, 0, 0, live[nohtbslice]['_rspec'] )
 562
 563     # The dat file doesnt have HTB for the slice but kern has HTB
 564     slicesnodat = Set(kernelhtbs.keys()) - Set(slices.keys())
 565     logger.log( "bwmon: Found %s slices with HTBs but not in dat" % slicesnodat.__len__() )
 566     for slicenodat in slicesnodat:
 567         # But slice is running
 568         if live.has_key(slicenodat):
 569             # init the slice.  which means start accounting over since kernel
 570             # htb was already there.
 571             slices[slicenodat] = Slice(slicenodat,
 572                 live[slicenodat]['name'],
 573                 live[slicenodat]['_rspec'])
 574         else: bwlimit.off(slicenodat) # Abandoned.  it doesnt exist at PLC or the dat
 575
 576     # Get new slices.
 577     # Slices in GetSlivers but not running HTBs
 578     newslicesxids = Set(live.keys()) - Set(kernelhtbs.keys())
 579     logger.log("bwmon:  Found %s new slices" % newslicesxids.__len__())
 580
 581     # Setup new slices
 582     for newslice in newslicesxids:
 583         # Delegated slices dont have xids (which are uids) since they haven't been
 584         # instantiated yet.
 585         if newslice != None and live[newslice].has_key('_rspec') == True:
 586             # Check to see if we recently deleted this slice.
 587             if live[newslice]['name'] not in deaddb.keys():
 588                 logger.log( "bwmon: New Slice %s" % live[newslice]['name'] )
 589                 # _rspec is the computed rspec:  NM retrieved data from PLC, computed loans
 590                 # and made a dict of computed values.
 591                 slices[newslice] = Slice(newslice, live[newslice]['name'], live[newslice]['_rspec'])
 592                 slices[newslice].reset( 0, 0, 0, 0, live[newslice]['_rspec'] )
 593             # Double check time for dead slice in deaddb is within 24hr recording period.
 594             elif (time.time() <= (deaddb[live[newslice]['name']]['slice'].time + period)):
 595                 deadslice = deaddb[live[newslice]['name']]
 596                 logger.log("bwmon: Reinstantiating deleted slice %s" % live[newslice]['name'])
 597                 slices[newslice] = deadslice['slice']
 598                 slices[newslice].xid = newslice
 599                 # Start the HTB
 600                 slices[newslice].reset(deadslice['slice'].MaxRate,
 601                                     deadslice['slice'].Maxi2Rate,
 602                                     deadslice['htb']['usedbytes'],
 603                                     deadslice['htb']['usedi2bytes'],
 604                                     live[newslice]['_rspec'])
 605                 # Bring up to date
 606                 slices[newslice].update(deadslice['slice'].MaxRate,
 607                                     deadslice['slice'].Maxi2Rate,
 608                                     deadslice['htb']['usedbytes'],
 609                                     deadslice['htb']['usedi2bytes'],
 610                                     deadslice['htb']['share'],
 611                                     live[newslice]['_rspec'])
 612                 # Since the slice has been reinitialed, remove from dead database.
 613                 del deaddb[deadslice]
 614         else:
 615             logger.log("bwmon  Slice %s doesn't have xid.  Must be delegated."\
 616                        "Skipping." % live[newslice]['name'])
 617
 618     # Move dead slices that exist in the pickle file, but
 619     # aren't instantiated by PLC into the dead dict until
 620     # recording period is over.  This is to avoid the case where a slice is dynamically created
 621     # and destroyed then recreated to get around byte limits.
 622     deadxids = Set(slices.keys()) - Set(live.keys())
 623     logger.log("bwmon:  Found %s dead slices" % (deadxids.__len__() - 2))
 624     for deadxid in deadxids:
 625         if deadxid == root_xid or deadxid == default_xid:
 626             continue
 627         logger.log("bwmon:  removing dead slice  %s " % deadxid)
 628         if slices.has_key(deadxid):
 629             # add slice (by name) to deaddb
 630             deaddb[slices[deadxid].name] = {'slice': slices[deadxid], 'htb': kernelhtbs[deadxid]}
 631             del slices[deadxid]
 632         if kernelhtbs.has_key(deadxid):
 633             bwlimit.off(deadxid)
 634
 635         # Clean up deaddb
 636         for (deadslice, deadhtb) in deaddb.iteritems():
 637                 if (time.time() >= (deadslice.time() + period)):
 638                         logger.log("bwmon: Removing dead slice %s from dat." % deadslice.name)
 639                         del deaddb[deadslice.name]
 640
 641     # Get actual running values from tc since we've added and removed buckets.
 642     # Update slice totals and bandwidth. {xid: {values}}
 643     kernelhtbs = gethtbs(root_xid, default_xid)
 644     logger.log("bwmon:  now %s running HTBs" % kernelhtbs.keys().__len__())
 645
 646     for (xid, slice) in slices.iteritems():
 647         # Monitor only the specified slices
 648         if xid == root_xid or xid == default_xid: continue
 649         if names and name not in names:
 650             continue
 651
 652         if (time.time() >= (slice.time + period)) or \
 653             (kernelhtbs[xid]['usedbytes'] < slice.bytes) or \
 654             (kernelhtbs[xid]['usedi2bytes'] < slice.i2bytes):
 655             # Reset to defaults every 24 hours or if it appears
 656             # that the byte counters have overflowed (or, more
 657             # likely, the node was restarted or the HTB buckets
 658             # were re-initialized).
 659             slice.reset(kernelhtbs[xid]['maxrate'], \
 660                 kernelhtbs[xid]['maxexemptrate'], \
 661                 kernelhtbs[xid]['usedbytes'], \
 662                 kernelhtbs[xid]['usedi2bytes'], \
 663                 live[xid]['_rspec'])
 664         else:
 665             if debug:  logger.log("bwmon: Updating slice %s" % slice.name)
 666             # Update byte counts
 667             slice.update(kernelhtbs[xid]['maxrate'], \
 668                 kernelhtbs[xid]['maxexemptrate'], \
 669                 kernelhtbs[xid]['usedbytes'], \
 670                 kernelhtbs[xid]['usedi2bytes'], \
 671                 kernelhtbs[xid]['share'],
 672                 live[xid]['_rspec'])
 673
 674     logger.log("bwmon:  Saving %s slices in %s" % (slices.keys().__len__(),datafile))
 675     f = open(datafile, "w")
 676     pickle.dump((version, slices, deaddb), f)
 677     f.close()
 678
 679 lock = threading.Event()
 680 def run():
 681     """When run as a thread, wait for event, lock db, deep copy it, release it, run bwmon.GetSlivers(), then go back to waiting."""
 682     if debug:  logger.log("bwmon:  Thread started")
 683     while True:
 684         lock.wait()
 685         if debug: logger.log("bwmon:  Event received.  Running.")
 686         database.db_lock.acquire()
 687         nmdbcopy = copy.deepcopy(database.db)
 688         database.db_lock.release()
 689         try:  sync(nmdbcopy)
 690         except: logger.log_exc()
 691         lock.clear()
 692
 693 def start(*args):
 694     tools.as_daemon_thread(run)
 695
 696 def GetSlivers(*args):
 697     pass