bwmon/monitor.py

   1 # -*- coding: utf-8 -*-
   2 #
   3 # Copyright 2010 Thomas Perl and Stefan Kögl. All rights reserved.
   4 #
   5 # Developed for a practical course (Large-scaled distributed computing) at the
   6 # University of Technology Vienna in the 2010 summer term.
   7 #
   8 # Redistribution and use in source and binary forms, with or without
   9 # modification, are permitted provided that the following conditions are met:
  10 #
  11 #    1. Redistributions of source code must retain the above copyright notice,
  12 #       this list of conditions and the following disclaimer.
  13 #
  14 #    2. Redistributions in binary form must reproduce the above copyright notice,
  15 #       this list of conditions and the following disclaimer in the documentation
  16 #       and/or other materials provided with the distribution.
  17 #
  18 # THIS SOFTWARE IS PROVIDED BY THE AUTHORS ``AS IS'' AND ANY EXPRESS OR IMPLIED
  19 # WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  20 # MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
  21 # EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
  22 # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
  24 # PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
  25 # LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
  26 # OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
  27 # ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  28 #
  29
  30 from __future__ import absolute_import
  31
  32 from bwmon import proc
  33 from bwmon import util
  34 from bwmon import model
  35
  36 import collections
  37 import time
  38 import sys
  39 import copy
  40 import re
  41
  42 BANDWIDTH, TRAFFIC = range(2)
  43
  44 class Monitor(object):
  45     """The ip_conntrack-based, system-wide bandwidth monitor
  46
  47     This object implements the higher-level management functions
  48     for the ip_conntrack-based monitoring method (system-wide,
  49     non-shaping, Linux-specific).
  50
  51     The monitor has a default update frequency of 1 second.
  52     """
  53     DEFAULT_UPDATE_FREQUENCY = 1
  54
  55     def __init__(self, lookback=True, ignorelocal=False):
  56         """Create a new Monitor object
  57
  58         This initializes the monitor object. In case the
  59         lookback value is False, this call already takes
  60         the first measurement from ip_conntrack.
  61
  62         @param lookback: indicates if data already existing in ip_conntrack should be considered (True) or ignored (False)
  63         @param ignorelocal: indicates if the Monitor should ignore loopback traffic (True) or include it in the calculations (False)
  64         """
  65         self.fd_map = {}
  66         self.sample_time = time.time()
  67         self.conntrack = {}
  68         self.last_conntrack = {}
  69         self.init_conntrack = {} if lookback else proc.parse_ip_conntrack()
  70         self.connections = {}
  71         self.update_frequency = self.DEFAULT_UPDATE_FREQUENCY
  72         self.entries = model.MonitorEntryCollection(self.update_frequency)
  73         self.include_filter = []
  74         self.exclude_filter = []
  75         self.ignorelocal = ignorelocal
  76
  77     def update(self, entry_collection):
  78         """Update this monitor from a MonitorEntryCollection
  79
  80         This function gets a file descriptor to process name
  81         mapping, and re-parses the ip_conntrack data. The
  82         current time is saved as sample time, and the data is
  83         then converted into an aggregated monitoring value.
  84
  85         @param entry_collection: The collection from where to take data
  86         """
  87         self.fd_map.update(proc.get_fd_map())
  88         self.last_conntrack = copy.deepcopy(self.conntrack)
  89         self.conntrack.update( self.sub_conntrack(proc.parse_ip_conntrack(), self.init_conntrack) )
  90         self.connections.update(proc.get_connections())
  91         entry_collection.expire()
  92         self.sample_time = time.time()
  93         self.convert(entry_collection)
  94
  95     def sub_conntrack(self, conntrack_current, conntrack_init):
  96         """Subtract inital conntrack data from a measurement
  97
  98         @param conntrack_current: The current measurement
  99         @param conntrack_init: The initally measured data
 100         @return: The converted conntrack data
 101         """
 102         for (k, v_init) in conntrack_init.iteritems():
 103             if k in conntrack_current:
 104                 v_current = conntrack_current[k]
 105                 v_current['bytes'] = int(v_current['bytes']) - int(v_init['bytes'])
 106                 v_current['packets'] = int(v_current['packets']) - int(v_init['packets'])
 107                 conntrack_current[k] = v_current
 108
 109         return conntrack_current
 110
 111     def set_filter(self, include_filter, exclude_filter):
 112         """Apply inclusive and exclusive filters on this Monitor
 113
 114         This function takes two lists of regular expression strings,
 115         and will compile them into regular expression objects. The
 116         filters will be used to exclude and include commands from
 117         the monitoring output (see the config file docs for details).
 118
 119         @param include_filter: A list of regular expressions to include
 120         @param exclude_filter: A list of regular expressions to exclude
 121         """
 122         self.include_filter = [re.compile(f) for f in include_filter] if include_filter else []
 123         self.exclude_filter = [re.compile(f) for f in exclude_filter] if exclude_filter else []
 124
 125     def convert(self, entry_collection):
 126         """Apply a per-process merge of a MonitorEntryCollection
 127
 128         This function takes a MonitorEntryCollection object
 129         and merges new connections into it.
 130
 131         @param entry_collection: A MonitorEntryCollection object
 132         """
 133         entries = collections.defaultdict(lambda: (0, 0))
 134         for con in self.connections.itervalues():
 135             inode = con.get('inode', None)
 136             process = self.fd_map.get(inode, None)
 137
 138             if process is None:
 139                 continue
 140
 141             if self.include_filter and not any([f.search(process['cmd']) for f in self.include_filter]):
 142                 continue
 143
 144             if self.exclude_filter and any([f.search(process['cmd']) for f in self.exclude_filter]):
 145                 continue
 146
 147             if self.ignorelocal and islocal(con['remote']) and islocal(con['local']):
 148                 continue
 149
 150             key_in  = proc.ip_hash(con['remote'], con['local'])
 151             key_out = proc.ip_hash(con['local'], con['remote'])
 152             keys = {'in': key_in, 'out': key_out}
 153             new_byte = {'in': 0, 'out': 0}
 154
 155             for direction in ('in', 'out'):
 156                 k = keys[direction]
 157                 if k in self.conntrack:
 158                     try:
 159                         if key_in in self.last_conntrack:
 160                             diff = (int(self.conntrack[k]['bytes']) -
 161                                     int(self.last_conntrack[k]['bytes']))
 162                         else:
 163                             diff = int(self.conntrack[k]['bytes'])
 164                     except KeyError:
 165                         print >>sys.stderr, ("WARNING: No 'bytes' field in output (use '"
 166                                 "sysctl -w net.netfilter.nf_conntrack_acct=1"
 167                                 "' to enable accounting.")
 168                         sys.exit(1)
 169
 170                     new_byte[direction] = diff
 171
 172             current_in, current_out = entries[process['cmd']]
 173             new_in, new_out = (new_byte['in'], new_byte['out'])
 174
 175             entries[process['cmd']] = (current_in + new_in, current_out + new_out)
 176
 177         for key in entries:
 178             new_in, new_out = entries[key]
 179             old_in, old_out, timestamp = entry_collection.get_last_bytes(key)
 180             entry = model.MonitorEntry(key, old_in + new_in, old_out + new_out, self.sample_time)
 181             entry_collection.add(entry)
 182
 183     def output(self, mode=TRAFFIC):
 184         """Print the current status to standard output
 185
 186         @param mode: Monitoring mode (BANDWIDTH or TRAFFIC [=default])
 187         """
 188         util.clear()
 189         if mode == TRAFFIC:
 190             entries = sorted(self.entries.get_traffic())
 191         else:
 192             entries = sorted(self.entries.get_usage())
 193
 194         for bytes_in, bytes_out, cmd in entries:
 195             if bytes_in or bytes_out:
 196                 if len(cmd) > 60:
 197                     cmd = cmd[:57] + '...'
 198                 print '%10d / %10d -- %s' % (bytes_in, bytes_out, cmd)
 199                 sys.stdout.flush()
 200
 201     def loop(self, mode):
 202         """The mainloop of a standalone monitor
 203
 204         @param mode: Monitoring mode (BANDWIDTH or TRAFFIC [=default])
 205         """
 206         while True:
 207             self.update(self.entries)
 208             self.output(mode)
 209             time.sleep(self.update_frequency)
 210
 211     def close(self):
 212         """Close this bandwidth monitor"""
 213         pass
 214
 215
 216 def islocal(ip):
 217     """Check if an IP is the local host
 218
 219     @return: True if the IP is in the loopback interface
 220     """
 221     return ip.startswith('127.0.0.') or ip.startswith('0.0.0.0')
 222