bwmon.py

   1 #!/usr/bin/python
   2 #
   3 # Average bandwidth monitoring script. Run periodically via cron(8) to
   4 # enforce a soft limit on daily bandwidth usage for each slice. If a
   5 # slice is found to have exceeded its daily bandwidth usage when the
   6 # script is run, its instantaneous rate will be capped at the desired
   7 # average rate. Thus, in the worst case, a slice will only be able to
   8 # send a little more than twice its average daily limit.
   9 #
  10 # Two separate limits are enforced, one for destinations exempt from
  11 # the node bandwidth cap, and the other for all other destinations.
  12 #
  13 # Mark Huang <mlhuang@cs.princeton.edu>
  14 # Andy Bavier <acb@cs.princeton.edu>
  15 # Faiyaz Ahmed <faiyaza@cs.princeton.edu>
  16 # Copyright (C) 2004-2006 The Trustees of Princeton University
  17 #
  18 # $Id: bwmon.py,v 1.12 2007/03/06 20:46:54 faiyaza Exp $
  19 #
  20
  21 import os
  22 import sys
  23 import time
  24 import pickle
  25
  26 import socket
  27 import bwlimit
  28 import logger
  29
  30 from sets import Set
  31
  32 try:
  33     sys.path.append("/etc/planetlab")
  34     from plc_config import *
  35 except:
  36     logger.log("bwmon:  Warning: Configuration file /etc/planetlab/plc_config.py not found")
  37     PLC_NAME = "PlanetLab"
  38     PLC_SLICE_PREFIX = "pl"
  39     PLC_MAIL_SUPPORT_ADDRESS = "support@planet-lab.org"
  40     PLC_MAIL_SLICE_ADDRESS = "SLICE@slices.planet-lab.org"
  41
  42 # Constants
  43 seconds_per_day = 24 * 60 * 60
  44 bits_per_byte = 8
  45
  46 # Defaults
  47 debug = False
  48 verbose = False
  49 datafile = "/var/lib/misc/bwmon.dat"
  50 #nm = None
  51
  52 # Burst to line rate (or node cap).  Set by NM. in KBit/s
  53 default_MaxRate = int(bwlimit.get_bwcap() / 1000)
  54 default_Maxi2Rate = int(bwlimit.bwmax / 1000)
  55 # Min rate 8 bits/s
  56 default_MinRate = 0
  57 default_Mini2Rate = 0
  58 # 5.4 Gbyte per day. 5.4 * 1024 k * 1024M * 1024G
  59 # 5.4 Gbyte per day max allowed transfered per recording period
  60 default_MaxKByte = 5662310
  61 default_ThreshKByte = int(.8 * default_MaxKByte)
  62 # 16.4 Gbyte per day max allowed transfered per recording period to I2
  63 default_Maxi2KByte = 17196646
  64 default_Threshi2KByte = int(.8 * default_Maxi2KByte)
  65 # Default share quanta
  66 default_Share = 1
  67
  68 # Average over 1 day
  69 period = 1 * seconds_per_day
  70
  71 # Message template
  72 template = \
  73 """
  74 The slice %(slice)s has transmitted more than %(bytes)s from
  75 %(hostname)s to %(class)s destinations
  76 since %(since)s.
  77
  78 Its maximum %(class)s burst rate will be capped at %(new_maxrate)s/s
  79 until %(until)s.
  80
  81 Please reduce the average %(class)s transmission rate
  82 of the slice to %(limit)s per %(period)s.
  83
  84 """.lstrip()
  85
  86 footer = \
  87 """
  88 %(date)s %(hostname)s bwcap %(slice)s
  89 """.lstrip()
  90
  91 def format_bytes(bytes, si = True):
  92     """
  93     Formats bytes into a string
  94     """
  95     if si:
  96         kilo = 1000.
  97     else:
  98         # Officially, a kibibyte
  99         kilo = 1024.
 100
 101     if bytes >= (kilo * kilo * kilo):
 102         return "%.1f GB" % (bytes / (kilo * kilo * kilo))
 103     elif bytes >= 1000000:
 104         return "%.1f MB" % (bytes / (kilo * kilo))
 105     elif bytes >= 1000:
 106         return "%.1f KB" % (bytes / kilo)
 107     else:
 108         return "%.0f bytes" % bytes
 109
 110 def format_period(seconds):
 111     """
 112     Formats a period in seconds into a string
 113     """
 114
 115     if seconds == (24 * 60 * 60):
 116         return "day"
 117     elif seconds == (60 * 60):
 118         return "hour"
 119     elif seconds > (24 * 60 * 60):
 120         return "%.1f days" % (seconds / 24. / 60. / 60.)
 121     elif seconds > (60 * 60):
 122         return "%.1f hours" % (seconds / 60. / 60.)
 123     elif seconds > (60):
 124         return "%.1f minutes" % (seconds / 60.)
 125     else:
 126         return "%.0f seconds" % seconds
 127
 128 def slicemail(slice, subject, body):
 129     sendmail = os.popen("/usr/sbin/sendmail -N never -t -f%s" % PLC_MAIL_SUPPORT_ADDRESS, "w")
 130
 131     # PLC has a separate list for pl_mom messages
 132     if PLC_MAIL_SUPPORT_ADDRESS == "support@planet-lab.org":
 133         to = ["pl-mom@planet-lab.org"]
 134     else:
 135         to = [PLC_MAIL_SUPPORT_ADDRESS]
 136
 137     if slice is not None and slice != "root":
 138         to.append(PLC_MAIL_SLICE_ADDRESS.replace("SLICE", slice))
 139
 140     header = {'from': "%s Support <%s>" % (PLC_NAME, PLC_MAIL_SUPPORT_ADDRESS),
 141               'to': ", ".join(to),
 142               'version': sys.version.split(" ")[0],
 143               'subject': subject}
 144
 145     # Write headers
 146     sendmail.write(
 147 """
 148 Content-type: text/plain
 149 From: %(from)s
 150 Reply-To: %(from)s
 151 To: %(to)s
 152 X-Mailer: Python/%(version)s
 153 Subject: %(subject)s
 154
 155 """.lstrip() % header)
 156
 157     # Write body
 158     sendmail.write(body)
 159     # Done
 160     sendmail.close()
 161
 162
 163 class Slice:
 164     """
 165     Stores the last recorded bandwidth parameters of a slice.
 166
 167     xid - slice context/VServer ID
 168     name - slice name
 169     time - beginning of recording period in UNIX seconds
 170     bytes - low bandwidth bytes transmitted at the beginning of the recording period
 171     i2bytes - high bandwidth bytes transmitted at the beginning of the recording period (for I2 -F)
 172     ByteMax - total volume of data allowed
 173     ByteThresh - After thresh, cap node to (maxbyte - bytes)/(time left in period)
 174     ExemptByteMax - Same as above, but for i2.
 175     ExemptByteThresh - i2 ByteThresh
 176     maxrate - max_rate slice attribute.
 177     maxexemptrate - max_exempt_rate slice attribute.
 178     self.emailed = did we 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 = default_ThreshKByte
 194         self.Maxi2KByte = default_Maxi2KByte
 195         self.Threshi2KByte = default_Threshi2KByte
 196         self.Share = default_Share
 197         self.Sharei2 = default_Share
 198         self.emailed = False
 199
 200         self.updateSliceAttributes(rspec)
 201         bwlimit.set(xid = self.xid,
 202                 minrate = self.MinRate * 1000,
 203                 maxrate = self.MaxRate * 1000,
 204                 maxexemptrate = self.Maxi2Rate * 1000,
 205                 minexemptrate = self.Mini2Rate * 1000,
 206                 share = self.Share)
 207
 208     def __repr__(self):
 209         return self.name
 210
 211     def updateSliceAttributes(self, rspec):
 212
 213         # Get attributes
 214
 215         # Sanity check plus policy decision for MinRate:
 216         # Minrate cant be greater than 25% of MaxRate or NodeCap.
 217         MinRate = int(rspec.get("net_min_rate", default_MinRate))
 218         if MinRate > int(.25 * default_MaxRate):
 219             MinRate = int(.25 * default_MaxRate)
 220         if MinRate != self.MinRate:
 221             self.MinRate = MinRate
 222             logger.log("bwmon:  Updating %s. Min Rate = %s" %(self.name, self.MinRate))
 223
 224         MaxRate = int(rspec.get('net_max_rate', bwlimit.get_bwcap() / 1000))
 225         if MaxRate != self.MaxRate:
 226             self.MaxRate = MaxRate
 227             logger.log("bwmon:  Updating %s. Max Rate = %s" %(self.name, self.MaxRate))
 228
 229         Mini2Rate = int(rspec.get('net_i2_min_rate', default_Mini2Rate))
 230         if Mini2Rate != self.Mini2Rate:
 231             self.Mini2Rate = Mini2Rate
 232             logger.log("bwmon:  Updating %s. Min i2 Rate = %s" %(self.name, self.Mini2Rate))
 233
 234         Maxi2Rate = int(rspec.get('net_i2_max_rate', bwlimit.bwmax / 1000))
 235         if Maxi2Rate != self.Maxi2Rate:
 236             self.Maxi2Rate = Maxi2Rate
 237             logger.log("bwmon:  Updating %s. Max i2 Rate = %s" %(self.name, self.Maxi2Rate))
 238
 239         MaxKByte = int(rspec.get('net_max_kbyte', default_MaxKByte))
 240         if MaxKByte != self.MaxKByte:
 241             self.MaxKByte = MaxKByte
 242             logger.log("bwmon:  Updating %s. Max KByte lim = %s" %(self.name, self.MaxKByte))
 243
 244         Maxi2KByte = int(rspec.get('net_i2_max_kbyte', default_Maxi2KByte))
 245         if Maxi2KByte != self.Maxi2KByte:
 246             self.Maxi2KByte = Maxi2KByte
 247             logger.log("bwmon:  Updating %s. Max i2 KByte = %s" %(self.name, self.Maxi2KByte))
 248
 249         ThreshKByte = int(rspec.get('net_thresh_kbyte', default_ThreshKByte))
 250         if ThreshKByte != self.ThreshKByte:
 251             self.ThreshKByte = ThreshKByte
 252             logger.log("bwmon:  Updating %s. Thresh KByte = %s" %(self.name, self.ThreshKByte))
 253
 254         Threshi2KByte = int(rspec.get('net_i2_thresh_kbyte', default_Threshi2KByte))
 255         if Threshi2KByte != self.Threshi2KByte:
 256             self.Threshi2KByte = Threshi2KByte
 257             logger.log("bwmon:  Updating %s. i2 Thresh KByte = %s" %(self.name, self.Threshi2KByte))
 258
 259         Share = int(rspec.get('net_share', default_Share))
 260         if Share != self.Share:
 261             self.Share = Share
 262             logger.log("bwmon:  Updating %s. Net Share = %s" %(self.name, self.Share))
 263
 264         Sharei2 = int(rspec.get('net_i2_share', default_Share))
 265         if Sharei2 != self.Sharei2:
 266             self.Sharei2 = Sharei2
 267             logger.log("bwmon:  Updating %s. Net i2 Share = %s" %(self.name, self.i2Share))
 268
 269
 270     def reset(self, runningmaxrate, runningmaxi2rate, usedbytes, usedi2bytes, rspec):
 271         """
 272         Begin a new recording period. Remove caps by restoring limits
 273         to their default values.
 274         """
 275
 276         # Query Node Manager for max rate overrides
 277         self.updateSliceAttributes(rspec)
 278
 279         # Reset baseline time
 280         self.time = time.time()
 281
 282         # Reset baseline byte coutns
 283         self.bytes = usedbytes
 284         self.i2bytes = usedi2bytes
 285
 286         # Reset email
 287         self.emailed = False
 288         maxrate = self.MaxRate * 1000
 289         maxi2rate = self.Maxi2Rate * 1000
 290         # Reset rates.
 291         if (self.MaxRate != runningmaxrate) or (self.Maxi2Rate != runningmaxi2rate):
 292             logger.log("bwmon:  %s reset to %s/%s" % \
 293                   (self.name,
 294                    bwlimit.format_tc_rate(maxrate),
 295                    bwlimit.format_tc_rate(maxi2rate)))
 296             bwlimit.set(xid = self.xid,
 297                 minrate = self.MinRate * 1000,
 298                 maxrate = self.MaxRate * 1000,
 299                 maxexemptrate = self.Maxi2Rate * 1000,
 300                 minexemptrate = self.Mini2Rate * 1000,
 301                 share = self.Share)
 302
 303     def update(self, runningmaxrate, runningmaxi2rate, usedbytes, usedi2bytes, rspec):
 304         """
 305         Update byte counts and check if byte limits have been
 306         exceeded.
 307         """
 308
 309         # Query Node Manager for max rate overrides
 310         self.updateSliceAttributes(rspec)
 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 usedbytes >= (self.bytes + (self.ThreshKByte * 1024)):
 321             sum = self.bytes + (self.ThreshKByte * 1024)
 322             maxbyte = self.MaxKByte * 1024
 323             bytesused = usedbytes - self.bytes
 324             timeused = int(time.time() - self.time)
 325             new_maxrate = int(((maxbyte - bytesused) * 8)/(period - timeused))
 326             if new_maxrate < (self.MinRate * 1000):
 327                 new_maxrate = self.MinRate * 1000
 328         else:
 329             new_maxrate = self.MaxRate * 1000
 330
 331         # Format template parameters for low bandwidth message
 332         params['class'] = "low bandwidth"
 333         params['bytes'] = format_bytes(usedbytes - self.bytes)
 334         params['limit'] = format_bytes(self.MaxKByte * 1024)
 335         params['new_maxrate'] = bwlimit.format_tc_rate(new_maxrate)
 336
 337         if verbose:
 338             logger.log("bwmon:  %(slice)s %(class)s " \
 339                   "%(bytes)s of %(limit)s (%(new_maxrate)s/s maxrate)" % \
 340                   params)
 341
 342         # Cap low bandwidth burst rate
 343         if new_maxrate != runningmaxrate:
 344             message += template % params
 345             logger.log("bwmon:   ** %(slice)s %(class)s capped at %(new_maxrate)s/s " % params)
 346
 347         if usedi2bytes >= (self.i2bytes + (self.Threshi2KByte * 1024)):
 348             maxi2byte = self.Maxi2KByte * 1024
 349             i2bytesused = usedi2bytes - self.i2bytes
 350             timeused = int(time.time() - self.time)
 351             new_maxi2rate = int(((maxi2byte - i2bytesused) * 8)/(period - timeused))
 352             if new_maxi2rate < (self.Mini2Rate * 1000):
 353                 new_maxi2rate = self.Mini2Rate * 1000
 354         else:
 355             new_maxi2rate = self.Maxi2Rate * 1000
 356
 357         # Format template parameters for high bandwidth message
 358         params['class'] = "high bandwidth"
 359         params['bytes'] = format_bytes(usedi2bytes - self.i2bytes)
 360         params['limit'] = format_bytes(self.Maxi2KByte * 1024)
 361         params['new_maxexemptrate'] = bwlimit.format_tc_rate(new_maxi2rate)
 362
 363         if verbose:
 364             logger.log("bwmon:  %(slice)s %(class)s " \
 365                   "%(bytes)s of %(limit)s (%(new_maxrate)s/s maxrate)" % params)
 366
 367         # Cap high bandwidth burst rate
 368         if new_maxi2rate != runningmaxi2rate:
 369             message += template % params
 370             logger.log("bwmon:  %(slice)s %(class)s capped at %(new_maxexemptrate)s/s" % params)
 371
 372         # Apply parameters
 373         if new_maxrate != runningmaxrate or new_maxi2rate != runningmaxi2rate:
 374             bwlimit.set(xid = self.xid, maxrate = new_maxrate, maxexemptrate = new_maxi2rate)
 375
 376         # Notify slice
 377         if message and self.emailed == False:
 378             subject = "pl_mom capped bandwidth of slice %(slice)s on %(hostname)s" % params
 379             if debug:
 380                 logger.log("bwmon:  "+ subject)
 381                 logger.log("bwmon:  "+ message + (footer % params))
 382             else:
 383                 self.emailed = True
 384                 slicemail(self.name, subject, message + (footer % params))
 385
 386 def GetSlivers(db):
 387     # Defaults
 388     global datafile, \
 389         period, \
 390         default_MaxRate, \
 391         default_Maxi2Rate, \
 392         default_MinRate, \
 393         default_MaxKByte,\
 394         default_ThreshKByte,\
 395         default_Maxi2KByte,\
 396         default_Threshi2KByte,\
 397         default_Share,\
 398         verbose
 399
 400     # All slices
 401     names = []
 402
 403     # Incase the limits have changed.
 404     default_MaxRate = int(bwlimit.get_bwcap() / 1000)
 405     default_Maxi2Rate = int(bwlimit.bwmax / 1000)
 406
 407     # Incase default isn't set yet.
 408     if default_MaxRate == -1:
 409         default_MaxRate = 1000000
 410
 411     try:
 412         f = open(datafile, "r+")
 413         logger.log("bwmon:  Loading %s" % datafile)
 414         (version, slices) = pickle.load(f)
 415         f.close()
 416         # Check version of data file
 417         if version != "$Id: bwmon.py,v 1.12 2007/03/06 20:46:54 faiyaza Exp $":
 418             logger.log("bwmon:  Not using old version '%s' data file %s" % (version, datafile))
 419             raise Exception
 420     except Exception:
 421         version = "$Id: bwmon.py,v 1.12 2007/03/06 20:46:54 faiyaza Exp $"
 422         slices = {}
 423
 424     # Get/set special slice IDs
 425     root_xid = bwlimit.get_xid("root")
 426     default_xid = bwlimit.get_xid("default")
 427
 428     # Since root is required for sanity, its not in the API/plc database, so pass {}
 429     # to use defaults.
 430     if root_xid not in slices.keys():
 431         slices[root_xid] = Slice(root_xid, "root", {})
 432         slices[root_xid].reset(0, 0, 0, 0, {})
 433
 434     # Used by bwlimit.  pass {} since there is no rspec (like above).
 435     if default_xid not in slices.keys():
 436         slices[default_xid] = Slice(default_xid, "default", {})
 437         slices[default_xid].reset(0, 0, 0, 0, {})
 438
 439     live = {}
 440     # Get running slivers that should be on this node (from plc). {xid: name}
 441     for sliver in db.keys():
 442         live[bwlimit.get_xid(sliver)] = sliver
 443
 444     # Setup new slices.
 445     # live.xids - runing.xids = new.xids
 446     newslicesxids = Set(live.keys()) - Set(slices.keys())
 447     for newslicexid in newslicesxids:
 448         # Delegated slices dont have xids (which are uids) since they haven't been
 449         # instantiated yet.
 450         if newslicexid != None and db[live[newslicexid]].has_key('_rspec') == True:
 451             logger.log("bwmon: New Slice %s" % live[newslicexid])
 452             # _rspec is the computed rspec:  NM retrieved data from PLC, computed loans
 453             # and made a dict of computed values.
 454             rspec = db[live[newslicexid]]['_rspec']
 455             slices[newslicexid] = Slice(newslicexid, live[newslicexid], rspec)
 456             slices[newslicexid].reset(0, 0, 0, 0, rspec)
 457         else:
 458             logger.log("bwmon  Slice %s doesn't have xid.  Must be delegated.  Skipping." % live[newslicexid])
 459     # Get actual running values from tc.
 460     # Update slice totals and bandwidth.
 461     for params in bwlimit.get():
 462         (xid, share,
 463          minrate, maxrate,
 464          minexemptrate, maxexemptrate,
 465          usedbytes, usedi2bytes) = params
 466
 467         # Ignore root and default buckets
 468         if xid == root_xid or xid == default_xid:
 469             continue
 470
 471         name = bwlimit.get_slice(xid)
 472         if name is None:
 473             # Orphaned (not associated with a slice) class
 474             name = "%d?" % xid
 475             bwlimit.off(xid)
 476
 477         # Monitor only the specified slices
 478         if names and name not in names:
 479             continue
 480         #slices is populated from the pickle file
 481         #xid is populated from bwlimit (read from /etc/passwd)
 482         if slices.has_key(xid):
 483             slice = slices[xid]
 484             if time.time() >= (slice.time + period) or \
 485                usedbytes < slice.bytes or usedi2bytes < slice.i2bytes:
 486                 # Reset to defaults every 24 hours or if it appears
 487                 # that the byte counters have overflowed (or, more
 488                 # likely, the node was restarted or the HTB buckets
 489                 # were re-initialized).
 490                 slice.reset(maxrate, \
 491                     maxexemptrate, \
 492                     usedbytes, \
 493                     usedi2bytes, \
 494                     db[slice.name]['_rspec'])
 495             else:
 496                 # Update byte counts
 497                 slice.update(maxrate, \
 498                     maxexemptrate, \
 499                     usedbytes, \
 500                     usedi2bytes, \
 501                     db[slice.name]['_rspec'])
 502         else:
 503             # Just in case.  Probably (hopefully) this will never happen.
 504             # New slice, initialize state
 505             logger.log("bwmon: New Slice %s" % name)
 506             slice = slices[xid] = Slice(xid, name, db[slice.name]['_rspec'])
 507             slice.reset(maxrate, \
 508                 maxexemptrate, \
 509                 usedbytes, \
 510                 usedi2bytes, \
 511                 db[slice.name]['_rspec'])
 512
 513     # Delete dead slices
 514     dead = Set(slices.keys()) - Set(live.keys())
 515     for xid in dead:
 516         if xid == root_xid or xid == default_xid:
 517             continue
 518         del slices[xid]
 519         bwlimit.off(xid)
 520
 521     logger.log("bwmon:  Saving %s" % datafile)
 522     f = open(datafile, "w")
 523     pickle.dump((version, slices), f)
 524     f.close()
 525
 526
 527 #def GetSlivers(data):
 528 #   for sliver in data['slivers']:
 529 #       if sliver.has_key('attributes'):
 530 #          print sliver
 531 #           for attribute in sliver['attributes']:
 532 #               if attribute['name'] == "KByteThresh": print attribute['value']
 533
 534 #def start(options, config):
 535 #    pass