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