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.18 2007/04/25 22:19:59 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             if verbose:
 321                 logger.log("bwmon: %s over thresh %s" \
 322                   % (self.name, format_bytes(self.ThreshKByte * 1024)))
 323             sum = self.bytes + (self.ThreshKByte * 1024)
 324             maxbyte = self.MaxKByte * 1024
 325             bytesused = usedbytes - self.bytes
 326             timeused = int(time.time() - self.time)
 327             new_maxrate = int(((maxbyte - bytesused) * 8)/(period - timeused))
 328             if new_maxrate < (self.MinRate * 1000):
 329                 new_maxrate = self.MinRate * 1000
 330         else:
 331             new_maxrate = self.MaxRate * 1000
 332
 333         # Format template parameters for low bandwidth message
 334         params['class'] = "low bandwidth"
 335         params['bytes'] = format_bytes(usedbytes - self.bytes)
 336         params['limit'] = format_bytes(self.MaxKByte * 1024)
 337         params['thresh'] = format_bytes(self.ThreshKByte * 1024)
 338         params['new_maxrate'] = bwlimit.format_tc_rate(new_maxrate)
 339
 340         if verbose:
 341             logger.log("bwmon:  %(slice)s %(class)s " \
 342                   "%(bytes)s of %(limit)s max %(thresh)s thresh (%(new_maxrate)s/s maxrate)" % \
 343                   params)
 344
 345         # Cap low bandwidth burst rate
 346         if new_maxrate != runningmaxrate:
 347             message += template % params
 348             logger.log("bwmon:   ** %(slice)s %(class)s capped at %(new_maxrate)s/s " % params)
 349
 350         if usedi2bytes >= (self.i2bytes + (self.Threshi2KByte * 1024)):
 351             maxi2byte = self.Maxi2KByte * 1024
 352             i2bytesused = usedi2bytes - self.i2bytes
 353             timeused = int(time.time() - self.time)
 354             new_maxi2rate = int(((maxi2byte - i2bytesused) * 8)/(period - timeused))
 355             if new_maxi2rate < (self.Mini2Rate * 1000):
 356                 new_maxi2rate = self.Mini2Rate * 1000
 357         else:
 358             new_maxi2rate = self.Maxi2Rate * 1000
 359
 360         # Format template parameters for high bandwidth message
 361         params['class'] = "high bandwidth"
 362         params['bytes'] = format_bytes(usedi2bytes - self.i2bytes)
 363         params['limit'] = format_bytes(self.Maxi2KByte * 1024)
 364         params['new_maxexemptrate'] = bwlimit.format_tc_rate(new_maxi2rate)
 365
 366         if verbose:
 367             logger.log("bwmon:  %(slice)s %(class)s " \
 368                   "%(bytes)s of %(limit)s (%(new_maxrate)s/s maxrate)" % params)
 369
 370         # Cap high bandwidth burst rate
 371         if new_maxi2rate != runningmaxi2rate:
 372             message += template % params
 373             logger.log("bwmon:  %(slice)s %(class)s capped at %(new_maxexemptrate)s/s" % params)
 374
 375         # Apply parameters
 376         if new_maxrate != runningmaxrate or new_maxi2rate != runningmaxi2rate:
 377             bwlimit.set(xid = self.xid, maxrate = new_maxrate, maxexemptrate = new_maxi2rate)
 378
 379         # Notify slice
 380         if message and self.emailed == False:
 381             subject = "pl_mom capped bandwidth of slice %(slice)s on %(hostname)s" % params
 382             if debug:
 383                 logger.log("bwmon:  "+ subject)
 384                 logger.log("bwmon:  "+ message + (footer % params))
 385             else:
 386                 self.emailed = True
 387                 slicemail(self.name, subject, message + (footer % params))
 388
 389 def GetSlivers(db):
 390     # Defaults
 391     global datafile, \
 392         period, \
 393         default_MaxRate, \
 394         default_Maxi2Rate, \
 395         default_MinRate, \
 396         default_MaxKByte,\
 397         default_ThreshKByte,\
 398         default_Maxi2KByte,\
 399         default_Threshi2KByte,\
 400         default_Share,\
 401         verbose
 402
 403     # All slices
 404     names = []
 405
 406     # Incase the limits have changed.
 407     default_MaxRate = int(bwlimit.get_bwcap() / 1000)
 408     default_Maxi2Rate = int(bwlimit.bwmax / 1000)
 409
 410     # Incase default isn't set yet.
 411     if default_MaxRate == -1:
 412         default_MaxRate = 1000000
 413
 414     try:
 415         f = open(datafile, "r+")
 416         logger.log("bwmon:  Loading %s" % datafile)
 417         (version, slices) = pickle.load(f)
 418         f.close()
 419         # Check version of data file
 420         if version != "$Id: bwmon.py,v 1.18 2007/04/25 22:19:59 faiyaza Exp $":
 421             logger.log("bwmon:  Not using old version '%s' data file %s" % (version, datafile))
 422             raise Exception
 423     except Exception:
 424         version = "$Id: bwmon.py,v 1.18 2007/04/25 22:19:59 faiyaza Exp $"
 425         slices = {}
 426
 427     # Get/set special slice IDs
 428     root_xid = bwlimit.get_xid("root")
 429     default_xid = bwlimit.get_xid("default")
 430
 431     # Since root is required for sanity, its not in the API/plc database, so pass {}
 432     # to use defaults.
 433     if root_xid not in slices.keys():
 434         slices[root_xid] = Slice(root_xid, "root", {})
 435         slices[root_xid].reset(0, 0, 0, 0, {})
 436
 437     # Used by bwlimit.  pass {} since there is no rspec (like above).
 438     if default_xid not in slices.keys():
 439         slices[default_xid] = Slice(default_xid, "default", {})
 440         slices[default_xid].reset(0, 0, 0, 0, {})
 441
 442     live = {}
 443     # Get running slivers that should be on this node (from plc). {xid: name}
 444     for sliver in db.keys():
 445         live[bwlimit.get_xid(sliver)] = sliver
 446
 447     # Setup new slices.
 448     # live.xids - runing(slices).xids = new.xids
 449     newslicesxids = []
 450     for plcxid in live.keys():
 451         if plcxid not in slices.keys():
 452             newslicesxids.append(plcxid)
 453
 454     #newslicesxids = Set(live.keys()) - Set(slices.keys())
 455     for newslicexid in newslicesxids:
 456         # Delegated slices dont have xids (which are uids) since they haven't been
 457         # instantiated yet.
 458         if newslicexid != None and db[live[newslicexid]].has_key('_rspec') == True:
 459             logger.log("bwmon: New Slice %s" % live[newslicexid])
 460             # _rspec is the computed rspec:  NM retrieved data from PLC, computed loans
 461             # and made a dict of computed values.
 462             rspec = db[live[newslicexid]]['_rspec']
 463             slices[newslicexid] = Slice(newslicexid, live[newslicexid], rspec)
 464             slices[newslicexid].reset(0, 0, 0, 0, rspec)
 465         else:
 466             logger.log("bwmon  Slice %s doesn't have xid.  Must be delegated.  Skipping." % live[newslicexid])
 467
 468     # ...mlhuang's abortion....
 469     # Get actual running values from tc.
 470     # Update slice totals and bandwidth.
 471     for params in bwlimit.get():
 472         (xid, share,
 473          minrate, maxrate,
 474          minexemptrate, maxexemptrate,
 475          usedbytes, usedi2bytes) = params
 476
 477         # Ignore root and default buckets
 478         if xid == root_xid or xid == default_xid:
 479             continue
 480
 481         name = bwlimit.get_slice(xid)
 482         if name is None:
 483             # Orphaned (not associated with a slice) class
 484             name = "%d?" % xid
 485             bwlimit.off(xid)
 486
 487         # Monitor only the specified slices
 488         if names and name not in names:
 489             continue
 490         #slices is populated from the pickle file
 491         #xid is populated from bwlimit (read from /etc/passwd)
 492         if slices.has_key(xid):
 493             slice = slices[xid]
 494             # Old slices werent being instanciated correctly because
 495             # the HTBs were still pleasent, but the slice in bwmon would
 496             # have the byte counts set to 0.  The next time update was run
 497             # the real byte count would be sent to update, causing the bw cap.
 498
 499             if time.time() >= (slice.time + period) or \
 500                usedbytes < slice.bytes or \
 501                usedi2bytes < slice.i2bytes or \
 502                xid in newslicesxids:
 503                 # Reset to defaults every 24 hours or if it appears
 504                 # that the byte counters have overflowed (or, more
 505                 # likely, the node was restarted or the HTB buckets
 506                 # were re-initialized).
 507                 slice.reset(maxrate, \
 508                     maxexemptrate, \
 509                     usedbytes, \
 510                     usedi2bytes, \
 511                     db[slice.name]['_rspec'])
 512             else:
 513                 # Update byte counts
 514                 slice.update(maxrate, \
 515                     maxexemptrate, \
 516                     usedbytes, \
 517                     usedi2bytes, \
 518                     db[slice.name]['_rspec'])
 519         else:
 520             # Just in case.  Probably (hopefully) this will never happen.
 521             # New slice, initialize state
 522             logger.log("bwmon: Deleting orphaned slice xid %s" % xid)
 523             bwlimit.off(xid)
 524
 525     # Delete dead slices
 526     dead = Set(slices.keys()) - Set(live.keys())
 527     for xid in dead:
 528         if xid == root_xid or xid == default_xid:
 529             continue
 530         del slices[xid]
 531         bwlimit.off(xid)
 532
 533     logger.log("bwmon:  Saving %s" % datafile)
 534     f = open(datafile, "w")
 535     pickle.dump((version, slices), f)
 536     f.close()
 537
 538
 539 #def GetSlivers(data):
 540 #   for sliver in data['slivers']:
 541 #       if sliver.has_key('attributes'):
 542 #          print sliver
 543 #           for attribute in sliver['attributes']:
 544 #               if attribute['name'] == "KByteThresh": print attribute['value']
 545
 546 #def start(options, config):
 547 #    pass