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.6 2006/07/10 15:19:35 faiyaza Exp $
  19 #
  20
  21 import syslog
  22 import os
  23 import sys
  24 import getopt
  25 import time
  26 import pickle
  27
  28 import socket
  29 import xmlrpclib
  30 import bwlimit
  31
  32 from sets import Set
  33
  34 # Utility functions
  35 from pl_mom import *
  36
  37 # Constants
  38 seconds_per_day = 24 * 60 * 60
  39 bits_per_byte = 8
  40
  41 # Defaults
  42 debug = False
  43 verbose = 0
  44 datafile = "/var/lib/misc/bwmon.dat"
  45 nm = None
  46
  47 default_maxrate = bwlimit.get_bwcap()
  48
  49 default_maxexemptrate = bwlimit.bwmax
  50
  51 # 500 Kbit or 5.4 GB per day
  52 default_avgrate = 500000
  53
  54 # 1.5 Mbit or 16.4 GB per day
  55 default_avgexemptrate = 1500000
  56
  57 # Average over 1 day
  58 period = 1 * seconds_per_day
  59
  60 # Message template
  61 template = \
  62 """
  63 The slice %(slice)s has transmitted more than %(bytes)s from
  64 %(hostname)s to %(class)s destinations
  65 since %(since)s.
  66
  67 Its maximum %(class)s burst rate will be capped at %(avgrate)s
  68 until %(until)s.
  69
  70 Please reduce the average %(class)s transmission rate
  71 of the slice to %(avgrate)s, or %(limit)s per %(period)s.
  72
  73 """.lstrip()
  74
  75 footer = \
  76 """
  77 %(date)s %(hostname)s bwcap %(slice)s
  78 """.lstrip()
  79
  80 class Slice:
  81     """
  82     Stores the last recorded bandwidth parameters of a slice.
  83
  84     xid - slice context/VServer ID
  85     name - slice name
  86     time - beginning of recording period in UNIX seconds
  87     bytes - low bandwidth bytes transmitted at the beginning of the recording period
  88     exemptbytes - high bandwidth bytes transmitted at the beginning of the recording period (for I2 -F)
  89     avgrate - average low bandwidth rate to enforce over the recording period
  90     avgexemptrate - average high bandwidth rate to enforce over the recording period (for I2 -F)
  91     last_avgrate - last recorded avgrate from NM
  92     last_maxrate - last recorded maxrate from NM
  93     last_avgexemptrate - last recorded avgexemptrate from NM
  94     last_maxexemptrate - last recorded maxexemptrate from NM
  95     """
  96
  97     def __init__(self, xid, name, maxrate, maxexemptrate, bytes, exemptbytes):
  98         self.xid = xid
  99         self.name = name
 100         self._maxrate = default_maxrate
 101         self.last_maxrate = default_maxrate
 102         self.avgrate = default_avgrate
 103         self.last_avgrate = default_avgrate
 104         self.avgexemptrate = default_avgexemptrate
 105         self.last_avgexemptrate = default_avgexemptrate
 106         self.maxexemptrate = default_maxexemptrate
 107         self.last_maxexemptrate = default_maxexemptrate
 108         self.reset(maxrate, maxexemptrate, bytes, exemptbytes)
 109
 110     def __repr__(self):
 111         return self.name
 112
 113     def reset(self, maxrate, maxexemptrate, bytes, exemptbytes):
 114         """
 115         Begin a new recording period. Remove caps by restoring limits
 116         to their default values.
 117         """
 118
 119         # Reset baseline time
 120         self.time = time.time()
 121
 122         # Reset baseline byte coutns
 123         self.bytes = bytes
 124         self.exemptbytes = exemptbytes
 125
 126         #If NM except"ns below, and new_max* doesn't get set, use last.
 127         new_maxrate = self.last_maxrate
 128         new_maxexemptrate = self.last_maxexemptrate
 129
 130         # Query Node Manager for max rate overrides
 131         try:
 132                 vals = nm.query(self.name, [('nm_net_max_rate', self.last_maxrate),
 133                                 ('nm_net_max_exempt_rate', self.last_maxexemptrate),
 134                                 ('nm_net_avg_rate', self.last_avgrate),
 135                                 ('nm_net_avg_exempt_rate', self.last_avgexemptrate)])
 136                 (new_maxrate, new_maxexemptrate,
 137                         self.last_avgrate, self.last_avgexemptrate) = vals
 138                 #If NM is alive, and there is a cap, update new
 139                 self.last_maxrate = new_maxrate
 140                 self.last_maxexemptrate = new_maxexemptrate
 141
 142         except Exception, err:
 143                 print "Warning: Exception received while querying NM:", err
 144
 145         if new_maxrate != maxrate or new_maxexemptrate != maxexemptrate:
 146             print "%s reset to %s/%s" % \
 147                   (self.name,
 148                    bwlimit.format_tc_rate(new_maxrate),
 149                    bwlimit.format_tc_rate(new_maxexemptrate))
 150             bwlimit.set(xid = self.xid, maxrate = new_maxrate, maxexemptrate = new_maxexemptrate)
 151
 152     def update(self, maxrate, maxexemptrate, bytes, exemptbytes):
 153         """
 154         Update byte counts and check if average rates have been
 155         exceeded. In the worst case (instantaneous usage of the entire
 156         average daily byte limit at the beginning of the recording
 157         period), the slice will be immediately capped and will get to
 158         send twice the average daily byte limit. In the common case,
 159         it will get to send slightly more than the average daily byte
 160         limit.
 161         """
 162
 163         # Query Node Manager for max average rate overrides
 164         try:
 165                 (self.avgrate, self.avgexemptrate) = nm.query(self.name,
 166                         [('nm_net_avg_rate', self.last_avgrate),
 167                         ('nm_net_avg_exempt_rate', self.last_avgexemptrate)])
 168                 #If NM is alive, and there is a cap, update new
 169                 self.last_avgexemptrate = self.avgexemptrate
 170                 self.last_avgrate = self.avgrate
 171         except Exception, err:
 172                 print "Warning: Exception received while querying NM:", err
 173
 174         # Prepare message parameters from the template
 175         message = ""
 176         params = {'slice': self.name, 'hostname': socket.gethostname(),
 177                   'since': time.asctime(time.gmtime(self.time)) + " GMT",
 178                   'until': time.asctime(time.gmtime(self.time + period)) + " GMT",
 179                   'date': time.asctime(time.gmtime()) + " GMT",
 180                   'period': format_period(period)}
 181
 182         bytelimit = self.avgrate * period / bits_per_byte
 183         if bytes >= (self.bytes + bytelimit) and \
 184            maxrate > self.avgrate:
 185             new_maxrate = self.avgrate
 186         else:
 187             new_maxrate = maxrate
 188
 189         # Format template parameters for low bandwidth message
 190         params['class'] = "low bandwidth"
 191         params['bytes'] = format_bytes(bytes - self.bytes)
 192         params['maxrate'] = bwlimit.format_tc_rate(maxrate)
 193         params['limit'] = format_bytes(bytelimit)
 194         params['avgrate'] = bwlimit.format_tc_rate(self.avgrate)
 195
 196         if verbose:
 197             print "%(slice)s %(class)s " \
 198                   "%(bytes)s, %(limit)s (%(maxrate)s max/%(avgrate)s avg)" % \
 199                   params
 200
 201         # Cap low bandwidth burst rate
 202         if new_maxrate != maxrate:
 203             message += template % params
 204             print "%(slice)s %(class)s capped at %(avgrate)s (%(bytes)s/%(limit)s)" % params
 205
 206         exemptbytelimit = self.avgexemptrate * period / bits_per_byte
 207         if exemptbytes >= (self.exemptbytes + exemptbytelimit) and \
 208            maxexemptrate > self.avgexemptrate:
 209             new_maxexemptrate = self.avgexemptrate
 210         else:
 211             new_maxexemptrate = maxexemptrate
 212
 213         # Format template parameters for high bandwidth message
 214         params['class'] = "high bandwidth"
 215         params['bytes'] = format_bytes(exemptbytes - self.exemptbytes)
 216         params['maxrate'] = bwlimit.format_tc_rate(maxexemptrate)
 217         params['limit'] = format_bytes(exemptbytelimit)
 218         params['avgrate'] = bwlimit.format_tc_rate(self.avgexemptrate)
 219
 220         if verbose:
 221             print "%(slice)s %(class)s " \
 222                   "%(bytes)s, %(limit)s (%(maxrate)s max /%(avgrate)s avg)" % \
 223                   params
 224
 225         # Cap high bandwidth burst rate
 226         if new_maxexemptrate != maxexemptrate:
 227             message += template % params
 228             print "%(slice)s %(class)s capped at %(avgrate)s (%(bytes)s/%(limit)s)" % params
 229
 230         # Apply parameters
 231         if new_maxrate != maxrate or new_maxexemptrate != maxexemptrate:
 232             bwlimit.set(xid = self.xid, maxrate = new_maxrate, maxexemptrate = new_maxexemptrate)
 233
 234         # Notify slice
 235         if message:
 236             subject = "pl_mom capped bandwidth of slice %(slice)s on %(hostname)s" % params
 237             if debug:
 238                 print subject
 239                 print message + (footer % params)
 240             else:
 241                 slicemail(self.name, subject, message + (footer % params))
 242
 243 def usage():
 244     print """
 245 Usage: %s [OPTIONS]...
 246
 247 Options:
 248         -d, --debug             Enable debugging (default: %s)
 249         -v, --verbose           Increase verbosity level (default: %d)
 250         -f, --file=FILE         Data file (default: %s)
 251         -s, --slice=SLICE       Constrain monitoring to these slices (default: all)
 252         -p, --period=SECONDS    Interval in seconds over which to enforce average byte limits (default: %s)
 253         -h, --help              This message
 254 """.lstrip() % (sys.argv[0], debug, verbose, datafile, format_period(period))
 255
 256 def main():
 257     # Defaults
 258     global debug, verbose, datafile, period, nm
 259     # All slices
 260     names = []
 261
 262     try:
 263         longopts = ["debug", "verbose", "file=", "slice=", "period=", "help"]
 264         (opts, argv) = getopt.getopt(sys.argv[1:], "dvf:s:p:h", longopts)
 265     except getopt.GetoptError, err:
 266         print "Error: " + err.msg
 267         usage()
 268         sys.exit(1)
 269
 270     for (opt, optval) in opts:
 271         if opt == "-d" or opt == "--debug":
 272             debug = True
 273         elif opt == "-v" or opt == "--verbose":
 274             verbose += 1
 275             bwlimit.verbose = verbose - 1
 276         elif opt == "-f" or opt == "--file":
 277             datafile = optval
 278         elif opt == "-s" or opt == "--slice":
 279             names.append(optval)
 280         elif opt == "-p" or opt == "--period":
 281             period = int(optval)
 282         else:
 283             usage()
 284             sys.exit(0)
 285
 286     # Check if we are already running
 287     writepid("bwmon")
 288
 289     if not debug:
 290         # Redirect stdout and stderr to syslog
 291         syslog.openlog("bwmon")
 292         sys.stdout = sys.stderr = Logger()
 293
 294     try:
 295         f = open(datafile, "r+")
 296         if verbose:
 297             print "Loading %s" % datafile
 298         (version, slices) = pickle.load(f)
 299         f.close()
 300         # Check version of data file
 301         if version != "$Id: bwmon.py,v 1.6 2006/07/10 15:19:35 faiyaza Exp $":
 302             print "Not using old version '%s' data file %s" % (version, datafile)
 303             raise Exception
 304     except Exception:
 305         version = "$Id: bwmon.py,v 1.6 2006/07/10 15:19:35 faiyaza Exp $"
 306         slices = {}
 307
 308     # Get special slice IDs
 309     root_xid = bwlimit.get_xid("root")
 310     default_xid = bwlimit.get_xid("default")
 311
 312     #Open connection to Node Manager
 313     nm = NM()
 314
 315     live = []
 316     for params in bwlimit.get():
 317         (xid, share,
 318          minrate, maxrate,
 319          minexemptrate, maxexemptrate,
 320          bytes, exemptbytes) = params
 321         live.append(xid)
 322
 323         # Ignore root and default buckets
 324         if xid == root_xid or xid == default_xid:
 325             continue
 326
 327         name = bwlimit.get_slice(xid)
 328         if name is None:
 329             # Orphaned (not associated with a slice) class
 330             name = "%d?" % xid
 331
 332         # Monitor only the specified slices
 333         if names and name not in names:
 334             continue
 335
 336         #slices is populated from the pickle file
 337         #xid is populated from bwlimit (read from /etc/passwd)
 338         if slices.has_key(xid):
 339             slice = slices[xid]
 340             if time.time() >= (slice.time + period) or \
 341                bytes < slice.bytes or exemptbytes < slice.exemptbytes:
 342                 # Reset to defaults every 24 hours or if it appears
 343                 # that the byte counters have overflowed (or, more
 344                 # likely, the node was restarted or the HTB buckets
 345                 # were re-initialized).
 346                 slice.reset(maxrate, maxexemptrate, bytes, exemptbytes)
 347             else:
 348                 # Update byte counts
 349                 slice.update(maxrate, maxexemptrate, bytes, exemptbytes)
 350         else:
 351             # New slice, initialize state
 352             slice = slices[xid] = Slice(xid, name, maxrate, maxexemptrate, bytes, exemptbytes)
 353
 354     # Delete dead slices
 355     dead = Set(slices.keys()) - Set(live)
 356     for xid in dead:
 357         del slices[xid]
 358
 359     if verbose:
 360         print "Saving %s" % datafile
 361     f = open(datafile, "w")
 362     pickle.dump((version, slices), f)
 363     f.close()
 364
 365     removepid("bwmon")
 366
 367 if __name__ == '__main__':
 368     main()