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, 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         # Prepare message parameters from the template
 370         #message = ""
 371         #params = {'slice': self.name, 'hostname': socket.gethostname(),
 372         #          'since': time.asctime(time.gmtime(self.time)) + " GMT",
 373         #          'until': time.asctime(time.gmtime(self.time + period)) + " GMT",
 374         #          'date': time.asctime(time.gmtime()) + " GMT",
 375         #          'period': format_period(period)}
 376
 377         # Check limits.
 378         if usedbytes >= (self.bytes + (self.ThreshKByte * 1024)):
 379             sum = self.bytes + (self.ThreshKByte * 1024)
 380             maxbyte = self.MaxKByte * 1024
 381             bytesused = usedbytes - self.bytes
 382             timeused = int(time.time() - self.time)
 383             # Calcuate new rate.
 384             new_maxrate = int(((maxbyte - bytesused) * 8)/(period - timeused))
 385             # Never go under MinRate
 386             if new_maxrate < (self.MinRate * 1000):
 387                 new_maxrate = self.MinRate * 1000
 388             # State information.  I'm capped.
 389             self.capped = True
 390         else:
 391             # Sanity Check
 392             new_maxrate = self.MaxRate * 1000
 393             self.capped = False
 394
 395         ## Format template parameters for low bandwidth message
 396         #params['class'] = "low bandwidth"
 397         #params['bytes'] = format_bytes(usedbytes - self.bytes)
 398         #params['limit'] = format_bytes(self.MaxKByte * 1024)
 399         #params['thresh'] = format_bytes(self.ThreshKByte * 1024)
 400         #params['new_maxrate'] = bwlimit.format_tc_rate(new_maxrate)
 401
 402         # Cap low bandwidth burst rate
 403         #if new_maxrate != runningmaxrate:
 404         #    message += template % params
 405         #    logger.log("bwmon:   ** %(slice)s %(class)s capped at %(new_maxrate)s/s " % params)
 406
 407         if usedi2bytes >= (self.i2bytes + (self.Threshi2KByte * 1024)):
 408             maxi2byte = self.Maxi2KByte * 1024
 409             i2bytesused = usedi2bytes - self.i2bytes
 410             timeused = int(time.time() - self.time)
 411             # Calcuate New Rate.
 412             new_maxi2rate = int(((maxi2byte - i2bytesused) * 8)/(period - timeused))
 413             # Never go under MinRate
 414             if new_maxi2rate < (self.Mini2Rate * 1000):
 415                 new_maxi2rate = self.Mini2Rate * 1000
 416             # State information.  I'm capped.
 417             self.capped = True
 418         else:
 419             # Sanity
 420             new_maxi2rate = self.Maxi2Rate * 1000
 421             self.capped = False
 422
 423         # Format template parameters for high bandwidth message
 424         #params['class'] = "high bandwidth"
 425         #params['bytes'] = format_bytes(usedi2bytes - self.i2bytes)
 426         #params['limit'] = format_bytes(self.Maxi2KByte * 1024)
 427         #params['new_maxexemptrate'] = bwlimit.format_tc_rate(new_maxi2rate)
 428
 429         # Cap high bandwidth burst rate
 430         #if new_maxi2rate != runningmaxi2rate:
 431         #    message += template % params
 432         #    logger.log("bwmon:  %(slice)s %(class)s capped at %(new_maxexemptrate)s/s" % params)
 433
 434         # Apply parameters
 435         if new_maxrate != runningmaxrate or new_maxi2rate != runningmaxi2rate:
 436             bwlimit.set(xid = self.xid, maxrate = new_maxrate, maxexemptrate = new_maxi2rate)
 437
 438         # Notify slice
 439         if self.capped == True and self.emailed == False:
 440             self.notify(newmaxrate, newmaxexemptrate, usedbytes, usedi2bytes)
 441         #    subject = "pl_mom capped bandwidth of slice %(slice)s on %(hostname)s" % params
 442         #    if debug:
 443         #        logger.log("bwmon:  "+ subject)
 444         #        logger.log("bwmon:  "+ message + (footer % params))
 445         #    else:
 446         #        self.emailed = True
 447         #        slicemail(self.name, subject, message + (footer % params))
 448
 449 def gethtbs(root_xid, default_xid):
 450     """
 451     Return dict {xid: {*rates}} of running htbs as reported by tc that have names.
 452     Turn off HTBs without names.
 453     """
 454     livehtbs = {}
 455     for params in bwlimit.get():
 456         (xid, share,
 457          minrate, maxrate,
 458          minexemptrate, maxexemptrate,
 459          usedbytes, usedi2bytes) = params
 460
 461         name = bwlimit.get_slice(xid)
 462
 463         if (name is None) \
 464         and (xid != root_xid) \
 465         and (xid != default_xid):
 466             # Orphaned (not associated with a slice) class
 467             name = "%d?" % xid
 468             logger.log("bwmon:  Found orphaned HTB %s. Removing." %name)
 469             bwlimit.off(xid)
 470
 471         livehtbs[xid] = {'share': share,
 472             'minrate': minrate,
 473             'maxrate': maxrate,
 474             'maxexemptrate': maxexemptrate,
 475             'minexemptrate': minexemptrate,
 476             'usedbytes': usedbytes,
 477             'name': name,
 478             'usedi2bytes': usedi2bytes}
 479
 480     return livehtbs
 481
 482 def sync(nmdbcopy):
 483     """
 484     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.
 485     """
 486     # Defaults
 487     global datafile, \
 488         period, \
 489         default_MaxRate, \
 490         default_Maxi2Rate, \
 491         default_MinRate, \
 492         default_MaxKByte,\
 493         default_ThreshKByte,\
 494         default_Maxi2KByte,\
 495         default_Threshi2KByte,\
 496         default_Share,\
 497         verbose
 498
 499     # All slices
 500     names = []
 501     # Incase the limits have changed.
 502     default_MaxRate = int(bwlimit.get_bwcap() / 1000)
 503     default_Maxi2Rate = int(bwlimit.bwmax / 1000)
 504
 505     # Incase default isn't set yet.
 506     if default_MaxRate == -1:
 507         default_MaxRate = 1000000
 508
 509     try:
 510         f = open(datafile, "r+")
 511         logger.log("bwmon:  Loading %s" % datafile)
 512         (version, slices, deaddb) = pickle.load(f)
 513         f.close()
 514         # Check version of data file
 515         if version != "$Id$":
 516             logger.log("bwmon:  Not using old version '%s' data file %s" % (version, datafile))
 517             raise Exception
 518     except Exception:
 519         version = "$Id$"
 520         slices = {}
 521         deaddb = {}
 522
 523     # Get/set special slice IDs
 524     root_xid = bwlimit.get_xid("root")
 525     default_xid = bwlimit.get_xid("default")
 526
 527     # Since root is required for sanity, its not in the API/plc database, so pass {}
 528     # to use defaults.
 529     if root_xid not in slices.keys():
 530         slices[root_xid] = Slice(root_xid, "root", {})
 531         slices[root_xid].reset(0, 0, 0, 0, {})
 532
 533     # Used by bwlimit.  pass {} since there is no rspec (like above).
 534     if default_xid not in slices.keys():
 535         slices[default_xid] = Slice(default_xid, "default", {})
 536         slices[default_xid].reset(0, 0, 0, 0, {})
 537
 538     live = {}
 539     # Get running slivers that should be on this node (from plc). {xid: name}
 540     # db keys on name, bwmon keys on xid.  db doesnt have xid either.
 541     for plcSliver in nmdbcopy.keys():
 542         live[bwlimit.get_xid(plcSliver)] = nmdbcopy[plcSliver]
 543
 544     logger.log("bwmon:  Found %s instantiated slices" % live.keys().__len__())
 545     logger.log("bwmon:  Found %s slices in dat file" % slices.values().__len__())
 546
 547     # Get actual running values from tc.
 548     # Update slice totals and bandwidth. {xid: {values}}
 549     kernelhtbs = gethtbs(root_xid, default_xid)
 550     logger.log("bwmon:  Found %s running HTBs" % kernelhtbs.keys().__len__())
 551
 552     # The dat file has HTBs for slices, but the HTBs aren't running
 553     nohtbslices =  Set(slices.keys()) - Set(kernelhtbs.keys())
 554     logger.log( "bwmon:  Found %s slices in dat but not running." % nohtbslices.__len__() )
 555     # Reset tc counts.
 556     for nohtbslice in nohtbslices:
 557         if live.has_key(nohtbslice):
 558             slices[nohtbslice].reset( 0, 0, 0, 0, live[nohtbslice]['_rspec'] )
 559
 560     # The dat file doesnt have HTB for the slice but kern has HTB
 561     slicesnodat = Set(kernelhtbs.keys()) - Set(slices.keys())
 562     logger.log( "bwmon: Found %s slices with HTBs but not in dat" % slicesnodat.__len__() )
 563     for slicenodat in slicesnodat:
 564         # But slice is running
 565         if live.has_key(slicenodat):
 566             # init the slice.  which means start accounting over since kernel
 567             # htb was already there.
 568             slices[slicenodat] = Slice(slicenodat,
 569                 live[slicenodat]['name'],
 570                 live[slicenodat]['_rspec'])
 571         else: bwlimit.off(slicenodat) # Abandoned.  it doesnt exist at PLC or the dat
 572
 573     # Get new slices.
 574     # Slices in GetSlivers but not running HTBs
 575     newslicesxids = Set(live.keys()) - Set(kernelhtbs.keys())
 576     logger.log("bwmon:  Found %s new slices" % newslicesxids.__len__())
 577
 578     # Setup new slices
 579     for newslice in newslicesxids:
 580         # Delegated slices dont have xids (which are uids) since they haven't been
 581         # instantiated yet.
 582         if newslice != None and live[newslice].has_key('_rspec') == True:
 583             # Check to see if we recently deleted this slice.
 584             if live[newslice]['name'] not in deaddb.keys():
 585                 logger.log( "bwmon: New Slice %s" % live[newslice]['name'] )
 586                 # _rspec is the computed rspec:  NM retrieved data from PLC, computed loans
 587                 # and made a dict of computed values.
 588                 slices[newslice] = Slice(newslice, live[newslice]['name'], live[newslice]['_rspec'])
 589                 slices[newslice].reset( 0, 0, 0, 0, live[newslice]['_rspec'] )
 590             # Double check time for dead slice in deaddb is within 24hr recording period.
 591             elif (time.time() <= (deaddb[live[newslice]['name']]['slice'].time + period)):
 592                 deadslice = deaddb[live[newslice]['name']]
 593                 logger.log("bwmon: Reinstantiating deleted slice %s" % live[newslice]['name'])
 594                 slices[newslice] = deadslice['slice']
 595                 slices[newslice].xid = newslice
 596                 # Start the HTB
 597                 slices[newslice].reset(deadslice['slice'].MaxRate,
 598                                     deadslice['slice'].Maxi2Rate,
 599                                     deadslice['htb']['usedbytes'],
 600                                     deadslice['htb']['usedi2bytes'],
 601                                     live[newslice]['_rspec'])
 602                 # Bring up to date
 603                 slices[newslice].update(deadslice['slice'].MaxRate,
 604                                     deadslice['slice'].Maxi2Rate,
 605                                     deadslice['htb']['usedbytes'],
 606                                     deadslice['htb']['usedi2bytes'],
 607                                     live[newslice]['_rspec'])
 608                 # Since the slice has been reinitialed, remove from dead database.
 609                 del deaddb[deadslice]
 610         else:
 611             logger.log("bwmon  Slice %s doesn't have xid.  Must be delegated."\
 612                        "Skipping." % live[newslice]['name'])
 613
 614     # Move dead slices that exist in the pickle file, but
 615     # aren't instantiated by PLC into the dead dict until
 616     # recording period is over.  This is to avoid the case where a slice is dynamically created
 617     # and destroyed then recreated to get around byte limits.
 618     deadxids = Set(slices.keys()) - Set(live.keys())
 619     logger.log("bwmon:  Found %s dead slices" % (deadxids.__len__() - 2))
 620     for deadxid in deadxids:
 621         if deadxid == root_xid or deadxid == default_xid:
 622             continue
 623         logger.log("bwmon:  removing dead slice  %s " % deadxid)
 624         if slices.has_key(deadxid):
 625             # add slice (by name) to deaddb
 626             deaddb[slices[deadxid].name] = {'slice': slices[deadxid], 'htb': kernelhtbs[deadxid]}
 627             del slices[deadxid]
 628         if kernelhtbs.has_key(deadxid):
 629             bwlimit.off(deadxid)
 630
 631         # Clean up deaddb
 632         for (deadslice, deadhtb) in deaddb.iteritems():
 633                 if (time.time() >= (deadslice.time() + period)):
 634                         logger.log("bwmon: Removing dead slice %s from dat." % deadslice.name)
 635                         del deaddb[deadslice.name]
 636
 637     # Get actual running values from tc since we've added and removed buckets.
 638     # Update slice totals and bandwidth. {xid: {values}}
 639     kernelhtbs = gethtbs(root_xid, default_xid)
 640     logger.log("bwmon:  now %s running HTBs" % kernelhtbs.keys().__len__())
 641
 642     for (xid, slice) in slices.iteritems():
 643         # Monitor only the specified slices
 644         if xid == root_xid or xid == default_xid: continue
 645         if names and name not in names:
 646             continue
 647
 648         if (time.time() >= (slice.time + period)) or \
 649             (kernelhtbs[xid]['usedbytes'] < slice.bytes) or \
 650             (kernelhtbs[xid]['usedi2bytes'] < slice.i2bytes):
 651             # Reset to defaults every 24 hours or if it appears
 652             # that the byte counters have overflowed (or, more
 653             # likely, the node was restarted or the HTB buckets
 654             # were re-initialized).
 655             slice.reset(kernelhtbs[xid]['maxrate'], \
 656                 kernelhtbs[xid]['maxexemptrate'], \
 657                 kernelhtbs[xid]['usedbytes'], \
 658                 kernelhtbs[xid]['usedi2bytes'], \
 659                 live[xid]['_rspec'])
 660         else:
 661             if debug:  logger.log("bwmon: Updating slice %s" % slice.name)
 662             # Update byte counts
 663             slice.update(kernelhtbs[xid]['maxrate'], \
 664                 kernelhtbs[xid]['maxexemptrate'], \
 665                 kernelhtbs[xid]['usedbytes'], \
 666                 kernelhtbs[xid]['usedi2bytes'], \
 667                 live[xid]['_rspec'])
 668
 669     logger.log("bwmon:  Saving %s slices in %s" % (slices.keys().__len__(),datafile))
 670     f = open(datafile, "w")
 671     pickle.dump((version, slices, deaddb), f)
 672     f.close()
 673
 674 lock = threading.Event()
 675 def run():
 676     """When run as a thread, wait for event, lock db, deep copy it, release it, run bwmon.GetSlivers(), then go back to waiting."""
 677     if debug:  logger.log("bwmon:  Thread started")
 678     while True:
 679         lock.wait()
 680         if debug: logger.log("bwmon:  Event received.  Running.")
 681         database.db_lock.acquire()
 682         nmdbcopy = copy.deepcopy(database.db)
 683         database.db_lock.release()
 684         try:  sync(nmdbcopy)
 685         except: logger.log_exc()
 686         lock.clear()
 687
 688 def start(*args):
 689     tools.as_daemon_thread(run)
 690
 691 def GetSlivers(*args):
 692     pass