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                     if key_in in self.last_conntrack:
 159                         new_byte[direction] = int(self.conntrack[k]['bytes']) - int(self.last_conntrack[k]['bytes'])
 160                     else:
 161                         new_byte[direction] = int(self.conntrack[k]['bytes'])
 162
 163             current_in, current_out = entries[process['cmd']]
 164             new_in, new_out = (new_byte['in'], new_byte['out'])
 165
 166             entries[process['cmd']] = (current_in + new_in, current_out + new_out)
 167
 168         for key in entries:
 169             new_in, new_out = entries[key]
 170             old_in, old_out, timestamp = entry_collection.get_last_bytes(key)
 171             entry = model.MonitorEntry(key, old_in + new_in, old_out + new_out, self.sample_time)
 172             entry_collection.add(entry)
 173
 174     def output(self, mode=TRAFFIC):
 175         """Print the current status to standard output
 176
 177         @param mode: Monitoring mode (BANDWIDTH or TRAFFIC [=default])
 178         """
 179         util.clear()
 180         if mode == TRAFFIC:
 181             entries = sorted(self.entries.get_traffic())
 182         else:
 183             entries = sorted(self.entries.get_usage())
 184
 185         for bytes_in, bytes_out, cmd in entries:
 186             if bytes_in or bytes_out:
 187                 if len(cmd) > 60:
 188                     cmd = cmd[:57] + '...'
 189                 print '%10d / %10d -- %s' % (bytes_in, bytes_out, cmd)
 190                 sys.stdout.flush()
 191
 192     def loop(self, mode):
 193         """The mainloop of a standalone monitor
 194
 195         @param mode: Monitoring mode (BANDWIDTH or TRAFFIC [=default])
 196         """
 197         while True:
 198             self.update(self.entries)
 199             self.output(mode)
 200             time.sleep(self.update_frequency)
 201
 202     def close(self):
 203         """Close this bandwidth monitor"""
 204         pass
 205
 206
 207 def islocal(ip):
 208     """Check if an IP is the local host
 209
 210     @return: True if the IP is in the loopback interface
 211     """
 212     return ip.startswith('127.0.0.') or ip.startswith('0.0.0.0')
 213