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