bwmon/monitor.py

   1 # -*- coding: utf-8 -*-
   2
   3 from __future__ import absolute_import
   4
   5 from bwmon import proc
   6 from bwmon import util
   7 from bwmon import model
   8
   9 import collections
  10 import time
  11 import sys
  12 import copy
  13 import re
  14
  15 BANDWIDTH, TRAFFIC = range(2)
  16
  17 class Monitor(object):
  18     """The ip_conntrack-based, system-wide bandwidth monitor
  19
  20     This object implements the higher-level management functions
  21     for the ip_conntrack-based monitoring method (system-wide,
  22     non-shaping, Linux-specific).
  23
  24     The monitor has a default update frequency of 1 second.
  25     """
  26     DEFAULT_UPDATE_FREQUENCY = 1
  27
  28     def __init__(self, lookback=True, ignorelocal=False):
  29         """Create a new Monitor object
  30
  31         This initializes the monitor object. In case the
  32         lookback value is False, this call already takes
  33         the first measurement from ip_conntrack.
  34
  35         @param lookback: TODO
  36         @param ignorelocal: TODO
  37         """
  38         self.fd_map = {}
  39         self.sample_time = time.time()
  40         self.conntrack = {}
  41         self.last_conntrack = {}
  42         self.init_conntrack = {} if lookback else proc.parse_ip_conntrack()
  43         self.connections = {}
  44         self.update_frequency = self.DEFAULT_UPDATE_FREQUENCY
  45         self.entries = model.MonitorEntryCollection(self.update_frequency)
  46         self.include_filter = []
  47         self.exclude_filter = []
  48         self.ignorelocal = ignorelocal
  49
  50     def update(self, entry_collection):
  51         """Update this monitor from a MonitorEntryCollection
  52
  53         This function gets a file descriptor to process name
  54         mapping, and re-parses the ip_conntrack data. The
  55         current time is saved as sample time, and the data is
  56         then converted into an aggregated monitoring value.
  57
  58         @param entry_collection: The collection from where to take data
  59         """
  60         self.fd_map.update(proc.get_fd_map())
  61         self.last_conntrack = copy.deepcopy(self.conntrack)
  62         self.conntrack.update( self.sub_conntrack(proc.parse_ip_conntrack(), self.init_conntrack) )
  63         self.connections.update(proc.get_connections())
  64         entry_collection.expire()
  65         self.sample_time = time.time()
  66         self.convert(entry_collection)
  67
  68     def sub_conntrack(self, conntrack_current, conntrack_init):
  69         """Subtract inital conntrack data from a measurement
  70
  71         @param conntrack_current: The current measurement
  72         @param conntrack_init: The initally measured data
  73         @return: The converted conntrack data
  74         """
  75         for (k, v_init) in conntrack_init.iteritems():
  76             if k in conntrack_current:
  77                 v_current = conntrack_current[k]
  78                 v_current['bytes'] = int(v_current['bytes']) - int(v_init['bytes'])
  79                 v_current['packets'] = int(v_current['packets']) - int(v_init['packets'])
  80                 conntrack_current[k] = v_current
  81
  82         return conntrack_current
  83
  84     def set_filter(self, include_filter, exclude_filter):
  85         """Apply inclusive and exclusive filters on this Monitor
  86
  87         This function takes two lists of regular expression strings,
  88         and will compile them into regular expression objects. The
  89         filters will be used to exclude and include commands from
  90         the monitoring output (see the config file docs for details).
  91
  92         @param include_filter: A list of regular expressions to include
  93         @param exclude_filter: A list of regular expressions to exclude
  94         """
  95         self.include_filter = [re.compile(f) for f in include_filter] if include_filter else []
  96         self.exclude_filter = [re.compile(f) for f in exclude_filter] if exclude_filter else []
  97
  98     def convert(self, entry_collection):
  99         """Apply a per-process merge of a MonitorEntryCollection
 100
 101         This function takes a MonitorEntryCollection object
 102         and merges new connections into it.
 103
 104         @param entry_collection: A MonitorEntryCollection object
 105         """
 106         entries = collections.defaultdict(lambda: (0, 0))
 107         for con in self.connections.itervalues():
 108             inode = con.get('inode', None)
 109             process = self.fd_map.get(inode, None)
 110
 111             if process is None:
 112                 continue
 113
 114             if self.include_filter and not any([f.search(process['cmd']) for f in self.include_filter]):
 115                 continue
 116
 117             if self.exclude_filter and any([f.search(process['cmd']) for f in self.exclude_filter]):
 118                 continue
 119
 120             if self.ignorelocal and islocal(con['remote']) and islocal(con['local']):
 121                 continue
 122
 123             key_in  = proc.ip_hash(con['remote'], con['local'])
 124             key_out = proc.ip_hash(con['local'], con['remote'])
 125             keys = {'in': key_in, 'out': key_out}
 126             new_byte = {'in': 0, 'out': 0}
 127
 128             for direction in ('in', 'out'):
 129                 k = keys[direction]
 130                 if k in self.conntrack:
 131                     if key_in in self.last_conntrack:
 132                         new_byte[direction] = int(self.conntrack[k]['bytes']) - int(self.last_conntrack[k]['bytes'])
 133                     else:
 134                         new_byte[direction] = int(self.conntrack[k]['bytes'])
 135
 136             current_in, current_out = entries[process['cmd']]
 137             new_in, new_out = (new_byte['in'], new_byte['out'])
 138
 139             entries[process['cmd']] = (current_in + new_in, current_out + new_out)
 140
 141         for key in entries:
 142             new_in, new_out = entries[key]
 143             old_in, old_out, timestamp = entry_collection.get_last_bytes(key)
 144             entry = model.MonitorEntry(key, old_in + new_in, old_out + new_out, self.sample_time)
 145             entry_collection.add(entry)
 146
 147     def output(self, mode=TRAFFIC):
 148         """Print the current status to standard output
 149
 150         @param mode: Monitoring mode (BANDWIDTH or TRAFFIC [=default])
 151         """
 152         util.clear()
 153         if mode == TRAFFIC:
 154             entries = sorted(self.entries.get_traffic())
 155         else:
 156             entries = sorted(self.entries.get_usage())
 157
 158         for bytes_in, bytes_out, cmd in entries:
 159             if bytes_in or bytes_out:
 160                 if len(cmd) > 60:
 161                     cmd = cmd[:57] + '...'
 162                 print '%10d / %10d -- %s' % (bytes_in, bytes_out, cmd)
 163                 sys.stdout.flush()
 164
 165     def loop(self, mode):
 166         """The mainloop of a standalone monitor
 167
 168         @param mode: Monitoring mode (BANDWIDTH or TRAFFIC [=default])
 169         """
 170         while True:
 171             self.update(self.entries)
 172             self.output(mode)
 173             time.sleep(self.update_frequency)
 174
 175     def close(self):
 176         """Close this bandwidth monitor"""
 177         pass
 178
 179
 180 def islocal(ip):
 181     """Check if an IP is the local host
 182
 183     @return: True if the IP is in the loopback interface
 184     """
 185     return ip.startswith('127.0.0.') or ip.startswith('0.0.0.0')
 186