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         # Get attributes
 213
 214         # Sanity check plus policy decision for MinRate:
 215         # Minrate cant be greater than 25% of MaxRate or NodeCap.
 216         MinRate = int(rspec.get("net_min_rate", default_MinRate))
 217         if MinRate > int(.25 * default_MaxRate):
 218             MinRate = int(.25 * default_MaxRate)
 219         if MinRate != self.MinRate:
 220             self.MinRate = MinRate
 221             logger.log("bwmon:  Updating %s: Min Rate = %s" %(self.name, self.MinRate))
 222
 223         MaxRate = int(rspec.get('net_max_rate', bwlimit.get_bwcap() / 1000))
 224         if MaxRate != self.MaxRate:
 225             self.MaxRate = MaxRate
 226             logger.log("bwmon:  Updating %s: Max Rate = %s" %(self.name, self.MaxRate))
 227
 228         Mini2Rate = int(rspec.get('net_i2_min_rate', default_Mini2Rate))
 229         if Mini2Rate != self.Mini2Rate:
 230             self.Mini2Rate = Mini2Rate
 231             logger.log("bwmon:  Updating %s: Min i2 Rate = %s" %(self.name, self.Mini2Rate))
 232
 233         Maxi2Rate = int(rspec.get('net_i2_max_rate', bwlimit.bwmax / 1000))
 234         if Maxi2Rate != self.Maxi2Rate:
 235             self.Maxi2Rate = Maxi2Rate
 236             logger.log("bwmon:  Updating %s: Max i2 Rate = %s" %(self.name, self.Maxi2Rate))
 237
 238         MaxKByte = int(rspec.get('net_max_kbyte', default_MaxKByte))
 239         if MaxKByte != self.MaxKByte:
 240             self.MaxKByte = MaxKByte
 241             logger.log("bwmon:  Updating %s: Max KByte lim = %s" %(self.name, self.MaxKByte))
 242
 243         Maxi2KByte = int(rspec.get('net_i2_max_kbyte', default_Maxi2KByte))
 244         if Maxi2KByte != self.Maxi2KByte:
 245             self.Maxi2KByte = Maxi2KByte
 246             logger.log("bwmon:  Updating %s: Max i2 KByte = %s" %(self.name, self.Maxi2KByte))
 247
 248         ThreshKByte = int(rspec.get('net_thresh_kbyte', default_ThreshKByte))
 249         if ThreshKByte != self.ThreshKByte:
 250             self.ThreshKByte = ThreshKByte
 251             logger.log("bwmon:  Updating %s: Thresh KByte = %s" %(self.name, self.ThreshKByte))
 252
 253         Threshi2KByte = int(rspec.get('net_i2_thresh_kbyte', default_Threshi2KByte))
 254         if Threshi2KByte != self.Threshi2KByte:
 255             self.Threshi2KByte = Threshi2KByte
 256             logger.log("bwmon:  Updating %s: i2 Thresh KByte = %s" %(self.name, self.Threshi2KByte))
 257
 258         Share = int(rspec.get('net_share', default_Share))
 259         if Share != self.Share:
 260             self.Share = Share
 261             logger.log("bwmon:  Updating %s: Net Share = %s" %(self.name, self.Share))
 262
 263         Sharei2 = int(rspec.get('net_i2_share', default_Share))
 264         if Sharei2 != self.Sharei2:
 265             self.Sharei2 = Sharei2
 266             logger.log("bwmon:  Updating %s: Net i2 Share = %s" %(self.name, self.i2Share))
 267
 268
 269     def reset(self, runningmaxrate, runningmaxi2rate, usedbytes, usedi2bytes, rspec):
 270         """
 271         Begin a new recording period. Remove caps by restoring limits
 272         to their default values.
 273         """
 274
 275         # Query Node Manager for max rate overrides
 276         self.updateSliceAttributes(rspec)
 277
 278         # Reset baseline time
 279         self.time = time.time()
 280
 281         # Reset baseline byte coutns
 282         self.bytes = usedbytes
 283         self.i2bytes = usedi2bytes
 284
 285         # Reset email
 286         self.emailed = False
 287         maxrate = self.MaxRate * 1000
 288         maxi2rate = self.Maxi2Rate * 1000
 289         # Reset rates.
 290         if (self.MaxRate != runningmaxrate) or (self.Maxi2Rate != runningmaxi2rate):
 291             logger.log("bwmon:  %s reset to %s/%s" % \
 292                   (self.name,
 293                    bwlimit.format_tc_rate(maxrate),
 294                    bwlimit.format_tc_rate(maxi2rate)))
 295             bwlimit.set(xid = self.xid,
 296                 minrate = self.MinRate * 1000,
 297                 maxrate = self.MaxRate * 1000,
 298                 maxexemptrate = self.Maxi2Rate * 1000,
 299                 minexemptrate = self.Mini2Rate * 1000,
 300                 share = self.Share)
 301
 302     def update(self, runningmaxrate, runningmaxi2rate, usedbytes, usedi2bytes, rspec):
 303         """
 304         Update byte counts and check if byte limits have been
 305         exceeded.
 306         """
 307
 308         # Query Node Manager for max rate overrides
 309         self.updateSliceAttributes(rspec)
 310
 311         # Prepare message parameters from the template
 312         message = ""
 313         params = {'slice': self.name, 'hostname': socket.gethostname(),
 314                   'since': time.asctime(time.gmtime(self.time)) + " GMT",
 315                   'until': time.asctime(time.gmtime(self.time + period)) + " GMT",
 316                   'date': time.asctime(time.gmtime()) + " GMT",
 317                   'period': format_period(period)}
 318
 319         if usedbytes >= (self.bytes + (self.ThreshKByte * 1024)):
 320             sum = self.bytes + (self.ThreshKByte * 1024)
 321             maxbyte = self.MaxKByte * 1024
 322             bytesused = usedbytes - self.bytes
 323             timeused = int(time.time() - self.time)
 324             new_maxrate = int(((maxbyte - bytesused) * 8)/(period - timeused))
 325             if new_maxrate < (self.MinRate * 1000):
 326                 new_maxrate = self.MinRate * 1000
 327         else:
 328             new_maxrate = self.MaxRate * 1000
 329
 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         if verbose:
 337             logger.log("bwmon:  %(slice)s %(class)s " \
 338                   "%(bytes)s of %(limit)s (%(new_maxrate)s/s maxrate)" % \
 339                   params)
 340
 341         # Cap low bandwidth burst rate
 342         if new_maxrate != runningmaxrate:
 343             message += template % params
 344             logger.log("bwmon:   ** %(slice)s %(class)s capped at %(new_maxrate)s/s " % params)
 345
 346         if usedi2bytes >= (self.i2bytes + (self.Threshi2KByte * 1024)):
 347             maxi2byte = self.Maxi2KByte * 1024
 348             i2bytesused = usedi2bytes - self.i2bytes
 349             timeused = int(time.time() - self.time)
 350             new_maxi2rate = int(((maxi2byte - i2bytesused) * 8)/(period - timeused))
 351             if new_maxi2rate < (self.Mini2Rate * 1000):
 352                 new_maxi2rate = self.Mini2Rate * 1000
 353         else:
 354             new_maxi2rate = self.Maxi2Rate * 1000
 355
 356         # Format template parameters for high bandwidth message
 357         params['class'] = "high bandwidth"
 358         params['bytes'] = format_bytes(usedi2bytes - self.i2bytes)
 359         params['limit'] = format_bytes(self.Maxi2KByte * 1024)
 360         params['new_maxexemptrate'] = bwlimit.format_tc_rate(new_maxi2rate)
 361
 362         if verbose:
 363             logger.log("bwmon:  %(slice)s %(class)s " \
 364                   "%(bytes)s of %(limit)s (%(new_maxrate)s/s maxrate)" % params)
 365
 366         # Cap high bandwidth burst rate
 367         if new_maxi2rate != runningmaxi2rate:
 368             message += template % params
 369             logger.log("bwmon:  %(slice)s %(class)s capped at %(new_maxexemptrate)s/s" % params)
 370
 371         # Apply parameters
 372         if new_maxrate != runningmaxrate or new_maxi2rate != runningmaxi2rate:
 373             bwlimit.set(xid = self.xid, maxrate = new_maxrate, maxexemptrate = new_maxi2rate)
 374
 375         # Notify slice
 376         if message and self.emailed == False:
 377             subject = "pl_mom capped bandwidth of slice %(slice)s on %(hostname)s" % params
 378             if debug:
 379                 logger.log("bwmon:  "+ subject)
 380                 logger.log("bwmon:  "+ message + (footer % params))
 381             else:
 382                 self.emailed = True
 383                 slicemail(self.name, subject, message + (footer % params))
 384
 385 def GetSlivers(db):
 386     # Defaults
 387     global datafile, \
 388         period, \
 389         default_MaxRate, \
 390         default_Maxi2Rate, \
 391         default_MinRate, \
 392         default_MaxKByte,\
 393         default_ThreshKByte,\
 394         default_Maxi2KByte,\
 395         default_Threshi2KByte,\
 396         default_Share,\
 397         verbose
 398
 399     # All slices
 400     names = []
 401
 402     # Incase the limits have changed.
 403     default_MaxRate = int(bwlimit.get_bwcap() / 1000)
 404     default_Maxi2Rate = int(bwlimit.bwmax / 1000)
 405
 406     # Incase default isn't set yet.
 407     if default_MaxRate == -1:
 408         default_MaxRate = 1000000
 409
 410     try:
 411         f = open(datafile, "r+")
 412         logger.log("bwmon:  Loading %s" % datafile)
 413         (version, slices) = pickle.load(f)
 414         f.close()
 415         # Check version of data file
 416         if version != "$Id: bwmon.py,v 1.12 2007/03/06 20:46:54 faiyaza Exp $":
 417             logger.log("bwmon:  Not using old version '%s' data file %s" % (version, datafile))
 418             raise Exception
 419     except Exception:
 420         version = "$Id: bwmon.py,v 1.12 2007/03/06 20:46:54 faiyaza Exp $"
 421         slices = {}
 422
 423     # Get/set special slice IDs
 424     root_xid = bwlimit.get_xid("root")
 425     default_xid = bwlimit.get_xid("default")
 426
 427     # Since root is required for sanity, its not in the API/plc database, so pass {}
 428     # to use defaults.
 429     if root_xid not in slices.keys():
 430         slices[root_xid] = Slice(root_xid, "root", {})
 431         slices[root_xid].reset(0, 0, 0, 0, {})
 432
 433     # Used by bwlimit.  pass {} since there is no rspec (like above).
 434     if default_xid not in slices.keys():
 435         slices[default_xid] = Slice(default_xid, "default", {})
 436         slices[default_xid].reset(0, 0, 0, 0, {})
 437
 438     live = {}
 439     # Get running slivers that should be on this node (from plc). {xid: name}
 440     for sliver in db.keys():
 441         live[bwlimit.get_xid(sliver)] = sliver
 442
 443     # Setup new slices.
 444     # live.xids - runing.xids = new.xids
 445     newslicesxids = Set(live.keys()) - Set(slices.keys())
 446     for newslicexid in newslicesxids:
 447         # Delegated slices dont have xids (which are uids) since they haven't been
 448         # instantiated yet.
 449         if newslicexid != None and db[live[newslicexid]].has_key('_rspec') == True:
 450             logger.log("bwmon: New Slice %s" % live[newslicexid])
 451             # _rspec is the computed rspec:  NM retrieved data from PLC, computed loans
 452             # and made a dict of computed values.
 453             rspec = db[live[newslicexid]]['_rspec']
 454             slices[newslicexid] = Slice(newslicexid, live[newslicexid], rspec)
 455             slices[newslicexid].reset(0, 0, 0, 0, rspec)
 456         else:
 457             logger.log("bwmon  Slice %s doesn't have xid.  Must be delegated.  Skipping." % live[newslicexid])
 458     # Get actual running values from tc.
 459     # Update slice totals and bandwidth.
 460     for params in bwlimit.get():
 461         (xid, share,
 462          minrate, maxrate,
 463          minexemptrate, maxexemptrate,
 464          usedbytes, usedi2bytes) = params
 465
 466         # Ignore root and default buckets
 467         if xid == root_xid or xid == default_xid:
 468             continue
 469
 470         name = bwlimit.get_slice(xid)
 471         if name is None:
 472             # Orphaned (not associated with a slice) class
 473             name = "%d?" % xid
 474             bwlimit.off(xid)
 475
 476         # Monitor only the specified slices
 477         if names and name not in names:
 478             continue
 479         #slices is populated from the pickle file
 480         #xid is populated from bwlimit (read from /etc/passwd)
 481         if slices.has_key(xid):
 482             slice = slices[xid]
 483             if time.time() >= (slice.time + period) or \
 484                usedbytes < slice.bytes or usedi2bytes < slice.i2bytes:
 485                 # Reset to defaults every 24 hours or if it appears
 486                 # that the byte counters have overflowed (or, more
 487                 # likely, the node was restarted or the HTB buckets
 488                 # were re-initialized).
 489                 slice.reset(maxrate, \
 490                     maxexemptrate, \
 491                     usedbytes, \
 492                     usedi2bytes, \
 493                     db[slice.name]['_rspec'])
 494             else:
 495                 # Update byte counts
 496                 slice.update(maxrate, \
 497                     maxexemptrate, \
 498                     usedbytes, \
 499                     usedi2bytes, \
 500                     db[slice.name]['_rspec'])
 501         else:
 502             # Just in case.  Probably (hopefully) this will never happen.
 503             # New slice, initialize state
 504             logger.log("bwmon: New Slice %s" % name)
 505             slice = slices[xid] = Slice(xid, name, db[slice.name]['_rspec'])
 506             slice.reset(maxrate, \
 507                 maxexemptrate, \
 508                 usedbytes, \
 509                 usedi2bytes, \
 510                 db[slice.name]['_rspec'])
 511
 512     # Delete dead slices
 513     dead = Set(slices.keys()) - Set(live.keys())
 514     for xid in dead:
 515         if xid == root_xid or xid == default_xid:
 516             continue
 517         del slices[xid]
 518         bwlimit.off(xid)
 519
 520     logger.log("bwmon:  Saving %s" % datafile)
 521     f = open(datafile, "w")
 522     pickle.dump((version, slices), f)
 523     f.close()
 524
 525
 526 #def GetSlivers(data):
 527 #   for sliver in data['slivers']:
 528 #       if sliver.has_key('attributes'):
 529 #          print sliver
 530 #           for attribute in sliver['attributes']:
 531 #               if attribute['name'] == "KByteThresh": print attribute['value']
 532
 533 #def start(options, config):
 534 #    pass